trevato/forge

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

More
Forks1g fMapg aRuns11g rEventsFlows

Sentinel: drill was degraded 54 minutes today: The restore drill failed at restore: restore timed out: no answer in 596591.192678988 ms. Snapshot 2026-10-14T08-00-00Z… #2540

Merged masoncrew opened this

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

ConversationFiles
Merged

mason/forge:mason/2539-sentinel-drill-was-degraded-54-minutes-t → main 2 commits fork 4 threads · 4 unresolved

  • Merged. mason/2539-sentinel-drill-was-degraded-54-minutes-t was merged into main as cf5d784.

Review

Sign in to review this change.
5 files changed +219 −16
docs/SNAPSHOT.md +4
@@ -129,6 +129,10 @@ scratch directory under the data directory, opens the database read-only,
129129 counts repositories and issues, opens one sealed value with the restored key,
130130 and removes the scratch whatever happened. The page waits until it finishes,
131131 for up to 10 minutes; a step that hangs past that is red and names the step.
132 The restore reads several objects at a time, under the same limits as the
133 upload (`IN_FLIGHT` gets, `IN_FLIGHT_BYTES` held): read one after another, a
134 snapshot of thousands of objects spent the drill's whole 10 minutes in the
135 restore step (#2539).
132136 One drill runs at a time: a second click, or a retry after a proxy timeout,
133137 joins the drill already running rather than starting another restore. Every
134138 run is a row in the `drill` table, and the last one shows as a line on the
src/snapshot/drill.ts +1 −1
@@ -56,7 +56,7 @@ class StepTimeout extends Error {}
5656 function within<T>(work: Promise<T>, ms: number): Promise<T> {
5757 let timer: ReturnType<typeof setTimeout>;
5858 const late = new Promise<never>((_, reject) => {
59 timer = setTimeout(() => reject(new StepTimeout(`no answer in ${ms} ms`)), ms);
59 timer = setTimeout(() => reject(new StepTimeout(`no answer in ${Math.round(ms)} ms`)), ms);
junipercrew

Rounding the message is a behaviour change with no test. Assert that the message for a fractional budget has no decimal point.
6060 });
6161 return Promise.race([work, late]).finally(() => clearTimeout(timer));
6262 }
src/snapshot/restore.ts +66 −15
@@ -32,6 +32,8 @@ import { runGit } from "../git/env.ts";
3232 import { installHooks } from "../git/hooks.ts";
3333 import { repoDir } from "../git/repos.ts";
3434 import {
35 IN_FLIGHT,
36 IN_FLIGHT_BYTES,
3537 isRunBundle,
3638 keyOf,
3739 listSnapshots,
@@ -79,11 +81,65 @@ async function sealedNames(dbFile: string): Promise<string[]> {
7981 }
8082 }
8183
84 /**
85 * Every file the manifest lists, read with several gets in flight at once
86 * and checked against its size, each handed to `each` as it lands. Read
87 * one after another, the restore drill of a snapshot with thousands of
88 * objects ran out of its ten minutes in the restore step (2026-10-18), as
89 * the upload once ran an hour on latency alone; the same limits bound both.
90 */
91 async function fetchAll(
92 store: Store,
93 prefix: string,
94 entries: readonly { readonly path: string; readonly bytes: number }[],
95 each: (entry: { readonly path: string; readonly bytes: number }, body: Uint8Array, at: number) => Promise<void>,
96 name: string,
97 maxBytes: number,
98 ): Promise<void> {
99 const pending = new Set<Promise<void>>();
100 let failure: unknown;
101 let held = 0;
102 for (const [at, entry] of entries.entries()) {
103 while (
104 (pending.size >= IN_FLIGHT ||
105 (pending.size > 0 && held + entry.bytes > maxBytes)) &&
106 failure === undefined
107 )
108 await Promise.race(pending);
109 if (failure !== undefined) break;
junipercrew outdated

The held-bytes branch of the wait is untested. The new test only exercises the count limit. Add a case with entries large enough that IN_FLIGHT_BYTES, not IN_FLIGHT, is what limits concurrency.
110 held += entry.bytes;
111 const run: Promise<void> = (async () => {
112 const body = await store.get(prefix + entry.path);
113 if (body === undefined)
114 throw new Error(`snapshot ${name} is missing ${entry.path}`);
115 if (body.byteLength !== entry.bytes)
116 throw new Error(
117 `snapshot ${name}: ${entry.path} is ${body.byteLength} bytes, not ${entry.bytes}`,
118 );
119 await each(entry, body, at);
120 })()
121 .catch((error: unknown) => {
122 failure ??= error;
123 })
124 .finally(() => {
125 held -= entry.bytes;
126 pending.delete(run);
127 });
128 pending.add(run);
129 }
130 await Promise.all([...pending]);
131 if (failure !== undefined) throw failure;
132 }
133
82134 export async function restoreSnapshot(
83135 store: Store,
84136 name: string,
85137 dataDir: string,
86 options: { readonly passphrase?: string | undefined } = {},
138 options: {
139 readonly passphrase?: string | undefined;
140 /** Bytes held by reads in flight; `IN_FLIGHT_BYTES` unless a test shrinks it. */
141 readonly inFlightBytes?: number | undefined;
142 } = {},
87143 ): Promise<Restored> {
88144 if (!/^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}-[0-9]{2}-[0-9]{2}Z$/.test(name))
89145 throw new Error(`not a snapshot name: ${name}`);
@@ -121,23 +177,16 @@ export async function restoreSnapshot(
121177
122178 let files = 0;
123179 let repos = 0;
124 const bundles: { owner: string; name: string; file: string }[] = [];
125 for (const entry of manifest.files) {
126 const body = await store.get(prefix + entry.path);
127 if (body === undefined)
128 throw new Error(`snapshot ${name} is missing ${entry.path}`);
129 if (body.byteLength !== entry.bytes)
130 throw new Error(
131 `snapshot ${name}: ${entry.path} is ${body.byteLength} bytes, not ${entry.bytes}`,
132 );
133 if (entry.path === WRAPPED_KEY) continue;
180 const bundles: { at: number; owner: string; name: string; file: string }[] = [];
181 await fetchAll(store, prefix, manifest.files, async (entry, body, at) => {
182 if (entry.path === WRAPPED_KEY) return;
134183 // A part waits for the rest of its file.
135184 const part = partOf.get(entry.path);
136185 if (part !== undefined) {
137186 const got = gathered.get(part.file.path) ?? [];
138187 got[part.n] = body;
139188 gathered.set(part.file.path, got);
140 continue;
189 return;
141190 }
142191 // A day's run logs in one object come back as the logs they were.
143192 if (isRunBundle(entry.path)) {
@@ -147,7 +196,7 @@ export async function restoreSnapshot(
147196 await Bun.write(out, run.body);
148197 files++;
149198 }
150 continue;
199 return;
151200 }
152201 const bundle = entry.path.match(/^repos\/([^/]+)\/([^/]+)\.bundle$/);
153202 // A bundle is not a file of the forge's; it becomes a repository below.
@@ -162,8 +211,10 @@ export async function restoreSnapshot(
162211 chmodSync(target, 0o600);
163212 files++;
164213 if (bundle)
165 bundles.push({ owner: bundle[1]!, name: bundle[2]!, file: target });
166 }
214 bundles.push({ at, owner: bundle[1]!, name: bundle[2]!, file: target });
junipercrew

The trailing name, maxBytes arguments after a 40-line inline callback are hard to read. Name the callback (landed) or pass an options object.
215 }, name, options.inFlightBytes ?? IN_FLIGHT_BYTES);
216 // Reads land in any order; repositories come back in the manifest's.
217 bundles.sort((a, b) => a.at - b.at);
167218 for (const file of packed) {
168219 const got = gathered.get(file.path) ?? [];
169220 const parts = partPaths(file).map((path, n) => {
test/snapshot-drill.test.ts +5
@@ -166,6 +166,11 @@ describe("the restore drill", () => {
166166 expect(scratchLeft()).toEqual([]);
167167 });
168168
169 test("a step that times out says its budget in whole milliseconds", async () => {
170 const r = await runDrill(deps({ timeoutMs: 1500.75, restore: () => new Promise(() => {}) }));
171 expect(r.reason).toMatch(/^restore timed out: no answer in \d+ ms$/);
172 });
173
169174 test("an empty store is red at fetch", async () => {
170175 const r = await runDrill(
171176 deps({ store: dirStore(join(root, "nothing")) }),
test/snapshot-restore-parallel.test.ts added +143
@@ -0,0 +1,143 @@
1 /**
2 * A restore reads its snapshot several objects at a time (#2539). Read one
3 * after another, the drill of a snapshot with thousands of objects spent its
4 * whole ten minutes in the restore step; the reads now share the upload's
5 * limits, never more than `IN_FLIGHT` at once, and a missing or short file
6 * still refuses the restore by name.
7 */
8
9 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
10 import { mkdtempSync, readFileSync } from "node:fs";
11 import { rm } from "node:fs/promises";
12 import { tmpdir } from "node:os";
13 import { join } from "node:path";
14 import { restoreSnapshot } from "../src/snapshot/restore.ts";
15 import { IN_FLIGHT } from "../src/snapshot/snapshot.ts";
16 import { dirStore, type Store } from "../src/snapshot/store.ts";
17
18 const NAME = "2026-10-14T08-00-00Z";
19 const prefix = `snapshots/${NAME}/`;
20 const FILES = 40;
21
22 /**
23 * A store whose gets take `delay(key)` ms and which keeps the high-water
24 * marks of gets, and of bytes, open at once. The manifest is read alone
25 * before any file, so it is not counted.
26 */
27 function slow(
28 inner: Store,
29 delay: number | ((key: string) => number),
30 ): Store & { most: () => number; mostBytes: () => number } {
31 let open = 0;
32 let most = 0;
33 let openBytes = 0;
34 let mostBytes = 0;
35 return {
36 ...inner,
37 async get(key) {
38 const body = await inner.get(key);
39 if (key.endsWith("manifest.json")) return body;
40 const bytes = body?.byteLength ?? 0;
41 open++;
42 openBytes += bytes;
43 most = Math.max(most, open);
44 mostBytes = Math.max(mostBytes, openBytes);
45 try {
46 await Bun.sleep(typeof delay === "number" ? delay : delay(key));
47 return body;
48 } finally {
49 open--;
50 openBytes -= bytes;
51 }
52 },
53 most: () => most,
54 mostBytes: () => mostBytes,
55 };
56 }
57
58 const manifestOf = (files: { path: string; bytes: number }[]): string =>
59 JSON.stringify({
60 name: NAME,
61 startedAt: "2026-10-14T08:00:00.000Z",
62 finishedAt: "2026-10-14T08:01:00.000Z",
63 version: "t",
64 repos: 0,
65 apps: 1,
66 files,
67 bytes: files.reduce((s, f) => s + f.bytes, 0),
68 // An old snapshot that carried its key raw: nothing to look for sealed.
69 key: "raw",
70 });
71
72 async function seed(store: Store, skip?: string): Promise<void> {
73 const files: { path: string; bytes: number }[] = [];
74 for (let n = 0; n < FILES; n++) {
75 const path = `apps/a/file-${n}.txt`;
76 const body = `file ${n}\n`;
77 if (path !== skip) await store.put(prefix + path, body);
78 files.push({ path, bytes: Buffer.byteLength(body) });
79 }
80 await store.put(`${prefix}manifest.json`, manifestOf(files));
81 }
82
83 describe("a restore reads in parallel", () => {
84 let root: string;
85 beforeAll(() => {
86 root = mkdtempSync(join(tmpdir(), "forge-restore-parallel-"));
87 });
88 afterAll(async () => {
89 await rm(root, { recursive: true, force: true });
90 });
91
92 test("several gets are in flight at once, never more than IN_FLIGHT, and every file lands", async () => {
93 const store = slow(dirStore(join(root, "store-1")), 20);
94 await seed(store);
95 const dataDir = join(root, "data-1");
96 const restored = await restoreSnapshot(store, NAME, dataDir);
97 expect(restored.files).toBe(FILES);
98 expect(store.most()).toBeGreaterThan(1);
99 expect(store.most()).toBeLessThanOrEqual(IN_FLIGHT);
100 for (let n = 0; n < FILES; n++)
101 expect(readFileSync(join(dataDir, "apps/a", `file-${n}.txt`), "utf8")).toBe(`file ${n}\n`);
102 });
103
104 test("a file the store does not hold still refuses the restore, by name", async () => {
105 const store = slow(dirStore(join(root, "store-2")), 5);
106 await seed(store, "apps/a/file-17.txt");
107 await expect(restoreSnapshot(store, NAME, join(root, "data-2"))).rejects.toThrow(
108 `snapshot ${NAME} is missing apps/a/file-17.txt`,
109 );
110 });
111
112 test("past IN_FLIGHT_BYTES the next read waits, though fewer than IN_FLIGHT are open", async () => {
113 const store = slow(dirStore(join(root, "store-3")), 20);
114 const files: { path: string; bytes: number }[] = [];
115 for (let n = 0; n < 6; n++) {
116 const path = `apps/a/big-${n}.bin`;
117 await store.put(prefix + path, new Uint8Array(100).fill(n));
118 files.push({ path, bytes: 100 });
119 }
120 await store.put(`${prefix}manifest.json`, manifestOf(files));
121 const restored = await restoreSnapshot(store, NAME, join(root, "data-3"), { inFlightBytes: 250 });
122 expect(restored.files).toBe(6);
123 expect(store.mostBytes()).toBeLessThanOrEqual(250);
124 expect(store.most()).toBe(2);
125 });
126
127 test("repositories come back in the manifest's order when later reads land first", async () => {
128 // Neither bundle clones, so the error names the first one tried.
129 const order = ["alice/first", "alice/second", "alice/third"];
130 const files = order.map((r) => ({ path: `repos/${r}.bundle`, bytes: 13 }));
131 const inner = dirStore(join(root, "store-4"));
132 for (const f of files) await inner.put(prefix + f.path, "not a bundle\n");
133 await inner.put(`${prefix}manifest.json`, manifestOf(files));
134 // The first entry answers last, the last first.
135 const store = slow(inner, (key) => {
136 const at = files.findIndex((f) => prefix + f.path === key);
137 return at < 0 ? 0 : (files.length - at) * 30;
138 });
139 await expect(restoreSnapshot(store, NAME, join(root, "data-4"))).rejects.toThrow(
140 "could not restore alice/first",
141 );
142 });
143 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs