// The on-disk content and thumbnail cache keyed by fileID (issue #46). // // Layout under `cacheDirectory`: `originals/.` and // `thumbnails/.`, flat directories at 0700 with files at 0600. // Content appears only by the streaming atomic writer's rename (the download // layer, #40), so a file that exists is whole — "present means complete". The // directory listing taken at `open()` is the record of what is cached, and the // orphan temp files a crashed write may have left are reaped there. // // A fetch goes through the shared request pools (#45): the content pool for // originals, the thumbnail pool for thumbnails. The pool limits concurrency, // orders on-demand work ahead of background, and dedups by key so a fileID // requested twice while the first is still in flight downloads once. // // Integrity. The reused streaming decrypt is the enforced guarantee: every // chunk is authenticated and the writer renames the file into place only once // the stream ends on TAG_FINAL, so a truncated or corrupt fetch throws and // nothing is stored. On top of that this module refuses to record a stored file // that came out empty. The design also asks for a content-hash comparison // against `FileMetadata.hash` (with a `fileSize` fallback); that is deferred — // see the PR — because the exact hash construction cannot be confirmed against // the repo's fixtures and `FileBlob.size` is the encrypted object size, not the // decrypted length this layer has. import { existsSync, statSync } from "node:fs"; import { chmod, mkdir, readdir, rm, stat, statfs, utimes, } from "node:fs/promises"; import { dirname, extname, join } from "node:path"; import type { ApiClient } from "../api/client.js"; import { downloadFile, downloadThumbnail, type ProgressCallback, } from "../download/index.js"; import type { EnteFile } from "../model/types.js"; import type { Priority, RequestPools } from "./pools.js"; const DIR_MODE = 0o700; const FILE_MODE = 0o600; const TEMP_PREFIX = ".quak-"; const TEMP_SUFFIX = ".tmp"; const GIB = 1024 * 1024 * 1024; // Owner ruling (#36): bound the originals cache at 100 GiB, but back off when // the volume has under 50 GiB free so the cache never crowds the disk. export const DEFAULT_ORIGINALS_MAX_BYTES = 100 * GIB; export const DEFAULT_FREE_BELOW_BYTES = 50 * GIB; // Ente thumbnails are always JPEG, so the cache stores them with a fixed // extension rather than deriving one from the (image or video) title. const THUMBNAIL_EXT = ".jpg"; type Kind = "original" | "thumbnail"; // The priority a caller attaches to a thumbnail prefetch. The pool has two // tiers, so this three-value surface collapses onto them: only a currently // visible thumbnail preempts (on-demand); "ahead" prefetch and speculative // "background" work both yield to it. export type ThumbnailPriority = "visible" | "ahead" | "background"; const poolPriorityOf = (priority: ThumbnailPriority): Priority => priority === "visible" ? "on-demand" : "background"; export interface ContentResult { path: string; bytes: number; } // Progress for a single `original`/`thumbnail` call. A present file emits one // `skipped` event and nothing else; a fetched file emits `downloading` as // plaintext lands and a final `done`. export type ContentEvent = | { status: "skipped"; bytes: number } | { status: "downloading"; bytesDone: number } | { status: "done"; bytes: number }; export interface ContentOptions { onProgress?: (event: ContentEvent) => void; } // The Photo-facing content surface (the read wrappers call these). The cache // implements it; a library opened without a content source leaves it absent. export interface PhotoContent { original(fileID: number, opts?: ContentOptions): Promise; thumbnail(fileID: number, opts?: ContentOptions): Promise; } export interface EnsureResult { fileID: number; path?: string; error?: string; } export interface EnsureEvent { fileID: number; status: "skipped" | "done" | "failed" | "aborted"; path?: string; error?: string; } export interface EnsureOptions { fileIDs: number[]; priority: ThumbnailPriority; signal?: AbortSignal; onProgress?: (event: EnsureEvent) => void; } export interface ThumbnailsAPI { ensure(args: EnsureOptions): Promise; } // The byte source the cache fetches through. The real implementation streams // and decrypts to the destination via the download layer; tests inject a // stand-in so the cache logic runs with no crypto and no network. Pool routing, // dedup, present-checks and integrity live in the cache, not here. export interface ContentSource { original(args: { file: EnteFile; destination: string; onProgress?: ProgressCallback; }): Promise<{ bytesWritten: number }>; thumbnail(args: { file: EnteFile; destination: string; onProgress?: ProgressCallback; }): Promise<{ bytesWritten: number }>; } // The production source: each fetch is the download layer's request + // streaming decrypt + atomic write + retry as one unit. export const makeDownloadContentSource = (api: ApiClient): ContentSource => ({ original: ({ file, destination, onProgress }) => downloadFile(api, file, destination, onProgress), thumbnail: ({ file, destination, onProgress }) => downloadThumbnail(api, file, destination, onProgress), }); export interface CachedPaths { originalPath?: string; thumbnailPath?: string; } // The slice of `fs.statfs` the eviction limit needs: `bavail` is the blocks // available to an unprivileged writer and `bsize` their size, so // `bavail * bsize` is the free byte count. Injectable so tests drive the // adaptive limit without a real volume. export interface StatFsResult { bsize: number; bavail: number; } export type StatFsFn = (path: string) => Promise; const realStatFs: StatFsFn = async (path) => { const s = await statfs(path); return { bsize: s.bsize, bavail: s.bavail }; }; // The current usage and effective limit of the originals cache, in bytes. // `limitBytes` is the adaptive ceiling last computed (see `originalsLimit`). export interface OriginalsStatus { usedBytes: number; limitBytes?: number; } export interface ContentCacheOptions { pools: RequestPools; source: ContentSource; cacheDirectory: string; // The backup destination (issue-level `downloadDirectory`). An original // already stored there by a backup counts as present, so the cache serves // it rather than fetching a second copy. downloadDirectory?: string; // Resolve any membership of a file; every membership shares the underlying // content key, so any one decrypts the same bytes. getFile: (fileID: number) => EnteFile | undefined; // Hard ceiling on `cacheDirectory/originals` (default 100 GiB) and the free // space to protect on the volume (default 50 GiB). The effective limit is // the lesser of the ceiling and what fits above the protected free space. cacheOriginalsMaxBytes?: number; freeBelowBytes?: number; // Whether an original is pinned (favorites + latest week; the precache unit // #48 supplies the set). Pinned originals are never evicted; when only // pinned originals remain the cache runs over-limit until the set shrinks. isPinned?: (fileID: number) => boolean; // Free-space probe on the volume holding `cacheDirectory`; defaults to the // real `fs.statfs`. statfs?: StatFsFn; } // Thrown inside a pooled task to drop a queued fetch that was aborted before it // started running. Never escapes `ensureThumbnails`. class AbortDrop extends Error { constructor() { super("aborted"); this.name = "AbortDrop"; } } const originalName = (file: EnteFile): string => { const ext = extname(file.metadata.title || "") || ".bin"; return `${file.id}${ext}`; }; // The fileID a cache filename encodes, or undefined when the name is not one // the cache writes (``). const fileIDFromName = (name: string): number | undefined => { const base = name.slice(0, name.length - extname(name).length); if (!/^\d+$/.test(base)) return undefined; const id = Number(base); return Number.isSafeInteger(id) ? id : undefined; }; // Size of a regular file, or undefined if it is absent (or not a regular file). const fileSize = (path: string): number | undefined => { try { const s = statSync(path); return s.isFile() ? s.size : undefined; } catch { return undefined; } }; export class ContentCache implements PhotoContent, ThumbnailsAPI { private readonly pools: RequestPools; private readonly source: ContentSource; private readonly downloadDirectory?: string; private readonly getFile: (fileID: number) => EnteFile | undefined; private readonly originalsDir: string; private readonly thumbnailsDir: string; // fileID -> absolute path of the cached bytes, seeded from the directory // listing at open() and extended as fetches store new files. private readonly originals = new Map(); private readonly thumbnails = new Map(); private readonly maxOriginalsBytes: number; private readonly freeBelowBytes: number; private readonly isPinned: (fileID: number) => boolean; private readonly statfs: StatFsFn; // The last measured usage and effective limit, refreshed at open() and after // every original write; exposed through `originalsStatus`. private originalsUsedBytes = 0; private originalsLimitBytes?: number; // Serializes limit enforcement so concurrent original writes never race on // the map or delete each other's just-freed room. private enforcing: Promise = Promise.resolve(); constructor(opts: ContentCacheOptions) { this.pools = opts.pools; this.source = opts.source; this.downloadDirectory = opts.downloadDirectory; this.getFile = opts.getFile; this.originalsDir = join(opts.cacheDirectory, "originals"); this.thumbnailsDir = join(opts.cacheDirectory, "thumbnails"); this.maxOriginalsBytes = opts.cacheOriginalsMaxBytes ?? DEFAULT_ORIGINALS_MAX_BYTES; this.freeBelowBytes = opts.freeBelowBytes ?? DEFAULT_FREE_BELOW_BYTES; this.isPinned = opts.isPinned ?? (() => false); this.statfs = opts.statfs ?? realStatFs; } // Prepare the cache directories, reap orphan temp files, and take the // record of what is already cached. Called once before the cache serves. async open(): Promise { await this.ensureDir(this.originalsDir); await this.ensureDir(this.thumbnailsDir); await this.scan(this.originalsDir, this.originals); await this.scan(this.thumbnailsDir, this.thumbnails); // Publish the current usage and limit without evicting; a restart // reuses whatever survived on disk. Eviction only ever fires on a write. await this.refreshOriginalsLimit(); } // The current originals usage and effective limit, both in bytes, as of the // last write or open. `status().originalsLimitBytes` surfaces this. originalsStatus(): OriginalsStatus { return { usedBytes: this.originalsUsedBytes, limitBytes: this.originalsLimitBytes, }; } // The cache paths known for a file, for the record projection to expose as // `originalPath`/`thumbnailPath`. pathsFor(fileID: number): CachedPaths { const out: CachedPaths = {}; const original = this.originals.get(fileID); if (original !== undefined) out.originalPath = original; const thumbnail = this.thumbnails.get(fileID); if (thumbnail !== undefined) out.thumbnailPath = thumbnail; return out; } async original( fileID: number, opts?: ContentOptions, ): Promise { return this.get(fileID, "original", "on-demand", opts?.onProgress); } async thumbnail( fileID: number, opts?: ContentOptions, ): Promise { return this.get(fileID, "thumbnail", "on-demand", opts?.onProgress); } async ensure(args: EnsureOptions): Promise { return this.ensureThumbnails(args); } async ensureThumbnails(args: EnsureOptions): Promise { const priority = poolPriorityOf(args.priority); // Dedup the request list so a repeated fileID is fetched once and // reported once, in first-requested order. const seen = new Set(); const unique: number[] = []; for (const id of args.fileIDs) { if (!seen.has(id)) { seen.add(id); unique.push(id); } } return Promise.all( unique.map((fileID) => this.ensureOne(fileID, priority, args.signal, args.onProgress), ), ); } private async ensureOne( fileID: number, priority: Priority, signal: AbortSignal | undefined, onProgress: ((event: EnsureEvent) => void) | undefined, ): Promise { try { const result = await this.acquire( fileID, "thumbnail", priority, signal, ); const status = result.cached ? "skipped" : "done"; onProgress?.({ fileID, status, path: result.path }); return { fileID, path: result.path }; } catch (err) { if (err instanceof AbortDrop) { onProgress?.({ fileID, status: "aborted" }); return { fileID, error: "aborted" }; } const error = err instanceof Error ? err.message : String(err); onProgress?.({ fileID, status: "failed", error }); return { fileID, error }; } } private async get( fileID: number, kind: Kind, priority: Priority, onProgress: ((event: ContentEvent) => void) | undefined, ): Promise { const onByte: ProgressCallback | undefined = onProgress ? (bytesDone) => onProgress({ status: "downloading", bytesDone }) : undefined; const result = await this.acquire(fileID, kind, priority, undefined, { onByte, }); onProgress?.( result.cached ? { status: "skipped", bytes: result.bytes } : { status: "done", bytes: result.bytes }, ); return { path: result.path, bytes: result.bytes }; } // The core: return the cached path if present, else fetch through the pool, // store, and return it. `cached` distinguishes a present hit (no network, // no download event) from a fresh fetch. private async acquire( fileID: number, kind: Kind, priority: Priority, signal: AbortSignal | undefined, opts?: { onByte?: ProgressCallback }, ): Promise<{ path: string; bytes: number; cached: boolean }> { const file = this.getFile(fileID); if (!file) throw new Error(`content cache: unknown file ${fileID}`); const known = kind === "original" ? this.originals : this.thumbnails; const cached = known.get(fileID); if (cached !== undefined) { const size = fileSize(cached); if (size !== undefined && size > 0) { // Returning an original's path is a use: bump its mtime so LRU // order reflects it and survives a restart with no ledger. if ( kind === "original" && dirname(cached) === this.originalsDir ) await this.touch(cached); return { path: cached, bytes: size, cached: true }; } // A recorded file that has since gone re-fetches below. known.delete(fileID); } // An original a backup already stored counts as present. if (kind === "original" && this.downloadDirectory !== undefined) { const backupPath = join( this.downloadDirectory, "originals", originalName(file), ); const size = fileSize(backupPath); if (size !== undefined && size > 0) { this.originals.set(fileID, backupPath); return { path: backupPath, bytes: size, cached: true }; } } const dir = kind === "original" ? this.originalsDir : this.thumbnailsDir; const dest = kind === "original" ? join(dir, originalName(file)) : join(dir, `${fileID}${THUMBNAIL_EXT}`); const pool = kind === "original" ? this.pools.content : this.pools.thumbnails; return pool.run( async () => { // Dropping queued work on abort: a task still waiting for a slot // when the signal fired sees it here and never touches the // network. A task already past this point is in flight and runs // to completion. if (signal?.aborted) throw new AbortDrop(); await this.download(file, dest, kind, opts?.onByte); await chmod(dest, FILE_MODE); const size = (await stat(dest)).size; if (size === 0) { throw new Error( `content cache: ${kind} ${fileID} stored empty`, ); } known.set(fileID, dest); // A fresh original may have crossed the limit; make room by // evicting least-recently-used originals (never this one, just // touched). Thumbnails are never bounded or evicted. if (kind === "original") await this.enforceOriginalsLimit(); return { path: dest, bytes: size, cached: false }; }, { priority, key: fileID }, ); } private async download( file: EnteFile, destination: string, kind: Kind, onProgress: ProgressCallback | undefined, ): Promise { const args = { file, destination, onProgress }; const result = kind === "original" ? await this.source.original(args) : await this.source.thumbnail(args); return result.bytesWritten; } // Best-effort bump of a file's mtime to now; a failed touch must never fail // the read it accompanies. private async touch(path: string): Promise { const now = new Date(); await utimes(path, now, now).catch(() => undefined); } // Every stored original that lives under `originalsDir` (a backup-directory // hit recorded in the map is excluded), with its size and mtime. Entries // whose file has vanished are dropped from the map. Backups and thumbnails // are never counted. private async measureOriginals(): Promise<{ entries: { fileID: number; path: string; size: number; mtimeMs: number; }[]; used: number; }> { const entries: { fileID: number; path: string; size: number; mtimeMs: number; }[] = []; let used = 0; for (const [fileID, path] of this.originals) { if (dirname(path) !== this.originalsDir) continue; try { const s = await stat(path); entries.push({ fileID, path, size: s.size, mtimeMs: s.mtimeMs, }); used += s.size; } catch { this.originals.delete(fileID); } } return { entries, used }; } // The effective ceiling on originals: the configured max, but no more than // what fits once the protected free space is set aside. `used + free` is the // volume space the cache could occupy; subtracting `freeBelowBytes` leaves // the reserve untouched. Clamped at zero. private async originalsLimit(used: number): Promise { const { bsize, bavail } = await this.statfs(this.originalsDir); const free = bsize * bavail; const adaptive = used + free - this.freeBelowBytes; return Math.max(0, Math.min(this.maxOriginalsBytes, adaptive)); } // Recompute and publish usage and limit without evicting (used at open()). private refreshOriginalsLimit(): Promise { return this.serializeEnforce(async () => { const { used } = await this.measureOriginals(); this.originalsUsedBytes = used; this.originalsLimitBytes = await this.originalsLimit(used); }); } // Evict least-recently-used originals until usage fits the limit. Pinned // originals are skipped; when only pinned originals remain the cache stays // over-limit until the pinned set shrinks. private enforceOriginalsLimit(): Promise { return this.serializeEnforce(async () => { const { entries, used } = await this.measureOriginals(); const limit = await this.originalsLimit(used); let remaining = used; if (remaining > limit) { const evictable = entries .filter((e) => !this.isPinned(e.fileID)) .sort((a, b) => a.mtimeMs - b.mtimeMs); for (const e of evictable) { if (remaining <= limit) break; await rm(e.path, { force: true }); this.originals.delete(e.fileID); remaining -= e.size; } } this.originalsUsedBytes = remaining; this.originalsLimitBytes = limit; }); } // Run limit work one at a time; failures are swallowed so a transient // statfs or unlink error never rejects the read or write that triggered it. private serializeEnforce(work: () => Promise): Promise { const next = this.enforcing.then(work).catch(() => undefined); this.enforcing = next; return next; } private async ensureDir(dir: string): Promise { // chmod after mkdir so the mode is tightened even when the directory // already existed with a looser one; mkdir alone would not. await mkdir(dir, { recursive: true, mode: DIR_MODE }); await chmod(dir, DIR_MODE); } private async scan(dir: string, into: Map): Promise { let entries: string[]; try { entries = await readdir(dir); } catch { return; } for (const name of entries) { if (name.startsWith(TEMP_PREFIX) && name.endsWith(TEMP_SUFFIX)) { await rm(join(dir, name), { force: true }).catch( () => undefined, ); continue; } const id = fileIDFromName(name); const path = join(dir, name); if (id !== undefined && existsSync(path)) into.set(id, path); } } }