1// implements the file viewer's cache. there are two primary exports:
2// - `cache.read(...)` to read a file, caching it
3// - TODO: `cache.prefetch(...)` to prefetch a file, placing it in the disk cache.
4//
5// in local mode, the cache layer reads directly the mounted network share.
6// when unavailable, it fetches from the Source of Truth, caching its
7// results on disk. one `Lru` is used to keep the most relevant files in
8// the store, while another one is used to re-use file descriptors.
9//
10// the cache handles concurrent requests to the same disk entry by placing
11// its file descriptor in "w+" mode for reading and writing. this way, the
12// second request can read from the same file descriptor that the first
13// request is writing into. large files are never buffered entirely in
14// memory, which should keep memory usage low.
15//
16// as a measure to make this potentially more memory efficient, the buffer
17// objects given by node:fs/node:http streams -- which are re-allocated for
18// every read -- are weakly held when written and reused when reading. this
19// somewhat avoids file reads in concurrent access situations.
20const console = log.scoped("cache");
21
22export type Source =
23 | "local" // local mode
24 | "hit" // cache hit
25 | "pending" // cache hit but the file is still being loaded
26 | "ignored" // file too big
27 | "contended" // could not write disk because too many ongoing downloads
28 | "ratelimit" // the user is ratelimited for inserting files
29 | "miss"; // cache miss, disk cache is created
30
31// to download protected files from the source of truth -- for example files
32// used on the private blog for friends -- its token is required.
33const token = process.env.CLOVER_SOT_KEY ?? null;
34const sotUrl = new URL(
35 process.env.CLOVER_SOT_URL ?? "https://db.paperclover.net",
36).toString().replace(/\/+$/g, "");
37if (!token) console.warn("Missing CLOVER_SOT_KEY, cannot host protected files");
38
39const diskCacheRoot = openDiskCache();
40const diskCacheState = diskCacheRoot.join("state.json");
41const capacityInBytes = 8 * 1024 * 1024 * 1024;
42
43const derivedFileRoot = new Path(paths.derivedFileRoot).ifExistsSync();
44const rawFileRoot = new Path(paths.rawFileRoot).ifExistsSync();
45
46const diskCache = new Lru<Path, DiskFile>({
47 capacity: capacityInBytes,
48 sizeFn: ({ size }) => size,
49 delete: (_, path) => deleteCachedPath(path),
50});
51
52// When reading from the disk cache, file descriptors are ref-counted in `fds`,
53// and held open in the `fdsCache` lru when no references.
54const fdsCache = new Lru<Path, number>({
55 capacity: 64,
56 evict: (fd) => fsCallbacks.closeSync(fd),
57});
58const fds = new Map<Path, RefCountFile>();
59
60// Revive disk cache state
61{
62 const revived = new Set([diskCacheState.base]);
63 try {
64 const state = diskCacheState
65 .readIfExistsSync<Lru.Serialized<string, number>>("json") ?? [];
66 diskCache.revive(state, {
67 map(base: string, size: number) {
68 const path = diskCacheRoot.join(base);
69 try {
70 if (path.statSync().size !== size) return null;
71 } catch {
72 return null;
73 }
74 revived.add(base);
75 return [
76 path,
77 { size, written: new async.Watch(size), memory: new Map() },
78 ];
79 },
80 });
81 } catch (err) {
82 console.warn("Failed to restore file viewer cache", err);
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 }
91}
92
93export interface Options {
94 file: MediaFile;
95 /** If set, fetches the derived asset instead. */
96 derived: string | null;
97 size: number;
98 hash: string;
99 start: number;
100 /** inclusive, as in an HTTP range. ex: [10, 30] skips ten bytes then grabs 21 */
101 end: number;
102 noWrite?: boolean;
103 localOnly?: boolean;
104}
105
106/**
107 * Read a value from file store, potentially fetching from the Source of Truth.
108 * `null` if a derived asset does not exist
109 */
110export async function read(
111 options: Options,
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 );
118 const rel = derived ? `${file.path}$/${derived}` : file.path;
119
120 // Serve from local network share if it is mounted.
121 const localFile = derived
122 ? derivedFileRoot?.join(derived)
123 : rawFileRoot?.join(file.path);
124 local: {
125 if (!localFile) break local;
126 if (process.env.CLOVER_SOT_FORCE) break local;
127 try {
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 };
136 } catch (err) {
137 if (exception.code(err) !== "ENOENT") throw err;
138 console.warn(`missing local file: ${localFile}`);
139 }
140 }
141
142 if (localOnly) return null;
143
144 const suffix = derived ? ":/" + derived.split("/").slice(1).join("/") : "";
145 const url = `${sotUrl}/file${file.path}${suffix}`;
146
147 // Disk cache
148 const diskPath = diskCacheRoot.join(hash);
149 let diskFile = diskCache.get(diskPath);
150 let src: Source = "hit";
151 if (diskFile) {
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}`);
158 return {
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",
181 };
182 }
183
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";
190 }
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 });
200 return {
201 stream: stream.Readable.toWeb(diskStream),
202 src,
203 size: diskFile.size,
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);
211 }
212}
213
214/** Fill a new disk cache entry. Readers follow along through `diskFile.written`. */
215async function download(
216 rel: string,
217 url: string,
218 diskPath: Path,
219 diskFile: DiskFile,
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 });
224 try {
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+");
230 });
231
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 {
273 lock();
274 }
275}
276
277async 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");
292 }
293 return response;
294}
295
296/** Stream from the Source of Truth, skipping the disk cache. */
297async function fetchAndSlice(
298 rel: string,
299 url: string,
300 size: number,
301 start: number,
302 end: number,
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);
325}
326
327/** `end` is inclusive. Returning early destroys the rest of `readable`. */
328async function* sliceStream(
329 readable: AsyncIterable<Buffer>,
330 start: number,
331 end: number,
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}
344
345interface DiskFile {
346 size: number;
347 /** bytes on disk so far, or `Infinity` once the download has failed */
348 written: async.Watch<number>;
349 /** chunks recently written or read, keyed by their offset in the file */
350 memory: Map<number, WeakRef<Buffer>>;
351}
352
353function httpGet(url: string, options: https.RequestOptions) {
354 return new Promise<http.IncomingMessage>((resolve, reject) => {
355 const request = (url.startsWith("https://") ? https : http).get(
356 url,
357 options,
358 );
359 request.on("response", resolve);
360 request.on("error", reject);
361 request.on("timeout", () => {
362 request.destroy(new Error(`Source of Truth timed out: ${url}`));
363 });
364 });
365}
366
367type Memory = DiskFile["memory"];
368const finalization = new FinalizationRegistry(
369 ([memory, offset, ref]: [Memory, number, WeakRef<Buffer>]) => {
370 if (memory.get(offset) === ref) memory.delete(offset);
371 },
372);
373
374function 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
380const openFile = util.promisify(fsCallbacks.open);
381
382function deleteCachedPath(path: Path) {
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) {
387 fdsCache.delete(path);
388 fsCallbacks.closeSync(parked);
389 }
390 try {
391 path.deleteSync();
392 } catch {}
393}
394
395interface RefCountFile {
396 fd: Promise<number>;
397 refs: number;
398}
399
400function openDiskCache() {
401 const clover = ".clover", fileCache = "file-viewer";
402 const split = import.meta.filename.split(Path.sep);
403 const index = split.indexOf(".clover");
404 return index === -1
405 ? Path.createDirSync(clover, fileCache)
406 : Path.createDirSync(split.slice(0, index + 1).join(Path.sep), fileCache);
407}
408
409function writeDiskCache() {
410 const serialized = diskCache.serialize({
411 map(path, { size, written }) {
412 if (size !== written.value) return null;
413 return [path.base, size];
414 },
415 });
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 }
424}
425
426/** `open` only runs when `path` has no holders and no parked descriptor. */
427async 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;
444 try {
445 fd = await shared.fd;
446 } catch (err) {
447 shared.refs -= 1;
448 if (fds.get(path) === shared) fds.delete(path);
449 throw err;
450 }
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 };
461}
462
463interface ReadStreamOptions {
464 start: number;
465 end: number;
466 ref?: DiskFile;
467 lock?: Disposable;
468}
469
470/** Disposes `lock` when the stream closes, but not if opening it throws. */
471async function readStream(
472 path: Path,
473 { start, end, ref, lock }: ReadStreamOptions,
474) {
475 const handle = await getHandle(path, () => openFile(path.toString(), "r"));
476
477 let read = fsCallbacks.read;
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(
483 fd: number,
484 buf: Buffer,
485 offset: 0,
486 len: number,
487 pos: number,
488 cb: (er: Error | null, bytesRead: number, buf: Buffer) => void,
489 ) {
490 ASSERT(pos != null);
491 ASSERT(offset === 0);
492 written.until((v) => v >= Math.min(size, pos + len)).then(
493 (newLength) => {
494 if (newLength > size) return cb(new Error("Broken Pipe"), 0, buf);
495
496 // Try to re-use uncollected buffers in memory
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);
509 }
510
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;
515 }
516 if (filled === len) return cb(null, len, buf);
517
518 // Read from disk
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 });
524 },
525 );
526 } as typeof fsCallbacks.read;
527 }
528
529 return fs.createReadStream(path.toString(), {
530 fd: handle.fd,
531 start,
532 end,
533 fs: {
534 close(_: number, cb: (er: Error | null) => void) {
535 handle.release();
536 lock?.[Symbol.dispose]();
537 cb(null);
538 },
539 read,
540 },
541 });
542}
543
544import * as fs from "#sitegen/fs";
545import { Path } from "#sitegen/path";
546import { ASSERT } from "@clo/lib/assert";
547import * as async from "@clo/lib/async";
548import * as exception from "@clo/lib/exception";
549import * as log from "@clo/lib/log";
550import { Lru } from "@clo/lib/Lru";
551import * as fsCallbacks from "node:fs";
552import * as http from "node:http";
553import * as https from "node:https";
554import * as stream from "node:stream";
555import * as consumers from "node:stream/consumers";
556import * as streamPromises from "node:stream/promises";
557import { ReadableStream } from "node:stream/web";
558import * as util from "node:util";
559import type { MediaFile } from "./models/media-files.ts";
560import * as paths from "./paths.ts";