diff --git a/.changeset/calm-mice-admit.md b/.changeset/calm-mice-admit.md new file mode 100644 index 0000000000..c80cde1154 --- /dev/null +++ b/.changeset/calm-mice-admit.md @@ -0,0 +1,5 @@ +--- +"emdash": patch +--- + +Adds resource limits to durable Media Usage indexing so already-oversized changed projections are rejected before the worker updates their index. diff --git a/packages/core/package.json b/packages/core/package.json index 644565052f..d007b89f1b 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -205,6 +205,7 @@ "typecheck": "tsgo --noEmit", "check": "publint && attw --pack --ignore-rules=cjs-resolves-to-esm --ignore-rules=no-resolution --ignore-rules=internal-resolution-error", "test": "vitest", + "test:workerd": "vitest run --config vitest.workerd.config.ts", "test:smoke": "vitest run --config vitest.smoke.config.ts", "test:integration": "vitest run --config vitest.integration.config.ts", "test:repro": "vitest run --config vitest.repro.config.ts" @@ -278,6 +279,7 @@ "devDependencies": { "@apidevtools/swagger-parser": "^12.1.0", "@arethetypeswrong/cli": "catalog:", + "@cloudflare/vitest-pool-workers": "catalog:", "@emdash-cms/blocks": "workspace:*", "@types/better-sqlite3": "^7.6.12", "@types/pg": "^8.16.0", diff --git a/packages/core/src/database/repositories/media-usage.ts b/packages/core/src/database/repositories/media-usage.ts index f958cc2c45..657d431b9f 100644 --- a/packages/core/src/database/repositories/media-usage.ts +++ b/packages/core/src/database/repositories/media-usage.ts @@ -33,7 +33,6 @@ type MediaUsageSourceNullableStringColumn = | "updated_at" | "last_attempted_at" | "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( @@ -184,6 +183,12 @@ export interface MediaUsageGuardedAttemptResult { source: MediaUsageSource | null; } +export interface MediaUsageSourceGenerationDeletionMeasurement { + occurrenceCount: number; + occurrenceBytes: number; + exceedsOccurrenceLimit: boolean; +} + export interface MediaUsageCleanupCursor { createdAt: string; id: string; @@ -537,6 +542,38 @@ export class MediaUsageRepository { return sources; } + async measureSourceGenerationDeletion( + sourceKey: string, + generation: string, + maxOccurrences: number, + ): Promise { + if (!Number.isSafeInteger(maxOccurrences) || maxOccurrences < 0) { + throw new Error("Media usage deletion measurement requires a non-negative row limit"); + } + const payload = sql` + COALESCE(field_slug, '') || COALESCE(field_path, '') || + COALESCE(reference_type, '') || COALESCE(media_id, '') || + COALESCE(provider, '') || COALESCE(provider_asset_id, '') || + COALESCE(media_kind, '') || COALESCE(mime_type, '') + `; + const occurrenceBytes = isPostgres(this.db) + ? sql`octet_length(${payload})` + : sql`length(CAST(${payload} AS BLOB))`; + const rows = await this.db + .selectFrom("_emdash_media_usage") + .select(occurrenceBytes.as("occurrence_bytes")) + .where("source_key", "=", sourceKey) + .where("generation", "=", generation) + .limit(maxOccurrences + 1) + .execute(); + + return { + occurrenceCount: rows.length, + occurrenceBytes: rows.reduce((total, row) => total + Number(row.occurrence_bytes), 0), + exceedsOccurrenceLimit: rows.length > maxOccurrences, + }; + } + async replaceSourceIfMatching( source: MediaUsageSourceInput, occurrences: readonly MediaUsageOccurrenceInput[], diff --git a/packages/core/src/media/usage/content-refresh.ts b/packages/core/src/media/usage/content-refresh.ts index e0d93cf4c3..6f648c425d 100644 --- a/packages/core/src/media/usage/content-refresh.ts +++ b/packages/core/src/media/usage/content-refresh.ts @@ -1,7 +1,10 @@ import { sql, type Kysely } from "kysely"; import { tableExists } from "../../database/dialect-helpers.js"; -import { MediaUsageRepository } from "../../database/repositories/media-usage.js"; +import { + MediaUsageRepository, + type MediaUsageSource, +} from "../../database/repositories/media-usage.js"; import type { Database } from "../../database/types.js"; import { validateIdentifier } from "../../database/validate.js"; import { isI18nEnabled } from "../../i18n/config.js"; @@ -9,6 +12,7 @@ import { loadContentMediaUsageFields } from "./content-fields.js"; import { CONTENT_SOURCE_SCHEMA_VERSION, loadContentMediaUsageSnapshots, + type ContentMediaUsageSnapshot, } from "./content-snapshots.js"; import { buildContentMediaUsageSourceKey, @@ -22,6 +26,28 @@ const CONTENT_USAGE_LOCKS_KEY = Symbol.for("emdash.mediaUsage.contentLocks"); const CONTENT_USAGE_COLLECTION_LOCKS_KEY = Symbol.for("emdash.mediaUsage.collectionLocks"); const CONTENT_USAGE_REFRESH_MAX_ATTEMPTS = 2; +export const MEDIA_USAGE_PROJECTION_ADMISSION_LIMITS = Object.freeze({ + maxOccurrenceMutationUnitsPerClaim: 12, + maxProjectionMutationBytesPerClaim: 512 * 1024, +}); + +export interface ContentMediaUsageAdmissionBudget { + remainingOccurrenceMutationUnits: number; + remainingProjectionMutationBytes: number; + hasReservedMutation: boolean; +} + +export type ContentMediaUsageProjectionAdmissionResult = + | { + outcome: "admitted"; + noOpSourceKeys: ReadonlySet; + absentSources: MediaUsageSource[]; + occurrenceMutationUnits: number; + projectionMutationBytes: number; + } + | { outcome: "intrinsic_resource_limit" } + | { outcome: "claim_budget_deferred" }; + // These maps only de-dupe usage work inside the current isolate/process. Cross-worker // correctness comes from expected-generation guards on repository writes. @@ -33,11 +59,13 @@ export type ContentMediaUsageRefreshErrorCode = | "CONTENT_USAGE_REFRESH_ERROR" | "CONTENT_USAGE_DELETE_ERROR" | "CONTENT_USAGE_GENERATION_CONFLICT" + | "CONTENT_USAGE_RESOURCE_LIMIT" | "CONTENT_USAGE_STALE"; interface ContentMediaUsageRefreshOptions { collectionId?: string; durableWork?: boolean; + admissionBudget?: ContentMediaUsageAdmissionBudget; } export interface ContentMediaUsageRefreshResult { @@ -55,6 +83,133 @@ const ZERO_RESULT: ContentMediaUsageRefreshResult = { failedSourceCount: 0, }; +export function createContentMediaUsageAdmissionBudget(): ContentMediaUsageAdmissionBudget { + return { + remainingOccurrenceMutationUnits: + MEDIA_USAGE_PROJECTION_ADMISSION_LIMITS.maxOccurrenceMutationUnitsPerClaim, + remainingProjectionMutationBytes: + MEDIA_USAGE_PROJECTION_ADMISSION_LIMITS.maxProjectionMutationBytesPerClaim, + hasReservedMutation: false, + }; +} + +export async function planContentMediaUsageProjectionAdmission( + repo: MediaUsageRepository, + snapshots: readonly ContentMediaUsageSnapshot[], + observedSources: ReadonlyMap, + canonicalSourceKeys: readonly string[], + budget: ContentMediaUsageAdmissionBudget, +): Promise { + const snapshotSourceKeys = new Set(snapshots.map((snapshot) => snapshot.source.sourceKey)); + const absentSources = canonicalSourceKeys + .filter((sourceKey) => !snapshotSourceKeys.has(sourceKey)) + .map((sourceKey) => observedSources.get(sourceKey)) + .filter((source): source is MediaUsageSource => source !== undefined); + let deletionOccurrenceUnits = 0; + let deletionBytes = 0; + for (const source of absentSources) { + const measurement = await repo.measureSourceGenerationDeletion( + source.sourceKey, + source.currentGeneration, + MEDIA_USAGE_PROJECTION_ADMISSION_LIMITS.maxOccurrenceMutationUnitsPerClaim, + ); + if (measurement.exceedsOccurrenceLimit) { + return budget.hasReservedMutation + ? { outcome: "claim_budget_deferred" } + : { outcome: "intrinsic_resource_limit" }; + } + deletionOccurrenceUnits += measurement.occurrenceCount; + deletionBytes += storedMediaUsageSourceByteLength(source) + measurement.occurrenceBytes * 2; + } + + const noOpSourceKeys = new Set(); + let cost = projectionAdmissionCost( + snapshots, + noOpSourceKeys, + deletionOccurrenceUnits, + deletionBytes, + ); + if (exceedsProjectionAdmissionLimits(cost)) { + for (const snapshot of snapshots) { + const expectedSource = observedSources.get(snapshot.source.sourceKey); + if ( + expectedSource && + (await repo.projectionMatchesExpectedSource(snapshot.source, expectedSource)) + ) { + noOpSourceKeys.add(snapshot.source.sourceKey); + } + } + cost = projectionAdmissionCost( + snapshots, + noOpSourceKeys, + deletionOccurrenceUnits, + deletionBytes, + ); + } + + if (exceedsProjectionAdmissionLimits(cost)) { + return budget.hasReservedMutation + ? { outcome: "claim_budget_deferred" } + : { outcome: "intrinsic_resource_limit" }; + } + if ( + cost.occurrenceMutationUnits > budget.remainingOccurrenceMutationUnits || + cost.projectionMutationBytes > budget.remainingProjectionMutationBytes + ) { + return { outcome: "claim_budget_deferred" }; + } + + budget.remainingOccurrenceMutationUnits -= cost.occurrenceMutationUnits; + budget.remainingProjectionMutationBytes -= cost.projectionMutationBytes; + if (cost.occurrenceMutationUnits > 0 || cost.projectionMutationBytes > 0) { + budget.hasReservedMutation = true; + } + return { + outcome: "admitted", + noOpSourceKeys, + absentSources, + ...cost, + }; +} + +interface ProjectionAdmissionCost { + occurrenceMutationUnits: number; + projectionMutationBytes: number; +} + +function projectionAdmissionCost( + snapshots: readonly ContentMediaUsageSnapshot[], + noOpSourceKeys: ReadonlySet, + deletionOccurrenceUnits: number, + deletionBytes: number, +): ProjectionAdmissionCost { + return snapshots.reduce( + (cost, snapshot) => { + if (noOpSourceKeys.has(snapshot.source.sourceKey)) return cost; + cost.occurrenceMutationUnits += snapshot.occurrences.length; + cost.projectionMutationBytes += snapshot.projectionByteLength; + return cost; + }, + { + occurrenceMutationUnits: deletionOccurrenceUnits, + projectionMutationBytes: deletionBytes, + }, + ); +} + +function exceedsProjectionAdmissionLimits(cost: ProjectionAdmissionCost): boolean { + return ( + cost.occurrenceMutationUnits > + MEDIA_USAGE_PROJECTION_ADMISSION_LIMITS.maxOccurrenceMutationUnitsPerClaim || + cost.projectionMutationBytes > + MEDIA_USAGE_PROJECTION_ADMISSION_LIMITS.maxProjectionMutationBytesPerClaim + ); +} + +function storedMediaUsageSourceByteLength(source: MediaUsageSource): number { + return new TextEncoder().encode(JSON.stringify(source)).byteLength; +} + export async function refreshContentMediaUsage( db: Kysely, collectionSlug: string, @@ -94,10 +249,12 @@ async function refreshContentMediaUsageUnlocked( ): Promise { try { let conflictResult: ContentMediaUsageRefreshResult | null = null; + if (options.durableWork) options.admissionBudget = createContentMediaUsageAdmissionBudget(); for (let attempt = 0; attempt < CONTENT_USAGE_REFRESH_MAX_ATTEMPTS; attempt++) { const result = await refreshContentMediaUsageAttempt(db, collectionSlug, contentId, options); if (result.errorCode !== "CONTENT_USAGE_GENERATION_CONFLICT") return result; conflictResult = result; + if (options.admissionBudget?.hasReservedMutation) break; } if (options.durableWork) { @@ -136,12 +293,8 @@ async function refreshContentMediaUsageAttempt( options: ContentMediaUsageRefreshOptions, ): Promise { const repo = new MediaUsageRepository(db); - const observedSources = await loadObservedContentSources( - repo, - collectionSlug, - contentId, - options.collectionId, - ); + const canonicalSourceKeys = contentSourceKeys(collectionSlug, contentId, options.collectionId); + const observedSources = await repo.findSources(canonicalSourceKeys); const snapshotsResult = await loadContentMediaUsageSnapshots( db, collectionSlug, @@ -154,9 +307,19 @@ async function refreshContentMediaUsageAttempt( if (!(await contentCollectionExists(db, collectionSlug, options.collectionId))) { return generationConflictResult({ refreshedSourceCount: 0, deletedSourceCount: 0 }); } - return deleteCanonicalContentSourcesIfAbsent( + if (!options.admissionBudget) + throw new Error("Durable media usage work requires an admission budget"); + const admission = await planContentMediaUsageProjectionAdmission( repo, + [], observedSources, + canonicalSourceKeys, + options.admissionBudget, + ); + if (admission.outcome !== "admitted") return admissionFailureResult(admission.outcome); + return deleteCanonicalContentSourcesIfAbsent( + repo, + admission.absentSources, collectionSlug, contentId, ); @@ -180,8 +343,27 @@ async function refreshContentMediaUsageAttempt( const deletedSourceCount = await repo.deleteContentSources(collectionSlug, contentId); return { ...ZERO_RESULT, deletedSourceCount }; } + const admission = options.admissionBudget + ? await planContentMediaUsageProjectionAdmission( + repo, + snapshotsResult.snapshots, + observedSources, + canonicalSourceKeys, + options.admissionBudget, + ) + : null; + if (admission && admission.outcome !== "admitted") { + return admissionFailureResult(admission.outcome); + } let refreshedSourceCount = 0; for (const snapshot of snapshotsResult.snapshots) { + if ( + admission?.outcome === "admitted" && + admission.noOpSourceKeys.has(snapshot.source.sourceKey) + ) { + refreshedSourceCount++; + continue; + } const result = await repo.replaceSourceIfMatching( snapshot.source, snapshot.occurrences, @@ -210,20 +392,16 @@ async function refreshContentMediaUsageAttempt( const expectedSourceKeys = new Set( snapshotsResult.snapshots.map((snapshot) => snapshot.source.sourceKey), ); - const absentSourceKeys = MEDIA_USAGE_CONTENT_SOURCE_VARIANTS.map((sourceVariant) => - buildContentMediaUsageSourceKey({ - collectionId: options.collectionId, - collectionSlug, - contentId, - sourceVariant, - }), - ).filter((sourceKey) => !expectedSourceKeys.has(sourceKey)); + const absentSources = + admission?.outcome === "admitted" + ? admission.absentSources + : canonicalSourceKeys + .filter((sourceKey) => !expectedSourceKeys.has(sourceKey)) + .map((sourceKey) => observedSources.get(sourceKey)) + .filter((source): source is MediaUsageSource => source !== undefined); let deletedSourceCount = 0; - for (const sourceKey of absentSourceKeys) { - const expectedSource = observedSources.get(sourceKey); - if (!expectedSource) continue; - - const result = await repo.deleteSourceIfMatching(sourceKey, expectedSource); + for (const expectedSource of absentSources) { + const result = await repo.deleteSourceIfMatching(expectedSource.sourceKey, expectedSource); if (result.deleted) { deletedSourceCount++; continue; @@ -244,13 +422,12 @@ async function refreshContentMediaUsageAttempt( }; } -async function loadObservedContentSources( - repo: MediaUsageRepository, +function contentSourceKeys( collectionSlug: string, contentId: string, collectionId?: string, -): Promise>> { - const sourceKeys = MEDIA_USAGE_CONTENT_SOURCE_VARIANTS.map((sourceVariant) => +): string[] { + return MEDIA_USAGE_CONTENT_SOURCE_VARIANTS.map((sourceVariant) => buildContentMediaUsageSourceKey({ collectionId, collectionSlug, @@ -258,7 +435,18 @@ async function loadObservedContentSources( sourceVariant, }), ); - return repo.findSources(sourceKeys); +} + +function admissionFailureResult( + outcome: "intrinsic_resource_limit" | "claim_budget_deferred", +): ContentMediaUsageRefreshResult { + return { + ...generationConflictResult({ refreshedSourceCount: 0, deletedSourceCount: 0 }), + errorCode: + outcome === "intrinsic_resource_limit" + ? "CONTENT_USAGE_RESOURCE_LIMIT" + : "CONTENT_USAGE_GENERATION_CONFLICT", + }; } async function markGenerationConflict( @@ -540,12 +728,12 @@ function snapshotFailureResult( async function deleteCanonicalContentSourcesIfAbsent( repo: MediaUsageRepository, - observedSources: Awaited>, + observedSources: readonly MediaUsageSource[], collectionSlug: string, contentId: string, ): Promise { let deletedSourceCount = 0; - for (const source of observedSources.values()) { + for (const source of observedSources) { const result = await repo.deleteSourceIfMatchingContentAbsent( source.sourceKey, source, diff --git a/packages/core/src/media/usage/content-snapshots.ts b/packages/core/src/media/usage/content-snapshots.ts index 1e8d76b19f..0011ab2bde 100644 --- a/packages/core/src/media/usage/content-snapshots.ts +++ b/packages/core/src/media/usage/content-snapshots.ts @@ -54,6 +54,7 @@ export interface ContentMediaUsageSnapshot { source: MediaUsageSourceInput; occurrences: MediaUsageOccurrenceInput[]; fields: readonly ContentMediaUsageField[]; + projectionByteLength: number; } export interface LoadContentMediaUsageSnapshotsOptions { @@ -106,17 +107,19 @@ export async function loadContentMediaUsageSnapshots( sourceVariant: "columns", revisionId: columnsRevisionId, }); - columnsSource.sourceFingerprint = await buildMediaUsageProjectionFingerprint({ + const columnsProjection = await buildMediaUsageProjectionFingerprint({ collectionId, source: columnsSource, occurrences, extractionFields: discovery.extractionFields, }); + columnsSource.sourceFingerprint = columnsProjection.fingerprint; const snapshots: ContentMediaUsageSnapshot[] = [ { source: columnsSource, occurrences, fields: discovery.extractionFields, + projectionByteLength: columnsProjection.byteLength, }, ]; @@ -180,16 +183,18 @@ export async function loadContentMediaUsageSnapshots( revisionId: draftRevisionId, contentSlug: draftContentSlug, }); - draftSource.sourceFingerprint = await buildMediaUsageProjectionFingerprint({ + const draftProjection = await buildMediaUsageProjectionFingerprint({ collectionId, source: draftSource, occurrences: draftOccurrences, extractionFields: discovery.extractionFields, }); + draftSource.sourceFingerprint = draftProjection.fingerprint; snapshots.push({ source: draftSource, occurrences: draftOccurrences, fields: discovery.extractionFields, + projectionByteLength: draftProjection.byteLength, }); } diff --git a/packages/core/src/media/usage/projection-fingerprint.ts b/packages/core/src/media/usage/projection-fingerprint.ts index e609593dbf..bb40063e9b 100644 --- a/packages/core/src/media/usage/projection-fingerprint.ts +++ b/packages/core/src/media/usage/projection-fingerprint.ts @@ -15,9 +15,14 @@ export interface MediaUsageProjectionFingerprintInput { extractionFields: readonly MediaUsageExtractionField[]; } +export interface MediaUsageProjectionFingerprint { + fingerprint: string; + byteLength: number; +} + export async function buildMediaUsageProjectionFingerprint( input: MediaUsageProjectionFingerprintInput, -): Promise { +): Promise { if (!input.collectionId) { throw new Error("Media usage projection fingerprints require a collection identity"); } @@ -59,11 +64,15 @@ export async function buildMediaUsageProjectionFingerprint( }, occurrences: canonicalOccurrences, }); - const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(payload)); + const encodedPayload = new TextEncoder().encode(payload); + const digest = await crypto.subtle.digest("SHA-256", encodedPayload); const hex = Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join( "", ); - return `${FINGERPRINT_PREFIX}${hex}`; + return { + fingerprint: `${FINGERPRINT_PREFIX}${hex}`, + byteLength: encodedPayload.byteLength, + }; } function normalizeExtractionFields( diff --git a/packages/core/src/media/usage/work-processor.ts b/packages/core/src/media/usage/work-processor.ts index ceefbfe52d..6ef62480c3 100644 --- a/packages/core/src/media/usage/work-processor.ts +++ b/packages/core/src/media/usage/work-processor.ts @@ -166,7 +166,10 @@ async function processCandidate( } const errorCode = processingErrorCode(refresh.errorCode); - if (claimed.attemptCount + 1 >= MEDIA_USAGE_WORK_PROCESSING_LIMITS.maxAttempts) { + const terminal = + errorCode === "MEDIA_USAGE_RESOURCE_LIMIT" || + claimed.attemptCount + 1 >= MEDIA_USAGE_WORK_PROCESSING_LIMITS.maxAttempts; + if (terminal) { const failed = await repo.failWork({ ...lease, errorCode }); if (failed) { await new MediaUsageRepository(db).recordIncrementalFailure({ @@ -240,5 +243,6 @@ function processingErrorCode(errorCode: ContentMediaUsageRefreshErrorCode | unde if (errorCode === "CONTENT_USAGE_GENERATION_CONFLICT") { return "MEDIA_USAGE_GENERATION_CONFLICT"; } + if (errorCode === "CONTENT_USAGE_RESOURCE_LIMIT") return "MEDIA_USAGE_RESOURCE_LIMIT"; return "MEDIA_USAGE_PROCESSING_FAILED"; } diff --git a/packages/core/tests/integration/database/media-usage-content-snapshots.test.ts b/packages/core/tests/integration/database/media-usage-content-snapshots.test.ts index 86c27b1e68..9467cce9a3 100644 --- a/packages/core/tests/integration/database/media-usage-content-snapshots.test.ts +++ b/packages/core/tests/integration/database/media-usage-content-snapshots.test.ts @@ -440,6 +440,10 @@ describeEachDialect("content media usage snapshots", (dialect) => { expect(firstOverlay.source.sourceFingerprint).not.toBe(firstColumns.source.sourceFingerprint); expect(secondColumns.source.sourceFingerprint).toBe(firstColumns.source.sourceFingerprint); expect(secondOverlay.source.sourceFingerprint).toBe(firstOverlay.source.sourceFingerprint); + expect(firstColumns.projectionByteLength).toBeGreaterThan(0); + expect(firstOverlay.projectionByteLength).toBeGreaterThan(0); + expect(secondColumns.projectionByteLength).toBe(firstColumns.projectionByteLength); + expect(secondOverlay.projectionByteLength).toBe(firstOverlay.projectionByteLength); }); it("changes fingerprints when V1-visible source metadata changes", async () => { diff --git a/packages/core/tests/integration/database/media-usage-projection-admission-measurement.test.ts b/packages/core/tests/integration/database/media-usage-projection-admission-measurement.test.ts new file mode 100644 index 0000000000..5b34c9724e --- /dev/null +++ b/packages/core/tests/integration/database/media-usage-projection-admission-measurement.test.ts @@ -0,0 +1,295 @@ +import Database from "better-sqlite3"; +import { Kysely, SqliteDialect, sql } from "kysely"; +import { afterAll, beforeAll, expect, it, vi } from "vitest"; + +vi.mock( + "virtual:emdash/object-cache", + () => ({ createObjectCache: undefined, objectCacheConfig: {} }), + { virtual: true }, +); + +import { runMigrations } from "../../../src/database/migrations/runner.js"; +import type { Database as DatabaseSchema } from "../../../src/database/types.js"; +import { waitForDeferredTasks } from "../../../src/deferred-tasks.js"; +import { setI18nConfig } from "../../../src/i18n/config.js"; +import { buildContentMediaUsageSourceKey } from "../../../src/media/usage/source-key.js"; +import { processMediaUsageWorkAfterWrite } from "../../../src/media/usage/work-processor.js"; +import { createTestRuntime } from "../../utils/mcp-runtime.js"; +import { + addMediaUsageMeasurementDraft, + createMediaUsageAdmissionFixture, + insertMediaUsageMeasurementEntry, + mediaUsageMeasurementData as mediaData, +} from "../../utils/media-usage-admission-fixture.js"; + +interface CapturedQuery { + sql: string; + parameters: readonly unknown[]; +} + +interface Measurement { + d1Queries: number; + maxBinds: number; + maxSqlBytes: number; + changedRows: number; + durationMs: number; +} + +interface MeasurementRow extends Measurement { + path: "processor" | "processor-bytes" | "runtime-conflict" | "runtime-create" | "runtime-update"; + totalOccurrences: number; + liveOccurrences: number; + draftOccurrences: number | null; + outcome: string; + payloadBytes?: number; +} + +let sqlite: Database.Database; +let db: Kysely; +let captured: CapturedQuery[]; +let fixture: Awaited>; + +beforeAll(async () => { + setI18nConfig(null); + captured = []; + sqlite = new Database(":memory:"); + db = new Kysely({ + dialect: new SqliteDialect({ database: sqlite }), + log(event) { + if (event.level === "query") { + captured.push({ sql: event.query.sql, parameters: event.query.parameters }); + } + }, + }); + await runMigrations(db); + fixture = await createActiveFixture(); +}); + +afterAll(async () => { + setI18nConfig(null); + await db.destroy(); +}); + +it("reports current projection costs at occurrence and byte boundaries", async () => { + const rows: MeasurementRow[] = []; + for (const totalOccurrences of [0, 1, 3, 4, 6, 9, 12, 15, 18, 21, 24, 27, 30]) { + for (const [liveOccurrences, draftOccurrences] of candidateSplits(totalOccurrences)) { + const contentId = `processor-${totalOccurrences}-${liveOccurrences}-${draftOccurrences ?? "none"}`; + await insertEntry(contentId, mediaData(liveOccurrences, `${contentId}-live`)); + if (draftOccurrences !== null) { + await addDraft(contentId, mediaData(draftOccurrences, `${contentId}-draft`)); + } + + const result = await measure(() => + processMediaUsageWorkAfterWrite(db, fixture.collectionSlug, contentId), + ); + const expectedOutcome = totalOccurrences <= 12 ? "completed" : "failed"; + expect(result.value.outcome).toBe(expectedOutcome); + rows.push({ + path: "processor", + totalOccurrences, + liveOccurrences, + draftOccurrences, + outcome: result.value.outcome, + ...result.measurement, + }); + } + } + for (const payloadBytes of [64, 128, 256, 512, 768, 1024].map((size) => size * 1024)) { + const contentId = `bytes-${payloadBytes}`; + const title = "é".repeat(payloadBytes / 2); + await insertEntry(contentId, mediaData(0, contentId), title); + const result = await measure(() => + processMediaUsageWorkAfterWrite(db, fixture.collectionSlug, contentId), + ); + const expectedOutcome = payloadBytes < 512 * 1024 ? "completed" : "failed"; + expect(result.value.outcome).toBe(expectedOutcome); + rows.push({ + path: "processor-bytes", + totalOccurrences: 0, + liveOccurrences: 0, + draftOccurrences: null, + payloadBytes, + outcome: result.value.outcome, + ...result.measurement, + }); + } + + const runtime = createTestRuntime(db); + for (const totalOccurrences of [0, 3, 6, 9, 12, 15, 18, 21, 24, 27, 30]) { + const createSlug = `runtime-create-${totalOccurrences}`; + const created = await measure(() => + runtime.handleContentCreate(fixture.collectionSlug, { + slug: createSlug, + data: { title: createSlug, ...mediaData(totalOccurrences, createSlug) }, + }), + ); + expect(created.value.success).toBe(true); + if (!created.value.success) throw new Error(created.value.error.message); + const createOutcome = await workOutcome(created.value.data.item.id); + expect(createOutcome).toBe(totalOccurrences <= 12 ? "completed" : "failed"); + rows.push({ + path: "runtime-create", + totalOccurrences, + liveOccurrences: totalOccurrences, + draftOccurrences: null, + outcome: createOutcome, + ...created.measurement, + }); + + for (const [liveOccurrences, draftOccurrences] of candidateSplits(totalOccurrences)) { + if (draftOccurrences === null) continue; + const slug = `runtime-update-${totalOccurrences}-${liveOccurrences}-${draftOccurrences}`; + const initial = await runtime.handleContentCreate(fixture.collectionSlug, { + slug, + data: { title: slug, ...mediaData(liveOccurrences, `${slug}-live`) }, + }); + expect(initial.success).toBe(true); + if (!initial.success) throw new Error(initial.error.message); + + const updated = await measure(() => + runtime.handleContentUpdate(fixture.collectionSlug, initial.data.item.id, { + data: mediaData(draftOccurrences, `${slug}-draft`), + }), + ); + expect(updated.value.success).toBe(true); + const updateOutcome = await workOutcome(initial.data.item.id); + expect(updateOutcome).toBe( + liveOccurrences <= 12 && draftOccurrences <= 12 ? "completed" : "failed", + ); + rows.push({ + path: "runtime-update", + totalOccurrences, + liveOccurrences, + draftOccurrences, + outcome: updateOutcome, + ...updated.measurement, + }); + } + } + for (const draftOccurrences of [3, 6, 9, 12]) { + const slug = `runtime-conflict-${draftOccurrences}`; + const created = await runtime.handleContentCreate(fixture.collectionSlug, { + slug, + data: { title: slug, ...mediaData(0, `${slug}-live`) }, + }); + expect(created.success).toBe(true); + if (!created.success) throw new Error(created.error.message); + const initial = await runtime.handleContentUpdate( + fixture.collectionSlug, + created.data.item.id, + { data: mediaData(draftOccurrences, `${slug}-draft-initial`) }, + ); + expect(initial.success).toBe(true); + + const triggerName = `measure_admission_conflict_${draftOccurrences}`; + const draftSourceKey = buildContentMediaUsageSourceKey({ + collectionId: fixture.collectionId, + collectionSlug: fixture.collectionSlug, + contentId: created.data.item.id, + sourceVariant: "draft_overlay", + }); + await sql` + CREATE TRIGGER ${sql.ref(triggerName)} + AFTER INSERT ON _emdash_media_usage_generation_writes + WHEN NEW.source_key = ${sql.lit(draftSourceKey)} + BEGIN + UPDATE _emdash_media_usage_sources + SET updated_at = updated_at || 'x' + WHERE source_key = NEW.source_key; + END + `.execute(db); + try { + const result = await measure(() => + runtime.handleContentUpdate(fixture.collectionSlug, created.data.item.id, { + data: mediaData(draftOccurrences, `${slug}-draft-changed`), + }), + ); + expect(result.value.success).toBe(true); + const conflictOutcome = await workOutcome(created.data.item.id); + expect(conflictOutcome).toBe("retry"); + rows.push({ + path: "runtime-conflict", + totalOccurrences: draftOccurrences, + liveOccurrences: 0, + draftOccurrences, + outcome: conflictOutcome, + ...result.measurement, + }); + } finally { + await sql`DROP TRIGGER ${sql.ref(triggerName)}`.execute(db); + } + } + + const worstByBoundary = new Map(); + for (const row of rows) { + const key = + row.path === "processor-bytes" + ? `${row.path}:${row.payloadBytes}` + : `${row.path}:${row.totalOccurrences}`; + const current = worstByBoundary.get(key); + if (!current || row.d1Queries > current.d1Queries) worstByBoundary.set(key, row); + } + console.info(`PR1_MEASUREMENTS=${JSON.stringify([...worstByBoundary.values()])}`); + expect(rows.every((row) => row.maxBinds <= 100)).toBe(true); +}); + +function candidateSplits(totalOccurrences: number): Array<[number, number | null]> { + const splits: Array<[number, number | null]> = [[totalOccurrences, null]]; + for (const liveOccurrences of [0, 1, 3, Math.floor(totalOccurrences / 2), totalOccurrences]) { + if (liveOccurrences < 0 || liveOccurrences > totalOccurrences) continue; + splits.push([liveOccurrences, totalOccurrences - liveOccurrences]); + } + return [...new Map(splits.map((split) => [split.join(":"), split])).values()]; +} + +async function createActiveFixture() { + return createMediaUsageAdmissionFixture(db, "admission_measurement"); +} + +async function insertEntry( + contentId: string, + data: ReturnType, + title = contentId, +): Promise { + await insertMediaUsageMeasurementEntry(db, fixture, contentId, data, title); +} + +async function addDraft(contentId: string, data: ReturnType): Promise { + await addMediaUsageMeasurementDraft(db, fixture, contentId, data); +} + +async function measure(operation: () => Promise): Promise<{ + value: T; + measurement: Measurement; +}> { + captured = []; + const beforeChanges = totalChanges(); + const startedAt = performance.now(); + const value = await operation(); + await waitForDeferredTasks(); + return { + value, + measurement: { + d1Queries: captured.filter((query) => !/^(?:begin|commit|rollback)$/i.test(query.sql.trim())) + .length, + maxBinds: Math.max(0, ...captured.map((query) => query.parameters.length)), + maxSqlBytes: Math.max(0, ...captured.map((query) => Buffer.byteLength(query.sql))), + changedRows: totalChanges() - beforeChanges, + durationMs: Number((performance.now() - startedAt).toFixed(3)), + }, + }; +} + +function totalChanges(): number { + return (sqlite.prepare("SELECT total_changes() AS count").get() as { count: number }).count; +} + +async function workOutcome(contentId: string): Promise { + const work = await db + .selectFrom("_emdash_media_usage_work") + .select("state") + .where("content_id", "=", contentId) + .executeTakeFirst(); + return work?.state ?? "completed"; +} diff --git a/packages/core/tests/integration/database/media-usage-projection-admission-runtime.test.ts b/packages/core/tests/integration/database/media-usage-projection-admission-runtime.test.ts new file mode 100644 index 0000000000..e26fa81e6d --- /dev/null +++ b/packages/core/tests/integration/database/media-usage-projection-admission-runtime.test.ts @@ -0,0 +1,349 @@ +import type { + KyselyPlugin, + PluginTransformQueryArgs, + PluginTransformResultArgs, + QueryResult, + RootOperationNode, + UnknownRow, +} from "kysely"; +import { sql } from "kysely"; +import { afterEach, beforeEach, expect, it } from "vitest"; + +import { MediaUsageRepository } from "../../../src/database/repositories/media-usage.js"; +import { refreshContentMediaUsage } from "../../../src/media/usage/content-refresh.js"; +import { loadContentMediaUsageSnapshots } from "../../../src/media/usage/content-snapshots.js"; +import { buildContentMediaUsageSourceKey } from "../../../src/media/usage/source-key.js"; +import { processMediaUsageWorkAfterWrite } from "../../../src/media/usage/work-processor.js"; +import { + addMediaUsageMeasurementDraft, + createMediaUsageAdmissionFixture, + insertMediaUsageMeasurementEntry, + mediaUsageMeasurementData, +} from "../../utils/media-usage-admission-fixture.js"; +import { + describeEachDialect, + setupForDialect, + teardownForDialect, + type DialectTestContext, +} from "../../utils/test-db.js"; + +describeEachDialect("media usage projection admission runtime", (dialect) => { + let ctx: DialectTestContext; + let fixture: Awaited>; + let repo: MediaUsageRepository; + + beforeEach(async () => { + ctx = await setupForDialect(dialect); + fixture = await createMediaUsageAdmissionFixture(ctx.db, "admission_runtime"); + repo = new MediaUsageRepository(ctx.db); + }); + + afterEach(async () => { + await teardownForDialect(ctx); + }); + + it("terminally rejects an oversized replacement before publishing either variant", async () => { + await insertEntry("oversized-variants", 6); + await addMediaUsageMeasurementDraft( + ctx.db, + fixture, + "oversized-variants", + mediaUsageMeasurementData(7, "oversized-draft"), + ); + + const result = await processMediaUsageWorkAfterWrite( + ctx.db, + fixture.collectionSlug, + "oversized-variants", + ); + + expect(result.outcome).toBe("failed"); + expect(await workRow("oversized-variants")).toEqual( + expect.objectContaining({ + state: "failed", + attempt_count: 1, + last_error_code: "MEDIA_USAGE_RESOURCE_LIMIT", + }), + ); + expect((await repo.findSources(sourceKeys("oversized-variants"))).size).toBe(0); + expect(await occurrenceCount("oversized-variants")).toBe(0); + expect(await coverage()).toEqual( + expect.objectContaining({ + status: "partial", + last_error_code: "MEDIA_USAGE_RESOURCE_LIMIT", + }), + ); + }); + + it("admits the occurrence boundary and keeps an oversized exact projection as a no-op", async () => { + await insertEntry("boundary", 12); + expect( + (await processMediaUsageWorkAfterWrite(ctx.db, fixture.collectionSlug, "boundary")).outcome, + ).toBe("completed"); + expect(await currentOccurrenceCount("boundary", "columns")).toBe(12); + + await insertEntry("oversized-no-op", 13); + const snapshots = await loadSnapshots("oversized-no-op"); + const snapshot = snapshots.find((candidate) => candidate.source.sourceVariant === "columns"); + if (!snapshot) throw new Error("Missing columns snapshot"); + const stored = await repo.replaceSource(snapshot.source, snapshot.occurrences); + + const result = await processMediaUsageWorkAfterWrite( + ctx.db, + fixture.collectionSlug, + "oversized-no-op", + ); + + expect(result.outcome).toBe("completed"); + expect((await repo.findSource(snapshot.source.sourceKey))?.currentGeneration).toBe( + stored.currentGeneration, + ); + expect(await currentOccurrenceCount("oversized-no-op", "columns")).toBe(13); + }); + + it("rejects oversized source bytes without mutation", async () => { + await insertMediaUsageMeasurementEntry( + ctx.db, + fixture, + "oversized-bytes", + mediaUsageMeasurementData(0, "oversized-bytes"), + "é".repeat(300_000), + ); + expect( + (await processMediaUsageWorkAfterWrite(ctx.db, fixture.collectionSlug, "oversized-bytes")) + .outcome, + ).toBe("failed"); + expect(await repo.findSource(sourceKey("oversized-bytes", "columns"))).toBeNull(); + }); + + it("rejects oversized absent-source cleanup without mutation", async () => { + await insertEntry("oversized-delete", 13); + const snapshots = await loadSnapshots("oversized-delete"); + const snapshot = snapshots.find((candidate) => candidate.source.sourceVariant === "columns"); + if (!snapshot) throw new Error("Missing columns snapshot"); + const stored = await repo.replaceSource(snapshot.source, snapshot.occurrences); + await sql` + DELETE FROM ${sql.ref(fixture.tableName)} + WHERE id = 'oversized-delete' + `.execute(ctx.db); + + const deleted = await processMediaUsageWorkAfterWrite( + ctx.db, + fixture.collectionSlug, + "oversized-delete", + ); + + expect(deleted.outcome).toBe("failed"); + expect((await repo.findSource(snapshot.source.sourceKey))?.currentGeneration).toBe( + stored.currentGeneration, + ); + expect(await currentOccurrenceCount("oversized-delete", "columns")).toBe(13); + }); + + it("leaves synchronous backwards-compatible refresh outside admission", async () => { + await insertEntry("legacy-refresh", 13); + + const result = await refreshContentMediaUsage(ctx.db, fixture.collectionSlug, "legacy-refresh"); + + expect(result.success).toBe(true); + const count = await ctx.db + .selectFrom("_emdash_media_usage") + .select((eb) => eb.fn.countAll().as("count")) + .where("source_key", "=", "content:admission_runtime:legacy-refresh:columns") + .executeTakeFirstOrThrow(); + expect(Number(count.count)).toBe(13); + }); + + it("defers immediately after a reserved conflict inside the 40-query envelope", async () => { + await insertEntry("conflict", 0); + await addMediaUsageMeasurementDraft( + ctx.db, + fixture, + "conflict", + mediaUsageMeasurementData(12, "draft-initial"), + ); + expect( + (await processMediaUsageWorkAfterWrite(ctx.db, fixture.collectionSlug, "conflict")).outcome, + ).toBe("completed"); + await ctx.db + .updateTable("revisions") + .set({ data: JSON.stringify(mediaUsageMeasurementData(12, "draft-changed")) }) + .where("id", "=", "revision-conflict") + .execute(); + await sql` + UPDATE ${sql.ref(fixture.tableName)} + SET updated_at = '2026-08-11T12:00:00.000Z' + WHERE id = 'conflict' + `.execute(ctx.db); + await installDraftConflictTrigger(sourceKey("conflict", "draft_overlay")); + const counter = new QueryCountingPlugin(); + const draftRowsBefore = await occurrenceCountForSource(sourceKey("conflict", "draft_overlay")); + + try { + const result = await processMediaUsageWorkAfterWrite( + ctx.db.withPlugin(counter), + fixture.collectionSlug, + "conflict", + ); + expect(result.outcome).toBe("retry"); + expect(counter.count).toBeLessThanOrEqual(40); + expect( + (await occurrenceCountForSource(sourceKey("conflict", "draft_overlay"))) - draftRowsBefore, + ).toBe(12); + expect(await workRow("conflict")).toEqual( + expect.objectContaining({ + state: "retry", + attempt_count: 1, + last_error_code: "MEDIA_USAGE_GENERATION_CONFLICT", + }), + ); + } finally { + await removeDraftConflictTrigger(); + } + }); + + async function insertEntry(contentId: string, occurrences: number): Promise { + await insertMediaUsageMeasurementEntry( + ctx.db, + fixture, + contentId, + mediaUsageMeasurementData(occurrences, contentId), + ); + } + + async function loadSnapshots(contentId: string) { + const result = await loadContentMediaUsageSnapshots( + ctx.db, + fixture.collectionSlug, + contentId, + undefined, + { collectionId: fixture.collectionId, identityVersion: 1 }, + ); + if (!result.success) throw new Error(result.error); + return result.snapshots; + } + + function sourceKey(contentId: string, sourceVariant: "columns" | "draft_overlay"): string { + return buildContentMediaUsageSourceKey({ + collectionId: fixture.collectionId, + collectionSlug: fixture.collectionSlug, + contentId, + sourceVariant, + }); + } + + function sourceKeys(contentId: string): string[] { + return [sourceKey(contentId, "columns"), sourceKey(contentId, "draft_overlay")]; + } + + async function occurrenceCount(contentId: string): Promise { + const row = await ctx.db + .selectFrom("_emdash_media_usage") + .select((eb) => eb.fn.countAll().as("count")) + .where("source_key", "in", sourceKeys(contentId)) + .executeTakeFirstOrThrow(); + return Number(row.count); + } + + async function occurrenceCountForSource(sourceKeyValue: string): Promise { + const row = await ctx.db + .selectFrom("_emdash_media_usage") + .select((eb) => eb.fn.countAll().as("count")) + .where("source_key", "=", sourceKeyValue) + .executeTakeFirstOrThrow(); + return Number(row.count); + } + + async function currentOccurrenceCount( + contentId: string, + sourceVariant: "columns" | "draft_overlay", + ): Promise { + const source = await repo.findSource(sourceKey(contentId, sourceVariant)); + if (!source) return 0; + const row = await ctx.db + .selectFrom("_emdash_media_usage") + .select((eb) => eb.fn.countAll().as("count")) + .where("source_key", "=", source.sourceKey) + .where("generation", "=", source.currentGeneration) + .executeTakeFirstOrThrow(); + return Number(row.count); + } + + function workRow(contentId: string) { + return ctx.db + .selectFrom("_emdash_media_usage_work") + .selectAll() + .where("content_id", "=", contentId) + .executeTakeFirstOrThrow(); + } + + function coverage() { + return ctx.db + .selectFrom("_emdash_media_usage_index_status") + .selectAll() + .where("collection_id", "=", fixture.collectionId) + .executeTakeFirstOrThrow(); + } + + async function installDraftConflictTrigger(draftSourceKey: string): Promise { + if (ctx.dialect === "postgres") { + await sql` + CREATE FUNCTION media_usage_admission_draft_conflict() + RETURNS trigger + LANGUAGE plpgsql + AS $$ + BEGIN + IF NEW.source_key = ${sql.lit(draftSourceKey)} THEN + UPDATE _emdash_media_usage_sources + SET updated_at = updated_at || 'x' + WHERE source_key = NEW.source_key; + END IF; + RETURN NEW; + END; + $$ + `.execute(ctx.db); + await sql` + CREATE TRIGGER media_usage_admission_draft_conflict + AFTER INSERT ON _emdash_media_usage_generation_writes + FOR EACH ROW EXECUTE FUNCTION media_usage_admission_draft_conflict() + `.execute(ctx.db); + return; + } + + await sql` + CREATE TRIGGER media_usage_admission_draft_conflict + AFTER INSERT ON _emdash_media_usage_generation_writes + WHEN NEW.source_key = ${sql.lit(draftSourceKey)} + BEGIN + UPDATE _emdash_media_usage_sources + SET updated_at = updated_at || 'x' + WHERE source_key = NEW.source_key; + END + `.execute(ctx.db); + } + + async function removeDraftConflictTrigger(): Promise { + if (ctx.dialect === "postgres") { + await sql` + DROP TRIGGER media_usage_admission_draft_conflict + ON _emdash_media_usage_generation_writes + `.execute(ctx.db); + await sql`DROP FUNCTION media_usage_admission_draft_conflict()`.execute(ctx.db); + return; + } + await sql`DROP TRIGGER media_usage_admission_draft_conflict`.execute(ctx.db); + } +}); + +class QueryCountingPlugin implements KyselyPlugin { + count = 0; + + transformQuery(args: PluginTransformQueryArgs): RootOperationNode { + this.count++; + return args.node; + } + + transformResult(args: PluginTransformResultArgs): Promise> { + return Promise.resolve(args.result); + } +} diff --git a/packages/core/tests/integration/database/media-usage-projection-admission.test.ts b/packages/core/tests/integration/database/media-usage-projection-admission.test.ts new file mode 100644 index 0000000000..c5fc2f5e38 --- /dev/null +++ b/packages/core/tests/integration/database/media-usage-projection-admission.test.ts @@ -0,0 +1,224 @@ +import { afterEach, beforeEach, expect, it } from "vitest"; + +import { + MediaUsageRepository, + type MediaUsageOccurrenceInput, + type MediaUsageSourceInput, +} from "../../../src/database/repositories/media-usage.js"; +import { + createContentMediaUsageAdmissionBudget, + MEDIA_USAGE_PROJECTION_ADMISSION_LIMITS, + planContentMediaUsageProjectionAdmission, +} from "../../../src/media/usage/content-refresh.js"; +import type { ContentMediaUsageSnapshot } from "../../../src/media/usage/content-snapshots.js"; +import { buildMediaUsageProjectionFingerprint } from "../../../src/media/usage/projection-fingerprint.js"; +import { + buildContentMediaUsageSourceKey, + type MediaUsageContentSourceVariant, +} from "../../../src/media/usage/source-key.js"; +import { + describeEachDialect, + setupForDialect, + teardownForDialect, + type DialectTestContext, +} from "../../utils/test-db.js"; + +const COLLECTION_ID = "collection-admission"; +const COLLECTION_SLUG = "admission"; +const EXTRACTION_FIELDS = [{ slug: "gallery", type: "image" as const }]; + +describeEachDialect("content media usage projection admission", (dialect) => { + let ctx: DialectTestContext; + let repo: MediaUsageRepository; + + beforeEach(async () => { + ctx = await setupForDialect(dialect); + repo = new MediaUsageRepository(ctx.db); + await ctx.db + .insertInto("_emdash_collections") + .values({ id: COLLECTION_ID, slug: COLLECTION_SLUG, label: "Admission" }) + .execute(); + }); + + afterEach(async () => { + await teardownForDialect(ctx); + }); + + it("admits a whole entry only when the combined variant plan fits", async () => { + const withinBudget = await plan([ + await snapshot("combined-12", "columns", 6), + await snapshot("combined-12", "draft_overlay", 6), + ]); + const overBudget = await plan([ + await snapshot("combined-13", "columns", 6), + await snapshot("combined-13", "draft_overlay", 7), + ]); + + expect(withinBudget.outcome).toBe("admitted"); + expect(overBudget.outcome).toBe("intrinsic_resource_limit"); + }); + + it("recomputes an oversized mixed plan after proving its replacement is unchanged", async () => { + const contentId = "mixed-no-op-delete"; + const unchangedColumns = await snapshot(contentId, "columns", 0, "é".repeat(270_000)); + const absentDraft = await snapshot(contentId, "draft_overlay", 1, "Small draft"); + await repo.replaceSource(unchangedColumns.source, unchangedColumns.occurrences); + await repo.replaceSource(absentDraft.source, absentDraft.occurrences); + const observedSources = await repo.findSources(canonicalSourceKeys(contentId)); + const budget = createContentMediaUsageAdmissionBudget(); + + const result = await planContentMediaUsageProjectionAdmission( + repo, + [unchangedColumns], + observedSources, + canonicalSourceKeys(contentId), + budget, + ); + + expect(result.outcome).toBe("admitted"); + if (result.outcome !== "admitted") throw new Error(result.outcome); + expect(result.noOpSourceKeys).toEqual(new Set([unchangedColumns.source.sourceKey])); + expect(result.absentSources.map((source) => source.sourceKey)).toEqual([ + absentDraft.source.sourceKey, + ]); + expect(result.occurrenceMutationUnits).toBe(1); + expect(budget.remainingOccurrenceMutationUnits).toBe( + MEDIA_USAGE_PROJECTION_ADMISSION_LIMITS.maxOccurrenceMutationUnitsPerClaim - 1, + ); + }); + + it("rejects deletion when the stored source row alone exceeds the byte limit", async () => { + const contentId = "oversized-source-delete"; + const absentDraft = await snapshot(contentId, "draft_overlay", 0, "é".repeat(300_000)); + await repo.replaceSource(absentDraft.source, []); + const observedSources = await repo.findSources(canonicalSourceKeys(contentId)); + const budget = createContentMediaUsageAdmissionBudget(); + + const result = await planContentMediaUsageProjectionAdmission( + repo, + [], + observedSources, + canonicalSourceKeys(contentId), + budget, + ); + + expect(result.outcome).toBe("intrinsic_resource_limit"); + expect(budget.hasReservedMutation).toBe(false); + }); + + it("reserves one claim budget and defers later intrinsic excess as a conflict", async () => { + const budget = createContentMediaUsageAdmissionBudget(); + const first = await planContentMediaUsageProjectionAdmission( + repo, + [await snapshot("reserved-small", "columns", 1)], + new Map(), + canonicalSourceKeys("reserved-small"), + budget, + ); + const oversized = [await snapshot("reserved-large", "columns", 13)]; + const deferred = await planContentMediaUsageProjectionAdmission( + repo, + oversized, + new Map(), + canonicalSourceKeys("reserved-large"), + budget, + ); + const freshClaim = await planContentMediaUsageProjectionAdmission( + repo, + oversized, + new Map(), + canonicalSourceKeys("reserved-large"), + createContentMediaUsageAdmissionBudget(), + ); + + expect(first.outcome).toBe("admitted"); + expect(budget.hasReservedMutation).toBe(true); + expect(budget.remainingOccurrenceMutationUnits).toBe(11); + expect(deferred.outcome).toBe("claim_budget_deferred"); + expect(freshClaim.outcome).toBe("intrinsic_resource_limit"); + }); + + async function plan(snapshots: readonly ContentMediaUsageSnapshot[]) { + const contentId = snapshots[0]?.source.contentId; + if (!contentId) throw new Error("Expected a content identity"); + return planContentMediaUsageProjectionAdmission( + repo, + snapshots, + new Map(), + canonicalSourceKeys(contentId), + createContentMediaUsageAdmissionBudget(), + ); + } +}); + +async function snapshot( + contentId: string, + sourceVariant: MediaUsageContentSourceVariant, + occurrenceCount: number, + contentTitle = contentId, +): Promise { + const source: MediaUsageSourceInput = { + sourceKey: buildContentMediaUsageSourceKey({ + collectionId: COLLECTION_ID, + collectionSlug: COLLECTION_SLUG, + contentId, + sourceVariant, + }), + sourceType: "content", + collectionId: COLLECTION_ID, + collectionSlug: COLLECTION_SLUG, + contentId, + sourceVariant, + locale: "en", + translationGroup: `translation-${contentId}`, + contentSlug: contentId, + contentTitle, + contentStatus: sourceVariant === "columns" ? "published" : "draft", + contentScheduledAt: null, + contentDeletedAt: null, + revisionId: sourceVariant === "draft_overlay" ? `revision-${contentId}` : null, + schemaVersion: 1, + sourceUpdatedAt: "2026-08-11T00:00:00.000Z", + sourceVersion: 1, + identityVersion: 1, + }; + const occurrences = Array.from({ length: occurrenceCount }, (_, index) => occurrence(index)); + const projection = await buildMediaUsageProjectionFingerprint({ + collectionId: COLLECTION_ID, + source, + occurrences, + extractionFields: EXTRACTION_FIELDS, + }); + source.sourceFingerprint = projection.fingerprint; + return { + source, + occurrences, + fields: EXTRACTION_FIELDS, + projectionByteLength: projection.byteLength, + }; +} + +function occurrence(index: number): MediaUsageOccurrenceInput { + return { + fieldSlug: "gallery", + fieldPath: `gallery[${index}]`, + occurrenceIndex: index, + referenceType: "image_field", + mediaId: `media-${index}`, + provider: "local", + providerAssetId: `media-${index}`, + mediaKind: "image", + mimeType: "image/webp", + }; +} + +function canonicalSourceKeys(contentId: string): string[] { + return (["columns", "draft_overlay"] as const).map((sourceVariant) => + buildContentMediaUsageSourceKey({ + collectionId: COLLECTION_ID, + collectionSlug: COLLECTION_SLUG, + contentId, + sourceVariant, + }), + ); +} 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 e132923f39..038f81635d 100644 --- a/packages/core/tests/integration/database/media-usage-repository.test.ts +++ b/packages/core/tests/integration/database/media-usage-repository.test.ts @@ -1503,6 +1503,35 @@ describeEachDialect("MediaUsageRepository", (dialect) => { expect(rows).toHaveLength(SQL_BATCH_SIZE + 7); expect(rows.every((row) => row.generation === source.currentGeneration)).toBe(true); }); + + it("bounds current-generation deletion measurement without counting stale generations", async () => { + const sourceInput = contentSource("entry-admission", "draft_overlay"); + await repo.replaceSource( + sourceInput, + Array.from({ length: 30 }, (_, index) => + occurrence(`stale-${index}`, `stale-media-${index}`), + ), + ); + const current = await repo.replaceSource( + sourceInput, + Array.from({ length: 13 }, (_, index) => + occurrence(`current-${index}`, `current-media-${index}`), + ), + ); + + const measurement = await repo.measureSourceGenerationDeletion( + current.sourceKey, + current.currentGeneration, + 12, + ); + + expect(measurement).toEqual({ + occurrenceCount: 13, + occurrenceBytes: expect.any(Number), + exceedsOccurrenceLimit: true, + }); + expect(measurement.occurrenceBytes).toBeGreaterThan(0); + }); }); function contentSource( diff --git a/packages/core/tests/unit/media/usage-projection-fingerprint.test.ts b/packages/core/tests/unit/media/usage-projection-fingerprint.test.ts index efdd72b436..2a9bd4d515 100644 --- a/packages/core/tests/unit/media/usage-projection-fingerprint.test.ts +++ b/packages/core/tests/unit/media/usage-projection-fingerprint.test.ts @@ -46,21 +46,38 @@ const extractionFields = [{ slug: "gallery", type: "image" as const }]; it("is order-independent but changes for every projection input class", async () => { const baseline = await fingerprint(); - expect(await fingerprint({ occurrences: occurrences.toReversed() })).toBe(baseline); - expect(await fingerprint({ collectionId: "collection-2" })).not.toBe(baseline); - expect(await fingerprint({ source: { ...source, contentTitle: "Changed title" } })).not.toBe( - baseline, + expect((await fingerprint({ occurrences: occurrences.toReversed() })).fingerprint).toBe( + baseline.fingerprint, ); + expect((await fingerprint({ collectionId: "collection-2" })).fingerprint).not.toBe( + baseline.fingerprint, + ); + expect( + (await fingerprint({ source: { ...source, contentTitle: "Changed title" } })).fingerprint, + ).not.toBe(baseline.fingerprint); expect( - await fingerprint({ - occurrences: [{ ...occurrences[0]!, mediaId: "changed-media" }, occurrences[1]!], - }), - ).not.toBe(baseline); + ( + await fingerprint({ + occurrences: [{ ...occurrences[0]!, mediaId: "changed-media" }, occurrences[1]!], + }) + ).fingerprint, + ).not.toBe(baseline.fingerprint); expect( - await fingerprint({ - extractionFields: [...extractionFields, { slug: "hero", type: "image" as const }], - }), - ).not.toBe(baseline); + ( + await fingerprint({ + extractionFields: [...extractionFields, { slug: "hero", type: "image" as const }], + }) + ).fingerprint, + ).not.toBe(baseline.fingerprint); + expect(baseline.fingerprint).toMatch(/^media-usage-projection:v1:sha256:[a-f0-9]{64}$/); + expect(baseline.byteLength).toBeGreaterThan(0); +}); + +it("returns the reused UTF-8 payload length with the fingerprint", async () => { + const ascii = await fingerprint({ source: { ...source, contentTitle: "e" } }); + const multibyte = await fingerprint({ source: { ...source, contentTitle: "é" } }); + + expect(multibyte.byteLength).toBe(ascii.byteLength + 1); }); it("refuses to mint a current fingerprint without immutable collection identity", async () => { diff --git a/packages/core/tests/utils/media-usage-admission-fixture.ts b/packages/core/tests/utils/media-usage-admission-fixture.ts new file mode 100644 index 0000000000..6318390a9a --- /dev/null +++ b/packages/core/tests/utils/media-usage-admission-fixture.ts @@ -0,0 +1,130 @@ +import { sql, type Kysely } from "kysely"; + +import type { Database } from "../../src/database/types.js"; +import { installMediaUsageCaptureTriggers } from "../../src/media/usage/capture-triggers.js"; +import { SchemaRegistry } from "../../src/schema/registry.js"; + +export interface MediaUsageAdmissionFixture { + collectionId: string; + collectionSlug: string; + tableName: string; +} + +export async function createMediaUsageAdmissionFixture( + db: Kysely, + collectionSlug: string, +): Promise { + const registry = new SchemaRegistry(db); + await registry.createCollection({ slug: collectionSlug, label: "Admission measurement" }); + await registry.createField(collectionSlug, { slug: "title", label: "Title", type: "string" }); + await registry.createField(collectionSlug, { + slug: "body", + label: "Body", + type: "portableText", + }); + await registry.createField(collectionSlug, { + slug: "sections", + label: "Sections", + type: "repeater", + validation: { subFields: [{ slug: "image", type: "image", label: "Image" }] }, + }); + const collection = await registry.getCollection(collectionSlug); + if (!collection) throw new Error("Expected admission measurement collection"); + + await db + .updateTable("_emdash_media_usage_index_status") + .set({ + collection_id: collection.id, + status: "complete", + completed_at: "2026-08-11T00:00:00.000Z", + reconciliation_required: 0, + capture_state: "installing", + }) + .where("adapter_id", "=", "content-media") + .where("scope_type", "=", "collection") + .where("scope_key", "=", collectionSlug) + .execute(); + await installMediaUsageCaptureTriggers(db, { + collectionId: collection.id, + collectionSlug, + }); + await db + .updateTable("_emdash_media_usage_index_status") + .set({ capture_state: "active" }) + .where("collection_id", "=", collection.id) + .execute(); + await db + .updateTable("_emdash_media_usage_activation") + .set({ state: "active", activated_at: "2026-08-11T00:00:00.000Z" }) + .execute(); + + return { + collectionId: collection.id, + collectionSlug, + tableName: `ec_${collectionSlug}`, + }; +} + +export async function insertMediaUsageMeasurementEntry( + db: Kysely, + fixture: MediaUsageAdmissionFixture, + contentId: string, + data: ReturnType, + title = contentId, +): Promise { + await sql` + INSERT INTO ${sql.ref(fixture.tableName)} (id, slug, status, title, body, sections) + VALUES ( + ${contentId}, + ${contentId}, + 'published', + ${title}, + ${JSON.stringify(data.body)}, + ${JSON.stringify(data.sections)} + ) + `.execute(db); +} + +export async function addMediaUsageMeasurementDraft( + db: Kysely, + fixture: MediaUsageAdmissionFixture, + contentId: string, + data: ReturnType, +): Promise { + const revisionId = `revision-${contentId}`; + await db + .insertInto("revisions") + .values({ + id: revisionId, + collection: fixture.collectionSlug, + entry_id: contentId, + data: JSON.stringify(data), + author_id: null, + }) + .execute(); + await sql` + UPDATE ${sql.ref(fixture.tableName)} + SET draft_revision_id = ${revisionId} + WHERE id = ${contentId} + `.execute(db); +} + +export function mediaUsageMeasurementData(count: number, prefix: string) { + const portableTextCount = Math.ceil(count / 2); + const repeaterCount = Math.floor(count / 2); + return { + body: Array.from({ length: portableTextCount }, (_, index) => ({ + _type: "image", + _key: `body-${index}`, + asset: { _ref: `${prefix}-body-${index}` }, + })), + sections: Array.from({ length: repeaterCount }, (_, index) => ({ + _key: `section-${index}`, + image: { + id: `${prefix}-section-${index}`, + provider: "local", + mimeType: "image/webp", + }, + })), + }; +} diff --git a/packages/core/tests/workerd/media-usage-projection-admission-d1.test.ts b/packages/core/tests/workerd/media-usage-projection-admission-d1.test.ts new file mode 100644 index 0000000000..e1b94be650 --- /dev/null +++ b/packages/core/tests/workerd/media-usage-projection-admission-d1.test.ts @@ -0,0 +1,135 @@ +import { env } from "cloudflare:test"; +import { Kysely } from "kysely"; +import { afterAll, beforeAll, expect, it } from "vitest"; + +import { RawBindingD1Dialect } from "../../../cloudflare/src/db/d1-dialect.js"; +import { runMigrations } from "../../src/database/migrations/runner.js"; +import type { Database } from "../../src/database/types.js"; +import { waitForDeferredTasks } from "../../src/deferred-tasks.js"; +import { setI18nConfig } from "../../src/i18n/config.js"; +import { createTestRuntime } from "../utils/mcp-runtime.js"; +import { + createMediaUsageAdmissionFixture, + mediaUsageMeasurementData, +} from "../utils/media-usage-admission-fixture.js"; + +declare module "cloudflare:test" { + interface ProvidedEnv { + DB: D1Database; + } +} + +interface CapturedD1Query { + sql: string; + binds: number; + rowsRead: number; + rowsWritten: number; + durationMs: number; +} + +let db: Kysely; +let captured: CapturedD1Query[]; +let fixture: Awaited>; + +beforeAll(async () => { + setI18nConfig(null); + captured = []; + db = new Kysely({ + dialect: new RawBindingD1Dialect({ database: captureD1(env.DB) }), + }); + await runMigrations(db); + fixture = await createMediaUsageAdmissionFixture(db, "admission_d1"); +}); + +afterAll(async () => { + setI18nConfig(null); + await db.destroy(); +}); + +it("reports real D1 metadata for the approved full runtime boundary", async () => { + const runtime = createTestRuntime(db); + const created = await runtime.handleContentCreate(fixture.collectionSlug, { + slug: "boundary-d1", + data: { + title: "Boundary D1", + ...mediaUsageMeasurementData(0, "boundary-d1-live"), + }, + }); + expect(created.success).toBe(true); + if (!created.success) throw new Error(created.error.message); + + const measurement = await measure(() => + runtime.handleContentUpdate(fixture.collectionSlug, created.data.item.id, { + data: mediaUsageMeasurementData(12, "boundary-d1-draft"), + }), + ); + expect(measurement.value.success).toBe(true); + expect(measurement.d1Queries).toBeLessThanOrEqual(40); + expect(measurement.rowsWritten).toBeLessThanOrEqual(122); + expect(measurement.maxBinds).toBeLessThanOrEqual(100); + expect(measurement.maxSqlBytes).toBeLessThan(100 * 1024); + const { value: _value, ...evidence } = measurement; + console.info(`PR1_D1_MEASUREMENT=${JSON.stringify(evidence)}`); +}); + +async function measure(operation: () => Promise) { + captured = []; + const startedAt = performance.now(); + const value = await operation(); + await waitForDeferredTasks(); + return { + value, + d1Queries: captured.length, + rowsRead: captured.reduce((total, query) => total + query.rowsRead, 0), + rowsWritten: captured.reduce((total, query) => total + query.rowsWritten, 0), + d1DurationMs: captured.reduce((total, query) => total + query.durationMs, 0), + wallDurationMs: Number((performance.now() - startedAt).toFixed(3)), + maxBinds: Math.max(0, ...captured.map((query) => query.binds)), + maxSqlBytes: Math.max( + 0, + ...captured.map((query) => new TextEncoder().encode(query.sql).byteLength), + ), + }; +} + +function captureD1(database: D1Database): D1Database { + return new Proxy(database, { + get(target, property) { + if (property === "prepare") { + return (query: string) => captureStatement(target.prepare(query), query, 0); + } + const value: unknown = Reflect.get(target, property, target); + return typeof value === "function" ? value.bind(target) : value; + }, + }); +} + +function captureStatement( + statement: D1PreparedStatement, + query: string, + binds: number, +): D1PreparedStatement { + return new Proxy(statement, { + get(target, property) { + if (property === "bind") { + return (...values: unknown[]) => + captureStatement(target.bind(...values), query, values.length); + } + if (property === "all") { + return async () => { + const result = await target.all(); + captured.push({ + sql: query, + binds, + rowsRead: result.meta.rows_read, + rowsWritten: result.meta.rows_written, + durationMs: result.meta.duration, + }); + return result; + }; + } + const value: unknown = Reflect.get(target, property, target); + return typeof value === "function" ? value.bind(target) : value; + }, + }); +} diff --git a/packages/core/vitest.config.ts b/packages/core/vitest.config.ts index 437e0f9daf..2e93f28864 100644 --- a/packages/core/vitest.config.ts +++ b/packages/core/vitest.config.ts @@ -46,6 +46,7 @@ export default defineConfig({ // The fixture has symlinked node_modules that contain test files // from transitive deps (zod, emdash) — exclude them too. exclude: [ + "tests/workerd/**", // Render tests import .astro components and need the Astro Vite // plugin -- run them via the dedicated repro config (test:repro), // not this plain-node config which cannot transform .astro. diff --git a/packages/core/vitest.workerd.config.ts b/packages/core/vitest.workerd.config.ts new file mode 100644 index 0000000000..6f22f6aab4 --- /dev/null +++ b/packages/core/vitest.workerd.config.ts @@ -0,0 +1,39 @@ +import { cloudflareTest } from "@cloudflare/vitest-pool-workers"; +import { defineConfig } from "vitest/config"; + +const virtualStubs: Record = { + "virtual:emdash/wait-until": "export const waitUntil = undefined;", + "virtual:emdash/scheduler": "export const createScheduler = null;", + "virtual:emdash/config": "export default {};", + "virtual:emdash/env": "export const env = undefined;", + "virtual:emdash/object-cache": + "export const createObjectCache = undefined; export const objectCacheConfig = {};", +}; + +export default defineConfig({ + plugins: [ + { + name: "emdash-virtual-stubs", + resolveId(id) { + if (Object.hasOwn(virtualStubs, id)) return `\0${id}`; + return null; + }, + load(id) { + if (!id.startsWith("\0virtual:emdash/")) return null; + return virtualStubs[id.slice(1)] ?? null; + }, + }, + cloudflareTest({ + miniflare: { + compatibilityDate: "2026-05-14", + compatibilityFlags: ["nodejs_compat"], + d1Databases: ["DB"], + }, + }), + ], + test: { + include: ["tests/workerd/**/*.test.ts"], + testTimeout: 30_000, + hookTimeout: 30_000, + }, +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 78dad0ddaa..86c735d63d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1833,6 +1833,9 @@ importers: '@arethetypeswrong/cli': specifier: 'catalog:' version: 0.18.2 + '@cloudflare/vitest-pool-workers': + specifier: 'catalog:' + version: 0.16.3(@vitest/runner@4.1.5)(@vitest/snapshot@4.1.5)(vitest@4.1.5(@opentelemetry/api@1.9.0)(@types/node@24.10.13)(jsdom@26.1.0)(vite@8.0.11(@types/node@24.10.13)(esbuild@0.28.1)(jiti@2.7.0)(tsx@4.21.0)(yaml@2.9.0))) '@emdash-cms/blocks': specifier: workspace:* version: link:../blocks