On-disk content and thumbnail cache with per-photo fetch and prefetch (closes #46)
check / check (push) Successful in 33s
check / check (push) Successful in 33s
Adds the on-disk content and thumbnail cache keyed by fileID: originals/ and thumbnails/ under cacheDirectory, present-means-complete (streaming atomic rename), orphan temp reaping on open. Photo.original/thumbnail return a cached path with no network when present, else fetch through the shared request pool; thumbnails.ensure drives the thumbnail pool with priority, dedup and AbortSignal. One shared RequestPools set serves both the ML fetch and the content cache. Content-hash integrity is deferred (#68); authenticated streaming decrypt guarantees integrity now. Model: opus-4-8
This commit was merged in pull request #66.
This commit is contained in:
@@ -0,0 +1,416 @@
|
||||
// The on-disk content and thumbnail cache keyed by fileID (issue #46).
|
||||
//
|
||||
// Layout under `cacheDirectory`: `originals/<fileID>.<ext>` and
|
||||
// `thumbnails/<fileID>.<ext>`, 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 } from "node:fs/promises";
|
||||
import { 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";
|
||||
// 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<ContentResult>;
|
||||
thumbnail(fileID: number, opts?: ContentOptions): Promise<ContentResult>;
|
||||
}
|
||||
|
||||
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<EnsureResult[]>;
|
||||
}
|
||||
|
||||
// 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;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
// 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 (`<digits><ext>`).
|
||||
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<number, string>();
|
||||
private readonly thumbnails = new Map<number, string>();
|
||||
|
||||
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");
|
||||
}
|
||||
|
||||
// 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<void> {
|
||||
await this.ensureDir(this.originalsDir);
|
||||
await this.ensureDir(this.thumbnailsDir);
|
||||
await this.scan(this.originalsDir, this.originals);
|
||||
await this.scan(this.thumbnailsDir, this.thumbnails);
|
||||
}
|
||||
|
||||
// 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<ContentResult> {
|
||||
return this.get(fileID, "original", "on-demand", opts?.onProgress);
|
||||
}
|
||||
|
||||
async thumbnail(
|
||||
fileID: number,
|
||||
opts?: ContentOptions,
|
||||
): Promise<ContentResult> {
|
||||
return this.get(fileID, "thumbnail", "on-demand", opts?.onProgress);
|
||||
}
|
||||
|
||||
async ensure(args: EnsureOptions): Promise<EnsureResult[]> {
|
||||
return this.ensureThumbnails(args);
|
||||
}
|
||||
|
||||
async ensureThumbnails(args: EnsureOptions): Promise<EnsureResult[]> {
|
||||
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<number>();
|
||||
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<EnsureResult> {
|
||||
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<ContentResult> {
|
||||
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)
|
||||
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);
|
||||
return { path: dest, bytes: size, cached: false };
|
||||
},
|
||||
{ priority, key: fileID },
|
||||
);
|
||||
}
|
||||
|
||||
private async download(
|
||||
file: EnteFile,
|
||||
destination: string,
|
||||
kind: Kind,
|
||||
onProgress: ProgressCallback | undefined,
|
||||
): Promise<number> {
|
||||
const args = { file, destination, onProgress };
|
||||
const result =
|
||||
kind === "original"
|
||||
? await this.source.original(args)
|
||||
: await this.source.thumbnail(args);
|
||||
return result.bytesWritten;
|
||||
}
|
||||
|
||||
private async ensureDir(dir: string): Promise<void> {
|
||||
// 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<number, string>): Promise<void> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user