4ab6b5f · 3f51c6c · 08b0f8d
@@ -6,7 +6,7 @@
- **Catch-up.** A boot with the newest snapshot older than the last 08:00 UTC slot takes one two minutes later (`snapshotDue`). A snapshot cut short by a release has no manifest, so it counts as lost.
- **In the terminal.** The Resilience dialog (palette: Resilience, admins) opens with the same verdict, each lane's newest, and every reason. On the web it is home's backup tile and `/settings/backups`; for an agent, `verdict` in the MCP `backups` tool.
- **Off the serving thread.** `VACUUM INTO` runs in a worker (`vacuum.ts`).
- **Not done, tracked:** snapshots outside the serving process or resumable (#8828; the off-process probe is below). Catch-up retakes a lost snapshot whole.
- **Not done, tracked:** snapshots outside the serving process (#8828); resuming across a release is below, and so is the off-process probe.
## Platform backup lanes (#8811)
@@ -57,8 +57,21 @@
unreadable or empty; or the newest manifest's `finishedAt` older than the
limit (default 36 h: one nightly plus grace). Use read-only credentials.
Logic lives in `src/snapshot/probe.ts`; tests in `test/backup-probe.test.ts`.
### Not done yet
## Resuming across a release (#8828)
Snapshots are still taken inside the serving process; a release mid-snapshot
is caught up by retaking it whole after boot. Resuming across a release (or
moving the snapshot into its own process) is a separate, larger change, tracked in #8828.
A snapshot cut short has no manifest, but it is no longer retaken whole.
`takeSnapshot` writes `partial.json` under the snapshot's prefix (after the
first landed object, every 100 more, every 15 s while something new landed,
and on failure; a release kills the process without running any cleanup, so
the clock is what leaves it): the repository bundles and run bundles that
landed, with their sizes and SHA-256 digests. The catch-up after boot (`takeIfDue`) finds the
newest prefix newer than the newest manifest, begun within 24 h, that holds a
journal, and takes the snapshot again under that name and start time with
`resume`. A bundle or run bundle the journal lists with the same size and digest is
not put again; the database, key, ssh files, app data and recent run logs always are,
since they change while the forge runs. The journal is deleted once the
manifest is written. `pruneSnapshots` removes a prefix with no manifest that
is older than the newest whole snapshot or past the window.
Not done: the snapshot still runs in the serving process (the database copy is
in a worker). Tests: `test/snapshot-resume.test.ts`.
@@ -14,9 +14,13 @@
import type { App } from "../app.ts";
import { createLog } from "../kernel/log.ts";
import { firstSeen, reportVerdict, seedFirstSeen } from "./verdict.ts";
import {
  JOURNAL,
  KEEP,
  listSnapshots,
  pruneSnapshots,
  RESUME_WINDOW_MS,
  SNAPSHOT_PREFIX,
  snapshotStart,
  takeSnapshot,
  type Manifest,
  type Progress,
@@ -164,7 +168,7 @@
export function startSnapshots(
  let running: Progress | undefined;
  let last: { manifest?: Manifest; error?: string; at?: string } = {};
  const take = (): Promise<Manifest> => {
  const take = (resume?: Date): Promise<Manifest> => {
    // Whether this run's store is off the machine, for home's backup age (#3701).
    let offBox: boolean | undefined;
    // After the one before it, however that ended: chained on a failure as it
@@ -186,6 +190,7 @@
export function startSnapshots(
          reposRoot: app.reposRoot,
          version: options.version,
          passphrase: options.passphrase,
          resume: resume !== undefined,
          onLand: (p) => {
            running = p;
            if (Date.now() - toldAt < (options.progressEventMs ?? PROGRESS_EVENT_MS)) return;
@@ -194,6 +199,7 @@
export function startSnapshots(
          },
        },
        store,
        resume,
      ).finally(() => {
        running = undefined;
      });
@@ -232,6 +238,21 @@
export function startSnapshots(
    return inFlight;
  };
  // The newest snapshot a release cut short that is newer than the newest
  // whole one and not stale: its journal says what need not go up again.
  const cutShort = async (store: Store, newestWhole: string | undefined): Promise<Date | undefined> => {
    try {
      const names = (await store.names(SNAPSHOT_PREFIX)).sort().reverse();
      const name = names.find((n) => newestWhole === undefined || n > newestWhole);
      const at = name === undefined ? undefined : snapshotStart(name);
      if (name === undefined || at === undefined) return undefined;
      if ((options.now ?? Date.now)() - at.getTime() > RESUME_WINDOW_MS) return undefined;
      return (await store.get(`${SNAPSHOT_PREFIX}${name}/${JOURNAL}`)) === undefined ? undefined : at;
    } catch {
      return undefined;
    }
  };
  // A restart takes the night's snapshot if the store does not hold it: a
  // release at 08:21 killed the one in flight and nothing took it again (#8742).
  // The manifest is written last, so a snapshot cut short is not one.
@@ -247,8 +268,9 @@
export function startSnapshots(
      return false;
    }
    if (stopped || active > 0 || !snapshotDue(newest?.finishedAt, (options.now ?? Date.now)())) return false;
    log.info("snapshot due or lost; taking one", { newest: newest?.name ?? "none" });
    await take().catch(() => {});
    const cut = await cutShort(store, newest?.name);
    log.info("snapshot due or lost; taking one", { newest: newest?.name ?? "none", resuming: cut?.toISOString() ?? "no" });
    await take(cut).catch(() => {});
    return true;
  };
@@ -171,6 +171,59 @@
export interface Snapshotter {
  readonly progressEvery?: { objects?: number; ms?: number } | undefined;
  /** Override of the database's part size, for tests. */
  readonly partBytes?: number | undefined;
  /**
   * Carry on a snapshot a release cut short: `at` is its start, and the
   * repository bundles and run bundles its journal says landed are not put
   * again. Everything else (the database, the key, live files) is taken anew.
   */
  readonly resume?: boolean | undefined;
  /** Override of how often a changed journal is written, for tests. */
  readonly journalEveryMs?: number | undefined;
}
/** The journal of what landed of a snapshot still being taken; gone once its manifest is. */
export const JOURNAL = "partial.json";
/** The journal is rewritten after this many more objects have landed. */
export const JOURNAL_EVERY = 100;
/** And no later than this after something landed: a release kills the process between writes. */
export const JOURNAL_MS = 15_000;
/** A snapshot cut short is carried on, not retaken, if it began within this long. */
export const RESUME_WINDOW_MS = 24 * 3_600_000;
/** The start a snapshot's name stands for: the inverse of `snapshotName`. */
export function snapshotStart(name: string): Date | undefined {
  const m = /^(\d{4}-\d{2}-\d{2})T(\d{2})-(\d{2})-(\d{2})Z$/.exec(name);
  if (m === null) return undefined;
  const at = new Date(`${m[1]}T${m[2]}:${m[3]}:${m[4]}Z`);
  return Number.isNaN(at.getTime()) ? undefined : at;
}
/** Whether a path is written once and never changed after, so a resume may skip it. */
const resumable = (path: string): boolean => path.startsWith("repos/") || isRunBundle(path);
/** What the journal remembers of an object: its size and the digest of its bytes. */
interface Landed {
  readonly bytes: number;
  readonly sha256: string;
}
const digest = (body: Uint8Array | string): string =>
  new Bun.CryptoHasher("sha256").update(body).digest("hex");
async function readJournal(store: Store, prefix: string): Promise<Map<string, Landed>> {
  const out = new Map<string, Landed>();
  try {
    const body = await store.get(prefix + JOURNAL);
    if (body === undefined) return out;
    const list = JSON.parse(new TextDecoder().decode(body)) as unknown;
    if (!Array.isArray(list)) return out;
    for (const e of list as { path?: unknown; bytes?: unknown; sha256?: unknown }[])
      if (typeof e?.path === "string" && typeof e.bytes === "number" && typeof e.sha256 === "string")
        out.set(e.path, { bytes: e.bytes, sha256: e.sha256 });
  } catch {
    // An unreadable journal is no journal: everything goes up again.
  }
  return out;
}
/** `2026-09-22T09-00-00Z`: sortable, and a name a bucket allows. */
@@ -240,6 +293,30 @@
export async function takeSnapshot(
  };
  const ticker = setInterval(say, everyMs);
  const pending = new Set<Promise<void>>();
  const before = forge.resume ? await readJournal(store, prefix) : new Map<string, Landed>();
  const journal = new Map<string, Landed>();
  let journaled = 0;
  let unwritten = false;
  let writing: Promise<void> = Promise.resolve();
  let finished = false;
  // Best effort: a journal not written costs a retake of some objects.
  const writeJournal = (): void => {
    journaled = 0;
    unwritten = false;
    const body = JSON.stringify([...journal].map(([path, l]) => ({ path, ...l })));
    writing = writing.then(() => store.put(prefix + JOURNAL, body)).catch(() => {});
  };
  // A release kills the process without running a `finally`: what landed must
  // already be written, so a changed journal goes out on a clock too.
  const journalTicker = setInterval(() => {
    if (unwritten) writeJournal();
  }, forge.journalEveryMs ?? JOURNAL_MS);
  const remember = (path: string, l: Landed): void => {
    journal.set(path, l);
    unwritten = true;
    // The first object landed is journaled at once; then every N more.
    if (journal.size === 1 || ++journaled >= JOURNAL_EVERY) writeJournal();
  };
  let failure: Error | undefined;
  let held = 0;
  // Hands an object to the queue: waits for a free slot, then lets the put
@@ -257,8 +334,16 @@
export async function takeSnapshot(
    )
      await Promise.race(pending);
    if (failure !== undefined) throw failure;
    held += bytes;
    files.push({ path, bytes });
    const sha256 = resumable(path) ? digest(body) : "";
    const had = before.get(path);
    if (had !== undefined && had.bytes === bytes && had.sha256 === sha256) {
      journal.set(path, had);
      landed++;
      landedBytes += bytes;
      return;
    }
    held += bytes;
    const run: Promise<void> = withRetries(prefix + path, () =>
      store.put(prefix + path, body),
    )
@@ -266,6 +351,7 @@
export async function takeSnapshot(
        () => {
          landed++;
          landedBytes += bytes;
          if (resumable(path)) remember(path, { bytes, sha256 });
          forge.onLand?.({
            objects: landed,
            bytes: landedBytes,
@@ -455,7 +541,11 @@
export async function takeSnapshot(
      `${prefix}manifest.json`,
      JSON.stringify(manifest, null, 2),
    );
    finished = true;
    await writing;
    await store.delete(prefix + JOURNAL).catch(() => {});
    log.info("snapshot taken", {
      resumed: before.size,
      name,
      where: store.where,
      repos: bundled,
@@ -467,8 +557,11 @@
export async function takeSnapshot(
    return manifest;
  } finally {
    clearInterval(ticker);
    clearInterval(journalTicker);
    // Puts still running when one failed must not outlive the scratch.
    await Promise.allSettled([...pending]);
    if (!finished && journal.size > 0) writeJournal();
    await writing;
    rmSync(scratch, { recursive: true, force: true });
  }
}
@@ -498,13 +591,29 @@
export async function listSnapshots(store: Store): Promise<Manifest[]> {
  return out.sort((a, b) => (a.name < b.name ? 1 : a.name > b.name ? -1 : 0));
}
/** Delete every snapshot past the newest `keep`. Returns what went. */
/**
 * Delete every snapshot past the newest `keep`, and every prefix a release
 * cut short that nothing will resume: one with no manifest that is older
 * than the newest whole snapshot, or began more than a resume window ago.
 * Returns what went.
 */
export async function pruneSnapshots(
  store: Store,
  keep = KEEP,
  now = Date.now(),
): Promise<string[]> {
  const names = (await store.names(SNAPSHOT_PREFIX)).sort().reverse();
  const gone = names.slice(keep);
  const whole: string[] = [];
  const cut: string[] = [];
  for (const name of names)
    ((await store.get(`${SNAPSHOT_PREFIX}${name}/manifest.json`)) === undefined ? cut : whole).push(name);
  const newest = whole[0];
  const stale = cut.filter((name) => {
    const at = snapshotStart(name);
    if (at === undefined) return false;
    return (newest !== undefined && name < newest) || now - at.getTime() > RESUME_WINDOW_MS;
  });
  const gone = [...whole.slice(keep), ...stale];
  for (const name of gone)
    for (const key of await store.list(`${SNAPSHOT_PREFIX}${name}/`))
      await store.delete(key);
@@ -0,0 +1,243 @@
/**
 * A release mid-snapshot (#8828): the snapshot a restart finds without a
 * manifest is carried on under its own name, and the repository bundles
 * that already landed are not put again.
 */
import { afterAll, beforeAll, expect, test } from "bun:test";
import { mkdtempSync } from "node:fs";
import { rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { commitToBranch, createRepo } from "../src/actions/repos.ts";
import { createApp, type App } from "../src/app.ts";
import { createUser } from "../src/core/users.ts";
import { startSnapshots } from "../src/snapshot/service.ts";
import {
  JOURNAL,
  JOURNAL_EVERY,
  listSnapshots,
  pruneSnapshots,
  RESUME_WINDOW_MS,
  snapshotName,
  snapshotStart,
  takeSnapshot,
} from "../src/snapshot/snapshot.ts";
import { dirStore, type Store } from "../src/snapshot/store.ts";
let root: string;
let app: App;
beforeAll(async () => {
  root = mkdtempSync(join(tmpdir(), "forge-resume-"));
  app = createApp({ dataDir: join(root, "live"), port: 0 });
  const a = await createUser(app.db, {
    origin: "seed",
    trusted: true,
    username: "alice",
    email: "[email protected]",
    password: "a-good-passphrase",
    adminIfFirst: true,
  });
  if (!a.ok) throw new Error(a.error);
  for (const name of ["one", "two"]) {
    const repo = await createRepo(app, a.value, { name, visibility: "private" });
    if (!repo.ok) throw new Error(String(repo.error));
    const c = await commitToBranch(app, a.value, repo.value, {
      message: "first",
      files: [{ path: "README.md", content: `# ${name}\n` }],
    });
    if (!c.ok) throw new Error(JSON.stringify(c.error));
  }
});
afterAll(async () => {
  app.close();
  await rm(root, { recursive: true, force: true });
});
const forge = (resume?: boolean, journalEveryMs?: number) => ({
  db: app.db,
  dataDir: app.config.dataDir,
  reposRoot: app.reposRoot,
  version: "test",
  onProgress: () => {},
  resume,
  journalEveryMs,
});
/** A store that counts the keys put under it. */
function counting(inner: Store, puts: string[]): Store {
  return {
    ...inner,
    put: (key, body) => {
      puts.push(key);
      return inner.put(key, body);
    },
  };
}
test("a name gives back the moment it stands for", () => {
  const at = new Date("2026-09-22T09:00:00Z");
  expect(snapshotStart(snapshotName(at))?.getTime()).toBe(at.getTime());
  expect(snapshotStart("not-a-snapshot")).toBeUndefined();
});
test("a snapshot cut short is finished under its name without putting its bundles again", async () => {
  const inner = dirStore(join(root, "store"));
  const at = new Date("2026-11-02T08:00:00Z");
  const prefix = `snapshots/${snapshotName(at)}/`;
  // The first attempt dies when the manifest is due: its journal is written on the way out.
  const dying: Store = {
    ...inner,
    put: (key, body) => {
      if (key.endsWith("manifest.json")) throw new Error("the release came");
      return inner.put(key, body);
    },
  };
  await expect(takeSnapshot(forge(), dying, at)).rejects.toThrow("the release came");
  expect(await inner.get(prefix + "manifest.json")).toBeUndefined();
  const journal = JSON.parse(new TextDecoder().decode(await inner.get(prefix + JOURNAL))) as { path: string }[];
  expect(journal.map((e) => e.path).sort()).toEqual(["repos/alice/one.bundle", "repos/alice/two.bundle"]);
  const puts: string[] = [];
  const manifest = await takeSnapshot(forge(true), counting(inner, puts), at);
  expect(puts.some((k) => k.includes("/repos/"))).toBe(false);
  expect(puts.some((k) => k.includes("forge.db"))).toBe(true);
  expect(manifest.name).toBe(snapshotName(at));
  expect(manifest.files.map((f) => f.path)).toContain("repos/alice/one.bundle");
  expect(manifest.repos).toBe(2);
  // Every object the manifest lists is in the store at its size, and the journal is gone.
  for (const f of manifest.files) expect((await inner.get(prefix + f.path))?.byteLength).toBe(f.bytes);
  expect(await inner.get(prefix + JOURNAL)).toBeUndefined();
});
test("without resume, everything goes up again", async () => {
  const inner = dirStore(join(root, "store2"));
  const puts: string[] = [];
  await takeSnapshot(forge(), counting(inner, puts), new Date("2026-11-02T09:00:00Z"));
  expect(puts.some((k) => k.includes("/repos/"))).toBe(true);
});
const HOUR = 3_600_000;
/** Takes a snapshot at `at` that dies before its manifest, leaving a journal. */
async function cutShortAt(inner: Store, at: Date): Promise<void> {
  const dying: Store = {
    ...inner,
    put: (key, body) => {
      if (key.endsWith("manifest.json")) throw new Error("the release came");
      return inner.put(key, body);
    },
  };
  await expect(takeSnapshot(forge(), dying, at)).rejects.toThrow("the release came");
}
test("a kill that runs no finally still leaves the journal, written on a clock", async () => {
  const inner = dirStore(join(root, "store-kill"));
  const at = new Date("2026-11-02T08:00:00Z");
  const prefix = `snapshots/${snapshotName(at)}/`;
  // Everything but the bundles hangs for ever: the process is killed there.
  const hung: Store = {
    ...inner,
    put: (key, body) => (key.includes("/repos/") || key.endsWith(JOURNAL) ? inner.put(key, body) : new Promise<void>(() => {})),
  };
  void takeSnapshot(forge(false, 20), hung, at).catch(() => {});
  let journal: { path: string }[] = [];
  for (let i = 0; i < 100 && journal.length < 2; i++) {
    await Bun.sleep(50);
    const body = await inner.get(prefix + JOURNAL);
    if (body !== undefined) journal = JSON.parse(new TextDecoder().decode(body)) as { path: string }[];
  }
  expect(journal.map((e) => e.path).sort()).toEqual(["repos/alice/one.bundle", "repos/alice/two.bundle"]);
  expect(await inner.get(prefix + "manifest.json")).toBeUndefined();
}, 15_000);
test("a bundle of the same size but other content is put again", async () => {
  const inner = dirStore(join(root, "store-stale"));
  const at = new Date("2026-11-02T08:00:00Z");
  const prefix = `snapshots/${snapshotName(at)}/`;
  await cutShortAt(inner, at);
  const body = JSON.parse(new TextDecoder().decode(await inner.get(prefix + JOURNAL))) as { path: string; bytes: number; sha256: string }[];
  // The journal remembers a different content of the very size the bundle has now.
  const forged = body.map((e) => (e.path === "repos/alice/one.bundle" ? { ...e, sha256: "0".repeat(64) } : e));
  await inner.put(prefix + JOURNAL, JSON.stringify(forged));
  const puts: string[] = [];
  await takeSnapshot(forge(true), counting(inner, puts), at);
  expect(puts.filter((k) => k.includes("/repos/"))).toEqual([prefix + "repos/alice/one.bundle"]);
});
test("the journal is rewritten after every N landed objects", async () => {
  expect(JOURNAL_EVERY).toBeGreaterThan(1);
  const inner = dirStore(join(root, "store-every"));
  const puts: string[] = [];
  // The clock is far off, so only the first object and the way out write it.
  await expect(
    takeSnapshot(forge(false, 3_600_000), counting({ ...inner, put: (k, b) => (k.endsWith("manifest.json") ? Promise.reject(new Error("x")) : inner.put(k, b)) }, puts), new Date("2026-11-02T08:30:00Z")),
  ).rejects.toThrow("x");
  expect(puts.filter((k) => k.endsWith(JOURNAL)).length).toBeGreaterThanOrEqual(1);
});
async function due(inner: Store, nowMs: number) {
  const puts: string[] = [];
  const counted = counting(inner, puts);
  const again = startSnapshots(app, counted, { version: "t", nightly: false, now: () => nowMs });
  try {
    expect(await again.takeIfDue()).toBe(true);
  } finally {
    await again.stop();
  }
  return { puts, manifests: await listSnapshots(inner) };
}
test("takeIfDue resumes the newest prefix newer than the newest manifest", async () => {
  const inner = dirStore(join(root, "store-due1"));
  const old = new Date("2026-11-02T08:00:00Z");
  const prior = "2026-11-01T08-00-00Z";
  await inner.put(`snapshots/${prior}/manifest.json`, JSON.stringify({ name: prior, files: [], finishedAt: "2026-11-01T08:05:00Z" }));
  await cutShortAt(inner, old);
  const { puts, manifests } = await due(inner, old.getTime() + 2 * HOUR);
  expect(manifests[0]?.name).toBe(snapshotName(old));
  expect(puts.some((k) => k.includes("/repos/"))).toBe(false);
});
test("takeIfDue ignores a prefix older than the resume window", async () => {
  const inner = dirStore(join(root, "store-due2"));
  const old = new Date("2026-11-02T08:00:00Z");
  await cutShortAt(inner, old);
  const { puts, manifests } = await due(inner, old.getTime() + RESUME_WINDOW_MS + HOUR);
  expect(manifests.map((m) => m.name)).not.toContain(snapshotName(old));
  expect(puts.some((k) => k.includes("/repos/"))).toBe(true);
});
test("takeIfDue ignores a prefix with no journal", async () => {
  const inner = dirStore(join(root, "store-due3"));
  const old = new Date("2026-11-02T08:00:00Z");
  await inner.put(`snapshots/${snapshotName(old)}/forge.db`, "x");
  const { puts, manifests } = await due(inner, old.getTime() + 2 * HOUR);
  expect(manifests.map((m) => m.name)).not.toContain(snapshotName(old));
  expect(puts.some((k) => k.includes("/repos/"))).toBe(true);
});
test("a plain due snapshot takes fresh", async () => {
  const inner = dirStore(join(root, "store-due4"));
  const { puts, manifests } = await due(inner, Date.now());
  expect(manifests).toHaveLength(1);
  expect(puts.some((k) => k.includes("/repos/"))).toBe(true);
});
test("pruning removes a cut-short prefix nothing will resume, and keeps one that may be", async () => {
  const s = dirStore(join(root, "store-prune"));
  const whole = new Date("2026-11-02T08:00:00Z");
  await s.put(`snapshots/${snapshotName(whole)}/manifest.json`, "{}");
  const behind = `snapshots/2026-11-01T08-00-00Z/`;
  const stale = `snapshots/2026-10-01T08-00-00Z/`;
  const ahead = `snapshots/2026-11-02T20-00-00Z/`;
  for (const p of [behind, stale, ahead]) {
    await s.put(p + JOURNAL, "[]");
    await s.put(p + "repos/a.bundle", "x");
  }
  const gone = await pruneSnapshots(s, 14, whole.getTime() + 13 * HOUR);
  expect(gone.sort()).toEqual(["2026-10-01T08-00-00Z", "2026-11-01T08-00-00Z"]);
  expect(await s.list(ahead)).toHaveLength(2);
  expect(await s.list(behind)).toHaveLength(0);
});
		@@ -6,7 +6,7 @@
6	6	- Catch-up. A boot with the newest snapshot older than the last 08:00 UTC slot takes one two minutes later (`snapshotDue`). A snapshot cut short by a release has no manifest, so it counts as lost.
7	7	- In the terminal. The Resilience dialog (palette: Resilience, admins) opens with the same verdict, each lane's newest, and every reason. On the web it is home's backup tile and `/settings/backups`; for an agent, `verdict` in the MCP `backups` tool.
8	8	- Off the serving thread. `VACUUM INTO` runs in a worker (`vacuum.ts`).
9		- Not done, tracked: snapshots outside the serving process or resumable (#8828; the off-process probe is below). Catch-up retakes a lost snapshot whole.
	9	- Not done, tracked: snapshots outside the serving process (#8828); resuming across a release is below, and so is the off-process probe.
10	10
11	11	## Platform backup lanes (#8811)
12	12
		@@ -57,8 +57,21 @@ unreadable or empty; or the newest manifest's `finishedAt` older than the
57	57	limit (default 36 h: one nightly plus grace). Use read-only credentials.
58	58	Logic lives in `src/snapshot/probe.ts`; tests in `test/backup-probe.test.ts`.
59	59
60		### Not done yet
	60	## Resuming across a release (#8828)
61	61
62		Snapshots are still taken inside the serving process; a release mid-snapshot
63		is caught up by retaking it whole after boot. Resuming across a release (or
64		moving the snapshot into its own process) is a separate, larger change, tracked in #8828.
	62	A snapshot cut short has no manifest, but it is no longer retaken whole.
	63	`takeSnapshot` writes `partial.json` under the snapshot's prefix (after the
	64	first landed object, every 100 more, every 15 s while something new landed,
	65	and on failure; a release kills the process without running any cleanup, so
	66	the clock is what leaves it): the repository bundles and run bundles that
	67	landed, with their sizes and SHA-256 digests. The catch-up after boot (`takeIfDue`) finds the
	68	newest prefix newer than the newest manifest, begun within 24 h, that holds a
	69	journal, and takes the snapshot again under that name and start time with
	70	`resume`. A bundle or run bundle the journal lists with the same size and digest is
	71	not put again; the database, key, ssh files, app data and recent run logs always are,
	72	since they change while the forge runs. The journal is deleted once the
	73	manifest is written. `pruneSnapshots` removes a prefix with no manifest that
	74	is older than the newest whole snapshot or past the window.
	75
	76	Not done: the snapshot still runs in the serving process (the database copy is
	77	in a worker). Tests: `test/snapshot-resume.test.ts`.
		@@ -14,9 +14,13 @@ import type { App } from "../app.ts";
14	14	import { createLog } from "../kernel/log.ts";
15	15	import { firstSeen, reportVerdict, seedFirstSeen } from "./verdict.ts";
16	16	import {
	17	JOURNAL,
17	18	KEEP,
18	19	listSnapshots,
19	20	pruneSnapshots,
	21	RESUME_WINDOW_MS,
	22	SNAPSHOT_PREFIX,
	23	snapshotStart,
20	24	takeSnapshot,
21	25	type Manifest,
22	26	type Progress,
		@@ -164,7 +168,7 @@ export function startSnapshots(
164	168	let running: Progress \| undefined;
165	169	let last: { manifest?: Manifest; error?: string; at?: string } = {};
166	170
167		const take = (): Promise<Manifest> => {
	171	const take = (resume?: Date): Promise<Manifest> => {
168	172	// Whether this run's store is off the machine, for home's backup age (#3701).
169	173	let offBox: boolean \| undefined;
170	174	// After the one before it, however that ended: chained on a failure as it
		@@ -186,6 +190,7 @@ export function startSnapshots(
186	190	reposRoot: app.reposRoot,
187	191	version: options.version,
188	192	passphrase: options.passphrase,
	193	resume: resume !== undefined,
189	194	onLand: (p) => {
190	195	running = p;
191	196	if (Date.now() - toldAt < (options.progressEventMs ?? PROGRESS_EVENT_MS)) return;
		@@ -194,6 +199,7 @@ export function startSnapshots(
194	199	},
195	200	},
196	201	store,
	202	resume,
197	203	).finally(() => {
198	204	running = undefined;
199	205	});
		@@ -232,6 +238,21 @@ export function startSnapshots(
232	238	return inFlight;
233	239	};
234	240
	241	// The newest snapshot a release cut short that is newer than the newest
	242	// whole one and not stale: its journal says what need not go up again.
	243	const cutShort = async (store: Store, newestWhole: string \| undefined): Promise<Date \| undefined> => {
	244	try {
	245	const names = (await store.names(SNAPSHOT_PREFIX)).sort().reverse();
	246	const name = names.find((n) => newestWhole === undefined \|\| n > newestWhole);
	247	const at = name === undefined ? undefined : snapshotStart(name);
	248	if (name === undefined \|\| at === undefined) return undefined;
	249	if ((options.now ?? Date.now)() - at.getTime() > RESUME_WINDOW_MS) return undefined;
	250	return (await store.get(`${SNAPSHOT_PREFIX}${name}/${JOURNAL}`)) === undefined ? undefined : at;
	251	} catch {
	252	return undefined;
	253	}
	254	};
	255
235	256	// A restart takes the night's snapshot if the store does not hold it: a
236	257	// release at 08:21 killed the one in flight and nothing took it again (#8742).
237	258	// The manifest is written last, so a snapshot cut short is not one.
		@@ -247,8 +268,9 @@ export function startSnapshots(
247	268	return false;
248	269	}
249	270	if (stopped \|\| active > 0 \|\| !snapshotDue(newest?.finishedAt, (options.now ?? Date.now)())) return false;
250		log.info("snapshot due or lost; taking one", { newest: newest?.name ?? "none" });
251		await take().catch(() => {});
	271	const cut = await cutShort(store, newest?.name);
	272	log.info("snapshot due or lost; taking one", { newest: newest?.name ?? "none", resuming: cut?.toISOString() ?? "no" });
	273	await take(cut).catch(() => {});
252	274	return true;
253	275	};
254	276
		@@ -171,6 +171,59 @@ export interface Snapshotter {
171	171	readonly progressEvery?: { objects?: number; ms?: number } \| undefined;
172	172	/** Override of the database's part size, for tests. */
173	173	readonly partBytes?: number \| undefined;
	174	/**
	175	* Carry on a snapshot a release cut short: `at` is its start, and the
	176	* repository bundles and run bundles its journal says landed are not put
	177	* again. Everything else (the database, the key, live files) is taken anew.
	178	*/
	179	readonly resume?: boolean \| undefined;
	180	/** Override of how often a changed journal is written, for tests. */
	181	readonly journalEveryMs?: number \| undefined;
	182	}
	183
	184	/** The journal of what landed of a snapshot still being taken; gone once its manifest is. */
	185	export const JOURNAL = "partial.json";
	186	/** The journal is rewritten after this many more objects have landed. */
	187	export const JOURNAL_EVERY = 100;
	188	/** And no later than this after something landed: a release kills the process between writes. */
	189	export const JOURNAL_MS = 15_000;
	190	/** A snapshot cut short is carried on, not retaken, if it began within this long. */
	191	export const RESUME_WINDOW_MS = 24 * 3_600_000;
	192
	193	/** The start a snapshot's name stands for: the inverse of `snapshotName`. */
	194	export function snapshotStart(name: string): Date \| undefined {
	195	const m = /^(\d{4}-\d{2}-\d{2})T(\d{2})-(\d{2})-(\d{2})Z$/.exec(name);
	196	if (m === null) return undefined;
	197	const at = new Date(`${m[1]}T${m[2]}:${m[3]}:${m[4]}Z`);
	198	return Number.isNaN(at.getTime()) ? undefined : at;
	199	}
	200
	201	/** Whether a path is written once and never changed after, so a resume may skip it. */
	202	const resumable = (path: string): boolean => path.startsWith("repos/") \|\| isRunBundle(path);
	203
	204	/** What the journal remembers of an object: its size and the digest of its bytes. */
	205	interface Landed {
	206	readonly bytes: number;
	207	readonly sha256: string;
	208	}
	209
	210	const digest = (body: Uint8Array \| string): string =>
	211	new Bun.CryptoHasher("sha256").update(body).digest("hex");
	212
	213	async function readJournal(store: Store, prefix: string): Promise<Map<string, Landed>> {
	214	const out = new Map<string, Landed>();
	215	try {
	216	const body = await store.get(prefix + JOURNAL);
	217	if (body === undefined) return out;
	218	const list = JSON.parse(new TextDecoder().decode(body)) as unknown;
	219	if (!Array.isArray(list)) return out;
	220	for (const e of list as { path?: unknown; bytes?: unknown; sha256?: unknown }[])
	221	if (typeof e?.path === "string" && typeof e.bytes === "number" && typeof e.sha256 === "string")
	222	out.set(e.path, { bytes: e.bytes, sha256: e.sha256 });
	223	} catch {
	224	// An unreadable journal is no journal: everything goes up again.
	225	}
	226	return out;
174	227	}
175	228
176	229	/** `2026-09-22T09-00-00Z`: sortable, and a name a bucket allows. */
		@@ -240,6 +293,30 @@ export async function takeSnapshot(
240	293	};
241	294	const ticker = setInterval(say, everyMs);
242	295	const pending = new Set<Promise<void>>();
	296	const before = forge.resume ? await readJournal(store, prefix) : new Map<string, Landed>();
	297	const journal = new Map<string, Landed>();
	298	let journaled = 0;
	299	let unwritten = false;
	300	let writing: Promise<void> = Promise.resolve();
	301	let finished = false;
	302	// Best effort: a journal not written costs a retake of some objects.
	303	const writeJournal = (): void => {
	304	journaled = 0;
	305	unwritten = false;
	306	const body = JSON.stringify([...journal].map(([path, l]) => ({ path, ...l })));
	307	writing = writing.then(() => store.put(prefix + JOURNAL, body)).catch(() => {});
	308	};
	309	// A release kills the process without running a `finally`: what landed must
	310	// already be written, so a changed journal goes out on a clock too.
	311	const journalTicker = setInterval(() => {
	312	if (unwritten) writeJournal();
	313	}, forge.journalEveryMs ?? JOURNAL_MS);
	314	const remember = (path: string, l: Landed): void => {
	315	journal.set(path, l);
	316	unwritten = true;
	317	// The first object landed is journaled at once; then every N more.
	318	if (journal.size === 1 \|\| ++journaled >= JOURNAL_EVERY) writeJournal();
	319	};
243	320	let failure: Error \| undefined;
244	321	let held = 0;
245	322	// Hands an object to the queue: waits for a free slot, then lets the put
		@@ -257,8 +334,16 @@ export async function takeSnapshot(
257	334	)
258	335	await Promise.race(pending);
259	336	if (failure !== undefined) throw failure;
260		held += bytes;
261	337	files.push({ path, bytes });
	338	const sha256 = resumable(path) ? digest(body) : "";
	339	const had = before.get(path);
	340	if (had !== undefined && had.bytes === bytes && had.sha256 === sha256) {
	341	journal.set(path, had);
	342	landed++;
	343	landedBytes += bytes;
	344	return;
	345	}
	346	held += bytes;
262	347	const run: Promise<void> = withRetries(prefix + path, () =>
263	348	store.put(prefix + path, body),
264	349	)
		@@ -266,6 +351,7 @@ export async function takeSnapshot(
266	351	() => {
267	352	landed++;
268	353	landedBytes += bytes;
	354	if (resumable(path)) remember(path, { bytes, sha256 });
269	355	forge.onLand?.({
270	356	objects: landed,
271	357	bytes: landedBytes,
		@@ -455,7 +541,11 @@ export async function takeSnapshot(
455	541	`${prefix}manifest.json`,
456	542	JSON.stringify(manifest, null, 2),
457	543	);
	544	finished = true;
	545	await writing;
	546	await store.delete(prefix + JOURNAL).catch(() => {});
458	547	log.info("snapshot taken", {
	548	resumed: before.size,
459	549	name,
460	550	where: store.where,
461	551	repos: bundled,
		@@ -467,8 +557,11 @@ export async function takeSnapshot(
467	557	return manifest;
468	558	} finally {
469	559	clearInterval(ticker);
	560	clearInterval(journalTicker);
470	561	// Puts still running when one failed must not outlive the scratch.
471	562	await Promise.allSettled([...pending]);
	563	if (!finished && journal.size > 0) writeJournal();
	564	await writing;
472	565	rmSync(scratch, { recursive: true, force: true });
473	566	}
474	567	}
		@@ -498,13 +591,29 @@ export async function listSnapshots(store: Store): Promise<Manifest[]> {
498	591	return out.sort((a, b) => (a.name < b.name ? 1 : a.name > b.name ? -1 : 0));
499	592	}
500	593
501		/** Delete every snapshot past the newest `keep`. Returns what went. */
	594	/**
	595	* Delete every snapshot past the newest `keep`, and every prefix a release
	596	* cut short that nothing will resume: one with no manifest that is older
	597	* than the newest whole snapshot, or began more than a resume window ago.
	598	* Returns what went.
	599	*/
502	600	export async function pruneSnapshots(
503	601	store: Store,
504	602	keep = KEEP,
	603	now = Date.now(),
505	604	): Promise<string[]> {
506	605	const names = (await store.names(SNAPSHOT_PREFIX)).sort().reverse();
507		const gone = names.slice(keep);
	606	const whole: string[] = [];
	607	const cut: string[] = [];
	608	for (const name of names)
	609	((await store.get(`${SNAPSHOT_PREFIX}${name}/manifest.json`)) === undefined ? cut : whole).push(name);
	610	const newest = whole[0];
	611	const stale = cut.filter((name) => {
	612	const at = snapshotStart(name);
	613	if (at === undefined) return false;
	614	return (newest !== undefined && name < newest) \|\| now - at.getTime() > RESUME_WINDOW_MS;
	615	});
	616	const gone = [...whole.slice(keep), ...stale];
508	617	for (const name of gone)
509	618	for (const key of await store.list(`${SNAPSHOT_PREFIX}${name}/`))
510	619	await store.delete(key);
		@@ -0,0 +1,243 @@
	1	/**
	2	* A release mid-snapshot (#8828): the snapshot a restart finds without a
	3	* manifest is carried on under its own name, and the repository bundles
	4	* that already landed are not put again.
	5	*/
	6	import { afterAll, beforeAll, expect, test } from "bun:test";
	7	import { mkdtempSync } from "node:fs";
	8	import { rm } from "node:fs/promises";
	9	import { tmpdir } from "node:os";
	10	import { join } from "node:path";
	11	import { commitToBranch, createRepo } from "../src/actions/repos.ts";
	12	import { createApp, type App } from "../src/app.ts";
	13	import { createUser } from "../src/core/users.ts";
	14	import { startSnapshots } from "../src/snapshot/service.ts";
	15	import {
	16	JOURNAL,
	17	JOURNAL_EVERY,
	18	listSnapshots,
	19	pruneSnapshots,
	20	RESUME_WINDOW_MS,
	21	snapshotName,
	22	snapshotStart,
	23	takeSnapshot,
	24	} from "../src/snapshot/snapshot.ts";
	25	import { dirStore, type Store } from "../src/snapshot/store.ts";
	26
	27	let root: string;
	28	let app: App;
	29
	30	beforeAll(async () => {
	31	root = mkdtempSync(join(tmpdir(), "forge-resume-"));
	32	app = createApp({ dataDir: join(root, "live"), port: 0 });
	33	const a = await createUser(app.db, {
	34	origin: "seed",
	35	trusted: true,
	36	username: "alice",
	37	email: "[email protected]",
	38	password: "a-good-passphrase",
	39	adminIfFirst: true,
	40	});
	41	if (!a.ok) throw new Error(a.error);
	42	for (const name of ["one", "two"]) {
	43	const repo = await createRepo(app, a.value, { name, visibility: "private" });
	44	if (!repo.ok) throw new Error(String(repo.error));
	45	const c = await commitToBranch(app, a.value, repo.value, {
	46	message: "first",
	47	files: [{ path: "README.md", content: `# ${name}\n` }],
	48	});
	49	if (!c.ok) throw new Error(JSON.stringify(c.error));
	50	}
	51	});
	52
	53	afterAll(async () => {
	54	app.close();
	55	await rm(root, { recursive: true, force: true });
	56	});
	57
	58	const forge = (resume?: boolean, journalEveryMs?: number) => ({
	59	db: app.db,
	60	dataDir: app.config.dataDir,
	61	reposRoot: app.reposRoot,
	62	version: "test",
	63	onProgress: () => {},
	64	resume,
	65	journalEveryMs,
	66	});
	67
	68	/** A store that counts the keys put under it. */
	69	function counting(inner: Store, puts: string[]): Store {
	70	return {
	71	...inner,
	72	put: (key, body) => {
	73	puts.push(key);
	74	return inner.put(key, body);
	75	},
	76	};
	77	}
	78
	79	test("a name gives back the moment it stands for", () => {
	80	const at = new Date("2026-09-22T09:00:00Z");
	81	expect(snapshotStart(snapshotName(at))?.getTime()).toBe(at.getTime());
	82	expect(snapshotStart("not-a-snapshot")).toBeUndefined();
	83	});
	84
	85	test("a snapshot cut short is finished under its name without putting its bundles again", async () => {
	86	const inner = dirStore(join(root, "store"));
	87	const at = new Date("2026-11-02T08:00:00Z");
	88	const prefix = `snapshots/${snapshotName(at)}/`;
	89	// The first attempt dies when the manifest is due: its journal is written on the way out.
	90	const dying: Store = {
	91	...inner,
	92	put: (key, body) => {
	93	if (key.endsWith("manifest.json")) throw new Error("the release came");
	94	return inner.put(key, body);
	95	},
	96	};
	97	await expect(takeSnapshot(forge(), dying, at)).rejects.toThrow("the release came");
	98	expect(await inner.get(prefix + "manifest.json")).toBeUndefined();
	99	const journal = JSON.parse(new TextDecoder().decode(await inner.get(prefix + JOURNAL))) as { path: string }[];
	100	expect(journal.map((e) => e.path).sort()).toEqual(["repos/alice/one.bundle", "repos/alice/two.bundle"]);
	101
	102	const puts: string[] = [];
	103	const manifest = await takeSnapshot(forge(true), counting(inner, puts), at);
	104	expect(puts.some((k) => k.includes("/repos/"))).toBe(false);
	105	expect(puts.some((k) => k.includes("forge.db"))).toBe(true);
	106	expect(manifest.name).toBe(snapshotName(at));
	107	expect(manifest.files.map((f) => f.path)).toContain("repos/alice/one.bundle");
	108	expect(manifest.repos).toBe(2);
	109	// Every object the manifest lists is in the store at its size, and the journal is gone.
	110	for (const f of manifest.files) expect((await inner.get(prefix + f.path))?.byteLength).toBe(f.bytes);
	111	expect(await inner.get(prefix + JOURNAL)).toBeUndefined();
	112	});
	113
	114	test("without resume, everything goes up again", async () => {
	115	const inner = dirStore(join(root, "store2"));
	116	const puts: string[] = [];
	117	await takeSnapshot(forge(), counting(inner, puts), new Date("2026-11-02T09:00:00Z"));
	118	expect(puts.some((k) => k.includes("/repos/"))).toBe(true);
	119	});
	120
	121	const HOUR = 3_600_000;
	122
	123	/** Takes a snapshot at `at` that dies before its manifest, leaving a journal. */
	124	async function cutShortAt(inner: Store, at: Date): Promise<void> {
	125	const dying: Store = {
	126	...inner,
	127	put: (key, body) => {
	128	if (key.endsWith("manifest.json")) throw new Error("the release came");
	129	return inner.put(key, body);
	130	},
	131	};
	132	await expect(takeSnapshot(forge(), dying, at)).rejects.toThrow("the release came");
	133	}
	134
	135	test("a kill that runs no finally still leaves the journal, written on a clock", async () => {
	136	const inner = dirStore(join(root, "store-kill"));
	137	const at = new Date("2026-11-02T08:00:00Z");
	138	const prefix = `snapshots/${snapshotName(at)}/`;
	139	// Everything but the bundles hangs for ever: the process is killed there.
	140	const hung: Store = {
	141	...inner,
	142	put: (key, body) => (key.includes("/repos/") \|\| key.endsWith(JOURNAL) ? inner.put(key, body) : new Promise<void>(() => {})),
	143	};
	144	void takeSnapshot(forge(false, 20), hung, at).catch(() => {});
	145	let journal: { path: string }[] = [];
	146	for (let i = 0; i < 100 && journal.length < 2; i++) {
	147	await Bun.sleep(50);
	148	const body = await inner.get(prefix + JOURNAL);
	149	if (body !== undefined) journal = JSON.parse(new TextDecoder().decode(body)) as { path: string }[];
	150	}
	151	expect(journal.map((e) => e.path).sort()).toEqual(["repos/alice/one.bundle", "repos/alice/two.bundle"]);
	152	expect(await inner.get(prefix + "manifest.json")).toBeUndefined();
	153	}, 15_000);
	154
	155	test("a bundle of the same size but other content is put again", async () => {
	156	const inner = dirStore(join(root, "store-stale"));
	157	const at = new Date("2026-11-02T08:00:00Z");
	158	const prefix = `snapshots/${snapshotName(at)}/`;
	159	await cutShortAt(inner, at);
	160	const body = JSON.parse(new TextDecoder().decode(await inner.get(prefix + JOURNAL))) as { path: string; bytes: number; sha256: string }[];
	161	// The journal remembers a different content of the very size the bundle has now.
	162	const forged = body.map((e) => (e.path === "repos/alice/one.bundle" ? { ...e, sha256: "0".repeat(64) } : e));
	163	await inner.put(prefix + JOURNAL, JSON.stringify(forged));
	164	const puts: string[] = [];
	165	await takeSnapshot(forge(true), counting(inner, puts), at);
	166	expect(puts.filter((k) => k.includes("/repos/"))).toEqual([prefix + "repos/alice/one.bundle"]);
	167	});
	168
	169	test("the journal is rewritten after every N landed objects", async () => {
	170	expect(JOURNAL_EVERY).toBeGreaterThan(1);
	171	const inner = dirStore(join(root, "store-every"));
	172	const puts: string[] = [];
	173	// The clock is far off, so only the first object and the way out write it.
	174	await expect(
	175	takeSnapshot(forge(false, 3_600_000), counting({ ...inner, put: (k, b) => (k.endsWith("manifest.json") ? Promise.reject(new Error("x")) : inner.put(k, b)) }, puts), new Date("2026-11-02T08:30:00Z")),
	176	).rejects.toThrow("x");
	177	expect(puts.filter((k) => k.endsWith(JOURNAL)).length).toBeGreaterThanOrEqual(1);
	178	});
	179
	180	async function due(inner: Store, nowMs: number) {
	181	const puts: string[] = [];
	182	const counted = counting(inner, puts);
	183	const again = startSnapshots(app, counted, { version: "t", nightly: false, now: () => nowMs });
	184	try {
	185	expect(await again.takeIfDue()).toBe(true);
	186	} finally {
	187	await again.stop();
	188	}
	189	return { puts, manifests: await listSnapshots(inner) };
	190	}
	191
	192	test("takeIfDue resumes the newest prefix newer than the newest manifest", async () => {
	193	const inner = dirStore(join(root, "store-due1"));
	194	const old = new Date("2026-11-02T08:00:00Z");
	195	const prior = "2026-11-01T08-00-00Z";
	196	await inner.put(`snapshots/${prior}/manifest.json`, JSON.stringify({ name: prior, files: [], finishedAt: "2026-11-01T08:05:00Z" }));
	197	await cutShortAt(inner, old);
	198	const { puts, manifests } = await due(inner, old.getTime() + 2 * HOUR);
	199	expect(manifests[0]?.name).toBe(snapshotName(old));
	200	expect(puts.some((k) => k.includes("/repos/"))).toBe(false);
	201	});
	202
	203	test("takeIfDue ignores a prefix older than the resume window", async () => {
	204	const inner = dirStore(join(root, "store-due2"));
	205	const old = new Date("2026-11-02T08:00:00Z");
	206	await cutShortAt(inner, old);
	207	const { puts, manifests } = await due(inner, old.getTime() + RESUME_WINDOW_MS + HOUR);
	208	expect(manifests.map((m) => m.name)).not.toContain(snapshotName(old));
	209	expect(puts.some((k) => k.includes("/repos/"))).toBe(true);
	210	});
	211
	212	test("takeIfDue ignores a prefix with no journal", async () => {
	213	const inner = dirStore(join(root, "store-due3"));
	214	const old = new Date("2026-11-02T08:00:00Z");
	215	await inner.put(`snapshots/${snapshotName(old)}/forge.db`, "x");
	216	const { puts, manifests } = await due(inner, old.getTime() + 2 * HOUR);
	217	expect(manifests.map((m) => m.name)).not.toContain(snapshotName(old));
	218	expect(puts.some((k) => k.includes("/repos/"))).toBe(true);
	219	});
	220
	221	test("a plain due snapshot takes fresh", async () => {
	222	const inner = dirStore(join(root, "store-due4"));
	223	const { puts, manifests } = await due(inner, Date.now());
	224	expect(manifests).toHaveLength(1);
	225	expect(puts.some((k) => k.includes("/repos/"))).toBe(true);
	226	});
	227
	228	test("pruning removes a cut-short prefix nothing will resume, and keeps one that may be", async () => {
	229	const s = dirStore(join(root, "store-prune"));
	230	const whole = new Date("2026-11-02T08:00:00Z");
	231	await s.put(`snapshots/${snapshotName(whole)}/manifest.json`, "{}");
	232	const behind = `snapshots/2026-11-01T08-00-00Z/`;
	233	const stale = `snapshots/2026-10-01T08-00-00Z/`;
	234	const ahead = `snapshots/2026-11-02T20-00-00Z/`;
	235	for (const p of [behind, stale, ahead]) {
	236	await s.put(p + JOURNAL, "[]");
	237	await s.put(p + "repos/a.bundle", "x");
	238	}
	239	const gone = await pruneSnapshots(s, 14, whole.getTime() + 13 * HOUR);
	240	expect(gone.sort()).toEqual(["2026-10-01T08-00-00Z", "2026-11-01T08-00-00Z"]);
	241	expect(await s.list(ahead)).toHaveLength(2);
	242	expect(await s.list(behind)).toHaveLength(0);
	243	});