c06069d · 81e4865
@@ -11,9 +11,12 @@
 * the parts back in order, joins them, and unpacks to exactly the bytes
 * that went in, or refuses.
 *
 * Plain data in, plain data out: no store, no files.
 * Plain data in, plain data out: `pack` and `unpack` touch no store and no files; `unpacker` writes the one file it is given.
 */
import { createWriteStream, renameSync, rmSync } from "node:fs";
import { createZstdDecompress } from "node:zlib";
/** A part's size: one put carries it within its own deadline at the floor rate (27 s at 5 Mbit/s). */
export const PART_BYTES = 16 * 1_048_576;
@@ -92,6 +95,110 @@
export async function unpack(
  return body;
}
export interface Unpacker {
  /** Hand over part `n`; resolves once part `n` itself is written, which waits for every part before it. */
  feed(n: number, body: Uint8Array): Promise<void>;
  /** Close the file, check its size, and put it at `target`; the bytes it unpacked to. */
  finish(): Promise<number>;
  /** Drop what was written, after a failure elsewhere. */
  abort(): void;
}
/**
 * A packed file unpacking as its parts arrive: each part, once every part
 * before it has come, goes into a zstd stream that writes the file out. No
 * whole compressed copy is held or written, only the parts waiting on an
 * earlier one; the file is written beside its target and renamed into place
 * when it checks out, so a refusal leaves nothing at `target`.
 */
export function unpacker(p: PackedFile, target: string): Unpacker {
  const partial = `${target}.partial`;
  const waiting = new Map<number, Uint8Array>();
  const waiters = new Map<number, { ok: () => void; no: (e: unknown) => void }>();
  const dec = createZstdDecompress();
  const out = createWriteStream(partial);
  let failure: unknown;
  // The first failure stands; every part still waiting for its turn hears of it.
  const fail = (e: unknown): void => {
    failure ??= e;
    for (const w of waiters.values()) w.no(failure);
    waiters.clear();
    waiting.clear();
    drop();
  };
  // Nothing of a refused file stays: the stream may still be opening, so the
  // file goes again once it has closed.
  const drop = (): void => {
    dec.destroy();
    out.destroy();
    out.once("close", () => rmSync(partial, { force: true }));
    rmSync(partial, { force: true });
  };
  dec.on("error", fail);
  out.on("error", fail);
  dec.pipe(out);
      
        junipercrew
        25d ago
        outdated
      
      Resolves when the chain drains, not when part n is written, if an earlier part is still missing. Make it wait for part n, or bound waiting.
    
  let next = 0;
  let packedSoFar = 0;
  let chain: Promise<void> = Promise.resolve();
  const write = (body: Uint8Array): Promise<void> =>
    new Promise((resolve, reject) => {
      if (failure !== undefined) return reject(failure);
      dec.write(body, (e) => (e ? reject(e) : resolve()));
    });
  const drain = async (): Promise<void> => {
    for (let b = waiting.get(next); b !== undefined; b = waiting.get(next)) {
      const n = next++;
      waiting.delete(n);
      packedSoFar += b.byteLength;
      await write(b);
      waiters.get(n)?.ok();
      waiters.delete(n);
    }
  };
  const abort = (): void => {
    fail(new Error(`${p.path} was abandoned`));
  };
  return {
    feed(n, body) {
      if (!Number.isSafeInteger(n) || n < 0 || n >= p.parts)
        return Promise.reject(new Error(`${p.path} has no part ${n}`));
      if (n < next || waiting.has(n))
        return Promise.reject(new Error(`${p.path} part ${n} came twice`));
      if (failure !== undefined) return Promise.reject(failure);
      return new Promise<void>((ok, no) => {
        waiters.set(n, { ok, no });
        waiting.set(n, body);
        // The chain never rejects: a failure is `failure`, told to every waiter.
        chain = chain.then(drain).catch(fail);
      });
    },
    async finish() {
      try {
        await chain;
        if (failure !== undefined) throw failure;
        if (next !== p.parts)
          throw new Error(`${p.path} came back in ${next} parts, not ${p.parts}`);
        if (packedSoFar !== p.packedBytes)
          throw new Error(`${p.path}'s parts are ${packedSoFar} bytes, not ${p.packedBytes}`);
        await new Promise<void>((resolve, reject) => {
          out.on("finish", resolve);
          out.on("error", reject);
          dec.on("error", reject);
          dec.end();
        });
        if (out.bytesWritten !== p.bytes)
          throw new Error(`${p.path} unpacks to ${out.bytesWritten} bytes, not ${p.bytes}`);
        renameSync(partial, target);
        return out.bytesWritten;
      } catch (error) {
        abort();
        throw error;
      }
    },
    abort,
  };
}
/** Whether a manifest's `packed` entry reads whole: the restore trusts nothing else. */
export function isPackedFile(v: unknown): v is PackedFile {
  if (typeof v !== "object" || v === null) return false;
@@ -20,8 +20,9 @@
 * `FORGE_SNAPSHOT_DIR` for a directory store.
 *
 * A file that went up compressed in parts (the database, `parts.ts`)
 * comes back whole: its parts read in order, joined, unpacked, and checked
 * against the size the manifest gives it.
 * comes back whole: its parts decompressed as they arrive, in order, into
 * the file, and checked against the size the manifest gives it. No whole
 * compressed copy is held or written.
 *
 * The key comes back only from its wrapped form and the operator's
 * passphrase (`FORGE_SNAPSHOT_PASSPHRASE`). A wrong passphrase refuses the
@@ -46,7 +47,7 @@
import {
  type Manifest,
} from "./snapshot.ts";
import { unwrapKey, WRAPPED_KEY } from "./wrap.ts";
import { isPackedFile, partPaths, unpack, type PackedFile } from "./parts.ts";
import { isPackedFile, partPaths, unpacker, type PackedFile, type Unpacker } from "./parts.ts";
import { resolveDestination } from "./destination.ts";
import type { Store } from "./store.ts";
@@ -121,6 +122,7 @@
async function fetchAll(
  each: (entry: { readonly path: string; readonly bytes: number }, body: Uint8Array, at: number) => Promise<void>,
  name: string,
  maxBytes: number,
  onFailure: () => void,
): Promise<void> {
  const pending = new Set<Promise<void>>();
  let failure: unknown;
@@ -146,6 +148,8 @@
async function fetchAll(
    })()
      .catch((error: unknown) => {
        failure ??= error;
        // Parts held for an earlier one would wait for ever.
        onFailure();
      })
      .finally(() => {
        held -= entry.bytes;
@@ -204,12 +208,42 @@
export async function restoreSnapshot(
  const packed: readonly unknown[] = manifest.packed ?? [];
  if (!packed.every(isPackedFile))
    throw new Error(`snapshot ${name}: a packed file in its manifest does not read whole`);
  const partOf = new Map<string, { file: PackedFile; n: number }>();
  // A part the manifest does not list is refused before anything is written.
  const listed = new Set(manifest.files.map((f) => f.path));
  for (const file of packed)
    for (const [n, path] of partPaths(file).entries())
      partOf.set(path, { file, n });
  const gathered = new Map<string, Uint8Array[]>();
    for (const path of partPaths(file))
      if (!listed.has(path)) throw new Error(`snapshot ${name} is missing ${path}`);
  // Each packed file unpacks as its parts arrive, in order, to disk.
  mkdirSync(dataDir, { recursive: true });
  const unpackers: { file: PackedFile; unpacker: Unpacker }[] = [];
  const partOf = new Map<string, { file: PackedFile; unpacker: Unpacker; n: number }>();
  for (const file of packed) {
    const target = join(dataDir, file.path);
    mkdirSync(dirname(target), { recursive: true });
    const made = unpacker(file, target);
    unpackers.push({ file, unpacker: made });
    for (const [n, path] of partPaths(file).entries())
      partOf.set(path, { file, unpacker: made, n });
  }
      
        junipercrew
        25d ago
        outdated
      
      The try body is not indented; run the formatter.
    
  const abandon = (): void => {
    for (const u of unpackers) u.unpacker.abort();
  };
  // A part waits in memory for the parts before it, so the parts are read
  // in part order: a manifest that lists them otherwise cannot stall the reads.
  const order = [...manifest.files];
  const slots = order.flatMap((f, i) => (partOf.has(f.path) ? [i] : []));
  const inPartOrder = slots
    .map((i) => order[i])
    .sort((x, y) => {
      const px = partOf.get(x?.path ?? "");
      const py = partOf.get(y?.path ?? "");
      if (px === undefined || py === undefined) return 0;
      return px.file.path === py.file.path ? px.n - py.n : px.file.path < py.file.path ? -1 : 1;
    });
  for (const [k, i] of slots.entries()) {
    const entry = inPartOrder[k];
    if (entry !== undefined) order[i] = entry;
  }
  let files = 0;
  let repos = 0;
@@ -217,65 +251,63 @@
export async function restoreSnapshot(
  let fetched = 0;
  let fetchedBytes = 0;
  say("files", 0, manifest.files.length, 0);
  await fetchAll(store, prefix, manifest.files, async (entry, body, at) => {
    fetched++;
    fetchedBytes += entry.bytes;
    say("files", fetched, manifest.files.length, fetchedBytes);
    if (entry.path === WRAPPED_KEY) return;
    // A part waits for the rest of its file.
    const part = partOf.get(entry.path);
    if (part !== undefined) {
      const got = gathered.get(part.file.path) ?? [];
      got[part.n] = body;
      gathered.set(part.file.path, got);
      return;
    }
    // A day's run logs in one object come back as the logs they were.
    if (isRunBundle(entry.path)) {
      for (const run of unpackRuns(body)) {
        const out = join(dataDir, "runs", run.path);
        mkdirSync(dirname(out), { recursive: true });
        await Bun.write(out, run.body);
        files++;
  try {
    await fetchAll(store, prefix, order, async (entry, body, at) => {
      fetched++;
      fetchedBytes += entry.bytes;
      say("files", fetched, manifest.files.length, fetchedBytes);
      if (entry.path === WRAPPED_KEY) return;
      // A part goes into its file's decompression as soon as its turn comes.
      const part = partOf.get(entry.path);
      if (part !== undefined) {
        await part.unpacker.feed(part.n, body);
        return;
      }
      return;
    }
    const bundle = entry.path.match(/^repos\/([^/]+)\/([^/]+)\.bundle$/);
    // A bundle is not a file of the forge's; it becomes a repository below.
    const target = bundle
      ? join(dataDir, "restore-bundles", `${bundle[1]}--${bundle[2]}.bundle`)
      : join(dataDir, entry.path);
    mkdirSync(dirname(target), { recursive: true });
    await Bun.write(target, body);
    // A key the rest of the machine could read is one the forge refuses to
    // use; the snapshot carries bytes, not modes, so the modes are set here.
    if (entry.path === "secret.key" || entry.path.startsWith("ssh/"))
      chmodSync(target, 0o600);
    files++;
    if (bundle)
      bundles.push({ at, owner: bundle[1]!, name: bundle[2]!, file: target });
  }, name, options.inFlightBytes ?? IN_FLIGHT_BYTES);
      // A day's run logs in one object come back as the logs they were.
      if (isRunBundle(entry.path)) {
        for (const run of unpackRuns(body)) {
          const out = join(dataDir, "runs", run.path);
          mkdirSync(dirname(out), { recursive: true });
          await Bun.write(out, run.body);
          files++;
        }
        return;
      
        junipercrew
        25d ago
        outdated
      
      Build a Set of manifest paths once instead of scanning manifest.files for every part.
    
      }
      const bundle = entry.path.match(/^repos\/([^/]+)\/([^/]+)\.bundle$/);
      // A bundle is not a file of the forge's; it becomes a repository below.
      const target = bundle
        ? join(dataDir, "restore-bundles", `${bundle[1]}--${bundle[2]}.bundle`)
        : join(dataDir, entry.path);
      mkdirSync(dirname(target), { recursive: true });
      await Bun.write(target, body);
      // A key the rest of the machine could read is one the forge refuses to
      // use; the snapshot carries bytes, not modes, so the modes are set here.
      if (entry.path === "secret.key" || entry.path.startsWith("ssh/"))
        chmodSync(target, 0o600);
      files++;
      if (bundle)
        bundles.push({ at, owner: bundle[1]!, name: bundle[2]!, file: target });
    }, name, options.inFlightBytes ?? IN_FLIGHT_BYTES, abandon);
  } catch (error) {
    abandon();
    throw error;
  }
  // Reads land in any order; repositories come back in the manifest's.
  bundles.sort((a, b) => a.at - b.at);
  let unpacked = 0;
  let unpackedBytes = 0;
  say("unpack", 0, packed.length, 0);
  for (const file of packed) {
    const got = gathered.get(file.path) ?? [];
    const parts = partPaths(file).map((path, n) => {
      const body = got[n];
      if (body === undefined)
        throw new Error(`snapshot ${name} is missing ${path}`);
      return body;
    });
    gathered.delete(file.path);
    const target = join(dataDir, file.path);
    mkdirSync(dirname(target), { recursive: true });
    const whole = await unpack(file, parts);
    await Bun.write(target, whole);
  for (const { unpacker: u } of unpackers) {
    let whole: number;
    try {
      whole = await u.finish();
    } catch (error) {
      abandon();
      throw error;
    }
    files++;
    unpacked++;
    unpackedBytes += whole.byteLength;
    unpackedBytes += whole;
    say("unpack", unpacked, packed.length, unpackedBytes);
  }
  // Each repository is its own clone, so several at once: a forge of
@@ -0,0 +1,178 @@
/**
 * The restore drill streams the snapshot into decompression (#2743): a
 * packed file's parts are unpacked as they arrive, in order, so no whole
 * compressed copy is held or written, the restored bytes match, and the
 * last part's download is the only wait left.
 */
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { existsSync, mkdtempSync, readdirSync, statSync } from "node:fs";
import { rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { pack, partPath, unpacker, type PackedFile } from "../src/snapshot/parts.ts";
import { restoreSnapshot } from "../src/snapshot/restore.ts";
import { dirStore, type Store } from "../src/snapshot/store.ts";
const NAME = "2026-10-19T08-00-00Z";
const prefix = `snapshots/${NAME}/`;
/** Bytes under `dir`, every file counted. */
function diskBytes(dir: string): number {
  let total = 0;
  const walk = (d: string): void => {
    for (const name of readdirSync(d)) {
      const path = join(d, name);
      const s = statSync(path);
      if (s.isDirectory()) walk(path);
      else total += s.size;
    }
  };
  try {
    walk(dir);
  } catch {}
  return total;
}
describe("a packed file unpacks as its parts arrive", () => {
  let root: string;
  beforeAll(() => {
    root = mkdtempSync(join(tmpdir(), "forge-restore-streamed-"));
  });
  afterAll(async () => {
    await rm(root, { recursive: true, force: true });
  });
  test("the file is already growing while the last part is still downloading; peak disk stays under 1.2x; bytes match", async () => {
    const db = crypto.getRandomValues(new Uint8Array(1_500_000));
    const { packed, parts } = await pack("forge.db", db, 64 * 1024);
    expect(parts.length).toBeGreaterThan(10);
    const inner = dirStore(join(root, "store"));
    for (const [i, part] of parts.entries()) await inner.put(prefix + partPath("forge.db", i), part);
    const files = parts.map((part, i) => ({ path: partPath("forge.db", i), bytes: part.byteLength }));
    await inner.put(
      `${prefix}manifest.json`,
      JSON.stringify({
        name: NAME,
        startedAt: "2026-10-19T08:00:00.000Z",
        finishedAt: "2026-10-19T08:01:00.000Z",
        version: "t",
        repos: 0,
        apps: 0,
        files,
        bytes: files.reduce((n, f) => n + f.bytes, 0),
        key: "none",
        packed: [packed],
      }),
    );
    const dataDir = join(root, "data");
    const last = prefix + partPath("forge.db", parts.length - 1);
    let release!: () => void;
    const gate = new Promise<void>((r) => (release = r));
    let peak = 0;
    let grewBeforeLast = 0;
    const store: Store = {
      ...inner,
      get: async (key) => {
        peak = Math.max(peak, diskBytes(dataDir));
        if (key !== last) return inner.get(key);
        // Everything else has landed; wait for the file to show it was unpacked on arrival.
        for (let i = 0; i < 100 && diskBytes(dataDir) === 0; i++) await Bun.sleep(20);
        grewBeforeLast = diskBytes(dataDir);
        await gate;
        return inner.get(key);
      },
    };
    const done = restoreSnapshot(store, NAME, dataDir);
    await Bun.sleep(400);
    release();
    await done;
    peak = Math.max(peak, diskBytes(dataDir));
    expect(grewBeforeLast).toBeGreaterThan(0);
    expect(peak).toBeLessThanOrEqual(db.byteLength * 1.2);
    expect(new Uint8Array(await Bun.file(join(dataDir, "forge.db")).arrayBuffer())).toEqual(db);
    expect(readdirSync(dataDir)).toEqual(["forge.db"]);
  });
});
describe("unpacker refuses what does not add up, and leaves nothing behind", () => {
  let root: string;
  let n = 0;
  beforeAll(() => {
    root = mkdtempSync(join(tmpdir(), "forge-unpacker-"));
  });
  afterAll(async () => {
    await rm(root, { recursive: true, force: true });
  });
  /** Feed `parts` in order to an unpacker of `file`, finish it, and say what it refused with. */
  async function refusal(file: PackedFile, parts: Uint8Array[]): Promise<{ error: string; target: string }> {
    const target = join(root, `out-${n++}`);
    const u = unpacker(file, target);
    let error = "";
    try {
      for (const [i, part] of parts.entries()) await u.feed(i, part);
      await u.finish();
    } catch (e) {
      error = e instanceof Error ? e.message : String(e);
    }
    await Bun.sleep(30);
    return { error, target };
  }
  const gone = (target: string): void => {
    expect(existsSync(target)).toBe(false);
    expect(existsSync(`${target}.partial`)).toBe(false);
  };
  const db = crypto.getRandomValues(new Uint8Array(200_000));
  test("a short set of parts", async () => {
    const { packed, parts } = await pack("forge.db", db, 16 * 1024);
    const r = await refusal(packed, parts.slice(0, -1));
    expect(r.error).toContain(`came back in ${parts.length - 1} parts, not ${parts.length}`);
    gone(r.target);
  });
  test("a wrong packed size", async () => {
    const { packed, parts } = await pack("forge.db", db, 16 * 1024);
    const r = await refusal({ ...packed, packedBytes: packed.packedBytes + 1 }, parts);
    expect(r.error).toContain("bytes, not");
    gone(r.target);
  });
  test("a wrong unpacked size", async () => {
    const { packed, parts } = await pack("forge.db", db, 16 * 1024);
    const r = await refusal({ ...packed, bytes: packed.bytes + 1 }, parts);
    expect(r.error).toContain(`unpacks to ${db.byteLength} bytes, not ${db.byteLength + 1}`);
    gone(r.target);
  });
  test("a corrupt part", async () => {
    const { packed, parts } = await pack("forge.db", db, 16 * 1024);
    const bad = parts.map((p) => new Uint8Array(p));
    bad[0] = new Uint8Array(bad[0]!.byteLength).fill(0xab);
    const r = await refusal(packed, bad);
    expect(r.error).not.toBe("");
    gone(r.target);
  });
  test("a part fed early resolves only once it is written, and a failure frees it", async () => {
    const { packed, parts } = await pack("forge.db", db, 16 * 1024);
    const target = join(root, `out-${n++}`);
    const u = unpacker(packed, target);
    let settled = false;
    const early = u.feed(2, parts[2]!).then(
      () => (settled = true),
      () => (settled = true),
    );
    await Bun.sleep(50);
    expect(settled).toBe(false);
    await u.feed(0, parts[0]!);
    await Bun.sleep(50);
    expect(settled).toBe(false);
    u.abort();
    await early;
    await Bun.sleep(30);
    expect(settled).toBe(true);
    gone(target);
  });
});
		@@ -11,9 +11,12 @@
11	11	* the parts back in order, joins them, and unpacks to exactly the bytes
12	12	* that went in, or refuses.
13	13	*
14		* Plain data in, plain data out: no store, no files.
	14	* Plain data in, plain data out: `pack` and `unpack` touch no store and no files; `unpacker` writes the one file it is given.
15	15	*/
16	16
	17	import { createWriteStream, renameSync, rmSync } from "node:fs";
	18	import { createZstdDecompress } from "node:zlib";
	19
17	20	/** A part's size: one put carries it within its own deadline at the floor rate (27 s at 5 Mbit/s). */
18	21	export const PART_BYTES = 16 * 1_048_576;
19	22
		@@ -92,6 +95,110 @@ export async function unpack(
92	95	return body;
93	96	}
94	97
	98	export interface Unpacker {
	99	/** Hand over part `n`; resolves once part `n` itself is written, which waits for every part before it. */
	100	feed(n: number, body: Uint8Array): Promise<void>;
	101	/** Close the file, check its size, and put it at `target`; the bytes it unpacked to. */
	102	finish(): Promise<number>;
	103	/** Drop what was written, after a failure elsewhere. */
	104	abort(): void;
	105	}
	106
	107	/**
	108	* A packed file unpacking as its parts arrive: each part, once every part
	109	* before it has come, goes into a zstd stream that writes the file out. No
	110	* whole compressed copy is held or written, only the parts waiting on an
	111	* earlier one; the file is written beside its target and renamed into place
	112	* when it checks out, so a refusal leaves nothing at `target`.
	113	*/
	114	export function unpacker(p: PackedFile, target: string): Unpacker {
	115	const partial = `${target}.partial`;
	116	const waiting = new Map<number, Uint8Array>();
	117	const waiters = new Map<number, { ok: () => void; no: (e: unknown) => void }>();
	118	const dec = createZstdDecompress();
	119	const out = createWriteStream(partial);
	120	let failure: unknown;
	121	// The first failure stands; every part still waiting for its turn hears of it.
	122	const fail = (e: unknown): void => {
	123	failure ??= e;
	124	for (const w of waiters.values()) w.no(failure);
	125	waiters.clear();
	126	waiting.clear();
	127	drop();
	128	};
	129	// Nothing of a refused file stays: the stream may still be opening, so the
	130	// file goes again once it has closed.
	131	const drop = (): void => {
	132	dec.destroy();
	133	out.destroy();
	134	out.once("close", () => rmSync(partial, { force: true }));
	135	rmSync(partial, { force: true });
	136	};
	137	dec.on("error", fail);
	138	out.on("error", fail);
	139	dec.pipe(out);
		junipercrew 25d ago outdated Resolves when the chain drains, not when part n is written, if an earlier part is still missing. Make it wait for part n, or bound `waiting`.
	140	let next = 0;
	141	let packedSoFar = 0;
	142	let chain: Promise<void> = Promise.resolve();
	143	const write = (body: Uint8Array): Promise<void> =>
	144	new Promise((resolve, reject) => {
	145	if (failure !== undefined) return reject(failure);
	146	dec.write(body, (e) => (e ? reject(e) : resolve()));
	147	});
	148	const drain = async (): Promise<void> => {
	149	for (let b = waiting.get(next); b !== undefined; b = waiting.get(next)) {
	150	const n = next++;
	151	waiting.delete(n);
	152	packedSoFar += b.byteLength;
	153	await write(b);
	154	waiters.get(n)?.ok();
	155	waiters.delete(n);
	156	}
	157	};
	158	const abort = (): void => {
	159	fail(new Error(`${p.path} was abandoned`));
	160	};
	161	return {
	162	feed(n, body) {
	163	if (!Number.isSafeInteger(n) \|\| n < 0 \|\| n >= p.parts)
	164	return Promise.reject(new Error(`${p.path} has no part ${n}`));
	165	if (n < next \|\| waiting.has(n))
	166	return Promise.reject(new Error(`${p.path} part ${n} came twice`));
	167	if (failure !== undefined) return Promise.reject(failure);
	168	return new Promise<void>((ok, no) => {
	169	waiters.set(n, { ok, no });
	170	waiting.set(n, body);
	171	// The chain never rejects: a failure is `failure`, told to every waiter.
	172	chain = chain.then(drain).catch(fail);
	173	});
	174	},
	175	async finish() {
	176	try {
	177	await chain;
	178	if (failure !== undefined) throw failure;
	179	if (next !== p.parts)
	180	throw new Error(`${p.path} came back in ${next} parts, not ${p.parts}`);
	181	if (packedSoFar !== p.packedBytes)
	182	throw new Error(`${p.path}'s parts are ${packedSoFar} bytes, not ${p.packedBytes}`);
	183	await new Promise<void>((resolve, reject) => {
	184	out.on("finish", resolve);
	185	out.on("error", reject);
	186	dec.on("error", reject);
	187	dec.end();
	188	});
	189	if (out.bytesWritten !== p.bytes)
	190	throw new Error(`${p.path} unpacks to ${out.bytesWritten} bytes, not ${p.bytes}`);
	191	renameSync(partial, target);
	192	return out.bytesWritten;
	193	} catch (error) {
	194	abort();
	195	throw error;
	196	}
	197	},
	198	abort,
	199	};
	200	}
	201
95	202	/** Whether a manifest's `packed` entry reads whole: the restore trusts nothing else. */
96	203	export function isPackedFile(v: unknown): v is PackedFile {
97	204	if (typeof v !== "object" \|\| v === null) return false;
		@@ -20,8 +20,9 @@
20	20	* `FORGE_SNAPSHOT_DIR` for a directory store.
21	21	*
22	22	* A file that went up compressed in parts (the database, `parts.ts`)
23		* comes back whole: its parts read in order, joined, unpacked, and checked
24		* against the size the manifest gives it.
	23	* comes back whole: its parts decompressed as they arrive, in order, into
	24	* the file, and checked against the size the manifest gives it. No whole
	25	* compressed copy is held or written.
25	26	*
26	27	* The key comes back only from its wrapped form and the operator's
27	28	* passphrase (`FORGE_SNAPSHOT_PASSPHRASE`). A wrong passphrase refuses the
		@@ -46,7 +47,7 @@ import {
46	47	type Manifest,
47	48	} from "./snapshot.ts";
48	49	import { unwrapKey, WRAPPED_KEY } from "./wrap.ts";
49		import { isPackedFile, partPaths, unpack, type PackedFile } from "./parts.ts";
	50	import { isPackedFile, partPaths, unpacker, type PackedFile, type Unpacker } from "./parts.ts";
50	51	import { resolveDestination } from "./destination.ts";
51	52	import type { Store } from "./store.ts";
52	53
		@@ -121,6 +122,7 @@ async function fetchAll(
121	122	each: (entry: { readonly path: string; readonly bytes: number }, body: Uint8Array, at: number) => Promise<void>,
122	123	name: string,
123	124	maxBytes: number,
	125	onFailure: () => void,
124	126	): Promise<void> {
125	127	const pending = new Set<Promise<void>>();
126	128	let failure: unknown;
		@@ -146,6 +148,8 @@ async function fetchAll(
146	148	})()
147	149	.catch((error: unknown) => {
148	150	failure ??= error;
	151	// Parts held for an earlier one would wait for ever.
	152	onFailure();
149	153	})
150	154	.finally(() => {
151	155	held -= entry.bytes;
		@@ -204,12 +208,42 @@ export async function restoreSnapshot(
204	208	const packed: readonly unknown[] = manifest.packed ?? [];
205	209	if (!packed.every(isPackedFile))
206	210	throw new Error(`snapshot ${name}: a packed file in its manifest does not read whole`);
207		const partOf = new Map<string, { file: PackedFile; n: number }>();
	211	// A part the manifest does not list is refused before anything is written.
	212	const listed = new Set(manifest.files.map((f) => f.path));
208	213	for (const file of packed)
209		for (const [n, path] of partPaths(file).entries())
210		partOf.set(path, { file, n });
211		const gathered = new Map<string, Uint8Array[]>();
	214	for (const path of partPaths(file))
	215	if (!listed.has(path)) throw new Error(`snapshot ${name} is missing ${path}`);
	216	// Each packed file unpacks as its parts arrive, in order, to disk.
212	217	mkdirSync(dataDir, { recursive: true });
	218	const unpackers: { file: PackedFile; unpacker: Unpacker }[] = [];
	219	const partOf = new Map<string, { file: PackedFile; unpacker: Unpacker; n: number }>();
	220	for (const file of packed) {
	221	const target = join(dataDir, file.path);
	222	mkdirSync(dirname(target), { recursive: true });
	223	const made = unpacker(file, target);
	224	unpackers.push({ file, unpacker: made });
	225	for (const [n, path] of partPaths(file).entries())
	226	partOf.set(path, { file, unpacker: made, n });
	227	}
		junipercrew 25d ago outdated The try body is not indented; run the formatter.
	228	const abandon = (): void => {
	229	for (const u of unpackers) u.unpacker.abort();
	230	};
	231	// A part waits in memory for the parts before it, so the parts are read
	232	// in part order: a manifest that lists them otherwise cannot stall the reads.
	233	const order = [...manifest.files];
	234	const slots = order.flatMap((f, i) => (partOf.has(f.path) ? [i] : []));
	235	const inPartOrder = slots
	236	.map((i) => order[i])
	237	.sort((x, y) => {
	238	const px = partOf.get(x?.path ?? "");
	239	const py = partOf.get(y?.path ?? "");
	240	if (px === undefined \|\| py === undefined) return 0;
	241	return px.file.path === py.file.path ? px.n - py.n : px.file.path < py.file.path ? -1 : 1;
	242	});
	243	for (const [k, i] of slots.entries()) {
	244	const entry = inPartOrder[k];
	245	if (entry !== undefined) order[i] = entry;
	246	}
213	247
214	248	let files = 0;
215	249	let repos = 0;
		@@ -217,65 +251,63 @@ export async function restoreSnapshot(
217	251	let fetched = 0;
218	252	let fetchedBytes = 0;
219	253	say("files", 0, manifest.files.length, 0);
220		await fetchAll(store, prefix, manifest.files, async (entry, body, at) => {
221		fetched++;
222		fetchedBytes += entry.bytes;
223		say("files", fetched, manifest.files.length, fetchedBytes);
224		if (entry.path === WRAPPED_KEY) return;
225		// A part waits for the rest of its file.
226		const part = partOf.get(entry.path);
227		if (part !== undefined) {
228		const got = gathered.get(part.file.path) ?? [];
229		got[part.n] = body;
230		gathered.set(part.file.path, got);
231		return;
232		}
233		// A day's run logs in one object come back as the logs they were.
234		if (isRunBundle(entry.path)) {
235		for (const run of unpackRuns(body)) {
236		const out = join(dataDir, "runs", run.path);
237		mkdirSync(dirname(out), { recursive: true });
238		await Bun.write(out, run.body);
239		files++;
	254	try {
	255	await fetchAll(store, prefix, order, async (entry, body, at) => {
	256	fetched++;
	257	fetchedBytes += entry.bytes;
	258	say("files", fetched, manifest.files.length, fetchedBytes);
	259	if (entry.path === WRAPPED_KEY) return;
	260	// A part goes into its file's decompression as soon as its turn comes.
	261	const part = partOf.get(entry.path);
	262	if (part !== undefined) {
	263	await part.unpacker.feed(part.n, body);
	264	return;
240	265	}
241		return;
242		}
243		const bundle = entry.path.match(/^repos\/([^/]+)\/([^/]+)\.bundle$/);
244		// A bundle is not a file of the forge's; it becomes a repository below.
245		const target = bundle
246		? join(dataDir, "restore-bundles", `${bundle[1]}--${bundle[2]}.bundle`)
247		: join(dataDir, entry.path);
248		mkdirSync(dirname(target), { recursive: true });
249		await Bun.write(target, body);
250		// A key the rest of the machine could read is one the forge refuses to
251		// use; the snapshot carries bytes, not modes, so the modes are set here.
252		if (entry.path === "secret.key" \|\| entry.path.startsWith("ssh/"))
253		chmodSync(target, 0o600);
254		files++;
255		if (bundle)
256		bundles.push({ at, owner: bundle[1]!, name: bundle[2]!, file: target });
257		}, name, options.inFlightBytes ?? IN_FLIGHT_BYTES);
	266	// A day's run logs in one object come back as the logs they were.
	267	if (isRunBundle(entry.path)) {
	268	for (const run of unpackRuns(body)) {
	269	const out = join(dataDir, "runs", run.path);
	270	mkdirSync(dirname(out), { recursive: true });
	271	await Bun.write(out, run.body);
	272	files++;
	273	}
	274	return;
		junipercrew 25d ago outdated Build a Set of manifest paths once instead of scanning manifest.files for every part.
	275	}
	276	const bundle = entry.path.match(/^repos\/([^/]+)\/([^/]+)\.bundle$/);
	277	// A bundle is not a file of the forge's; it becomes a repository below.
	278	const target = bundle
	279	? join(dataDir, "restore-bundles", `${bundle[1]}--${bundle[2]}.bundle`)
	280	: join(dataDir, entry.path);
	281	mkdirSync(dirname(target), { recursive: true });
	282	await Bun.write(target, body);
	283	// A key the rest of the machine could read is one the forge refuses to
	284	// use; the snapshot carries bytes, not modes, so the modes are set here.
	285	if (entry.path === "secret.key" \|\| entry.path.startsWith("ssh/"))
	286	chmodSync(target, 0o600);
	287	files++;
	288	if (bundle)
	289	bundles.push({ at, owner: bundle[1]!, name: bundle[2]!, file: target });
	290	}, name, options.inFlightBytes ?? IN_FLIGHT_BYTES, abandon);
	291	} catch (error) {
	292	abandon();
	293	throw error;
	294	}
258	295	// Reads land in any order; repositories come back in the manifest's.
259	296	bundles.sort((a, b) => a.at - b.at);
260	297	let unpacked = 0;
261	298	let unpackedBytes = 0;
262	299	say("unpack", 0, packed.length, 0);
263		for (const file of packed) {
264		const got = gathered.get(file.path) ?? [];
265		const parts = partPaths(file).map((path, n) => {
266		const body = got[n];
267		if (body === undefined)
268		throw new Error(`snapshot ${name} is missing ${path}`);
269		return body;
270		});
271		gathered.delete(file.path);
272		const target = join(dataDir, file.path);
273		mkdirSync(dirname(target), { recursive: true });
274		const whole = await unpack(file, parts);
275		await Bun.write(target, whole);
	300	for (const { unpacker: u } of unpackers) {
	301	let whole: number;
	302	try {
	303	whole = await u.finish();
	304	} catch (error) {
	305	abandon();
	306	throw error;
	307	}
276	308	files++;
277	309	unpacked++;
278		unpackedBytes += whole.byteLength;
	310	unpackedBytes += whole;
279	311	say("unpack", unpacked, packed.length, unpackedBytes);
280	312	}
281	313	// Each repository is its own clone, so several at once: a forge of
		@@ -0,0 +1,178 @@
	1	/**
	2	* The restore drill streams the snapshot into decompression (#2743): a
	3	* packed file's parts are unpacked as they arrive, in order, so no whole
	4	* compressed copy is held or written, the restored bytes match, and the
	5	* last part's download is the only wait left.
	6	*/
	7
	8	import { afterAll, beforeAll, describe, expect, test } from "bun:test";
	9	import { existsSync, mkdtempSync, readdirSync, statSync } from "node:fs";
	10	import { rm } from "node:fs/promises";
	11	import { tmpdir } from "node:os";
	12	import { join } from "node:path";
	13	import { pack, partPath, unpacker, type PackedFile } from "../src/snapshot/parts.ts";
	14	import { restoreSnapshot } from "../src/snapshot/restore.ts";
	15	import { dirStore, type Store } from "../src/snapshot/store.ts";
	16
	17	const NAME = "2026-10-19T08-00-00Z";
	18	const prefix = `snapshots/${NAME}/`;
	19
	20	/** Bytes under `dir`, every file counted. */
	21	function diskBytes(dir: string): number {
	22	let total = 0;
	23	const walk = (d: string): void => {
	24	for (const name of readdirSync(d)) {
	25	const path = join(d, name);
	26	const s = statSync(path);
	27	if (s.isDirectory()) walk(path);
	28	else total += s.size;
	29	}
	30	};
	31	try {
	32	walk(dir);
	33	} catch {}
	34	return total;
	35	}
	36
	37	describe("a packed file unpacks as its parts arrive", () => {
	38	let root: string;
	39	beforeAll(() => {
	40	root = mkdtempSync(join(tmpdir(), "forge-restore-streamed-"));
	41	});
	42	afterAll(async () => {
	43	await rm(root, { recursive: true, force: true });
	44	});
	45
	46	test("the file is already growing while the last part is still downloading; peak disk stays under 1.2x; bytes match", async () => {
	47	const db = crypto.getRandomValues(new Uint8Array(1_500_000));
	48	const { packed, parts } = await pack("forge.db", db, 64 * 1024);
	49	expect(parts.length).toBeGreaterThan(10);
	50	const inner = dirStore(join(root, "store"));
	51	for (const [i, part] of parts.entries()) await inner.put(prefix + partPath("forge.db", i), part);
	52	const files = parts.map((part, i) => ({ path: partPath("forge.db", i), bytes: part.byteLength }));
	53	await inner.put(
	54	`${prefix}manifest.json`,
	55	JSON.stringify({
	56	name: NAME,
	57	startedAt: "2026-10-19T08:00:00.000Z",
	58	finishedAt: "2026-10-19T08:01:00.000Z",
	59	version: "t",
	60	repos: 0,
	61	apps: 0,
	62	files,
	63	bytes: files.reduce((n, f) => n + f.bytes, 0),
	64	key: "none",
	65	packed: [packed],
	66	}),
	67	);
	68	const dataDir = join(root, "data");
	69	const last = prefix + partPath("forge.db", parts.length - 1);
	70	let release!: () => void;
	71	const gate = new Promise<void>((r) => (release = r));
	72	let peak = 0;
	73	let grewBeforeLast = 0;
	74	const store: Store = {
	75	...inner,
	76	get: async (key) => {
	77	peak = Math.max(peak, diskBytes(dataDir));
	78	if (key !== last) return inner.get(key);
	79	// Everything else has landed; wait for the file to show it was unpacked on arrival.
	80	for (let i = 0; i < 100 && diskBytes(dataDir) === 0; i++) await Bun.sleep(20);
	81	grewBeforeLast = diskBytes(dataDir);
	82	await gate;
	83	return inner.get(key);
	84	},
	85	};
	86	const done = restoreSnapshot(store, NAME, dataDir);
	87	await Bun.sleep(400);
	88	release();
	89	await done;
	90	peak = Math.max(peak, diskBytes(dataDir));
	91	expect(grewBeforeLast).toBeGreaterThan(0);
	92	expect(peak).toBeLessThanOrEqual(db.byteLength * 1.2);
	93	expect(new Uint8Array(await Bun.file(join(dataDir, "forge.db")).arrayBuffer())).toEqual(db);
	94	expect(readdirSync(dataDir)).toEqual(["forge.db"]);
	95	});
	96	});
	97
	98	describe("unpacker refuses what does not add up, and leaves nothing behind", () => {
	99	let root: string;
	100	let n = 0;
	101	beforeAll(() => {
	102	root = mkdtempSync(join(tmpdir(), "forge-unpacker-"));
	103	});
	104	afterAll(async () => {
	105	await rm(root, { recursive: true, force: true });
	106	});
	107
	108	/** Feed `parts` in order to an unpacker of `file`, finish it, and say what it refused with. */
	109	async function refusal(file: PackedFile, parts: Uint8Array[]): Promise<{ error: string; target: string }> {
	110	const target = join(root, `out-${n++}`);
	111	const u = unpacker(file, target);
	112	let error = "";
	113	try {
	114	for (const [i, part] of parts.entries()) await u.feed(i, part);
	115	await u.finish();
	116	} catch (e) {
	117	error = e instanceof Error ? e.message : String(e);
	118	}
	119	await Bun.sleep(30);
	120	return { error, target };
	121	}
	122	const gone = (target: string): void => {
	123	expect(existsSync(target)).toBe(false);
	124	expect(existsSync(`${target}.partial`)).toBe(false);
	125	};
	126	const db = crypto.getRandomValues(new Uint8Array(200_000));
	127
	128	test("a short set of parts", async () => {
	129	const { packed, parts } = await pack("forge.db", db, 16 * 1024);
	130	const r = await refusal(packed, parts.slice(0, -1));
	131	expect(r.error).toContain(`came back in ${parts.length - 1} parts, not ${parts.length}`);
	132	gone(r.target);
	133	});
	134
	135	test("a wrong packed size", async () => {
	136	const { packed, parts } = await pack("forge.db", db, 16 * 1024);
	137	const r = await refusal({ ...packed, packedBytes: packed.packedBytes + 1 }, parts);
	138	expect(r.error).toContain("bytes, not");
	139	gone(r.target);
	140	});
	141
	142	test("a wrong unpacked size", async () => {
	143	const { packed, parts } = await pack("forge.db", db, 16 * 1024);
	144	const r = await refusal({ ...packed, bytes: packed.bytes + 1 }, parts);
	145	expect(r.error).toContain(`unpacks to ${db.byteLength} bytes, not ${db.byteLength + 1}`);
	146	gone(r.target);
	147	});
	148
	149	test("a corrupt part", async () => {
	150	const { packed, parts } = await pack("forge.db", db, 16 * 1024);
	151	const bad = parts.map((p) => new Uint8Array(p));
	152	bad[0] = new Uint8Array(bad[0]!.byteLength).fill(0xab);
	153	const r = await refusal(packed, bad);
	154	expect(r.error).not.toBe("");
	155	gone(r.target);
	156	});
	157
	158	test("a part fed early resolves only once it is written, and a failure frees it", async () => {
	159	const { packed, parts } = await pack("forge.db", db, 16 * 1024);
	160	const target = join(root, `out-${n++}`);
	161	const u = unpacker(packed, target);
	162	let settled = false;
	163	const early = u.feed(2, parts[2]!).then(
	164	() => (settled = true),
	165	() => (settled = true),
	166	);
	167	await Bun.sleep(50);
	168	expect(settled).toBe(false);
	169	await u.feed(0, parts[0]!);
	170	await Bun.sleep(50);
	171	expect(settled).toBe(false);
	172	u.abort();
	173	await early;
	174	await Bun.sleep(30);
	175	expect(settled).toBe(true);
	176	gone(target);
	177	});
	178	});