diff --git a/.changeset/cleanup-media-usage-generations.md b/.changeset/cleanup-media-usage-generations.md new file mode 100644 index 0000000000..42a3a594d6 --- /dev/null +++ b/.changeset/cleanup-media-usage-generations.md @@ -0,0 +1,5 @@ +--- +"emdash": patch +--- + +Fixes retention of superseded media-usage projection generations during scheduled maintenance. diff --git a/AGENTS.md b/AGENTS.md index d849e583bc..8cf4c634b6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -367,7 +367,7 @@ In libraries used in a Worker but not themselves Workers, install `@cloudflare/w # Testing - **Framework:** vitest. Tests in `packages/core/tests/`. -- **No mocks for the DB.** SQLite (`better-sqlite3`) by default. PostgreSQL parity tests via a real `pg` connection with per-test schema isolation (set `PG_CONNECTION_STRING` to opt in). +- **No mocks for the DB.** SQLite (`better-sqlite3`) by default. PostgreSQL parity tests via a real `pg` connection with per-test schema isolation (set `EMDASH_TEST_PG` to a connection string for a role with `CREATEDB` to opt in). - **Utilities:** `tests/utils/test-db.ts` exposes `setupTestDatabase()`, `setupTestDatabaseWithCollections()`, `teardownTestDatabase()` for SQLite and `setupTestPostgresDatabase()` etc. for Postgres. Dialect-agnostic: `setupForDialect`, `setupForDialectWithCollections`, `teardownForDialect`, plus `describeEachDialect(name, fn)`. Use the dialect wrapper for query-builder code -- regressions tend to be dialect-specific. - **Structure:** `tests/unit/`, `tests/integration/`, `tests/e2e/` (Playwright). Test files mirror source structure. Each test gets a fresh DB. diff --git a/packages/core/src/cleanup.ts b/packages/core/src/cleanup.ts index 1dc295d204..fe4227bd2a 100644 --- a/packages/core/src/cleanup.ts +++ b/packages/core/src/cleanup.ts @@ -17,6 +17,7 @@ import { MediaRepository } from "./database/repositories/media.js"; import { RevisionRepository } from "./database/repositories/revision.js"; import type { Database } from "./database/types.js"; import { removeUploadAttempt } from "./media/upload-attempts.js"; +import { cleanupMediaUsage } from "./media/usage/cleanup.js"; import type { Storage } from "./storage/types.js"; /** @@ -30,6 +31,7 @@ export interface CleanupResult { pendingUploadFiles: number; uploadAttempts: number; revisionsPruned: number; + mediaUsage: number; } /** Max revisions to keep per entry during periodic pruning */ @@ -60,6 +62,7 @@ export async function runSystemCleanup( pendingUploadFiles: -1, uploadAttempts: -1, revisionsPruned: -1, + mediaUsage: -1, }; // 1. Passkey challenges (expire after 60s, clean anything past 5 min) @@ -136,6 +139,13 @@ export async function runSystemCleanup( console.error("[cleanup] Failed to prune revisions:", error); } + try { + const mediaUsage = await cleanupMediaUsage(db); + result.mediaUsage = mediaUsage.status === "failed" ? -1 : mediaUsage.deletedRows; + } catch (error) { + console.error("[cleanup] Failed to clean media usage:", error); + } + return result; } diff --git a/packages/core/src/database/migrations/055_media_usage_cleanup.ts b/packages/core/src/database/migrations/055_media_usage_cleanup.ts new file mode 100644 index 0000000000..1f4b586c77 --- /dev/null +++ b/packages/core/src/database/migrations/055_media_usage_cleanup.ts @@ -0,0 +1,77 @@ +import { sql, type Kysely } from "kysely"; + +import { currentTimestamp } from "../dialect-helpers.js"; + +const CLEANUP_TASK_KEY = "projection_gc"; +const INITIAL_ELIGIBLE_AT = "1970-01-01T00:00:00.000Z"; + +export async function up(db: Kysely): Promise { + await db.schema + .createIndex("idx__emdash_media_usage_cleanup_scan") + .ifNotExists() + .on("_emdash_media_usage") + .columns(["created_at", "id", "source_key", "generation"]) + .execute(); + + await db.schema + .createTable("_emdash_media_usage_cleanup") + .ifNotExists() + .addColumn("task_key", "text", (c) => c.primaryKey()) + .addColumn("lease_token", "text") + .addColumn("lease_expires_at", "text") + .addColumn("next_eligible_at", "text", (c) => c.notNull()) + .addColumn("cursor_created_at", "text") + .addColumn("cursor_id", "text") + .addColumn("scan_before_at", "text") + .addColumn("consecutive_failures", "integer", (c) => c.notNull().defaultTo(0)) + .addColumn("last_started_at", "text") + .addColumn("last_completed_at", "text") + .addColumn("last_candidate_count", "integer", (c) => c.notNull().defaultTo(0)) + .addColumn("last_deleted_orphans", "integer", (c) => c.notNull().defaultTo(0)) + .addColumn("last_deleted_stale", "integer", (c) => c.notNull().defaultTo(0)) + .addColumn("last_deleted_abandoned", "integer", (c) => c.notNull().defaultTo(0)) + .addColumn("last_deleted_write_leases", "integer", (c) => c.notNull().defaultTo(0)) + .addColumn("last_backlog_lower_bound", "integer", (c) => c.notNull().defaultTo(0)) + .addColumn("last_scan_has_more", "integer", (c) => c.notNull().defaultTo(0)) + .addColumn("last_duration_ms", "integer", (c) => c.notNull().defaultTo(0)) + .addColumn("last_error_code", "text") + .addColumn("updated_at", "text", (c) => c.notNull().defaultTo(currentTimestamp(db))) + .execute(); + + await db.schema + .createTable("_emdash_media_usage_generation_writes") + .ifNotExists() + .addColumn("source_key", "text", (c) => c.notNull()) + .addColumn("generation", "text", (c) => c.notNull()) + .addColumn("lease_token", "text", (c) => c.primaryKey()) + .addColumn("expires_at", "text", (c) => c.notNull()) + .addColumn("created_at", "text", (c) => c.notNull().defaultTo(currentTimestamp(db))) + .addUniqueConstraint("_emdash_media_usage_generation_writes_source_generation", [ + "source_key", + "generation", + ]) + .execute(); + + await db.schema + .createIndex("idx__emdash_media_usage_generation_writes_expiry") + .ifNotExists() + .on("_emdash_media_usage_generation_writes") + .columns(["expires_at", "lease_token"]) + .execute(); + + await sql` + INSERT INTO _emdash_media_usage_cleanup (task_key, next_eligible_at) + VALUES (${CLEANUP_TASK_KEY}, ${INITIAL_ELIGIBLE_AT}) + ON CONFLICT (task_key) DO NOTHING + `.execute(db); +} + +export async function down(db: Kysely): Promise { + await db.schema + .dropIndex("idx__emdash_media_usage_generation_writes_expiry") + .ifExists() + .execute(); + await db.schema.dropTable("_emdash_media_usage_generation_writes").ifExists().execute(); + await db.schema.dropTable("_emdash_media_usage_cleanup").ifExists().execute(); + await db.schema.dropIndex("idx__emdash_media_usage_cleanup_scan").ifExists().execute(); +} diff --git a/packages/core/src/database/migrations/056_media_usage_cleanup_fence.ts b/packages/core/src/database/migrations/056_media_usage_cleanup_fence.ts new file mode 100644 index 0000000000..396ec45118 --- /dev/null +++ b/packages/core/src/database/migrations/056_media_usage_cleanup_fence.ts @@ -0,0 +1,295 @@ +import { sql, type Kysely } from "kysely"; + +import { columnExists, currentTimestamp, isPostgres } from "../dialect-helpers.js"; + +export async function up(db: Kysely): Promise { + if (!(await columnExists(db, "_emdash_media_usage", "cleanup_lease_token"))) { + await db.schema + .alterTable("_emdash_media_usage") + .addColumn("cleanup_lease_token", "text") + .execute(); + } + await db.schema + .createTable("_emdash_media_usage_cleanup_fence") + .ifNotExists() + .addColumn("task_key", "text", (c) => c.primaryKey()) + .addColumn("generation_floor", "text", (c) => c.notNull()) + .addColumn("updated_at", "text", (c) => c.notNull().defaultTo(currentTimestamp(db))) + .execute(); + + if (isPostgres(db)) { + await sql + .raw("DROP TRIGGER IF EXISTS emdash_media_usage_record_cleanup_fence ON _emdash_media_usage") + .execute(db); + await sql + .raw( + "DROP TRIGGER IF EXISTS emdash_media_usage_fence_source_generation_update ON _emdash_media_usage_sources", + ) + .execute(db); + await sql + .raw( + "DROP TRIGGER IF EXISTS emdash_media_usage_fence_source_generation_insert ON _emdash_media_usage_sources", + ) + .execute(db); + await sql + .raw( + "DROP TRIGGER IF EXISTS emdash_media_usage_lock_cleanup_update ON _emdash_media_usage_sources", + ) + .execute(db); + await sql + .raw( + "DROP TRIGGER IF EXISTS emdash_media_usage_lock_cleanup_insert ON _emdash_media_usage_sources", + ) + .execute(db); + await sql + .raw( + "DROP TRIGGER IF EXISTS emdash_media_usage_lock_cleanup_delete ON _emdash_media_usage_sources", + ) + .execute(db); + await sql + .raw(` + CREATE OR REPLACE FUNCTION emdash_media_usage_lock_cleanup() + RETURNS trigger + LANGUAGE plpgsql + AS $$ + BEGIN + PERFORM 1 + FROM _emdash_media_usage_cleanup AS cleanup + WHERE cleanup.task_key = 'projection_gc' + FOR SHARE; + RETURN NULL; + END; + $$ + `) + .execute(db); + await sql + .raw(` + CREATE OR REPLACE FUNCTION emdash_media_usage_fence_source_generation() + RETURNS trigger + LANGUAGE plpgsql + AS $$ + BEGIN + IF EXISTS ( + SELECT 1 + FROM _emdash_media_usage_cleanup_fence AS fence + WHERE fence.task_key = 'projection_gc' + AND NEW.current_generation <= fence.generation_floor + ) AND NOT EXISTS ( + SELECT 1 + FROM _emdash_media_usage_generation_writes AS writer + WHERE writer.source_key = NEW.source_key + AND writer.generation = NEW.current_generation + AND writer.expires_at::timestamptz > clock_timestamp() + ) THEN + RETURN NULL; + END IF; + RETURN NEW; + END; + $$ + `) + .execute(db); + await sql + .raw(` + CREATE TRIGGER emdash_media_usage_lock_cleanup_insert + BEFORE INSERT ON _emdash_media_usage_sources + FOR EACH STATEMENT + EXECUTE FUNCTION emdash_media_usage_lock_cleanup() + `) + .execute(db); + await sql + .raw(` + CREATE TRIGGER emdash_media_usage_lock_cleanup_update + BEFORE UPDATE OF current_generation ON _emdash_media_usage_sources + FOR EACH STATEMENT + EXECUTE FUNCTION emdash_media_usage_lock_cleanup() + `) + .execute(db); + await sql + .raw(` + CREATE TRIGGER emdash_media_usage_lock_cleanup_delete + BEFORE DELETE ON _emdash_media_usage_sources + FOR EACH STATEMENT + EXECUTE FUNCTION emdash_media_usage_lock_cleanup() + `) + .execute(db); + await sql + .raw(` + CREATE OR REPLACE FUNCTION emdash_media_usage_record_cleanup_fence() + RETURNS trigger + LANGUAGE plpgsql + AS $$ + BEGIN + IF OLD.cleanup_lease_token IS NOT NULL AND EXISTS ( + SELECT 1 + FROM _emdash_media_usage_cleanup AS cleanup + WHERE cleanup.task_key = 'projection_gc' + AND cleanup.lease_token = OLD.cleanup_lease_token + ) THEN + INSERT INTO _emdash_media_usage_cleanup_fence ( + task_key, + generation_floor, + updated_at + ) + VALUES ('projection_gc', OLD.generation, CURRENT_TIMESTAMP::text) + ON CONFLICT (task_key) DO UPDATE SET + generation_floor = CASE + WHEN EXCLUDED.generation_floor > _emdash_media_usage_cleanup_fence.generation_floor + THEN EXCLUDED.generation_floor + ELSE _emdash_media_usage_cleanup_fence.generation_floor + END, + updated_at = EXCLUDED.updated_at; + END IF; + RETURN OLD; + END; + $$ + `) + .execute(db); + await sql + .raw(` + CREATE TRIGGER emdash_media_usage_fence_source_generation_insert + BEFORE INSERT ON _emdash_media_usage_sources + FOR EACH ROW + EXECUTE FUNCTION emdash_media_usage_fence_source_generation() + `) + .execute(db); + await sql + .raw(` + CREATE TRIGGER emdash_media_usage_fence_source_generation_update + BEFORE UPDATE OF current_generation ON _emdash_media_usage_sources + FOR EACH ROW + EXECUTE FUNCTION emdash_media_usage_fence_source_generation() + `) + .execute(db); + await sql + .raw(` + CREATE TRIGGER emdash_media_usage_record_cleanup_fence + BEFORE DELETE ON _emdash_media_usage + FOR EACH ROW + WHEN (OLD.cleanup_lease_token IS NOT NULL) + EXECUTE FUNCTION emdash_media_usage_record_cleanup_fence() + `) + .execute(db); + return; + } + + await sql + .raw(` + CREATE TRIGGER IF NOT EXISTS emdash_media_usage_fence_source_generation_insert + BEFORE INSERT ON _emdash_media_usage_sources + WHEN EXISTS ( + SELECT 1 + FROM _emdash_media_usage_cleanup_fence AS fence + WHERE fence.task_key = 'projection_gc' + AND NEW.current_generation <= fence.generation_floor + ) AND NOT EXISTS ( + SELECT 1 + FROM _emdash_media_usage_generation_writes AS writer + WHERE writer.source_key = NEW.source_key + AND writer.generation = NEW.current_generation + AND writer.expires_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + ) + BEGIN + SELECT RAISE(IGNORE); + END + `) + .execute(db); + await sql + .raw(` + CREATE TRIGGER IF NOT EXISTS emdash_media_usage_fence_source_generation_update + BEFORE UPDATE OF current_generation ON _emdash_media_usage_sources + WHEN NEW.current_generation <> OLD.current_generation + AND EXISTS ( + SELECT 1 + FROM _emdash_media_usage_cleanup_fence AS fence + WHERE fence.task_key = 'projection_gc' + AND NEW.current_generation <= fence.generation_floor + ) AND NOT EXISTS ( + SELECT 1 + FROM _emdash_media_usage_generation_writes AS writer + WHERE writer.source_key = NEW.source_key + AND writer.generation = NEW.current_generation + AND writer.expires_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + ) + BEGIN + SELECT RAISE(IGNORE); + END + `) + .execute(db); + await sql + .raw(` + CREATE TRIGGER IF NOT EXISTS emdash_media_usage_record_cleanup_fence + BEFORE DELETE ON _emdash_media_usage + WHEN OLD.cleanup_lease_token IS NOT NULL AND EXISTS ( + SELECT 1 + FROM _emdash_media_usage_cleanup AS cleanup + WHERE cleanup.task_key = 'projection_gc' + AND cleanup.lease_token = OLD.cleanup_lease_token + ) + BEGIN + INSERT INTO _emdash_media_usage_cleanup_fence ( + task_key, + generation_floor, + updated_at + ) + VALUES ('projection_gc', OLD.generation, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) + ON CONFLICT (task_key) DO UPDATE SET + generation_floor = CASE + WHEN excluded.generation_floor > _emdash_media_usage_cleanup_fence.generation_floor + THEN excluded.generation_floor + ELSE _emdash_media_usage_cleanup_fence.generation_floor + END, + updated_at = excluded.updated_at; + END + `) + .execute(db); +} + +export async function down(db: Kysely): Promise { + if (isPostgres(db)) { + await sql + .raw("DROP TRIGGER IF EXISTS emdash_media_usage_record_cleanup_fence ON _emdash_media_usage") + .execute(db); + await sql + .raw( + "DROP TRIGGER IF EXISTS emdash_media_usage_fence_source_generation_update ON _emdash_media_usage_sources", + ) + .execute(db); + await sql + .raw( + "DROP TRIGGER IF EXISTS emdash_media_usage_fence_source_generation_insert ON _emdash_media_usage_sources", + ) + .execute(db); + await sql + .raw( + "DROP TRIGGER IF EXISTS emdash_media_usage_lock_cleanup_update ON _emdash_media_usage_sources", + ) + .execute(db); + await sql + .raw( + "DROP TRIGGER IF EXISTS emdash_media_usage_lock_cleanup_insert ON _emdash_media_usage_sources", + ) + .execute(db); + await sql + .raw( + "DROP TRIGGER IF EXISTS emdash_media_usage_lock_cleanup_delete ON _emdash_media_usage_sources", + ) + .execute(db); + await sql.raw("DROP FUNCTION IF EXISTS emdash_media_usage_record_cleanup_fence()").execute(db); + await sql + .raw("DROP FUNCTION IF EXISTS emdash_media_usage_fence_source_generation()") + .execute(db); + await sql.raw("DROP FUNCTION IF EXISTS emdash_media_usage_lock_cleanup()").execute(db); + } else { + await sql.raw("DROP TRIGGER IF EXISTS emdash_media_usage_record_cleanup_fence").execute(db); + await sql + .raw("DROP TRIGGER IF EXISTS emdash_media_usage_fence_source_generation_update") + .execute(db); + await sql + .raw("DROP TRIGGER IF EXISTS emdash_media_usage_fence_source_generation_insert") + .execute(db); + } + await db.schema.dropTable("_emdash_media_usage_cleanup_fence").ifExists().execute(); + if (await columnExists(db, "_emdash_media_usage", "cleanup_lease_token")) { + await db.schema.alterTable("_emdash_media_usage").dropColumn("cleanup_lease_token").execute(); + } +} diff --git a/packages/core/src/database/migrations/runner.ts b/packages/core/src/database/migrations/runner.ts index 5c1c7764ff..e17dfbe8ff 100644 --- a/packages/core/src/database/migrations/runner.ts +++ b/packages/core/src/database/migrations/runner.ts @@ -57,6 +57,8 @@ import * as m051 from "./051_content_taxonomies_denorm.js"; import * as m052 from "./052_media_usage_read_index.js"; import * as m053 from "./053_plugin_mcp_tools.js"; import * as m054 from "./054_media_upload_attempts.js"; +import * as m055 from "./055_media_usage_cleanup.js"; +import * as m056 from "./056_media_usage_cleanup_fence.js"; const MIGRATIONS: Readonly> = Object.freeze({ "001_initial": m001, @@ -112,6 +114,8 @@ const MIGRATIONS: Readonly> = Object.freeze({ "052_media_usage_read_index": m052, "053_plugin_mcp_tools": m053, "054_media_upload_attempts": m054, + "055_media_usage_cleanup": m055, + "056_media_usage_cleanup_fence": m056, }); /** Total number of registered migrations. Exported for use in tests. */ diff --git a/packages/core/src/database/repositories/media-usage.ts b/packages/core/src/database/repositories/media-usage.ts index 3ebd0fff68..5f02e4b9ac 100644 --- a/packages/core/src/database/repositories/media-usage.ts +++ b/packages/core/src/database/repositories/media-usage.ts @@ -1,8 +1,8 @@ import { sql, type ExpressionBuilder, - type Insertable, type Kysely, + type RawBuilder, type Selectable, type Transaction, type Updateable, @@ -12,6 +12,7 @@ import { ulid } from "ulidx"; import type { MediaUsageContentSourceVariant } from "../../media/usage/source-key.js"; import type { MediaKind, MediaUsageReferenceType } from "../../media/usage/types.js"; import { chunks, SQL_BATCH_SIZE } from "../../utils/chunks.js"; +import { isPostgres } from "../dialect-helpers.js"; import { withTransaction } from "../transaction.js"; import type { Database, @@ -32,10 +33,27 @@ type MediaUsageSourceNullableStringColumn = | "last_error_code"; const OCCURRENCE_BIND_COLUMNS = 13; +export const MEDIA_USAGE_GENERATION_WRITE_LEASE_MS = 60 * 60 * 1000; const OCCURRENCE_INSERT_BATCH_SIZE = Math.max( 1, Math.floor(SQL_BATCH_SIZE / OCCURRENCE_BIND_COLUMNS), ); + +function cleanupDeleteBatchSize(cleanupLease: MediaUsageCleanupLease | undefined): number { + return cleanupLease ? SQL_BATCH_SIZE - 3 : SQL_BATCH_SIZE; +} + +function canIssueCleanupStatement(canIssueStatement: (() => boolean) | undefined): boolean { + return canIssueStatement?.() ?? true; +} + +function cleanupDurationSeconds(value: number): number { + if (!Number.isSafeInteger(value) || value < 0) { + throw new Error("Media usage cleanup duration must be a non-negative whole number of seconds"); + } + return value; +} + const CONTENT_SOURCE_ELIGIBILITY = sql`( s.source_variant = 'draft_overlay' OR ( @@ -137,6 +155,53 @@ export interface MediaUsageGuardedAttemptResult { source: MediaUsageSource | null; } +export interface MediaUsageCleanupCursor { + createdAt: string; + id: string; +} + +export interface MediaUsageCleanupClaim { + leaseToken: string; + cursor: MediaUsageCleanupCursor | null; + claimedAt: string; + scanBeforeAt: string; + consecutiveFailures: number; +} + +export interface MediaUsageCleanupCandidate { + id: string; + sourceKey: string; + generation: string; + createdAt: string; + currentGeneration: string | null; + indexedAt: string | null; + writeLeaseExpiresAt: string | null; +} + +export interface MediaUsageCleanupLease { + leaseToken: string; +} + +export interface MediaUsageCleanupDeleteOptions { + candidateIds?: readonly string[]; + cleanupLease?: MediaUsageCleanupLease; + canIssueStatement?: () => boolean; +} + +export interface MediaUsageCleanupCompletion { + leaseToken: string; + nextCursor: MediaUsageCleanupCursor | null; + sweepComplete: boolean; + candidateCount: number; + deletedOrphans: number; + deletedStale: number; + deletedAbandoned: number; + deletedWriteLeases: number; + backlogLowerBound: number; + scanHasMore: boolean; + durationMs: number; +} + export interface MediaUsageIndexStatusRepairInput extends MediaUsageIndexStatusIdentity { runToken: string; schemaVersion?: number; @@ -329,11 +394,15 @@ export class MediaUsageRepository { occurrences: readonly MediaUsageOccurrenceInput[], ): Promise { const generation = ulid(); - const now = new Date().toISOString(); - await withTransaction(this.db, async (trx) => { - await this.insertOccurrences(trx, source.sourceKey, generation, occurrences, now); - await this.upsertSource(trx, source, generation, now); + await this.withGenerationWriteLease(source.sourceKey, generation, async (leaseToken, now) => { + await withTransaction(this.db, async (trx) => { + await this.insertOccurrences(trx, source.sourceKey, generation, occurrences, now); + const promoted = await this.upsertSource(trx, source, generation, now, leaseToken); + if (!promoted) { + throw new Error(`Media usage generation lease expired for ${source.sourceKey}`); + } + }); }); const replaced = await this.findSource(source.sourceKey); @@ -348,17 +417,23 @@ export class MediaUsageRepository { expectedCurrentGeneration: string | null, ): Promise { const generation = ulid(); - const now = new Date().toISOString(); - const row = this.buildSourceRow(source, generation, now); let replaced = false; - await withTransaction(this.db, async (trx) => { - await this.insertOccurrences(trx, source.sourceKey, generation, occurrences, now); - if (expectedCurrentGeneration === null) { - replaced = await this.insertSourceIfAbsent(trx, row); - return; - } - replaced = await this.updateSourceIfGeneration(trx, row, expectedCurrentGeneration); + await this.withGenerationWriteLease(source.sourceKey, generation, async (leaseToken, now) => { + const row = this.buildSourceRow(source, generation, now); + await withTransaction(this.db, async (trx) => { + await this.insertOccurrences(trx, source.sourceKey, generation, occurrences, now); + if (expectedCurrentGeneration === null) { + replaced = await this.insertSourceIfAbsent(trx, row, leaseToken); + return; + } + replaced = await this.updateSourceIfGeneration( + trx, + row, + expectedCurrentGeneration, + leaseToken, + ); + }); }); return { @@ -403,17 +478,18 @@ export class MediaUsageRepository { expectedSource: MediaUsageSource | null, ): Promise { const generation = ulid(); - const now = new Date().toISOString(); - const row = this.buildSourceRow(source, generation, now); let replaced = false; - await withTransaction(this.db, async (trx) => { - await this.insertOccurrences(trx, source.sourceKey, generation, occurrences, now); - if (expectedSource === null) { - replaced = await this.insertSourceIfAbsent(trx, row); - return; - } - replaced = await this.updateSourceIfMatching(trx, row, expectedSource); + await this.withGenerationWriteLease(source.sourceKey, generation, async (leaseToken, now) => { + const row = this.buildSourceRow(source, generation, now); + await withTransaction(this.db, async (trx) => { + await this.insertOccurrences(trx, source.sourceKey, generation, occurrences, now); + if (expectedSource === null) { + replaced = await this.insertSourceIfAbsent(trx, row, leaseToken); + return; + } + replaced = await this.updateSourceIfMatching(trx, row, expectedSource, leaseToken); + }); }); return { @@ -423,15 +499,19 @@ export class MediaUsageRepository { } async markSourceAttempted(source: MediaUsageSourceInput): Promise { - const now = new Date().toISOString(); - const row = this.buildAttemptedSourceRow(source, now); - const updates = this.attemptedSourceUpdateSet(source, row); - - await this.db - .insertInto("_emdash_media_usage_sources") - .values(row) - .onConflict((oc) => oc.column("source_key").doUpdateSet(updates)) - .execute(); + const generation = ulid(); + await this.withGenerationWriteLease(source.sourceKey, generation, async (leaseToken, now) => { + const row = this.buildAttemptedSourceRow(source, generation, now); + const updates = this.attemptedSourceUpdateSet(source, row); + const result = await this.db + .insertInto("_emdash_media_usage_sources") + .values(row) + .onConflict((oc) => oc.column("source_key").doUpdateSet(updates)) + .executeTakeFirst(); + if ((result.numInsertedOrUpdatedRows ?? 0n) <= 0n) { + throw new Error(`Media usage generation lease expired for ${source.sourceKey}`); + } + }); const attempted = await this.findSource(source.sourceKey); if (!attempted) { @@ -444,13 +524,21 @@ export class MediaUsageRepository { source: MediaUsageSourceInput, expectedSource: MediaUsageSource | null, ): Promise { - const now = new Date().toISOString(); - const row = this.buildAttemptedSourceRow(source, now); + const generation = ulid(); let attempted = false; if (expectedSource === null) { - attempted = await this.insertSourceIfAbsent(this.db, row); + await this.withGenerationWriteLease(source.sourceKey, generation, async (leaseToken, now) => { + const row = this.buildAttemptedSourceRow(source, generation, now); + const result = await this.db + .insertInto("_emdash_media_usage_sources") + .values(row) + .onConflict((oc) => oc.column("source_key").doNothing()) + .executeTakeFirst(); + attempted = (result.numInsertedOrUpdatedRows ?? 0n) > 0n; + }); } else { + const row = this.buildAttemptedSourceRow(source, generation, new Date().toISOString()); attempted = await this.updateAttemptedSourceIfMatching(this.db, source, row, expectedSource); } @@ -692,6 +780,7 @@ export class MediaUsageRepository { ): Promise { let deleted = false; await withTransaction(this.db, async (trx) => { + await this.lockCleanupBeforeSourceDelete(trx); const result = await trx .deleteFrom("_emdash_media_usage_sources") .where("source_key", "=", sourceKey) @@ -714,6 +803,7 @@ export class MediaUsageRepository { ): Promise { let deleted = false; await withTransaction(this.db, async (trx) => { + await this.lockCleanupBeforeSourceDelete(trx); const result = await trx .deleteFrom("_emdash_media_usage_sources") .where("source_key", "=", sourceKey) @@ -744,6 +834,7 @@ export class MediaUsageRepository { const tableName = `ec_${collectionSlug}`; let deleted = false; await withTransaction(this.db, async (trx) => { + await this.lockCleanupBeforeSourceDelete(trx); const result = await trx .deleteFrom("_emdash_media_usage_sources") .where("source_key", "=", sourceKey) @@ -814,125 +905,345 @@ export class MediaUsageRepository { return rows.map((row) => rowToSource(row)); } - async deleteOrphanOccurrencesOlderThan(cutoff: string, limit: number): Promise { + async claimMediaUsageCleanup(input: { + leaseToken: string; + leaseDurationSeconds: number; + nextEligibleDelaySeconds: number; + sweepSafetyWindowSeconds: number; + }): Promise { + const leaseDurationSeconds = cleanupDurationSeconds(input.leaseDurationSeconds); + const nextEligibleDelaySeconds = cleanupDurationSeconds(input.nextEligibleDelaySeconds); + const sweepSafetyWindowSeconds = cleanupDurationSeconds(input.sweepSafetyWindowSeconds); + const claimedAt = this.cleanupTimestampOffset(0); + const leaseExpiresAt = this.cleanupTimestampOffset(leaseDurationSeconds); + const nextEligibleAt = this.cleanupTimestampOffset(nextEligibleDelaySeconds); + const sweepBeforeAt = this.cleanupTimestampOffset(-sweepSafetyWindowSeconds); + const row = await this.db + .updateTable("_emdash_media_usage_cleanup") + .set({ + lease_token: input.leaseToken, + lease_expires_at: leaseExpiresAt, + next_eligible_at: nextEligibleAt, + last_started_at: claimedAt, + updated_at: claimedAt, + scan_before_at: sql`CASE + WHEN scan_before_at IS NULL THEN ${sweepBeforeAt} + ELSE scan_before_at + END`, + }) + .where("task_key", "=", "projection_gc") + .where(this.cleanupTimestampIsDue("next_eligible_at")) + .where((eb) => + eb.or([eb("lease_token", "is", null), this.cleanupTimestampIsDue("lease_expires_at")]), + ) + .returning([ + "cursor_created_at", + "cursor_id", + "last_started_at", + "scan_before_at", + "consecutive_failures", + ]) + .executeTakeFirst(); + if (!row) return null; + if (!row.last_started_at || !row.scan_before_at) { + throw new Error("Media usage cleanup claim did not persist its database timestamps"); + } + return { + leaseToken: input.leaseToken, + cursor: + row.cursor_created_at && row.cursor_id + ? { createdAt: row.cursor_created_at, id: row.cursor_id } + : null, + claimedAt: row.last_started_at, + scanBeforeAt: row.scan_before_at, + consecutiveFailures: row.consecutive_failures, + }; + } + + async findMediaUsageCleanupCandidates(input: { + cutoff: string; + cursor: MediaUsageCleanupCursor | null; + limit: number; + cleanupLease?: MediaUsageCleanupLease; + }): Promise { + let query = this.db + .selectFrom("_emdash_media_usage as u") + .leftJoin("_emdash_media_usage_sources as s", "s.source_key", "u.source_key") + .leftJoin("_emdash_media_usage_generation_writes as writer", (join) => + join + .onRef("writer.source_key", "=", "u.source_key") + .onRef("writer.generation", "=", "u.generation"), + ) + .select([ + "u.id as id", + "u.source_key as source_key", + "u.generation as generation", + "u.created_at as created_at", + "s.current_generation as current_generation", + "s.indexed_at as indexed_at", + "writer.expires_at as write_lease_expires_at", + ]) + .where("u.created_at", "<", input.cutoff) + .orderBy("u.created_at", "asc") + .orderBy("u.id", "asc") + .limit(Math.max(0, Math.floor(input.limit))); + if (input.cleanupLease) { + query = query.where(this.activeCleanupLeaseExpression(input.cleanupLease)); + } + + if (input.cursor) { + query = query.where((eb) => + eb.or([ + eb("u.created_at", ">", input.cursor!.createdAt), + eb.and([ + eb("u.created_at", "=", input.cursor!.createdAt), + eb("u.id", ">", input.cursor!.id), + ]), + ]), + ); + } + + const rows = await query.execute(); + return rows.map((row) => ({ + id: row.id, + sourceKey: row.source_key, + generation: row.generation, + createdAt: row.created_at, + currentGeneration: row.current_generation, + indexedAt: row.indexed_at, + writeLeaseExpiresAt: row.write_lease_expires_at, + })); + } + + async completeMediaUsageCleanup(input: MediaUsageCleanupCompletion): Promise { + const updates = { + lease_token: null, + lease_expires_at: null, + cursor_created_at: input.sweepComplete ? null : (input.nextCursor?.createdAt ?? null), + cursor_id: input.sweepComplete ? null : (input.nextCursor?.id ?? null), + ...(input.sweepComplete ? { scan_before_at: null } : {}), + consecutive_failures: 0, + last_completed_at: this.cleanupTimestampOffset(0), + last_candidate_count: input.candidateCount, + last_deleted_orphans: input.deletedOrphans, + last_deleted_stale: input.deletedStale, + last_deleted_abandoned: input.deletedAbandoned, + last_deleted_write_leases: input.deletedWriteLeases, + last_backlog_lower_bound: input.backlogLowerBound, + last_scan_has_more: input.scanHasMore ? 1 : 0, + last_duration_ms: input.durationMs, + last_error_code: null, + updated_at: this.cleanupTimestampOffset(0), + }; + const result = await this.db + .updateTable("_emdash_media_usage_cleanup") + .set(updates) + .where("task_key", "=", "projection_gc") + .where("lease_token", "=", input.leaseToken) + .where(this.cleanupLeaseExpiryIsInFuture("_emdash_media_usage_cleanup.lease_expires_at")) + .executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) > 0; + } + + async failMediaUsageCleanup(input: { + leaseToken: string; + retryDelaySeconds: number; + consecutiveFailures: number; + durationMs: number; + errorCode: string; + }): Promise { + const retryDelaySeconds = cleanupDurationSeconds(input.retryDelaySeconds); + const result = await this.db + .updateTable("_emdash_media_usage_cleanup") + .set({ + lease_token: null, + lease_expires_at: null, + next_eligible_at: this.cleanupTimestampOffset(retryDelaySeconds), + consecutive_failures: input.consecutiveFailures, + last_completed_at: this.cleanupTimestampOffset(0), + last_duration_ms: input.durationMs, + last_error_code: input.errorCode, + updated_at: this.cleanupTimestampOffset(0), + }) + .where("task_key", "=", "projection_gc") + .where("lease_token", "=", input.leaseToken) + .where(this.cleanupLeaseExpiryIsInFuture("_emdash_media_usage_cleanup.lease_expires_at")) + .executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) > 0; + } + + async deleteOrphanOccurrencesOlderThan( + cutoff: string, + limit: number, + options: MediaUsageCleanupDeleteOptions = {}, + ): Promise { const batchLimit = Math.floor(limit); if (batchLimit <= 0) return 0; + if (options.candidateIds) { + return this.deleteOrphanCandidateIds( + options.candidateIds.slice(0, batchLimit), + cutoff, + options.cleanupLease, + options.canIssueStatement, + ); + } + if (!canIssueCleanupStatement(options.canIssueStatement)) return 0; - const rows = await this.db + let query = this.db .selectFrom("_emdash_media_usage as u") .leftJoin("_emdash_media_usage_sources as s", (join) => join.onRef("s.source_key", "=", "u.source_key"), ) + .leftJoin("_emdash_media_usage_generation_writes as writer", (join) => + join + .onRef("writer.source_key", "=", "u.source_key") + .onRef("writer.generation", "=", "u.generation"), + ) .select("u.id") .where("s.source_key", "is", null) .where("u.created_at", "<", cutoff) + .where(this.noActiveGenerationWriteExpression("u")) .orderBy("u.created_at", "asc") .orderBy("u.id", "asc") - .limit(batchLimit) - .execute(); + .limit(batchLimit); + if (options.cleanupLease) { + query = query.where(this.activeCleanupLeaseExpression(options.cleanupLease)); + } + const rows = await query.execute(); - let deleted = 0; - for (const idBatch of chunks( + return this.deleteOrphanCandidateIds( rows.map((row) => row.id), - SQL_BATCH_SIZE, - )) { - const result = await this.db - .deleteFrom("_emdash_media_usage") - .where("id", "in", idBatch) - .where("created_at", "<", cutoff) - .where( - sql`NOT EXISTS (SELECT 1 FROM _emdash_media_usage_sources s WHERE s.source_key = _emdash_media_usage.source_key)`, - ) - .executeTakeFirst(); - deleted += Number(result.numDeletedRows ?? 0); - } - return deleted; + cutoff, + options.cleanupLease, + options.canIssueStatement, + ); } - async deleteStaleGenerationsOlderThan(cutoff: string, limit: number): Promise { + async deleteStaleGenerationsOlderThan( + cutoff: string, + limit: number, + options: MediaUsageCleanupDeleteOptions = {}, + ): Promise { const batchLimit = Math.floor(limit); if (batchLimit <= 0) return 0; + if (options.candidateIds) { + return this.deleteStaleCandidateIds( + options.candidateIds.slice(0, batchLimit), + cutoff, + options.cleanupLease, + options.canIssueStatement, + ); + } + if (!canIssueCleanupStatement(options.canIssueStatement)) return 0; - const rows = await this.db + let query = this.db .selectFrom("_emdash_media_usage as u") .innerJoin("_emdash_media_usage_sources as s", (join) => join.onRef("s.source_key", "=", "u.source_key"), ) + .leftJoin("_emdash_media_usage_generation_writes as writer", (join) => + join + .onRef("writer.source_key", "=", "u.source_key") + .onRef("writer.generation", "=", "u.generation"), + ) .select("u.id") .where("u.created_at", "<", cutoff) .whereRef("u.generation", "!=", "s.current_generation") .whereRef("u.created_at", "<", "s.indexed_at") + .where(this.noActiveGenerationWriteExpression("u")) .orderBy("u.created_at", "asc") .orderBy("u.id", "asc") - .limit(batchLimit) - .execute(); - - const ids = rows.map((row) => row.id); - if (ids.length === 0) return 0; - - let deleted = 0; - for (const idBatch of chunks(ids, SQL_BATCH_SIZE)) { - const result = await this.db - .deleteFrom("_emdash_media_usage") - .where("id", "in", idBatch) - .where("created_at", "<", cutoff) - .where((eb) => - eb.exists( - eb - .selectFrom("_emdash_media_usage_sources as s") - .select("s.source_key") - .whereRef("s.source_key", "=", "_emdash_media_usage.source_key") - .whereRef("s.current_generation", "!=", "_emdash_media_usage.generation") - .whereRef("_emdash_media_usage.created_at", "<", "s.indexed_at"), - ), - ) - .executeTakeFirst(); - deleted += Number(result.numDeletedRows ?? 0); + .limit(batchLimit); + if (options.cleanupLease) { + query = query.where(this.activeCleanupLeaseExpression(options.cleanupLease)); } - return deleted; + const rows = await query.execute(); + + return this.deleteStaleCandidateIds( + rows.map((row) => row.id), + cutoff, + options.cleanupLease, + options.canIssueStatement, + ); } - async deleteAbandonedGenerationsOlderThan(cutoff: string, limit: number): Promise { + async deleteAbandonedGenerationsOlderThan( + cutoff: string, + limit: number, + options: MediaUsageCleanupDeleteOptions = {}, + ): Promise { const batchLimit = Math.floor(limit); if (batchLimit <= 0) return 0; + if (options.candidateIds) { + return this.deleteAbandonedCandidateIds( + options.candidateIds.slice(0, batchLimit), + cutoff, + options.cleanupLease, + options.canIssueStatement, + ); + } + if (!canIssueCleanupStatement(options.canIssueStatement)) return 0; - const rows = await this.db + let query = this.db .selectFrom("_emdash_media_usage as u") .innerJoin("_emdash_media_usage_sources as s", (join) => join.onRef("s.source_key", "=", "u.source_key"), ) + .leftJoin("_emdash_media_usage_generation_writes as writer", (join) => + join + .onRef("writer.source_key", "=", "u.source_key") + .onRef("writer.generation", "=", "u.generation"), + ) .select("u.id") .where("u.created_at", "<", cutoff) .whereRef("u.generation", "!=", "s.current_generation") .whereRef("u.created_at", ">=", "s.indexed_at") + .where(this.noActiveGenerationWriteExpression("u")) .orderBy("u.created_at", "asc") .orderBy("u.id", "asc") - .limit(batchLimit) - .execute(); + .limit(batchLimit); + if (options.cleanupLease) { + query = query.where(this.activeCleanupLeaseExpression(options.cleanupLease)); + } + const rows = await query.execute(); - let deleted = 0; - for (const idBatch of chunks( + return this.deleteAbandonedCandidateIds( rows.map((row) => row.id), - SQL_BATCH_SIZE, - )) { - const result = await this.db - .deleteFrom("_emdash_media_usage") - .where("id", "in", idBatch) - .where("created_at", "<", cutoff) - .where((eb) => - eb.exists( - eb - .selectFrom("_emdash_media_usage_sources as s") - .select("s.source_key") - .whereRef("s.source_key", "=", "_emdash_media_usage.source_key") - .whereRef("s.current_generation", "!=", "_emdash_media_usage.generation") - .whereRef("_emdash_media_usage.created_at", ">=", "s.indexed_at"), - ), - ) - .executeTakeFirst(); - deleted += Number(result.numDeletedRows ?? 0); - } - return deleted; + cutoff, + options.cleanupLease, + options.canIssueStatement, + ); + } + + async deleteExpiredGenerationWriteLeases( + limit: number, + cleanupLease?: MediaUsageCleanupLease, + canIssueStatement?: () => boolean, + ): Promise { + const batchLimit = Math.floor(limit); + if (batchLimit <= 0 || !canIssueCleanupStatement(canIssueStatement)) return 0; + let query = this.db + .selectFrom("_emdash_media_usage_generation_writes") + .select("lease_token") + .where(this.generationWriteLeaseHasExpired("expires_at")) + .orderBy("expires_at", "asc") + .orderBy("lease_token", "asc") + .limit(batchLimit); + if (cleanupLease) query = query.where(this.activeCleanupLeaseExpression(cleanupLease)); + const rows = await query.execute(); + if (rows.length === 0 || !canIssueCleanupStatement(canIssueStatement)) return 0; + let deleteQuery = this.db + .deleteFrom("_emdash_media_usage_generation_writes") + .where( + "lease_token", + "in", + rows.map((row) => row.lease_token), + ) + .where(this.generationWriteLeaseHasExpired("expires_at")); + if (cleanupLease) + deleteQuery = deleteQuery.where(this.activeCleanupLeaseExpression(cleanupLease)); + const result = await deleteQuery.executeTakeFirst(); + return Number(result.numDeletedRows ?? 0); } async upsertIndexStatus(input: MediaUsageIndexStatusInput): Promise { @@ -1106,11 +1417,249 @@ export class MediaUsageRepository { .where(CONTENT_SOURCE_ELIGIBILITY); } + private async deleteOrphanCandidateIds( + ids: readonly string[], + cutoff: string, + cleanupLease?: MediaUsageCleanupLease, + canIssueStatement?: () => boolean, + ): Promise { + let deleted = 0; + for (const idBatch of chunks([...ids], cleanupDeleteBatchSize(cleanupLease))) { + if (!canIssueCleanupStatement(canIssueStatement)) break; + if (cleanupLease) { + await this.markOrphanCandidatesForCleanup(idBatch, cutoff, cleanupLease); + if (!canIssueCleanupStatement(canIssueStatement)) break; + } + let query = this.db + .deleteFrom("_emdash_media_usage") + .where("id", "in", idBatch) + .where("created_at", "<", cutoff) + .where( + sql`NOT EXISTS (SELECT 1 FROM _emdash_media_usage_sources source WHERE source.source_key = _emdash_media_usage.source_key)`, + ) + .where(this.noActiveGenerationWriteExpression()); + if (cleanupLease) { + query = query + .where("cleanup_lease_token", "=", cleanupLease.leaseToken) + .where(this.activeCleanupLeaseExpression(cleanupLease)); + } + const result = await query.executeTakeFirst(); + deleted += Number(result.numDeletedRows ?? 0); + } + return deleted; + } + + private async deleteStaleCandidateIds( + ids: readonly string[], + cutoff: string, + cleanupLease?: MediaUsageCleanupLease, + canIssueStatement?: () => boolean, + ): Promise { + let deleted = 0; + for (const idBatch of chunks([...ids], cleanupDeleteBatchSize(cleanupLease))) { + if (!canIssueCleanupStatement(canIssueStatement)) break; + if (cleanupLease) { + await this.markStaleCandidatesForCleanup(idBatch, cutoff, cleanupLease); + if (!canIssueCleanupStatement(canIssueStatement)) break; + } + let query = this.db + .deleteFrom("_emdash_media_usage") + .where("id", "in", idBatch) + .where("created_at", "<", cutoff) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_sources as source") + .select("source.source_key") + .whereRef("source.source_key", "=", "_emdash_media_usage.source_key") + .whereRef("source.current_generation", "!=", "_emdash_media_usage.generation") + .whereRef("_emdash_media_usage.created_at", "<", "source.indexed_at"), + ), + ) + .where(this.noActiveGenerationWriteExpression()); + if (cleanupLease) { + query = query + .where("cleanup_lease_token", "=", cleanupLease.leaseToken) + .where(this.activeCleanupLeaseExpression(cleanupLease)); + } + const result = await query.executeTakeFirst(); + deleted += Number(result.numDeletedRows ?? 0); + } + return deleted; + } + + private async deleteAbandonedCandidateIds( + ids: readonly string[], + cutoff: string, + cleanupLease?: MediaUsageCleanupLease, + canIssueStatement?: () => boolean, + ): Promise { + let deleted = 0; + for (const idBatch of chunks([...ids], cleanupDeleteBatchSize(cleanupLease))) { + if (!canIssueCleanupStatement(canIssueStatement)) break; + if (cleanupLease) { + await this.markAbandonedCandidatesForCleanup(idBatch, cutoff, cleanupLease); + if (!canIssueCleanupStatement(canIssueStatement)) break; + } + let query = this.db + .deleteFrom("_emdash_media_usage") + .where("id", "in", idBatch) + .where("created_at", "<", cutoff) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_sources as source") + .select("source.source_key") + .whereRef("source.source_key", "=", "_emdash_media_usage.source_key") + .whereRef("source.current_generation", "!=", "_emdash_media_usage.generation") + .whereRef("_emdash_media_usage.created_at", ">=", "source.indexed_at"), + ), + ) + .where(this.noActiveGenerationWriteExpression()); + if (cleanupLease) { + query = query + .where("cleanup_lease_token", "=", cleanupLease.leaseToken) + .where(this.activeCleanupLeaseExpression(cleanupLease)); + } + const result = await query.executeTakeFirst(); + deleted += Number(result.numDeletedRows ?? 0); + } + return deleted; + } + + private async markOrphanCandidatesForCleanup( + ids: readonly string[], + cutoff: string, + cleanupLease: MediaUsageCleanupLease, + ): Promise { + await this.db + .updateTable("_emdash_media_usage") + .set({ cleanup_lease_token: cleanupLease.leaseToken }) + .where("id", "in", ids) + .where("created_at", "<", cutoff) + .where( + sql`NOT EXISTS (SELECT 1 FROM _emdash_media_usage_sources source WHERE source.source_key = _emdash_media_usage.source_key)`, + ) + .where(this.noActiveGenerationWriteExpression()) + .where(this.activeCleanupLeaseExpression(cleanupLease)) + .execute(); + } + + private async markStaleCandidatesForCleanup( + ids: readonly string[], + cutoff: string, + cleanupLease: MediaUsageCleanupLease, + ): Promise { + await this.db + .updateTable("_emdash_media_usage") + .set({ cleanup_lease_token: cleanupLease.leaseToken }) + .where("id", "in", ids) + .where("created_at", "<", cutoff) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_sources as source") + .select("source.source_key") + .whereRef("source.source_key", "=", "_emdash_media_usage.source_key") + .whereRef("source.current_generation", "!=", "_emdash_media_usage.generation") + .whereRef("_emdash_media_usage.created_at", "<", "source.indexed_at"), + ), + ) + .where(this.noActiveGenerationWriteExpression()) + .where(this.activeCleanupLeaseExpression(cleanupLease)) + .execute(); + } + + private async markAbandonedCandidatesForCleanup( + ids: readonly string[], + cutoff: string, + cleanupLease: MediaUsageCleanupLease, + ): Promise { + await this.db + .updateTable("_emdash_media_usage") + .set({ cleanup_lease_token: cleanupLease.leaseToken }) + .where("id", "in", ids) + .where("created_at", "<", cutoff) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_sources as source") + .select("source.source_key") + .whereRef("source.source_key", "=", "_emdash_media_usage.source_key") + .whereRef("source.current_generation", "!=", "_emdash_media_usage.generation") + .whereRef("_emdash_media_usage.created_at", ">=", "source.indexed_at"), + ), + ) + .where(this.noActiveGenerationWriteExpression()) + .where(this.activeCleanupLeaseExpression(cleanupLease)) + .execute(); + } + + private noActiveGenerationWriteExpression(usageTable = "_emdash_media_usage") { + const sourceKey = sql.ref(`${usageTable}.source_key`); + const generation = sql.ref(`${usageTable}.generation`); + return sql`NOT EXISTS ( + SELECT 1 + FROM _emdash_media_usage_generation_writes AS writer + WHERE writer.source_key = ${sourceKey} + AND writer.generation = ${generation} + AND ${this.generationWriteLeaseExpiryIsInFuture("writer.expires_at")} + )`; + } + + private activeCleanupLeaseExpression(cleanupLease: MediaUsageCleanupLease) { + const rowLock = isPostgres(this.db) ? sql` FOR UPDATE` : sql``; + return sql`EXISTS ( + SELECT 1 + FROM _emdash_media_usage_cleanup AS cleanup + WHERE cleanup.task_key = 'projection_gc' + AND cleanup.lease_token = ${cleanupLease.leaseToken} + AND ${this.cleanupLeaseExpiryIsInFuture("cleanup.lease_expires_at")} + ${rowLock} + )`; + } + + private cleanupLeaseExpiryIsInFuture(column: string) { + const leaseExpiresAt = sql.ref(column); + return isPostgres(this.db) + ? sql`${leaseExpiresAt}::timestamptz > clock_timestamp()` + : sql`${leaseExpiresAt} > strftime('%Y-%m-%dT%H:%M:%fZ', 'now')`; + } + + private cleanupTimestampIsDue(column: string) { + const timestamp = sql.ref(column); + return isPostgres(this.db) + ? sql`${timestamp}::timestamptz <= clock_timestamp()` + : sql`${timestamp} <= strftime('%Y-%m-%dT%H:%M:%fZ', 'now')`; + } + + private cleanupTimestampOffset(offsetSeconds: number): RawBuilder { + if (isPostgres(this.db)) { + return sql`to_char( + (clock_timestamp() AT TIME ZONE 'UTC') + (${offsetSeconds} * INTERVAL '1 second'), + 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"' + )`; + } + return sql`strftime( + '%Y-%m-%dT%H:%M:%fZ', + 'now', + ${`${offsetSeconds >= 0 ? "+" : ""}${offsetSeconds} seconds`} + )`; + } + + private generationWriteLeaseHasExpired(column: string) { + const leaseExpiresAt = sql.ref(column); + return isPostgres(this.db) + ? sql`${leaseExpiresAt}::timestamptz <= clock_timestamp()` + : sql`${leaseExpiresAt} <= strftime('%Y-%m-%dT%H:%M:%fZ', 'now')`; + } + private async deleteSourceKeys(sourceKeys: readonly string[]): Promise { const uniqueSourceKeys = [...new Set(sourceKeys)]; if (uniqueSourceKeys.length === 0) return 0; return withTransaction(this.db, async (trx) => { + await this.lockCleanupBeforeSourceDelete(trx); let deleted = 0; for (const sourceKeyBatch of chunks(uniqueSourceKeys, SQL_BATCH_SIZE)) { const result = await trx @@ -1119,6 +1668,11 @@ export class MediaUsageRepository { .executeTakeFirst(); deleted += Number(result.numDeletedRows ?? 0); + await trx + .updateTable("_emdash_media_usage") + .set({ cleanup_lease_token: null }) + .where("source_key", "in", sourceKeyBatch) + .execute(); await trx .deleteFrom("_emdash_media_usage") .where("source_key", "in", sourceKeyBatch) @@ -1133,8 +1687,12 @@ export class MediaUsageRepository { sourceKey: string, generation: string, ): Promise { - // Guarded source deletes remove only the generation that won the source CAS; - // unmatched generations become invisible orphans reclaimed by age-gated cleanup. + await db + .updateTable("_emdash_media_usage") + .set({ cleanup_lease_token: null }) + .where("source_key", "=", sourceKey) + .where("generation", "=", generation) + .execute(); await db .deleteFrom("_emdash_media_usage") .where("source_key", "=", sourceKey) @@ -1142,6 +1700,16 @@ export class MediaUsageRepository { .execute(); } + private async lockCleanupBeforeSourceDelete(db: DatabaseExecutor): Promise { + if (!isPostgres(this.db)) return; + await sql` + SELECT 1 + FROM _emdash_media_usage_cleanup + WHERE task_key = 'projection_gc' + FOR SHARE + `.execute(db); + } + private async insertOccurrences( db: DatabaseExecutor, sourceKey: string, @@ -1177,38 +1745,212 @@ export class MediaUsageRepository { source: MediaUsageSourceInput, generation: string, now: string, - ): Promise { + leaseToken: string, + ): Promise { const row = this.buildSourceRow(source, generation, now); - - await db - .insertInto("_emdash_media_usage_sources") - .values(row) - .onConflict((oc) => oc.column("source_key").doUpdateSet(this.sourceUpdateSet(row))) - .execute(); + return this.persistSourceIfWriteLease( + db, + row, + leaseToken, + sql` + ON CONFLICT (source_key) DO UPDATE SET + source_type = excluded.source_type, + collection_slug = excluded.collection_slug, + content_id = excluded.content_id, + source_variant = excluded.source_variant, + locale = excluded.locale, + translation_group = excluded.translation_group, + content_slug = excluded.content_slug, + content_title = excluded.content_title, + content_status = excluded.content_status, + content_scheduled_at = excluded.content_scheduled_at, + content_deleted_at = excluded.content_deleted_at, + revision_id = excluded.revision_id, + current_generation = excluded.current_generation, + schema_version = excluded.schema_version, + source_updated_at = excluded.source_updated_at, + source_version = excluded.source_version, + source_fingerprint = excluded.source_fingerprint, + source_completeness = excluded.source_completeness, + last_attempted_at = excluded.last_attempted_at, + last_error_code = excluded.last_error_code, + indexed_at = excluded.indexed_at, + updated_at = excluded.updated_at + `, + ); } private async insertSourceIfAbsent( db: DatabaseExecutor, - row: Insertable, + row: ReturnType, + leaseToken: string, ): Promise { - const result = await db - .insertInto("_emdash_media_usage_sources") - .values(row) - .onConflict((oc) => oc.column("source_key").doNothing()) - .executeTakeFirst(); - return (result.numInsertedOrUpdatedRows ?? 0n) > 0n; + return this.persistSourceIfWriteLease( + db, + row, + leaseToken, + sql`ON CONFLICT (source_key) DO NOTHING`, + ); + } + + private async persistSourceIfWriteLease( + db: DatabaseExecutor, + row: ReturnType, + leaseToken: string, + conflict: RawBuilder, + ): Promise { + const result = await sql` + INSERT INTO _emdash_media_usage_sources ( + source_key, + source_type, + collection_slug, + content_id, + source_variant, + locale, + translation_group, + content_slug, + content_title, + content_status, + content_scheduled_at, + content_deleted_at, + revision_id, + current_generation, + schema_version, + source_updated_at, + source_version, + source_fingerprint, + source_completeness, + last_attempted_at, + last_error_code, + indexed_at, + updated_at + ) + SELECT + ${row.source_key}, + ${row.source_type}, + ${row.collection_slug}, + ${row.content_id}, + ${row.source_variant}, + ${row.locale}, + ${row.translation_group}, + ${row.content_slug}, + ${row.content_title}, + ${row.content_status}, + ${row.content_scheduled_at}, + ${row.content_deleted_at}, + ${row.revision_id}, + ${row.current_generation}, + ${row.schema_version}, + ${row.source_updated_at}, + ${row.source_version}, + ${row.source_fingerprint}, + ${row.source_completeness}, + ${row.last_attempted_at}, + ${row.last_error_code}, + ${row.indexed_at}, + ${row.updated_at} + WHERE EXISTS ( + SELECT 1 + FROM _emdash_media_usage_generation_writes + WHERE source_key = ${row.source_key} + AND generation = ${row.current_generation} + AND lease_token = ${leaseToken} + AND ${this.generationWriteLeaseExpiryIsInFuture("expires_at")} + ) + ${conflict} + `.execute(db); + return Number(result.numAffectedRows ?? 0) > 0; + } + + private generationWriteLeaseExpression( + row: ReturnType, + leaseToken: string, + ) { + return (eb: ExpressionBuilder) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_generation_writes") + .select("source_key") + .where("source_key", "=", row.source_key) + .where("generation", "=", row.current_generation) + .where("lease_token", "=", leaseToken) + .where( + this.generationWriteLeaseExpiryIsInFuture( + "_emdash_media_usage_generation_writes.expires_at", + ), + ), + ); + } + + private generationWriteLeaseExpiryIsInFuture(column: string) { + const leaseExpiresAt = sql.ref(column); + return isPostgres(this.db) + ? sql`${leaseExpiresAt}::timestamptz > clock_timestamp()` + : sql`${leaseExpiresAt} > strftime('%Y-%m-%dT%H:%M:%fZ', 'now')`; + } + + private async withGenerationWriteLease( + sourceKey: string, + generation: string, + write: (leaseToken: string, startedAt: string) => Promise, + ): Promise { + const leaseToken = ulid(); + const lease = await this.db + .insertInto("_emdash_media_usage_generation_writes") + .values({ + source_key: sourceKey, + generation, + lease_token: leaseToken, + expires_at: this.generationWriteLeaseTimestampOffset( + MEDIA_USAGE_GENERATION_WRITE_LEASE_MS / 1000, + ), + created_at: this.generationWriteLeaseTimestampOffset(0), + }) + .returning("created_at") + .executeTakeFirstOrThrow(); + + try { + return await write(leaseToken, lease.created_at); + } finally { + try { + await this.db + .deleteFrom("_emdash_media_usage_generation_writes") + .where("source_key", "=", sourceKey) + .where("generation", "=", generation) + .where("lease_token", "=", leaseToken) + .execute(); + } catch (error) { + console.error("[media-usage] Failed to release generation write lease:", error); + } + } + } + + private generationWriteLeaseTimestampOffset(offsetSeconds: number): RawBuilder { + if (isPostgres(this.db)) { + return sql`to_char( + (clock_timestamp() AT TIME ZONE 'UTC') + (${offsetSeconds} * INTERVAL '1 second'), + 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"' + )`; + } + return sql`strftime( + '%Y-%m-%dT%H:%M:%fZ', + 'now', + ${`${offsetSeconds >= 0 ? "+" : ""}${offsetSeconds} seconds`} + )`; } private async updateSourceIfGeneration( db: DatabaseExecutor, row: ReturnType, expectedCurrentGeneration: string, + leaseToken: string, ): Promise { const result = await db .updateTable("_emdash_media_usage_sources") .set(this.sourceUpdateSet(row)) .where("source_key", "=", row.source_key) .where("current_generation", "=", expectedCurrentGeneration) + .where(this.generationWriteLeaseExpression(row, leaseToken)) .executeTakeFirst(); return Number(result.numUpdatedRows ?? 0) > 0; } @@ -1217,12 +1959,14 @@ export class MediaUsageRepository { db: DatabaseExecutor, row: ReturnType, expectedSource: MediaUsageSource, + leaseToken: string, ): Promise { const result = await db .updateTable("_emdash_media_usage_sources") .set(this.sourceUpdateSet(row)) .where("source_key", "=", row.source_key) .where(this.sourceMatchExpression(expectedSource)) + .where(this.generationWriteLeaseExpression(row, leaseToken)) .executeTakeFirst(); return Number(result.numUpdatedRows ?? 0) > 0; } @@ -1313,7 +2057,7 @@ export class MediaUsageRepository { }; } - private buildAttemptedSourceRow(source: MediaUsageSourceInput, now: string) { + private buildAttemptedSourceRow(source: MediaUsageSourceInput, generation: string, now: string) { return { source_key: source.sourceKey, source_type: source.sourceType, @@ -1328,7 +2072,7 @@ export class MediaUsageRepository { content_scheduled_at: source.contentScheduledAt ?? null, content_deleted_at: source.contentDeletedAt ?? null, revision_id: source.revisionId ?? null, - current_generation: ulid(), + current_generation: generation, schema_version: source.schemaVersion ?? 1, source_updated_at: source.sourceUpdatedAt ?? null, source_version: source.sourceVersion ?? null, @@ -1567,6 +2311,7 @@ function rowToUsageRecord(row: JoinedUsageRow): MediaUsageRecord { media_kind: row.media_kind, mime_type: row.mime_type, created_at: row.occurrence_created_at, + cleanup_lease_token: null, }), }; } diff --git a/packages/core/src/database/types.ts b/packages/core/src/database/types.ts index 0b00056fc9..d0c5cf1921 100644 --- a/packages/core/src/database/types.ts +++ b/packages/core/src/database/types.ts @@ -131,6 +131,44 @@ export interface MediaUsageTable { media_kind: string | null; mime_type: string | null; created_at: Generated; + cleanup_lease_token: Generated; +} + +export interface MediaUsageCleanupTable { + task_key: string; + lease_token: string | null; + lease_expires_at: string | null; + next_eligible_at: string; + cursor_created_at: string | null; + cursor_id: string | null; + scan_before_at: string | null; + consecutive_failures: Generated; + last_started_at: string | null; + last_completed_at: string | null; + last_candidate_count: Generated; + last_deleted_orphans: Generated; + last_deleted_stale: Generated; + last_deleted_abandoned: Generated; + last_deleted_write_leases: Generated; + last_backlog_lower_bound: Generated; + last_scan_has_more: Generated; + last_duration_ms: Generated; + last_error_code: string | null; + updated_at: Generated; +} + +export interface MediaUsageGenerationWriteTable { + source_key: string; + generation: string; + lease_token: string; + expires_at: string; + created_at: Generated; +} + +export interface MediaUsageGenerationFenceTable { + task_key: string; + generation_floor: string; + updated_at: Generated; } export interface MediaUsageIndexStatusTable { @@ -507,6 +545,9 @@ export interface Database { _emdash_media_upload_attempts: MediaUploadAttemptTable; _emdash_media_usage_sources: MediaUsageSourceTable; _emdash_media_usage: MediaUsageTable; + _emdash_media_usage_cleanup: MediaUsageCleanupTable; + _emdash_media_usage_generation_writes: MediaUsageGenerationWriteTable; + _emdash_media_usage_cleanup_fence: MediaUsageGenerationFenceTable; _emdash_media_usage_index_status: MediaUsageIndexStatusTable; users: UserTable; credentials: CredentialTable; diff --git a/packages/core/src/media/usage/cleanup.ts b/packages/core/src/media/usage/cleanup.ts new file mode 100644 index 0000000000..11eea44c87 --- /dev/null +++ b/packages/core/src/media/usage/cleanup.ts @@ -0,0 +1,305 @@ +import type { Kysely } from "kysely"; +import { ulid } from "ulidx"; + +import { + MediaUsageRepository, + type MediaUsageCleanupCandidate, + type MediaUsageCleanupCursor, +} from "../../database/repositories/media-usage.js"; +import type { Database } from "../../database/types.js"; + +export const MEDIA_USAGE_CLEANUP_CANDIDATE_LIMIT = 250; +export const MEDIA_USAGE_CLEANUP_DELETE_LIMIT = 50; +export const MEDIA_USAGE_CLEANUP_WRITE_LEASE_DELETE_LIMIT = 49; +export const MEDIA_USAGE_CLEANUP_INTERVAL_MS = 60 * 1000; +export const MEDIA_USAGE_CLEANUP_LEASE_MS = 5 * 60 * 1000; +export const MEDIA_USAGE_CLEANUP_SAFETY_WINDOW_MS = 60 * 60 * 1000; +const MEDIA_USAGE_CLEANUP_TIME_BUDGET_MS = 5 * 1000; + +export interface MediaUsageCleanupResult { + status: "completed" | "failed" | "skipped"; + candidateRows: number; + deletedRows: number; + deletedOrphans: number; + deletedStale: number; + deletedAbandoned: number; + deletedWriteLeases: number; + backlogLowerBound: number; + scanHasMore: boolean; + durationMs: number; +} + +interface CleanupCandidates { + orphanIds: string[]; + staleIds: string[]; + abandonedIds: string[]; + entries: CleanupCandidateEntry[]; +} + +type CleanupTarget = "orphan" | "stale" | "abandoned"; + +interface CleanupCandidateEntry { + candidate: MediaUsageCleanupCandidate; + target: CleanupTarget | null; +} + +/** + * Reclaims a bounded window of obsolete media-usage occurrences. + * + * The persisted claim makes a cron tick single-flight across Worker isolates + * and Node processes. + */ +export async function cleanupMediaUsage(db: Kysely): Promise { + const startedMs = Date.now(); + const canIssueStatement = () => withinBudget(startedMs); + const repo = new MediaUsageRepository(db); + const leaseToken = ulid(); + const claim = await repo.claimMediaUsageCleanup({ + leaseToken, + leaseDurationSeconds: MEDIA_USAGE_CLEANUP_LEASE_MS / 1000, + nextEligibleDelaySeconds: MEDIA_USAGE_CLEANUP_INTERVAL_MS / 1000, + sweepSafetyWindowSeconds: MEDIA_USAGE_CLEANUP_SAFETY_WINDOW_MS / 1000, + }); + if (!claim) return emptyResult("skipped", elapsedSince(startedMs)); + + let candidateRows = 0; + let deletedOrphans = 0; + let deletedStale = 0; + let deletedAbandoned = 0; + let deletedWriteLeases = 0; + let backlogLowerBound = 0; + let scanHasMore = false; + let nextCursor = claim.cursor; + let sweepComplete = false; + + try { + if (canIssueStatement()) { + deletedWriteLeases = await repo.deleteExpiredGenerationWriteLeases( + MEDIA_USAGE_CLEANUP_WRITE_LEASE_DELETE_LIMIT, + cleanupLease(leaseToken), + canIssueStatement, + ); + } + + if (canIssueStatement()) { + const cutoff = claim.scanBeforeAt; + const candidates = await repo.findMediaUsageCleanupCandidates({ + cutoff, + cursor: claim.cursor, + limit: MEDIA_USAGE_CLEANUP_CANDIDATE_LIMIT, + cleanupLease: cleanupLease(leaseToken), + }); + candidateRows = candidates.length; + scanHasMore = candidates.length === MEDIA_USAGE_CLEANUP_CANDIDATE_LIMIT; + + const selected = selectCleanupCandidates(candidates, claim.claimedAt); + backlogLowerBound = + selected.orphanIds.length + selected.staleIds.length + selected.abandonedIds.length; + const completedTargets = new Set(); + let canContinue = true; + + if (canIssueStatement() && selected.orphanIds.length > 0) { + deletedOrphans = await repo.deleteOrphanOccurrencesOlderThan( + cutoff, + selected.orphanIds.length, + { + candidateIds: selected.orphanIds, + cleanupLease: cleanupLease(leaseToken), + canIssueStatement, + }, + ); + canContinue = deletedOrphans === selected.orphanIds.length; + if (canContinue) completedTargets.add("orphan"); + } + if (canContinue && canIssueStatement() && selected.staleIds.length > 0) { + deletedStale = await repo.deleteStaleGenerationsOlderThan( + cutoff, + selected.staleIds.length, + { + candidateIds: selected.staleIds, + cleanupLease: cleanupLease(leaseToken), + canIssueStatement, + }, + ); + canContinue = deletedStale === selected.staleIds.length; + if (canContinue) completedTargets.add("stale"); + } + if (canContinue && canIssueStatement() && selected.abandonedIds.length > 0) { + deletedAbandoned = await repo.deleteAbandonedGenerationsOlderThan( + cutoff, + selected.abandonedIds.length, + { + candidateIds: selected.abandonedIds, + cleanupLease: cleanupLease(leaseToken), + canIssueStatement, + }, + ); + if (deletedAbandoned === selected.abandonedIds.length) { + completedTargets.add("abandoned"); + } + } + const hasIncompleteTargets = selected.entries.some( + (entry) => entry.target !== null && !completedTargets.has(entry.target), + ); + sweepComplete = + !scanHasMore && selected.entries.length === candidates.length && !hasIncompleteTargets; + nextCursor = sweepComplete + ? null + : cursorAfterCompletedCandidates(selected, completedTargets, claim.cursor); + } + + const durationMs = elapsedSince(startedMs); + const completed = await repo.completeMediaUsageCleanup({ + leaseToken, + nextCursor, + sweepComplete, + candidateCount: candidateRows, + deletedOrphans, + deletedStale, + deletedAbandoned, + deletedWriteLeases, + backlogLowerBound, + scanHasMore, + durationMs, + }); + + return { + status: completed ? "completed" : "skipped", + candidateRows, + deletedRows: deletedOrphans + deletedStale + deletedAbandoned, + deletedOrphans, + deletedStale, + deletedAbandoned, + deletedWriteLeases, + backlogLowerBound, + scanHasMore, + durationMs, + }; + } catch (error) { + const durationMs = elapsedSince(startedMs); + const failures = Math.min(claim.consecutiveFailures + 1, 5); + try { + await repo.failMediaUsageCleanup({ + leaseToken, + retryDelaySeconds: failureDelayMs(failures) / 1000, + consecutiveFailures: failures, + durationMs, + errorCode: "MEDIA_USAGE_CLEANUP_FAILED", + }); + } catch (failureError) { + console.error("[media-usage-cleanup] Failed to record cleanup failure:", failureError); + } + console.error("[media-usage-cleanup] Cleanup failed:", error); + return { + ...emptyResult("failed", durationMs), + candidateRows, + deletedRows: deletedOrphans + deletedStale + deletedAbandoned, + deletedOrphans, + deletedStale, + deletedAbandoned, + deletedWriteLeases, + backlogLowerBound, + scanHasMore, + }; + } +} + +function selectCleanupCandidates( + candidates: readonly MediaUsageCleanupCandidate[], + activeLeaseAt: string, +): CleanupCandidates { + const orphanIds: string[] = []; + const staleIds: string[] = []; + const abandonedIds: string[] = []; + const entries: CleanupCandidateEntry[] = []; + + for (const candidate of candidates) { + if (hasActiveWriteLease(candidate, activeLeaseAt)) { + entries.push({ candidate, target: null }); + continue; + } + + const target = cleanupTarget(candidate); + if (target === null) { + entries.push({ candidate, target: null }); + continue; + } + if ( + orphanIds.length + staleIds.length + abandonedIds.length >= + MEDIA_USAGE_CLEANUP_DELETE_LIMIT + ) { + break; + } + if (target === "orphan") orphanIds.push(candidate.id); + if (target === "stale") staleIds.push(candidate.id); + if (target === "abandoned") abandonedIds.push(candidate.id); + entries.push({ candidate, target }); + } + + return { orphanIds, staleIds, abandonedIds, entries }; +} + +function cleanupTarget(candidate: MediaUsageCleanupCandidate): CleanupTarget | null { + if (candidate.currentGeneration === null) return "orphan"; + if (candidate.currentGeneration === candidate.generation || candidate.indexedAt === null) + return null; + return candidate.createdAt < candidate.indexedAt ? "stale" : "abandoned"; +} + +function hasActiveWriteLease( + candidate: MediaUsageCleanupCandidate, + activeLeaseAt: string, +): boolean { + return candidate.writeLeaseExpiresAt !== null && candidate.writeLeaseExpiresAt > activeLeaseAt; +} + +function cursorFor(candidate: MediaUsageCleanupCandidate): MediaUsageCleanupCursor { + return { createdAt: candidate.createdAt, id: candidate.id }; +} + +function cursorAfterCompletedCandidates( + selected: CleanupCandidates, + completedTargets: ReadonlySet, + priorCursor: MediaUsageCleanupCursor | null, +): MediaUsageCleanupCursor | null { + let cursor = priorCursor; + for (const entry of selected.entries) { + if (entry.target !== null && !completedTargets.has(entry.target)) break; + cursor = cursorFor(entry.candidate); + } + return cursor; +} + +function emptyResult( + status: Extract, + durationMs: number, +): MediaUsageCleanupResult { + return { + status, + candidateRows: 0, + deletedRows: 0, + deletedOrphans: 0, + deletedStale: 0, + deletedAbandoned: 0, + deletedWriteLeases: 0, + backlogLowerBound: 0, + scanHasMore: false, + durationMs, + }; +} + +function withinBudget(startedMs: number): boolean { + return elapsedSince(startedMs) < MEDIA_USAGE_CLEANUP_TIME_BUDGET_MS; +} + +function elapsedSince(startedMs: number): number { + return Math.max(0, Date.now() - startedMs); +} + +function failureDelayMs(consecutiveFailures: number): number { + return Math.min(2 ** (consecutiveFailures - 1), 15) * MEDIA_USAGE_CLEANUP_INTERVAL_MS; +} + +function cleanupLease(leaseToken: string) { + return { leaseToken }; +} diff --git a/packages/core/tests/integration/database/byline-fields-races.test.ts b/packages/core/tests/integration/database/byline-fields-races.test.ts index 3d345713d5..c7175e41a7 100644 --- a/packages/core/tests/integration/database/byline-fields-races.test.ts +++ b/packages/core/tests/integration/database/byline-fields-races.test.ts @@ -5,8 +5,8 @@ * exact +N increments — concurrent mutators can collapse one or both * bookends, which is fine per the registry's class JSDoc. * - * Activate Postgres parity by exporting `EMDASH_TEST_PG=1` and pointing - * `PG_CONNECTION_STRING` at a writable test database. + * Activate Postgres parity by setting `EMDASH_TEST_PG` to a connection string + * for a test role with `CREATEDB` privileges. */ import { beforeEach, afterEach, expect, it } from "vitest"; diff --git a/packages/core/tests/integration/database/media-usage-cleanup-plan.test.ts b/packages/core/tests/integration/database/media-usage-cleanup-plan.test.ts new file mode 100644 index 0000000000..77c78df9cb --- /dev/null +++ b/packages/core/tests/integration/database/media-usage-cleanup-plan.test.ts @@ -0,0 +1,597 @@ +import Database from "better-sqlite3"; +import { Kysely, SqliteDialect, sql } from "kysely"; +import { afterEach, beforeEach, expect, it, vi } from "vitest"; + +import { runMigrations } from "../../../src/database/migrations/runner.js"; +import { MediaUsageRepository } from "../../../src/database/repositories/media-usage.js"; +import type { Database as DatabaseSchema } from "../../../src/database/types.js"; +import { + cleanupMediaUsage, + MEDIA_USAGE_CLEANUP_CANDIDATE_LIMIT, + MEDIA_USAGE_CLEANUP_DELETE_LIMIT, +} from "../../../src/media/usage/cleanup.js"; +import { hasPgTestDatabase, setupForDialect, teardownForDialect } from "../../utils/test-db.js"; + +interface CapturedQuery { + sql: string; + parameters: readonly unknown[]; +} + +const MAX_CLEANUP_STATEMENTS_PER_TICK = 14; +const MAX_BIND_PARAMETERS_PER_CLEANUP_STATEMENT = 52; +const MAX_CLEANUP_ADMISSION_TIME_MS = 5_000; + +let sqlite: Database.Database; +let db: Kysely; +let repo: MediaUsageRepository; +let captured: CapturedQuery[]; +let afterQuery: ((query: CapturedQuery) => void) | undefined; + +beforeEach(async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date()); + captured = []; + afterQuery = undefined; + sqlite = new Database(":memory:"); + db = new Kysely({ + dialect: new SqliteDialect({ database: sqlite }), + log(event) { + if (event.level === "query") { + const query = { sql: event.query.sql, parameters: event.query.parameters }; + captured.push(query); + afterQuery?.(query); + } + }, + }); + await runMigrations(db); + repo = new MediaUsageRepository(db); +}); + +afterEach(async () => { + vi.useRealTimers(); + await db.destroy(); +}); + +it("uses an indexed, D1-compatible fixed statement and bind budget", async () => { + const stale = await repo.replaceSource( + contentSource("entry-budget"), + Array.from({ length: MEDIA_USAGE_CLEANUP_DELETE_LIMIT }, (_, index) => + occurrence(`media-stale-${index}`, `field-${index}`), + ), + ); + await repo.replaceSource(contentSource("entry-budget"), [occurrence("media-current")]); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", stale.currentGeneration) + .execute(); + captured = []; + + const result = await cleanupMediaUsage(db); + + expect(result).toEqual( + expect.objectContaining({ + status: "completed", + candidateRows: MEDIA_USAGE_CLEANUP_DELETE_LIMIT, + deletedRows: MEDIA_USAGE_CLEANUP_DELETE_LIMIT, + backlogLowerBound: MEDIA_USAGE_CLEANUP_DELETE_LIMIT, + }), + ); + expect(captured.length).toBeLessThanOrEqual(MAX_CLEANUP_STATEMENTS_PER_TICK); + for (const query of captured) { + expect(query.parameters.length).toBeLessThanOrEqual(MAX_BIND_PARAMETERS_PER_CLEANUP_STATEMENT); + } + + const candidateQuery = captured.find( + (query) => + query.sql.toLowerCase().includes("left join") && + query.sql.includes("_emdash_media_usage_generation_writes"), + ); + expect(candidateQuery).toBeDefined(); + const plan = explain(candidateQuery!); + expect(plan).toContain("idx__emdash_media_usage_cleanup_scan"); + expect(plan).not.toContain("USE TEMP B-TREE"); +}); + +it("uses the expiry index without sorting generation write leases", async () => { + await db + .insertInto("_emdash_media_usage_generation_writes") + .values( + Array.from({ length: 20 }, (_, index) => ({ + source_key: `expired-source-${index}`, + generation: `expired-generation-${index}`, + lease_token: `expired-writer-lease-${index}`, + expires_at: "2026-02-01T00:00:00.000Z", + created_at: "2026-02-01T00:00:00.000Z", + })), + ) + .execute(); + captured = []; + afterQuery = (query) => { + if (query.sql.startsWith('select "lease_token" from "_emdash_media_usage_generation_writes"')) { + vi.advanceTimersByTime(MAX_CLEANUP_ADMISSION_TIME_MS); + } + }; + + await cleanupMediaUsage(db); + + const expiryQuery = captured.find((query) => + query.sql.startsWith('select "lease_token" from "_emdash_media_usage_generation_writes"'), + ); + expect(expiryQuery).toBeDefined(); + const plan = explain(expiryQuery!); + expect(plan).toContain("idx__emdash_media_usage_generation_writes_expiry"); + expect(plan).not.toContain("USE TEMP B-TREE"); +}); + +it("does not dispatch a delete after marking consumes the time budget", async () => { + const stale = await repo.replaceSource(contentSource("entry-deadline"), [ + occurrence("media-before-deadline"), + ]); + await repo.replaceSource(contentSource("entry-deadline"), [occurrence("media-current")]); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", stale.currentGeneration) + .execute(); + captured = []; + let marked = false; + afterQuery = (query) => { + if (query.sql.startsWith('update "_emdash_media_usage" set "cleanup_lease_token"')) { + marked = true; + vi.advanceTimersByTime(MAX_CLEANUP_ADMISSION_TIME_MS); + } + }; + + const result = await cleanupMediaUsage(db); + + expect(marked).toBe(true); + expect(result).toEqual( + expect.objectContaining({ + candidateRows: 1, + deletedRows: 0, + durationMs: MAX_CLEANUP_ADMISSION_TIME_MS, + }), + ); + expect(captured.some((query) => query.sql.startsWith('delete from "_emdash_media_usage"'))).toBe( + false, + ); + expect( + await db + .selectFrom("_emdash_media_usage") + .select("id") + .where("generation", "=", stale.currentGeneration) + .execute(), + ).toHaveLength(1); +}); + +it.each([ + { + name: "full page after prior progress", + currentRows: MEDIA_USAGE_CLEANUP_CANDIDATE_LIMIT - MEDIA_USAGE_CLEANUP_DELETE_LIMIT, + hasPriorCursor: true, + }, + { name: "short page after prior progress", currentRows: 0, hasPriorCursor: true }, + { name: "short first page", currentRows: 0, hasPriorCursor: false }, +])("preserves finite sweep state for a partial $name", async ({ currentRows, hasPriorCursor }) => { + const stale = await repo.replaceSource( + contentSource("entry-partial-deadline"), + Array.from({ length: MEDIA_USAGE_CLEANUP_DELETE_LIMIT }, (_, index) => + occurrence(`media-stale-${index}`, `stale-${index}`), + ), + ); + const current = await repo.replaceSource(contentSource("entry-partial-deadline"), [ + ...(hasPriorCursor ? [occurrence("media-before-stale", "before-stale")] : []), + ...Array.from({ length: currentRows }, (_, index) => + occurrence(`media-current-${index}`, `current-${index}`), + ), + ]); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T18:00:00.000Z" }) + .where("generation", "=", stale.currentGeneration) + .execute(); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "!=", stale.currentGeneration) + .execute(); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T17:00:00.000Z" }) + .where("generation", "=", current.currentGeneration) + .where("field_path", "=", "before-stale") + .execute(); + const priorCursor = hasPriorCursor + ? await db + .selectFrom("_emdash_media_usage") + .select(["id", "created_at"]) + .where("generation", "=", current.currentGeneration) + .where("field_path", "=", "before-stale") + .executeTakeFirstOrThrow() + : null; + const scanBeforeAt = "2026-02-02T00:00:00.000Z"; + await db + .updateTable("_emdash_media_usage_cleanup") + .set({ + cursor_created_at: priorCursor?.created_at ?? null, + cursor_id: priorCursor?.id ?? null, + scan_before_at: scanBeforeAt, + }) + .where("task_key", "=", "projection_gc") + .execute(); + captured = []; + let firstDeleteCompleted = false; + afterQuery = (query) => { + if (!firstDeleteCompleted && query.sql.startsWith('delete from "_emdash_media_usage"')) { + firstDeleteCompleted = true; + vi.advanceTimersByTime(MAX_CLEANUP_ADMISSION_TIME_MS); + } + }; + + const result = await cleanupMediaUsage(db); + + expect(firstDeleteCompleted).toBe(true); + expect(result.candidateRows).toBe(MEDIA_USAGE_CLEANUP_DELETE_LIMIT + currentRows); + expect(result.deletedRows).toBeGreaterThan(0); + expect(result.deletedRows).toBeLessThan(MEDIA_USAGE_CLEANUP_DELETE_LIMIT); + afterQuery = undefined; + expect( + captured.filter((query) => + query.sql.startsWith('update "_emdash_media_usage" set "cleanup_lease_token"'), + ), + ).toHaveLength(1); + expect( + await db + .selectFrom("_emdash_media_usage_cleanup") + .select(["cursor_created_at", "cursor_id", "scan_before_at"]) + .where("task_key", "=", "projection_gc") + .executeTakeFirstOrThrow(), + ).toEqual({ + cursor_created_at: priorCursor?.created_at ?? null, + cursor_id: priorCursor?.id ?? null, + scan_before_at: scanBeforeAt, + }); + const remaining = await db + .selectFrom("_emdash_media_usage") + .select("cleanup_lease_token") + .where("generation", "=", stale.currentGeneration) + .execute(); + expect(remaining).toHaveLength(MEDIA_USAGE_CLEANUP_DELETE_LIMIT - result.deletedRows); + expect(remaining.every((row) => row.cleanup_lease_token === null)).toBe(true); +}); + +it("retains the sweep after reaching the delete cap on a short page", async () => { + const stale = await repo.replaceSource( + contentSource("entry-short-delete-cap"), + Array.from({ length: MEDIA_USAGE_CLEANUP_DELETE_LIMIT + 1 }, (_, index) => + occurrence(`media-stale-cap-${index}`, `stale-cap-${index}`), + ), + ); + const current = await repo.replaceSource(contentSource("entry-short-delete-cap"), [ + occurrence("media-before-stale-cap", "before-stale-cap"), + ]); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T18:00:00.000Z" }) + .where("generation", "=", stale.currentGeneration) + .execute(); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T17:00:00.000Z" }) + .where("generation", "=", current.currentGeneration) + .execute(); + const priorCursor = await db + .selectFrom("_emdash_media_usage") + .select(["id", "created_at"]) + .where("generation", "=", current.currentGeneration) + .executeTakeFirstOrThrow(); + const staleCandidates = await db + .selectFrom("_emdash_media_usage") + .select(["id", "created_at"]) + .where("generation", "=", stale.currentGeneration) + .orderBy("created_at", "asc") + .orderBy("id", "asc") + .execute(); + const expectedCursor = staleCandidates[MEDIA_USAGE_CLEANUP_DELETE_LIMIT - 1]!; + const expectedRemaining = staleCandidates[MEDIA_USAGE_CLEANUP_DELETE_LIMIT]!; + const scanBeforeAt = "2026-02-02T00:00:00.000Z"; + await db + .updateTable("_emdash_media_usage_cleanup") + .set({ + cursor_created_at: priorCursor.created_at, + cursor_id: priorCursor.id, + scan_before_at: scanBeforeAt, + }) + .where("task_key", "=", "projection_gc") + .execute(); + + const result = await cleanupMediaUsage(db); + + expect(result).toEqual( + expect.objectContaining({ + candidateRows: MEDIA_USAGE_CLEANUP_DELETE_LIMIT + 1, + deletedRows: MEDIA_USAGE_CLEANUP_DELETE_LIMIT, + scanHasMore: false, + }), + ); + expect( + await db + .selectFrom("_emdash_media_usage_cleanup") + .select(["cursor_created_at", "cursor_id", "scan_before_at"]) + .where("task_key", "=", "projection_gc") + .executeTakeFirstOrThrow(), + ).toEqual({ + cursor_created_at: expectedCursor.created_at, + cursor_id: expectedCursor.id, + scan_before_at: scanBeforeAt, + }); + expect( + await db + .selectFrom("_emdash_media_usage") + .select("id") + .where("generation", "=", stale.currentGeneration) + .execute(), + ).toEqual([{ id: expectedRemaining.id }]); +}); + +it("does not delete writer leases after their scan consumes the time budget", async () => { + await db + .insertInto("_emdash_media_usage_generation_writes") + .values({ + source_key: "expired-source", + generation: "expired-generation", + lease_token: "expired-writer-lease", + expires_at: "2026-02-01T00:00:00.000Z", + created_at: "2026-02-01T00:00:00.000Z", + }) + .execute(); + captured = []; + let scanned = false; + afterQuery = (query) => { + if (query.sql.startsWith('select "lease_token" from "_emdash_media_usage_generation_writes"')) { + scanned = true; + vi.advanceTimersByTime(MAX_CLEANUP_ADMISSION_TIME_MS); + } + }; + + const result = await cleanupMediaUsage(db); + + expect(scanned).toBe(true); + expect(result).toEqual( + expect.objectContaining({ + deletedWriteLeases: 0, + durationMs: MAX_CLEANUP_ADMISSION_TIME_MS, + }), + ); + expect( + captured.some((query) => + query.sql.startsWith('delete from "_emdash_media_usage_generation_writes"'), + ), + ).toBe(false); + afterQuery = undefined; + expect( + await db + .selectFrom("_emdash_media_usage_generation_writes") + .select("lease_token") + .where("lease_token", "=", "expired-writer-lease") + .execute(), + ).toHaveLength(1); + expect( + await db + .selectFrom("_emdash_media_usage_cleanup") + .select(["cursor_created_at", "cursor_id", "scan_before_at"]) + .where("task_key", "=", "projection_gc") + .executeTakeFirstOrThrow(), + ).toEqual({ + cursor_created_at: null, + cursor_id: null, + scan_before_at: expect.any(String), + }); +}); + +it("reserves a failure update within the fixed statement and bind budget", async () => { + const stale = await repo.replaceSource(contentSource("entry-stale"), [occurrence("media-stale")]); + await repo.replaceSource(contentSource("entry-stale"), [occurrence("media-current")]); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", stale.currentGeneration) + .execute(); + for (let index = 0; index < MEDIA_USAGE_CLEANUP_DELETE_LIMIT - 2; index++) { + await insertOccurrence(db, { + id: `orphan-${index}`, + sourceKey: `missing-source-${index}`, + generation: `orphan-generation-${index}`, + mediaId: `media-orphan-${index}`, + createdAt: "2026-02-01T18:00:00.000Z", + }); + } + const abandoned = await repo.replaceSource(contentSource("entry-abandoned"), [ + occurrence("media-abandoned-current"), + ]); + await db + .updateTable("_emdash_media_usage_sources") + .set({ indexed_at: "2026-02-01T18:00:00.000Z" }) + .where("source_key", "=", abandoned.sourceKey) + .execute(); + await insertOccurrence(db, { + id: "abandoned-occurrence", + sourceKey: abandoned.sourceKey, + generation: "abandoned-generation", + mediaId: "media-abandoned", + createdAt: "2026-02-01T19:00:00.000Z", + }); + await db + .insertInto("_emdash_media_usage_generation_writes") + .values({ + source_key: "expired-source", + generation: "expired-generation", + lease_token: "expired-writer-lease", + expires_at: "2026-02-01T00:00:00.000Z", + created_at: "2026-02-01T00:00:00.000Z", + }) + .execute(); + captured = []; + const original = MediaUsageRepository.prototype.completeMediaUsageCleanup; + vi.spyOn(MediaUsageRepository.prototype, "completeMediaUsageCleanup").mockImplementation( + async function (this: MediaUsageRepository, input) { + await original.call(this, input); + throw new Error("completion transport failure"); + }, + ); + vi.spyOn(console, "error").mockImplementation(() => undefined); + + const result = await cleanupMediaUsage(db); + expect(result).toEqual( + expect.objectContaining({ + status: "failed", + candidateRows: MEDIA_USAGE_CLEANUP_DELETE_LIMIT, + deletedRows: MEDIA_USAGE_CLEANUP_DELETE_LIMIT, + deletedOrphans: MEDIA_USAGE_CLEANUP_DELETE_LIMIT - 2, + deletedStale: 1, + deletedAbandoned: 1, + deletedWriteLeases: 1, + }), + ); + expect(captured.some((query) => query.parameters.includes("MEDIA_USAGE_CLEANUP_FAILED"))).toBe( + true, + ); + expect(captured.length).toBeLessThanOrEqual(MAX_CLEANUP_STATEMENTS_PER_TICK); + for (const query of captured) { + expect(query.parameters.length).toBeLessThanOrEqual(MAX_BIND_PARAMETERS_PER_CLEANUP_STATEMENT); + } +}); + +function contentSource(contentId: string) { + return { + sourceKey: `content:posts:${contentId}:columns`, + sourceType: "content", + collectionSlug: "posts", + contentId, + sourceVariant: "columns" as const, + contentStatus: "published", + }; +} + +function occurrence(mediaId: string, fieldPath = "hero") { + return { + fieldSlug: fieldPath, + fieldPath, + referenceType: "image_field" as const, + mediaId, + provider: "local", + providerAssetId: mediaId, + }; +} + +async function insertOccurrence( + database: Kysely, + input: { + id: string; + sourceKey: string; + generation: string; + mediaId: string; + createdAt: string; + }, +): Promise { + await database + .insertInto("_emdash_media_usage") + .values({ + id: input.id, + source_key: input.sourceKey, + generation: input.generation, + field_slug: "hero", + field_path: input.id, + occurrence_index: 0, + reference_type: "image_field", + media_id: input.mediaId, + provider: "local", + provider_asset_id: input.mediaId, + media_kind: "image", + mime_type: null, + created_at: input.createdAt, + }) + .execute(); +} + +function explain(query: CapturedQuery): string { + const rows = sqlite.prepare(`EXPLAIN QUERY PLAN ${query.sql}`).all(...query.parameters) as { + detail: string; + }[]; + return rows.map((row) => row.detail).join("\n"); +} + +it.skipIf(!hasPgTestDatabase)("uses the cleanup scan index in PostgreSQL", async () => { + const context = await setupForDialect("postgres"); + try { + const now = new Date(); + for (let batchStart = 0; batchStart < 6_000; batchStart += 1_000) { + await context.db + .insertInto("_emdash_media_usage") + .values( + Array.from({ length: 1_000 }, (_, offset) => { + const index = batchStart + offset; + return { + id: `plan-${index}`, + source_key: `plan-source-${index}`, + generation: `01K000000000000000000${String(index).padStart(3, "0")}`, + field_slug: "hero", + field_path: "hero", + occurrence_index: 0, + reference_type: "image_field", + media_id: null, + provider: "local", + provider_asset_id: `plan-media-${index}`, + media_kind: "image", + mime_type: null, + created_at: new Date(now.getTime() - (index + 2) * 60_000).toISOString(), + }; + }), + ) + .execute(); + } + + await sql`ANALYZE _emdash_media_usage`.execute(context.db); + await context.db + .updateTable("_emdash_media_usage_cleanup") + .set({ + lease_token: "plan-cleanup-lease", + lease_expires_at: "2100-01-01T00:00:00.000Z", + }) + .where("task_key", "=", "projection_gc") + .execute(); + const result = await sql<{ "QUERY PLAN": string }>` + EXPLAIN (COSTS OFF) + SELECT + u.id, + u.source_key, + u.generation, + u.created_at, + s.current_generation, + s.indexed_at, + writer.expires_at AS write_lease_expires_at + FROM _emdash_media_usage AS u + LEFT JOIN _emdash_media_usage_sources AS s ON s.source_key = u.source_key + LEFT JOIN _emdash_media_usage_generation_writes AS writer + ON writer.source_key = u.source_key AND writer.generation = u.generation + WHERE u.created_at < ${now.toISOString()} + AND EXISTS ( + SELECT 1 + FROM _emdash_media_usage_cleanup AS cleanup + WHERE cleanup.task_key = 'projection_gc' + AND cleanup.lease_token = 'plan-cleanup-lease' + AND cleanup.lease_expires_at::timestamptz > clock_timestamp() + FOR UPDATE + ) + ORDER BY u.created_at ASC, u.id ASC + LIMIT 250 + `.execute(context.db); + const plan = result.rows.map((row) => row["QUERY PLAN"]).join("\n"); + + expect(plan).toMatch(/Index(?: Only)? Scan using idx__emdash_media_usage_cleanup_scan/); + expect(plan).not.toContain("Sort"); + } finally { + await teardownForDialect(context); + } +}); diff --git a/packages/core/tests/integration/database/media-usage-cleanup.test.ts b/packages/core/tests/integration/database/media-usage-cleanup.test.ts new file mode 100644 index 0000000000..c2f16fff32 --- /dev/null +++ b/packages/core/tests/integration/database/media-usage-cleanup.test.ts @@ -0,0 +1,1095 @@ +import { sql, type Kysely, type Transaction } from "kysely"; +import { afterEach, beforeEach, expect, it, vi } from "vitest"; + +import { runSystemCleanup } from "../../../src/cleanup.js"; +import { MediaUsageRepository } from "../../../src/database/repositories/media-usage.js"; +import type { Database } from "../../../src/database/types.js"; +import { + cleanupMediaUsage, + MEDIA_USAGE_CLEANUP_CANDIDATE_LIMIT, + MEDIA_USAGE_CLEANUP_DELETE_LIMIT, + MEDIA_USAGE_CLEANUP_INTERVAL_MS, +} from "../../../src/media/usage/cleanup.js"; +import { + describeEachDialect, + setupForDialect, + teardownForDialect, + type DialectTestContext, +} from "../../utils/test-db.js"; + +const MAX_CLEANUP_ADMISSION_TIME_MS = 5_000; + +describeEachDialect("scheduled media usage cleanup", (dialect) => { + let ctx: DialectTestContext; + let db: Kysely; + let repo: MediaUsageRepository; + + beforeEach(async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date()); + ctx = await setupForDialect(dialect); + db = ctx.db; + repo = new MediaUsageRepository(db); + }); + + afterEach(async () => { + vi.restoreAllMocks(); + vi.useRealTimers(); + await teardownForDialect(ctx); + }); + + it("reclaims every superseded generation class through production maintenance", async () => { + const first = await repo.replaceSource(contentSource("entry-1"), [occurrence("media-stale")]); + const current = await repo.replaceSource(contentSource("entry-1"), [ + occurrence("media-current"), + ]); + + await db + .updateTable("_emdash_media_usage_sources") + .set({ indexed_at: "2026-02-01T20:00:00.000Z" }) + .where("source_key", "=", current.sourceKey) + .execute(); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", first.currentGeneration) + .execute(); + await insertOccurrence(db, { + id: "abandoned-occurrence", + sourceKey: current.sourceKey, + generation: "abandoned-generation", + mediaId: "media-abandoned", + createdAt: "2026-02-01T21:00:00.000Z", + }); + await insertOccurrence(db, { + id: "orphan-occurrence", + sourceKey: "missing-source", + generation: "orphan-generation", + mediaId: "media-orphan", + createdAt: "2026-02-01T18:00:00.000Z", + }); + + await runSystemCleanup(db); + + const remaining = await db + .selectFrom("_emdash_media_usage") + .select(["id", "generation", "media_id"]) + .orderBy("id", "asc") + .execute(); + expect(remaining).toEqual([ + expect.objectContaining({ + generation: current.currentGeneration, + media_id: "media-current", + }), + ]); + }); + + it("admits only one overlapping tick to the shared delete budget", async () => { + const stale = await repo.replaceSource( + contentSource("entry-overlap"), + Array.from({ length: MEDIA_USAGE_CLEANUP_DELETE_LIMIT + 20 }, (_, index) => + occurrence(`media-stale-${index}`, `field-${index}`), + ), + ); + await repo.replaceSource(contentSource("entry-overlap"), [occurrence("media-current")]); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", stale.currentGeneration) + .execute(); + + const results = await Promise.all([cleanupMediaUsage(db), cleanupMediaUsage(db)]); + + expect(results.filter((result) => result.status === "completed")).toHaveLength(1); + expect(results.filter((result) => result.status === "skipped")).toHaveLength(1); + const remaining = await db + .selectFrom("_emdash_media_usage") + .select(({ fn }) => fn.countAll().as("count")) + .where("generation", "=", stale.currentGeneration) + .executeTakeFirstOrThrow(); + expect(Number(remaining.count)).toBe(20); + }); + + it("does not let a fast scheduler clock steal an active cleanup lease", async () => { + const owner = await repo.claimMediaUsageCleanup({ + leaseToken: "database-clock-owner", + leaseDurationSeconds: 5 * 60, + nextEligibleDelaySeconds: 60, + sweepSafetyWindowSeconds: 60 * 60, + }); + expect(owner).not.toBeNull(); + + vi.advanceTimersByTime(6 * 60 * 1000); + const stolen = await repo.claimMediaUsageCleanup({ + leaseToken: "fast-scheduler", + leaseDurationSeconds: 5 * 60, + nextEligibleDelaySeconds: 60, + sweepSafetyWindowSeconds: 60 * 60, + }); + + expect(stolen).toBeNull(); + }); + + it("advances a bounded cursor past old live rows to later garbage", async () => { + const stale = await repo.replaceSource( + contentSource("entry-cursor"), + Array.from({ length: 5 }, (_, index) => + occurrence(`media-stale-cursor-${index}`, `stale-${index}`), + ), + ); + const current = await repo.replaceSource( + contentSource("entry-cursor"), + Array.from({ length: MEDIA_USAGE_CLEANUP_CANDIDATE_LIMIT }, (_, index) => + occurrence(`media-current-cursor-${index}`, `current-${index}`), + ), + ); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", current.currentGeneration) + .execute(); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:10:00.000Z" }) + .where("generation", "=", stale.currentGeneration) + .execute(); + + const first = await cleanupMediaUsage(db); + expect(first).toEqual( + expect.objectContaining({ + status: "completed", + candidateRows: MEDIA_USAGE_CLEANUP_CANDIDATE_LIMIT, + deletedRows: 0, + scanHasMore: true, + }), + ); + + await makeCleanupEligible(db); + const second = await cleanupMediaUsage(db); + expect(second).toEqual(expect.objectContaining({ status: "completed", deletedRows: 5 })); + expect( + await db + .selectFrom("_emdash_media_usage") + .select("id") + .where("generation", "=", stale.currentGeneration) + .execute(), + ).toEqual([]); + }); + + it("restarts a finite sweep before later work can strand newly stale rows", async () => { + const head = await repo.replaceSource( + contentSource("entry-sweep-head"), + Array.from({ length: MEDIA_USAGE_CLEANUP_CANDIDATE_LIMIT }, (_, index) => + occurrence(`media-sweep-head-${index}`, `head-${index}`), + ), + ); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", head.currentGeneration) + .execute(); + + expect((await cleanupMediaUsage(db)).candidateRows).toBe(MEDIA_USAGE_CLEANUP_CANDIDATE_LIMIT); + const firstSweep = await db + .selectFrom("_emdash_media_usage_cleanup") + .select("scan_before_at") + .where("task_key", "=", "projection_gc") + .executeTakeFirstOrThrow(); + expect(firstSweep.scan_before_at).not.toBeNull(); + + await repo.replaceSource(contentSource("entry-sweep-head"), [ + occurrence("media-sweep-current"), + ]); + vi.advanceTimersByTime(30 * 60 * 1000); + const laterCreatedAt = new Date( + Date.parse(firstSweep.scan_before_at!) + 15 * 60 * 1000, + ).toISOString(); + for (let index = 0; index < MEDIA_USAGE_CLEANUP_CANDIDATE_LIMIT; index += 1) { + await insertOccurrence(db, { + id: `sweep-later-${index}`, + sourceKey: `sweep-later-source-${index}`, + generation: `01K00000000000000000${String(index).padStart(3, "0")}`, + mediaId: `media-sweep-later-${index}`, + createdAt: laterCreatedAt, + }); + } + + await makeCleanupEligible(db); + expect(await cleanupMediaUsage(db)).toEqual( + expect.objectContaining({ candidateRows: 0, scanHasMore: false }), + ); + await makeCleanupEligible(db); + expect((await cleanupMediaUsage(db)).deletedRows).toBe(MEDIA_USAGE_CLEANUP_DELETE_LIMIT); + }); + + it("backs off after a failed tick and retries from the persisted state", async () => { + const source = await repo.replaceSource(contentSource("entry-failure"), [ + occurrence("media-before-failure"), + ]); + await repo.replaceSource(contentSource("entry-failure"), [occurrence("media-current")]); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", source.currentGeneration) + .execute(); + + vi.spyOn( + MediaUsageRepository.prototype, + "findMediaUsageCleanupCandidates", + ).mockImplementationOnce(async function (this: MediaUsageRepository, input) { + await Promise.resolve(); + throw new Error(`cleanup candidate failure at ${input.cutoff}`); + }); + vi.spyOn(console, "error").mockImplementation(() => undefined); + + const failed = await cleanupMediaUsage(db); + expect(failed.status).toBe("failed"); + const failureState = await db + .selectFrom("_emdash_media_usage_cleanup") + .select(["next_eligible_at", "consecutive_failures", "last_error_code"]) + .where("task_key", "=", "projection_gc") + .executeTakeFirstOrThrow(); + expect(failureState).toEqual( + expect.objectContaining({ + consecutive_failures: 1, + last_error_code: "MEDIA_USAGE_CLEANUP_FAILED", + }), + ); + expect(Date.parse(failureState.next_eligible_at)).toBeGreaterThan(Date.now()); + + expect((await cleanupMediaUsage(db)).status).toBe("skipped"); + await makeCleanupEligible(db); + expect((await cleanupMediaUsage(db)).deletedRows).toBe(1); + }); + + it("does not admit another cleanup statement after its time budget expires", async () => { + const stale = await repo.replaceSource(contentSource("entry-time-budget"), [ + occurrence("media-before-time-budget"), + ]); + await repo.replaceSource(contentSource("entry-time-budget"), [occurrence("media-current")]); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", stale.currentGeneration) + .execute(); + + const original = MediaUsageRepository.prototype.findMediaUsageCleanupCandidates; + vi.spyOn(MediaUsageRepository.prototype, "findMediaUsageCleanupCandidates").mockImplementation( + async function (this: MediaUsageRepository, input) { + const candidates = await original.call(this, input); + vi.advanceTimersByTime(MAX_CLEANUP_ADMISSION_TIME_MS); + return candidates; + }, + ); + + const result = await cleanupMediaUsage(db); + expect(result).toEqual( + expect.objectContaining({ + candidateRows: 1, + deletedRows: 0, + durationMs: MAX_CLEANUP_ADMISSION_TIME_MS, + }), + ); + expect( + await db + .selectFrom("_emdash_media_usage") + .select("id") + .where("generation", "=", stale.currentGeneration) + .execute(), + ).toHaveLength(1); + }); + + it("protects an active writer lease and reclaims its abandoned generation after expiry", async () => { + const stale = await repo.replaceSource(contentSource("entry-writer-lease"), [ + occurrence("media-stale-writer"), + ]); + await repo.replaceSource(contentSource("entry-writer-lease"), [occurrence("media-current")]); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", stale.currentGeneration) + .execute(); + await db + .insertInto("_emdash_media_usage_generation_writes") + .values({ + source_key: stale.sourceKey, + generation: stale.currentGeneration, + lease_token: "active-writer-lease", + expires_at: new Date(Date.now() + 60 * MEDIA_USAGE_CLEANUP_INTERVAL_MS).toISOString(), + created_at: new Date().toISOString(), + }) + .execute(); + + expect((await cleanupMediaUsage(db)).deletedRows).toBe(0); + expect( + await db + .selectFrom("_emdash_media_usage") + .select("id") + .where("generation", "=", stale.currentGeneration) + .execute(), + ).toHaveLength(1); + + await db + .updateTable("_emdash_media_usage_generation_writes") + .set({ expires_at: "1970-01-01T00:00:00.000Z" }) + .where("lease_token", "=", "active-writer-lease") + .execute(); + await makeCleanupEligible(db); + const reclaimed = await cleanupMediaUsage(db); + expect(reclaimed).toEqual(expect.objectContaining({ deletedRows: 1, deletedWriteLeases: 1 })); + }); + it("derives generation-write and occurrence timestamps from database time", async () => { + vi.setSystemTime(new Date("2099-01-01T00:00:00.000Z")); + let expiresAt: string | undefined; + let occurrenceCreatedAt: string | undefined; + const internals = repo as unknown as { + upsertSource( + transaction: Kysely | Transaction, + source: unknown, + generation: string, + now: string, + leaseToken: string, + ): Promise; + }; + const original = internals.upsertSource.bind(repo); + vi.spyOn(internals, "upsertSource").mockImplementation( + async (transaction, source, generation, now, leaseToken) => { + const [writeLease, storedOccurrence] = await Promise.all([ + transaction + .selectFrom("_emdash_media_usage_generation_writes") + .select("expires_at") + .where("lease_token", "=", leaseToken) + .executeTakeFirstOrThrow(), + transaction + .selectFrom("_emdash_media_usage") + .select("created_at") + .where("generation", "=", generation) + .executeTakeFirstOrThrow(), + ]); + expiresAt = writeLease.expires_at; + occurrenceCreatedAt = storedOccurrence.created_at; + return original(transaction, source, generation, now, leaseToken); + }, + ); + + await repo.replaceSource(contentSource("entry-database-writer-clock"), [ + occurrence("media-database-writer-clock"), + ]); + + const databaseNow = await databaseNowTimestamp(db, dialect); + const remainingMs = Date.parse(expiresAt!) - Date.parse(databaseNow); + expect(remainingMs).toBeGreaterThan(59 * 60 * 1000); + expect(remainingMs).toBeLessThan(61 * 60 * 1000); + const occurrenceAgeMs = Date.parse(occurrenceCreatedAt!) - Date.parse(databaseNow); + expect(occurrenceAgeMs).toBeGreaterThan(-60_000); + expect(occurrenceAgeMs).toBeLessThan(60_000); + }); + + it("fences a pre-lease writer after cleanup reclaims its generation", async () => { + const current = await repo.replaceSource(contentSource("entry-old-writer"), [ + occurrence("media-current"), + ]); + const oldGeneration = "01J00000000000000000000000"; + await insertOccurrence(db, { + id: "old-writer-occurrence", + sourceKey: current.sourceKey, + generation: oldGeneration, + mediaId: "media-old-writer", + createdAt: "2026-02-01T19:00:00.000Z", + }); + + expect((await cleanupMediaUsage(db)).deletedRows).toBe(1); + expect( + await db + .selectFrom("_emdash_media_usage_cleanup_fence") + .select("generation_floor") + .where("task_key", "=", "projection_gc") + .executeTakeFirstOrThrow(), + ).toEqual({ generation_floor: oldGeneration }); + + const promotion = await db + .updateTable("_emdash_media_usage_sources") + .set({ current_generation: oldGeneration }) + .where("source_key", "=", current.sourceKey) + .executeTakeFirst(); + expect(Number(promotion.numUpdatedRows ?? 0)).toBe(0); + expect(await repo.findCurrentUsageByMediaId("media-current")).toHaveLength(1); + }); + + it("does not advance the promotion fence when normal source deletion removes a cleanup-marked occurrence", async () => { + const current = await repo.replaceSource(contentSource("entry-normal-delete"), [ + occurrence("media-normal-delete"), + ]); + expect( + await repo.claimMediaUsageCleanup({ + leaseToken: "normal-delete-cleanup-owner", + leaseDurationSeconds: 5 * 60, + nextEligibleDelaySeconds: 60, + sweepSafetyWindowSeconds: 60 * 60, + }), + ).not.toBeNull(); + await db + .updateTable("_emdash_media_usage") + .set({ cleanup_lease_token: "normal-delete-cleanup-owner" }) + .where("source_key", "=", current.sourceKey) + .execute(); + + expect(await repo.deleteSource(current.sourceKey)).toBe(1); + expect(await db.selectFrom("_emdash_media_usage_cleanup_fence").selectAll().execute()).toEqual( + [], + ); + }); + + it("allows attempt-only sources behind the cleanup generation fence", async () => { + await db + .insertInto("_emdash_media_usage_cleanup_fence") + .values({ task_key: "projection_gc", generation_floor: "ZZZZZZZZZZZZZZZZZZZZZZZZZZ" }) + .execute(); + + const attempted = await repo.markSourceAttempted({ + ...contentSource("entry-attempt-only"), + lastErrorCode: "MEDIA_USAGE_INDEX_FAILED", + }); + const guarded = await repo.markSourceAttemptedIfMatching( + { + ...contentSource("entry-guarded-attempt-only"), + lastErrorCode: "MEDIA_USAGE_INDEX_FAILED", + }, + null, + ); + + expect(attempted.sourceKey).toBe("content:posts:entry-attempt-only:columns"); + expect(attempted.lastErrorCode).toBe("MEDIA_USAGE_INDEX_FAILED"); + expect(guarded).toEqual({ attempted: true, source: null }); + expect( + await repo.findSource("content:posts:entry-guarded-attempt-only:columns"), + ).not.toBeNull(); + expect( + await db + .selectFrom("_emdash_media_usage") + .select("id") + .where("source_key", "in", [ + attempted.sourceKey, + "content:posts:entry-guarded-attempt-only:columns", + ]) + .execute(), + ).toEqual([]); + }); + + it("allows unrelated PostgreSQL source writes to overlap", async () => { + if (dialect !== "postgres") return; + + let releaseFirst!: () => void; + const firstRelease = new Promise((resolve) => { + releaseFirst = resolve; + }); + let signalFirstInserted!: () => void; + const firstInserted = new Promise((resolve) => { + signalFirstInserted = resolve; + }); + const first = db.transaction().execute(async (trx) => { + await trx + .insertInto("_emdash_media_usage_sources") + .values({ + source_key: "concurrent-source-a", + source_type: "content", + source_variant: "columns", + current_generation: "concurrent-generation-a", + }) + .execute(); + signalFirstInserted(); + await firstRelease; + }); + await firstInserted; + + let secondError: unknown; + try { + await db.transaction().execute(async (trx) => { + await sql`SET LOCAL lock_timeout = '250ms'`.execute(trx); + await trx + .insertInto("_emdash_media_usage_sources") + .values({ + source_key: "concurrent-source-b", + source_type: "content", + source_variant: "columns", + current_generation: "concurrent-generation-b", + }) + .execute(); + }); + } catch (error) { + secondError = error; + } + releaseFirst(); + await first; + + expect(secondError).toBeUndefined(); + }); + + it("allows an unrelated PostgreSQL source write during source deletion", async () => { + if (dialect !== "postgres") return; + + const current = await repo.replaceSource(contentSource("entry-concurrent-delete"), [ + occurrence("media-concurrent-delete"), + ]); + const internals = repo as unknown as { + deleteSourceGenerationOccurrences( + database: Kysely | Transaction, + sourceKey: string, + generation: string, + ): Promise; + }; + const original = internals.deleteSourceGenerationOccurrences.bind(repo); + let releaseDelete!: () => void; + const deleteRelease = new Promise((resolve) => { + releaseDelete = resolve; + }); + let signalDeleteStarted!: () => void; + const deleteStarted = new Promise((resolve) => { + signalDeleteStarted = resolve; + }); + vi.spyOn(internals, "deleteSourceGenerationOccurrences").mockImplementation( + async (database, sourceKey, generation) => { + signalDeleteStarted(); + await deleteRelease; + return original(database, sourceKey, generation); + }, + ); + const deleting = repo.deleteSourceIfCurrent(current.sourceKey, current.currentGeneration); + await deleteStarted; + + let sourceWriteError: unknown; + try { + await db.transaction().execute(async (trx) => { + await sql`SET LOCAL lock_timeout = '250ms'`.execute(trx); + await trx + .insertInto("_emdash_media_usage_sources") + .values({ + source_key: "source-during-delete", + source_type: "content", + source_variant: "columns", + current_generation: "generation-during-delete", + }) + .execute(); + }); + } catch (error) { + sourceWriteError = error; + } + releaseDelete(); + + expect(await deleting).toEqual({ deleted: true, source: null }); + expect(sourceWriteError).toBeUndefined(); + }); + + it("makes PostgreSQL cleanup wait for an in-flight source write", async () => { + if (dialect !== "postgres") return; + + let releaseSourceWrite!: () => void; + const sourceWriteRelease = new Promise((resolve) => { + releaseSourceWrite = resolve; + }); + let signalSourceInserted!: () => void; + const sourceInserted = new Promise((resolve) => { + signalSourceInserted = resolve; + }); + const writing = db.transaction().execute(async (trx) => { + await trx + .insertInto("_emdash_media_usage_sources") + .values({ + source_key: "cleanup-fence-source", + source_type: "content", + source_variant: "columns", + current_generation: "cleanup-fence-generation", + }) + .execute(); + signalSourceInserted(); + await sourceWriteRelease; + }); + await sourceInserted; + + let claimError: unknown; + try { + await db.transaction().execute(async (trx) => { + await sql`SET LOCAL lock_timeout = '250ms'`.execute(trx); + await new MediaUsageRepository(trx).claimMediaUsageCleanup({ + leaseToken: "blocked-source-write-claimant", + leaseDurationSeconds: 5 * 60, + nextEligibleDelaySeconds: 60, + sweepSafetyWindowSeconds: 60 * 60, + }); + }); + } catch (error) { + claimError = error; + } + releaseSourceWrite(); + await writing; + const claim = await repo.claimMediaUsageCleanup({ + leaseToken: "source-write-claimant", + leaseDurationSeconds: 5 * 60, + nextEligibleDelaySeconds: 60, + sweepSafetyWindowSeconds: 60 * 60, + }); + + expect(claimError).toMatchObject({ code: "55P03" }); + expect(claim).not.toBeNull(); + }); + + it("takes the PostgreSQL cleanup lock before source deletion locks occurrences", async () => { + if (dialect !== "postgres") return; + + const current = await repo.replaceSource(contentSource("entry-delete-lock-order"), [ + occurrence("media-delete-lock-order"), + ]); + expect( + await repo.claimMediaUsageCleanup({ + leaseToken: "delete-lock-order-owner", + leaseDurationSeconds: 5 * 60, + nextEligibleDelaySeconds: 60, + sweepSafetyWindowSeconds: 60 * 60, + }), + ).not.toBeNull(); + await db + .updateTable("_emdash_media_usage") + .set({ cleanup_lease_token: "delete-lock-order-owner" }) + .where("source_key", "=", current.sourceKey) + .execute(); + + let releaseCleanupLock!: () => void; + const cleanupLockRelease = new Promise((resolve) => { + releaseCleanupLock = resolve; + }); + let signalCleanupLockAcquired!: () => void; + const cleanupLockAcquired = new Promise((resolve) => { + signalCleanupLockAcquired = resolve; + }); + const cleanupLockHolder = db.transaction().execute(async (trx) => { + await sql` + SELECT 1 + FROM _emdash_media_usage_cleanup + WHERE task_key = 'projection_gc' + FOR UPDATE + `.execute(trx); + signalCleanupLockAcquired(); + await cleanupLockRelease; + }); + await cleanupLockAcquired; + + let deletionFinished = false; + const deleting = repo.deleteSource(current.sourceKey).then((deleted) => { + deletionFinished = true; + return deleted; + }); + await sql`SELECT pg_sleep(0.1)`.execute(db); + const finishedWhileCleanupLockHeld = deletionFinished; + releaseCleanupLock(); + + expect(await deleting).toBe(1); + await cleanupLockHolder; + expect(finishedWhileCleanupLockHeld).toBe(false); + }); + + it("serializes previous-version PostgreSQL source deletion at the cleanup singleton", async () => { + if (dialect !== "postgres") return; + + const current = await repo.replaceSource(contentSource("entry-old-delete-lock-order"), [ + occurrence("media-old-delete-lock-order"), + ]); + expect( + await repo.claimMediaUsageCleanup({ + leaseToken: "old-delete-lock-order-owner", + leaseDurationSeconds: 5 * 60, + nextEligibleDelaySeconds: 60, + sweepSafetyWindowSeconds: 60 * 60, + }), + ).not.toBeNull(); + await db + .updateTable("_emdash_media_usage") + .set({ cleanup_lease_token: "old-delete-lock-order-owner" }) + .where("source_key", "=", current.sourceKey) + .execute(); + + let releaseCleanupLock!: () => void; + const cleanupLockRelease = new Promise((resolve) => { + releaseCleanupLock = resolve; + }); + let signalCleanupLockAcquired!: () => void; + const cleanupLockAcquired = new Promise((resolve) => { + signalCleanupLockAcquired = resolve; + }); + const cleanupLockHolder = db.transaction().execute(async (trx) => { + await sql` + SELECT 1 + FROM _emdash_media_usage_cleanup + WHERE task_key = 'projection_gc' + FOR UPDATE + `.execute(trx); + signalCleanupLockAcquired(); + await cleanupLockRelease; + }); + await cleanupLockAcquired; + + let deletionFinished = false; + const deleting = db + .transaction() + .execute(async (trx) => { + await trx + .deleteFrom("_emdash_media_usage_sources") + .where("source_key", "=", current.sourceKey) + .execute(); + await trx + .deleteFrom("_emdash_media_usage") + .where("source_key", "=", current.sourceKey) + .execute(); + }) + .then(() => { + deletionFinished = true; + return true; + }); + await sql`SELECT pg_sleep(0.1)`.execute(db); + const finishedWhileCleanupLockHeld = deletionFinished; + releaseCleanupLock(); + + await Promise.all([deleting, cleanupLockHolder]); + expect(finishedWhileCleanupLockHeld).toBe(false); + }); + + it("takes the PostgreSQL cleanup lock before source promotion locks the source row", async () => { + if (dialect !== "postgres") return; + + const current = await repo.replaceSource(contentSource("entry-promotion-lock-order"), [ + occurrence("media-before-promotion-lock"), + ]); + let signalCleanupLockAcquired!: () => void; + const cleanupLockAcquired = new Promise((resolve) => { + signalCleanupLockAcquired = resolve; + }); + let startSourceDelete!: () => void; + const sourceDeleteStart = new Promise((resolve) => { + startSourceDelete = resolve; + }); + const deleting = db.transaction().execute(async (trx) => { + await sql` + SELECT 1 + FROM _emdash_media_usage_cleanup + WHERE task_key = 'projection_gc' + FOR UPDATE + `.execute(trx); + signalCleanupLockAcquired(); + await sourceDeleteStart; + return trx + .deleteFrom("_emdash_media_usage_sources") + .where("source_key", "=", current.sourceKey) + .executeTakeFirst(); + }); + await cleanupLockAcquired; + + const replacing = repo.replaceSourceIfCurrent( + contentSource("entry-promotion-lock-order"), + [occurrence("media-after-promotion-lock")], + current.currentGeneration, + ); + await sql`SELECT pg_sleep(0.1)`.execute(db); + startSourceDelete(); + const outcomes = await Promise.allSettled([deleting, replacing]); + + expect(outcomes.every((outcome) => outcome.status === "fulfilled")).toBe(true); + if (outcomes[1]?.status === "fulfilled") { + expect(outcomes[1].value.replaced).toBe(false); + } + }); + + it("uses database time to reject delayed cleanup work after claim expiry", async () => { + const stale = await repo.replaceSource(contentSource("entry-delayed-cleanup"), [ + occurrence("media-before-expiry"), + ]); + await repo.replaceSource(contentSource("entry-delayed-cleanup"), [occurrence("media-current")]); + const staleOccurrence = await db + .selectFrom("_emdash_media_usage") + .select("id") + .where("generation", "=", stale.currentGeneration) + .executeTakeFirstOrThrow(); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("id", "=", staleOccurrence.id) + .execute(); + await db + .updateTable("_emdash_media_usage_cleanup") + .set({ + lease_token: "delayed-owner", + lease_expires_at: new Date(Date.now() - 1).toISOString(), + }) + .where("task_key", "=", "projection_gc") + .execute(); + + const delayedLease = { leaseToken: "delayed-owner" }; + const deleted = await repo.deleteStaleGenerationsOlderThan( + new Date(Date.now() + MEDIA_USAGE_CLEANUP_INTERVAL_MS).toISOString(), + 1, + { + candidateIds: [staleOccurrence.id], + cleanupLease: delayedLease, + }, + ); + + expect(deleted).toBe(0); + expect( + await db + .selectFrom("_emdash_media_usage") + .select("id") + .where("id", "=", staleOccurrence.id) + .executeTakeFirst(), + ).not.toBeNull(); + }); + + it("blocks a post-expiry claimant until an in-flight PostgreSQL delete completes", async () => { + if (dialect !== "postgres") return; + + const stale = await repo.replaceSource(contentSource("entry-postgres-delete-lock"), [ + occurrence("media-before-delete-lock"), + ]); + await repo.replaceSource(contentSource("entry-postgres-delete-lock"), [ + occurrence("media-current-delete-lock"), + ]); + const staleOccurrence = await db + .selectFrom("_emdash_media_usage") + .select("id") + .where("generation", "=", stale.currentGeneration) + .executeTakeFirstOrThrow(); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("id", "=", staleOccurrence.id) + .execute(); + await sql` + CREATE FUNCTION test_media_usage_cleanup_delete_delay() + RETURNS trigger AS $$ + BEGIN + PERFORM pg_sleep(0.8); + RETURN OLD; + END; + $$ LANGUAGE plpgsql + `.execute(db); + await sql` + CREATE TRIGGER test_media_usage_cleanup_delete_delay + BEFORE DELETE ON _emdash_media_usage + FOR EACH ROW EXECUTE FUNCTION test_media_usage_cleanup_delete_delay() + `.execute(db); + + const owner = await repo.claimMediaUsageCleanup({ + leaseToken: "delete-lock-owner", + leaseDurationSeconds: 5 * 60, + nextEligibleDelaySeconds: 60, + sweepSafetyWindowSeconds: 60 * 60, + }); + expect(owner).not.toBeNull(); + await sql` + UPDATE _emdash_media_usage_cleanup + SET lease_expires_at = to_char( + (clock_timestamp() AT TIME ZONE 'UTC') + INTERVAL '100 milliseconds', + 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"' + ), next_eligible_at = '1970-01-01T00:00:00.000Z' + WHERE task_key = 'projection_gc' + `.execute(db); + + const deleting = repo.deleteStaleGenerationsOlderThan("2100-01-01T00:00:00.000Z", 1, { + candidateIds: [staleOccurrence.id], + cleanupLease: { leaseToken: "delete-lock-owner" }, + }); + await sql`SELECT pg_sleep(0.2)`.execute(db); + let claimedBeforeDeleteFinished = false; + const nextClaim = repo + .claimMediaUsageCleanup({ + leaseToken: "post-expiry-claimant", + leaseDurationSeconds: 5 * 60, + nextEligibleDelaySeconds: 60, + sweepSafetyWindowSeconds: 60 * 60, + }) + .then((claim) => { + claimedBeforeDeleteFinished = true; + return claim; + }); + await sql`SELECT pg_sleep(0.2)`.execute(db); + + expect(claimedBeforeDeleteFinished).toBe(false); + expect(await deleting).toBe(1); + expect(await nextClaim).not.toBeNull(); + expect( + await db + .selectFrom("_emdash_media_usage_cleanup_fence") + .select("generation_floor") + .where("task_key", "=", "projection_gc") + .executeTakeFirstOrThrow(), + ).toEqual({ generation_floor: stale.currentGeneration }); + const promotion = await db + .updateTable("_emdash_media_usage_sources") + .set({ current_generation: stale.currentGeneration }) + .where("source_key", "=", stale.sourceKey) + .executeTakeFirst(); + expect(Number(promotion.numUpdatedRows ?? 0)).toBe(0); + }); + + it("stops deletion when a newer cleanup owner takes the lease", async () => { + const stale = await repo.replaceSource(contentSource("entry-cleanup-takeover"), [ + occurrence("media-stale"), + ]); + await repo.replaceSource(contentSource("entry-cleanup-takeover"), [ + occurrence("media-current"), + ]); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", stale.currentGeneration) + .execute(); + + const original = MediaUsageRepository.prototype.deleteStaleGenerationsOlderThan; + vi.spyOn(MediaUsageRepository.prototype, "deleteStaleGenerationsOlderThan").mockImplementation( + async function (this: MediaUsageRepository, cutoff, limit, options) { + await db + .updateTable("_emdash_media_usage_cleanup") + .set({ lease_token: "newer-owner", lease_expires_at: "2026-02-02T00:05:00.000Z" }) + .where("task_key", "=", "projection_gc") + .execute(); + return original.call(this, cutoff, limit, options); + }, + ); + + expect((await cleanupMediaUsage(db)).deletedRows).toBe(0); + expect( + await db + .selectFrom("_emdash_media_usage") + .select("id") + .where("generation", "=", stale.currentGeneration) + .execute(), + ).toHaveLength(1); + }); + + it("revalidates a candidate when concurrent publication makes it current", async () => { + const stale = await repo.replaceSource(contentSource("entry-publication-race"), [ + occurrence("media-before-publication"), + ]); + await repo.replaceSource(contentSource("entry-publication-race"), [ + occurrence("media-after-publication"), + ]); + await db + .updateTable("_emdash_media_usage") + .set({ created_at: "2026-02-01T19:00:00.000Z" }) + .where("generation", "=", stale.currentGeneration) + .execute(); + + const original = MediaUsageRepository.prototype.deleteStaleGenerationsOlderThan; + vi.spyOn(MediaUsageRepository.prototype, "deleteStaleGenerationsOlderThan").mockImplementation( + async function (this: MediaUsageRepository, cutoff, limit, options) { + await db + .updateTable("_emdash_media_usage_sources") + .set({ current_generation: stale.currentGeneration }) + .where("source_key", "=", stale.sourceKey) + .execute(); + return original.call(this, cutoff, limit, options); + }, + ); + + expect((await cleanupMediaUsage(db)).deletedRows).toBe(0); + expect(await repo.findCurrentUsageByMediaId("media-before-publication")).toHaveLength(1); + }); + + it("does not report completion after its cleanup claim is fenced by a newer owner", async () => { + const original = MediaUsageRepository.prototype.completeMediaUsageCleanup; + vi.spyOn(MediaUsageRepository.prototype, "completeMediaUsageCleanup").mockImplementation( + async function (this: MediaUsageRepository, input) { + await db + .updateTable("_emdash_media_usage_cleanup") + .set({ lease_token: "newer-owner", lease_expires_at: "2026-02-02T00:05:00.000Z" }) + .where("task_key", "=", "projection_gc") + .execute(); + return original.call(this, input); + }, + ); + + expect((await cleanupMediaUsage(db)).status).toBe("skipped"); + const state = await db + .selectFrom("_emdash_media_usage_cleanup") + .select("lease_token") + .where("task_key", "=", "projection_gc") + .executeTakeFirstOrThrow(); + expect(state.lease_token).toBe("newer-owner"); + }); +}); + +function contentSource(contentId: string) { + return { + sourceKey: `content:posts:${contentId}:columns`, + sourceType: "content", + collectionSlug: "posts", + contentId, + sourceVariant: "columns" as const, + contentStatus: "published", + }; +} + +function occurrence(mediaId: string, fieldPath = "hero") { + return { + fieldSlug: fieldPath, + fieldPath, + referenceType: "image_field" as const, + mediaId, + provider: "local", + providerAssetId: mediaId, + }; +} + +async function insertOccurrence( + db: Kysely, + input: { + id: string; + sourceKey: string; + generation: string; + mediaId: string; + createdAt: string; + }, +): Promise { + await db + .insertInto("_emdash_media_usage") + .values({ + id: input.id, + source_key: input.sourceKey, + generation: input.generation, + field_slug: "hero", + field_path: input.id, + occurrence_index: 0, + reference_type: "image_field", + media_id: input.mediaId, + provider: "local", + provider_asset_id: input.mediaId, + media_kind: "image", + mime_type: null, + created_at: input.createdAt, + }) + .execute(); +} + +async function makeCleanupEligible(db: Kysely): Promise { + await db + .updateTable("_emdash_media_usage_cleanup") + .set({ next_eligible_at: "1970-01-01T00:00:00.000Z" }) + .where("task_key", "=", "projection_gc") + .execute(); +} + +async function databaseNowTimestamp( + db: Kysely, + dialect: DialectTestContext["dialect"], +): Promise { + if (dialect === "postgres") { + const { rows } = await sql<{ value: string }>` + SELECT to_char( + clock_timestamp() AT TIME ZONE 'UTC', + 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"' + ) AS value + `.execute(db); + return rows[0]!.value; + } + const { rows } = await sql<{ value: string }>` + SELECT strftime('%Y-%m-%dT%H:%M:%fZ', 'now') AS value + `.execute(db); + return rows[0]!.value; +} diff --git a/packages/core/tests/integration/database/media-usage-migration.test.ts b/packages/core/tests/integration/database/media-usage-migration.test.ts index 9bada5dfb8..32ff9d4cfb 100644 --- a/packages/core/tests/integration/database/media-usage-migration.test.ts +++ b/packages/core/tests/integration/database/media-usage-migration.test.ts @@ -18,6 +18,7 @@ const EXPECTED_INDEXES = [ "idx__emdash_media_usage_provider_asset", "idx__emdash_media_usage_source_generation", "idx__emdash_media_usage_unique_occurrence", + "idx__emdash_media_usage_cleanup_scan", ] as const; describeEachDialect("media usage index migration", (dialect) => { diff --git a/packages/core/tests/integration/database/media-usage-repository.test.ts b/packages/core/tests/integration/database/media-usage-repository.test.ts index 48ba225512..60c9650b57 100644 --- a/packages/core/tests/integration/database/media-usage-repository.test.ts +++ b/packages/core/tests/integration/database/media-usage-repository.test.ts @@ -925,6 +925,23 @@ describeEachDialect("MediaUsageRepository", (dialect) => { expect(await repo.findCurrentUsageByMediaId("media-live")).toHaveLength(1); }); + it("does not promote a generation reclaimed during a D1-style write", async () => { + await repo.replaceSource(contentSource("entry1", "columns"), [ + occurrence("hero", "media-before-race"), + ]); + await installCleanupBeforePromotionTrigger(ctx); + vi.resetModules(); + const { MediaUsageRepository: D1LikeMediaUsageRepository } = + await import("../../../src/database/repositories/media-usage.js"); + const d1LikeRepo = new D1LikeMediaUsageRepository(withoutTransactions(ctx.db)); + + await d1LikeRepo.replaceSource(contentSource("entry1", "columns"), [ + occurrence("hero", "media-after-race"), + ]); + + expect(await repo.findCurrentUsageByMediaId("media-after-race")).toHaveLength(1); + }); + it("deletes content sources by collection", async () => { await repo.replaceSource(contentSource("entry1", "columns"), [ occurrence("hero", "media-live"), @@ -1463,3 +1480,50 @@ async function installSourceDeleteFailureTrigger(ctx: DialectTestContext): Promi END `.execute(ctx.db); } + +async function installCleanupBeforePromotionTrigger(ctx: DialectTestContext): Promise { + if (ctx.dialect === "postgres") { + await sql` + CREATE FUNCTION media_usage_cleanup_before_promotion() + RETURNS trigger + LANGUAGE plpgsql + AS $$ + BEGIN + DELETE FROM _emdash_media_usage AS usage + WHERE usage.source_key = NEW.source_key + AND usage.generation = NEW.current_generation + AND NOT EXISTS ( + SELECT 1 + FROM _emdash_media_usage_generation_writes AS writer + WHERE writer.source_key = usage.source_key + AND writer.generation = usage.generation + ); + RETURN NEW; + END; + $$ + `.execute(ctx.db); + await sql` + CREATE TRIGGER media_usage_cleanup_before_promotion + BEFORE UPDATE OF current_generation ON _emdash_media_usage_sources + FOR EACH ROW + EXECUTE FUNCTION media_usage_cleanup_before_promotion() + `.execute(ctx.db); + return; + } + + await sql` + CREATE TRIGGER media_usage_cleanup_before_promotion + BEFORE UPDATE OF current_generation ON _emdash_media_usage_sources + BEGIN + DELETE FROM _emdash_media_usage + WHERE source_key = NEW.source_key + AND generation = NEW.current_generation + AND NOT EXISTS ( + SELECT 1 + FROM _emdash_media_usage_generation_writes AS writer + WHERE writer.source_key = _emdash_media_usage.source_key + AND writer.generation = _emdash_media_usage.generation + ); + END + `.execute(ctx.db); +} diff --git a/packages/core/tests/integration/database/migrations.test.ts b/packages/core/tests/integration/database/migrations.test.ts index ce2e6de4c3..eb78c4edfa 100644 --- a/packages/core/tests/integration/database/migrations.test.ts +++ b/packages/core/tests/integration/database/migrations.test.ts @@ -51,6 +51,9 @@ describe("Database Migrations (Integration)", () => { "_emdash_byline_field_group_values", "_emdash_media_usage_sources", "_emdash_media_usage", + "_emdash_media_usage_cleanup", + "_emdash_media_usage_generation_writes", + "_emdash_media_usage_cleanup_fence", "_emdash_media_usage_index_status", ]; @@ -141,6 +144,8 @@ describe("Database Migrations (Integration)", () => { "052_media_usage_read_index", "053_plugin_mcp_tools", "054_media_upload_attempts", + "055_media_usage_cleanup", + "056_media_usage_cleanup_fence", ]; await db.deleteFrom("_emdash_migrations").where("name", "in", trailing).execute();