1// fuzzes `cache.read` against a fake Source of Truth and a `node:fs` that
2// tracks descriptor ownership and injects faults. every delivered byte is
3// compared against the known file contents, and at each quiescent point the
4// locks, descriptors, and cache directory must agree with the Lru.
5//
6// FUZZ_SEED, FUZZ_RUNS, and FUZZ_OPS reproduce a failure or run a soak, and
7// FUZZ_STATS prints what each run exercised.
8const h = vi.hoisted(() => ({
9 lrus: [] as unknown[],
10 fs: {
11 /** descriptors opened through the mock and not yet closed */
12 open: new Map<number, string>(),
13 violations: [] as string[],
14 random: Math.random as () => number,
15 lag: 0,
16 shortIo: 0,
17 readError: 0,
18 writeError: 0,
19 onFault: () => {},
20 },
21}));
22
23vi.mock("@clo/lib/Lru", async (importOriginal) => {
24 const real = await importOriginal<typeof import("@clo/lib/Lru")>();
25 class Tracked<K, V> extends real.Lru<K, V> {
26 constructor(opts: ConstructorParameters<typeof real.Lru<K, V>>[0]) {
27 super(opts);
28 h.lrus.push(this);
29 }
30 }
31 return { ...real, Lru: Tracked };
32});
33
34vi.mock("node:fs", async (importOriginal) => {
35 const real = await importOriginal<typeof import("node:fs")>();
36 const s = h.fs;
37 const roll = (chance: number) => chance > 0 && s.random() < chance;
38 const later = (fn: () => void) => {
39 if (roll(s.lag)) setTimeout(fn, Math.floor(s.random() * 3));
40 else fn();
41 };
42 const injected = (code: string, syscall: string) => {
43 s.onFault();
44 return Object.assign(new Error(`${code}: injected, ${syscall}`), { code, syscall });
45 };
46 type IoCallback = (err: Error | null, bytes: number, buf: Buffer) => void;
47 type Io = (
48 fd: number,
49 buf: Buffer,
50 off: number,
51 len: number,
52 pos: number,
53 cb: IoCallback,
54 ) => void;
55 const io =
56 (name: string, real: Io, errorChance: () => number, errorCode: string): Io => (fd, buf, off, len, pos, cb) => {
57 if (!s.open.has(fd)) s.violations.push(`${name} on fd ${fd} which is not open`);
58 // a real syscall never calls back synchronously, and a write stream that sees one writes past the failure
59 if (roll(errorChance())) return void setImmediate(() => later(() => cb(injected(errorCode, name), 0, buf)));
60 const partial = len > 1 && roll(s.shortIo) ? 1 + Math.floor(s.random() * (len - 1)) : len;
61 real(fd, buf, off, partial, pos, (err, bytes) => later(() => cb(err, bytes, buf)));
62 };
63 const overrides = {
64 open(path: string, flags: string, cb: (err: Error | null, fd: number) => void) {
65 real.open(path, flags, (err, fd) => {
66 if (!err) s.open.set(fd, path);
67 cb(err, fd);
68 });
69 },
70 close(fd: number, cb: (err: Error | null) => void) {
71 if (!s.open.delete(fd)) s.violations.push(`close of fd ${fd} which is not open`);
72 real.close(fd, cb);
73 },
74 closeSync(fd: number) {
75 if (!s.open.delete(fd)) s.violations.push(`close of fd ${fd} which is not open`);
76 real.closeSync(fd);
77 },
78 read: io("read", real.read, () => s.readError, "EIO"),
79 write: io("write", real.write, () => s.writeError, "ENOSPC"),
80 };
81 return { ...real, ...overrides, default: { ...real, ...overrides } };
82});
83
84const seed = Number(process.env.FUZZ_SEED ?? 1);
85const runs = Number(process.env.FUZZ_RUNS ?? 6);
86const opCount = Number(process.env.FUZZ_OPS ?? 250);
87const token = "fuzz-token";
88
89function prng(seed: number) {
90 const next = () => {
91 let t = (seed += 0x6D2B79F5);
92 t = Math.imul(t ^ (t >>> 15), t | 1);
93 t ^= t + Math.imul(t ^ (t >>> 7), t | 61);
94 return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
95 };
96 return Object.assign(next, {
97 int: (max: number) => Math.floor(next() * max),
98 roll: (chance: number) => chance > 0 && next() < chance,
99 pick: <T>(items: readonly T[]) => UNWRAP(items[Math.floor(next() * items.length)]),
100 });
101}
102type Random = ReturnType<typeof prng>;
103
104interface FakeFile {
105 file: MediaFile;
106 derived: string | null;
107 url: string;
108 hash: string;
109 data: Buffer;
110}
111
112function fakeFiles(random: Random): FakeFile[] {
113 const sizes = [0, 1, 2, 4096, 65535, 65536, 65537, 131072, 200_001, 300_000];
114 const files = sizes.map((size, i): FakeFile => {
115 const data = Buffer.alloc(size);
116 for (let b = 0; b < size; b++) data[b] = random.int(256);
117 const path = `/2026/friends/${i}/new home [rev ${i}].mp3`;
118 return {
119 file: { path } as MediaFile,
120 derived: null,
121 url: `/file${path}`,
122 hash: crypto.createHash("sha1").update(data).digest("hex"),
123 data,
124 };
125 });
126 // a derived asset, and a second path with the contents of another file
127 const source = UNWRAP(files[4]);
128 const twin = UNWRAP(files[7]);
129 const preview = Buffer.from(source.data.subarray(100, 30_100));
130 files.push({
131 ...source,
132 derived: `${source.hash}/preview.webp`,
133 url: `${source.url}:/preview.webp`,
134 hash: crypto.createHash("sha1").update(preview).digest("hex"),
135 data: preview,
136 }, {
137 ...twin,
138 file: { path: "/2026/copy of new home.mp3" } as MediaFile,
139 url: "/file/2026/copy of new home.mp3",
140 });
141 return files;
142}
143
144interface Chaos {
145 /** Source of Truth responds 500 */
146 status: number;
147 /** Source of Truth resets the connection mid-body */
148 abort: number;
149 /** Source of Truth serves a file shorter than the database says */
150 truncated: number;
151 /** a filesystem callback arrives a few milliseconds late */
152 lag: number;
153 shortIo: number;
154 readError: number;
155 writeError: number;
156}
157const calm: Chaos = {
158 status: 0,
159 abort: 0,
160 truncated: 0,
161 lag: 0.3,
162 shortIo: 0.1,
163 readError: 0,
164 writeError: 0,
165};
166const stormy: Chaos = {
167 ...calm,
168 status: 0.03,
169 abort: 0.05,
170 truncated: 0.02,
171 readError: 0.002,
172 writeError: 0.003,
173};
174
175interface DiskFile {
176 size: number;
177 written: { value: number };
178}
179
180class World {
181 violations: string[] = [];
182 /** the cache's own log, shown alongside a failure */
183 log: string[] = [];
184 /** what the run exercised, printed with FUZZ_STATS */
185 stats: Record<string, number> = {};
186 lastFault = -Infinity;
187 files: FakeFile[];
188 random: Random;
189 sotRandom: Random;
190 dir: string;
191 cacheDir: string;
192 rawDir: string | null;
193 server!: http.Server;
194 cache!: typeof import("./cache.ts");
195 diskCache!: Lru<Path, DiskFile>;
196 fdsCache!: Lru<Path, number>;
197
198 constructor(seed: number, public chaos: Chaos, local: boolean) {
199 this.random = prng(seed);
200 this.sotRandom = prng(seed ^ 0x5eed);
201 this.files = fakeFiles(prng(seed ^ 0xf11e));
202 this.dir = fs.mkdtempSync(path.join(os.tmpdir(), "clover-cache-fuzz-"));
203 this.cacheDir = path.join(this.dir, ".clover/file-viewer");
204 this.rawDir = local ? path.join(this.dir, "raw") : null;
205
206 const fsRandom = prng(seed ^ 0xd15c);
207 Object.assign(h.fs, chaos, { random: fsRandom, onFault: () => this.fault() });
208 h.fs.violations = this.violations;
209 }
210
211 fault() {
212 this.lastFault = performance.now();
213 }
214
215 count(what: string, by = 1) {
216 this.stats[what] = (this.stats[what] ?? 0) + by;
217 }
218
219 async start() {
220 this.server = http.createServer((req, res) => this.respond(req, res));
221 await new Promise<void>((resolve) => this.server.listen(0, "127.0.0.1", resolve));
222 const { port } = this.server.address() as net.AddressInfo;
223
224 // only some files are on the local share, the rest fall back to the network
225 if (this.rawDir) {
226 for (const [i, { file, derived, data }] of this.files.entries()) {
227 if (i % 3 === 0) continue;
228 const local = derived
229 ? path.join(this.dir, "derived", derived)
230 : path.join(this.rawDir, file.path);
231 fs.mkdirSync(path.dirname(local), { recursive: true });
232 fs.writeFileSync(local, data);
233 }
234 }
235
236 process.env.CLOVER_SOT_URL = `http://127.0.0.1:${port}`;
237 process.env.CLOVER_SOT_KEY = token;
238 process.env.CLOVER_FILE_RAW = this.rawDir ?? path.join(this.dir, "missing");
239 process.env.CLOVER_FILE_DERIVED = path.join(this.dir, "derived");
240 process.env.CLOVER_DB = path.join(this.dir, "db");
241 process.chdir(this.dir);
242 await this.boot();
243 }
244
245 /** load a fresh copy of the cache module, as a process restart would */
246 async boot() {
247 for (const fd of h.fs.open.keys()) realFs.closeSync(fd);
248 h.fs.open.clear();
249
250 vi.resetModules();
251 const log = await import("@clo/lib/log");
252 log.replaceGlobalMessageDestination((message) => {
253 this.log.push(log.formatAnsiMessage(message, false).trimEnd());
254 });
255 this.cache = await import("./cache.ts");
256 [this.diskCache, this.fdsCache] = h.lrus.slice(-2) as [Lru<Path, DiskFile>, Lru<Path, number>];
257 this.diskCache.capacity = 700_000;
258 this.fdsCache.capacity = 3;
259 }
260
261 async stop() {
262 await new Promise((resolve) => this.server.close(resolve));
263 this.server.closeAllConnections();
264 for (const fd of h.fs.open.keys()) realFs.closeSync(fd);
265 h.fs.open.clear();
266 fs.rmSync(this.dir, { recursive: true, force: true });
267 }
268
269 respond(req: http.IncomingMessage, res: http.ServerResponse) {
270 const { sotRandom: random, chaos } = this;
271 const found = this.files.find((f) => f.url === decodeURI(UNWRAP(req.url)));
272 if (!found) return void res.writeHead(404).end("not found");
273 if (req.headers.authorization !== token) {
274 this.violations.push(`request for ${found.url} did not carry the token`);
275 }
276 if (random.roll(chaos.status)) {
277 this.fault();
278 return void res.writeHead(500).end("injected");
279 }
280
281 let { data } = found;
282 let status = 200;
283 const headers: Record<string, string> = {};
284 // a server is free to ignore `Range`, so do so half of the time
285 const range = /^bytes=(\d+)-(\d+)$/.exec(req.headers.range ?? "");
286 if (range && random.roll(0.5)) {
287 status = 206;
288 headers["Content-Range"] = `bytes ${range[1]}-${range[2]}/${data.length}`;
289 data = data.subarray(Number(range[1]), Number(range[2]) + 1);
290 }
291 if (data.length > 0 && random.roll(chaos.truncated)) {
292 this.fault();
293 data = data.subarray(0, random.int(data.length));
294 }
295 const abortAt = random.roll(chaos.abort) ? random.int(data.length + 1) : Infinity;
296 headers["Content-Length"] = String(data.length);
297 res.writeHead(status, headers);
298
299 let sent = 0;
300 const pump = () => {
301 if (res.destroyed) return;
302 if (sent >= abortAt) {
303 this.fault();
304 return void res.destroy();
305 }
306 if (sent >= data.length) return void res.end();
307 const end = Math.min(data.length, abortAt, sent + 1 + random.int(90_000));
308 res.write(data.subarray(sent, sent = end));
309 if (random.roll(0.3)) setTimeout(pump, random.int(4));
310 else setImmediate(pump);
311 };
312 pump();
313 }
314
315 /** a request for the whole of `target` */
316 async request({ file, derived, hash, data }: FakeFile) {
317 const options = { file, derived, hash, size: data.length, start: 0, end: Infinity };
318 return UNWRAP(await this.cache.read(options));
319 }
320
321 async body(target: FakeFile) {
322 const { stream } = await this.request(target);
323 return Buffer.from(await new Response(stream as BodyInit).arrayBuffer());
324 }
325
326 /** one request, consumed the way `backend.ts` hands it to a `Response` */
327 async read(label: string) {
328 const { random } = this;
329 const target = random.pick(this.files);
330 const { data } = target;
331 let start = 0, end = Infinity;
332 if (data.length > 0 && random.roll(0.6)) {
333 start = random.int(data.length);
334 end = start + random.int(data.length - start);
335 }
336 const expected = data.subarray(start, end + 1);
337 const mode = random.pick(["drain", "drain", "slow", "cancel"] as const);
338 const cancelAt = mode === "cancel" ? random.int(expected.length + 1) : Infinity;
339 const what = `${label} ${target.url} [${start}, ${end}] ${mode}`;
340 const startedAt = performance.now();
341
342 let received = 0;
343 let src = "";
344 try {
345 const found = UNWRAP(
346 await this.cache.read({
347 file: target.file,
348 derived: target.derived,
349 size: data.length,
350 hash: target.hash,
351 start,
352 end,
353 noWrite: random.roll(0.1),
354 }),
355 );
356 src = found.src;
357 this.count(`src ${src}`);
358 if (found.size !== data.length) {
359 this.violations.push(`${what}: size ${found.size}, expected ${data.length}`);
360 }
361 const reader = UNWRAP(new Response(found.stream as BodyInit).body).getReader();
362 while (true) {
363 if (received >= cancelAt) {
364 this.count("cancelled");
365 return void await reader.cancel();
366 }
367 const { done, value } = await reader.read();
368 if (done) break;
369 const want = expected.subarray(received, received + value.length);
370 if (want.length !== value.length || !want.equals(value)) {
371 this.violations.push(`${what} (${src}): wrong bytes at offset ${received}`);
372 return void await reader.cancel();
373 }
374 received += value.length;
375 this.count("bytes verified", value.length);
376 if (mode === "slow") await delay(random.int(4));
377 }
378 if (received !== expected.length) {
379 this.violations.push(
380 `${what} (${src}): ended cleanly after ${received} of ${expected.length} bytes`,
381 );
382 }
383 } catch (err) {
384 // an injected fault may fail a request, but never corrupt one
385 if (this.lastFault >= startedAt - 500) return this.count("failed by a fault");
386 this.violations.push(`${what} (${src}): ${util.inspect(err)}`);
387 }
388 }
389
390 /** wait for the cache to go idle, then check it against the disk */
391 async settle(label: string) {
392 const { diskCache, fdsCache } = this;
393 const idle = () =>
394 diskCache.unlockedCapacity === diskCache.capacity
395 && h.fs.open.size === fdsCache.size
396 && [...fdsCache.values()].every((fd) => h.fs.open.has(fd));
397 for (let waited = 0; !idle() && waited < 4000; waited += 10) await delay(10);
398 if (!idle()) {
399 const locked = diskCache.capacity - diskCache.unlockedCapacity;
400 const leaked = [...h.fs.open].filter(([fd]) => !new Set(fdsCache.values()).has(fd));
401 this.violations.push(
402 `${label}: never went idle, ${locked} bytes locked, `
403 + `leaked descriptors ${util.inspect(leaked)}, `
404 + `descriptor cache ${util.inspect([...fdsCache.values()])}`,
405 );
406 return;
407 }
408
409 let used = 0;
410 const names = new Set<string>();
411 for (const [file, entry] of diskCache) {
412 used += entry.size;
413 if (names.has(file.base)) {
414 this.violations.push(`${label}: ${file.base} has two entries`);
415 }
416 names.add(file.base);
417 const known = this.files.find((f) => f.hash === file.base);
418 const onDisk = fs.existsSync(file.toString()) ? fs.readFileSync(file.toString()) : null;
419 if (entry.written.value !== entry.size) {
420 this.violations.push(`${label}: ${file.base} is idle but still pending`);
421 } else if (!onDisk) {
422 this.violations.push(`${label}: ${file.base} is cached but not on disk`);
423 } else if (!known || !known.data.equals(onDisk)) {
424 this.violations.push(`${label}: ${file.base} is corrupt on disk`);
425 }
426 }
427 if (used !== diskCache.used || used > diskCache.capacity) {
428 this.violations.push(`${label}: holds ${used} bytes, accounted ${diskCache.used}`);
429 }
430 for (const name of fs.readdirSync(this.cacheDir)) {
431 if (name !== "state.json" && !names.has(name)) {
432 this.violations.push(`${label}: ${name} is on disk but not in the cache`);
433 }
434 }
435 }
436
437 /** restart the process, optionally after damaging the cache directory */
438 async restart(label: string) {
439 const { random, cacheDir } = this;
440 const before = new Set([...this.diskCache.keys()].map((file) => file.base));
441 const damage = random.pick(["none", "none", "truncate", "stray", "state", "nostate"] as const);
442 const victim = before.size > 0 ? random.pick([...before]) : null;
443 const state = path.join(cacheDir, "state.json");
444 if (damage === "truncate" && victim) {
445 const { size } = fs.statSync(path.join(cacheDir, victim));
446 fs.truncateSync(path.join(cacheDir, victim), random.int(size));
447 } else if (damage === "stray") {
448 fs.writeFileSync(path.join(cacheDir, "0".repeat(40)), "stray");
449 } else if (damage === "state" && fs.existsSync(state)) {
450 fs.truncateSync(state, random.int(fs.statSync(state).size));
451 } else if (damage === "nostate") {
452 fs.rmSync(state, { force: true });
453 }
454
455 this.count(`restart ${damage}`);
456 await this.boot();
457 const after = new Set([...this.diskCache.keys()].map((file) => file.base));
458 if (damage === "none" && before.size > 0 && before.difference(after).size > 0) {
459 this.violations.push(
460 `${label}: restart lost ${before.difference(after).size} of ${before.size} entries`,
461 );
462 }
463 }
464
465 check() {
466 // the log module owns `console` here, so write to the descriptor
467 if (process.env.FUZZ_STATS) fs.writeSync(2, util.inspect(this.stats) + "\n");
468 if (this.violations.length === 0) return;
469 const shown = this.violations.slice(0, 8).join("\n ");
470 const log = this.log.slice(-40).join("\n ");
471 assert.fail(`${this.violations.length} violations, first:\n ${shown}\nlog:\n ${log}`);
472 }
473}
474
475async function withWorld<T>(
476 seed: number,
477 chaos: Chaos,
478 local: boolean,
479 fn: (world: World) => Promise<T>,
480) {
481 const cwd = process.cwd();
482 const world = new World(seed, chaos, local);
483 try {
484 await world.start();
485 return await fn(world);
486 } finally {
487 process.chdir(cwd);
488 await world.stop();
489 }
490}
491
492const quiet: Chaos = { ...calm, lag: 0, shortIo: 0 };
493
494describe("issue 106", () => {
495 test("a descriptor is not closed while a stream reads it", () =>
496 withWorld(106, quiet, false, async (world) => {
497 const big = UNWRAP(world.files.at(9));
498 const others = world.files.filter((f) => f !== big && f.data.length < 70_000);
499 world.diskCache.capacity = 8_000_000;
500
501 // once to fill the cache, once more to park the descriptor
502 await world.body(big);
503 await world.settle("fill");
504 await world.body(big);
505 await world.settle("park");
506
507 // a paused player holds this response open while a listing loads
508 const held = await world.request(big);
509 for (const other of others) await world.body(other);
510 for (const other of others) await world.body(other);
511
512 const body = Buffer.from(await new Response(held.stream as BodyInit).arrayBuffer());
513 assert.ok(body.equals(big.data), "held response was served the wrong bytes");
514 await world.settle("end");
515 world.check();
516 }));
517
518 test("a descriptor is not parked while another request is opening it", () =>
519 withWorld(106, quiet, false, async (world) => {
520 const big = UNWRAP(world.files.at(9));
521 const others = world.files.filter((f) => f !== big && f.data.length < 70_000);
522 world.diskCache.capacity = 8_000_000;
523 await world.body(big);
524 await world.settle("fill");
525
526 // once its buffers fill, the first stream has no read in flight, so
527 // cancelling it releases the descriptor before the second request resumes
528 const first = (await world.request(big)).stream.getReader();
529 await delay(50);
530 const second = world.request(big);
531 await first.cancel();
532
533 const held = await second;
534 for (const other of others) await world.body(other);
535 const body = Buffer.from(await new Response(held.stream as BodyInit).arrayBuffer());
536 assert.ok(body.equals(big.data), "held response was served the wrong bytes");
537 await world.settle("end");
538 world.check();
539 }));
540
541 test("a range that starts past its end leaks nothing", () =>
542 withWorld(106, quiet, false, async (world) => {
543 const { file, hash, data } = UNWRAP(world.files.at(8));
544 await world.body(UNWRAP(world.files.at(8)));
545 await world.settle("fill");
546
547 const backwards = { file, derived: null, hash, size: data.length, start: 46226599, end: 25705638 };
548 await assert.rejects(world.cache.read(backwards));
549 await world.settle("end");
550 world.check();
551 }));
552});
553
554describe("fuzz", () => {
555 const cases = Array.from({ length: runs }, (_, i) => ({
556 seed: seed + i,
557 chaos: (seed + i) % 2 === 0 ? "calm" : "stormy",
558 local: (seed + i) % 3 === 0,
559 }));
560 test.each(cases)(
561 "seed $seed, $chaos, local share $local",
562 ({ seed, chaos, local }) =>
563 withWorld(seed, chaos === "calm" ? calm : stormy, local, async (world) => {
564 const { random } = world;
565 const inflight = new Set<Promise<void>>();
566 for (let i = 0; i < opCount && world.violations.length === 0; i++) {
567 const roll = random();
568 if (roll < 0.03) {
569 await Promise.all(inflight);
570 await world.settle(`#${i} settle`);
571 if (world.violations.length === 0 && random.roll(0.5)) {
572 await world.restart(`#${i} restart`);
573 }
574 continue;
575 }
576 const op = world.read(`#${i}`).finally(() => inflight.delete(op));
577 inflight.add(op);
578 if (inflight.size >= 12 || random.roll(0.25)) await Promise.race(inflight);
579 else if (random.roll(0.4)) await delay(random.int(3));
580 }
581 await Promise.all(inflight);
582 await world.settle("end");
583 world.check();
584 }),
585 120_000,
586 );
587});
588
589const realFs = await vi.importActual<typeof import("node:fs")>("node:fs");
590
591import type { Path } from "#sitegen/path";
592import { UNWRAP } from "@clo/lib/assert";
593import { delay } from "@clo/lib/async";
594import type { Lru } from "@clo/lib/Lru";
595import { strict as assert } from "node:assert";
596import * as crypto from "node:crypto";
597import * as fs from "node:fs";
598import * as http from "node:http";
599import type * as net from "node:net";
600import * as os from "node:os";
601import * as path from "node:path";
602import * as util from "node:util";
603import { describe, test, vi } from "vitest";
604import type { MediaFile } from "./models/media-files.ts";