Diffinery

codician/diffinerypublic

Open-source, agent-first GitHub and GitHub Actions alternative that runs on your own Cloudflare account: git hosting, pull requests, code review and drop-in GitHub Actions YAML on Workers, Durable Objects, R2 and Containers.

Push path: a retried push keeps its staged body and its moved refs, and small pushes are not compressed twice (#192)

* Push path: the Worker alone deletes a staged push, a retry counts a ref already moved as updated, and small pushes keep their objects' zlib data
test-bot authored 1 parent b79522ccommit 1998051a0f75ce604d7db05ab303354c68caef3fBrowse files
7 files changed+71−28
Files (7)
modifiedapps/diffinery/src/git/push-contents.ts+3−0View file
@@ -32,6 +32,9 @@
3232 write: async () => {
3333 for (const o of objects) {
3434 if (o.deflated) await store.writeDeflated(o.oid, o.type, o.size!, o.deflated);
35+ // An object that isn't a delta keeps the zlib data it came with: compressing it again cost
36+ // CPU on every push for the same bytes.
37+ else if (o.zlib) await store.writeDeflated(o.oid, o.type, o.content.length, o.zlib, o.content);
3538 else await store.writeKnown(o.oid, o.type, o.content);
3639 }
3740 },
modifiedapps/diffinery/src/http/git.ts+14−12View file
@@ -57,22 +57,24 @@
5757 /**
5858 * A deploy resets every Durable Object, and one doing a push then fails it mid-way: git doesn't
5959 * retry a push, so it would print a 500. The push is handed over once more (to the new code; a
60- * large one is still staged in R2, as the reset repository never got to delete it). Nothing was
61- * lost or done twice: objects are stored by their ids, and refs move only at the end, each by
62- * compare-and-swap (one the reset push already moved is reported as stale, not moved again).
60+ * large one is still staged in R2: only this function deletes it, once the push is done either way).
61+ * Nothing is lost or moved twice: objects are stored by their ids, and refs move only at the end,
62+ * each by compare-and-swap; one the cut-off attempt already moved to the pushed commit is reported
63+ * as updated.
6364 */
6465 export async function receiveWithRetry(env: Env, repoId: string, stub: ReturnType<typeof repoStub>, body: Uint8Array | SpilledBody, caller: Caller, moved: boolean): Promise<Uint8Array> {
65- for (let attempt = 1; ; attempt++) {
66- try {
67- return await stub.gitReceivePack(body, caller, { moved });
68- } catch (e) {
69- if (attempt >= PUSH_TRIES || !resetByDeploy(e)) {
70- if (!(body instanceof Uint8Array)) await env.BUCKET.delete(body.spilled).catch(() => {});
71- throw e;
66+ try {
67+ for (let attempt = 1; ; attempt++) {
68+ try {
69+ return await stub.gitReceivePack(body, caller, { moved, retry: attempt > 1 });
70+ } catch (e) {
71+ if (attempt >= PUSH_TRIES || !resetByDeploy(e)) throw e;
72+ // A stub that saw its object reset is broken; a new one reaches the new code.
73+ stub = repoStub(env, repoId);
7274 }
73- // A stub that saw its object reset is broken; a new one reaches the new code.
74- stub = repoStub(env, repoId);
7575 }
76+ } finally {
77+ if (!(body instanceof Uint8Array)) await env.BUCKET.delete(body.spilled).catch(() => {});
7678 }
7779 }
7880
modifiedapps/diffinery/src/repo/RepoDO.ts+13−10View file
@@ -549,17 +549,17 @@
549549 return readable;
550550 }
551551
552- /** `moved`: the push came in under an old name (a transfer or rename), so say where the repository lives now, like GitHub. */
553- async gitReceivePack(input: Uint8Array | SpilledBody, caller: Caller, o: { moved?: boolean } = {}): Promise<Uint8Array> {
554- if (input instanceof Uint8Array) return this.receivePack(input, caller, o);
555- try {
556- return await this.receivePack(input, caller, o);
557- } finally {
558- await this.env.BUCKET.delete(input.spilled).catch(() => {});
559- }
552+ /**
553+ * `moved`: the push came in under an old name (a transfer or rename), so say where the repository
554+ * lives now, like GitHub. `retry`: the Worker hands over a push a deploy cut off (http/git.ts
555+ * receiveWithRetry), which may already have moved some refs. A body staged in R2 is the Worker's
556+ * to delete: deleted here, a push cut off after this point would find it gone on its retry.
557+ */
558+ async gitReceivePack(input: Uint8Array | SpilledBody, caller: Caller, o: { moved?: boolean; retry?: boolean } = {}): Promise<Uint8Array> {
559+ return this.receivePack(input, caller, o);
560560 }
561561
562- private async receivePack(input: Uint8Array | SpilledBody, caller: Caller, o: { moved?: boolean }): Promise<Uint8Array> {
562+ private async receivePack(input: Uint8Array | SpilledBody, caller: Caller, o: { moved?: boolean; retry?: boolean }): Promise<Uint8Array> {
563563 const size = input instanceof Uint8Array ? input.length : input.size;
564564 if (size > MAX_PUSH_BYTES) {
565565 const req = { commands: [], capabilities: new Set<string>(), pushOptions: [], pack: null };
@@ -687,7 +687,10 @@
687687 results.push({ ref: c.ref, ok: false, reason: "the repository's unreachable objects were purged during this push; push again" });
688688 continue;
689689 }
690- const applied = this.casRef(c.cmd.ref, c.cmd.oldOid, c.cmd.newOid);
690+ // On a retry, a ref the cut-off attempt already moved is where the push wants it: that is a
691+ // success, not stale info. Its push event goes out (again), as the cut-off attempt may have
692+ // stopped before sending it: a missing one would leave a pull request without its checks.
693+ const applied = this.casRef(c.cmd.ref, c.cmd.oldOid, c.cmd.newOid) || (!!o.retry && (this.getRef(c.cmd.ref) ?? ZERO_OID) === c.cmd.newOid);
691694 results.push(applied ? { ref: c.ref, ok: true } : { ref: c.ref, ok: false, reason: "fetch first (stale info)" });
692695 if (applied && importMode) {
693696 imported++;
modifiedapps/diffinery/src/repo/objects.ts+3−1View file
@@ -144,12 +144,14 @@
144144 /**
145145 * Store an object from zlib data that is already in hand (a large blob as it sits in a pushed
146146 * pack), without inflating or recompressing it. The data goes to R2 in slices, never copied whole.
147+ * `content`: the object inflated, when already in hand, to keep in the memory cache as writeKnown does.
147148 */
148- async writeDeflated(oid: string, type: ObjectType, size: number, deflated: Uint8Array): Promise<void> {
149+ async writeDeflated(oid: string, type: ObjectType, size: number, deflated: Uint8Array, content?: Uint8Array): Promise<void> {
149150 if (this.has(oid)) return;
150151 const r2 = deflated.length > INLINE_LIMIT;
151152 if (r2) await this.bucket.put(this.r2Key(oid), sized(slices(deflated), deflated.length));
152153 this.insert(oid, type, size, deflated.length, r2 ? null : deflated.slice());
154+ if (content) this.remember(oid, { type, content });
153155 }
154156
155157 /**
modifiedapps/diffinery/test/unit/push-retry.test.ts+15−4View file
@@ -9,10 +9,12 @@
99 function fakes(failures: number, message = RESET) {
1010 let calls = 0;
1111 const stubs: number[] = [];
12+ const retries: boolean[] = [];
1213 const deleted: string[] = [];
1314 const stub = (n: number) => ({
14- gitReceivePack: async () => {
15+ gitReceivePack: async (_body: unknown, _caller: unknown, o: { retry?: boolean }) => {
1516 stubs.push(n);
17+ retries.push(!!o.retry);
1618 if (calls++ < failures) throw new Error(message);
1719 return new Uint8Array([1]);
1820 },
@@ -22,20 +24,29 @@
2224 REPO: { idFromName: (name: string) => name, get: () => stub(++made) },
2325 BUCKET: { delete: async (key: string) => void deleted.push(key) },
2426 } as unknown as Env;
25- return { env, first: stub(0) as never, stubs, deleted };
27+ return { env, first: stub(0) as never, stubs, retries, deleted };
2628 }
2729
2830 const caller = { role: "write" } as unknown as Caller;
2931 const spilled = { spilled: "push/abc", size: 70 << 20 };
3032
3133 describe("a push cut off by a deploy", () => {
32- it("is handed to the new code once more, through a new stub", async () => {
34+ it("is handed to the new code once more, through a new stub, marked as a retry", async () => {
3335 const f = fakes(1);
3436 expect(await receiveWithRetry(f.env, "r1", f.first, spilled, caller, false)).toEqual(new Uint8Array([1]));
3537 expect(f.stubs).toEqual([0, 1]);
36- expect(f.deleted).toEqual([]);
38+ expect(f.retries).toEqual([false, true]);
39+ // The staged body stays for the retry and is deleted once the push is done.
40+ expect(f.deleted).toEqual(["push/abc"]);
3741 });
3842
43+ it("deletes the staged body after a push that went through the first time", async () => {
44+ const f = fakes(0);
45+ await receiveWithRetry(f.env, "r1", f.first, spilled, caller, false);
46+ expect(f.retries).toEqual([false]);
47+ expect(f.deleted).toEqual(["push/abc"]);
48+ });
49+
3950 it("fails after a second reset, and the staged body is deleted", async () => {
4051 const f = fakes(2);
4152 await expect(receiveWithRetry(f.env, "r1", f.first, spilled, caller, false)).rejects.toThrow(RESET);
modifiedpackages/git-core/src/pack.ts+8−1View file
@@ -16,6 +16,12 @@
1616 */
1717 deflated?: Uint8Array;
1818 size?: number;
19+ /**
20+ * Set on an object that isn't a delta (and not a large blob, which has `deflated`): its zlib data as
21+ * it sits in the pack (a view, not a copy), which a store that keeps objects zlib-compressed can
22+ * keep as is instead of compressing `content` again.
23+ */
24+ zlib?: Uint8Array;
1925 }
2026
2127 /** A large blob kept compressed: `content` inflates it when something needs the bytes. */
@@ -175,12 +181,13 @@
175181 hold(size + OBJECT_OVERHEAD);
176182 const { out, consumed } = inflateAt(pack, pos, size);
177183 if (out.length !== size) throw new Error(`size mismatch at ${offset}: ${out.length} != ${size}`);
184+ const zlib = pack.subarray(pos, pos + consumed);
178185 pos += consumed;
179186 if (baseOffset !== undefined) waitFor(childrenByOffset, baseOffset, { offset, data: out });
180187 else if (baseOid !== undefined) waitFor(childrenByOid, baseOid, { offset, data: out });
181188 else {
182189 const type = CODE_TYPES[typeCode]!;
183- add(offset, { oid: await hashObject(type, out), type, content: out });
190+ add(offset, { oid: await hashObject(type, out), type, content: out, zlib });
184191 }
185192 }
186193 if (pos !== pack.length - 20) throw new Error("trailing garbage in pack");
modifiedpackages/git-core/test/pack.test.ts+15−0View file
@@ -62,6 +62,21 @@
6262 for (const oid of expected) expect(store.objects.has(oid), oid).toBe(true);
6363 });
6464
65+ it("hands over the zlib data of every object that isn't a delta, which inflates to its content", async () => {
66+ const dir = makeRepo();
67+ const expected = git(dir, "rev-list", "--objects", "--all").toString().trim().split("\n").map((l) => l.split(" ")[0]!);
68+ const pack = execFileSync("git", ["pack-objects", "--stdout", "--delta-base-offset"], { cwd: dir, input: expected.join("\n") + "\n" });
69+ const objs = await parsePack(new Uint8Array(pack));
70+ const whole = objs.filter((o) => o.zlib);
71+ expect(whole.length).toBeGreaterThan(0);
72+ // The second commit's a.txt is a delta against the first: rebuilt, so it has no zlib data of its own.
73+ expect(objs.some((o) => !o.zlib)).toBe(true);
74+ for (const o of whole) {
75+ expect(inflate(o.zlib!)).toEqual(o.content);
76+ expect(await hashObject(o.type, inflate(o.zlib!))).toBe(o.oid);
77+ }
78+ });
79+
6580 it("keeps large blobs compressed, with the same oids, and still resolves deltas against them", async () => {
6681 const dir = makeRepo();
6782 const expected = git(dir, "rev-list", "--objects", "--all").toString().trim().split("\n").map((l) => l.split(" ")[0]!);