trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns17g rEventsFlows

A snapshot uploads in parallel and says how far it is: 15,422 serial puts took 1 h 47 min #1489

Merged scoutcrew opened this in answer to trevatoowner

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

scout/forge:scout/1193-a-snapshot-uploads-in-parallel-and-says → main 7 commits fork

  • Merged. scout/1193-a-snapshot-uploads-in-parallel-and-says was merged into main as e4ed92b.

Review

Sign in to review this change.
7 files changed +486 −15
README.md +6
@@ -365,6 +365,12 @@ it answers by default: `.well-known`, feeds, cards, the manifest.
365365 `docs/SNAPSHOT.md` is a forge back from a copy: what a nightly snapshot
366366 holds, where it goes, and the one command that restores it.
367367
368 A snapshot uploads eight objects at a time. Run logs older than a day go up
369 bundled per day (cut at 4 MB, so a long history is several objects, never one
370 huge one); restore unpacks them. While one runs, progress shows on
371 **Settings → Backups** and as a line in the log every 500 objects or minute.
372 The terminal has no backups view, so it has no readout.
373
368374 `docs/EXTRACT.md` is the forge on a machine of its own, and the platform it
369375 grew up in switched off: what a host owes it, the moves in order, and what is
370376 given up.
src/snapshot/restore.ts +18 −1
@@ -27,7 +27,14 @@ import { dirname, join } from "node:path";
2727 import { runGit } from "../git/env.ts";
2828 import { installHooks } from "../git/hooks.ts";
2929 import { repoDir } from "../git/repos.ts";
30 import { keyOf, listSnapshots, SNAPSHOT_PREFIX, type Manifest } from "./snapshot.ts";
30 import {
31 isRunBundle,
32 keyOf,
33 listSnapshots,
34 SNAPSHOT_PREFIX,
35 unpackRuns,
36 type Manifest,
37 } from "./snapshot.ts";
3138 import { unwrapKey, WRAPPED_KEY } from "./wrap.ts";
3239 import { resolveDestination } from "./destination.ts";
3340 import type { Store } from "./store.ts";
@@ -111,6 +118,16 @@ export async function restoreSnapshot(
111118 `snapshot ${name}: ${entry.path} is ${body.byteLength} bytes, not ${entry.bytes}`,
112119 );
113120 if (entry.path === WRAPPED_KEY) continue;
121 // A day's run logs in one object come back as the logs they were.
122 if (isRunBundle(entry.path)) {
123 for (const run of unpackRuns(body)) {
124 const out = join(dataDir, "runs", run.path);
125 mkdirSync(dirname(out), { recursive: true });
126 await Bun.write(out, run.body);
127 files++;
128 }
129 continue;
130 }
114131 const bundle = entry.path.match(/^repos\/([^/]+)\/([^/]+)\.bundle$/);
115132 // A bundle is not a file of the forge's; it becomes a repository below.
116133 const target = bundle
src/snapshot/service.ts +12 −1
@@ -16,6 +16,7 @@ import {
1616 pruneSnapshots,
1717 takeSnapshot,
1818 type Manifest,
19 type Progress,
1920 } from "./snapshot.ts";
2021 import type { Store } from "./store.ts";
2122
@@ -31,6 +32,8 @@ export interface SnapshotsHandle {
3132 readonly wraps: boolean;
3233 /** Take one now; waits for any in flight. */
3334 readonly now: () => Promise<Manifest>;
35 /** How far the snapshot being taken is; undefined when none is. */
36 readonly progress: () => Progress | undefined;
3437 /** The newest first, from the store. */
3538 readonly list: () => Promise<Manifest[]>;
3639 /** The last one taken by this process, and what it said. */
@@ -72,6 +75,7 @@ export function startSnapshots(
7275 };
7376 let stopped = false;
7477 let inFlight: Promise<Manifest> | undefined;
78 let running: Progress | undefined;
7579 let last: { manifest?: Manifest; error?: string; at?: string } = {};
7680
7781 const take = (): Promise<Manifest> => {
@@ -81,6 +85,7 @@ export function startSnapshots(
8185 const store = current();
8286 if (store === undefined)
8387 throw new Error("this forge has nowhere to put a snapshot");
88 running = { objects: 0, bytes: 0, elapsedMs: 0 };
8489 const manifest = await takeSnapshot(
8590 {
8691 db: app.db,
@@ -88,9 +93,14 @@ export function startSnapshots(
8893 reposRoot: app.reposRoot,
8994 version: options.version,
9095 passphrase: options.passphrase,
96 onLand: (p) => {
97 running = p;
98 },
9199 },
92100 store,
93 );
101 ).finally(() => {
102 running = undefined;
103 });
94104 await pruneSnapshots(store, KEEP);
95105 return manifest;
96106 });
@@ -130,6 +140,7 @@ export function startSnapshots(
130140 return store === undefined ? [] : listSnapshots(store);
131141 },
132142 last: () => last,
143 progress: () => running,
133144 forget: () => {
134145 resolved = undefined;
135146 },
src/snapshot/snapshot.ts +207 −11
@@ -12,8 +12,13 @@
1212 * checks, previews, pictures.
1313 *
1414 * A snapshot is named by the moment it began, in UTC, so a listing sorts
15 * itself. Nothing is compressed or bundled beyond git's own bundles: a
16 * restore is reading files back, and a person can read the manifest.
15 * itself. Nothing is compressed or bundled beyond git's own bundles and
16 * the crew's older run logs, one compressed bundle per day: a bucket costs
17 * a round trip per object, so thousands of tiny logs are one object a day.
18 * A restore is reading files back, and a person can read the manifest.
19 *
20 * Objects go up through a bounded queue, eight in flight; the manifest
21 * lists them in the order they were handed over, not the order they landed.
1722 */
1823
1924 import type { Database } from "bun:sqlite";
@@ -61,6 +66,86 @@ export interface Manifest {
6166 export const keyOf = (m: Manifest): "wrapped" | "none" | "raw" =>
6267 m.key ?? (m.files.some((f) => f.path === "secret.key") ? "raw" : "none");
6368
69 /** How far a running snapshot is. */
70 export interface Progress {
71 readonly objects: number;
72 readonly bytes: number;
73 readonly elapsedMs: number;
74 }
75
76 /** Puts in flight at once: latency, not bandwidth, was the hour. */
77 export const IN_FLIGHT = 8;
78 /** Bytes held in memory by puts in flight; past this the next file waits. */
79 export const IN_FLIGHT_BYTES = 64 * 1024 * 1024;
80 /** A progress line every this many objects or this many ms, whichever first. */
81 export const PROGRESS_OBJECTS = 500;
82 export const PROGRESS_MS = 60_000;
83 /** Run logs touched within this long stay one object each; older go in daily bundles. */
84 export const RUN_WINDOW_MS = 24 * 3_600_000;
85 /** A bundle is cut before its logs pass this many bytes, so one is never a day's whole history in memory. */
86 export const BUNDLE_BYTES = 4 * 1_048_576;
87 const BUNDLE_NAME = /^runs\/\d{4}-\d{2}-\d{2}(-\d+)?\.ndjson\.zst$/;
88
89 /** Whether a manifest path is a daily bundle of run logs. */
90 export const isRunBundle = (path: string): boolean => BUNDLE_NAME.test(path);
91
92 /** A bundle's body: one JSON line per log, its body in base64, zstd over all. */
93 export function packRuns(
94 logs: readonly { path: string; body: Uint8Array }[],
95 ): Uint8Array {
96 const text = logs
97 .map((l) =>
98 JSON.stringify({
99 path: l.path,
100 body: Buffer.from(l.body).toString("base64"),
101 }),
102 )
103 .join("\n");
104 return Bun.zstdCompressSync(text);
105 }
106
107 /** The logs in a bundle; paths that would leave the runs directory are refused. */
108 export function unpackRuns(
109 bundle: Uint8Array,
110 ): { path: string; body: Uint8Array }[] {
111 const text = new TextDecoder().decode(Bun.zstdDecompressSync(bundle));
112 const out: { path: string; body: Uint8Array }[] = [];
113 for (const line of text.split("\n")) {
114 if (line === "") continue;
115 const entry = JSON.parse(line) as { path?: unknown; body?: unknown };
116 if (typeof entry.path !== "string" || typeof entry.body !== "string")
117 throw new Error("a run bundle holds a line that is not a log");
118 if (
119 entry.path === "" ||
120 entry.path.startsWith("/") ||
121 entry.path.split("/").includes("..")
122 )
123 throw new Error(`a run bundle names ${entry.path}`);
124 out.push({
125 path: entry.path,
126 body: new Uint8Array(Buffer.from(entry.body, "base64")),
127 });
128 }
129 return out;
130 }
131
132 /** Three tries at one put; the last failure names the key. */
133 async function withRetries(
134 key: string,
135 put: () => Promise<void>,
136 ): Promise<void> {
137 let failure: unknown;
138 for (let attempt = 0; attempt < 3; attempt++) {
139 try {
140 return await put();
141 } catch (error) {
142 failure = error;
143 if (attempt < 2) await Bun.sleep(200 * 4 ** attempt);
144 }
145 }
146 throw new Error(`could not put ${key}: ${String(failure)}`);
147 }
148
64149 export interface Snapshotter {
65150 readonly db: Database;
66151 readonly dataDir: string;
@@ -68,6 +153,12 @@ export interface Snapshotter {
68153 readonly version: string;
69154 /** The operator's wrapping secret; without it the key stays out. */
70155 readonly passphrase?: string | undefined;
156 /** Told every so often how far the upload is; the default writes the progress line to the log. */
157 readonly onProgress?: ((p: Progress) => void) | undefined;
158 /** Told each time an object lands, for a page that shows the snapshot live. */
159 readonly onLand?: ((p: Progress) => void) | undefined;
160 /** Overrides of the progress cadence, for tests. */
161 readonly progressEvery?: { objects?: number; ms?: number } | undefined;
71162 }
72163
73164 /** `2026-09-22T09-00-00Z`: sortable, and a name a bucket allows. */
@@ -122,16 +213,76 @@ export async function takeSnapshot(
122213 const prefix = `${SNAPSHOT_PREFIX}${name}/`;
123214 const scratch = mkdtempSync(join(tmpdir(), "forge-snapshot-"));
124215 const files: SnapshotFile[] = [];
216 const started = Date.now();
217 const everyObjects = forge.progressEvery?.objects ?? PROGRESS_OBJECTS;
218 const everyMs = forge.progressEvery?.ms ?? PROGRESS_MS;
219 const report =
220 forge.onProgress ??
221 ((p: Progress) =>
222 log.info("in progress", {
223 objects: p.objects,
224 bytes: p.bytes,
225 elapsed: `${Math.round(p.elapsedMs / 1000)}s`,
226 }));
227 let landed = 0;
228 let landedBytes = 0;
229 let sinceReport = 0;
230 const say = (): void => {
231 sinceReport = 0;
232 report({
233 objects: landed,
234 bytes: landedBytes,
235 elapsedMs: Date.now() - started,
236 });
237 };
238 const ticker = setInterval(say, everyMs);
239 const pending = new Set<Promise<void>>();
240 let failure: Error | undefined;
241 let held = 0;
242 // Hands an object to the queue: waits for a free slot, then lets the put
243 // run on its own while the caller reads the next file.
125244 const put = async (
126245 path: string,
127246 body: Uint8Array | string,
128247 ): Promise<void> => {
129 await store.put(prefix + path, body);
130 files.push({
131 path,
132 bytes:
133 typeof body === "string" ? Buffer.byteLength(body) : body.byteLength,
134 });
248 const bytes =
249 typeof body === "string" ? Buffer.byteLength(body) : body.byteLength;
250 while (
251 (pending.size >= IN_FLIGHT ||
252 (pending.size > 0 && held + bytes > IN_FLIGHT_BYTES)) &&
253 failure === undefined
254 )
255 await Promise.race(pending);
256 if (failure !== undefined) throw failure;
257 held += bytes;
258 files.push({ path, bytes });
259 const run: Promise<void> = withRetries(prefix + path, () =>
260 store.put(prefix + path, body),
261 )
262 .then(
263 () => {
264 landed++;
265 landedBytes += bytes;
266 forge.onLand?.({
267 objects: landed,
268 bytes: landedBytes,
269 elapsedMs: Date.now() - started,
270 });
271 if (++sinceReport >= everyObjects) say();
272 },
273 (error) => {
274 failure ??= error as Error;
275 },
276 )
277 .finally(() => {
278 held -= bytes;
279 pending.delete(run);
280 });
281 pending.add(run);
282 };
283 const drain = async (): Promise<void> => {
284 while (pending.size > 0) await Promise.all([...pending]);
285 if (failure !== undefined) throw failure;
135286 };
136287 const putFile = async (path: string, file: string): Promise<void> => {
137288 const f = Bun.file(file);
@@ -214,9 +365,51 @@ export async function takeSnapshot(
214365 }
215366 }
216367
217 // The crew's run logs: history a person may want back.
218 for (const f of filesUnder(join(forge.dataDir, "runs")))
219 await putFile(`runs/${f}`, join(forge.dataDir, "runs", f));
368 // The crew's run logs: history a person may want back. Recent ones go
369 // up as they are; older ones, a day's worth to an object.
370 const runsDir = join(forge.dataDir, "runs");
371 const days = new Map<string, { path: string; size: number }[]>();
372 for (const f of filesUnder(runsDir)) {
373 const file = join(runsDir, f);
374 const age = statSync(file, { throwIfNoEntry: false });
375 if (age === undefined) continue;
376 if (Date.now() - age.mtimeMs < RUN_WINDOW_MS) {
377 await putFile(`runs/${f}`, file);
378 continue;
379 }
380 const day = age.mtime.toISOString().slice(0, 10);
381 const logs = days.get(day) ?? [];
382 logs.push({ path: f, size: age.size });
383 days.set(day, logs);
384 }
385 // Only one bundle's logs are in memory at a time.
386 for (const day of [...days.keys()].sort()) {
387 const chunks: { path: string; size: number }[][] = [[]];
388 let held = 0;
389 for (const entry of days.get(day)!) {
390 if (held > 0 && held + entry.size > BUNDLE_BYTES) {
391 chunks.push([]);
392 held = 0;
393 }
394 chunks.at(-1)!.push(entry);
395 held += entry.size;
396 }
397 for (const [n, chunk] of chunks.entries()) {
398 const bodies: { path: string; body: Uint8Array }[] = [];
399 for (const l of chunk) {
400 const f = Bun.file(join(runsDir, l.path));
401 if (await f.exists())
402 bodies.push({ path: l.path, body: new Uint8Array(await f.arrayBuffer()) });
403 }
404 if (bodies.length > 0)
405 await put(
406 `runs/${day}${n === 0 ? "" : `-${n + 1}`}.ndjson.zst`,
407 packRuns(bodies),
408 );
409 }
410 }
411
412 await drain();
220413
221414 const manifest: Manifest = {
222415 name,
@@ -244,6 +437,9 @@ export async function takeSnapshot(
244437 });
245438 return manifest;
246439 } finally {
440 clearInterval(ticker);
441 // Puts still running when one failed must not outlive the scratch.
442 await Promise.allSettled([...pending]);
247443 rmSync(scratch, { recursive: true, force: true });
248444 }
249445 }
src/web/pages/backups.ts +12 −1
@@ -31,10 +31,11 @@ function requireAdmin(ctx: Ctx): Response | undefined {
3131 return undefined;
3232 }
3333
34 function shell(ctx: Ctx, body: Html): Response {
34 function shell(ctx: Ctx, body: Html, moving = false): Response {
3535 return page({
3636 ctx,
3737 title: "Backups",
38 ...(moving ? { refresh: { seconds: 5 } } : {}),
3839 body: html`<div class="page-head"><h1>Backups</h1></div>
3940 ${settingsLayout(
4041 accountSections("backups", { admin: true }),
@@ -122,6 +123,15 @@ bun run src/snapshot/restore.ts ${snapshots[0].name} /path/to/empty/data</pre>`
122123 fourteen.
123124 </p>
124125 ${keyNote(svc.wraps, snapshots[0])}
126 ${when(
127 svc.progress(),
128 (p) =>
129 html`<p class="backups-progress">
130 ${icon("clock")} A snapshot is being taken: ${p.objects} objects,
131 ${fileSize(p.bytes)} so far, ${Math.round(p.elapsedMs / 1000)} s
132 in.
133 </p>`,
134 )}
125135 ${when(
126136 last.error,
127137 (error) =>
@@ -195,6 +205,7 @@ bun run src/snapshot/restore.ts ${snapshots[0].name} /path/to/empty/data</pre>`
195205 )
196206 }
197207 ${form}`,
208 svc.progress() !== undefined,
198209 );
199210 }
200211
test/snapshot-destination.test.ts +45
@@ -17,6 +17,7 @@ import {
1717 } from "../src/actions/destination.ts";
1818 import { createUser, type User } from "../src/core/users.ts";
1919 import { resolveDestination } from "../src/snapshot/destination.ts";
20 import { dirStore, type Store } from "../src/snapshot/store.ts";
2021 import { startSnapshots } from "../src/snapshot/service.ts";
2122 import { serve, type Servers } from "../src/http/server.ts";
2223 import { quietLogs } from "./helpers/globals.ts";
@@ -136,6 +137,50 @@ describe("the snapshot destination", () => {
136137 expect(sentence).not.toContain(input.accessKey);
137138 });
138139
140 test("the page says a snapshot is being taken, even before there is a first one", async () => {
141 const slow: Store = {
142 ...dirStore(mkdtempSync(join(tmpdir(), "forge-slow-"))),
143 where: "slow",
144 put: async () => {
145 await Bun.sleep(150);
146 },
147 };
148 const svc = startSnapshots(app, slow, { version: "t", nightly: false });
149 const servers: Servers = await serve(app);
150 try {
151 const signedIn = await fetch(`${servers.url}/signin`, {
152 method: "POST",
153 headers: { origin: servers.url },
154 body: new URLSearchParams({
155 username: "alice",
156 password: "a-good-passphrase",
157 }),
158 redirect: "manual",
159 });
160 const cookie = (signedIn.headers.get("set-cookie") ?? "").split(";")[0]!;
161 const page = async () =>
162 (
163 await fetch(`${servers.url}/settings/backups`, { headers: { cookie } })
164 ).text();
165 expect(await page()).not.toContain("A snapshot is being taken");
166 expect(await page()).not.toContain("http-equiv=\"refresh\"");
167 const taking = svc.now();
168 let html = "";
169 for (let i = 0; i < 100 && !html.includes("A snapshot is being taken"); i++) {
170 await Bun.sleep(50);
171 html = await page();
172 }
173 expect(html).toContain("A snapshot is being taken");
174 expect(html).toContain("http-equiv=\"refresh\"");
175 expect(html).toMatch(/[1-9]\d* objects/);
176 await taking;
177 expect(await page()).not.toContain("A snapshot is being taken");
178 } finally {
179 servers.stop();
180 await svc.stop();
181 }
182 }, 20_000);
183
139184 test("bad input is refused and non-admins are told nothing", async () => {
140185 for (const bad of [
141186 { ...input, endpoint: "ftp://x" },
test/snapshot.test.ts +186 −1
@@ -12,6 +12,7 @@ import {
1212 mkdirSync,
1313 mkdtempSync,
1414 readFileSync,
15 utimesSync,
1516 writeFileSync,
1617 } from "node:fs";
1718 import { rm } from "node:fs/promises";
@@ -31,13 +32,16 @@ import { repoDir } from "../src/git/repos.ts";
3132 import { restoreSnapshot } from "../src/snapshot/restore.ts";
3233 import { startSnapshots, untilNightly } from "../src/snapshot/service.ts";
3334 import {
35 isRunBundle,
3436 listSnapshots,
3537 pruneSnapshots,
3638 snapshotName,
39 packRuns,
3740 takeSnapshot,
41 unpackRuns,
3842 } from "../src/snapshot/snapshot.ts";
3943 import { unwrapKey, wrapKey } from "../src/snapshot/wrap.ts";
40 import { dirStore } from "../src/snapshot/store.ts";
44 import { dirStore, type Store } from "../src/snapshot/store.ts";
4145
4246 const PASS = "correct horse battery staple";
4347
@@ -408,3 +412,184 @@ describe("a forge back from a snapshot", () => {
408412 }
409413 });
410414 });
415
416 describe("a snapshot's uploads", () => {
417 let root: string;
418 let app: App;
419 beforeAll(() => {
420 root = mkdtempSync(join(tmpdir(), "forge-snapshot-queue-"));
421 app = createApp({ dataDir: join(root, "live"), port: 0 });
422 mkdirSync(join(app.config.dataDir, "runs"), { recursive: true });
423 });
424 afterAll(async () => {
425 app.close();
426 await rm(root, { recursive: true, force: true });
427 });
428 const forge = (extra: object = {}) => ({
429 db: app.db,
430 dataDir: app.config.dataDir,
431 reposRoot: app.reposRoot,
432 version: "t",
433 ...extra,
434 });
435 /** A store that holds nothing but counts, and answers after `ms`. */
436 const slowStore = (ms: number, failOn?: string) => {
437 const seen = { inFlight: 0, peak: 0, keys: [] as string[], started: [] as string[] };
438 const store: Store = {
439 ...dirStore(join(root, "unused")),
440 where: "fake",
441 async put(key) {
442 seen.started.push(key);
443 seen.inFlight++;
444 seen.peak = Math.max(seen.peak, seen.inFlight);
445 await Bun.sleep(ms);
446 seen.inFlight--;
447 if (failOn !== undefined && key.endsWith(failOn))
448 throw new Error("refused");
449 seen.keys.push(key);
450 },
451 };
452 return { store, seen };
453 };
454 const logsDir = () => join(app.config.dataDir, "runs");
455 const old = (name: string, body: string, day: string) => {
456 const file = join(logsDir(), name);
457 writeFileSync(file, body);
458 const when = new Date(`${day}T12:00:00Z`);
459 utimesSync(file, when, when);
460 };
461
462 test("at most eight puts are in flight, and the manifest keeps its order", async () => {
463 for (let i = 0; i < 40; i++)
464 writeFileSync(join(logsDir(), `new-${String(i).padStart(2, "0")}.log`), "x");
465 const { store, seen } = slowStore(15);
466 const manifest = await takeSnapshot(forge(), store);
467 expect(seen.peak).toBeGreaterThan(1);
468 expect(seen.peak).toBeLessThanOrEqual(8);
469 // Handed over in the manifest's order, whatever order they landed in.
470 expect(seen.started.slice(0, -1)).toEqual(
471 manifest.files.map((f) => `snapshots/${manifest.name}/${f.path}`),
472 );
473 expect(manifest.files.filter((f) => f.path.startsWith("runs/new-")).length).toBe(40);
474 expect(seen.keys.length).toBe(manifest.files.length + 1);
475 });
476
477 test("held bytes are bounded too: big files go through two at a time", async () => {
478 const big = "x".repeat(30 * 1024 * 1024);
479 for (let i = 0; i < 4; i++) writeFileSync(join(logsDir(), `big-${i}.log`), big);
480 const inner = slowStore(30).store;
481 let bigIn = 0;
482 let bigPeak = 0;
483 const counting = {
484 ...inner,
485 async put(...args: Parameters<typeof inner.put>) {
486 const isBig = args[0].includes("/runs/big-");
487 if (isBig) bigPeak = Math.max(bigPeak, ++bigIn);
488 try {
489 await inner.put(...args);
490 } finally {
491 if (isBig) bigIn--;
492 }
493 },
494 };
495 await takeSnapshot(forge(), counting);
496 expect(bigPeak).toBeLessThanOrEqual(2);
497 expect(bigPeak).toBeGreaterThan(0);
498 }, 20_000);
499
500 test("a failed put fails the snapshot and names its key", async () => {
501 const { store } = slowStore(1, "runs/new-07.log");
502 await expect(takeSnapshot(forge(), store)).rejects.toThrow(
503 /could not put snapshots\/.*runs\/new-07\.log/,
504 );
505 }, 15_000);
506
507 test("a slow store gets a progress line every so many objects", async () => {
508 const { store } = slowStore(10);
509 const lines: { objects: number; bytes: number }[] = [];
510 await takeSnapshot(
511 forge({
512 onProgress: (p: { objects: number; bytes: number }) => lines.push(p),
513 progressEvery: { objects: 10, ms: 60_000 },
514 }),
515 store,
516 );
517 expect(lines.length).toBeGreaterThanOrEqual(4);
518 expect(lines[0]!.objects).toBeGreaterThanOrEqual(10);
519 expect(lines.at(-1)!.bytes).toBeGreaterThan(0);
520 // And by the clock when objects are slow to land.
521 const ticks: number[] = [];
522 await takeSnapshot(
523 forge({
524 onProgress: (p: { objects: number }) => ticks.push(p.objects),
525 progressEvery: { objects: 100_000, ms: 20 },
526 }),
527 slowStore(30).store,
528 );
529 expect(ticks.length).toBeGreaterThan(0);
530 });
531
532 test("older run logs go up as one object a day and restore as the logs they were", async () => {
533 old("a.log", "alpha\n", "2026-10-10");
534 old("b.log", "beta\n", "2026-10-10");
535 old("c.log", "gamma\n", "2026-10-11");
536 const store = dirStore(join(root, "bundled"));
537 const manifest = await takeSnapshot(forge(), store);
538 const paths = manifest.files.map((f) => f.path);
539 expect(paths).toContain("runs/2026-10-10.ndjson.zst");
540 expect(paths).toContain("runs/2026-10-11.ndjson.zst");
541 expect(paths).not.toContain("runs/a.log");
542 const target = join(root, "restored");
543 await restoreSnapshot(store, manifest.name, target);
544 expect(readFileSync(join(target, "runs", "a.log"), "utf8")).toBe("alpha\n");
545 expect(readFileSync(join(target, "runs", "b.log"), "utf8")).toBe("beta\n");
546 expect(readFileSync(join(target, "runs", "c.log"), "utf8")).toBe("gamma\n");
547 expect(readFileSync(join(target, "runs", "new-03.log"), "utf8")).toBe("x");
548 });
549
550 test("a bundle that names a path outside runs is refused", () => {
551 const bad = packRuns([{ path: "../evil", body: new Uint8Array([1]) }]);
552 expect(() => unpackRuns(bad)).toThrow(/names/);
553 });
554
555 test("a bundle is cut before it grows past the cap, and every log still restores", async () => {
556 const big = "z".repeat(1_500_000);
557 for (const n of [1, 2, 3, 4]) old(`big-${n}.log`, big + n, "2026-10-09");
558 const store = dirStore(join(root, "capped"));
559 const manifest = await takeSnapshot(forge(), store);
560 const bundles = manifest.files
561 .map((f) => f.path)
562 .filter((p) => p.startsWith("runs/2026-10-09"));
563 expect(bundles.length).toBeGreaterThan(1);
564 expect(bundles.every(isRunBundle)).toBe(true);
565 const target = join(root, "capped-restored");
566 await restoreSnapshot(store, manifest.name, target);
567 for (const n of [1, 2, 3, 4])
568 expect(readFileSync(join(target, "runs", `big-${n}.log`), "utf8")).toBe(
569 big + n,
570 );
571 }, 15_000);
572
573 test("the service shows a running snapshot's progress while it runs", async () => {
574 const { store } = slowStore(40);
575 const svc = startSnapshots(app, store, { version: "t", nightly: false });
576 try {
577 expect(svc.progress()).toBeUndefined();
578 const taking = svc.now();
579 let seen: { objects: number; bytes: number } | undefined;
580 let done = false;
581 void taking.then(() => (done = true));
582 while (!done && (seen?.objects ?? 0) === 0) {
583 seen = svc.progress();
584 await Bun.sleep(10);
585 }
586 expect(seen).toBeDefined();
587 expect(seen!.objects).toBeGreaterThan(0);
588 expect(done).toBe(false);
589 await taking;
590 expect(svc.progress()).toBeUndefined();
591 } finally {
592 await svc.stop();
593 }
594 }, 15_000);
595 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs