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.

Large pushes into one repository take turns instead of resetting it (#243)

* Large pushes into one repository take turns instead of resetting it
test-bot authored 1 parent a7fb4e7commit 4798de7b2e20f76f4cf5e2564dcf6ef1f68c7797Browse files
7 files changed+251−24
Files (7)
modifiedapps/diffinery/docs/reference/limits.md+1−0View file
@@ -136,6 +136,7 @@
136136 | Limit | Value | Applies to |
137137 | --- | --- | --- |
138138 | Push size | 100 MB per push | The whole push request after decompression. A larger push is rejected with `push exceeds 100 MB; split it into smaller pushes`. Push in several steps (for example a range of commits at a time). |
139+| Pushes to one repository at once | Large ones unpack one at a time | Small pushes (up to 1 MiB) unpack side by side. A larger push needs the repository's full memory share to unpack, so large pushes to the same repository take turns, oldest first; one that waits keeps its connection open and goes through once the push before it is done. Pushes to different repositories don't wait for each other. |
139140 | Fetch request size | 16 MB | The request a client sends to fetch or clone (its wants and haves), after decompression. A larger one is refused with `413 fetch request exceeds 16 MB`. |
140141 | File size in a push | Up to the push size | A single file (blob) can be as large as fits in one push (GitHub allows files up to 100 MB). Files of 4 MiB or more are checked and stored from their compressed bytes, so they never need to fit in memory inflated. |
141142 | Unpacked size of a push | 512 MiB | What a push's pack inflates to, deltas resolved. A pack that unpacks to more is refused with `pack unpacks to more than 512 MiB; split it into smaller pushes`, however small it is on the wire: a long history of one large file can, since each version counts whole. |
modifiedapps/diffinery/src/git/push-contents.ts+3−3View file
@@ -144,12 +144,12 @@
144144 * A pack staged in R2, unpacked without holding it whole (see StreamedPack): every object is
145145 * checked as it is resolved, and read again from R2 to be stored.
146146 */
147-export function stagedContents(store: RepoObjectStore, bucket: R2Bucket, key: string, packOffset: number, packBytes: number, o: { largeObjectBytes: number }): Promise<PushContents> {
147+export function stagedContents(store: RepoObjectStore, bucket: R2Bucket, key: string, packOffset: number, packBytes: number, o: { largeObjectBytes: number; maxHeldBytes?: number }): Promise<PushContents> {
148148 return streamedContents(store, r2Source(bucket, key, packOffset, packBytes), o);
149149 }
150150
151151 /** A pack unpacked from `src` in pieces, one object at a time (see StreamedPack). */
152-export async function streamedContents(store: RepoObjectStore, src: PackSource, o: { largeObjectBytes: number }): Promise<PushContents> {
152+export async function streamedContents(store: RepoObjectStore, src: PackSource, o: { largeObjectBytes: number; maxHeldBytes?: number }): Promise<PushContents> {
153153 const packBytes = src.size;
154154 const hits: SecretHit[] = [];
155155 const opted: SecretHit[] = [];
@@ -157,7 +157,7 @@
157157 let refusal: string | null = null;
158158 let pack: StreamedPack | null = null;
159159 try {
160- const p = (pack = await StreamedPack.open(src, { external: repositoryBases(store), largeObjectBytes: o.largeObjectBytes, writesOver: INLINE_LIMIT }, async (obj, content, feed) => {
160+ const p = (pack = await StreamedPack.open(src, { external: repositoryBases(store), largeObjectBytes: o.largeObjectBytes, maxHeldBytes: o.maxHeldBytes, writesOver: INLINE_LIMIT }, async (obj, content, feed) => {
161161 const why = obj.type === "tree" ? fsckTree(obj.oid, content!, gitmodules) : content ? fsckSize(obj.oid, obj.type, content.length) : null;
162162 if (why) throw new FsckRefusal(why);
163163 if (scanned(obj.type)) {
modifiedapps/diffinery/src/http/git.ts+28−5View file
@@ -49,8 +49,8 @@
4949 // A GitHub import's own token, while that import is running and filling this repository, may push in import mode.
5050 const p = c.get("principal");
5151 if (p.type === "account" && p.importId && p.tokenId && (await importPushAllowed(c.env, p.importId, repo.id, p.tokenId))) caller.importId = p.importId;
52- const body = await readPushBody(c.env, repo.id, c.req.raw);
53- const out = await receiveWithRetry(c.env, repo.id, stub, body, caller, moved);
52+ const { body, release } = await readPushBody(c.env, repo.id, c.req.raw);
53+ const out = await receiveWithRetry(c.env, repo.id, stub, body, caller, moved).finally(release);
5454 c.executionCtx.waitUntil(c.get("dir").touchPushed(repo.id));
5555 return new Response(out, { headers: { "content-type": "application/x-git-receive-pack-result", ...noCache } });
5656 });
@@ -120,13 +120,34 @@
120120
121121 /** Pushes up to this size are handed to the repository in memory; larger ones go through R2. */
122122 const IN_MEMORY_PUSH = 16 * 1024 * 1024;
123+/**
124+ * Push bodies one Worker isolate holds in memory at once, across its requests (it has 128 MB, and a
125+ * body is held twice while it is joined): past this, a push goes through R2 even under IN_MEMORY_PUSH.
126+ */
127+const ISOLATE_PUSH_BYTES = 32 * 1024 * 1024;
128+let bufferedPushBytes = 0;
123129
124130 /**
125131 * A push's body. A large one is staged in R2 as it arrives (a Worker and a Durable Object each
126132 * have 128 MB of memory, and holding a 70 MB push twice, then copying it to the repository, does
127133 * not fit); the repository reads it from there once and deletes it.
128134 */
129-async function readPushBody(env: Env, repoId: string, req: Request): Promise<Uint8Array | SpilledBody> {
135+async function readPushBody(env: Env, repoId: string, req: Request): Promise<{ body: Uint8Array | SpilledBody; release: () => void }> {
136+ let counted = 0;
137+ const release = () => {
138+ bufferedPushBytes -= counted;
139+ counted = 0;
140+ };
141+ try {
142+ return { body: await readPushBodyCounted(env, repoId, req, (n) => ((counted += n), (bufferedPushBytes += n))), release };
143+ } catch (e) {
144+ release();
145+ throw e;
146+ }
147+}
148+
149+/** readPushBody's body; `count` is told of every byte held in memory (ISOLATE_PUSH_BYTES). */
150+async function readPushBodyCounted(env: Env, repoId: string, req: Request, count: (bytes: number) => void): Promise<Uint8Array | SpilledBody> {
130151 const tooLarge = "push exceeds 100 MB; split it into smaller pushes";
131152 let stream = req.body;
132153 if (!stream) return new Uint8Array(0);
@@ -139,8 +160,10 @@
139160 if (done) break;
140161 total += value.length;
141162 chunks.push(value);
142- if (total > IN_MEMORY_PUSH) {
143- // Too big to pass along in memory: what came so far, then the rest, goes to R2.
163+ count(value.length);
164+ if (total > IN_MEMORY_PUSH || bufferedPushBytes > ISOLATE_PUSH_BYTES) {
165+ // Too big to pass along in memory (or this isolate holds enough already): what came so far, then the rest, goes to R2.
166+ count(-total);
144167 const key = pushBodyKey(repoId);
145168 const rest = new ReadableStream<Uint8Array>({
146169 start(controller) {
addedapps/diffinery/src/lib/memory-pool.ts+64−0View file
@@ -1,0 +1,64 @@
1+/** A share of a MemoryPool: what one task may hold until it releases it. */
2+export interface Lease {
3+ readonly bytes: number;
4+ /** Waits for a larger share: this one is given back first, so nothing it held may still be in use. */
5+ grow(bytes: number): Promise<void>;
6+ /** Gives the share back; a second call does nothing. */
7+ release(): void;
8+}
9+
10+/**
11+ * Bytes that tasks in one isolate may hold at once (a Durable Object or a Worker has 128 MB, shared
12+ * by every request in flight there). take(n) waits until n bytes fit beside what is taken, behind
13+ * every earlier waiter, so a large task isn't kept waiting by a stream of small ones. A task asking
14+ * for more than the whole pool gets all of it, and runs alone.
15+ */
16+export class MemoryPool {
17+ private used = 0;
18+ private readonly queue: Array<{ n: number; go: () => void }> = [];
19+
20+ constructor(readonly size: number) {}
21+
22+ /** Whether `bytes` would be given at once. */
23+ fits(bytes: number): boolean {
24+ return !this.queue.length && this.used + Math.min(bytes, this.size) <= this.size;
25+ }
26+
27+ async take(bytes: number): Promise<Lease> {
28+ let n = await this.acquire(bytes);
29+ let held = true;
30+ const lease: Lease = {
31+ get bytes() {
32+ return held ? n : 0;
33+ },
34+ grow: async (more) => {
35+ if (held && more <= n) return;
36+ lease.release();
37+ n = await this.acquire(more);
38+ held = true;
39+ },
40+ release: () => {
41+ if (!held) return;
42+ held = false;
43+ this.used -= n;
44+ this.drain();
45+ },
46+ };
47+ return lease;
48+ }
49+
50+ private async acquire(bytes: number): Promise<number> {
51+ const n = Math.min(Math.max(0, bytes), this.size);
52+ if (this.fits(n)) this.used += n;
53+ else await new Promise<void>((go) => this.queue.push({ n, go }));
54+ return n;
55+ }
56+
57+ private drain(): void {
58+ while (this.queue.length && this.used + this.queue[0]!.n <= this.size) {
59+ const w = this.queue.shift()!;
60+ this.used += w.n;
61+ w.go();
62+ }
63+ }
64+}
modifiedapps/diffinery/src/repo/RepoDO.ts+57−16View file
@@ -27,6 +27,7 @@
2727 readTree,
2828 resolvePath,
2929 TooMuchMemory,
30+ MAX_HELD_BYTES,
3031 serializeCommit,
3132 walkCommits,
3233 type Commit,
@@ -53,7 +54,8 @@
5354 import { scanForSecrets, secretBypassReason, type SecretHit } from "../git/secret-scan.js";
5455 import { GHOST, HttpError, isValidRefName, now } from "../lib/util.js";
5556 import { R2Sink } from "../lib/r2-sink.js";
56-import { archivePrefix, deletePrefixes, repoPrefixes, type SpilledBody } from "../lib/storage-keys.js";
57+import { archivePrefix, deletePrefixes, pushBodyKey, repoPrefixes, type SpilledBody } from "../lib/storage-keys.js";
58+import { MemoryPool, type Lease } from "../lib/memory-pool.js";
5759 import { storageRefusal } from "../lib/quotas.js";
5860 import { Actions } from "./actions.js";
5961 import { AutoMerge, isPerson } from "./auto-merge.js";
@@ -77,6 +79,16 @@
7779 const MAX_PUSH_BYTES = 100 * 1024 * 1024;
7880 /** Pushed blobs this large are hashed, scanned and stored from their compressed bytes, never held inflated. */
7981 const LARGE_OBJECT_BYTES = 4 * 1024 * 1024;
82+/**
83+ * What pushes may hold at once in this object while they unpack: one push may hold MAX_HELD_BYTES, besides
84+ * a body of up to 16 MiB passed in memory (http/git.ts IN_MEMORY_PUSH), and the object has 128 MB. Without
85+ * it, two large pushes unpacking at once could go past that, and the reset drops every request in flight
86+ * here, other pushes and fetches included. Each push reserves what it may hold and waits its turn.
87+ */
88+const PUSH_MEMORY_BYTES = MAX_HELD_BYTES + 16 * 1024 * 1024;
89+/** A push with a body this small unpacks within SMALL_PUSH_HELD, beside others; one that needs more waits for a full share. */
90+const SMALL_PUSH_BYTES = 1024 * 1024;
91+const SMALL_PUSH_HELD = 8 * 1024 * 1024;
8092 /** A purge walks every reachable commit and tree while holding the repository: bounded so it finishes in time. */
8193 export const MAX_PURGE_OBJECTS = 300_000;
8294 /** Storage key that marks a deleted repository whose R2 objects are still being deleted. */
@@ -576,10 +588,30 @@
576588 * to delete: deleted here, a push cut off after this point would find it gone on its retry.
577589 */
578590 async gitReceivePack(input: Uint8Array | SpilledBody, caller: Caller, o: { moved?: boolean; retry?: boolean } = {}): Promise<Uint8Array> {
579- return this.receivePack(input, caller, o);
591+ const size = input instanceof Uint8Array ? input.length : input.size;
592+ const held = size <= SMALL_PUSH_BYTES ? SMALL_PUSH_HELD : MAX_HELD_BYTES;
593+ // A large body that has to wait goes to R2 meanwhile, so pushes waiting their turn don't hold their bodies.
594+ let staged: string | null = null;
595+ if (input instanceof Uint8Array && size > SMALL_PUSH_BYTES && !this.pushMemory.fits(size + held)) {
596+ staged = pushBodyKey(this.meta.repoId);
597+ await this.env.BUCKET.put(staged, input);
598+ input = { spilled: staged, size };
599+ }
600+ const body = input instanceof Uint8Array ? size : 0;
601+ const lease = await this.pushMemory.take(body + held);
602+ try {
603+ return await this.receivePack(input, caller, o, { held, grow: () => lease.grow(body + MAX_HELD_BYTES) });
604+ } finally {
605+ lease.release();
606+ if (staged) await this.env.BUCKET.delete(staged).catch(() => {});
607+ }
580608 }
581609
582- private async receivePack(input: Uint8Array | SpilledBody, caller: Caller, o: { moved?: boolean; retry?: boolean }): Promise<Uint8Array> {
610+ /** What pushes in flight here hold while they unpack (see PUSH_MEMORY_BYTES). */
611+ private readonly pushMemory = new MemoryPool(PUSH_MEMORY_BYTES);
612+
613+ /** `memory`: what this push may hold while it unpacks, and how to wait for a full share when a small push needs more. */
614+ private async receivePack(input: Uint8Array | SpilledBody, caller: Caller, o: { moved?: boolean; retry?: boolean }, memory: { held: number; grow: () => Promise<void> }): Promise<Uint8Array> {
583615 const size = input instanceof Uint8Array ? input.length : input.size;
584616 if (size > MAX_PUSH_BYTES) {
585617 const req = { commands: [], capabilities: new Set<string>(), pushOptions: [], pack: null };
@@ -619,20 +651,29 @@
619651 // Like git's quarantine: nothing from the pack is stored until push protection passes,
620652 // so a blocked push leaves no objects behind for a retry to build on.
621653 let contents: PushContents = packedContents(this.store, [], 0);
654+ const unpack = async (maxHeldBytes: number): Promise<PushContents> => {
655+ const opts = { largeObjectBytes: LARGE_OBJECT_BYTES, maxHeldBytes };
656+ if (!(input instanceof Uint8Array)) return packAt < input.size ? stagedContents(this.store, this.env.BUCKET, input.spilled, packAt, input.size - packAt, opts) : contents;
657+ const pack = req.pack;
658+ if (!pack?.length) return contents;
659+ return parsePack(pack, repositoryBases(this.store), opts).then(
660+ (objects) => packedContents(this.store, objects, pack.length),
661+ // Parsed whole, a pack holds every object at once: one that can't, such as a long history
662+ // of a large file, is unpacked in pieces instead, holding only the objects in use.
663+ (e: unknown) => {
664+ if (!(e instanceof TooMuchMemory)) throw e;
665+ return streamedContents(this.store, memorySource(pack), opts);
666+ },
667+ );
668+ };
622669 try {
623- if (!(input instanceof Uint8Array)) {
624- if (packAt < input.size) contents = await stagedContents(this.store, this.env.BUCKET, input.spilled, packAt, input.size - packAt, { largeObjectBytes: LARGE_OBJECT_BYTES });
625- } else if (req.pack && req.pack.length > 0) {
626- const pack = req.pack;
627- contents = await parsePack(pack, repositoryBases(this.store), { largeObjectBytes: LARGE_OBJECT_BYTES }).then(
628- (objects) => packedContents(this.store, objects, pack.length),
629- // Parsed whole, a pack holds every object at once: one that can't, such as a long history
630- // of a large file, is unpacked in pieces instead, holding only the objects in use.
631- (e: unknown) => {
632- if (!(e instanceof TooMuchMemory)) throw e;
633- return streamedContents(this.store, memorySource(pack), { largeObjectBytes: LARGE_OBJECT_BYTES });
634- },
635- );
670+ try {
671+ contents = await unpack(memory.held);
672+ } catch (e) {
673+ // A small push that needs more than its small share waits for a full one.
674+ if (!(e instanceof TooMuchMemory) || memory.held >= MAX_HELD_BYTES) throw e;
675+ await memory.grow();
676+ contents = await unpack(MAX_HELD_BYTES);
636677 }
637678 } catch (e) {
638679 if (resetByDeploy(e)) throw e;
modifiedapps/diffinery/test/e2e/e2e.test.mjs+38−0View file
@@ -2932,6 +2932,44 @@
29322932 assert.match(gitFails(src, ["push", "-q", `${base}/${me.login}/bigfile.git`, "main"]), /possible secrets detected/);
29332933 });
29342934
2935+test("large pushes: three large pushes into one repository at once all go in, and a fetch meanwhile works", async () => {
2936+ const me = await ok("GET", "/user");
2937+ await ok("POST", "/user/repos", { name: "pushes-at-once" });
2938+ const remote = `${base}/${me.login}/pushes-at-once.git`;
2939+ const src = join(work, "pushes-at-once");
2940+ mkdirSync(src, { recursive: true });
2941+ git(src, ["init", "-q", "-b", "main"]);
2942+ writeFileSync(join(src, "README.md"), "base\n");
2943+ git(src, ["add", "-A"]);
2944+ git(src, ["commit", "-q", "-m", "base"]);
2945+ git(src, ["push", "-q", remote, "main"]);
2946+ // Each branch: many medium files over a few commits, so unpacking holds a lot. Two packs stay under
2947+ // 16 MB (handed over in memory), one goes over (staged in R2), as when several agents push at once.
2948+ const sizes = { a: 180, b: 180, c: 420 };
2949+ const tips = {};
2950+ for (const [branch, count] of Object.entries(sizes)) {
2951+ git(src, ["checkout", "-q", "-b", branch, "main"]);
2952+ const files = Array.from({ length: count }, () => randomBytes(48 * 1024).toString("base64"));
2953+ mkdirSync(join(src, branch), { recursive: true });
2954+ for (let c = 0; c < 3; c++) {
2955+ files.forEach((f, i) => writeFileSync(join(src, branch, `f${i}.txt`), `${f}\nrevision ${c}\n${randomBytes(1024).toString("hex")}\n`));
2956+ git(src, ["add", "-A"]);
2957+ git(src, ["commit", "-q", "-m", `${branch} ${c}`]);
2958+ }
2959+ tips[branch] = git(src, ["rev-parse", "HEAD"]).trim();
2960+ }
2961+ git(src, ["checkout", "-q", "main"]);
2962+ const push = (branch) => gitAsync(src, ["-c", "http.postBuffer=1048576", "push", "-q", remote, branch]);
2963+ const fetchWhile = gitAsync(work, ["clone", "-q", remote, join(work, "pushes-at-once-early")]);
2964+ await Promise.all([push("a"), push("b"), push("c"), fetchWhile]);
2965+ const branches = (await ok("GET", `/repos/${me.login}/pushes-at-once/branches`)).map((b) => b.name).sort();
2966+ assert.deepEqual(branches, ["a", "b", "c", "main"]);
2967+ const out = join(work, "pushes-at-once-clone");
2968+ git(work, ["clone", "-q", "--mirror", remote, out]);
2969+ for (const [branch, tip] of Object.entries(tips)) assert.equal(git(out, ["rev-parse", `refs/heads/${branch}`]).trim(), tip);
2970+ git(out, ["fsck", "--full", "--strict"]);
2971+});
2972+
29352973 test("large pushes: a 40 MB history of many files goes in one push, read from R2 in ranges, and is still checked", async () => {
29362974 const me = await ok("GET", "/user");
29372975 await ok("POST", "/user/repos", { name: "bighistory" });
addedapps/diffinery/test/unit/memory-pool.test.ts+60−0View file
@@ -1,0 +1,60 @@
1+import { describe, expect, it } from "vitest";
2+import { MemoryPool } from "../../src/lib/memory-pool.js";
3+
4+const tick = () => new Promise((r) => setTimeout(r, 0));
5+
6+describe("MemoryPool", () => {
7+ it("gives shares that fit at once, and makes the rest wait, oldest first", async () => {
8+ const pool = new MemoryPool(100);
9+ const a = await pool.take(60);
10+ const b = await pool.take(30);
11+ expect(pool.fits(10)).toBe(true);
12+ const order: string[] = [];
13+ const c = pool.take(50).then((l) => (order.push("c"), l));
14+ // A small one that would fit still waits behind the large one, so the large one isn't starved.
15+ expect(pool.fits(10)).toBe(false);
16+ const d = pool.take(10).then((l) => (order.push("d"), l));
17+ await tick();
18+ expect(order).toEqual([]);
19+ a.release();
20+ a.release(); // a second release gives nothing back twice
21+ await tick();
22+ expect(order).toEqual(["c", "d"]);
23+ (await c).release();
24+ (await d).release();
25+ b.release();
26+ expect(pool.fits(100)).toBe(true);
27+ });
28+
29+ it("a share larger than the pool gets all of it, alone", async () => {
30+ const pool = new MemoryPool(100);
31+ const small = await pool.take(1);
32+ let got = false;
33+ const big = pool.take(500).then((l) => ((got = true), l));
34+ await tick();
35+ expect(got).toBe(false);
36+ small.release();
37+ const lease = await big;
38+ expect(lease.bytes).toBe(100);
39+ expect(pool.fits(1)).toBe(false);
40+ lease.release();
41+ expect(pool.fits(100)).toBe(true);
42+ });
43+
44+ it("grows a share by giving it back and waiting its turn for the larger one", async () => {
45+ const pool = new MemoryPool(100);
46+ const a = await pool.take(10);
47+ const b = await pool.take(50);
48+ let grown = false;
49+ const growing = a.grow(80).then(() => (grown = true));
50+ await tick();
51+ expect(grown).toBe(false);
52+ b.release();
53+ await growing;
54+ expect(a.bytes).toBe(80);
55+ await a.grow(20); // already large enough
56+ expect(a.bytes).toBe(80);
57+ a.release();
58+ expect(pool.fits(100)).toBe(true);
59+ });
60+});