Commit 27372f562b13b96c1bf165339923d5f402753428 | sitegen

author	clover caruso <git@paperclover.net> 2026-10-11 15:45:35-07:00
committer	clover caruso <git@paperclover.net> 2026-10-11 20:05:55-07:00
log	27372f562b13b96c1bf165339923d5f402753428
tree	617dee5a71d5e850efd832982fbdb7e41f920112
parent	4ceca6c30ef19c14cd4ba76a3cbd9e894f91fdc6
signature	Signed by SSH key SHA256:cOKiuRFOeSRxne6EWgHtdQQSlBxjOXm2hOCFnCdLQbQ (~clover)

fix(file-viewer): fix all the bugs

Closes #106

- EBADF: a descriptor reused from `fdsCache` stayed in that lru, so it was
  closed, or its number handed to another file, under a stream still reading
  it. Handles are counted before the first await and leave the lru while held.
- RangeError: `parseRangeHeader` never clamped or rejected ranges, and the
  throw leaked the entry lock. Unsatisfiable ranges are now a 416, and open
  and suffix ranges (`500-`, `-500`) parse instead of being ignored.
- `http.serve` piped bodies without cancelling them on disconnect or handling
  their errors: every aborted request leaked a descriptor and a lock, and any
  stream error was an uncaught exception.
- Every request, the first included, reads a download back from disk as it
  lands. This replaces the tee, where a slow first client throttled the
  download, and the bypass that sent pending requests upstream.
- A download that is cut short, stalls, or disagrees with the expected size
  drops its entry instead of staying pending forever.
- Uncached ranges were sent the whole file, since nothing implements
  `Clover-Start`. They use `Range`, sliced locally if the server ignores it.
- `Lru.delete` skipped the `delete` callback, leaving dropped entries on
  disk, and the revive compared base names to absolute paths, wiping the
  cache on every restart.
- Chunk reuse repeated bytes after a chunk at offset 0 and could hand out the
  uninitialized tail of a short read.
- `Path` deduplication broke when an old copy's finalizer ran after a newer
  copy registered, giving the cache two entries for one file.
- Source of Truth: requests sharing a descriptor shared its file offset and
  split the file between them. Reads are positioned and `Range` is supported.
- Ranges and `If-Range` on derived assets used the original file's size and
  hash.

Assisted-by: claude-opus-5.5

12 files changed, 1201 insertions(+), 379 deletions(-)

framework
- lib
  - path.test.ts
  - path.ts
lib
src
- file-viewer
- source-of-truth.test.ts
- source-of-truth.ts

framework/lib/path.test.ts created+22

...	...	@@ -0,0 +1,22 @@
	1	test("a path stays deduplicated after an older copy is collected", async () => {
	2	const raw = "/clover/path.test/deduplicated";
	3	(() => void new Path(raw))();
	4	// a WeakRef keeps its target alive until the task that created it ends
	5	await delay(0);
	6	collectGarbage();
	7
	8	// the first copy is collected, but its finalizer has not yet run
	9	const second = new Path(raw);
	10	await delay(10);
	11	assert.equal(new Path(raw), second);
	12	});
	13
	14	v8.setFlagsFromString("--expose-gc");
	15	const collectGarbage: () => void = vm.runInNewContext("gc");
	16
	17	import { delay } from "@clo/lib/async";
	18	import { strict as assert } from "node:assert";
	19	import * as v8 from "node:v8";
	20	import * as vm from "node:vm";
	21	import { test } from "vitest";
	22	import { Path } from "./path.ts";

framework/lib/path.ts+4-1

...	...	@@ -1,5 +1,8 @@
1	1	const unique = new Map<string, WeakRef<Path>>();
2		const final = new FinalizationRegistry((raw: string) => unique.delete(raw));
	2	// A finalizer can run after a newer `Path` has taken over the entry.
	3	const final = new FinalizationRegistry((raw: string) => {
	4	if (!unique.get(raw)?.deref()) unique.delete(raw);
	5	});
3	6
4	7	export class Path {
5	8	private raw: string;

lib/Lru.test.ts+22

...	...	@@ -148,6 +148,28 @@ test("revival", () => {
148	148	assert.equal(lru.get("d"), 4);
149	149	assert.equal(lru.get("e"), 5);
150	150	});
	151	test("delete callback sees every removal, evict only automatic ones", () => {
	152	const deleted: string[] = [];
	153	const evicted: string[] = [];
	154	const lru = new Lru<string, number>({
	155	capacity: 2,
	156	delete: (_, key) => deleted.push(key),
	157	evict: (_, key) => evicted.push(key),
	158	});
	159	lru.set("a", 1);
	160	lru.set("b", 2);
	161	lru.set("c", 3);
	162	assert.deepEqual([deleted, evicted], [["a"], ["a"]]);
	163
	164	assert.equal(lru.delete("b"), true);
	165	assert.equal(lru.delete("b"), false);
	166	assert.deepEqual([deleted, evicted], [["a", "b"], ["a"]]);
	167
	168	lru.set("d", 4);
	169	lru.clear();
	170	assert.deepEqual([deleted, evicted], [["a", "b", "c", "d"], ["a"]]);
	171	assert.equal(lru.used, 0);
	172	});
151	173
152	174	import assert from "node:assert";
153	175	import { test } from "vitest";

lib/Lru.ts+4

...	...	@@ -56,18 +56,22 @@ export class Lru<K, V> extends Map<K, V> {
56	56	}
57	57
58	58	override clear() {
	59	const removed = this.#deleteFn ? [...this] : [];
59	60	super.clear();
60	61	this.#order.clear();
61	62	this.#head = none;
62	63	this.#tail = none;
63	64	this.#used = 0;
	65	for (const [key, value] of removed) this.#deleteFn?.(value, key);
64	66	}
65	67	override delete(key: K): boolean {
66	68	const entry = this.#evict(key);
67	69	if (!entry) return false;
	70	const value = super.get(key) as V;
68	71	this.#used -= entry[0];
69	72	ASSERT(this.#order.delete(key));
70	73	ASSERT(super.delete(key));
	74	this.#deleteFn?.(value, key);
71	75	return true;
72	76	}
73	77	override get(key: K): V \| undefined {

lib/http.test.ts created+133

...	...	@@ -0,0 +1,133 @@
	1	describe("parseRangeHeader", () => {
	2	test("closed, open, and suffix ranges", () => {
	3	assert.deepEqual(http.parseRangeHeader("bytes=0-1", 100), [{ start: 0, end: 1 }]);
	4	assert.deepEqual(http.parseRangeHeader("bytes=10-", 100), [{ start: 10, end: 99 }]);
	5	assert.deepEqual(http.parseRangeHeader("bytes=-10", 100), [{ start: 90, end: 99 }]);
	6	assert.deepEqual(http.parseRangeHeader("bytes=-500", 100), [{ start: 0, end: 99 }]);
	7	assert.deepEqual(http.parseRangeHeader("bytes=50-500", 100), [{ start: 50, end: 99 }]);
	8	assert.deepEqual(http.parseRangeHeader("bytes=0-9, 20-29", 100), [
	9	{ start: 0, end: 9 },
	10	{ start: 20, end: 29 },
	11	]);
	12	});
	13
	14	test("ranges past the end are unsatisfiable", () => {
	15	// https://shale.paperclover.net/sitegen/issues/106
	16	assert.deepEqual(http.parseRangeHeader("bytes=46226599-", 25705639), []);
	17	assert.deepEqual(http.parseRangeHeader("bytes=100-200", 100), []);
	18	assert.deepEqual(http.parseRangeHeader("bytes=-0", 100), []);
	19	assert.deepEqual(http.parseRangeHeader("bytes=0-", 0), []);
	20	});
	21
	22	test("malformed headers are ignored", () => {
	23	for (
	24	const header of [
	25	"",
	26	"bytes",
	27	"bytes=",
	28	"bytes=-",
	29	"bytes=5-2",
	30	"bytes=a-b",
	31	"bytes=1.5-3",
	32	"bytes=0x10-",
	33	"bytes=--5",
	34	"bytes=1-2-3",
	35	"pages=0-1",
	36	"bytes=0-5,3-9",
	37	]
	38	) {
	39	assert.equal(http.parseRangeHeader(header, 100), null, header);
	40	}
	41	});
	42
	43	test("fuzz: every range it returns can be read", () => {
	44	const pieces = [
	45	"bytes",
	46	"=",
	47	"-",
	48	",",
	49	" ",
	50	"0",
	51	"1",
	52	"7",
	53	"99",
	54	"100",
	55	"101",
	56	"25705639",
	57	"9".repeat(30),
	58	"x",
	59	".",
	60	"e3",
	61	"-1",
	62	];
	63	let state = 106;
	64	const random = (max: number) => (state = (state * 48271) % 0x7fffffff) % max;
	65	for (let i = 0; i < 20_000; i++) {
	66	const size = [0, 1, 100, 25705639][random(4)]!;
	67	let header = random(4) === 0 ? "" : "bytes=";
	68	for (let n = random(7); n > 0; n--) header += pieces[random(pieces.length)];
	69
	70	for (const { start, end } of http.parseRangeHeader(header, size) ?? []) {
	71	assert.ok(
	72	Number.isInteger(start) && Number.isInteger(end)
	73	&& 0 <= start && start <= end && end < size,
	74	`${JSON.stringify(header)} of ${size} gave [${start}, ${end}]`,
	75	);
	76	}
	77	}
	78	});
	79	});
	80
	81	describe("serve", () => {
	82	test("a client that disconnects cancels the response body", async () => {
	83	const cancelled = Promise.withResolvers<void>();
	84	await using server = await http.serve({
	85	port: 0,
	86	respond: async () =>
	87	new Response(
	88	new ReadableStream({
	89	pull: (controller) => controller.enqueue(new Uint8Array(65536)),
	90	cancel: () => cancelled.resolve(),
	91	}),
	92	),
	93	});
	94
	95	const response = await new Promise<nodeHttp.IncomingMessage>((resolve) => nodeHttp.get(server.url, resolve));
	96	response.once("data", () => response.destroy());
	97	await cancelled.promise;
	98	});
	99
	100	test("a failing response body closes the connection", async () => {
	101	const failure = Object.assign(new Error("bad file descriptor"), { code: "EBADF" });
	102	const logged = vi.spyOn(console, "error").mockImplementation(() => {});
	103	let pulls = 0;
	104	await using server = await http.serve({
	105	port: 0,
	106	respond: async () =>
	107	new Response(
	108	new ReadableStream({
	109	pull(controller) {
	110	if (pulls++ === 0) controller.enqueue(new Uint8Array(16));
	111	else controller.error(failure);
	112	},
	113	}),
	114	{ headers: { "Content-Length": "65536" } },
	115	),
	116	});
	117
	118	await assert.rejects(async () => {
	119	const response = await new Promise<nodeHttp.IncomingMessage>((resolve, reject) =>
	120	nodeHttp.get(server.url, resolve).on("error", reject)
	121	);
	122	await consumers.buffer(response);
	123	}, { code: "ECONNRESET" });
	124	assert.deepEqual(logged.mock.calls, [[failure]]);
	125	logged.mockRestore();
	126	});
	127	});
	128
	129	import { strict as assert } from "node:assert";
	130	import * as nodeHttp from "node:http";
	131	import * as consumers from "node:stream/consumers";
	132	import { describe, test, vi } from "vitest";
	133	import * as http from "./http.ts";

lib/http.ts+22-10

...	...	@@ -67,7 +67,15 @@ export async function serve(
67	67	Array.from(response.headers.entries()),
68	68	);
69	69	if (response.body) {
70		stream.Readable.fromWeb(response.body as import("node:stream/web").ReadableStream).pipe(res);
	70	// unlike `pipe`, this cancels the body when the client goes away,
	71	// and a failing body closes the connection instead of the process.
	72	stream.pipeline(
	73	stream.Readable.fromWeb(response.body as import("node:stream/web").ReadableStream),
	74	res,
	75	(err) => {
	76	if (err && exception.code(err) !== "ERR_STREAM_PREMATURE_CLOSE") console.error(err);
	77	},
	78	);
71	79	} else {
72	80	res.end();
73	81	}
...	...	@@ -123,7 +131,9 @@ export async function serve(
123	131
124	132	/**
125	133	* parses the [`Range`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Guides/Range_requests)
126		* header, returning `null` if it is invalid. the `fileSize`
	134	* header for a resource of `fileSize` bytes. `null` means the header is
	135	* malformed and should be ignored. otherwise, the satisfiable ranges are
	136	* returned clamped to the resource, so an empty array is a `416` response.
127	137	*/
128	138	export function parseRangeHeader(
129	139	rangeHeader: string,
...	...	@@ -137,14 +147,16 @@ export function parseRangeHeader(
137	147	const rangeParts = ranges.split(",");
138	148
139	149	for (const range of rangeParts) {
140		const split = range.split("-");
141		const [start, end] = split;
142		if (!start \|\| !end) return null;
143		if (start === "" && end === "") return null;
144		result.push({
145		start: start === "" ? fileSize - +end : +start,
146		end: end === "" ? fileSize - 1 : +end,
147		});
	150	const match = /^\s(\d)-(\d)\s$/.exec(range);
	151	if (!match) return null;
	152	const [, first = "", last = ""] = match;
	153	if (first === "" && last === "") return null;
	154	if (first !== "" && last !== "" && +last < +first) return null;
	155
	156	// `-500` is the last 500 bytes, `500-` is everything after the first 500.
	157	const start = first === "" ? Math.max(0, fileSize - +last) : +first;
	158	const end = first === "" \|\| last === "" ? fileSize - 1 : Math.min(+last, fileSize - 1);
	159	if (start <= end) result.push({ start, end });
148	160	}
149	161
150	162	// validate that ranges do not intersect

lib/readme.changes.md+9

...	...	@@ -7,6 +7,15 @@
7	7	- `meta`'s auto-embed feature now sets all the twitter tags, not just thumbnail.
8	8	sorry i might have been dumb when doing the first version of this.
9	9
	10	### fixes
	11
	12	- `Lru`: `delete()` and `clear()` now call the `delete` callback, as documented.
	13	- `http.parseRangeHeader` accepts open and suffix ranges (`500-`, `-500`),
	14	clamps ranges to the resource, and returns an empty array when none of them
	15	can be satisfied.
	16	- `http.serve` cancels the response body when the client disconnects, and a
	17	failing body closes the connection instead of crashing the process.
	18
10	19	## v6
11	20
12	21	### breaking

src/file-viewer/backend.ts+14-10

...	...	@@ -151,7 +151,7 @@ app.get("/file/*", async (c, next) => {
151	151	let status = 200;
152	152	let rangeHeader = c.req.header("Range") ?? null;
153	153	const ifRangeHeader = c.req.header("If-Range");
154		if (ifRangeHeader && ifRangeOutdated(file, ifRangeHeader)) {
	154	if (ifRangeHeader && ifRangeOutdated(hash, date, ifRangeHeader)) {
155	155	// > If the condition is not fulfilled, the full resource is
156	156	// > sent back with a 200 OK status.
157	157	rangeHeader = null;
...	...	@@ -163,17 +163,21 @@ app.get("/file/*", async (c, next) => {
163	163	// Range requests
164	164	// https://developer.mozilla.org/en-US/docs/Web/HTTP/Range_requests
165	165	if (rangeHeader) {
166		const ranges = http.parseRangeHeader(rangeHeader, file.size);
	166	const ranges = http.parseRangeHeader(rangeHeader, size);
	167	if (ranges?.length === 0) {
	168	c.res = new Response(null, {
	169	status: 416,
	170	headers: { "Content-Range": `bytes */${size}` },
	171	});
	172	return;
	173	}
167	174	if (ranges?.[0] && ranges.length === 1) {
168	175	start = ranges[0].start;
169	176	end = ranges[0].end;
170	177
171	178	status = 206;
172		headers.set(
173		"Content-Range",
174		`bytes ${ranges[0].start}-${ranges[0].end}/${file.size}`,
175		);
176		headers.delete("Content-Length");
	179	headers.set("Content-Range", `bytes ${start}-${end}/${size}`);
	180	headers.set("Content-Length", String(end - start + 1));
177	181	} else {
178	182	// multi-ranges are not supported
179	183	// it could be done by calling `cache.read` multiple times.
...	...	@@ -209,13 +213,13 @@ app.get("/file/*", async (c, next) => {
209	213	});
210	214	});
211	215
212		function ifRangeOutdated(file: MediaFile, ifRangeHeader: string) {
	216	function ifRangeOutdated(hash: string, date: Date, ifRangeHeader: string) {
213	217	// etag
214	218	if (ifRangeHeader[0] === "\"") {
215		return ifRangeHeader.slice(1, -1) !== file.hash;
	219	return ifRangeHeader.slice(1, -1) !== hash;
216	220	}
217	221	// date
218		return new Date(ifRangeHeader) < file.date;
	222	return new Date(ifRangeHeader) < date;
219	223	}
220	224
221	225	function checkCotyledonCookie(c: Context) {

src/file-viewer/cache.test.ts created+603

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

src/file-viewer/cache.ts+304-357

...	...	@@ -53,45 +53,41 @@ const diskCache = new Lru<Path, DiskFile>({
53	53	// and held open in the `fdsCache` lru when no references.
54	54	const fdsCache = new Lru<Path, number>({
55	55	capacity: 64,
56		delete: (value) => void fsCallbacks.closeSync(value),
	56	evict: (fd) => fsCallbacks.closeSync(fd),
57	57	});
58		const fds = new Map<Path, Awaitable<RefCountFile>>();
59
60		const parallelRequests = new Map<string, Promise<http.IncomingMessage>>();
	58	const fds = new Map<Path, RefCountFile>();
61	59
62	60	// Revive disk cache state
63		revive: {
	61	{
	62	const revived = new Set([diskCacheState.base]);
64	63	try {
65	64	const state = diskCacheState
66		.readIfExistsSync<Lru.Serialized<string, number>>("json");
67		if (!state) break revive;
68		const missing = new Set(diskCacheRoot.readDirSync().map((c) => c.base));
69		missing.delete(diskCacheState.base);
	65	.readIfExistsSync<Lru.Serialized<string, number>>("json") ?? [];
70	66	diskCache.revive(state, {
71		map(k: string, v: number) {
72		const path = new Path(k);
	67	map(base: string, size: number) {
	68	const path = diskCacheRoot.join(base);
73	69	try {
74		const { size } = path.statSync();
75		if (size !== v) return null;
76		if (!missing.delete(k)) return null;
77		return [
78		path,
79		{ size, written: new async.Watch(size), memory: new Map() },
80		];
	70	if (path.statSync().size !== size) return null;
81	71	} catch {
82	72	return null;
83	73	}
	74	revived.add(base);
	75	return [
	76	path,
	77	{ size, written: new async.Watch(size), memory: new Map() },
	78	];
84	79	},
85	80	});
86		// Delete unreferenced files
87		for (const missingFile of missing) {
88		try {
89		diskCacheRoot.join(missingFile).deleteSync();
90		} catch {}
91		}
92	81	} catch (err) {
93	82	console.warn("Failed to restore file viewer cache", err);
94	83	}
	84	// Delete unreferenced files
	85	for (const file of diskCacheRoot.readDirSync()) {
	86	if (revived.has(file.base)) continue;
	87	try {
	88	file.deleteSync();
	89	} catch {}
	90	}
95	91	}
96	92
97	93	export interface Options {
...	...	@@ -101,7 +97,7 @@ export interface Options {
101	97	size: number;
102	98	hash: string;
103	99	start: number;
104		/** exclusive. ex: [10, 30] skips ten bytes then grabs 20 */
	100	/** inclusive, as in an HTTP range. ex: [10, 30] skips ten bytes then grabs 21 */
105	101	end: number;
106	102	noWrite?: boolean;
107	103	localOnly?: boolean;
...	...	@@ -112,8 +108,13 @@ export interface Options {
112	108	* `null` if a derived asset does not exist
113	109	*/
114	110	export async function read(
115		{ file, derived, hash, size, start, end, noWrite, localOnly }: Options,
	111	options: Options,
116	112	): Promise<{ stream: ReadableStream; src: Source; size: number } \| null> {
	113	const { file, derived, hash, size, start, end, noWrite, localOnly } = options;
	114	ASSERT(
	115	Number.isInteger(start) && start >= 0 && start <= end,
	116	() => `Invalid range [${start}, ${end}]`,
	117	);
117	118	const rel = derived ? `${file.path}$/${derived}` : file.path;
118	119
119	120	// Serve from local network share if it is mounted.
...	...	@@ -123,24 +124,19 @@ export async function read(
123	124	local: {
124	125	if (!localFile) break local;
125	126	if (process.env.CLOVER_SOT_FORCE) break local;
126		let size: number;
127	127	try {
128		({ size } = await localFile.stat());
	128	const { size } = await localFile.stat();
	129	const fileStream = await readStream(localFile, { start, end });
	130	console.info(`from local: ${rel} ` + localFile);
	131	return {
	132	stream: stream.Readable.toWeb(fileStream),
	133	size,
	134	src: "local",
	135	};
129	136	} catch (err) {
130		if (exception.code(err) === "ENOENT") {
131		console.warn(`missing local file: ${localFile}`);
132		break local;
133		}
134		throw err;
	137	if (exception.code(err) !== "ENOENT") throw err;
	138	console.warn(`missing local file: ${localFile}`);
135	139	}
136		const fileStream = await readStream(localFile, { start, end });
137		if (!fileStream) break local;
138		console.info(`from local: ${rel} ` + localFile);
139		return {
140		stream: stream.Readable.toWeb(fileStream),
141		size,
142		src: "local",
143		};
144	140	}
145	141
146	142	if (localOnly) return null;
...	...	@@ -151,200 +147,206 @@ export async function read(
151	147	// Disk cache
152	148	const diskPath = diskCacheRoot.join(hash);
153	149	let diskFile = diskCache.get(diskPath);
	150	let src: Source = "hit";
154	151	if (diskFile) {
155		const lock = diskCache.lock(diskPath);
156		const diskStream = await readStream(diskPath, {
157		ref: diskFile,
158		start: start,
159		end: end,
160		lock,
161		});
162		if (diskStream) {
163		const isPending = diskFile.written.value !== diskFile.size;
164
165		// TODO: fix this? i'm suspicious
166		if (isPending) {
167		diskStream.close();
168		return {
169		stream: await fetchAndSlice(rel, url, start, end),
170		size,
171		src: "contended",
172		};
173		}
174
175		console.info(`from ${isPending ? "pending " : ""}disk cache: ${rel}`);
	152	if (diskFile.written.value !== diskFile.size) src = "pending";
	153	console.info(`from ${src === "pending" ? "pending " : ""}disk cache: ${rel}`);
	154	} else {
	155	// Do not cache large assets
	156	if (size > 1_000_000_000) {
	157	console.info(`skipping disk cache due to large asset: ${rel}`);
176	158	return {
177		stream: stream.Readable.toWeb(diskStream),
178		src: isPending ? "pending" : "hit",
179		size: diskFile.size,
	159	stream: await fetchAndSlice(rel, url, size, start, end),
	160	size,
	161	src: "ignored",
	162	};
	163	}
	164	// Bail out of the disk cache if it is contended
	165	if (diskCache.unlockedCapacity < size) {
	166	console.info(`skipping disk cache due to contention: ${rel}`);
	167	return {
	168	stream: await fetchAndSlice(rel, url, size, start, end),
	169	size,
	170	src: "contended",
	171	};
	172	}
	173	// IPs which request too many cache misses are temporarily
	174	// banned from writing into the disk cache to avoid abuse.
	175	if (noWrite) {
	176	console.info(`skipping disk cache due to config: ${rel}`);
	177	return {
	178	stream: await fetchAndSlice(rel, url, size, start, end),
	179	size,
	180	src: "ratelimit",
180	181	};
181		} else {
182		diskCache.delete(diskPath);
183	182	}
184		}
185	183
186		// Do not cache large assets
187		if (size > 1_000_000_000) {
188		console.info(`skipping disk cache due to large asset: ${rel}`);
189		return {
190		stream: await fetchAndSlice(rel, url, start, end),
191		size,
192		src: "ignored",
193		};
	184	// Create a disk cache entry. Every request, this one included, reads the
	185	// file back as it lands so that a slow client never stalls the download.
	186	diskFile = { size, written: new async.Watch(0), memory: new Map() };
	187	diskCache.set(diskPath, diskFile);
	188	void download(rel, url, diskPath, diskFile);
	189	src = "miss";
194	190	}
195		// Bail out of the disk cache if it is contended
196		if (diskCache.unlockedCapacity < size) {
197		console.info(`skipping disk cache due to contention: ${rel}`);
198		return {
199		stream: await fetchAndSlice(rel, url, start, end),
200		size,
201		src: "contended",
202		};
203		}
204		// IPs which request too many cache misses are temporarily
205		// banned from writing into the disk cache to avoid abuse.
206		if (noWrite) {
207		console.info(`skipping disk cache due to config: ${rel}`);
	191
	192	const lock = diskCache.lock(diskPath);
	193	try {
	194	const diskStream = await readStream(diskPath, {
	195	ref: diskFile,
	196	start,
	197	end,
	198	lock,
	199	});
208	200	return {
209		stream: await fetchAndSlice(rel, url, start, end),
210		size,
211		src: "ratelimit",
	201	stream: stream.Readable.toWeb(diskStream),
	202	src,
	203	size: diskFile.size,
212	204	};
	205	} catch (err) {
	206	lock();
	207	if (src === "miss" \|\| exception.code(err) !== "ENOENT") throw err;
	208	// The file vanished from under its entry, so cache it again.
	209	if (diskCache.get(diskPath) === diskFile) diskCache.delete(diskPath);
	210	return read(options);
213	211	}
214
215		// Create a disk cache entry;
216		// Tee the stream to both the request and the filesystem.
217		diskFile = { size, written: new async.Watch(0), memory: new Map() };
218		diskCache.set(diskPath, diskFile);
219		const lock = diskCache.lock(diskPath);
220		return await fetchAndStore(
221		rel,
222		url,
223		size,
224		diskPath,
225		diskFile,
226		start,
227		end,
228		lock,
229		);
230	212	}
231	213
232		async function fetchAndStore(
	214	/** Fill a new disk cache entry. Readers follow along through `diskFile.written`. */
	215	async function download(
233	216	rel: string,
234	217	url: string,
235		size: number,
236	218	diskPath: Path,
237	219	diskFile: DiskFile,
238		start: number,
239		end: number,
240		lock: ts.Dispose,
241	220	) {
	221	const lock = diskCache.lock(diskPath);
	222	// Readers wait on this download, so a stalled connection has to fail it.
	223	const response = fetch(rel, url, { timeout: 30_000 });
242	224	try {
243		let response = await fetch(rel, url);
244		const { statusCode } = response;
245		if (statusCode !== 200) {
246		const text = await new Response(response as unknown as Blob).text();
247		const msg = `Source of Truth responded with status ${statusCode}: ${text}`;
248		console.error(msg);
249		throw new Error("Source of Truth is unavailable");
250		}
251		const writer = await writeStream(diskPath, response, diskFile, lock);
252		writer.on("error", () => {
253		diskCache.delete(diskPath);
254		diskFile.written.value = Infinity;
	225	// Opening only after the response arrives makes every reader sharing this
	226	// handle wait on the Source of Truth, and fail with it.
	227	const handle = await getHandle(diskPath, async () => {
	228	await response;
	229	return await openFile(diskPath.toString(), "w+");
255	230	});
256	231
257		return {
258		stream: sliceStream(response, start, end),
259		size,
260		src: "miss",
261		} as const;
262		} catch (e) {
	232	const { memory, written } = diskFile;
	233	const writer = fs.createWriteStream(diskPath.toString(), {
	234	start: 0,
	235	fd: handle.fd,
	236	fs: {
	237	close(_: number, cb: (er: Error \| null) => void) {
	238	handle.release();
	239	cb(null);
	240	},
	241	write(
	242	fd: number,
	243	data: Buffer,
	244	off: 0,
	245	size: number,
	246	pos: number \| undefined,
	247	cb: (er: Error \| null, bytesWritten: number, buffer: Buffer) => void,
	248	) {
	249	ASSERT(size === data.byteLength);
	250	ASSERT(pos != null);
	251	fsCallbacks.write(fd, data, off, size, pos, (er, bytes) => {
	252	if (!er && bytes) {
	253	remember(memory, pos, data);
	254	written.value += bytes;
	255	}
	256	cb(er, bytes, data);
	257	});
	258	},
	259	},
	260	});
	261	await streamPromises.pipeline(await response, writer);
	262
	263	if (written.value !== diskFile.size) {
	264	throw new Error(`Expected ${diskFile.size} bytes, received ${written.value}`);
	265	}
	266	writeDiskCache();
	267	} catch (err) {
	268	console.error(`Failed to cache ${rel}:`, err);
	269	response.then((opened) => opened.destroy(), () => {});
	270	if (diskCache.get(diskPath) === diskFile) diskCache.delete(diskPath);
	271	diskFile.written.value = Infinity;
	272	} finally {
263	273	lock();
264		throw e;
265	274	}
266	275	}
267	276
268		async function fetch(rel: string, url: string, headers?: {}) {
269		let promise = !headers ? parallelRequests.get(url) : null;
270		if (promise) {
271		console.info(`from other response: ${rel}`);
272		return await promise;
273		} else {
274		console.info(`cache miss: ${rel}`);
275		promise = httpGet(url, {
276		headers: {
277		...token ? { Authorization: token } : null,
278		...headers,
279		},
280		});
281		if (!headers) parallelRequests.set(url, promise);
282		const response = await promise;
283		if (!headers) parallelRequests.delete(url);
284		return response;
	277	async function fetch(rel: string, url: string, options: https.RequestOptions = {}) {
	278	console.info(`cache miss: ${rel}`);
	279	const response = await httpGet(url, {
	280	...options,
	281	headers: {
	282	...token ? { Authorization: token } : null,
	283	...options.headers,
	284	},
	285	});
	286	const { statusCode } = response;
	287	if (statusCode !== 200 && statusCode !== 206) {
	288	const text = await consumers.text(response);
	289	const msg = `Source of Truth responded with status ${statusCode}: ${text}`;
	290	console.error(msg);
	291	throw new Error("Source of Truth is unavailable");
285	292	}
	293	return response;
286	294	}
287	295
	296	/** Stream from the Source of Truth, skipping the disk cache. */
288	297	async function fetchAndSlice(
289	298	rel: string,
290	299	url: string,
	300	size: number,
291	301	start: number,
292	302	end: number,
293		) {
294		const headers = start !== 0 \|\| end !== Infinity
295		? { "Clover-Start": String(start), "Clover-End": String(end) }
296		: null;
297		return stream.Readable.toWeb(await fetch(rel, url, headers));
	303	): Promise<ReadableStream> {
	304	const ranged = start !== 0 \|\| end !== Infinity;
	305	const response = await fetch(
	306	rel,
	307	url,
	308	ranged ? { headers: { Range: `bytes=${start}-${end}` } } : {},
	309	);
	310	const partial = response.statusCode === 206;
	311
	312	// The caller has already promised a length, which a short body would break.
	313	const expected = partial ? end - start + 1 : size;
	314	const length = response.headers["content-length"];
	315	if (length !== undefined && Number(length) !== expected) {
	316	response.destroy();
	317	throw new Error(`Source of Truth has ${length} bytes of ${rel}, expected ${expected}`);
	318	}
	319
	320	// A server is free to ignore `Range` and respond with the whole file.
	321	if (ranged && !partial) {
	322	return ReadableStream.from(sliceStream(response, start, end));
	323	}
	324	return stream.Readable.toWeb(response);
298	325	}
299	326
300		export function sliceStream(
301		readable: stream.Readable,
	327	/** `end` is inclusive. Returning early destroys the rest of `readable`. */
	328	async function* sliceStream(
	329	readable: AsyncIterable<Buffer>,
302	330	start: number,
303	331	end: number,
304		): Readable {
305		const length = end === Infinity ? Infinity : end - start;
306		let recieved = 0;
307		let sent = 0;
308
309		const transform = new stream.Transform({
310		transform(chunk: Buffer, _, callback) {
311		const chunkEnd = recieved + chunk.length;
312
313		if (chunkEnd <= start) {
314		recieved = chunkEnd;
315		callback();
316		return;
317		}
318
319		if (sent >= length) {
320		callback();
321		return;
322		}
323
324		const sliceStart = Math.max(0, start - recieved);
325		const remaining = length - sent;
326		const sliceEnd = end === Infinity
327		? chunk.length
328		: Math.min(chunk.length, sliceStart + remaining);
329
330		const slice = chunk.slice(sliceStart, sliceEnd);
331		sent += slice.length;
332		recieved = chunkEnd;
333
334		this.push(slice);
335
336		callback();
337		},
338		});
339
340		readable.on("error", (err) => transform.destroy(err));
341
342		return readable.pipe(transform);
	332	) {
	333	let offset = 0;
	334	for await (const chunk of readable) {
	335	const slice = chunk.subarray(
	336	Math.max(0, start - offset),
	337	Math.max(0, end + 1 - offset),
	338	);
	339	offset += chunk.length;
	340	if (slice.length > 0) yield slice;
	341	if (offset > end) return;
	342	}
343	343	}
344	344
345	345	interface DiskFile {
346	346	size: number;
	347	/** bytes on disk so far, or `Infinity` once the download has failed */
347	348	written: async.Watch<number>;
	349	/** chunks recently written or read, keyed by their offset in the file */
348	350	memory: Map<number, WeakRef<Buffer>>;
349	351	}
350	352
...	...	@@ -356,28 +358,42 @@ function httpGet(url: string, options: https.RequestOptions) {
356	358	);
357	359	request.on("response", resolve);
358	360	request.on("error", reject);
	361	request.on("timeout", () => {
	362	request.destroy(new Error(`Source of Truth timed out: ${url}`));
	363	});
359	364	});
360	365	}
361	366
	367	type Memory = DiskFile["memory"];
362	368	const finalization = new FinalizationRegistry(
363		([map, key]: [Map<number, unknown>, number]) => {
364		map.delete(key);
	369	([memory, offset, ref]: [Memory, number, WeakRef<Buffer>]) => {
	370	if (memory.get(offset) === ref) memory.delete(offset);
365	371	},
366	372	);
367	373
	374	function remember(memory: Memory, offset: number, chunk: Buffer) {
	375	const ref = new WeakRef(chunk);
	376	memory.set(offset, ref);
	377	finalization.register(chunk, [memory, offset, ref]);
	378	}
	379
368	380	const openFile = util.promisify(fsCallbacks.open);
369	381
370	382	function deleteCachedPath(path: Path) {
371		try {
	383	// Streams still holding the descriptor close it instead of parking it.
	384	fds.delete(path);
	385	const parked = fdsCache.get(path);
	386	if (parked !== undefined) {
372	387	fdsCache.delete(path);
373		} catch {}
	388	fsCallbacks.closeSync(parked);
	389	}
374	390	try {
375	391	path.deleteSync();
376	392	} catch {}
377	393	}
378	394
379	395	interface RefCountFile {
380		fd: number;
	396	fd: Promise<number>;
381	397	refs: number;
382	398	}
383	399
...	...	@@ -394,42 +410,54 @@ function writeDiskCache() {
394	410	const serialized = diskCache.serialize({
395	411	map(path, { size, written }) {
396	412	if (size !== written.value) return null;
397		return [path.toString(), size];
	413	return [path.base, size];
398	414	},
399	415	});
400		diskCacheState.writeJsonFile(serialized)
401		.catch((err) => console.error(`Failed to write ${diskCacheState.relative()}:`, err));
	416	// Renamed into place so a crash cannot leave half a file to revive from.
	417	const partial = `${diskCacheState}.tmp`;
	418	try {
	419	fsCallbacks.writeFileSync(partial, JSON.stringify(serialized));
	420	fsCallbacks.renameSync(partial, diskCacheState.toString());
	421	} catch (err) {
	422	console.error(`Failed to write ${diskCacheState.relative()}:`, err);
	423	}
402	424	}
403	425
404		async function getHandle(path: Path, flag: string, unlockOnFail?: Disposable) {
405		let handle: RefCountFile \| null = null;
	426	/** `open` only runs when `path` has no holders and no parked descriptor. */
	427	async function getHandle(path: Path, open: () => Promise<number>) {
	428	let handle = fds.get(path);
	429	if (!handle) {
	430	const parked = fdsCache.get(path);
	431	if (parked !== undefined) fdsCache.delete(path);
	432	handle = {
	433	fd: parked === undefined ? open() : Promise.resolve(parked),
	434	refs: 0,
	435	};
	436	fds.set(path, handle);
	437	}
	438	const shared = handle;
	439	// Counted before the first await, or a release in between could park a
	440	// descriptor that is about to be used.
	441	shared.refs += 1;
	442
	443	let fd: number;
406	444	try {
407		handle = await fds.get(path) ?? null;
408		if (!handle) {
409		const cached = fdsCache.get(path);
410		if (cached) {
411		fds.delete(path);
412		fds.set(path, handle = { fd: cached, refs: 0 });
413		} else {
414		const promise = openFile(path.raw, flag)
415		.then((fd) => ({ fd, refs: 0 }))
416		.catch((err) => {
417		fds.delete(path);
418		throw err;
419		});
420		fds.set(path, promise);
421		fds.set(path, handle = await promise);
422		}
423		}
424		handle.refs += 1;
425		} catch (err: any) {
426		unlockOnFail?.[Symbol.dispose]();
427		if (err.code === "ENOENT") {
428		return null;
429		}
	445	fd = await shared.fd;
	446	} catch (err) {
	447	shared.refs -= 1;
	448	if (fds.get(path) === shared) fds.delete(path);
430	449	throw err;
431	450	}
432		return handle;
	451
	452	return {
	453	fd,
	454	release() {
	455	if ((shared.refs -= 1) > 0) return;
	456	if (fds.get(path) !== shared) return void fsCallbacks.closeSync(fd);
	457	fds.delete(path);
	458	fdsCache.set(path, fd);
	459	},
	460	};
433	461	}
434	462
435	463	interface ReadStreamOptions {
...	...	@@ -439,15 +467,19 @@ interface ReadStreamOptions {
439	467	lock?: Disposable;
440	468	}
441	469
442		async function readStream(path: Path, opts: ReadStreamOptions) {
443		const handle = await getHandle(path, "r", opts?.lock);
444		if (!handle) return null;
	470	/** Disposes `lock` when the stream closes, but not if opening it throws. */
	471	async function readStream(
	472	path: Path,
	473	{ start, end, ref, lock }: ReadStreamOptions,
	474	) {
	475	const handle = await getHandle(path, () => openFile(path.toString(), "r"));
445	476
446	477	let read = fsCallbacks.read;
447		if (opts?.ref) {
448		const { memory, written, size } = opts.ref;
449		let lastRead: number \| null = null;
450		read = function loop(
	478	if (ref) {
	479	const { memory, written, size } = ref;
	480	/** offset of the chunk in `memory` that the previous read stopped inside */
	481	let partial: number \| null = null;
	482	read = function(
451	483	fd: number,
452	484	buf: Buffer,
453	485	offset: 0,
...	...	@@ -459,155 +491,70 @@ async function readStream(path: Path, opts: ReadStreamOptions) {
459	491	ASSERT(offset === 0);
460	492	written.until((v) => v >= Math.min(size, pos + len)).then(
461	493	(newLength) => {
	494	if (newLength > size) return cb(new Error("Broken Pipe"), 0, buf);
	495
462	496	// Try to re-use uncollected buffers in memory
463		while (true) {
464		const weak = memory.get(lastRead ?? pos);
465		if (weak) {
466		let cachedBuf = weak.deref();
467		if (cachedBuf) {
468		if (lastRead) {
469		cachedBuf = cachedBuf.subarray(pos - lastRead);
470		}
471		const { byteLength } = cachedBuf;
472
473		if (byteLength === len && offset === 0) {
474		cb(null, len, cachedBuf);
475		return;
476		}
477		if (byteLength <= len) {
478		buf.set(cachedBuf, offset);
479		lastRead = null;
480		} else {
481		buf.set(cachedBuf.subarray(0, len), offset);
482		lastRead = lastRead ?? pos;
483		}
484		offset += byteLength;
485		len -= byteLength;
486		pos += byteLength;
487		if (len > 0) continue;
488		cb(null, offset + len, buf);
489		return;
490		} else {
491		memory.delete(pos);
492		}
	497	let filled = 0;
	498	while (filled < len) {
	499	const at = pos + filled;
	500	const key = partial ?? at;
	501	const chunk = memory.get(key)?.deref()?.subarray(at - key);
	502	partial = null;
	503	if (!chunk?.byteLength) {
	504	if (key === at) break;
	505	continue;
	506	}
	507	if (filled === 0 && chunk.byteLength === len) {
	508	return cb(null, len, chunk);
493	509	}
494	510
495		lastRead = null;
496		break;
	511	const slice = chunk.subarray(0, len - filled);
	512	buf.set(slice, filled);
	513	filled += slice.byteLength;
	514	if (slice.byteLength < chunk.byteLength) partial = key;
497	515	}
	516	if (filled === len) return cb(null, len, buf);
498	517
499	518	// Read from disk
500		if (newLength > size) cb(new Error("Broken Pipe"), 0, buf);
501		fsCallbacks.read(
502		fd,
503		buf,
504		offset,
505		len,
506		pos,
507		async (er, bytes, buf) => {
508		if (!er && bytes > 0) {
509		memory.set(pos - offset, new WeakRef(buf));
510		finalization.register(buf, [memory, pos]);
511		}
512		cb(er, offset + bytes, buf);
513		},
514		);
	519	fsCallbacks.read(fd, buf, filled, len - filled, pos + filled, (er, bytes) => {
	520	// A short read leaves the tail of `buf` uninitialized.
	521	if (!er && filled === 0 && bytes === len) remember(memory, pos, buf);
	522	cb(er, filled + bytes, buf);
	523	});
515	524	},
516	525	);
517	526	} as typeof fsCallbacks.read;
518	527	}
519	528
520		let closed = false;
521	529	return fs.createReadStream(path.toString(), {
522	530	fd: handle.fd,
523		start: opts?.start ?? 0,
524		end: opts?.end ?? Infinity,
525		fs: {
526		close: util.callbackify(async () => {
527		ASSERT(!closed);
528		if ((handle.refs -= 1) <= 0) {
529		fds.delete(path);
530		fdsCache.set(path, handle.fd);
531		}
532		closed = true;
533		opts?.lock?.[Symbol.dispose]();
534		}),
535		read,
536		},
537		});
538		}
539
540		async function writeStream(
541		path: Path,
542		data: stream.Readable,
543		file: DiskFile,
544		lock: Disposable,
545		) {
546		const handle = UNWRAP(await getHandle(path, "w+", lock));
547		let closed = false;
548
549		const { memory, written } = file;
550		const writer = fs.createWriteStream(path.toString(), {
551		start: 0,
552		fd: handle.fd,
	531	start,
	532	end,
553	533	fs: {
554		close: util.callbackify(async () => {
555		ASSERT(!closed);
556		if ((handle.refs -= 1) <= 0) {
557		fds.delete(path);
558		fdsCache.set(path, handle.fd);
559		}
560		closed = true;
561		lock[Symbol.dispose]();
562		writeDiskCache();
563		}),
564		write(
565		fd: number,
566		data: Buffer,
567		off: 0,
568		size: number,
569		pos: number \| undefined,
570		cb: (er: Error \| null, bytesWritten: number, buffer: Buffer) => void,
571		) {
572		ASSERT(size === data.byteLength);
573		ASSERT(pos != null);
574		fsCallbacks.write(
575		fd,
576		data,
577		off,
578		size,
579		pos,
580		(er, bytes, buf) => {
581		if (!er && bytes) {
582		memory.set(pos, new WeakRef(buf));
583		finalization.register(buf, [memory, pos]);
584		written.value += bytes;
585		}
586		cb(er, bytes, buf);
587		},
588		);
	534	close(_: number, cb: (er: Error \| null) => void) {
	535	handle.release();
	536	lock?.[Symbol.dispose]();
	537	cb(null);
589	538	},
	539	read,
590	540	},
591	541	});
592		data.pipe(writer);
593
594		return writer;
595	542	}
596	543
597	544	import * as fs from "#sitegen/fs";
598	545	import { Path } from "#sitegen/path";
599		import { ASSERT, UNWRAP } from "@clo/lib/assert";
	546	import { ASSERT } from "@clo/lib/assert";
600	547	import * as async from "@clo/lib/async";
601	548	import * as exception from "@clo/lib/exception";
602	549	import * as log from "@clo/lib/log";
603	550	import { Lru } from "@clo/lib/Lru";
604		import * as ts from "@clo/lib/ts";
605		import * as crypto from "node:crypto";
606	551	import * as fsCallbacks from "node:fs";
607	552	import * as http from "node:http";
608	553	import * as https from "node:https";
609	554	import * as stream from "node:stream";
610		import type { ReadableStream } from "node:stream/web";
	555	import * as consumers from "node:stream/consumers";
	556	import * as streamPromises from "node:stream/promises";
	557	import { ReadableStream } from "node:stream/web";
611	558	import * as util from "node:util";
612		import { MediaFile } from "./models/MediaFile.ts";
	559	import type { MediaFile } from "./models/MediaFile.ts";
613	560	import * as paths from "./paths.ts";

src/source-of-truth.test.ts created+51

...	...	@@ -0,0 +1,51 @@
	1	const dir = vi.hoisted(() => {
	2	const fs = process.getBuiltinModule("node:fs");
	3	const os = process.getBuiltinModule("node:os");
	4	const dir = fs.mkdtempSync(os.tmpdir() + "/clover-sot-test-");
	5	fs.mkdirSync(dir + "/raw");
	6	fs.mkdirSync(dir + "/derived");
	7	Object.assign(process.env, {
	8	CLOVER_DB: dir + "/db",
	9	CLOVER_FILE_RAW: dir + "/raw",
	10	CLOVER_FILE_DERIVED: dir + "/derived",
	11	CLOVER_SOT_KEY: "test",
	12	});
	13	return dir;
	14	});
	15	afterAll(() => fs.rmSync(dir, { recursive: true, force: true }));
	16
	17	const data = crypto.randomBytes(1_000_000);
	18	const url = "http://localhost/file/new%20home%20[rev%203].mp3";
	19	fs.writeFileSync(dir + "/raw/new home [rev 3].mp3", data);
	20	MediaFile.createFile({
	21	path: "/new home [rev 3].mp3",
	22	date: new Date(0),
	23	hash: crypto.createHash("sha1").update(data).digest("hex"),
	24	size: data.length,
	25	duration: 0,
	26	dimensions: "",
	27	contents: "",
	28	});
	29
	30	test("requests for the same file do not share a file offset", async () => {
	31	// the second request arrives once the first holds the descriptor
	32	const first = await app.fetch(new Request(url));
	33	const second = await app.fetch(new Request(url));
	34	const bodies = await Promise.all([first.arrayBuffer(), second.arrayBuffer()]);
	35	for (const body of bodies) assert.ok(data.equals(Buffer.from(body)));
	36	});
	37
	38	test("a range is served as partial content", async () => {
	39	const response = await app.fetch(new Request(url, { headers: { Range: "bytes=70000-" } }));
	40	assert.equal(response.status, 206);
	41	assert.equal(response.headers.get("Content-Range"), `bytes 70000-999999/1000000`);
	42	assert.equal(response.headers.get("Content-Length"), "930000");
	43	assert.ok(data.subarray(70000).equals(Buffer.from(await response.arrayBuffer())));
	44	});
	45
	46	import { strict as assert } from "node:assert";
	47	import * as crypto from "node:crypto";
	48	import * as fs from "node:fs";
	49	import { afterAll, test, vi } from "vitest";
	50	import { MediaFile } from "./file-viewer/models/MediaFile.ts";
	51	import app from "./source-of-truth.ts";

src/source-of-truth.ts+13-1

...	...	@@ -100,9 +100,21 @@ app.get("/file/*", async (c) => {
100	100	if (err.code === "ENOENT") return c.notFound();
101	101	throw err;
102	102	}
103		c.header("Content-Length", (await fstat(handle.fd)).size.toString());
	103	const { size } = await fstat(handle.fd);
	104	let start = 0;
	105	let end = Infinity;
	106	const ranges = http.parseRangeHeader(c.req.header("Range") ?? "", size);
	107	if (ranges?.[0] && ranges.length === 1) {
	108	({ start, end } = ranges[0]);
	109	c.status(206);
	110	c.header("Content-Range", `bytes ${start}-${end}/${size}`);
	111	}
	112	c.header("Content-Length", String(Math.min(end, size - 1) - start + 1));
104	113	const nodeStream = fs.createReadStream(fullPath, {
105	114	fd: handle.fd,
	115	// Positioned, or requests sharing this descriptor would share its offset.
	116	start,
	117	end,
106	118	fs: {
107	119	close: util.callbackify(async () => {
108	120	ASSERT(handle);