From 724315dd47c1431edd0daf556dadff8fa538ad9d Mon Sep 17 00:00:00 2001 From: Luke Melia Date: Thu, 20 Aug 2026 17:03:17 -0400 Subject: [PATCH 1/2] Add MediaCache: content-addressed media store with ledger and GC sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The storage foundation for derived media (screenshots): objects keyed by the sha256 of their output bytes behind a MediaCacheAdapter interface (S3 for deployed environments, local disk for dev/tests), a media_cache_ledger table as the store's only catalog, and a reconcile-style media-cache-gc queue job (cron-enqueued, coalesced) that reclaims superseded generations, captures of tombstoned instances, and idle on-demand captures — deleting objects before their ledger rows so a crashed sweep leaves retryable rows, never untracked bytes. Serving and capture are deliberately not wired up here; nothing writes to the store in production yet. Co-Authored-By: Claude Fable 5 --- ...67_schema.sql => 1787258803063_schema.sql} | 15 + .../1787258803063_add-media-cache-ledger.js | 52 +++ packages/postgres/pg-queue.ts | 1 + packages/realm-server/lib/cron-scheduler.ts | 31 ++ .../realm-server/lib/media-cache-gc-config.ts | 25 + packages/realm-server/media-cache/index.ts | 37 ++ .../media-cache/local-disk-adapter.ts | 73 +++ .../realm-server/media-cache/s3-adapter.ts | 134 ++++++ .../realm-server/scripts/media-cache-gc.ts | 41 ++ packages/realm-server/tests/index.ts | 2 + .../tests/media-cache-adapter-test.ts | 224 +++++++++ .../realm-server/tests/media-cache-gc-test.ts | 432 ++++++++++++++++++ .../tests/worker-job-registration-test.ts | 1 + packages/realm-server/worker.ts | 2 + packages/runtime-common/index.ts | 1 + packages/runtime-common/media-cache.ts | 354 ++++++++++++++ packages/runtime-common/tasks/index.ts | 5 + .../runtime-common/tasks/media-cache-gc.ts | 134 ++++++ packages/runtime-common/worker.ts | 10 + 19 files changed, 1574 insertions(+) rename packages/host/config/schema/{1785953445767_schema.sql => 1787258803063_schema.sql} (92%) create mode 100644 packages/postgres/migrations/1787258803063_add-media-cache-ledger.js create mode 100644 packages/realm-server/lib/media-cache-gc-config.ts create mode 100644 packages/realm-server/media-cache/index.ts create mode 100644 packages/realm-server/media-cache/local-disk-adapter.ts create mode 100644 packages/realm-server/media-cache/s3-adapter.ts create mode 100644 packages/realm-server/scripts/media-cache-gc.ts create mode 100644 packages/realm-server/tests/media-cache-adapter-test.ts create mode 100644 packages/realm-server/tests/media-cache-gc-test.ts create mode 100644 packages/runtime-common/media-cache.ts create mode 100644 packages/runtime-common/tasks/media-cache-gc.ts diff --git a/packages/host/config/schema/1785953445767_schema.sql b/packages/host/config/schema/1787258803063_schema.sql similarity index 92% rename from packages/host/config/schema/1785953445767_schema.sql rename to packages/host/config/schema/1787258803063_schema.sql index 5c87082c0c7..9d88dcae222 100644 --- a/packages/host/config/schema/1785953445767_schema.sql +++ b/packages/host/config/schema/1787258803063_schema.sql @@ -76,6 +76,21 @@ PRIMARY KEY ( id ) ); + CREATE TABLE IF NOT EXISTS media_cache_ledger ( + realm_url TEXT NOT NULL, + source_url TEXT NOT NULL, + capture_spec_hash TEXT NOT NULL, + source_generation INTEGER NOT NULL, + object_key TEXT NOT NULL, + source_content_hash TEXT, + lane TEXT NOT NULL, + content_type TEXT NOT NULL, + size_bytes NOT NULL, + created_at NOT NULL, + last_accessed_at NOT NULL, + PRIMARY KEY ( realm_url, source_url, capture_spec_hash, source_generation ) +); + CREATE TABLE IF NOT EXISTS module_transpile_cache ( realm_url TEXT NOT NULL, canonical_path TEXT NOT NULL, diff --git a/packages/postgres/migrations/1787258803063_add-media-cache-ledger.js b/packages/postgres/migrations/1787258803063_add-media-cache-ledger.js new file mode 100644 index 00000000000..b4528478dd2 --- /dev/null +++ b/packages/postgres/migrations/1787258803063_add-media-cache-ledger.js @@ -0,0 +1,52 @@ +exports.shorthands = undefined; + +// The MediaCache ledger: one row per derived-media capture (a screenshot of +// one source instance under one canonical capture spec at one generation), +// pointing at a content-addressed object (`object_key` = hash of the output +// bytes) in the configured media store. The ledger is the store's only +// catalog — GC reclaims objects by scanning these rows, never by enumerating +// the bucket — so every object write must be paired with a ledger row. +// +// Several rows may share one `object_key`: identical output bytes are stored +// once (dedupe-on-write) and the object is reclaimable only when its last +// referencing row is gone. +// +// `lane` separates GC policy: 'declared' rows (indexing-time declared +// screenshots) are superseded by newer generations of the same capture, +// while 'on-demand' rows (URL-DSL / POST captures) additionally age out by +// last access. `created_at` / `last_accessed_at` are unix-ms bigints like +// `prerendered_html.rendered_at` (pg returns them as JS strings). + +exports.up = (pgm) => { + pgm.createTable('media_cache_ledger', { + realm_url: { type: 'varchar', notNull: true }, + source_url: { type: 'varchar', notNull: true }, + capture_spec_hash: { type: 'varchar', notNull: true }, + source_generation: { type: 'integer', notNull: true }, + object_key: { type: 'varchar', notNull: true }, + // The source file's content hash for captures keyed by file content + // rather than by generation (FileDef posters); null otherwise. + source_content_hash: { type: 'varchar' }, + lane: { type: 'varchar', notNull: true }, + content_type: { type: 'varchar', notNull: true }, + size_bytes: { type: 'bigint', notNull: true }, + created_at: { type: 'bigint', notNull: true }, + last_accessed_at: { type: 'bigint', notNull: true }, + }); + pgm.addConstraint('media_cache_ledger', 'media_cache_ledger_pkey', { + primaryKey: [ + 'realm_url', + 'source_url', + 'capture_spec_hash', + 'source_generation', + ], + }); + // Reference counting at GC time: is this object's key still referenced? + pgm.createIndex('media_cache_ledger', ['object_key']); + // The on-demand age-out lane scans by lane + last access. + pgm.createIndex('media_cache_ledger', ['lane', 'last_accessed_at']); +}; + +exports.down = (pgm) => { + pgm.dropTable('media_cache_ledger'); +}; diff --git a/packages/postgres/pg-queue.ts b/packages/postgres/pg-queue.ts index ccfe8d930f2..dce2880c5b9 100644 --- a/packages/postgres/pg-queue.ts +++ b/packages/postgres/pg-queue.ts @@ -34,6 +34,7 @@ import { FROM_SCRATCH_JOB_TIMEOUT_SEC } from '@cardstack/runtime-common/tasks/in // coalesce handlers registered before publish() is called. import '@cardstack/runtime-common/tasks/copy'; import '@cardstack/runtime-common/tasks/full-reindex'; +import '@cardstack/runtime-common/tasks/media-cache-gc'; import '@cardstack/runtime-common/tasks/prerender-html'; import '@cardstack/runtime-common/tasks/prerender-html-reconcile'; import type { PgAdapter } from './pg-adapter.ts'; diff --git a/packages/realm-server/lib/cron-scheduler.ts b/packages/realm-server/lib/cron-scheduler.ts index ad9a0d956f9..207d486e58c 100644 --- a/packages/realm-server/lib/cron-scheduler.ts +++ b/packages/realm-server/lib/cron-scheduler.ts @@ -19,6 +19,12 @@ import { PRERENDER_HTML_RECONCILE_CRON_TZ, createPrerenderHtmlReconcileCronJob, } from './prerender-html-reconcile-config.ts'; +import { enqueueMediaCacheGc } from '../scripts/media-cache-gc.ts'; +import { + MEDIA_CACHE_GC_CRON_SCHEDULE, + MEDIA_CACHE_GC_CRON_TZ, + createMediaCacheGcCronJob, +} from './media-cache-gc-config.ts'; let log = logger('cron-scheduler'); @@ -39,6 +45,11 @@ export function startCronJobs(): void { if (prerenderHtmlReconcileJob) { jobs.push(prerenderHtmlReconcileJob); } + + let mediaCacheGcJob = startMediaCacheGcCron(); + if (mediaCacheGcJob) { + jobs.push(mediaCacheGcJob); + } } export function stopCronJobs(): void { @@ -121,3 +132,23 @@ function startPrerenderHtmlReconcileCron(): CronJob | undefined { ); return job; } + +function startMediaCacheGcCron(): CronJob | undefined { + let job = createMediaCacheGcCronJob( + async () => { + try { + await enqueueMediaCacheGc(); + } catch (error) { + Sentry.captureException(error); + log.error('media-cache-gc cron failed to enqueue job', error); + } + }, + { runOnInit: false }, + ); + + job.start(); + log.info( + `media-cache-gc cron scheduled for ${MEDIA_CACHE_GC_CRON_SCHEDULE} ${MEDIA_CACHE_GC_CRON_TZ}`, + ); + return job; +} diff --git a/packages/realm-server/lib/media-cache-gc-config.ts b/packages/realm-server/lib/media-cache-gc-config.ts new file mode 100644 index 00000000000..755dc8d4a69 --- /dev/null +++ b/packages/realm-server/lib/media-cache-gc-config.ts @@ -0,0 +1,25 @@ +import { CronJob } from 'cron'; + +// Daily is plenty: the sweep's min-age delay is measured in hours and the +// on-demand TTL in days, so nothing it reclaims is urgent. Off the top of +// the hour to stay clear of the hourly prerender-html reconcile. Cadence is +// a tuning knob via the env override. +export const MEDIA_CACHE_GC_CRON_SCHEDULE = + process.env.MEDIA_CACHE_GC_CRON_SCHEDULE ?? '30 2 * * *'; +export const MEDIA_CACHE_GC_CRON_TZ = + process.env.MEDIA_CACHE_GC_CRON_TZ ?? 'America/New_York'; + +export function createMediaCacheGcCronJob( + onTick: () => void, + options: { runOnInit?: boolean } = {}, +) { + return new CronJob( + MEDIA_CACHE_GC_CRON_SCHEDULE, + onTick, + null, + false, + MEDIA_CACHE_GC_CRON_TZ, + null, + options.runOnInit ?? false, + ); +} diff --git a/packages/realm-server/media-cache/index.ts b/packages/realm-server/media-cache/index.ts new file mode 100644 index 00000000000..8b8120ae4de --- /dev/null +++ b/packages/realm-server/media-cache/index.ts @@ -0,0 +1,37 @@ +import type { MediaCacheAdapter } from '@cardstack/runtime-common'; +import { logger } from '@cardstack/runtime-common'; +import { S3MediaCacheAdapter } from './s3-adapter.ts'; +import { LocalDiskMediaCacheAdapter } from './local-disk-adapter.ts'; + +export { S3MediaCacheAdapter } from './s3-adapter.ts'; +export { LocalDiskMediaCacheAdapter } from './local-disk-adapter.ts'; + +const log = logger('media-cache'); + +// The process-level choice of MediaCache object store: +// MEDIA_CACHE_BUCKET → S3 (deployed environments; optional +// MEDIA_CACHE_KEY_PREFIX / MEDIA_CACHE_REGION) +// MEDIA_CACHE_DIR → local disk (dev / tests) +// neither → no store; media-cache tasks no-op. +// Bucket wins if both are set, so a deployed env var can't be shadowed by a +// stray local one. +export function createMediaCacheAdapterFromEnv(): + | MediaCacheAdapter + | undefined { + let bucket = process.env.MEDIA_CACHE_BUCKET?.trim(); + if (bucket) { + return new S3MediaCacheAdapter({ + bucket, + keyPrefix: process.env.MEDIA_CACHE_KEY_PREFIX?.trim() || '', + region: process.env.MEDIA_CACHE_REGION?.trim() || undefined, + }); + } + let dir = process.env.MEDIA_CACHE_DIR?.trim(); + if (dir) { + return new LocalDiskMediaCacheAdapter({ dir }); + } + log.info( + 'neither MEDIA_CACHE_BUCKET nor MEDIA_CACHE_DIR is set; media cache is disabled for this process', + ); + return undefined; +} diff --git a/packages/realm-server/media-cache/local-disk-adapter.ts b/packages/realm-server/media-cache/local-disk-adapter.ts new file mode 100644 index 00000000000..2a06c70ab93 --- /dev/null +++ b/packages/realm-server/media-cache/local-disk-adapter.ts @@ -0,0 +1,73 @@ +// Local-disk MediaCache object store for dev and tests. Objects live at +// `//` — the two-character fan-out keeps any one +// directory from accumulating every object. No metadata is stored beside the +// bytes: the ledger's `content_type` column is the serving-path source of +// truth, so `head` reports size only. + +import { createReadStream } from 'node:fs'; +import { mkdir, rename, rm, stat, writeFile } from 'node:fs/promises'; +import { dirname, join } from 'node:path'; +import type { + MediaCacheAdapter, + MediaObjectStat, +} from '@cardstack/runtime-common'; + +export class LocalDiskMediaCacheAdapter implements MediaCacheAdapter { + #dir: string; + + constructor({ dir }: { dir: string }) { + this.#dir = dir; + } + + private pathFor(key: string): string { + return join(this.#dir, key.slice(0, 2), key); + } + + async put( + key: string, + bytes: Uint8Array, + _opts: { contentType: string }, + ): Promise { + let path = this.pathFor(key); + // The key is a hash of the bytes, so an existing file under this key + // already holds them — skip the write (dedupe-on-write). + if (await this.head(key)) { + return; + } + await mkdir(dirname(path), { recursive: true }); + // Write-then-rename so a reader never sees a half-written object and a + // crashed write leaves only a stray temp file, not a corrupt object. The + // temp name carries the pid so two processes writing the same key (both + // saw it absent) don't collide; rename is atomic, and last-writer-wins is + // harmless because both hold the same bytes. + let tempPath = `${path}.${process.pid}.tmp`; + await writeFile(tempPath, bytes); + await rename(tempPath, path); + } + + async head(key: string): Promise { + try { + let stats = await stat(this.pathFor(key)); + return { size: stats.size }; + } catch (error: any) { + if (error?.code === 'ENOENT') { + return undefined; + } + throw error; + } + } + + async getStream(key: string): Promise | undefined> { + // Existence-check first: createReadStream reports a missing file only as + // an async 'error' event, which would escape as an unhandled stream + // error rather than this interface's `undefined`. + if (!(await this.head(key))) { + return undefined; + } + return createReadStream(this.pathFor(key)); + } + + async delete(key: string): Promise { + await rm(this.pathFor(key), { force: true }); + } +} diff --git a/packages/realm-server/media-cache/s3-adapter.ts b/packages/realm-server/media-cache/s3-adapter.ts new file mode 100644 index 00000000000..1154ecd9f5f --- /dev/null +++ b/packages/realm-server/media-cache/s3-adapter.ts @@ -0,0 +1,134 @@ +// S3-backed MediaCache object store. The bucket holds only content-addressed +// derived media (screenshots); the `media_cache_ledger` table is the sole +// catalog of what's in here, so nothing in this adapter ever lists the +// bucket. Region/credential resolution mirrors `prerender/artifact-sink.ts`: +// in ECS the write grant rides on the task role, which the SDK resolves from +// the container credentials endpoint — no keys are configured here. + +import { + DeleteObjectCommand, + GetObjectCommand, + HeadObjectCommand, + PutObjectCommand, + S3Client, +} from '@aws-sdk/client-s3'; +import type { + MediaCacheAdapter, + MediaObjectStat, +} from '@cardstack/runtime-common'; +import type { Readable } from 'stream'; + +const DEFAULT_REGION = 'us-east-1'; + +export class S3MediaCacheAdapter implements MediaCacheAdapter { + #client: S3Client; + #bucket: string; + #keyPrefix: string; + + constructor({ + bucket, + region, + keyPrefix = '', + client, + }: { + bucket: string; + region?: string; + // Namespaces this store's objects within a shared bucket. Applied to + // every operation, so it never leaks above the adapter. + keyPrefix?: string; + // Injectable for tests; production callers omit it. + client?: S3Client; + }) { + this.#bucket = bucket; + this.#keyPrefix = keyPrefix; + this.#client = + client ?? + new S3Client({ + region: region ?? process.env.AWS_REGION?.trim() ?? DEFAULT_REGION, + }); + } + + private objectKey(key: string): string { + return `${this.#keyPrefix}${key}`; + } + + async put( + key: string, + bytes: Uint8Array, + opts: { contentType: string }, + ): Promise { + // The key is a hash of the bytes, so an existing object under this key + // already holds them — skip the upload (dedupe-on-write). + if (await this.head(key)) { + return; + } + await this.#client.send( + new PutObjectCommand({ + Bucket: this.#bucket, + Key: this.objectKey(key), + Body: bytes, + ContentType: opts.contentType, + }), + ); + } + + async head(key: string): Promise { + try { + let response = await this.#client.send( + new HeadObjectCommand({ + Bucket: this.#bucket, + Key: this.objectKey(key), + }), + ); + return { + size: response.ContentLength ?? 0, + contentType: response.ContentType, + }; + } catch (error: any) { + if (isMissingObjectError(error)) { + return undefined; + } + throw error; + } + } + + async getStream(key: string): Promise | undefined> { + try { + let response = await this.#client.send( + new GetObjectCommand({ + Bucket: this.#bucket, + Key: this.objectKey(key), + }), + ); + // In node the SDK's Body is a Readable, which is an + // AsyncIterable — exactly the interface's stream shape. + return (response.Body as Readable | undefined) ?? undefined; + } catch (error: any) { + if (isMissingObjectError(error)) { + return undefined; + } + throw error; + } + } + + async delete(key: string): Promise { + // S3 DeleteObject on a missing key succeeds, giving the interface's + // idempotent-delete contract for free. + await this.#client.send( + new DeleteObjectCommand({ + Bucket: this.#bucket, + Key: this.objectKey(key), + }), + ); + } +} + +// HeadObject reports absence as `NotFound`, GetObject as `NoSuchKey`; some +// SDK paths surface only the bare 404 status. +function isMissingObjectError(error: any): boolean { + return ( + error?.name === 'NotFound' || + error?.name === 'NoSuchKey' || + error?.$metadata?.httpStatusCode === 404 + ); +} diff --git a/packages/realm-server/scripts/media-cache-gc.ts b/packages/realm-server/scripts/media-cache-gc.ts new file mode 100644 index 00000000000..21547586fdf --- /dev/null +++ b/packages/realm-server/scripts/media-cache-gc.ts @@ -0,0 +1,41 @@ +import '../instrument.ts'; +import '../setup-logger.ts'; // This should be first +import { logger, systemInitiatedPriority } from '@cardstack/runtime-common'; +import { PgAdapter, PgQueuePublisher } from '@cardstack/postgres'; +import * as Sentry from '@sentry/node'; + +const log = logger('media-cache-gc'); +const MEDIA_CACHE_GC_JOB_TIMEOUT_SEC = 10 * 60; + +// Enqueue the GC sweep rather than sweeping inline in the worker-manager +// process: a worker scans the ledger and deletes reclaimed rows/objects. The +// sweep runs at the background tier (priority 0) so it never competes with +// indexing or user work. +export async function enqueueMediaCacheGc({ + priority = systemInitiatedPriority, + migrateDB, +}: { + priority?: number; + migrateDB?: boolean; +} = {}) { + let dbAdapter = new PgAdapter({ autoMigrate: migrateDB || undefined }); + let queue = new PgQueuePublisher(dbAdapter); + + try { + await queue.publish({ + jobType: 'media-cache-gc', + concurrencyGroup: 'media-cache-gc', + timeout: MEDIA_CACHE_GC_JOB_TIMEOUT_SEC, + priority, + args: {}, + }); + log.info('enqueued media-cache-gc job'); + } catch (error) { + Sentry.captureException(error); + log.error('failed to enqueue media-cache-gc job', error); + throw error; + } finally { + await queue.destroy(); + await dbAdapter.close(); + } +} diff --git a/packages/realm-server/tests/index.ts b/packages/realm-server/tests/index.ts index b2ad2fcf514..af71c108033 100644 --- a/packages/realm-server/tests/index.ts +++ b/packages/realm-server/tests/index.ts @@ -261,6 +261,8 @@ const ALL_TEST_FILES: string[] = [ './prerender-html-split-test', './prerender-html-split-integration-test', './prerender-html-reconcile-test', + './media-cache-adapter-test', + './media-cache-gc-test', './prerender-server-test', './prerender-manager-test', './prerender-host-shell-recycle-test', diff --git a/packages/realm-server/tests/media-cache-adapter-test.ts b/packages/realm-server/tests/media-cache-adapter-test.ts new file mode 100644 index 00000000000..2750b01d507 --- /dev/null +++ b/packages/realm-server/tests/media-cache-adapter-test.ts @@ -0,0 +1,224 @@ +import QUnit from 'qunit'; +const { module, test } = QUnit; +import { basename } from 'path'; +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { computeMediaCacheKey } from '@cardstack/runtime-common'; +import { LocalDiskMediaCacheAdapter } from '../media-cache/local-disk-adapter.ts'; +import { S3MediaCacheAdapter } from '../media-cache/s3-adapter.ts'; + +async function collectBytes( + stream: AsyncIterable, +): Promise { + let chunks: Uint8Array[] = []; + for await (let chunk of stream) { + chunks.push(chunk); + } + return Buffer.concat(chunks); +} + +module(basename(import.meta.filename), function () { + module('computeMediaCacheKey', function () { + test('is the sha256 hex of the bytes', async function (assert) { + // Pinned against `echo -n "boxel" | shasum -a 256`, so the content + // address never silently changes shape — object keys, ledger rows, and + // (later) ETags all depend on it. + assert.strictEqual( + await computeMediaCacheKey(new TextEncoder().encode('boxel')), + '1c1ab2b8fae6c953de3694b4bb8b1dc9295d4ce0bb71e24d284ee4012611579c', + ); + }); + + test('distinguishes different bytes', async function (assert) { + assert.notStrictEqual( + await computeMediaCacheKey(new Uint8Array([1, 2, 3])), + await computeMediaCacheKey(new Uint8Array([1, 2, 4])), + ); + }); + }); + + module('LocalDiskMediaCacheAdapter', function (hooks) { + let dir: string; + let adapter: LocalDiskMediaCacheAdapter; + let bytes = new TextEncoder().encode('png-bytes'); + let key: string; + + hooks.beforeEach(async function () { + dir = await mkdtemp(join(tmpdir(), 'media-cache-test-')); + adapter = new LocalDiskMediaCacheAdapter({ dir }); + key = await computeMediaCacheKey(bytes); + }); + + hooks.afterEach(async function () { + await rm(dir, { recursive: true, force: true }); + }); + + test('round-trips an object', async function (assert) { + await adapter.put(key, bytes, { contentType: 'image/png' }); + + let stat = await adapter.head(key); + assert.strictEqual(stat?.size, bytes.length, 'head reports the size'); + + let stream = await adapter.getStream(key); + assert.ok(stream, 'the object streams'); + assert.deepEqual( + [...(await collectBytes(stream!))], + [...bytes], + 'the streamed bytes are the stored bytes', + ); + + await adapter.delete(key); + assert.strictEqual(await adapter.head(key), undefined, 'deleted'); + }); + + test('reports absence rather than erroring', async function (assert) { + assert.strictEqual(await adapter.head('0'.repeat(64)), undefined); + assert.strictEqual(await adapter.getStream('0'.repeat(64)), undefined); + await adapter.delete('0'.repeat(64)); // idempotent no-op + assert.ok(true, 'deleting a missing key does not throw'); + }); + + test('put is dedupe-on-write: an existing object is not rewritten', async function (assert) { + await adapter.put(key, bytes, { contentType: 'image/png' }); + // Scribble on the stored file, then re-put the same key. A correct + // adapter treats key-exists as bytes-present and skips the write, so + // the scribble survives — proof the second put was a no-op. + let path = join(dir, key.slice(0, 2), key); + await writeFile(path, 'scribble'); + await adapter.put(key, bytes, { contentType: 'image/png' }); + assert.strictEqual( + await readFile(path, 'utf8'), + 'scribble', + 'the second put did not rewrite the object', + ); + }); + + test('objects fan out under a two-character prefix directory', async function (assert) { + await adapter.put(key, bytes, { contentType: 'image/png' }); + let path = join(dir, key.slice(0, 2), key); + assert.deepEqual( + [...(await readFile(path))], + [...bytes], + 'the object lives at //', + ); + }); + }); + + module('S3MediaCacheAdapter', function (hooks) { + // The stub stands in for S3Client: it records every command and answers + // from a scripted head/get response, so these tests pin the adapter's + // command construction and error mapping without any network. + let sent: { name: string; input: any }[]; + let headResult: (() => any) | undefined; + let getResult: (() => any) | undefined; + let adapter: S3MediaCacheAdapter; + + function notFound(name: string) { + return Object.assign(new Error(name), { + name, + $metadata: { httpStatusCode: 404 }, + }); + } + + hooks.beforeEach(function () { + sent = []; + headResult = undefined; + getResult = undefined; + let client = { + send: async (command: any) => { + sent.push({ name: command.constructor.name, input: command.input }); + switch (command.constructor.name) { + case 'HeadObjectCommand': + if (!headResult) { + throw notFound('NotFound'); + } + return headResult(); + case 'GetObjectCommand': + if (!getResult) { + throw notFound('NoSuchKey'); + } + return getResult(); + default: + return {}; + } + }, + }; + adapter = new S3MediaCacheAdapter({ + bucket: 'test-bucket', + keyPrefix: 'media/', + client: client as any, + }); + }); + + test('put uploads a missing object with its content type', async function (assert) { + await adapter.put('abc123', new Uint8Array([1, 2]), { + contentType: 'image/png', + }); + assert.deepEqual( + sent.map((s) => s.name), + ['HeadObjectCommand', 'PutObjectCommand'], + 'head-checks then uploads', + ); + let put = sent[1].input; + assert.strictEqual(put.Bucket, 'test-bucket'); + assert.strictEqual(put.Key, 'media/abc123', 'the key prefix is applied'); + assert.strictEqual(put.ContentType, 'image/png'); + assert.deepEqual([...put.Body], [1, 2]); + }); + + test('put is dedupe-on-write: an existing object is not re-uploaded', async function (assert) { + headResult = () => ({ ContentLength: 2 }); + await adapter.put('abc123', new Uint8Array([1, 2]), { + contentType: 'image/png', + }); + assert.deepEqual( + sent.map((s) => s.name), + ['HeadObjectCommand'], + 'no upload was issued', + ); + }); + + test('head reports size and content type, and absence as undefined', async function (assert) { + headResult = () => ({ ContentLength: 42, ContentType: 'image/webp' }); + assert.deepEqual(await adapter.head('abc123'), { + size: 42, + contentType: 'image/webp', + }); + assert.strictEqual(sent[0].input.Key, 'media/abc123'); + + headResult = undefined; + assert.strictEqual(await adapter.head('missing'), undefined); + }); + + test('getStream returns the body and maps a missing key to undefined', async function (assert) { + let body = (async function* () { + yield new Uint8Array([9]); + })(); + getResult = () => ({ Body: body }); + assert.strictEqual(await adapter.getStream('abc123'), body); + + getResult = undefined; + assert.strictEqual(await adapter.getStream('missing'), undefined); + }); + + test('delete issues a DeleteObjectCommand under the prefixed key', async function (assert) { + await adapter.delete('abc123'); + assert.deepEqual( + sent.map((s) => s.name), + ['DeleteObjectCommand'], + ); + assert.strictEqual(sent[0].input.Key, 'media/abc123'); + }); + + test('non-404 errors propagate', async function (assert) { + headResult = () => { + throw Object.assign(new Error('AccessDenied'), { + name: 'AccessDenied', + $metadata: { httpStatusCode: 403 }, + }); + }; + await assert.rejects(adapter.head('abc123'), /AccessDenied/); + }); + }); +}); diff --git a/packages/realm-server/tests/media-cache-gc-test.ts b/packages/realm-server/tests/media-cache-gc-test.ts new file mode 100644 index 00000000000..bfce1ee8a56 --- /dev/null +++ b/packages/realm-server/tests/media-cache-gc-test.ts @@ -0,0 +1,432 @@ +import QUnit from 'qunit'; +const { module, test } = QUnit; +import { basename } from 'path'; +import type { PgAdapter } from '@cardstack/postgres'; +import type { + DefinitionLookup, + IndexWriter, + MediaCacheAdapter, + MediaCacheLane, + Prerenderer, + QueuePublisher, + VirtualNetwork, +} from '@cardstack/runtime-common'; +import { + asExpressions, + computeMediaCacheKey, + insert, + logger, + mediaCacheGc, + putMedia, + query, + touchMediaCacheEntry, +} from '@cardstack/runtime-common'; + +import { setupDB } from './helpers/index.ts'; + +const HOUR = 60 * 60 * 1000; +const DAY = 24 * HOUR; + +// In-memory MediaCacheAdapter: enough store to observe what the sweep +// deletes, plus scriptable per-key delete failures. +class FakeMediaCacheAdapter implements MediaCacheAdapter { + objects = new Map(); + deleted: string[] = []; + failDeletesFor = new Set(); + + async put(key: string, bytes: Uint8Array, _opts: { contentType: string }) { + if (!this.objects.has(key)) { + this.objects.set(key, bytes); + } + } + async head(key: string) { + let bytes = this.objects.get(key); + return bytes ? { size: bytes.length } : undefined; + } + async getStream(key: string) { + let bytes = this.objects.get(key); + if (!bytes) { + return undefined; + } + return (async function* () { + yield bytes; + })(); + } + async delete(key: string) { + if (this.failDeletesFor.has(key)) { + throw new Error(`simulated delete failure for ${key}`); + } + this.deleted.push(key); + this.objects.delete(key); + } +} + +module(basename(import.meta.filename), function (hooks) { + let dbAdapter: PgAdapter; + let adapter: FakeMediaCacheAdapter; + + setupDB(hooks, { + beforeEach: async ( + _dbAdapter: PgAdapter, + _publisher: QueuePublisher, + ): Promise => { + dbAdapter = _dbAdapter; + adapter = new FakeMediaCacheAdapter(); + }, + }); + + function runGc( + opts: { mediaCacheAdapter: MediaCacheAdapter | undefined } = { + mediaCacheAdapter: adapter, + }, + ) { + return mediaCacheGc({ + reportStatus: () => {}, + log: logger('media-cache-gc-test'), + dbAdapter, + mediaCacheAdapter: opts.mediaCacheAdapter, + queuePublisher: null as unknown as QueuePublisher, + indexWriter: null as unknown as IndexWriter, + prerenderer: null as unknown as Prerenderer, + definitionLookup: null as unknown as DefinitionLookup, + virtualNetwork: null as unknown as VirtualNetwork, + matrixURL: 'http://localhost:8008', + getReader: () => { + throw new Error('getReader is not used by media-cache-gc'); + }, + getAuthedFetch: async () => globalThis.fetch, + createPrerenderAuth: () => '', + })({}); + } + + async function seedLedgerRow({ + realmURL = 'http://test-realm/a/', + sourceURL = 'http://test-realm/a/card-1', + captureSpecHash = 'spec-1', + sourceGeneration, + objectKey, + lane = 'declared' as MediaCacheLane, + createdAt, + lastAccessedAt = createdAt, + }: { + realmURL?: string; + sourceURL?: string; + captureSpecHash?: string; + sourceGeneration: number; + objectKey: string; + lane?: MediaCacheLane; + createdAt: number; + lastAccessedAt?: number; + }) { + let { nameExpressions, valueExpressions } = asExpressions({ + realm_url: realmURL, + source_url: sourceURL, + capture_spec_hash: captureSpecHash, + source_generation: sourceGeneration, + object_key: objectKey, + lane, + content_type: 'image/png', + size_bytes: 3, + created_at: createdAt, + last_accessed_at: lastAccessedAt, + }); + await query( + dbAdapter, + insert('media_cache_ledger', nameExpressions, valueExpressions), + ); + adapter.objects.set(objectKey, new Uint8Array([1, 2, 3])); + } + + async function seedTombstone(sourceURL: string, realmURL: string) { + let { nameExpressions, valueExpressions } = asExpressions({ + url: sourceURL, + file_alias: sourceURL, + realm_url: realmURL, + type: 'instance', + generation: 1, + is_deleted: true, + }); + await query( + dbAdapter, + insert('boxel_index', nameExpressions, valueExpressions), + ); + } + + async function ledgerRows(): Promise< + { source_generation: number; object_key: string }[] + > { + return (await query(dbAdapter, [ + `SELECT source_generation, object_key FROM media_cache_ledger ORDER BY source_generation`, + ])) as { source_generation: number; object_key: string }[]; + } + + test('reclaims a superseded generation and its orphaned object', async function (assert) { + let now = Date.now(); + await seedLedgerRow({ + sourceGeneration: 1, + objectKey: 'old-object', + createdAt: now - 3 * DAY, + }); + await seedLedgerRow({ + sourceGeneration: 2, + objectKey: 'new-object', + createdAt: now - 2 * DAY, + }); + + let result = await runGc(); + + assert.strictEqual(result.rowsDeleted, 1); + assert.strictEqual(result.objectsDeleted, 1); + assert.deepEqual(adapter.deleted, ['old-object']); + assert.deepEqual( + (await ledgerRows()).map((row) => Number(row.source_generation)), + [2], + 'only the superseding row survives', + ); + }); + + test('a row is not superseded until its successor has aged past min-age', async function (assert) { + let now = Date.now(); + await seedLedgerRow({ + sourceGeneration: 1, + objectKey: 'old-object', + createdAt: now - 3 * DAY, + }); + // The gen-2 capture just landed: a serve that resolved gen 1 moments ago + // may still be streaming, so gen 1 lingers for the min-age window. + await seedLedgerRow({ + sourceGeneration: 2, + objectKey: 'new-object', + createdAt: now - 1 * HOUR, + }); + + let result = await runGc(); + + assert.strictEqual(result.rowsDeleted, 0, 'nothing reclaimed yet'); + assert.deepEqual(adapter.deleted, []); + }); + + test('a young row is never collected, whatever its lane', async function (assert) { + let now = Date.now(); + await seedLedgerRow({ + sourceGeneration: 1, + objectKey: 'young-object', + lane: 'on-demand', + createdAt: now - 1 * HOUR, + // Nonsense on purpose: even an ancient last-access cannot reclaim a + // row younger than min-age. + lastAccessedAt: now - 400 * DAY, + }); + + let result = await runGc(); + + assert.strictEqual(result.rowsDeleted, 0); + }); + + test('reclaims captures of a tombstoned source instance', async function (assert) { + let now = Date.now(); + await seedLedgerRow({ + sourceURL: 'http://test-realm/a/deleted-card', + sourceGeneration: 5, + objectKey: 'tombstoned-object', + createdAt: now - 2 * DAY, + }); + await seedTombstone( + 'http://test-realm/a/deleted-card', + 'http://test-realm/a/', + ); + + let result = await runGc(); + + assert.strictEqual(result.rowsDeleted, 1); + assert.deepEqual(adapter.deleted, ['tombstoned-object']); + }); + + test('ages out idle on-demand captures but never declared ones', async function (assert) { + let now = Date.now(); + await seedLedgerRow({ + captureSpecHash: 'spec-on-demand', + sourceGeneration: 1, + objectKey: 'idle-on-demand', + lane: 'on-demand', + createdAt: now - 60 * DAY, + lastAccessedAt: now - 45 * DAY, + }); + await seedLedgerRow({ + captureSpecHash: 'spec-on-demand-active', + sourceGeneration: 1, + objectKey: 'active-on-demand', + lane: 'on-demand', + createdAt: now - 60 * DAY, + lastAccessedAt: now - 1 * DAY, + }); + await seedLedgerRow({ + captureSpecHash: 'spec-declared', + sourceGeneration: 1, + objectKey: 'idle-declared', + lane: 'declared', + createdAt: now - 60 * DAY, + lastAccessedAt: now - 45 * DAY, + }); + + let result = await runGc(); + + assert.strictEqual(result.rowsDeleted, 1); + assert.deepEqual( + adapter.deleted, + ['idle-on-demand'], + 'only the idle on-demand capture is reclaimed', + ); + }); + + test('an object still referenced by a surviving row keeps its bytes', async function (assert) { + let now = Date.now(); + // Two captures produced identical bytes (dedupe): the superseded row is + // pruned, but the object stays because another capture still points at it. + await seedLedgerRow({ + captureSpecHash: 'spec-a', + sourceGeneration: 1, + objectKey: 'shared-object', + createdAt: now - 3 * DAY, + }); + await seedLedgerRow({ + captureSpecHash: 'spec-a', + sourceGeneration: 2, + objectKey: 'spec-a-gen2', + createdAt: now - 2 * DAY, + }); + await seedLedgerRow({ + captureSpecHash: 'spec-b', + sourceGeneration: 1, + objectKey: 'shared-object', + createdAt: now - 3 * DAY, + }); + + let result = await runGc(); + + assert.strictEqual(result.rowsDeleted, 1, 'the superseded row is pruned'); + assert.strictEqual(result.objectsDeleted, 0, 'but its object survives'); + assert.ok(adapter.objects.has('shared-object')); + }); + + test('a failed object delete keeps the rows for the next sweep', async function (assert) { + let now = Date.now(); + await seedLedgerRow({ + sourceGeneration: 1, + objectKey: 'stubborn-object', + createdAt: now - 3 * DAY, + }); + await seedLedgerRow({ + sourceGeneration: 2, + objectKey: 'new-object', + createdAt: now - 2 * DAY, + }); + adapter.failDeletesFor.add('stubborn-object'); + + let result = await runGc(); + + assert.strictEqual(result.objectDeleteFailures, 1); + assert.strictEqual(result.rowsDeleted, 0); + assert.strictEqual( + (await ledgerRows()).length, + 2, + 'the failed object keeps its ledger row as the retry path', + ); + + // The failure clears (transient S3 trouble): the next sweep re-finds the + // same candidate and completes the reclaim. + adapter.failDeletesFor.clear(); + let retry = await runGc(); + assert.strictEqual(retry.rowsDeleted, 1); + assert.deepEqual(adapter.deleted, ['stubborn-object']); + }); + + test('no-ops without a configured adapter', async function (assert) { + let now = Date.now(); + await seedLedgerRow({ + sourceGeneration: 1, + objectKey: 'old-object', + createdAt: now - 3 * DAY, + }); + await seedLedgerRow({ + sourceGeneration: 2, + objectKey: 'new-object', + createdAt: now - 2 * DAY, + }); + + let result = await runGc({ mediaCacheAdapter: undefined }); + + assert.deepEqual(result, { + rowsDeleted: 0, + objectsDeleted: 0, + objectDeleteFailures: 0, + }); + assert.strictEqual((await ledgerRows()).length, 2, 'nothing was touched'); + }); + + module('putMedia and touchMediaCacheEntry', function () { + let entryKey = { + realmURL: 'http://test-realm/a/', + sourceURL: 'http://test-realm/a/card-1', + captureSpecHash: 'spec-1', + sourceGeneration: 1, + }; + + test('stores the object under its content address and records the ledger row', async function (assert) { + let bytes = new Uint8Array([1, 2, 3, 4]); + let { objectKey, sizeBytes } = await putMedia(dbAdapter, adapter, { + ...entryKey, + bytes, + contentType: 'image/png', + lane: 'on-demand', + }); + + assert.strictEqual(objectKey, await computeMediaCacheKey(bytes)); + assert.strictEqual(sizeBytes, 4); + assert.deepEqual([...adapter.objects.get(objectKey)!], [...bytes]); + let rows = await ledgerRows(); + assert.strictEqual(rows.length, 1); + assert.strictEqual(rows[0].object_key, objectKey); + }); + + test('a re-capture upserts its row, repointing at the new bytes', async function (assert) { + let first = await putMedia(dbAdapter, adapter, { + ...entryKey, + bytes: new Uint8Array([1]), + contentType: 'image/png', + lane: 'on-demand', + }); + let second = await putMedia(dbAdapter, adapter, { + ...entryKey, + bytes: new Uint8Array([2]), + contentType: 'image/png', + lane: 'on-demand', + }); + + assert.notStrictEqual(first.objectKey, second.objectKey); + let rows = await ledgerRows(); + assert.strictEqual(rows.length, 1, 'still one row for the capture'); + assert.strictEqual( + rows[0].object_key, + second.objectKey, + 'the row points at the latest bytes', + ); + }); + + test('touch bumps last_accessed_at', async function (assert) { + await putMedia(dbAdapter, adapter, { + ...entryKey, + bytes: new Uint8Array([1]), + contentType: 'image/png', + lane: 'on-demand', + }); + let later = Date.now() + 5000; + await touchMediaCacheEntry(dbAdapter, entryKey, later); + + let [row] = (await query(dbAdapter, [ + `SELECT last_accessed_at FROM media_cache_ledger`, + ])) as { last_accessed_at: number | string }[]; + assert.strictEqual(Number(row.last_accessed_at), later); + }); + }); +}); diff --git a/packages/realm-server/tests/worker-job-registration-test.ts b/packages/realm-server/tests/worker-job-registration-test.ts index 23f8ff989d3..445aafeee72 100644 --- a/packages/realm-server/tests/worker-job-registration-test.ts +++ b/packages/realm-server/tests/worker-job-registration-test.ts @@ -64,6 +64,7 @@ module(basename(import.meta.filename), function () { 'full-reindex', 'incremental-index', 'lint-source', + 'media-cache-gc', 'prerender-html-reconcile', 'prerender_html', 'run-command', diff --git a/packages/realm-server/worker.ts b/packages/realm-server/worker.ts index 87751c131a0..9a51e0e15f6 100644 --- a/packages/realm-server/worker.ts +++ b/packages/realm-server/worker.ts @@ -48,6 +48,7 @@ import { } from '@cardstack/postgres'; import { createRemotePrerenderer } from './prerender/remote-prerenderer.ts'; import { buildCreatePrerenderAuth } from './prerender/auth.ts'; +import { createMediaCacheAdapterFromEnv } from './media-cache/index.ts'; import { finalizeChildReservationAsFailure } from './lib/finalize-child-fatal-failure.ts'; let log = logger('worker'); @@ -215,6 +216,7 @@ let autoMigrate = migrateDB || undefined; prerenderer, createPrerenderAuth, indexJobsOnly, + mediaCacheAdapter: createMediaCacheAdapterFromEnv(), }); await worker.run(); diff --git a/packages/runtime-common/index.ts b/packages/runtime-common/index.ts index 30ee4b5e0bc..c1b8516ff1d 100644 --- a/packages/runtime-common/index.ts +++ b/packages/runtime-common/index.ts @@ -1062,6 +1062,7 @@ export * from './matrix-client.ts'; export * from './queue.ts'; export * from './job-utils.ts'; export * from './prerender-html-reconcile.ts'; +export * from './media-cache.ts'; export * from './expression.ts'; export * from './searchable-parity.ts'; export * from './infer-content-type.ts'; diff --git a/packages/runtime-common/media-cache.ts b/packages/runtime-common/media-cache.ts new file mode 100644 index 00000000000..da1b0ca89f3 --- /dev/null +++ b/packages/runtime-common/media-cache.ts @@ -0,0 +1,354 @@ +import { Sha256 } from '@aws-crypto/sha256-js'; +import type { DBAdapter } from './db.ts'; +import { + addExplicitParens, + asExpressions, + param, + query, + separatedByCommas, + upsert, + type Expression, +} from './expression.ts'; +import { uint8ArrayToHex } from './index.ts'; + +// The MediaCache: a content-addressed store for derived media (screenshots), +// split across two channels that this module keeps consistent: +// +// * objects — the output bytes, held in a `MediaCacheAdapter` (S3 in +// deployment, local disk in dev/tests) under a key that is the hash of +// the bytes themselves, so identical output is stored exactly once; +// * the ledger — one `media_cache_ledger` row per capture (source +// instance × canonical capture spec × generation) pointing at its +// object. The ledger is the store's only catalog: GC reclaims objects by +// scanning ledger rows, never by enumerating the bucket, so every object +// write is paired with a ledger write here. +// +// This module is browser-safe: it holds the adapter interface, the key +// derivation, the put/touch orchestration, and the GC sweep's read side. The +// adapters themselves are node-only and live in the realm-server package. + +// What an adapter knows about a stored object. Content type is recorded here +// for bucket-browsing convenience only — the ledger's `content_type` column +// is the serving-path source of truth (the local-disk adapter stores no +// metadata at all and never reports one). +export interface MediaObjectStat { + size: number; + contentType?: string; +} + +// One backing store for MediaCache objects. Implementations must be safe to +// call with keys they have never seen: `head`/`getStream` report absence with +// `undefined`, and `delete` of a missing key is a successful no-op (the GC +// sweep re-deletes keys whose ledger rows survived a crashed sweep). +// +// `put` must leave the object durable before resolving, and may skip the +// upload when the key is already stored — the key is a hash of the bytes, so +// an existing object under the same key already holds them (dedupe-on-write). +export interface MediaCacheAdapter { + put( + key: string, + bytes: Uint8Array, + opts: { contentType: string }, + ): Promise; + head(key: string): Promise; + getStream(key: string): Promise | undefined>; + delete(key: string): Promise; +} + +// GC policy differs by how a capture came to exist. A 'declared' capture +// (an indexing-time declared screenshot) lives until a newer generation +// supersedes it or its source is deleted; an 'on-demand' capture (URL DSL / +// POST) additionally ages out when unused, since nothing re-creates demand +// for it except another request. +export type MediaCacheLane = 'declared' | 'on-demand'; + +// The identity of one ledger row: one capture of one source instance under +// one canonical spec at one generation. +export interface MediaCacheEntryKey { + realmURL: string; + sourceURL: string; + captureSpecHash: string; + sourceGeneration: number; +} + +export interface MediaCacheEntry extends MediaCacheEntryKey { + objectKey: string; + sourceContentHash: string | null; + lane: MediaCacheLane; + contentType: string; + sizeBytes: number; + createdAt: number; + lastAccessedAt: number; +} + +// The content address for a blob of output bytes. sha256 rather than the +// realm-file `computeContentHash`: that one samples large content (head + +// tail) to bound synchronous main-thread cost, which is the wrong trade for +// a true content address — a false key match here would serve one capture's +// bytes under another's identity forever. +export async function computeMediaCacheKey(bytes: Uint8Array): Promise { + let hash = new Sha256(); + hash.update(bytes); + return uint8ArrayToHex(await hash.digest()); +} + +// Stores one capture: derives the content address, makes the object durable, +// then records the ledger row. Object-before-ledger so a ledger row never +// points at bytes that aren't durable; a crash between the two leaves an +// unreferenced object that the capture's own retry re-references (same +// bytes → same key). The ledger write is an upsert on the capture identity: +// a re-capture that produced different bytes repoints the row at the new +// object, and the old object is reclaimed by the GC sweep once nothing else +// references it. +export async function putMedia( + dbAdapter: DBAdapter, + adapter: MediaCacheAdapter, + { + bytes, + contentType, + realmURL, + sourceURL, + captureSpecHash, + sourceGeneration, + sourceContentHash = null, + lane, + }: MediaCacheEntryKey & { + bytes: Uint8Array; + contentType: string; + lane: MediaCacheLane; + sourceContentHash?: string | null; + }, +): Promise<{ objectKey: string; sizeBytes: number }> { + let objectKey = await computeMediaCacheKey(bytes); + await adapter.put(objectKey, bytes, { contentType }); + let now = Date.now(); + let { nameExpressions, valueExpressions } = asExpressions({ + realm_url: realmURL, + source_url: sourceURL, + capture_spec_hash: captureSpecHash, + source_generation: sourceGeneration, + object_key: objectKey, + source_content_hash: sourceContentHash, + lane, + content_type: contentType, + size_bytes: bytes.length, + created_at: now, + last_accessed_at: now, + }); + await query( + dbAdapter, + upsert( + 'media_cache_ledger', + 'media_cache_ledger_pkey', + nameExpressions, + valueExpressions, + ), + ); + return { objectKey, sizeBytes: bytes.length }; +} + +// Serving-path bump for the on-demand age-out lane: reading a capture resets +// its idle clock. Coarse is fine — the lane's TTL is measured in days, so +// callers may throttle their bumps. +export async function touchMediaCacheEntry( + dbAdapter: DBAdapter, + entryKey: MediaCacheEntryKey, + now = Date.now(), +): Promise { + await query(dbAdapter, [ + `UPDATE media_cache_ledger SET last_accessed_at =`, + param(now), + `WHERE realm_url =`, + param(entryKey.realmURL), + `AND source_url =`, + param(entryKey.sourceURL), + `AND capture_spec_hash =`, + param(entryKey.captureSpecHash), + `AND source_generation =`, + param(entryKey.sourceGeneration), + ] as Expression); +} + +// --------------------------------------------------------------------------- +// GC sweep read side — mirrors `prerender-html-reconcile.ts`: pure queries +// plus a pure planning step; the enqueue/delete orchestration lives in the +// `media-cache-gc` task. A healthy, fully-live ledger yields no candidates. +// --------------------------------------------------------------------------- + +// No row younger than this is ever collected, whatever lane it is in: the +// delay is what keeps the sweep from racing an in-flight capture or serve +// (a row is written moments after its object; a serve streams moments after +// its ledger read). +export const MEDIA_CACHE_GC_MIN_AGE_MS = 24 * 60 * 60 * 1000; +// The on-demand lane's idle TTL: a DSL/POST capture nobody has requested for +// this long is reclaimed. Anything needing a longer-lived artifact should be +// a declared screenshot, which never ages out. +export const MEDIA_CACHE_ON_DEMAND_TTL_MS = 30 * 24 * 60 * 60 * 1000; + +export type MediaCacheGcReason = 'tombstoned' | 'superseded' | 'expired'; + +export interface MediaCacheGcCandidate extends MediaCacheEntryKey { + objectKey: string; + reason: MediaCacheGcReason; +} + +// Ledger rows the sweep may reclaim, oldest reasons first: +// - 'tombstoned': the source instance is deleted (its `boxel_index` row is +// a tombstone) — the capture inherits the deletion. +// - 'superseded': a newer-generation row exists for the same capture +// identity, and has for at least the min-age (so a serve that resolved +// the old row just before the swap can still finish streaming). +// - 'expired': an on-demand capture idle past the TTL. +// Every arm additionally requires the row itself to be older than min-age. +// The jsonb-free SQL here is still Postgres-shaped (row-value EXISTS, +// bigint arithmetic); like the reconcile scans, the GC task runs solely +// behind the Postgres queue. +export async function findMediaCacheGcCandidates( + dbAdapter: DBAdapter, + { + now = Date.now(), + minAgeMs = MEDIA_CACHE_GC_MIN_AGE_MS, + onDemandTtlMs = MEDIA_CACHE_ON_DEMAND_TTL_MS, + }: { now?: number; minAgeMs?: number; onDemandTtlMs?: number } = {}, +): Promise { + let minAgeCutoff = now - minAgeMs; + let idleCutoff = now - onDemandTtlMs; + let rows = (await query(dbAdapter, [ + `SELECT realm_url, source_url, capture_spec_hash, source_generation, object_key, + CASE + WHEN EXISTS ( + SELECT 1 FROM boxel_index i + WHERE i.url = r.source_url AND i.realm_url = r.realm_url + AND i.type = 'instance' AND i.is_deleted IS TRUE + ) THEN 'tombstoned' + WHEN EXISTS ( + SELECT 1 FROM media_cache_ledger n + WHERE n.realm_url = r.realm_url AND n.source_url = r.source_url + AND n.capture_spec_hash = r.capture_spec_hash + AND n.source_generation > r.source_generation + AND n.created_at <`, + param(minAgeCutoff), + `) THEN 'superseded' + ELSE 'expired' + END AS reason + FROM media_cache_ledger r + WHERE r.created_at <`, + param(minAgeCutoff), + `AND ( + EXISTS ( + SELECT 1 FROM boxel_index i + WHERE i.url = r.source_url AND i.realm_url = r.realm_url + AND i.type = 'instance' AND i.is_deleted IS TRUE + ) + OR EXISTS ( + SELECT 1 FROM media_cache_ledger n + WHERE n.realm_url = r.realm_url AND n.source_url = r.source_url + AND n.capture_spec_hash = r.capture_spec_hash + AND n.source_generation > r.source_generation + AND n.created_at <`, + param(minAgeCutoff), + `) + OR (r.lane = 'on-demand' AND r.last_accessed_at <`, + param(idleCutoff), + `))`, + ] as Expression)) as { + realm_url: string; + source_url: string; + capture_spec_hash: string; + source_generation: number | string; + object_key: string; + reason: MediaCacheGcReason; + }[]; + return rows.map((row) => ({ + realmURL: row.realm_url, + sourceURL: row.source_url, + captureSpecHash: row.capture_spec_hash, + sourceGeneration: Number(row.source_generation), + objectKey: row.object_key, + reason: row.reason, + })); +} + +// Total ledger reference counts for the given object keys — candidates and +// survivors alike — so the planner can tell which objects the sweep's row +// deletions would orphan. +export async function findMediaCacheKeyReferenceCounts( + dbAdapter: DBAdapter, + objectKeys: string[], +): Promise> { + let counts = new Map(); + if (objectKeys.length === 0) { + return counts; + } + let rows = (await query(dbAdapter, [ + `SELECT object_key, COUNT(*) AS refs FROM media_cache_ledger WHERE object_key IN`, + ...addExplicitParens( + separatedByCommas([...new Set(objectKeys)].map((key) => [param(key)])), + ), + `GROUP BY object_key`, + ] as Expression)) as { object_key: string; refs: number | string }[]; + for (let row of rows) { + counts.set(row.object_key, Number(row.refs)); + } + return counts; +} + +export interface MediaCacheGcPlan { + rows: MediaCacheGcCandidate[]; + // Object keys whose every ledger reference is in `rows` — deleting the + // rows orphans the object, so the object goes too. + objectKeys: string[]; +} + +// Pure reconciliation: an object is reclaimed only when this sweep's row +// deletions remove its last ledger reference; an object some surviving row +// still points at (dedupe across captures) keeps its bytes. +export function planMediaCacheGcDeletions( + candidates: MediaCacheGcCandidate[], + referenceCounts: Map, +): MediaCacheGcPlan { + let candidateRefs = new Map(); + for (let candidate of candidates) { + candidateRefs.set( + candidate.objectKey, + (candidateRefs.get(candidate.objectKey) ?? 0) + 1, + ); + } + let objectKeys: string[] = []; + for (let [key, candidateCount] of candidateRefs) { + if ((referenceCounts.get(key) ?? 0) <= candidateCount) { + objectKeys.push(key); + } + } + return { rows: candidates, objectKeys }; +} + +// Removes the given ledger rows by exact identity. Chunked so a large sweep +// never builds one unbounded statement. +const DELETE_ROWS_CHUNK_SIZE = 200; +export async function deleteMediaCacheRows( + dbAdapter: DBAdapter, + rows: MediaCacheEntryKey[], +): Promise { + for (let i = 0; i < rows.length; i += DELETE_ROWS_CHUNK_SIZE) { + let chunk = rows.slice(i, i + DELETE_ROWS_CHUNK_SIZE); + await query(dbAdapter, [ + `DELETE FROM media_cache_ledger + WHERE (realm_url, source_url, capture_spec_hash, source_generation) IN`, + ...addExplicitParens( + separatedByCommas( + chunk.map((row) => + addExplicitParens( + separatedByCommas([ + [param(row.realmURL)], + [param(row.sourceURL)], + [param(row.captureSpecHash)], + [param(row.sourceGeneration)], + ]), + ), + ), + ), + ), + ] as Expression); + } +} diff --git a/packages/runtime-common/tasks/index.ts b/packages/runtime-common/tasks/index.ts index 54e125cdc31..d23dd72c229 100644 --- a/packages/runtime-common/tasks/index.ts +++ b/packages/runtime-common/tasks/index.ts @@ -10,6 +10,7 @@ import type { VirtualNetwork, } from '../index.ts'; import type { JobInfo, IndexingProgressEvent } from '../worker.ts'; +import type { MediaCacheAdapter } from '../media-cache.ts'; import type { RealmEventContent } from '@cardstack/base/matrix-event'; export type * from './lint.ts'; export * from '#lint-task'; @@ -17,6 +18,7 @@ export * from './full-reindex.ts'; export * from './daily-credit-grant.ts'; export * from './copy.ts'; export * from './indexer.ts'; +export * from './media-cache-gc.ts'; export * from './prerender-html.ts'; export * from './prerender-html-reconcile.ts'; export * from './run-command.ts'; @@ -33,6 +35,9 @@ export interface TaskArgs { virtualNetwork: VirtualNetwork; log: LoggerInstance; matrixURL: string; + // The MediaCache's object store. Optional: a worker process without one + // configured still registers media-cache jobs, whose tasks then no-op. + mediaCacheAdapter?: MediaCacheAdapter; getReader(fetch: typeof global.fetch, realmURL: string): Reader; getAuthedFetch(args: WorkerArgs): Promise; createPrerenderAuth(userId: string, permissions: RealmPermissions): string; diff --git a/packages/runtime-common/tasks/media-cache-gc.ts b/packages/runtime-common/tasks/media-cache-gc.ts new file mode 100644 index 00000000000..e66a5471238 --- /dev/null +++ b/packages/runtime-common/tasks/media-cache-gc.ts @@ -0,0 +1,134 @@ +import type * as JSONTypes from 'json-typescript'; +import type { Task } from './index.ts'; +import { jobIdentity } from '../index.ts'; +import { + registerQueueJobDefinition, + type QueueCoalesceContext, + type QueueCoalesceDecision, +} from '../queue.ts'; +import { + deleteMediaCacheRows, + findMediaCacheGcCandidates, + findMediaCacheKeyReferenceCounts, + planMediaCacheGcDeletions, + type MediaCacheGcReason, +} from '../media-cache.ts'; + +// The cron enqueues this job with no arguments; it sweeps the whole ledger. +type MediaCacheGcArgs = JSONTypes.Object; + +export interface MediaCacheGcResult extends JSONTypes.Object { + rowsDeleted: number; + objectsDeleted: number; + // Objects whose adapter delete failed this sweep. Their ledger rows are + // kept so the next sweep re-finds and retries them. + objectDeleteFailures: number; +} + +export { mediaCacheGc }; + +// The fixed concurrency group already serializes execution to one sweep at a +// time; this additionally collapses a queued tick into any pending or +// in-flight sweep so overlapping cron ticks never pile up. The sweep carries +// no args, so there is nothing to merge — a twin simply wins. +function chooseMediaCacheGcCoalesceDecision( + context: QueueCoalesceContext, +): QueueCoalesceDecision { + let { incoming, candidates, inFlightCandidates } = context; + let twin = + candidates.find((candidate) => candidate.jobType === incoming.jobType) ?? + inFlightCandidates.find( + (candidate) => candidate.jobType === incoming.jobType, + ); + if (!twin) { + return { type: 'insert' }; + } + return { type: 'join', jobId: twin.id }; +} + +registerQueueJobDefinition({ + jobType: 'media-cache-gc', + coalesce: chooseMediaCacheGcCoalesceDecision, +}); + +// Reconcile-style GC for the MediaCache: reclaims ledger rows whose capture +// is superseded by a newer generation, whose source instance is tombstoned, +// or (on-demand lane) idle past the TTL — then deletes each object whose +// last ledger reference those rows held. Objects are deleted before their +// rows so a sweep that dies mid-way leaves rows behind for the next sweep to +// re-find, never bytes the ledger no longer knows about (the ledger is the +// only catalog — an object with no row is unreachable to GC forever). +// Adapter deletes are idempotent, so the re-found rows re-delete harmlessly. +// A sweep over a healthy, fully-live ledger deletes nothing. +const mediaCacheGc: Task = ({ + dbAdapter, + mediaCacheAdapter, + reportStatus, + log, +}) => + async function (args) { + let { jobInfo } = args; + reportStatus(jobInfo, 'start'); + + if (!mediaCacheAdapter) { + log.info( + `${jobIdentity(jobInfo)} media-cache gc: no media cache adapter configured; nothing to sweep`, + ); + reportStatus(jobInfo, 'finish'); + return { rowsDeleted: 0, objectsDeleted: 0, objectDeleteFailures: 0 }; + } + + let candidates = await findMediaCacheGcCandidates(dbAdapter); + if (candidates.length === 0) { + log.debug(`${jobIdentity(jobInfo)} media-cache gc: no candidates`); + reportStatus(jobInfo, 'finish'); + return { rowsDeleted: 0, objectsDeleted: 0, objectDeleteFailures: 0 }; + } + + let referenceCounts = await findMediaCacheKeyReferenceCounts( + dbAdapter, + candidates.map((candidate) => candidate.objectKey), + ); + let plan = planMediaCacheGcDeletions(candidates, referenceCounts); + + let objectsDeleted = 0; + let failedKeys = new Set(); + for (let objectKey of plan.objectKeys) { + try { + await mediaCacheAdapter.delete(objectKey); + objectsDeleted++; + } catch (error: any) { + failedKeys.add(objectKey); + log.error( + `${jobIdentity(jobInfo)} media-cache gc: failed to delete object ${objectKey}`, + error, + ); + } + } + + // A row whose object delete failed must survive: it is the ledger's only + // record that the object exists, and the next sweep's retry path. + let rowsToDelete = plan.rows.filter( + (row) => !failedKeys.has(row.objectKey), + ); + await deleteMediaCacheRows(dbAdapter, rowsToDelete); + + let byReason = new Map(); + for (let row of rowsToDelete) { + byReason.set(row.reason, (byReason.get(row.reason) ?? 0) + 1); + } + log.info( + `${jobIdentity(jobInfo)} media-cache gc: deleted ${rowsToDelete.length} ledger row(s) ` + + `(${[...byReason].map(([reason, count]) => `${count} ${reason}`).join(', ') || 'none'}) ` + + `and ${objectsDeleted} object(s)` + + (failedKeys.size > 0 + ? `; ${failedKeys.size} object delete(s) failed and were kept for the next sweep` + : ''), + ); + reportStatus(jobInfo, 'finish'); + return { + rowsDeleted: rowsToDelete.length, + objectsDeleted, + objectDeleteFailures: failedKeys.size, + }; + }; diff --git a/packages/runtime-common/worker.ts b/packages/runtime-common/worker.ts index ff8a31cc5dd..4251e47f778 100644 --- a/packages/runtime-common/worker.ts +++ b/packages/runtime-common/worker.ts @@ -26,6 +26,7 @@ import { CachingDefinitionLookup, } from './index.ts'; import { MatrixClient } from './matrix-client.ts'; +import type { MediaCacheAdapter } from './media-cache.ts'; import * as Tasks from './tasks/index.ts'; import type { WorkerArgs, TaskArgs } from './tasks/index.ts'; import type { RealmEventContent } from '@cardstack/base/matrix-event'; @@ -153,6 +154,7 @@ export class Worker { #reportRealmEvent: ((event: RealmEventContent) => void) | undefined; #realmServerMatrixUsername; #indexJobsOnly: boolean; + #mediaCacheAdapter: MediaCacheAdapter | undefined; #createPrerenderAuth: ( userId: string, permissions: RealmPermissions, @@ -173,6 +175,7 @@ export class Worker { prerenderer, createPrerenderAuth, indexJobsOnly, + mediaCacheAdapter, }: { indexWriter: IndexWriter; queue: QueueRunner; @@ -189,6 +192,9 @@ export class Worker { // When true, register handlers only for INDEX_JOB_TYPES so this worker // is a dedicated indexing lane — see INDEX_JOB_TYPES above. indexJobsOnly?: boolean; + // The MediaCache object store, absent when the process has none + // configured (media-cache tasks then no-op). + mediaCacheAdapter?: MediaCacheAdapter; createPrerenderAuth: ( userId: string, permissions: RealmPermissions, @@ -208,6 +214,7 @@ export class Worker { this.#prerenderer = prerenderer; this.#createPrerenderAuth = createPrerenderAuth; this.#indexJobsOnly = indexJobsOnly ?? false; + this.#mediaCacheAdapter = mediaCacheAdapter; } async run() { @@ -232,6 +239,7 @@ export class Worker { reportProgress: this.reportProgress.bind(this), reportRealmEvent: this.reportRealmEvent.bind(this), createPrerenderAuth: this.#createPrerenderAuth, + mediaCacheAdapter: this.#mediaCacheAdapter, }; let registrations: Record Promise | unknown> = { @@ -255,6 +263,8 @@ export class Worker { `prerender-html-reconcile`, Tasks['prerenderHtmlReconcile'](taskArgs), ), + 'media-cache-gc': () => + this.#queue.register(`media-cache-gc`, Tasks['mediaCacheGc'](taskArgs)), 'copy-index': () => this.#queue.register(`copy-index`, Tasks['copy'](taskArgs)), 'lint-source': () => From 23be42f28e2750320338970c1ed7c17620e36f89 Mon Sep 17 00:00:00 2001 From: Luke Melia Date: Thu, 20 Aug 2026 17:56:04 -0400 Subject: [PATCH 2/2] Address review: background-tier GC priority, inline reclaim of repointed objects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The GC enqueue now uses systemInitiatedPrerenderHtmlPriority (0, the all-priority pool's floor) as its comment always claimed, matching the prerender-html reconcile sibling instead of running co-equal with indexing. putMedia now reclaims the object a repointing upsert strips of its last ledger reference: the sweep can never see an unreferenced object (the ledger is its only catalog), so without this a same-identity re-capture that changed bytes leaked the old object permanently. Reclaim is inline, refcount-guarded, and best-effort — a failed delete is logged as a leak and never fails the put. Co-Authored-By: Claude Fable 5 --- .../realm-server/scripts/media-cache-gc.ts | 12 ++-- .../realm-server/tests/media-cache-gc-test.ts | 60 +++++++++++++++++++ packages/runtime-common/media-cache.ts | 59 +++++++++++++++++- 3 files changed, 125 insertions(+), 6 deletions(-) diff --git a/packages/realm-server/scripts/media-cache-gc.ts b/packages/realm-server/scripts/media-cache-gc.ts index 21547586fdf..400a15ff2c6 100644 --- a/packages/realm-server/scripts/media-cache-gc.ts +++ b/packages/realm-server/scripts/media-cache-gc.ts @@ -1,6 +1,9 @@ import '../instrument.ts'; import '../setup-logger.ts'; // This should be first -import { logger, systemInitiatedPriority } from '@cardstack/runtime-common'; +import { + logger, + systemInitiatedPrerenderHtmlPriority, +} from '@cardstack/runtime-common'; import { PgAdapter, PgQueuePublisher } from '@cardstack/postgres'; import * as Sentry from '@sentry/node'; @@ -9,10 +12,11 @@ const MEDIA_CACHE_GC_JOB_TIMEOUT_SEC = 10 * 60; // Enqueue the GC sweep rather than sweeping inline in the worker-manager // process: a worker scans the ledger and deletes reclaimed rows/objects. The -// sweep runs at the background tier (priority 0) so it never competes with -// indexing or user work. +// sweep runs at the background tier (priority 0, the all-priority pool's +// floor — the same tier the prerender-html reconcile scan uses) so it never +// competes with indexing or user work. export async function enqueueMediaCacheGc({ - priority = systemInitiatedPriority, + priority = systemInitiatedPrerenderHtmlPriority, migrateDB, }: { priority?: number; diff --git a/packages/realm-server/tests/media-cache-gc-test.ts b/packages/realm-server/tests/media-cache-gc-test.ts index bfce1ee8a56..17731505de1 100644 --- a/packages/realm-server/tests/media-cache-gc-test.ts +++ b/packages/realm-server/tests/media-cache-gc-test.ts @@ -411,6 +411,66 @@ module(basename(import.meta.filename), function (hooks) { second.objectKey, 'the row points at the latest bytes', ); + assert.notOk( + adapter.objects.has(first.objectKey), + 'the repointed-away object is reclaimed inline — the GC sweep could never find it', + ); + assert.deepEqual(adapter.deleted, [first.objectKey]); + }); + + test('a repoint keeps the prior object when another capture still names it', async function (assert) { + let sharedBytes = new Uint8Array([1]); + await putMedia(dbAdapter, adapter, { + ...entryKey, + bytes: sharedBytes, + contentType: 'image/png', + lane: 'on-demand', + }); + // A second capture identity produced identical bytes (dedupe), so the + // object is shared. + let other = await putMedia(dbAdapter, adapter, { + ...entryKey, + captureSpecHash: 'spec-2', + bytes: sharedBytes, + contentType: 'image/png', + lane: 'on-demand', + }); + await putMedia(dbAdapter, adapter, { + ...entryKey, + bytes: new Uint8Array([2]), + contentType: 'image/png', + lane: 'on-demand', + }); + + assert.ok( + adapter.objects.has(other.objectKey), + 'the shared object survives the repoint', + ); + assert.deepEqual(adapter.deleted, []); + }); + + test('a failed repoint reclaim does not fail the put', async function (assert) { + let first = await putMedia(dbAdapter, adapter, { + ...entryKey, + bytes: new Uint8Array([1]), + contentType: 'image/png', + lane: 'on-demand', + }); + adapter.failDeletesFor.add(first.objectKey); + + let second = await putMedia(dbAdapter, adapter, { + ...entryKey, + bytes: new Uint8Array([2]), + contentType: 'image/png', + lane: 'on-demand', + }); + + let rows = await ledgerRows(); + assert.strictEqual( + rows[0].object_key, + second.objectKey, + 'the new capture is recorded despite the failed reclaim', + ); }); test('touch bumps last_accessed_at', async function (assert) { diff --git a/packages/runtime-common/media-cache.ts b/packages/runtime-common/media-cache.ts index da1b0ca89f3..cfb9900e528 100644 --- a/packages/runtime-common/media-cache.ts +++ b/packages/runtime-common/media-cache.ts @@ -1,5 +1,6 @@ import { Sha256 } from '@aws-crypto/sha256-js'; import type { DBAdapter } from './db.ts'; +import { logger } from './log.ts'; import { addExplicitParens, asExpressions, @@ -11,6 +12,8 @@ import { } from './expression.ts'; import { uint8ArrayToHex } from './index.ts'; +const log = logger('media-cache'); + // The MediaCache: a content-addressed store for derived media (screenshots), // split across two channels that this module keeps consistent: // @@ -98,8 +101,17 @@ export async function computeMediaCacheKey(bytes: Uint8Array): Promise { // unreferenced object that the capture's own retry re-references (same // bytes → same key). The ledger write is an upsert on the capture identity: // a re-capture that produced different bytes repoints the row at the new -// object, and the old object is reclaimed by the GC sweep once nothing else -// references it. +// object. +// +// A repoint strips the prior object of this row's reference, and the GC +// sweep structurally cannot recover it — the ledger is the store's only +// catalog, so an object no row names is invisible to every later sweep. +// The put path therefore reclaims inline: once the repointing upsert lands, +// the prior object is deleted (best-effort) unless some other row still +// names it. A failed or crash-interrupted delete permanently leaks that one +// object; this is the same one-object crash exposure the put's own +// object-before-ledger ordering already carries, and is logged when +// observed. export async function putMedia( dbAdapter: DBAdapter, adapter: MediaCacheAdapter, @@ -121,6 +133,17 @@ export async function putMedia( ): Promise<{ objectKey: string; sizeBytes: number }> { let objectKey = await computeMediaCacheKey(bytes); await adapter.put(objectKey, bytes, { contentType }); + let priorRows = (await query(dbAdapter, [ + `SELECT object_key FROM media_cache_ledger WHERE realm_url =`, + param(realmURL), + `AND source_url =`, + param(sourceURL), + `AND capture_spec_hash =`, + param(captureSpecHash), + `AND source_generation =`, + param(sourceGeneration), + ] as Expression)) as { object_key: string }[]; + let priorObjectKey = priorRows[0]?.object_key; let now = Date.now(); let { nameExpressions, valueExpressions } = asExpressions({ realm_url: realmURL, @@ -144,9 +167,41 @@ export async function putMedia( valueExpressions, ), ); + if (priorObjectKey && priorObjectKey !== objectKey) { + await reclaimRepointedObject(dbAdapter, adapter, priorObjectKey); + } return { objectKey, sizeBytes: bytes.length }; } +// Reclaims the object a repointing upsert stripped of its reference — unless +// another capture's row still names it (dedupe across captures), in which +// case the bytes stay and the sweep handles them when their last row goes. +// Best-effort by design: a failed delete cannot fail the put (the new +// capture is already durable and recorded), but the orphan is then +// permanently unreachable, so it is logged. Shares the put/GC family of +// accepted races: a concurrent dedupe-put that re-references this key inside +// the check-then-delete window loses its bytes and self-heals on its next +// put, serving misses until then. +async function reclaimRepointedObject( + dbAdapter: DBAdapter, + adapter: MediaCacheAdapter, + objectKey: string, +): Promise { + let refs = await findMediaCacheKeyReferenceCounts(dbAdapter, [objectKey]); + if ((refs.get(objectKey) ?? 0) > 0) { + return; + } + try { + await adapter.delete(objectKey); + } catch (e) { + log.warn( + `failed to delete repointed-away media cache object ${objectKey}; ` + + `no ledger row references it, so these bytes are leaked`, + e, + ); + } +} + // Serving-path bump for the on-demand age-out lane: reading a capture resets // its idle clock. Coarse is fine — the lane's TTL is measured in days, so // callers may throttle their bumps.