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.

Imports: large files go first and whole, and batches of files are sent once (#197)

* Imports: large files go first and whole, and batches of files are sent once
test-bot authored 1 parent 9702f3ccommit 560b60e15ad9f9c1341f0844f3483f3bc12f8d51Browse files
4 files changed+118−15
Files (4)
modifiedapps/diffinery/test/e2e/e2e.test.mjs+4−1View file
@@ -2463,7 +2463,7 @@
24632463 const transfer = async (id) => {
24642464 await until(`import ${id} to wait for an agent`, async () => (await ok("GET", `/diffinery/imports/${id}`)).git?.phase === "waiting", 120_000);
24652465 const { token: pushToken } = await ok("POST", `/diffinery/imports/${id}/claim`);
2466- const r = await spawnAsync(process.execPath, [AGENT], { DIFFINERY_API_URL: base, DIFFINERY_IMPORT_ID: id, DIFFINERY_IMPORT_TOKEN: pushToken, DIFFINERY_IMPORT_CHUNK_BYTES: String(160 * 1024), DIFFINERY_WORK_ROOT: join(work, `import-${id}`) }, 240_000);
2466+ const r = await spawnAsync(process.execPath, [AGENT], { DIFFINERY_API_URL: base, DIFFINERY_IMPORT_ID: id, DIFFINERY_IMPORT_TOKEN: pushToken, DIFFINERY_IMPORT_CHUNK_BYTES: String(160 * 1024), DIFFINERY_IMPORT_LARGE_BLOB_BYTES: String(100 * 1024), DIFFINERY_WORK_ROOT: join(work, `import-${id}`) }, 240_000);
24672467 const out = `${r.stdout}${r.stderr}`;
24682468 assert.ok(!out.includes(GH_TOKEN) && !out.includes(pushToken), "the agent never prints tokens");
24692469 return { ...r, out, pushToken };
@@ -2539,6 +2539,9 @@
25392539 assert.equal(sink.last_response.status, "unused");
25402540 assert.ok(done.steps.every((s) => s.status === "done" || s.status === "skipped"), JSON.stringify(done.steps));
25412541 assert.ok(done.summary.git.chunks >= 3, `history went in chunks: ${JSON.stringify(done.summary.git)}`);
2542+ // The data files count as large here (DIFFINERY_IMPORT_LARGE_BLOB_BYTES): they went first, whole,
2543+ // under temporary refs naming trees, which the server takes and the import removes at the end.
2544+ assert.match(agent.out, /4 version\(s\) of large files/);
25422545 assert.equal(done.summary.git.branches, 3);
25432546 assert.equal(done.summary.git.tags, 2);
25442547 assert.equal(done.summary.git.lfsObjects, 1);
modifiedcontainers/runner/src/importer.ts+57−13View file
@@ -29,12 +29,21 @@
2929 workRoot: string;
3030 /** Overrides the spec's push size budget (tests use a small one to exercise chunking). */
3131 chunkBytes?: number;
32+ /** Overrides LARGE_BLOB_BYTES (tests use a small one). */
33+ largeBlobBytes?: number;
3234 log?: (line: string) => void;
3335 }
3436
3537 /** Chunk refs; only these are deleted at the end (pull request heads stay until the server has created refs/pull/N/head). */
3638 const TEMP_PREFIX = `${IMPORT_REF_PREFIX}tmp/`;
3739 const MIN_BUDGET = 64 * 1024;
40+/**
41+ * Files this large are pushed first, each version whole: sent as a delta, the server would hold the
42+ * version it is based on and the one it rebuilds at once, which it can't for files past about 30 MB
43+ * (a 32 MB file with a few versions stopped tactfolio's import). Whole, a large file is hashed and
44+ * stored as it streams in, up to the 100 MB a push may carry.
45+ */
46+const LARGE_BLOB_BYTES = 16 * 1024 * 1024;
3847 const REF_BATCH = 200;
3948
4049 export class ImportError extends Error {}
@@ -99,7 +108,7 @@
99108 };
100109 let result: ImportGitResult;
101110 try {
102- result = await transfer(spec, { workRoot: o.workRoot, chunkBytes: o.chunkBytes, progress, log });
111+ result = await transfer(spec, { workRoot: o.workRoot, chunkBytes: o.chunkBytes, largeBlobBytes: o.largeBlobBytes, progress, log });
103112 } catch (e) {
104113 result = emptyResult();
105114 result.ok = false;
@@ -126,6 +135,8 @@
126135 export interface TransferOptions {
127136 workRoot: string;
128137 chunkBytes?: number;
138+ /** Overrides LARGE_BLOB_BYTES (tests use a small one). */
139+ largeBlobBytes?: number;
129140 progress: (p: ImportProgress, force?: boolean) => Promise<void>;
130141 log: (line: string) => void;
131142 }
@@ -205,6 +216,7 @@
205216 const pushOpts = ["-o", "diffinery-import", ...(spec.secrets === "bypass" ? ["-o", `allow-secrets=${spec.bypassReason}`] : [])];
206217 const pusher = new ChunkPusher(git, spec, pushOpts, o, result);
207218 await pusher.loadRemoteState();
219+ await pusher.pushLargeBlobs(ordered.map((n) => refs.get(n)!.commit).filter((c): c is string => !!c), o.largeBlobBytes ?? LARGE_BLOB_BYTES);
208220 for (const name of ordered) {
209221 const ref = refs.get(name)!;
210222 if (ref.commit) await pusher.ensurePushed(ref.commit, name);
@@ -330,7 +342,8 @@
330342 const present = await this.git.run(["cat-file", "--batch-check=%(objectname) %(objecttype)"], { input: `${oids.join("\n")}\n` });
331343 for (const line of present.stdout.split("\n")) {
332344 const [oid, type] = line.split(" ");
333- if (type === "commit") this.addHave(oid!);
345+ // Chunks of history are commits; batches of files are trees (see pushBlobs).
346+ if (type === "commit" || type === "tree") this.addHave(oid!);
334347 }
335348 for (const line of r.stdout.split("\n")) {
336349 const ref = line.split("\t")[1];
@@ -386,7 +399,28 @@
386399 }
387400 }
388401
389- /** Upload the blobs `commit` adds through throwaway commits of flat trees, so the commit itself stays small. */
402+ /**
403+ * Every blob of `minBytes` or more in the history of `tips` that the server doesn't have yet, pushed
404+ * before the history and with no deltas (see LARGE_BLOB_BYTES). The history's pushes then leave
405+ * them out, as the server has them under a temporary ref.
406+ */
407+ async pushLargeBlobs(tips: string[], minBytes: number): Promise<void> {
408+ if (!tips.length) return;
409+ // Blobs of minBytes or more are the ones the filter leaves out, printed with a leading "~".
410+ const listed = await this.git.lines(["rev-list", "--objects", `--filter=blob:limit=${minBytes}`, "--filter-print-omitted", "--stdin"], `${[...new Set(tips)].join("\n")}\n${this.notHaves()}`);
411+ const omitted = listed.filter((l) => l.startsWith("~")).map((l) => l.slice(1).split(" ")[0]!);
412+ if (!omitted.length) return;
413+ const check = await this.git.lines(["cat-file", "--batch-check=%(objectname) %(objecttype) %(objectsize)"], `${omitted.join("\n")}\n`);
414+ const large = check
415+ .map((l) => l.split(" "))
416+ .filter((p) => p[1] === "blob")
417+ .map((p) => ({ oid: p[0]!, size: Number(p[2]) }));
418+ if (!large.length) return;
419+ this.o.log(`${large.length} version(s) of large files (${fmtBytes(minBytes)} or more) go first, whole`);
420+ await this.pushBlobs(large, "large files", true);
421+ }
422+
423+ /** Upload the blobs `commit` adds in batches (see pushBlobs), so the commit itself stays small. */
390424 private async pushObjectBatches(commit: string, label: string): Promise<void> {
391425 const listed = await this.git.lines(["rev-list", "--objects", "--stdin"], `${commit}\n${this.notHaves()}`);
392426 const oids = listed.map((l) => l.split(" ")[0]!);
@@ -395,25 +429,33 @@
395429 .map((l) => l.split(" "))
396430 .filter((p) => p[1] === "blob")
397431 .map((p) => ({ oid: p[0]!, size: Number(p[2]) }));
398- // Batched at the budget as it stands: when the server refuses a batch the budget shrinks, and
399- // what is left is batched again at the new one. A lone object is retried as is (it can't be
400- // split further), until the budget can't shrink any more.
401- for (let left = blobs, n = 1; left.length; n++) {
432+ await this.pushBlobs(blobs, label, false);
433+ }
434+
435+ /**
436+ * Blobs pushed in batches, each a flat tree under a temporary ref. A tree, not a commit holding
437+ * it: git leaves out of a later push what a ref on the server reaches, but the files of an
438+ * unrelated commit it would send again. Batched at the budget as it stands: when the server
439+ * refuses a batch the budget shrinks, and what is left is batched again at the new one. A lone
440+ * object is retried as is (it can't be split further), until the budget can't shrink any more.
441+ * `whole`: without deltas.
442+ */
443+ private async pushBlobs(blobs: Array<{ oid: string; size: number }>, label: string, whole: boolean): Promise<void> {
444+ for (let left = blobs; left.length; ) {
402445 const batch = batchBySize(left, this.budget)[0]!;
403446 const tree = (await this.git.run(["mktree"], { input: batch.map((b, j) => `100644 blob ${b.oid}\tobject-${j}\n`).join("") })).stdout.trim();
404- const carrier = (await this.git.run(["commit-tree", tree, "-m", `Diffinery import: objects for ${commit.slice(0, 12)} (${n})`])).stdout.trim();
405447 const size = batch.reduce((s, b) => s + b.size, 0);
406- if (await this.pushTemp(carrier, size, label)) left = left.slice(batch.length);
448+ if (await this.pushTemp(tree, size, label, whole)) left = left.slice(batch.length);
407449 }
408450 }
409451
410452 /** Push one chunk to a temporary ref. Returns false when the budget was lowered and the caller should re-plan. */
411- private async pushTemp(oid: string, size: number, label: string): Promise<boolean> {
453+ private async pushTemp(oid: string, size: number, label: string, whole = false): Promise<boolean> {
412454 if (this.haveSet.has(oid)) return true;
413455 const ref = `${TEMP_PREFIX}c${++this.temp}`;
414456 const n = this.result.chunks + 1;
415457 await this.o.progress({ phase: "push", message: `Pushing history of ${label.replace(/^refs\/(heads|tags)\//, "")}: chunk ${n} (${fmtBytes(size)})`, chunks: this.result.chunks, bytes: this.result.bytes });
416- const out = await this.push([`${oid}:${ref}`]);
458+ const out = await this.push([`${oid}:${ref}`], whole);
417459 if (out.code === 0 && !parsePorcelain(out.stdout).some((s) => !s.ok)) {
418460 this.tempRefs.push(ref);
419461 this.addHave(oid);
@@ -455,8 +497,10 @@
455497 }
456498
457499 /** `git push` to Diffinery in import mode; collects push protection findings from the output. */
458- async push(refspecs: string[]): Promise<{ code: number; stdout: string; stderr: string }> {
459- const out = await this.git.run(["push", "--porcelain", "--no-verify", ...this.pushOpts, this.spec.target.pushUrl, ...refspecs], { allowFail: true });
500+ async push(refspecs: string[], whole = false): Promise<{ code: number; stdout: string; stderr: string }> {
501+ // `whole`: no deltas at all, neither new ones (pack.window=0) nor those the mirror already
502+ // stores (pack.depth=0), so none against what the server has either.
503+ const out = await this.git.run([...(whole ? ["-c", "pack.window=0", "-c", "pack.depth=0"] : []), "push", "--porcelain", "--no-verify", ...this.pushOpts, this.spec.target.pushUrl, ...refspecs], { allowFail: true });
460504 if (/Push protection bypassed/i.test(out.stderr)) for (const f of parseSecretFindings(out.stderr)) this.findings.set(`${f.kind}:${f.blob}`, f);
461505 return out;
462506 }
modifiedcontainers/runner/src/main.ts+2−1View file
@@ -105,7 +105,8 @@
105105 const token = env.DIFFINERY_IMPORT_TOKEN;
106106 if (!api || !token) fail("DIFFINERY_API_URL and DIFFINERY_IMPORT_TOKEN are required in import mode");
107107 const chunk = Number(env.DIFFINERY_IMPORT_CHUNK_BYTES);
108- const result = await runImportAgent({ apiUrl: api, importId: env.DIFFINERY_IMPORT_ID, token, workRoot, chunkBytes: Number.isFinite(chunk) && chunk > 0 ? chunk : undefined });
108+ const large = Number(env.DIFFINERY_IMPORT_LARGE_BLOB_BYTES);
109+ const result = await runImportAgent({ apiUrl: api, importId: env.DIFFINERY_IMPORT_ID, token, workRoot, chunkBytes: Number.isFinite(chunk) && chunk > 0 ? chunk : undefined, largeBlobBytes: Number.isFinite(large) && large > 0 ? large : undefined });
109110 process.stdout.write(`[diffinery-import] ${result.ok ? "finished" : `failed: ${result.error}`}\n`);
110111 process.exit(result.ok ? 0 : 1);
111112 }
modifiedcontainers/runner/test/import.test.ts+55−0View file
@@ -149,6 +149,58 @@
149149 expect(log.every((l) => l.endsWith("diffinery-import"))).toBe(true);
150150 expect(log.some((l) => Number(l.split(" ")[0]) > 153600)).toBe(true);
151151 }, 30_000);
152+ it("pushes every version of a large file first and whole, so the server never rebuilds one from a delta", async () => {
153+ const root = mkdtempSync(path.join(tmpdir(), "diffinery-import-"));
154+ const work = path.join(root, "work");
155+ g(root, "init", "-q", "-b", "main", work);
156+ // A large text file in six versions (git stores them as deltas of each other), and small random files.
157+ let lines = Array.from({ length: 6000 }, (_, i) => `line ${i} ${randomBytes(12).toString("hex")}`);
158+ for (let v = 0; v < 6; v++) {
159+ lines = lines.map((l, i) => (i % 500 === v ? `${l} v${v}` : l));
160+ writeFileSync(path.join(work, "data.txt"), lines.join("\n"));
161+ writeFileSync(path.join(work, `small${v}.bin`), randomBytes(2048));
162+ g(work, "add", "-A");
163+ g(work, "commit", "-q", "-m", `v${v}`);
164+ }
165+ const source = path.join(root, "source.git");
166+ g(root, "clone", "-q", "--bare", work, source);
167+ g(source, "repack", "-adq");
168+ const target = path.join(root, "target.git");
169+ g(root, "init", "-q", "--bare", target);
170+ g(target, "config", "receive.advertisePushOptions", "true");
171+ // Pushes are kept as packs (not unpacked to loose objects), so the hook can look at them.
172+ g(target, "config", "receive.unpackLimit", "1");
173+ // The "server" refuses a push that holds a file as a delta, as Diffinery does past its memory.
174+ const hook = path.join(target, "hooks", "pre-receive");
175+ writeFileSync(
176+ hook,
177+ `#!/bin/sh
178+for idx in "$GIT_QUARANTINE_PATH"/pack/*.idx; do
179+ [ -f "$idx" ] || continue
180+ if git verify-pack -v "$idx" | awk '$2 == "blob" && NF >= 7 { found = 1 } END { exit !found }'; then
181+ echo "pack needs more than 80 MiB of memory to unpack; split it into smaller pushes"; exit 1
182+ fi
183+done
184+`,
185+ );
186+ chmodSync(hook, 0o755);
187+ const spec: ImportSpec = {
188+ importId: "t3",
189+ source: { fullName: "octo/data", cloneUrl: `file://${source}`, defaultBranch: "main" },
190+ githubToken: null,
191+ target: { fullName: "me/data", pushUrl: `file://${target}` },
192+ pushToken: "dfy_test",
193+ lfs: false,
194+ secrets: "bypass",
195+ bypassReason: "Imported from GitHub octo/data",
196+ chunkBytes: 1024 * 1024,
197+ };
198+ const result = await transfer(spec, { workRoot: path.join(root, "agent"), largeBlobBytes: 64 * 1024, progress: async () => {}, log: () => {} });
199+ expect(result.ok).toBe(true);
200+ expect(g(target, "rev-parse", "refs/heads/main")).toBe(g(work, "rev-parse", "HEAD"));
201+ g(target, "fsck", "--no-progress");
202+ });
203+
152204 it("splits a refused batch of a large commit's files again, rather than retrying it whole", async () => {
153205 const root = mkdtempSync(path.join(tmpdir(), "diffinery-import-"));
154206 const work = path.join(root, "work");
@@ -162,6 +214,9 @@
162214 const target = path.join(root, "target.git");
163215 g(root, "init", "-q", "--bare", target);
164216 g(target, "config", "receive.advertisePushOptions", "true");
217+ // Every push is kept as the pack that came (git otherwise stores only the objects it lacks), so
218+ // the commit itself must come without the files its batches already brought.
219+ g(target, "config", "receive.unpackLimit", "1");
165220 const hook = path.join(target, "hooks", "pre-receive");
166221 writeFileSync(
167222 hook,