From 27372f562b13b96c1bf165339923d5f402753428 Mon Sep 17 00:00:00 2001 From: clover caruso Date: Sun, 11 Oct 2026 15:45:35 -0700 Subject: [PATCH] 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 --- framework/lib/path.test.ts | 22 ++ framework/lib/path.ts | 5 +- lib/Lru.test.ts | 22 ++ lib/Lru.ts | 4 + lib/http.test.ts | 133 +++++++ lib/http.ts | 32 +- lib/readme.changes.md | 9 + src/file-viewer/backend.ts | 24 +- src/file-viewer/cache.test.ts | 603 ++++++++++++++++++++++++++++++ src/file-viewer/cache.ts | 665 ++++++++++++++++------------------ src/source-of-truth.test.ts | 51 +++ src/source-of-truth.ts | 14 +- 12 files changed, 1203 insertions(+), 381 deletions(-) create mode 100644 framework/lib/path.test.ts create mode 100644 lib/http.test.ts create mode 100644 src/file-viewer/cache.test.ts create mode 100644 src/source-of-truth.test.ts diff --git a/framework/lib/path.test.ts b/framework/lib/path.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..7c818f287ac2585c87e953ea50527e4bbe8d5df1 --- /dev/null +++ b/framework/lib/path.test.ts @@ -0,0 +1,22 @@ +test("a path stays deduplicated after an older copy is collected", async () => { + const raw = "/clover/path.test/deduplicated"; + (() => void new Path(raw))(); + // a WeakRef keeps its target alive until the task that created it ends + await delay(0); + collectGarbage(); + + // the first copy is collected, but its finalizer has not yet run + const second = new Path(raw); + await delay(10); + assert.equal(new Path(raw), second); +}); + +v8.setFlagsFromString("--expose-gc"); +const collectGarbage: () => void = vm.runInNewContext("gc"); + +import { delay } from "@clo/lib/async"; +import { strict as assert } from "node:assert"; +import * as v8 from "node:v8"; +import * as vm from "node:vm"; +import { test } from "vitest"; +import { Path } from "./path.ts"; diff --git a/framework/lib/path.ts b/framework/lib/path.ts index 9007cab3699224c7f109728491b9d215f9e6bb90..98c4e6d4f5ca29207ca572b407333fd37ae779e5 100644 --- a/framework/lib/path.ts +++ b/framework/lib/path.ts @@ -1,5 +1,8 @@ const unique = new Map>(); -const final = new FinalizationRegistry((raw: string) => unique.delete(raw)); +// A finalizer can run after a newer `Path` has taken over the entry. +const final = new FinalizationRegistry((raw: string) => { + if (!unique.get(raw)?.deref()) unique.delete(raw); +}); export class Path { private raw: string; diff --git a/lib/Lru.test.ts b/lib/Lru.test.ts index 54984fdf5c1b49cca5cb7abfd090b55eff727d0b..64105dfddd91fa98d0b1da8528b878ea5a6511ea 100644 --- a/lib/Lru.test.ts +++ b/lib/Lru.test.ts @@ -148,6 +148,28 @@ test("revival", () => { assert.equal(lru.get("d"), 4); assert.equal(lru.get("e"), 5); }); +test("delete callback sees every removal, evict only automatic ones", () => { + const deleted: string[] = []; + const evicted: string[] = []; + const lru = new Lru({ + capacity: 2, + delete: (_, key) => deleted.push(key), + evict: (_, key) => evicted.push(key), + }); + lru.set("a", 1); + lru.set("b", 2); + lru.set("c", 3); + assert.deepEqual([deleted, evicted], [["a"], ["a"]]); + + assert.equal(lru.delete("b"), true); + assert.equal(lru.delete("b"), false); + assert.deepEqual([deleted, evicted], [["a", "b"], ["a"]]); + + lru.set("d", 4); + lru.clear(); + assert.deepEqual([deleted, evicted], [["a", "b", "c", "d"], ["a"]]); + assert.equal(lru.used, 0); +}); import assert from "node:assert"; import { test } from "vitest"; diff --git a/lib/Lru.ts b/lib/Lru.ts index 6a324e1c0677ad319cbcd13e78461ea83dd77aa8..2b824f748624ec5b024494d448a489ea5efdf2c6 100644 --- a/lib/Lru.ts +++ b/lib/Lru.ts @@ -56,18 +56,22 @@ export class Lru extends Map { } override clear() { + const removed = this.#deleteFn ? [...this] : []; super.clear(); this.#order.clear(); this.#head = none; this.#tail = none; this.#used = 0; + for (const [key, value] of removed) this.#deleteFn?.(value, key); } override delete(key: K): boolean { const entry = this.#evict(key); if (!entry) return false; + const value = super.get(key) as V; this.#used -= entry[0]; ASSERT(this.#order.delete(key)); ASSERT(super.delete(key)); + this.#deleteFn?.(value, key); return true; } override get(key: K): V | undefined { diff --git a/lib/http.test.ts b/lib/http.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..5b02e4e8fb8ded7de817b743065578a7671e91bc --- /dev/null +++ b/lib/http.test.ts @@ -0,0 +1,133 @@ +describe("parseRangeHeader", () => { + test("closed, open, and suffix ranges", () => { + assert.deepEqual(http.parseRangeHeader("bytes=0-1", 100), [{ start: 0, end: 1 }]); + assert.deepEqual(http.parseRangeHeader("bytes=10-", 100), [{ start: 10, end: 99 }]); + assert.deepEqual(http.parseRangeHeader("bytes=-10", 100), [{ start: 90, end: 99 }]); + assert.deepEqual(http.parseRangeHeader("bytes=-500", 100), [{ start: 0, end: 99 }]); + assert.deepEqual(http.parseRangeHeader("bytes=50-500", 100), [{ start: 50, end: 99 }]); + assert.deepEqual(http.parseRangeHeader("bytes=0-9, 20-29", 100), [ + { start: 0, end: 9 }, + { start: 20, end: 29 }, + ]); + }); + + test("ranges past the end are unsatisfiable", () => { + // https://shale.paperclover.net/sitegen/issues/106 + assert.deepEqual(http.parseRangeHeader("bytes=46226599-", 25705639), []); + assert.deepEqual(http.parseRangeHeader("bytes=100-200", 100), []); + assert.deepEqual(http.parseRangeHeader("bytes=-0", 100), []); + assert.deepEqual(http.parseRangeHeader("bytes=0-", 0), []); + }); + + test("malformed headers are ignored", () => { + for ( + const header of [ + "", + "bytes", + "bytes=", + "bytes=-", + "bytes=5-2", + "bytes=a-b", + "bytes=1.5-3", + "bytes=0x10-", + "bytes=--5", + "bytes=1-2-3", + "pages=0-1", + "bytes=0-5,3-9", + ] + ) { + assert.equal(http.parseRangeHeader(header, 100), null, header); + } + }); + + test("fuzz: every range it returns can be read", () => { + const pieces = [ + "bytes", + "=", + "-", + ",", + " ", + "0", + "1", + "7", + "99", + "100", + "101", + "25705639", + "9".repeat(30), + "x", + ".", + "e3", + "-1", + ]; + let state = 106; + const random = (max: number) => (state = (state * 48271) % 0x7fffffff) % max; + for (let i = 0; i < 20_000; i++) { + const size = [0, 1, 100, 25705639][random(4)]!; + let header = random(4) === 0 ? "" : "bytes="; + for (let n = random(7); n > 0; n--) header += pieces[random(pieces.length)]; + + for (const { start, end } of http.parseRangeHeader(header, size) ?? []) { + assert.ok( + Number.isInteger(start) && Number.isInteger(end) + && 0 <= start && start <= end && end < size, + `${JSON.stringify(header)} of ${size} gave [${start}, ${end}]`, + ); + } + } + }); +}); + +describe("serve", () => { + test("a client that disconnects cancels the response body", async () => { + const cancelled = Promise.withResolvers(); + await using server = await http.serve({ + port: 0, + respond: async () => + new Response( + new ReadableStream({ + pull: (controller) => controller.enqueue(new Uint8Array(65536)), + cancel: () => cancelled.resolve(), + }), + ), + }); + + const response = await new Promise((resolve) => nodeHttp.get(server.url, resolve)); + response.once("data", () => response.destroy()); + await cancelled.promise; + }); + + test("a failing response body closes the connection", async () => { + const failure = Object.assign(new Error("bad file descriptor"), { code: "EBADF" }); + const logged = vi.spyOn(console, "error").mockImplementation(() => {}); + let pulls = 0; + await using server = await http.serve({ + port: 0, + respond: async () => + new Response( + new ReadableStream({ + pull(controller) { + if (pulls++ === 0) controller.enqueue(new Uint8Array(16)); + else controller.error(failure); + }, + }), + { headers: { "Content-Length": "65536" } }, + ), + }); + + await assert.rejects(async () => { + const response = await new Promise((resolve, reject) => + nodeHttp.get(server.url, resolve).on("error", reject) + ); + await consumers.buffer(response); + }, { code: "ECONNRESET" }); + assert.deepEqual(logged.mock.calls, [[failure]]); + logged.mockRestore(); + }); +}); + +import { strict as assert } from "node:assert"; +import * as nodeHttp from "node:http"; +import * as consumers from "node:stream/consumers"; +import { describe, test, vi } from "vitest"; +import * as http from "./http.ts"; diff --git a/lib/http.ts b/lib/http.ts index 46e8222956dc8627fdb83e4ec37466bca119c86c..d86c25fbecb4c98f1068dac5efb4c8064ba83cf8 100644 --- a/lib/http.ts +++ b/lib/http.ts @@ -67,7 +67,15 @@ export async function serve( Array.from(response.headers.entries()), ); if (response.body) { - stream.Readable.fromWeb(response.body as import("node:stream/web").ReadableStream).pipe(res); + // unlike `pipe`, this cancels the body when the client goes away, + // and a failing body closes the connection instead of the process. + stream.pipeline( + stream.Readable.fromWeb(response.body as import("node:stream/web").ReadableStream), + res, + (err) => { + if (err && exception.code(err) !== "ERR_STREAM_PREMATURE_CLOSE") console.error(err); + }, + ); } else { res.end(); } @@ -123,7 +131,9 @@ export async function serve( /** * parses the [`Range`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Guides/Range_requests) - * header, returning `null` if it is invalid. the `fileSize` + * header for a resource of `fileSize` bytes. `null` means the header is + * malformed and should be ignored. otherwise, the satisfiable ranges are + * returned clamped to the resource, so an empty array is a `416` response. */ export function parseRangeHeader( rangeHeader: string, @@ -137,14 +147,16 @@ export function parseRangeHeader( const rangeParts = ranges.split(","); for (const range of rangeParts) { - const split = range.split("-"); - const [start, end] = split; - if (!start || !end) return null; - if (start === "" && end === "") return null; - result.push({ - start: start === "" ? fileSize - +end : +start, - end: end === "" ? fileSize - 1 : +end, - }); + const match = /^\s*(\d*)-(\d*)\s*$/.exec(range); + if (!match) return null; + const [, first = "", last = ""] = match; + if (first === "" && last === "") return null; + if (first !== "" && last !== "" && +last < +first) return null; + + // `-500` is the last 500 bytes, `500-` is everything after the first 500. + const start = first === "" ? Math.max(0, fileSize - +last) : +first; + const end = first === "" || last === "" ? fileSize - 1 : Math.min(+last, fileSize - 1); + if (start <= end) result.push({ start, end }); } // validate that ranges do not intersect diff --git a/lib/readme.changes.md b/lib/readme.changes.md index 0f40490c4dca90b0ac11b53ddedc42906dbc7d67..b22e94df24e88a37c2af9b905fb6ff97b709b279 100644 --- a/lib/readme.changes.md +++ b/lib/readme.changes.md @@ -7,6 +7,15 @@ - `meta`'s auto-embed feature now sets all the twitter tags, not just thumbnail. sorry i might have been dumb when doing the first version of this. +### fixes + +- `Lru`: `delete()` and `clear()` now call the `delete` callback, as documented. +- `http.parseRangeHeader` accepts open and suffix ranges (`500-`, `-500`), + clamps ranges to the resource, and returns an empty array when none of them + can be satisfied. +- `http.serve` cancels the response body when the client disconnects, and a + failing body closes the connection instead of crashing the process. + ## v6 ### breaking diff --git a/src/file-viewer/backend.ts b/src/file-viewer/backend.ts index ceac957312822e64a48e58d75db1aad58200b0c2..2780f0939125d3c578d0f1d6e85ad0459e43e603 100644 --- a/src/file-viewer/backend.ts +++ b/src/file-viewer/backend.ts @@ -151,7 +151,7 @@ app.get("/file/*", async (c, next) => { let status = 200; let rangeHeader = c.req.header("Range") ?? null; const ifRangeHeader = c.req.header("If-Range"); - if (ifRangeHeader && ifRangeOutdated(file, ifRangeHeader)) { + if (ifRangeHeader && ifRangeOutdated(hash, date, ifRangeHeader)) { // > If the condition is not fulfilled, the full resource is // > sent back with a 200 OK status. rangeHeader = null; @@ -163,17 +163,21 @@ app.get("/file/*", async (c, next) => { // Range requests // https://developer.mozilla.org/en-US/docs/Web/HTTP/Range_requests if (rangeHeader) { - const ranges = http.parseRangeHeader(rangeHeader, file.size); + const ranges = http.parseRangeHeader(rangeHeader, size); + if (ranges?.length === 0) { + c.res = new Response(null, { + status: 416, + headers: { "Content-Range": `bytes */${size}` }, + }); + return; + } if (ranges?.[0] && ranges.length === 1) { start = ranges[0].start; end = ranges[0].end; status = 206; - headers.set( - "Content-Range", - `bytes ${ranges[0].start}-${ranges[0].end}/${file.size}`, - ); - headers.delete("Content-Length"); + headers.set("Content-Range", `bytes ${start}-${end}/${size}`); + headers.set("Content-Length", String(end - start + 1)); } else { // multi-ranges are not supported // it could be done by calling `cache.read` multiple times. @@ -209,13 +213,13 @@ app.get("/file/*", async (c, next) => { }); }); -function ifRangeOutdated(file: MediaFile, ifRangeHeader: string) { +function ifRangeOutdated(hash: string, date: Date, ifRangeHeader: string) { // etag if (ifRangeHeader[0] === "\"") { - return ifRangeHeader.slice(1, -1) !== file.hash; + return ifRangeHeader.slice(1, -1) !== hash; } // date - return new Date(ifRangeHeader) < file.date; + return new Date(ifRangeHeader) < date; } function checkCotyledonCookie(c: Context) { diff --git a/src/file-viewer/cache.test.ts b/src/file-viewer/cache.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..35aef5f265ce03fb15631609d14b3dab5dccd29c --- /dev/null +++ b/src/file-viewer/cache.test.ts @@ -0,0 +1,603 @@ +// fuzzes `cache.read` against a fake Source of Truth and a `node:fs` that +// tracks descriptor ownership and injects faults. every delivered byte is +// compared against the known file contents, and at each quiescent point the +// locks, descriptors, and cache directory must agree with the Lru. +// +// FUZZ_SEED, FUZZ_RUNS, and FUZZ_OPS reproduce a failure or run a soak, and +// FUZZ_STATS prints what each run exercised. +const h = vi.hoisted(() => ({ + lrus: [] as unknown[], + fs: { + /** descriptors opened through the mock and not yet closed */ + open: new Map(), + violations: [] as string[], + random: Math.random as () => number, + lag: 0, + shortIo: 0, + readError: 0, + writeError: 0, + onFault: () => {}, + }, +})); + +vi.mock("@clo/lib/Lru", async (importOriginal) => { + const real = await importOriginal(); + class Tracked extends real.Lru { + constructor(opts: ConstructorParameters>[0]) { + super(opts); + h.lrus.push(this); + } + } + return { ...real, Lru: Tracked }; +}); + +vi.mock("node:fs", async (importOriginal) => { + const real = await importOriginal(); + const s = h.fs; + const roll = (chance: number) => chance > 0 && s.random() < chance; + const later = (fn: () => void) => { + if (roll(s.lag)) setTimeout(fn, Math.floor(s.random() * 3)); + else fn(); + }; + const injected = (code: string, syscall: string) => { + s.onFault(); + return Object.assign(new Error(`${code}: injected, ${syscall}`), { code, syscall }); + }; + type IoCallback = (err: Error | null, bytes: number, buf: Buffer) => void; + type Io = ( + fd: number, + buf: Buffer, + off: number, + len: number, + pos: number, + cb: IoCallback, + ) => void; + const io = + (name: string, real: Io, errorChance: () => number, errorCode: string): Io => (fd, buf, off, len, pos, cb) => { + if (!s.open.has(fd)) s.violations.push(`${name} on fd ${fd} which is not open`); + if (roll(errorChance())) return later(() => cb(injected(errorCode, name), 0, buf)); + const partial = len > 1 && roll(s.shortIo) ? 1 + Math.floor(s.random() * (len - 1)) : len; + real(fd, buf, off, partial, pos, (err, bytes) => later(() => cb(err, bytes, buf))); + }; + const overrides = { + open(path: string, flags: string, cb: (err: Error | null, fd: number) => void) { + real.open(path, flags, (err, fd) => { + if (!err) s.open.set(fd, path); + cb(err, fd); + }); + }, + close(fd: number, cb: (err: Error | null) => void) { + if (!s.open.delete(fd)) s.violations.push(`close of fd ${fd} which is not open`); + real.close(fd, cb); + }, + closeSync(fd: number) { + if (!s.open.delete(fd)) s.violations.push(`close of fd ${fd} which is not open`); + real.closeSync(fd); + }, + read: io("read", real.read, () => s.readError, "EIO"), + write: io("write", real.write, () => s.writeError, "ENOSPC"), + }; + return { ...real, ...overrides, default: { ...real, ...overrides } }; +}); + +const seed = Number(process.env.FUZZ_SEED ?? 1); +const runs = Number(process.env.FUZZ_RUNS ?? 6); +const opCount = Number(process.env.FUZZ_OPS ?? 250); +const token = "fuzz-token"; + +function prng(seed: number) { + const next = () => { + let t = (seed += 0x6D2B79F5); + t = Math.imul(t ^ (t >>> 15), t | 1); + t ^= t + Math.imul(t ^ (t >>> 7), t | 61); + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; + return Object.assign(next, { + int: (max: number) => Math.floor(next() * max), + roll: (chance: number) => chance > 0 && next() < chance, + pick: (items: readonly T[]) => UNWRAP(items[Math.floor(next() * items.length)]), + }); +} +type Random = ReturnType; + +interface FakeFile { + file: MediaFile; + derived: string | null; + url: string; + hash: string; + data: Buffer; +} + +function fakeFiles(random: Random): FakeFile[] { + const sizes = [0, 1, 2, 4096, 65535, 65536, 65537, 131072, 200_001, 300_000]; + const files = sizes.map((size, i): FakeFile => { + const data = Buffer.alloc(size); + for (let b = 0; b < size; b++) data[b] = random.int(256); + const path = `/2026/friends/${i}/new home [rev ${i}].mp3`; + return { + file: { path } as MediaFile, + derived: null, + url: `/file${path}`, + hash: crypto.createHash("sha1").update(data).digest("hex"), + data, + }; + }); + // a derived asset, and a second path with the contents of another file + const source = UNWRAP(files[4]); + const twin = UNWRAP(files[7]); + const preview = Buffer.from(source.data.subarray(100, 30_100)); + files.push({ + ...source, + derived: `${source.hash}/preview.webp`, + url: `${source.url}:/preview.webp`, + hash: crypto.createHash("sha1").update(preview).digest("hex"), + data: preview, + }, { + ...twin, + file: { path: "/2026/copy of new home.mp3" } as MediaFile, + url: "/file/2026/copy of new home.mp3", + }); + return files; +} + +interface Chaos { + /** Source of Truth responds 500 */ + status: number; + /** Source of Truth resets the connection mid-body */ + abort: number; + /** Source of Truth serves a file shorter than the database says */ + truncated: number; + /** a filesystem callback arrives a few milliseconds late */ + lag: number; + shortIo: number; + readError: number; + writeError: number; +} +const calm: Chaos = { + status: 0, + abort: 0, + truncated: 0, + lag: 0.3, + shortIo: 0.1, + readError: 0, + writeError: 0, +}; +const stormy: Chaos = { + ...calm, + status: 0.03, + abort: 0.05, + truncated: 0.02, + readError: 0.002, + writeError: 0.003, +}; + +interface DiskFile { + size: number; + written: { value: number }; +} + +class World { + violations: string[] = []; + /** the cache's own log, shown alongside a failure */ + log: string[] = []; + /** what the run exercised, printed with FUZZ_STATS */ + stats: Record = {}; + lastFault = -Infinity; + files: FakeFile[]; + random: Random; + sotRandom: Random; + dir: string; + cacheDir: string; + rawDir: string | null; + server!: http.Server; + cache!: typeof import("./cache.ts"); + diskCache!: Lru; + fdsCache!: Lru; + + constructor(seed: number, public chaos: Chaos, local: boolean) { + this.random = prng(seed); + this.sotRandom = prng(seed ^ 0x5eed); + this.files = fakeFiles(prng(seed ^ 0xf11e)); + this.dir = fs.mkdtempSync(path.join(os.tmpdir(), "clover-cache-fuzz-")); + this.cacheDir = path.join(this.dir, ".clover/file-viewer"); + this.rawDir = local ? path.join(this.dir, "raw") : null; + + const fsRandom = prng(seed ^ 0xd15c); + Object.assign(h.fs, chaos, { random: fsRandom, onFault: () => this.fault() }); + h.fs.violations = this.violations; + } + + fault() { + this.lastFault = performance.now(); + } + + count(what: string, by = 1) { + this.stats[what] = (this.stats[what] ?? 0) + by; + } + + async start() { + this.server = http.createServer((req, res) => this.respond(req, res)); + await new Promise((resolve) => this.server.listen(0, "127.0.0.1", resolve)); + const { port } = this.server.address() as net.AddressInfo; + + // only some files are on the local share, the rest fall back to the network + if (this.rawDir) { + for (const [i, { file, derived, data }] of this.files.entries()) { + if (i % 3 === 0) continue; + const local = derived + ? path.join(this.dir, "derived", derived) + : path.join(this.rawDir, file.path); + fs.mkdirSync(path.dirname(local), { recursive: true }); + fs.writeFileSync(local, data); + } + } + + process.env.CLOVER_SOT_URL = `http://127.0.0.1:${port}`; + process.env.CLOVER_SOT_KEY = token; + process.env.CLOVER_FILE_RAW = this.rawDir ?? path.join(this.dir, "missing"); + process.env.CLOVER_FILE_DERIVED = path.join(this.dir, "derived"); + process.env.CLOVER_DB = path.join(this.dir, "db"); + process.chdir(this.dir); + await this.boot(); + } + + /** load a fresh copy of the cache module, as a process restart would */ + async boot() { + for (const fd of h.fs.open.keys()) realFs.closeSync(fd); + h.fs.open.clear(); + + vi.resetModules(); + const log = await import("@clo/lib/log"); + log.replaceGlobalMessageDestination((message) => { + this.log.push(log.formatAnsiMessage(message, false).trimEnd()); + }); + this.cache = await import("./cache.ts"); + [this.diskCache, this.fdsCache] = h.lrus.slice(-2) as [Lru, Lru]; + this.diskCache.capacity = 700_000; + this.fdsCache.capacity = 3; + } + + async stop() { + await new Promise((resolve) => this.server.close(resolve)); + this.server.closeAllConnections(); + for (const fd of h.fs.open.keys()) realFs.closeSync(fd); + h.fs.open.clear(); + fs.rmSync(this.dir, { recursive: true, force: true }); + } + + respond(req: http.IncomingMessage, res: http.ServerResponse) { + const { sotRandom: random, chaos } = this; + const found = this.files.find((f) => f.url === decodeURI(UNWRAP(req.url))); + if (!found) return void res.writeHead(404).end("not found"); + if (req.headers.authorization !== token) { + this.violations.push(`request for ${found.url} did not carry the token`); + } + if (random.roll(chaos.status)) { + this.fault(); + return void res.writeHead(500).end("injected"); + } + + let { data } = found; + let status = 200; + const headers: Record = {}; + // a server is free to ignore `Range`, so do so half of the time + const range = /^bytes=(\d+)-(\d+)$/.exec(req.headers.range ?? ""); + if (range && random.roll(0.5)) { + status = 206; + headers["Content-Range"] = `bytes ${range[1]}-${range[2]}/${data.length}`; + data = data.subarray(Number(range[1]), Number(range[2]) + 1); + } + if (data.length > 0 && random.roll(chaos.truncated)) { + this.fault(); + data = data.subarray(0, random.int(data.length)); + } + const abortAt = random.roll(chaos.abort) ? random.int(data.length + 1) : Infinity; + headers["Content-Length"] = String(data.length); + res.writeHead(status, headers); + + let sent = 0; + const pump = () => { + if (res.destroyed) return; + if (sent >= abortAt) { + this.fault(); + return void res.destroy(); + } + if (sent >= data.length) return void res.end(); + const end = Math.min(data.length, abortAt, sent + 1 + random.int(90_000)); + res.write(data.subarray(sent, sent = end)); + if (random.roll(0.3)) setTimeout(pump, random.int(4)); + else setImmediate(pump); + }; + pump(); + } + + /** a request for the whole of `target` */ + async request({ file, derived, hash, data }: FakeFile) { + const options = { file, derived, hash, size: data.length, start: 0, end: Infinity }; + return UNWRAP(await this.cache.read(options)); + } + + async body(target: FakeFile) { + const { stream } = await this.request(target); + return Buffer.from(await new Response(stream as BodyInit).arrayBuffer()); + } + + /** one request, consumed the way `backend.ts` hands it to a `Response` */ + async read(label: string) { + const { random } = this; + const target = random.pick(this.files); + const { data } = target; + let start = 0, end = Infinity; + if (data.length > 0 && random.roll(0.6)) { + start = random.int(data.length); + end = start + random.int(data.length - start); + } + const expected = data.subarray(start, end + 1); + const mode = random.pick(["drain", "drain", "slow", "cancel"] as const); + const cancelAt = mode === "cancel" ? random.int(expected.length + 1) : Infinity; + const what = `${label} ${target.url} [${start}, ${end}] ${mode}`; + const startedAt = performance.now(); + + let received = 0; + let src = ""; + try { + const found = UNWRAP( + await this.cache.read({ + file: target.file, + derived: target.derived, + size: data.length, + hash: target.hash, + start, + end, + noWrite: random.roll(0.1), + }), + ); + src = found.src; + this.count(`src ${src}`); + if (found.size !== data.length) { + this.violations.push(`${what}: size ${found.size}, expected ${data.length}`); + } + const reader = UNWRAP(new Response(found.stream as BodyInit).body).getReader(); + while (true) { + if (received >= cancelAt) { + this.count("cancelled"); + return void await reader.cancel(); + } + const { done, value } = await reader.read(); + if (done) break; + const want = expected.subarray(received, received + value.length); + if (want.length !== value.length || !want.equals(value)) { + this.violations.push(`${what} (${src}): wrong bytes at offset ${received}`); + return void await reader.cancel(); + } + received += value.length; + this.count("bytes verified", value.length); + if (mode === "slow") await delay(random.int(4)); + } + if (received !== expected.length) { + this.violations.push( + `${what} (${src}): ended cleanly after ${received} of ${expected.length} bytes`, + ); + } + } catch (err) { + // an injected fault may fail a request, but never corrupt one + if (this.lastFault >= startedAt - 500) return this.count("failed by a fault"); + this.violations.push(`${what} (${src}): ${util.inspect(err)}`); + } + } + + /** wait for the cache to go idle, then check it against the disk */ + async settle(label: string) { + const { diskCache, fdsCache } = this; + const idle = () => + diskCache.unlockedCapacity === diskCache.capacity + && h.fs.open.size === fdsCache.size + && [...fdsCache.values()].every((fd) => h.fs.open.has(fd)); + for (let waited = 0; !idle() && waited < 4000; waited += 10) await delay(10); + if (!idle()) { + const locked = diskCache.capacity - diskCache.unlockedCapacity; + const leaked = [...h.fs.open].filter(([fd]) => !new Set(fdsCache.values()).has(fd)); + this.violations.push( + `${label}: never went idle, ${locked} bytes locked, ` + + `leaked descriptors ${util.inspect(leaked)}, ` + + `descriptor cache ${util.inspect([...fdsCache.values()])}`, + ); + return; + } + + let used = 0; + const names = new Set(); + for (const [file, entry] of diskCache) { + used += entry.size; + if (names.has(file.base)) { + this.violations.push(`${label}: ${file.base} has two entries`); + } + names.add(file.base); + const known = this.files.find((f) => f.hash === file.base); + const onDisk = fs.existsSync(file.toString()) ? fs.readFileSync(file.toString()) : null; + if (entry.written.value !== entry.size) { + this.violations.push(`${label}: ${file.base} is idle but still pending`); + } else if (!onDisk) { + this.violations.push(`${label}: ${file.base} is cached but not on disk`); + } else if (!known || !known.data.equals(onDisk)) { + this.violations.push(`${label}: ${file.base} is corrupt on disk`); + } + } + if (used !== diskCache.used || used > diskCache.capacity) { + this.violations.push(`${label}: holds ${used} bytes, accounted ${diskCache.used}`); + } + for (const name of fs.readdirSync(this.cacheDir)) { + if (name !== "state.json" && !names.has(name)) { + this.violations.push(`${label}: ${name} is on disk but not in the cache`); + } + } + } + + /** restart the process, optionally after damaging the cache directory */ + async restart(label: string) { + const { random, cacheDir } = this; + const before = new Set([...this.diskCache.keys()].map((file) => file.base)); + const damage = random.pick(["none", "none", "truncate", "stray", "state", "nostate"] as const); + const victim = before.size > 0 ? random.pick([...before]) : null; + const state = path.join(cacheDir, "state.json"); + if (damage === "truncate" && victim) { + const { size } = fs.statSync(path.join(cacheDir, victim)); + fs.truncateSync(path.join(cacheDir, victim), random.int(size)); + } else if (damage === "stray") { + fs.writeFileSync(path.join(cacheDir, "0".repeat(40)), "stray"); + } else if (damage === "state" && fs.existsSync(state)) { + fs.truncateSync(state, random.int(fs.statSync(state).size)); + } else if (damage === "nostate") { + fs.rmSync(state, { force: true }); + } + + this.count(`restart ${damage}`); + await this.boot(); + const after = new Set([...this.diskCache.keys()].map((file) => file.base)); + if (damage === "none" && before.size > 0 && before.difference(after).size > 0) { + this.violations.push( + `${label}: restart lost ${before.difference(after).size} of ${before.size} entries`, + ); + } + } + + check() { + // the log module owns `console` here, so write to the descriptor + if (process.env.FUZZ_STATS) fs.writeSync(2, util.inspect(this.stats) + "\n"); + if (this.violations.length === 0) return; + const shown = this.violations.slice(0, 8).join("\n "); + const log = this.log.slice(-40).join("\n "); + assert.fail(`${this.violations.length} violations, first:\n ${shown}\nlog:\n ${log}`); + } +} + +async function withWorld( + seed: number, + chaos: Chaos, + local: boolean, + fn: (world: World) => Promise, +) { + const cwd = process.cwd(); + const world = new World(seed, chaos, local); + try { + await world.start(); + return await fn(world); + } finally { + process.chdir(cwd); + await world.stop(); + } +} + +const quiet: Chaos = { ...calm, lag: 0, shortIo: 0 }; + +describe("issue 106", () => { + test("a descriptor is not closed while a stream reads it", () => + withWorld(106, quiet, false, async (world) => { + const big = UNWRAP(world.files.at(9)); + const others = world.files.filter((f) => f !== big && f.data.length < 70_000); + world.diskCache.capacity = 8_000_000; + + // once to fill the cache, once more to park the descriptor + await world.body(big); + await world.settle("fill"); + await world.body(big); + await world.settle("park"); + + // a paused player holds this response open while a listing loads + const held = await world.request(big); + for (const other of others) await world.body(other); + for (const other of others) await world.body(other); + + const body = Buffer.from(await new Response(held.stream as BodyInit).arrayBuffer()); + assert.ok(body.equals(big.data), "held response was served the wrong bytes"); + await world.settle("end"); + world.check(); + })); + + test("a descriptor is not parked while another request is opening it", () => + withWorld(106, quiet, false, async (world) => { + const big = UNWRAP(world.files.at(9)); + const others = world.files.filter((f) => f !== big && f.data.length < 70_000); + world.diskCache.capacity = 8_000_000; + await world.body(big); + await world.settle("fill"); + + // once its buffers fill, the first stream has no read in flight, so + // cancelling it releases the descriptor before the second request resumes + const first = (await world.request(big)).stream.getReader(); + await delay(50); + const second = world.request(big); + await first.cancel(); + + const held = await second; + for (const other of others) await world.body(other); + const body = Buffer.from(await new Response(held.stream as BodyInit).arrayBuffer()); + assert.ok(body.equals(big.data), "held response was served the wrong bytes"); + await world.settle("end"); + world.check(); + })); + + test("a range that starts past its end leaks nothing", () => + withWorld(106, quiet, false, async (world) => { + const { file, hash, data } = UNWRAP(world.files.at(8)); + await world.body(UNWRAP(world.files.at(8))); + await world.settle("fill"); + + const backwards = { file, derived: null, hash, size: data.length, start: 46226599, end: 25705638 }; + await assert.rejects(world.cache.read(backwards)); + await world.settle("end"); + world.check(); + })); +}); + +describe("fuzz", () => { + const cases = Array.from({ length: runs }, (_, i) => ({ + seed: seed + i, + chaos: (seed + i) % 2 === 0 ? "calm" : "stormy", + local: (seed + i) % 3 === 0, + })); + test.each(cases)( + "seed $seed, $chaos, local share $local", + ({ seed, chaos, local }) => + withWorld(seed, chaos === "calm" ? calm : stormy, local, async (world) => { + const { random } = world; + const inflight = new Set>(); + for (let i = 0; i < opCount && world.violations.length === 0; i++) { + const roll = random(); + if (roll < 0.03) { + await Promise.all(inflight); + await world.settle(`#${i} settle`); + if (world.violations.length === 0 && random.roll(0.5)) { + await world.restart(`#${i} restart`); + } + continue; + } + const op = world.read(`#${i}`).finally(() => inflight.delete(op)); + inflight.add(op); + if (inflight.size >= 12 || random.roll(0.25)) await Promise.race(inflight); + else if (random.roll(0.4)) await delay(random.int(3)); + } + await Promise.all(inflight); + await world.settle("end"); + world.check(); + }), + 120_000, + ); +}); + +const realFs = await vi.importActual("node:fs"); + +import type { Path } from "#sitegen/path"; +import { UNWRAP } from "@clo/lib/assert"; +import { delay } from "@clo/lib/async"; +import type { Lru } from "@clo/lib/Lru"; +import { strict as assert } from "node:assert"; +import * as crypto from "node:crypto"; +import * as fs from "node:fs"; +import * as http from "node:http"; +import type * as net from "node:net"; +import * as os from "node:os"; +import * as path from "node:path"; +import * as util from "node:util"; +import { describe, test, vi } from "vitest"; +import type { MediaFile } from "./models/MediaFile.ts"; diff --git a/src/file-viewer/cache.ts b/src/file-viewer/cache.ts index ce24b81971e9952a92bfe5cf1d276fb1237b77a3..f1fc1b234e75f2849d3a200a0c450cdfb37f79fa 100644 --- a/src/file-viewer/cache.ts +++ b/src/file-viewer/cache.ts @@ -53,45 +53,41 @@ const diskCache = new Lru({ // and held open in the `fdsCache` lru when no references. const fdsCache = new Lru({ capacity: 64, - delete: (value) => void fsCallbacks.closeSync(value), + evict: (fd) => fsCallbacks.closeSync(fd), }); -const fds = new Map>(); - -const parallelRequests = new Map>(); +const fds = new Map(); // Revive disk cache state -revive: { +{ + const revived = new Set([diskCacheState.base]); try { const state = diskCacheState - .readIfExistsSync>("json"); - if (!state) break revive; - const missing = new Set(diskCacheRoot.readDirSync().map((c) => c.base)); - missing.delete(diskCacheState.base); + .readIfExistsSync>("json") ?? []; diskCache.revive(state, { - map(k: string, v: number) { - const path = new Path(k); + map(base: string, size: number) { + const path = diskCacheRoot.join(base); try { - const { size } = path.statSync(); - if (size !== v) return null; - if (!missing.delete(k)) return null; - return [ - path, - { size, written: new async.Watch(size), memory: new Map() }, - ]; + if (path.statSync().size !== size) return null; } catch { return null; } + revived.add(base); + return [ + path, + { size, written: new async.Watch(size), memory: new Map() }, + ]; }, }); - // Delete unreferenced files - for (const missingFile of missing) { - try { - diskCacheRoot.join(missingFile).deleteSync(); - } catch {} - } } catch (err) { console.warn("Failed to restore file viewer cache", err); } + // Delete unreferenced files + for (const file of diskCacheRoot.readDirSync()) { + if (revived.has(file.base)) continue; + try { + file.deleteSync(); + } catch {} + } } export interface Options { @@ -101,7 +97,7 @@ export interface Options { size: number; hash: string; start: number; - /** exclusive. ex: [10, 30] skips ten bytes then grabs 20 */ + /** inclusive, as in an HTTP range. ex: [10, 30] skips ten bytes then grabs 21 */ end: number; noWrite?: boolean; localOnly?: boolean; @@ -112,8 +108,13 @@ export interface Options { * `null` if a derived asset does not exist */ export async function read( - { file, derived, hash, size, start, end, noWrite, localOnly }: Options, + options: Options, ): Promise<{ stream: ReadableStream; src: Source; size: number } | null> { + const { file, derived, hash, size, start, end, noWrite, localOnly } = options; + ASSERT( + Number.isInteger(start) && start >= 0 && start <= end, + () => `Invalid range [${start}, ${end}]`, + ); const rel = derived ? `${file.path}$/${derived}` : file.path; // Serve from local network share if it is mounted. @@ -123,24 +124,19 @@ export async function read( local: { if (!localFile) break local; if (process.env.CLOVER_SOT_FORCE) break local; - let size: number; try { - ({ size } = await localFile.stat()); + const { size } = await localFile.stat(); + const fileStream = await readStream(localFile, { start, end }); + console.info(`from local: ${rel} ` + localFile); + return { + stream: stream.Readable.toWeb(fileStream), + size, + src: "local", + }; } catch (err) { - if (exception.code(err) === "ENOENT") { - console.warn(`missing local file: ${localFile}`); - break local; - } - throw err; + if (exception.code(err) !== "ENOENT") throw err; + console.warn(`missing local file: ${localFile}`); } - const fileStream = await readStream(localFile, { start, end }); - if (!fileStream) break local; - console.info(`from local: ${rel} ` + localFile); - return { - stream: stream.Readable.toWeb(fileStream), - size, - src: "local", - }; } if (localOnly) return null; @@ -151,200 +147,206 @@ export async function read( // Disk cache const diskPath = diskCacheRoot.join(hash); let diskFile = diskCache.get(diskPath); + let src: Source = "hit"; if (diskFile) { - const lock = diskCache.lock(diskPath); + if (diskFile.written.value !== diskFile.size) src = "pending"; + console.info(`from ${src === "pending" ? "pending " : ""}disk cache: ${rel}`); + } else { + // Do not cache large assets + if (size > 1_000_000_000) { + console.info(`skipping disk cache due to large asset: ${rel}`); + return { + stream: await fetchAndSlice(rel, url, size, start, end), + size, + src: "ignored", + }; + } + // Bail out of the disk cache if it is contended + if (diskCache.unlockedCapacity < size) { + console.info(`skipping disk cache due to contention: ${rel}`); + return { + stream: await fetchAndSlice(rel, url, size, start, end), + size, + src: "contended", + }; + } + // IPs which request too many cache misses are temporarily + // banned from writing into the disk cache to avoid abuse. + if (noWrite) { + console.info(`skipping disk cache due to config: ${rel}`); + return { + stream: await fetchAndSlice(rel, url, size, start, end), + size, + src: "ratelimit", + }; + } + + // Create a disk cache entry. Every request, this one included, reads the + // file back as it lands so that a slow client never stalls the download. + diskFile = { size, written: new async.Watch(0), memory: new Map() }; + diskCache.set(diskPath, diskFile); + void download(rel, url, diskPath, diskFile); + src = "miss"; + } + + const lock = diskCache.lock(diskPath); + try { const diskStream = await readStream(diskPath, { ref: diskFile, - start: start, - end: end, + start, + end, lock, }); - if (diskStream) { - const isPending = diskFile.written.value !== diskFile.size; - - // TODO: fix this? i'm suspicious - if (isPending) { - diskStream.close(); - return { - stream: await fetchAndSlice(rel, url, start, end), - size, - src: "contended", - }; - } - - console.info(`from ${isPending ? "pending " : ""}disk cache: ${rel}`); - return { - stream: stream.Readable.toWeb(diskStream), - src: isPending ? "pending" : "hit", - size: diskFile.size, - }; - } else { - diskCache.delete(diskPath); - } - } - - // Do not cache large assets - if (size > 1_000_000_000) { - console.info(`skipping disk cache due to large asset: ${rel}`); - return { - stream: await fetchAndSlice(rel, url, start, end), - size, - src: "ignored", - }; - } - // Bail out of the disk cache if it is contended - if (diskCache.unlockedCapacity < size) { - console.info(`skipping disk cache due to contention: ${rel}`); - return { - stream: await fetchAndSlice(rel, url, start, end), - size, - src: "contended", - }; - } - // IPs which request too many cache misses are temporarily - // banned from writing into the disk cache to avoid abuse. - if (noWrite) { - console.info(`skipping disk cache due to config: ${rel}`); return { - stream: await fetchAndSlice(rel, url, start, end), - size, - src: "ratelimit", + stream: stream.Readable.toWeb(diskStream), + src, + size: diskFile.size, }; + } catch (err) { + lock(); + if (src === "miss" || exception.code(err) !== "ENOENT") throw err; + // The file vanished from under its entry, so cache it again. + if (diskCache.get(diskPath) === diskFile) diskCache.delete(diskPath); + return read(options); } - - // Create a disk cache entry; - // Tee the stream to both the request and the filesystem. - diskFile = { size, written: new async.Watch(0), memory: new Map() }; - diskCache.set(diskPath, diskFile); - const lock = diskCache.lock(diskPath); - return await fetchAndStore( - rel, - url, - size, - diskPath, - diskFile, - start, - end, - lock, - ); } -async function fetchAndStore( +/** Fill a new disk cache entry. Readers follow along through `diskFile.written`. */ +async function download( rel: string, url: string, - size: number, diskPath: Path, diskFile: DiskFile, - start: number, - end: number, - lock: ts.Dispose, ) { + const lock = diskCache.lock(diskPath); + // Readers wait on this download, so a stalled connection has to fail it. + const response = fetch(rel, url, { timeout: 30_000 }); try { - let response = await fetch(rel, url); - const { statusCode } = response; - if (statusCode !== 200) { - const text = await new Response(response as unknown as Blob).text(); - const msg = `Source of Truth responded with status ${statusCode}: ${text}`; - console.error(msg); - throw new Error("Source of Truth is unavailable"); + // Opening only after the response arrives makes every reader sharing this + // handle wait on the Source of Truth, and fail with it. + const handle = await getHandle(diskPath, async () => { + await response; + return await openFile(diskPath.toString(), "w+"); + }); + + const { memory, written } = diskFile; + const writer = fs.createWriteStream(diskPath.toString(), { + start: 0, + fd: handle.fd, + fs: { + close(_: number, cb: (er: Error | null) => void) { + handle.release(); + cb(null); + }, + write( + fd: number, + data: Buffer, + off: 0, + size: number, + pos: number | undefined, + cb: (er: Error | null, bytesWritten: number, buffer: Buffer) => void, + ) { + ASSERT(size === data.byteLength); + ASSERT(pos != null); + fsCallbacks.write(fd, data, off, size, pos, (er, bytes) => { + if (!er && bytes) { + remember(memory, pos, data); + written.value += bytes; + } + cb(er, bytes, data); + }); + }, + }, + }); + await streamPromises.pipeline(await response, writer); + + if (written.value !== diskFile.size) { + throw new Error(`Expected ${diskFile.size} bytes, received ${written.value}`); } - const writer = await writeStream(diskPath, response, diskFile, lock); - writer.on("error", () => { - diskCache.delete(diskPath); - diskFile.written.value = Infinity; - }); - - return { - stream: sliceStream(response, start, end), - size, - src: "miss", - } as const; - } catch (e) { + writeDiskCache(); + } catch (err) { + console.error(`Failed to cache ${rel}:`, err); + response.then((opened) => opened.destroy(), () => {}); + if (diskCache.get(diskPath) === diskFile) diskCache.delete(diskPath); + diskFile.written.value = Infinity; + } finally { lock(); - throw e; } } -async function fetch(rel: string, url: string, headers?: {}) { - let promise = !headers ? parallelRequests.get(url) : null; - if (promise) { - console.info(`from other response: ${rel}`); - return await promise; - } else { - console.info(`cache miss: ${rel}`); - promise = httpGet(url, { - headers: { - ...token ? { Authorization: token } : null, - ...headers, - }, - }); - if (!headers) parallelRequests.set(url, promise); - const response = await promise; - if (!headers) parallelRequests.delete(url); - return response; +async function fetch(rel: string, url: string, options: https.RequestOptions = {}) { + console.info(`cache miss: ${rel}`); + const response = await httpGet(url, { + ...options, + headers: { + ...token ? { Authorization: token } : null, + ...options.headers, + }, + }); + const { statusCode } = response; + if (statusCode !== 200 && statusCode !== 206) { + const text = await consumers.text(response); + const msg = `Source of Truth responded with status ${statusCode}: ${text}`; + console.error(msg); + throw new Error("Source of Truth is unavailable"); } + return response; } +/** Stream from the Source of Truth, skipping the disk cache. */ async function fetchAndSlice( rel: string, url: string, + size: number, start: number, end: number, -) { - const headers = start !== 0 || end !== Infinity - ? { "Clover-Start": String(start), "Clover-End": String(end) } - : null; - return stream.Readable.toWeb(await fetch(rel, url, headers)); +): Promise { + const ranged = start !== 0 || end !== Infinity; + const response = await fetch( + rel, + url, + ranged ? { headers: { Range: `bytes=${start}-${end}` } } : {}, + ); + const partial = response.statusCode === 206; + + // The caller has already promised a length, which a short body would break. + const expected = partial ? end - start + 1 : size; + const length = response.headers["content-length"]; + if (length !== undefined && Number(length) !== expected) { + response.destroy(); + throw new Error(`Source of Truth has ${length} bytes of ${rel}, expected ${expected}`); + } + + // A server is free to ignore `Range` and respond with the whole file. + if (ranged && !partial) { + return ReadableStream.from(sliceStream(response, start, end)); + } + return stream.Readable.toWeb(response); } -export function sliceStream( - readable: stream.Readable, +/** `end` is inclusive. Returning early destroys the rest of `readable`. */ +async function* sliceStream( + readable: AsyncIterable, start: number, end: number, -): Readable { - const length = end === Infinity ? Infinity : end - start; - let recieved = 0; - let sent = 0; - - const transform = new stream.Transform({ - transform(chunk: Buffer, _, callback) { - const chunkEnd = recieved + chunk.length; - - if (chunkEnd <= start) { - recieved = chunkEnd; - callback(); - return; - } - - if (sent >= length) { - callback(); - return; - } - - const sliceStart = Math.max(0, start - recieved); - const remaining = length - sent; - const sliceEnd = end === Infinity - ? chunk.length - : Math.min(chunk.length, sliceStart + remaining); - - const slice = chunk.slice(sliceStart, sliceEnd); - sent += slice.length; - recieved = chunkEnd; - - this.push(slice); - - callback(); - }, - }); - - readable.on("error", (err) => transform.destroy(err)); - - return readable.pipe(transform); +) { + let offset = 0; + for await (const chunk of readable) { + const slice = chunk.subarray( + Math.max(0, start - offset), + Math.max(0, end + 1 - offset), + ); + offset += chunk.length; + if (slice.length > 0) yield slice; + if (offset > end) return; + } } interface DiskFile { size: number; + /** bytes on disk so far, or `Infinity` once the download has failed */ written: async.Watch; + /** chunks recently written or read, keyed by their offset in the file */ memory: Map>; } @@ -356,28 +358,42 @@ function httpGet(url: string, options: https.RequestOptions) { ); request.on("response", resolve); request.on("error", reject); + request.on("timeout", () => { + request.destroy(new Error(`Source of Truth timed out: ${url}`)); + }); }); } +type Memory = DiskFile["memory"]; const finalization = new FinalizationRegistry( - ([map, key]: [Map, number]) => { - map.delete(key); + ([memory, offset, ref]: [Memory, number, WeakRef]) => { + if (memory.get(offset) === ref) memory.delete(offset); }, ); +function remember(memory: Memory, offset: number, chunk: Buffer) { + const ref = new WeakRef(chunk); + memory.set(offset, ref); + finalization.register(chunk, [memory, offset, ref]); +} + const openFile = util.promisify(fsCallbacks.open); function deleteCachedPath(path: Path) { - try { + // Streams still holding the descriptor close it instead of parking it. + fds.delete(path); + const parked = fdsCache.get(path); + if (parked !== undefined) { fdsCache.delete(path); - } catch {} + fsCallbacks.closeSync(parked); + } try { path.deleteSync(); } catch {} } interface RefCountFile { - fd: number; + fd: Promise; refs: number; } @@ -394,42 +410,54 @@ function writeDiskCache() { const serialized = diskCache.serialize({ map(path, { size, written }) { if (size !== written.value) return null; - return [path.toString(), size]; + return [path.base, size]; }, }); - diskCacheState.writeJsonFile(serialized) - .catch((err) => console.error(`Failed to write ${diskCacheState.relative()}:`, err)); + // Renamed into place so a crash cannot leave half a file to revive from. + const partial = `${diskCacheState}.tmp`; + try { + fsCallbacks.writeFileSync(partial, JSON.stringify(serialized)); + fsCallbacks.renameSync(partial, diskCacheState.toString()); + } catch (err) { + console.error(`Failed to write ${diskCacheState.relative()}:`, err); + } } -async function getHandle(path: Path, flag: string, unlockOnFail?: Disposable) { - let handle: RefCountFile | null = null; +/** `open` only runs when `path` has no holders and no parked descriptor. */ +async function getHandle(path: Path, open: () => Promise) { + let handle = fds.get(path); + if (!handle) { + const parked = fdsCache.get(path); + if (parked !== undefined) fdsCache.delete(path); + handle = { + fd: parked === undefined ? open() : Promise.resolve(parked), + refs: 0, + }; + fds.set(path, handle); + } + const shared = handle; + // Counted before the first await, or a release in between could park a + // descriptor that is about to be used. + shared.refs += 1; + + let fd: number; try { - handle = await fds.get(path) ?? null; - if (!handle) { - const cached = fdsCache.get(path); - if (cached) { - fds.delete(path); - fds.set(path, handle = { fd: cached, refs: 0 }); - } else { - const promise = openFile(path.raw, flag) - .then((fd) => ({ fd, refs: 0 })) - .catch((err) => { - fds.delete(path); - throw err; - }); - fds.set(path, promise); - fds.set(path, handle = await promise); - } - } - handle.refs += 1; - } catch (err: any) { - unlockOnFail?.[Symbol.dispose](); - if (err.code === "ENOENT") { - return null; - } + fd = await shared.fd; + } catch (err) { + shared.refs -= 1; + if (fds.get(path) === shared) fds.delete(path); throw err; } - return handle; + + return { + fd, + release() { + if ((shared.refs -= 1) > 0) return; + if (fds.get(path) !== shared) return void fsCallbacks.closeSync(fd); + fds.delete(path); + fdsCache.set(path, fd); + }, + }; } interface ReadStreamOptions { @@ -439,15 +467,19 @@ interface ReadStreamOptions { lock?: Disposable; } -async function readStream(path: Path, opts: ReadStreamOptions) { - const handle = await getHandle(path, "r", opts?.lock); - if (!handle) return null; +/** Disposes `lock` when the stream closes, but not if opening it throws. */ +async function readStream( + path: Path, + { start, end, ref, lock }: ReadStreamOptions, +) { + const handle = await getHandle(path, () => openFile(path.toString(), "r")); let read = fsCallbacks.read; - if (opts?.ref) { - const { memory, written, size } = opts.ref; - let lastRead: number | null = null; - read = function loop( + if (ref) { + const { memory, written, size } = ref; + /** offset of the chunk in `memory` that the previous read stopped inside */ + let partial: number | null = null; + read = function( fd: number, buf: Buffer, offset: 0, @@ -459,155 +491,70 @@ async function readStream(path: Path, opts: ReadStreamOptions) { ASSERT(offset === 0); written.until((v) => v >= Math.min(size, pos + len)).then( (newLength) => { + if (newLength > size) return cb(new Error("Broken Pipe"), 0, buf); + // Try to re-use uncollected buffers in memory - while (true) { - const weak = memory.get(lastRead ?? pos); - if (weak) { - let cachedBuf = weak.deref(); - if (cachedBuf) { - if (lastRead) { - cachedBuf = cachedBuf.subarray(pos - lastRead); - } - const { byteLength } = cachedBuf; - - if (byteLength === len && offset === 0) { - cb(null, len, cachedBuf); - return; - } - if (byteLength <= len) { - buf.set(cachedBuf, offset); - lastRead = null; - } else { - buf.set(cachedBuf.subarray(0, len), offset); - lastRead = lastRead ?? pos; - } - offset += byteLength; - len -= byteLength; - pos += byteLength; - if (len > 0) continue; - cb(null, offset + len, buf); - return; - } else { - memory.delete(pos); - } + let filled = 0; + while (filled < len) { + const at = pos + filled; + const key = partial ?? at; + const chunk = memory.get(key)?.deref()?.subarray(at - key); + partial = null; + if (!chunk?.byteLength) { + if (key === at) break; + continue; + } + if (filled === 0 && chunk.byteLength === len) { + return cb(null, len, chunk); } - lastRead = null; - break; + const slice = chunk.subarray(0, len - filled); + buf.set(slice, filled); + filled += slice.byteLength; + if (slice.byteLength < chunk.byteLength) partial = key; } + if (filled === len) return cb(null, len, buf); // Read from disk - if (newLength > size) cb(new Error("Broken Pipe"), 0, buf); - fsCallbacks.read( - fd, - buf, - offset, - len, - pos, - async (er, bytes, buf) => { - if (!er && bytes > 0) { - memory.set(pos - offset, new WeakRef(buf)); - finalization.register(buf, [memory, pos]); - } - cb(er, offset + bytes, buf); - }, - ); + fsCallbacks.read(fd, buf, filled, len - filled, pos + filled, (er, bytes) => { + // A short read leaves the tail of `buf` uninitialized. + if (!er && filled === 0 && bytes === len) remember(memory, pos, buf); + cb(er, filled + bytes, buf); + }); }, ); } as typeof fsCallbacks.read; } - let closed = false; return fs.createReadStream(path.toString(), { fd: handle.fd, - start: opts?.start ?? 0, - end: opts?.end ?? Infinity, + start, + end, fs: { - close: util.callbackify(async () => { - ASSERT(!closed); - if ((handle.refs -= 1) <= 0) { - fds.delete(path); - fdsCache.set(path, handle.fd); - } - closed = true; - opts?.lock?.[Symbol.dispose](); - }), + close(_: number, cb: (er: Error | null) => void) { + handle.release(); + lock?.[Symbol.dispose](); + cb(null); + }, read, }, }); } -async function writeStream( - path: Path, - data: stream.Readable, - file: DiskFile, - lock: Disposable, -) { - const handle = UNWRAP(await getHandle(path, "w+", lock)); - let closed = false; - - const { memory, written } = file; - const writer = fs.createWriteStream(path.toString(), { - start: 0, - fd: handle.fd, - fs: { - close: util.callbackify(async () => { - ASSERT(!closed); - if ((handle.refs -= 1) <= 0) { - fds.delete(path); - fdsCache.set(path, handle.fd); - } - closed = true; - lock[Symbol.dispose](); - writeDiskCache(); - }), - write( - fd: number, - data: Buffer, - off: 0, - size: number, - pos: number | undefined, - cb: (er: Error | null, bytesWritten: number, buffer: Buffer) => void, - ) { - ASSERT(size === data.byteLength); - ASSERT(pos != null); - fsCallbacks.write( - fd, - data, - off, - size, - pos, - (er, bytes, buf) => { - if (!er && bytes) { - memory.set(pos, new WeakRef(buf)); - finalization.register(buf, [memory, pos]); - written.value += bytes; - } - cb(er, bytes, buf); - }, - ); - }, - }, - }); - data.pipe(writer); - - return writer; -} - import * as fs from "#sitegen/fs"; import { Path } from "#sitegen/path"; -import { ASSERT, UNWRAP } from "@clo/lib/assert"; +import { ASSERT } from "@clo/lib/assert"; import * as async from "@clo/lib/async"; import * as exception from "@clo/lib/exception"; import * as log from "@clo/lib/log"; import { Lru } from "@clo/lib/Lru"; -import * as ts from "@clo/lib/ts"; -import * as crypto from "node:crypto"; import * as fsCallbacks from "node:fs"; import * as http from "node:http"; import * as https from "node:https"; import * as stream from "node:stream"; -import type { ReadableStream } from "node:stream/web"; +import * as consumers from "node:stream/consumers"; +import * as streamPromises from "node:stream/promises"; +import { ReadableStream } from "node:stream/web"; import * as util from "node:util"; -import { MediaFile } from "./models/MediaFile.ts"; +import type { MediaFile } from "./models/MediaFile.ts"; import * as paths from "./paths.ts"; diff --git a/src/source-of-truth.test.ts b/src/source-of-truth.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..eaa10fc88a38d02e5447035473b3b01949c6ec73 --- /dev/null +++ b/src/source-of-truth.test.ts @@ -0,0 +1,51 @@ +const dir = vi.hoisted(() => { + const fs = process.getBuiltinModule("node:fs"); + const os = process.getBuiltinModule("node:os"); + const dir = fs.mkdtempSync(os.tmpdir() + "/clover-sot-test-"); + fs.mkdirSync(dir + "/raw"); + fs.mkdirSync(dir + "/derived"); + Object.assign(process.env, { + CLOVER_DB: dir + "/db", + CLOVER_FILE_RAW: dir + "/raw", + CLOVER_FILE_DERIVED: dir + "/derived", + CLOVER_SOT_KEY: "test", + }); + return dir; +}); +afterAll(() => fs.rmSync(dir, { recursive: true, force: true })); + +const data = crypto.randomBytes(1_000_000); +const url = "http://localhost/file/new%20home%20[rev%203].mp3"; +fs.writeFileSync(dir + "/raw/new home [rev 3].mp3", data); +MediaFile.createFile({ + path: "/new home [rev 3].mp3", + date: new Date(0), + hash: crypto.createHash("sha1").update(data).digest("hex"), + size: data.length, + duration: 0, + dimensions: "", + contents: "", +}); + +test("requests for the same file do not share a file offset", async () => { + // the second request arrives once the first holds the descriptor + const first = await app.fetch(new Request(url)); + const second = await app.fetch(new Request(url)); + const bodies = await Promise.all([first.arrayBuffer(), second.arrayBuffer()]); + for (const body of bodies) assert.ok(data.equals(Buffer.from(body))); +}); + +test("a range is served as partial content", async () => { + const response = await app.fetch(new Request(url, { headers: { Range: "bytes=70000-" } })); + assert.equal(response.status, 206); + assert.equal(response.headers.get("Content-Range"), `bytes 70000-999999/1000000`); + assert.equal(response.headers.get("Content-Length"), "930000"); + assert.ok(data.subarray(70000).equals(Buffer.from(await response.arrayBuffer()))); +}); + +import { strict as assert } from "node:assert"; +import * as crypto from "node:crypto"; +import * as fs from "node:fs"; +import { afterAll, test, vi } from "vitest"; +import { MediaFile } from "./file-viewer/models/MediaFile.ts"; +import app from "./source-of-truth.ts"; diff --git a/src/source-of-truth.ts b/src/source-of-truth.ts index 0d5d7723b923d5837f93247d11ffd7847e880ace..90754af552d387345b8427731f0c14b55bb30b2b 100644 --- a/src/source-of-truth.ts +++ b/src/source-of-truth.ts @@ -100,9 +100,21 @@ app.get("/file/*", async (c) => { if (err.code === "ENOENT") return c.notFound(); throw err; } - c.header("Content-Length", (await fstat(handle.fd)).size.toString()); + const { size } = await fstat(handle.fd); + let start = 0; + let end = Infinity; + const ranges = http.parseRangeHeader(c.req.header("Range") ?? "", size); + if (ranges?.[0] && ranges.length === 1) { + ({ start, end } = ranges[0]); + c.status(206); + c.header("Content-Range", `bytes ${start}-${end}/${size}`); + } + c.header("Content-Length", String(Math.min(end, size - 1) - start + 1)); const nodeStream = fs.createReadStream(fullPath, { fd: handle.fd, + // Positioned, or requests sharing this descriptor would share its offset. + start, + end, fs: { close: util.callbackify(async () => { ASSERT(handle); -- 2.54.0