diff --git a/scripts/check-adapter-inventory.ts b/scripts/check-adapter-inventory.ts index eac6727..c520019 100644 --- a/scripts/check-adapter-inventory.ts +++ b/scripts/check-adapter-inventory.ts @@ -199,7 +199,9 @@ async function main(): Promise { // 24 → 23: `browser-transfer/resumable-upload/indexeddb-checkpoint-store.ts`가 // IndexedDB 커널로 옮겨가면서 자기 abort 리스너를 지웠다. // 23 → 22: `storage/indexeddb/indexeddb-maintenance.ts`가 같은 이유로 지웠다. - const HAND_ROLLED_ABORT_CEILING = 22; + // 22 → 21: `storage/indexeddb/indexeddb-runtime.ts`가 같은 이유로 지웠다. + // 이것으로 IndexedDB 어댑터 4벌이 모두 커널을 쓴다. + const HAND_ROLLED_ABORT_CEILING = 21; const handRolledScan = spawnSync( "git", ["grep", "-l", 'addEventListener("abort"', "--", "src/adapters"], diff --git a/src/adapters/storage/indexeddb/indexeddb-runtime.ts b/src/adapters/storage/indexeddb/indexeddb-runtime.ts index 3d29f99..1a1e3c0 100644 --- a/src/adapters/storage/indexeddb/indexeddb-runtime.ts +++ b/src/adapters/storage/indexeddb/indexeddb-runtime.ts @@ -12,6 +12,7 @@ import type { IndexedDbWriteReceipt, } from "../../../application/ports/browser-file-storage/indexeddb-port.ts"; import type { + BrowserDataFailure, BrowserDataOperation, BrowserDataResult, } from "../../../application/ports/browser-file-storage/shared.ts"; @@ -20,6 +21,20 @@ import { browserDataFailure, browserDataSuccess, } from "../../browser-file-storage/result.ts"; +import { snapshotAbortTimers } from "../../platform/abortable-operation.ts"; +import { + createIndexedDbConnection, + openIndexedDbDatabase, + type IndexedDbTranslate, +} from "../../platform/indexeddb-connection.ts"; +import { + runIndexedDbTransaction, + walkIndexedDbCursor, + type IndexedDbBudget, + type IndexedDbCursorStep, + type IndexedDbCursorVisit, + type IndexedDbTransactionContext, +} from "../../platform/indexeddb-transaction.ts"; import { mapIndexedDbException } from "./indexeddb-failure.ts"; import { createIndexedDbDatasetBinding, @@ -39,7 +54,6 @@ import type { IndexedDbObservation, IndexedDbQueryPlan, IndexedDbRuntimeDependencies, - IndexedDbScheduler, } from "./indexeddb-types.ts"; type StoredRecord = Readonly<{ @@ -82,11 +96,10 @@ type ReceiptLookup = result: BrowserDataResult; }>; -type TransactionContext = Readonly<{ - succeed(value: Value): void; - fail(result: BrowserDataResult): void; - requestFailed(error: unknown): void; -}>; +type TransactionContext = IndexedDbTransactionContext< + Value, + BrowserDataFailure +>; const SAFE_DATABASE_NAME = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,63}$/u; const OPAQUE_SHA256_FINGERPRINT = /^[a-f0-9]{64}$/u; @@ -105,17 +118,6 @@ const MAX_MEASURED_RECORD_BYTES = 2_147_483_647; const MAX_QUERY_SCANNED_ROWS = 5_000; const MAX_IDEMPOTENCY_RECEIPTS = 1_000_000; -function defaultScheduler(): IndexedDbScheduler { - return Object.freeze({ - setTimeout: (callback, milliseconds) => - globalThis.setTimeout(callback, milliseconds), - clearTimeout: (handle) => - globalThis.clearTimeout( - handle as ReturnType, - ), - }); -} - function countBucket(count: number): IndexedDbCountBucket { if (count <= 0) return "0"; if (count === 1) return "1"; @@ -403,6 +405,43 @@ function unavailable( }); } +/** + * The kernel's `translate` and `context.fail` take the failure on its own while + * every helper above builds a whole `Result`. All of them only ever build the + * failure arm, so the branch below is a narrowing rather than a claim. + */ +function failureOf(result: BrowserDataResult): BrowserDataFailure { + if (result.ok) { + throw new TypeError("A browser data failure was expected."); + } + return result.error; +} + +function mappedFailure( + error: unknown, + operation: BrowserDataOperation, +): BrowserDataFailure { + return failureOf(mapIndexedDbException(error, operation)); +} + +function corruptFailure( + operation: BrowserDataOperation, +): BrowserDataFailure { + return failureOf(corruptData(operation)); +} + +function conflictFailure( + operation: BrowserDataOperation, +): BrowserDataFailure { + return failureOf(readwriteConflict(operation)); +} + +function invalidFailure( + operation: BrowserDataOperation, +): BrowserDataFailure { + return failureOf(invalidInput(operation)); +} + export function createIndexedDbRuntime( inputDependencies: IndexedDbRuntimeDependencies, ): IndexedDbRepositoryPort { @@ -572,7 +611,21 @@ export function createIndexedDbRuntime( (typeof globalThis.IDBKeyRange === "undefined" ? undefined : globalThis.IDBKeyRange); - const scheduler = dependencies.scheduler ?? defaultScheduler(); + // X-AUDIT-02. The timer callables are captured once, bound to their receiver, + // so replacing a global after composition cannot change how an open already + // in flight is bounded. The kernel throws at construction if a positive + // deadline arrives without them. + const timers = snapshotAbortTimers( + dependencies.scheduler ?? { + setTimeout: (callback: () => void, milliseconds: number): unknown => + globalThis.setTimeout(callback, milliseconds), + clearTimeout: (handle: unknown): void => { + globalThis.clearTimeout( + handle as ReturnType, + ); + }, + }, + ); const blockedTimeoutMs = dependencies.blockedTimeoutMs ?? 10_000; const nowEpochMilliseconds = dependencies.nowEpochMilliseconds ?? Date.now; @@ -589,11 +642,6 @@ export function createIndexedDbRuntime( kind: "CLOSED", reason: "NOT_OPENED", }); - let connection: IDBDatabase | null = null; - let openingRequest: IDBOpenDBRequest | null = null; - let openingPromise: Promise> | null = null; - let activeOpeningGeneration: object | null = null; - let cancelPendingOpen: (() => void) | null = null; let disposed = false; function observe(event: IndexedDbObservation): void { @@ -634,215 +682,154 @@ export function createIndexedDbRuntime( } } - function handleVersionChange(db: IDBDatabase): void { - if (connection !== db || disposed) return; - connection = null; - db.close(); - const next = Object.freeze({ - kind: "CLOSED" as const, - reason: "VERSION_CHANGE" as const, - }); - updateStatus(next); - try { - dependencies.onVersionChange?.(next); - } catch { - // Connection closure is independent from the notification callback. - } - } - - function handleForcedClose(db: IDBDatabase): void { - if (connection !== db || disposed) return; - connection = null; - updateStatus({ kind: "CLOSED", reason: "FORCED" }); - } - - function waitForOpeningAttempt( - attempt: Promise>, - signal: AbortSignal | undefined, - ): Promise> { - const operation = "INDEXEDDB_OPEN" as const; - const cancelled = abortedResult(signal, operation); - if (cancelled) return Promise.resolve(cancelled); - - return new Promise>((resolve) => { - let callerSettled = false; - const finishCaller = (result: BrowserDataResult) => { - if (callerSettled) return; - callerSettled = true; - signal?.removeEventListener("abort", onAbort); - resolve(result); - }; - function onAbort(): void { - finishCaller(browserDataFailure("ABORTED", operation)); - } - - signal?.addEventListener("abort", onAbort, { once: true }); - void attempt.then(finishCaller, () => finishCaller(unavailable(operation))); - }); - } - - function startOpeningAttempt( - availableFactory: IDBFactory, - generation: object, - ): Promise> { - const operation = "INDEXEDDB_OPEN" as const; - - return new Promise>((resolve) => { - const settleNativeRequest = () => { - if (activeOpeningGeneration !== generation) return; - activeOpeningGeneration = null; - openingRequest = null; - openingPromise = null; - cancelPendingOpen = null; - }; - let request: IDBOpenDBRequest; - try { - request = availableFactory.open( - databaseName, - dependencies.schemaVersion, - ); - } catch (error) { - const result = mapIndexedDbException(error, operation); - updateStatus({ kind: "CLOSED", reason: "NOT_OPENED" }); - resolve(result); - queueMicrotask(settleNativeRequest); - return; - } - - openingRequest = request; - let requestSettled = false; - let migrationFailed = false; - let policyBindingRejected = false; - let appliedMigrations = 0; - let blockedTimer: unknown; - - const clearBlockedTimer = () => { - if (blockedTimer !== undefined) { - scheduler.clearTimeout(blockedTimer); - blockedTimer = undefined; - } - }; - const finishOpeningAttempt = (result: BrowserDataResult) => { - if (requestSettled) return; - requestSettled = true; - clearBlockedTimer(); - resolve(result); - }; - const settleLateRequest = () => { - settleNativeRequest(); - }; - const abortUpgrade = () => { - try { - request.transaction?.abort(); - } catch { - // An open request cannot otherwise be cancelled. - } - }; - cancelPendingOpen = () => { - abortUpgrade(); - finishOpeningAttempt(unavailable(operation)); - }; - - request.onupgradeneeded = (event) => { - const transaction = request.transaction; - if (!transaction || event.newVersion === null) { - migrationFailed = true; - abortUpgrade(); - return; - } - try { - appliedMigrations = applyIndexedDbMigrations( - request.result, - transaction, - event.oldVersion, - event.newVersion, - dependencies.migrations, - ); - queueIndexedDbUpgradeBinding( - transaction, - dependencies.governanceStore, - expectedBinding, - event.oldVersion, - () => { - policyBindingRejected = true; - }, - ); - } catch { - migrationFailed = true; - abortUpgrade(); - } - }; - - request.onblocked = (event) => { - updateStatus({ - kind: "BLOCKED", - currentVersion: event.oldVersion, - targetVersion: event.newVersion ?? dependencies.schemaVersion, - }); - observe({ - operation, - outcome: "BLOCKED", - schemaVersion: dependencies.schemaVersion, - countBucket: "0", - failureCode: "BLOCKED", - }); - if (blockedTimer === undefined) { - blockedTimer = scheduler.setTimeout(() => { - finishOpeningAttempt( - browserDataFailure("BLOCKED", operation, { - retryable: true, - recovery: "RELOAD_OTHER_CONTEXTS", - }), - ); - }, blockedTimeoutMs); - } - }; - - request.onerror = () => { - settleLateRequest(); - try { - request.result.close(); - } catch { - // Failed native open requests do not expose a result. - } - if (!disposed) { - updateStatus({ kind: "CLOSED", reason: "NOT_OPENED" }); - } - if (migrationFailed) { - const result = browserDataFailure( - "MIGRATION_FAILED", - "INDEXEDDB_MIGRATE", - { recovery: "READ_ONLY" }, - ); - observeResult("INDEXEDDB_MIGRATE", result); - finishOpeningAttempt(result); - return; - } - if (policyBindingRejected) { - finishOpeningAttempt( - browserDataFailure("POLICY_REJECTED", operation, { - recovery: storagePolicySnapshot.unavailableFallback, + /** + * STO-06. The runtime keeps `mapIndexedDbException`, which maintenance and + * the OPFS journal share, rather than the checkpoint store's own table: the + * same native error is a different answer in the two families, and unifying + * them is a separate change from moving the mechanics onto the kernel. + * + * A translator per operation, because a read, a write and an open carry + * different labels. `CLOSED` is mapped explicitly: closing the handle while + * an open is in flight has always ended that open with UNAVAILABLE, and + * letting it fall in with the caller's own abort would change the code a + * caller sees when `close()` races its `open()`. + */ + function translateFor( + operation: BrowserDataOperation, + ): IndexedDbTranslate { + return (cause) => { + switch (cause.kind) { + case "NATIVE_EXCEPTION": + return mappedFailure(cause.error, operation); + case "BLOCKED": + case "BLOCKED_DEADLINE": + // The same answer whether the blocked event is itself terminal (a + // `blockedTimeoutMs` of 0) or a deadline ran out: the conflict that + // holds the old version open is the same one. + return failureOf( + browserDataFailure("BLOCKED", operation, { + retryable: true, + recovery: "RELOAD_OTHER_CONTEXTS", }), ); - return; + case "CALLER_ABORT": + return failureOf(browserDataFailure("ABORTED", operation)); + case "CLOSED": + case "NO_VALUE_PRODUCED": + // A closed handle and a transaction that committed without producing + // a value are both UNAVAILABLE here, and both are retried rather + // than reconciled. + return failureOf(unavailable(operation)); + case "UPGRADE_REJECTED": + // Any rejected upgrade is a migration failure, and it is labelled + // against INDEXEDDB_MIGRATE rather than the operation that asked to + // open. + return failureOf( + browserDataFailure("MIGRATION_FAILED", "INDEXEDDB_MIGRATE", { + recovery: "READ_ONLY", + }), + ); + case "ADMISSION_REJECTED": + // Only the runtime knows whether a rejected admission was the + // governance binding or a store the schema never created; the kernel + // keeps `detail` opaque so the two keep their own codes. + return cause.detail === "POLICY" + ? failureOf( + browserDataFailure("POLICY_REJECTED", operation, { + recovery: storagePolicySnapshot.unavailableFallback, + }), + ) + : mappedFailure(cause.detail, "INDEXEDDB_MIGRATE"); + case "UNSUPPORTED": + return failureOf( + browserDataFailure("UNSUPPORTED", operation, { + recovery: "ONLINE_ONLY", + }), + ); + default: { + const exhaustive: never = cause; + return exhaustive; } - finishOpeningAttempt(mapIndexedDbException(request.error, operation)); - }; + } + }; + } - request.onsuccess = () => { - const opened = request.result; - void (async () => { - if (requestSettled || disposed) { - settleLateRequest(); - opened.close(); - if (!disposed) { - updateStatus({ kind: "CLOSED", reason: "NOT_OPENED" }); - } - return; + /** + * The handle owns the cached connection, the single-flight open and the + * `versionchange`/`close` invalidation this file used to wire by hand. The + * status broadcast stays here because it is the runtime's own contract: the + * kernel never learns what OPENING, BLOCKED or READY mean. + */ + const connection = createIndexedDbConnection({ + translate: translateFor("INDEXEDDB_OPEN"), + open: (signal) => { + const operation = "INDEXEDDB_OPEN" as const; + if (factory === undefined) { + return Promise.resolve( + browserDataFailure("UNSUPPORTED", operation, { + recovery: "ONLINE_ONLY", + }), + ); + } + // Per attempt rather than per caller: concurrent callers join one open, + // so the migration count and the binding verdict belong to the attempt. + let appliedMigrations = 0; + let policyBindingRejected = false; + updateStatus({ + kind: "OPENING", + targetVersion: dependencies.schemaVersion, + }); + return openIndexedDbDatabase({ + factory, + databaseName, + version: dependencies.schemaVersion, + translate: translateFor(operation), + signal, + blockedTimeoutMs, + timers, + onBlocked: (event) => { + updateStatus({ + kind: "BLOCKED", + currentVersion: event.oldVersion, + targetVersion: event.newVersion ?? dependencies.schemaVersion, + }); + observe({ + operation, + outcome: "BLOCKED", + schemaVersion: dependencies.schemaVersion, + countBucket: "0", + failureCode: "BLOCKED", + }); + }, + upgrade: ({ database, transaction, oldVersion, newVersion }) => { + try { + appliedMigrations = applyIndexedDbMigrations( + database, + transaction, + oldVersion, + newVersion, + dependencies.migrations, + ); + queueIndexedDbUpgradeBinding( + transaction, + dependencies.governanceStore, + expectedBinding, + oldVersion, + () => { + policyBindingRejected = true; + }, + ); + } catch { + // No detail: a migration that threw and a versionchange this + // runtime never planned are the same MIGRATION_FAILED answer. + return { kind: "REJECTED" }; } + return { kind: "APPLIED" }; + }, + admit: async (database) => { try { assertIndexedDbRuntimeStores( - opened, + database, dependencies.recordStore, dependencies.governanceStore, dependencies.retentionStore, @@ -852,80 +839,90 @@ export function createIndexedDbRuntime( dependencies.idempotencyExpiryIndex, ); } catch (error) { - settleLateRequest(); - opened.close(); - updateStatus({ kind: "CLOSED", reason: "NOT_OPENED" }); - const result = mapIndexedDbException( - error, - "INDEXEDDB_MIGRATE", - ); - observeResult("INDEXEDDB_MIGRATE", result); - finishOpeningAttempt(result); - return; + return { kind: "REJECT", detail: error }; } + // No signal: an open the caller gave up on is already ended by the + // kernel's own abort path. const binding = await verifyIndexedDbDatasetBinding( - opened, + database, dependencies.governanceStore, expectedBinding, undefined, ); - settleLateRequest(); - if (!binding.ok) { - opened.close(); - if (!disposed) { - updateStatus({ kind: "CLOSED", reason: "NOT_OPENED" }); - } - if (!requestSettled) { - finishOpeningAttempt( - binding.reason === "ABORTED" - ? browserDataFailure("ABORTED", operation) - : binding.reason === "NATIVE_ERROR" - ? mapIndexedDbException(binding.error, operation) - : browserDataFailure( - "POLICY_REJECTED", - operation, - { - recovery: - storagePolicySnapshot.unavailableFallback, - }, - ), - ); - } - return; + if (binding.ok) return { kind: "ADMIT" }; + return binding.reason === "ABORTED" + ? { kind: "FAIL", cause: { kind: "CALLER_ABORT" } } + : binding.reason === "NATIVE_ERROR" + ? { + kind: "FAIL", + cause: { kind: "NATIVE_EXCEPTION", error: binding.error }, + } + : { kind: "REJECT", detail: "POLICY" }; + }, + }).then((opened) => { + // `queueIndexedDbUpgradeBinding` reports a mismatch from inside a + // request handler, which runs after `upgrade` has already returned. + // The upgrade transaction aborts and the open then fails as an + // AbortError, so the rejection is applied here instead — and a + // migration failure still outranks it, exactly as the hand-written + // `onerror` ordered the two. + const result: BrowserDataResult = + !opened.ok && + policyBindingRejected && + opened.error.code !== "MIGRATION_FAILED" + ? browserDataFailure("POLICY_REJECTED", operation, { + recovery: storagePolicySnapshot.unavailableFallback, + }) + : opened; + if (disposed) return result; + if (!result.ok) { + updateStatus({ kind: "CLOSED", reason: "NOT_OPENED" }); + // A failed migration is observed against INDEXEDDB_MIGRATE as well + // as against the open the caller asked for. + if (result.error.operation === "INDEXEDDB_MIGRATE") { + observeResult("INDEXEDDB_MIGRATE", result); } - if (requestSettled || disposed) { - opened.close(); - if (!disposed) { - updateStatus({ kind: "CLOSED", reason: "NOT_OPENED" }); - } - return; - } - connection = opened; - opened.onversionchange = () => handleVersionChange(opened); - opened.onclose = () => handleForcedClose(opened); - updateStatus({ - kind: "READY", - schemaVersion: opened.version, - }); - if (appliedMigrations > 0) { - observeResult( - "INDEXEDDB_MIGRATE", - browserDataSuccess(undefined), - appliedMigrations, - ); - } - finishOpeningAttempt(browserDataSuccess(undefined)); - })(); - }; - }); - } + return result; + } + updateStatus({ + kind: "READY", + schemaVersion: result.value.version, + }); + if (appliedMigrations > 0) { + observeResult( + "INDEXEDDB_MIGRATE", + browserDataSuccess(undefined), + appliedMigrations, + ); + } + return result; + }); + }, + onVersionChange: () => { + // The kernel has already closed the connection and dropped the cached + // handle, so only the announcement is left here. + const next = Object.freeze({ + kind: "CLOSED" as const, + reason: "VERSION_CHANGE" as const, + }); + updateStatus(next); + try { + dependencies.onVersionChange?.(next); + } catch { + // Connection closure is independent from the notification callback. + } + }, + onForcedClose: () => { + updateStatus({ kind: "CLOSED", reason: "FORCED" }); + }, + }); async function open( signal?: AbortSignal, ): Promise> { const operation = "INDEXEDDB_OPEN" as const; if (disposed) return observeResult(operation, unavailable(operation)); - if (connection) { + if (connection.current()) { return observeResult(operation, browserDataSuccess(undefined)); } const cancelled = abortedResult(signal, operation); @@ -938,41 +935,24 @@ export function createIndexedDbRuntime( }), ); } - - if (!openingPromise) { - updateStatus({ - kind: "OPENING", - targetVersion: dependencies.schemaVersion, - }); - const generation = {}; - activeOpeningGeneration = generation; - const attempt = startOpeningAttempt(factory, generation); - openingPromise = attempt; - } - - const result = await waitForOpeningAttempt(openingPromise, signal); - return observeResult(operation, result); - } - - function createTransaction( - db: IDBDatabase, - stores: readonly string[], - mode: IDBTransactionMode, - ): IDBTransaction { - const durability = - mode === "readonly" - ? dependencies.durability?.read ?? "default" - : dependencies.durability?.write ?? "strict"; - try { - return db.transaction([...stores], mode, { durability }); - } catch (error) { - if (error instanceof TypeError) { - return db.transaction([...stores], mode); - } - throw error; - } + const acquired = await connection.acquire(signal); + return observeResult( + operation, + acquired.ok ? browserDataSuccess(undefined) : acquired, + ); } + /** + * Durability is named explicitly on both sides, so a read stays on the + * engine default while a write is strict: a record that is only queued when + * the tab goes away would leave a receipt claiming a revision that was never + * stored. The kernel falls back to the no-options form on an engine that + * rejects the bag. + * + * The caller's abort is checked before the connection so a cancelled call + * reports ABORTED rather than UNAVAILABLE, which is the order this file has + * always used. + */ function runTransaction( stores: readonly string[], mode: "readonly" | "readwrite", @@ -985,91 +965,20 @@ export function createIndexedDbRuntime( ): Promise> { const cancelled = abortedResult(signal, operation); if (cancelled) return Promise.resolve(cancelled); - const db = connection; + const db = connection.current(); if (!db || disposed) return Promise.resolve(unavailable(operation)); - let transaction: IDBTransaction; - try { - transaction = createTransaction(db, stores, mode); - } catch (error) { - return Promise.resolve(mapIndexedDbException(error, operation)); - } - - return new Promise>((resolve) => { - let candidate: BrowserDataResult | undefined; - let requestError: unknown; - let callerAborted = false; - let settled = false; - - const finish = (result: BrowserDataResult) => { - if (settled) return; - settled = true; - signal?.removeEventListener("abort", onAbort); - resolve(result); - }; - function onAbort(): void { - callerAborted = true; - try { - transaction.abort(); - } catch { - // If complete won the race, its result remains authoritative. - } - } - - transaction.oncomplete = () => { - finish(candidate ?? unavailable(operation)); - }; - transaction.onerror = () => { - requestError ??= transaction.error; - }; - transaction.onabort = () => { - if (callerAborted) { - finish(browserDataFailure("ABORTED", operation)); - return; - } - if (candidate && !candidate.ok) { - finish(candidate); - return; - } - finish( - mapIndexedDbException( - requestError ?? transaction.error, - operation, - ), - ); - }; - signal?.addEventListener("abort", onAbort, { once: true }); - - const context: TransactionContext = Object.freeze({ - succeed(value) { - if (!candidate) candidate = browserDataSuccess(value); - }, - fail(result) { - if (!candidate) candidate = result; - try { - transaction.abort(); - } catch { - // Completion or a prior abort already owns the result. - } - }, - requestFailed(error) { - requestError ??= error; - if (!candidate) { - candidate = mapIndexedDbException(error, operation); - } - }, - }); - - try { - queue(transaction, context); - } catch (error) { - context.fail(mapIndexedDbException(error, operation)); - try { - transaction.abort(); - } catch { - finish(candidate ?? unavailable(operation)); - } - } + return runIndexedDbTransaction({ + database: db, + stores, + mode, + translate: translateFor(operation), + signal, + durability: + mode === "readonly" + ? dependencies.durability?.read ?? "default" + : dependencies.durability?.write ?? "strict", + queue, }); } @@ -1138,7 +1047,7 @@ export function createIndexedDbRuntime( } const decoded = decodeRecord(request.result, key); if (!decoded.ok) { - context.fail(decoded); + context.fail(failureOf(decoded)); return; } const retentionRequest = transaction @@ -1153,14 +1062,14 @@ export function createIndexedDbRuntime( key, ) ) { - context.fail(corruptData(operation)); + context.fail(corruptFailure(operation)); return; } if ( storagePolicySnapshot.retention.kind === "TTL" && retentionRequest.result.eligibleAtEpochMs === undefined ) { - context.fail(corruptData(operation)); + context.fail(corruptFailure(operation)); return; } if ( @@ -1169,9 +1078,8 @@ export function createIndexedDbRuntime( retentionRequest.result.eligibleAtEpochMs <= readEpochMs ) { context.fail( - browserDataFailure( - "EXPIRED_RESOURCE", - operation, + failureOf( + browserDataFailure("EXPIRED_RESOURCE", operation), ), ); return; @@ -1369,162 +1277,152 @@ export function createIndexedDbRuntime( let lastCursor: IndexedDbCursor | null = null; let lastScannedCursor: IndexedDbCursor | null = null; let scannedRows = 0; + // The two ways a page stops short resume from different places: a scan + // that ran out of budget resumes from the last row it looked at, while + // a full page resumes from the last row it kept. + let stoppedAtScanLimit = false; - request.onerror = () => context.requestFailed(request.error); - request.onsuccess = () => { - if (signal?.aborted) { - try { - transaction.abort(); - } catch { - // Transaction completion decides the race. + walkIndexedDbCursor({ + request, + sink: context, + translate: translateFor(operation), + // No budget: the page is bounded by rows scanned and rows kept, + // both of which are decided against the caller's plan below. + signal, + visit: ({ cursor: nativeCursor, resume }) => { + if ( + !isBoundedCursorKey(nativeCursor.key) || + !isBoundedCursorKey(nativeCursor.primaryKey) + ) { + context.fail(corruptFailure(operation)); + // `fail` aborts, so the pump has nowhere to go. Suspending + // without ever resuming says that without asking for another + // row or claiming a summary the scan never reached. + return { kind: "SUSPEND" }; } - return; - } - const nativeCursor = request.result; - if (!nativeCursor) { - context.succeed( - Object.freeze({ - items: Object.freeze(items), - nextCursor: null, - }), - ); - return; - } - if ( - !isBoundedCursorKey(nativeCursor.key) || - !isBoundedCursorKey(nativeCursor.primaryKey) - ) { - context.fail(corruptData(operation)); - return; - } - const currentCursor: IndexedDbCursor = Object.freeze({ - indexKey: normalizeCursorKey(nativeCursor.key), - primaryKey: normalizeCursorKey(nativeCursor.primaryKey), - }); - if ( - validated.value.cursor && - factory && - !cursorAfter( - factory, - currentCursor, - validated.value.cursor, - direction, - ) - ) { - try { + const currentCursor: IndexedDbCursor = Object.freeze({ + indexKey: normalizeCursorKey(nativeCursor.key), + primaryKey: normalizeCursorKey(nativeCursor.primaryKey), + }); + if ( + validated.value.cursor && + factory && + !cursorAfter( + factory, + currentCursor, + validated.value.cursor, + direction, + ) + ) { + // Rows at or before the caller's cursor are skipped rather than + // counted, so a resumed page never returns what it already did. + // A throwing `cmp` is reported by the pump, which maps it the + // same way the hand-written catch did. const sameIndexKey = factory.cmp( nativeCursorKey(currentCursor.indexKey), - nativeCursorKey( - validated.value.cursor.indexKey, - ), + nativeCursorKey(validated.value.cursor.indexKey), ) === 0; const samePrimaryKey = factory.cmp( nativeCursorKey(currentCursor.primaryKey), - nativeCursorKey( - validated.value.cursor.primaryKey, - ), + nativeCursorKey(validated.value.cursor.primaryKey), ) === 0; if (sameIndexKey && samePrimaryKey) { - nativeCursor.continue(); - } else if (validated.value.plan.index) { - nativeCursor.continuePrimaryKey( - nativeCursorKey( - validated.value.cursor.indexKey, - ), - nativeCursorKey( - validated.value.cursor.primaryKey, - ), - ); - } else { - nativeCursor.continue( - nativeCursorKey( - validated.value.cursor.primaryKey, - ), - ); + return { kind: "CONTINUE" }; } - } catch (error) { - context.fail( - mapIndexedDbException(error, operation), - ); + return validated.value.plan.index + ? { + kind: "CONTINUE_PRIMARY", + key: nativeCursorKey( + validated.value.cursor.indexKey, + ), + primaryKey: nativeCursorKey( + validated.value.cursor.primaryKey, + ), + } + : { + kind: "CONTINUE_FROM", + key: nativeCursorKey( + validated.value.cursor.primaryKey, + ), + }; } - return; - } - if (scannedRows >= scanLimit) { + if (scannedRows >= scanLimit) { + stoppedAtScanLimit = true; + return { kind: "STOP" }; + } + if (items.length >= validated.value.plan.limit) { + return { kind: "STOP" }; + } + if (!isStoredRecord(nativeCursor.value)) { + context.fail(corruptFailure(operation)); + return { kind: "SUSPEND" }; + } + const decoded = decodeRecord( + nativeCursor.value, + nativeCursor.value.key, + ); + if (!decoded.ok) { + context.fail(failureOf(decoded)); + return { kind: "SUSPEND" }; + } + scannedRows += 1; + lastScannedCursor = currentCursor; + const recordKey = nativeCursor.value.key; + const retentionRequest = transaction + .objectStore(dependencies.retentionStore) + .get(recordKey); + retentionRequest.onerror = () => + context.requestFailed(retentionRequest.error); + retentionRequest.onsuccess = () => { + if ( + !isStoredRetentionRecord( + retentionRequest.result, + recordKey, + ) + ) { + context.fail(corruptFailure(operation)); + return; + } + const expired = + storagePolicySnapshot.retention.kind === "TTL" && + retentionRequest.result.eligibleAtEpochMs !== + undefined && + retentionRequest.result.eligibleAtEpochMs <= + queryEpochMs; + if ( + storagePolicySnapshot.retention.kind === "TTL" && + retentionRequest.result.eligibleAtEpochMs === undefined + ) { + context.fail(corruptFailure(operation)); + return; + } + if (!expired) { + items.push(decoded.value.value); + lastCursor = currentCursor; + } + resume({ kind: "CONTINUE" }); + }; + // A row only counts once its retention sidecar has been read, so + // the cursor stays parked until `resume`. + return { kind: "SUSPEND" }; + }, + done: (summary) => { + if (summary.reason === "ABORTED") return; context.succeed( Object.freeze({ items: Object.freeze(items), - nextCursor: lastScannedCursor, + nextCursor: + summary.reason === "EXHAUSTED" + ? null + : stoppedAtScanLimit + ? lastScannedCursor + : lastCursor, }), ); - return; - } - if (items.length >= validated.value.plan.limit) { - context.succeed( - Object.freeze({ - items: Object.freeze(items), - nextCursor: lastCursor, - }), - ); - return; - } - if (!isStoredRecord(nativeCursor.value)) { - context.fail(corruptData(operation)); - return; - } - const decoded = decodeRecord( - nativeCursor.value, - nativeCursor.value.key, - ); - if (!decoded.ok) { - context.fail(decoded); - return; - } - scannedRows += 1; - lastScannedCursor = currentCursor; - const recordKey = nativeCursor.value.key; - const retentionRequest = transaction - .objectStore(dependencies.retentionStore) - .get(recordKey); - retentionRequest.onerror = () => - context.requestFailed(retentionRequest.error); - retentionRequest.onsuccess = () => { - if ( - !isStoredRetentionRecord( - retentionRequest.result, - recordKey, - ) - ) { - context.fail(corruptData(operation)); - return; - } - const expired = - storagePolicySnapshot.retention.kind === "TTL" && - retentionRequest.result.eligibleAtEpochMs !== - undefined && - retentionRequest.result.eligibleAtEpochMs <= - queryEpochMs; - if ( - storagePolicySnapshot.retention.kind === "TTL" && - retentionRequest.result.eligibleAtEpochMs === undefined - ) { - context.fail(corruptData(operation)); - return; - } - if (!expired) { - items.push(decoded.value.value); - lastCursor = currentCursor; - } - try { - nativeCursor.continue(); - } catch (error) { - context.fail( - mapIndexedDbException(error, operation), - ); - } - }; - }; + }, + }); }, ); return observeResult(operation, result, result.ok ? result.value.items.length : 0); @@ -1772,9 +1670,7 @@ export function createIndexedDbRuntime( try { recordRequest = records.get(input.key); } catch (error) { - context.fail( - mapIndexedDbException(error, operation), - ); + context.fail(mappedFailure(error, operation)); return; } recordRequest.onerror = () => @@ -1782,7 +1678,7 @@ export function createIndexedDbRuntime( recordRequest.onsuccess = () => { const raw = recordRequest.result; if (raw !== undefined && !isStoredRecord(raw, input.key)) { - context.fail(corruptData(operation)); + context.fail(corruptFailure(operation)); return; } if ( @@ -1791,12 +1687,12 @@ export function createIndexedDbRuntime( (raw === undefined || raw.revision !== input.expectedRevision)) ) { - context.fail(readwriteConflict(operation)); + context.fail(conflictFailure(operation)); return; } const revision = raw === undefined ? 1 : raw.revision + 1; if (!Number.isSafeInteger(revision)) { - context.fail(corruptData(operation)); + context.fail(corruptFailure(operation)); return; } const sidecarRequest = retention.get(input.key); @@ -1813,7 +1709,7 @@ export function createIndexedDbRuntime( input.key, )) ) { - context.fail(corruptData(operation)); + context.fail(corruptFailure(operation)); return; } const budgetRequest = governance.get( @@ -1823,7 +1719,7 @@ export function createIndexedDbRuntime( context.requestFailed(budgetRequest.error); budgetRequest.onsuccess = () => { if (!isStoredDatasetBudget(budgetRequest.result)) { - context.fail(corruptData(operation)); + context.fail(corruptFailure(operation)); return; } const previousBytes = @@ -1850,11 +1746,13 @@ export function createIndexedDbRuntime( storagePolicySnapshot.hardBudgetBytes || receiptCount > dependencies.maxIdempotencyReceipts - ? browserDataFailure( - "LIMIT_EXCEEDED", - operation, + ? failureOf( + browserDataFailure( + "LIMIT_EXCEEDED", + operation, + ), ) - : corruptData(operation), + : corruptFailure(operation), ); return; } @@ -1868,9 +1766,7 @@ export function createIndexedDbRuntime( try { writeRequest = records.put(stored); } catch (error) { - context.fail( - mapIndexedDbException(error, operation), - ); + context.fail(mappedFailure(error, operation)); return; } writeRequest.onerror = () => @@ -1957,7 +1853,7 @@ export function createIndexedDbRuntime( if (replay.result.ok) { context.succeed(replay.result.value); } else { - context.fail(replay.result); + context.fail(failureOf(replay.result)); } return; } @@ -1968,9 +1864,7 @@ export function createIndexedDbRuntime( input.idempotencyKey, ); } catch (error) { - context.fail( - mapIndexedDbException(error, operation), - ); + context.fail(mappedFailure(error, operation)); return; } deleteRequest.onerror = () => @@ -2045,9 +1939,7 @@ export function createIndexedDbRuntime( try { recordRequest = records.get(input.key); } catch (error) { - context.fail( - mapIndexedDbException(error, operation), - ); + context.fail(mappedFailure(error, operation)); return; } recordRequest.onerror = () => @@ -2057,7 +1949,7 @@ export function createIndexedDbRuntime( !isStoredRecord(recordRequest.result, input.key) || recordRequest.result.revision !== input.expectedRevision ) { - context.fail(readwriteConflict(operation)); + context.fail(conflictFailure(operation)); return; } const revision = input.expectedRevision + 1; @@ -2071,7 +1963,7 @@ export function createIndexedDbRuntime( input.key, ) ) { - context.fail(corruptData(operation)); + context.fail(corruptFailure(operation)); return; } const previousBytes = @@ -2086,7 +1978,7 @@ export function createIndexedDbRuntime( !isStoredDatasetBudget(budgetRequest.result) || budgetRequest.result.usedBytes < previousBytes ) { - context.fail(corruptData(operation)); + context.fail(corruptFailure(operation)); return; } const usedBytes = @@ -2099,9 +1991,11 @@ export function createIndexedDbRuntime( dependencies.maxIdempotencyReceipts ) { context.fail( - browserDataFailure( - "LIMIT_EXCEEDED", - operation, + failureOf( + browserDataFailure( + "LIMIT_EXCEEDED", + operation, + ), ), ); return; @@ -2171,7 +2065,7 @@ export function createIndexedDbRuntime( if (replay.result.ok) { context.succeed(replay.result.value); } else { - context.fail(replay.result); + context.fail(failureOf(replay.result)); } return; } @@ -2182,9 +2076,7 @@ export function createIndexedDbRuntime( input.idempotencyKey, ); } catch (error) { - context.fail( - mapIndexedDbException(error, operation), - ); + context.fail(mappedFailure(error, operation)); return; } deleteRequest.onerror = () => @@ -2318,10 +2210,12 @@ export function createIndexedDbRuntime( (transaction, context) => { if (!keyRange) { context.fail( - browserDataFailure( - "UNSUPPORTED", - "INDEXEDDB_WRITE", - { recovery: "ONLINE_ONLY" }, + failureOf( + browserDataFailure( + "UNSUPPORTED", + "INDEXEDDB_WRITE", + { recovery: "ONLINE_ONLY" }, + ), ), ); return; @@ -2339,7 +2233,7 @@ export function createIndexedDbRuntime( try { range = keyRange.upperBound(cutoffEpochMs); } catch { - context.fail(invalidInput("INDEXEDDB_WRITE")); + context.fail(invalidFailure("INDEXEDDB_WRITE")); return; } const request = retention @@ -2359,117 +2253,115 @@ export function createIndexedDbRuntime( budgetExhausted, }), ); - request.onerror = () => - context.requestFailed(request.error); - request.onsuccess = () => { - if (input.signal?.aborted) { - try { - transaction.abort(); - } catch { - // The transaction completion event decides the race. + // The sweep bounds on rows it deleted, not on rows it read, so the + // counters stay here while the kernel drives the cursor. The time + // check comes first so a batch that ran out of both still reports + // `budgetExhausted`, exactly as the hand-written check did. + const budget: IndexedDbBudget = { + admit: () => { + const clock = monotonicClock(); + if (!clock.ok) return clock; + if (clock.value >= deadline) { + return browserDataSuccess("TIME_BUDGET" as const); } - return; - } - const cursor = request.result; - if (!cursor) { - succeed("COMPLETE", false); - return; - } - const clock = monotonicClock(); - if (!clock.ok) { - context.fail(clock); - return; - } - if ( - deletedRows >= input.maxRows || - clock.value >= deadline - ) { - succeed("MORE", clock.value >= deadline); - return; - } - if ( - !isStoredRetentionRecord( - cursor.value, - String(cursor.primaryKey), - ) || - cursor.value.eligibleAtEpochMs === undefined || - cursor.value.eligibleAtEpochMs > cutoffEpochMs || - (storagePolicySnapshot.retention.kind === - "UNTIL_SYNCED" && - cursor.value.synchronization !== "CONFIRMED") - ) { - context.fail(corruptData("INDEXEDDB_WRITE")); - return; - } - scannedRows += 1; - const measuredBytes = cursor.value.measuredBytes; - let recordDelete: IDBRequest; - try { - recordDelete = records.delete(cursor.value.recordKey); - } catch (error) { - context.fail( - mapIndexedDbException(error, "INDEXEDDB_WRITE"), + return browserDataSuccess( + deletedRows >= input.maxRows + ? ("ROW_BUDGET" as const) + : ("CONTINUE" as const), ); - return; - } - recordDelete.onerror = () => - context.requestFailed(recordDelete.error); - recordDelete.onsuccess = () => { - let sidecarDelete: IDBRequest; - try { - sidecarDelete = retention.delete( - cursor.value.recordKey, - ); - } catch (error) { - context.fail( - mapIndexedDbException(error, "INDEXEDDB_WRITE"), - ); - return; + }, + }; + walkIndexedDbCursor({ + request, + sink: context, + translate: translateFor("INDEXEDDB_WRITE"), + budget, + signal: input.signal, + visit: ({ cursor, resume }) => { + if ( + !isStoredRetentionRecord( + cursor.value, + String(cursor.primaryKey), + ) || + cursor.value.eligibleAtEpochMs === undefined || + cursor.value.eligibleAtEpochMs > cutoffEpochMs || + (storagePolicySnapshot.retention.kind === + "UNTIL_SYNCED" && + cursor.value.synchronization !== "CONFIRMED") + ) { + context.fail(corruptFailure("INDEXEDDB_WRITE")); + // `fail` aborts, so the pump has nowhere to go. Suspending + // without ever resuming says that without claiming a summary + // the sweep never reached. + return { kind: "SUSPEND" }; } - sidecarDelete.onerror = () => - context.requestFailed(sidecarDelete.error); - sidecarDelete.onsuccess = () => { - const budgetRequest = governance.get( - INDEXEDDB_DATASET_BUDGET_KEY, - ); - budgetRequest.onerror = () => - context.requestFailed(budgetRequest.error); - budgetRequest.onsuccess = () => { - if ( - !isStoredDatasetBudget(budgetRequest.result) || - budgetRequest.result.usedBytes < measuredBytes - ) { - context.fail(corruptData("INDEXEDDB_WRITE")); - return; - } - const storedBudgetRequest = governance.put({ - bindingKey: INDEXEDDB_DATASET_BUDGET_KEY, - budgetVersion: 1, - usedBytes: - budgetRequest.result.usedBytes - measuredBytes, - receiptCount: budgetRequest.result.receiptCount, - } satisfies StoredDatasetBudget); - storedBudgetRequest.onerror = () => - context.requestFailed( - storedBudgetRequest.error, - ); - storedBudgetRequest.onsuccess = () => { - deletedRows += 1; - try { - cursor.continue(); - } catch (error) { - context.fail( - mapIndexedDbException( - error, - "INDEXEDDB_WRITE", - ), - ); + scannedRows += 1; + const measuredBytes = cursor.value.measuredBytes; + let recordDelete: IDBRequest; + try { + recordDelete = records.delete(cursor.value.recordKey); + } catch (error) { + context.fail(mappedFailure(error, "INDEXEDDB_WRITE")); + return { kind: "SUSPEND" }; + } + recordDelete.onerror = () => + context.requestFailed(recordDelete.error); + recordDelete.onsuccess = () => { + let sidecarDelete: IDBRequest; + try { + sidecarDelete = retention.delete( + cursor.value.recordKey, + ); + } catch (error) { + context.fail(mappedFailure(error, "INDEXEDDB_WRITE")); + return; + } + sidecarDelete.onerror = () => + context.requestFailed(sidecarDelete.error); + sidecarDelete.onsuccess = () => { + const budgetRequest = governance.get( + INDEXEDDB_DATASET_BUDGET_KEY, + ); + budgetRequest.onerror = () => + context.requestFailed(budgetRequest.error); + budgetRequest.onsuccess = () => { + if ( + !isStoredDatasetBudget(budgetRequest.result) || + budgetRequest.result.usedBytes < measuredBytes + ) { + context.fail(corruptFailure("INDEXEDDB_WRITE")); + return; } + const storedBudgetRequest = governance.put({ + bindingKey: INDEXEDDB_DATASET_BUDGET_KEY, + budgetVersion: 1, + usedBytes: + budgetRequest.result.usedBytes - measuredBytes, + receiptCount: budgetRequest.result.receiptCount, + } satisfies StoredDatasetBudget); + storedBudgetRequest.onerror = () => + context.requestFailed( + storedBudgetRequest.error, + ); + storedBudgetRequest.onsuccess = () => { + deletedRows += 1; + resume({ kind: "CONTINUE" }); + }; }; }; }; - }; - }; + // A row is only counted once its whole delete chain has been + // queued, so the cursor stays parked until `resume`. + return { kind: "SUSPEND" }; + }, + done: (summary) => { + if (summary.reason === "ABORTED") return; + succeed( + summary.reason === "EXHAUSTED" ? "COMPLETE" : "MORE", + summary.reason === "TIME_BUDGET", + ); + }, + }); }, ); } @@ -2517,6 +2409,52 @@ export function createIndexedDbRuntime( }), ); + // The purge bounds on rows it deleted, not on rows it read, so the + // counters stay here. The time check comes first so a pass that ran + // out of both still reports `budgetExhausted`. + const budget: IndexedDbBudget = { + admit: () => { + const clock = monotonicClock(); + if (!clock.ok) return clock; + if (clock.value >= deadline) { + return browserDataSuccess("TIME_BUDGET" as const); + } + return browserDataSuccess( + deletedRows >= input.maxRows + ? ("ROW_BUDGET" as const) + : ("CONTINUE" as const), + ); + }, + }; + + /** + * The four passes differ only in which rows they delete and in what + * runs once a store is empty, so the budget and the "this pass ran + * out" receipt are shared. No signal is passed: the partition purge + * has never checked one per row, and the transaction's own abort + * listener already ends a cancelled batch. + */ + const walkPurge = ( + request: IDBRequest, + visit: (visit: IndexedDbCursorVisit) => IndexedDbCursorStep, + exhausted: () => void, + ) => { + walkIndexedDbCursor({ + request, + sink: context, + translate: translateFor("INDEXEDDB_WRITE"), + budget, + visit, + done: (summary) => { + if (summary.reason === "EXHAUSTED") { + exhausted(); + return; + } + succeed("MORE", summary.reason === "TIME_BUDGET"); + }, + }); + }; + const processMetadataStore = (storeIndex: number) => { const storeName = dependencies.lifecycleMetadataStores[storeIndex]; @@ -2525,78 +2463,88 @@ export function createIndexedDbRuntime( return; } const store = transaction.objectStore(storeName); - const request = store.openCursor(); - request.onerror = () => - context.requestFailed(request.error); - request.onsuccess = () => { - const cursor = request.result; - if (!cursor) { - processMetadataStore(storeIndex + 1); - return; - } - const clock = monotonicClock(); - if (!clock.ok) { - context.fail(clock); - return; - } - if ( - deletedRows >= input.maxRows || - clock.value >= deadline - ) { - succeed("MORE", clock.value >= deadline); - return; - } - scannedRows += 1; - const deletion = store.delete(cursor.primaryKey); - deletion.onerror = () => - context.requestFailed(deletion.error); - deletion.onsuccess = () => { - deletedRows += 1; - try { - cursor.continue(); - } catch (error) { - context.fail( - mapIndexedDbException( - error, - "INDEXEDDB_WRITE", - ), - ); - } - }; - }; + walkPurge( + store.openCursor(), + ({ cursor, resume }) => { + scannedRows += 1; + const deletion = store.delete(cursor.primaryKey); + deletion.onerror = () => + context.requestFailed(deletion.error); + deletion.onsuccess = () => { + deletedRows += 1; + resume({ kind: "CONTINUE" }); + }; + // The row only counts once its deletion is queued, so the + // cursor stays parked until `resume`. + return { kind: "SUSPEND" }; + }, + () => processMetadataStore(storeIndex + 1), + ); }; const processReceipts = () => { - const request = receipts.openCursor(); - request.onerror = () => - context.requestFailed(request.error); - request.onsuccess = () => { - const cursor = request.result; - if (!cursor) { - succeed("COMPLETE", false); - return; - } - const clock = monotonicClock(); - if (!clock.ok) { - context.fail(clock); - return; - } - if ( - deletedRows >= input.maxRows || - clock.value >= deadline - ) { - succeed("MORE", clock.value >= deadline); - return; - } - if (!isStoredReceipt(cursor.value)) { - context.fail(corruptData("INDEXEDDB_WRITE")); - return; - } - scannedRows += 1; - const deletion = receipts.delete(cursor.primaryKey); - deletion.onerror = () => - context.requestFailed(deletion.error); - deletion.onsuccess = () => { + walkPurge( + receipts.openCursor(), + ({ cursor, resume }) => { + if (!isStoredReceipt(cursor.value)) { + context.fail(corruptFailure("INDEXEDDB_WRITE")); + // `fail` aborts, so the pump has nowhere to go. + return { kind: "SUSPEND" }; + } + scannedRows += 1; + const deletion = receipts.delete(cursor.primaryKey); + deletion.onerror = () => + context.requestFailed(deletion.error); + deletion.onsuccess = () => { + const budgetRequest = governance.get( + INDEXEDDB_DATASET_BUDGET_KEY, + ); + budgetRequest.onerror = () => + context.requestFailed(budgetRequest.error); + budgetRequest.onsuccess = () => { + if ( + !isStoredDatasetBudget(budgetRequest.result) || + budgetRequest.result.receiptCount < 1 + ) { + context.fail(corruptFailure("INDEXEDDB_WRITE")); + return; + } + const storedBudgetRequest = governance.put({ + ...budgetRequest.result, + receiptCount: + budgetRequest.result.receiptCount - 1, + } satisfies StoredDatasetBudget); + storedBudgetRequest.onerror = () => + context.requestFailed( + storedBudgetRequest.error, + ); + storedBudgetRequest.onsuccess = () => { + deletedRows += 1; + resume({ kind: "CONTINUE" }); + }; + }; + }; + return { kind: "SUSPEND" }; + }, + () => succeed("COMPLETE", false), + ); + }; + + const processOrphanedRetention = () => { + walkPurge( + retention.openCursor(), + ({ cursor, resume }) => { + if ( + !isStoredRetentionRecord( + cursor.value, + String(cursor.primaryKey), + ) + ) { + context.fail(corruptFailure("INDEXEDDB_WRITE")); + return { kind: "SUSPEND" }; + } + scannedRows += 1; + const measuredBytes = cursor.value.measuredBytes; const budgetRequest = governance.get( INDEXEDDB_DATASET_BUDGET_KEY, ); @@ -2605,208 +2553,104 @@ export function createIndexedDbRuntime( budgetRequest.onsuccess = () => { if ( !isStoredDatasetBudget(budgetRequest.result) || - budgetRequest.result.receiptCount < 1 + budgetRequest.result.usedBytes < measuredBytes ) { - context.fail(corruptData("INDEXEDDB_WRITE")); + context.fail(corruptFailure("INDEXEDDB_WRITE")); return; } - const storedBudgetRequest = governance.put({ - ...budgetRequest.result, - receiptCount: - budgetRequest.result.receiptCount - 1, - } satisfies StoredDatasetBudget); - storedBudgetRequest.onerror = () => - context.requestFailed( - storedBudgetRequest.error, - ); - storedBudgetRequest.onsuccess = () => { - deletedRows += 1; - try { - cursor.continue(); - } catch (error) { - context.fail( - mapIndexedDbException( - error, - "INDEXEDDB_WRITE", - ), - ); - } + const deletion = retention.delete(cursor.primaryKey); + deletion.onerror = () => + context.requestFailed(deletion.error); + deletion.onsuccess = () => { + const budgetWrite = governance.put({ + ...budgetRequest.result, + usedBytes: + budgetRequest.result.usedBytes - measuredBytes, + } satisfies StoredDatasetBudget); + budgetWrite.onerror = () => + context.requestFailed(budgetWrite.error); + budgetWrite.onsuccess = () => { + deletedRows += 1; + resume({ kind: "CONTINUE" }); + }; }; }; - }; - }; + return { kind: "SUSPEND" }; + }, + () => processMetadataStore(0), + ); }; - const processOrphanedRetention = () => { - const request = retention.openCursor(); - request.onerror = () => - context.requestFailed(request.error); - request.onsuccess = () => { - const cursor = request.result; - if (!cursor) { - processMetadataStore(0); - return; - } - const clock = monotonicClock(); - if (!clock.ok) { - context.fail(clock); - return; - } - if ( - deletedRows >= input.maxRows || - clock.value >= deadline - ) { - succeed("MORE", clock.value >= deadline); - return; - } - if ( - !isStoredRetentionRecord( - cursor.value, - String(cursor.primaryKey), - ) - ) { - context.fail(corruptData("INDEXEDDB_WRITE")); - return; + walkPurge( + records.openCursor(), + ({ cursor, resume }) => { + if (!isStoredRecord(cursor.value)) { + context.fail(corruptFailure("INDEXEDDB_WRITE")); + return { kind: "SUSPEND" }; } scannedRows += 1; - const measuredBytes = cursor.value.measuredBytes; - const budgetRequest = governance.get( - INDEXEDDB_DATASET_BUDGET_KEY, - ); - budgetRequest.onerror = () => - context.requestFailed(budgetRequest.error); - budgetRequest.onsuccess = () => { + const key = cursor.value.key; + const sidecarRequest = retention.get(key); + sidecarRequest.onerror = () => + context.requestFailed(sidecarRequest.error); + sidecarRequest.onsuccess = () => { if ( - !isStoredDatasetBudget(budgetRequest.result) || - budgetRequest.result.usedBytes < measuredBytes + !isStoredRetentionRecord( + sidecarRequest.result, + key, + ) ) { - context.fail(corruptData("INDEXEDDB_WRITE")); + context.fail(corruptFailure("INDEXEDDB_WRITE")); return; } - const deletion = retention.delete(cursor.primaryKey); - deletion.onerror = () => - context.requestFailed(deletion.error); - deletion.onsuccess = () => { - const budgetWrite = governance.put({ - ...budgetRequest.result, - usedBytes: - budgetRequest.result.usedBytes - measuredBytes, - } satisfies StoredDatasetBudget); - budgetWrite.onerror = () => - context.requestFailed(budgetWrite.error); - budgetWrite.onsuccess = () => { - deletedRows += 1; - try { - cursor.continue(); - } catch (error) { - context.fail( - mapIndexedDbException( - error, - "INDEXEDDB_WRITE", - ), - ); - } - }; - }; - }; - }; - }; - - const request = records.openCursor(); - request.onerror = () => - context.requestFailed(request.error); - request.onsuccess = () => { - const cursor = request.result; - if (!cursor) { - processOrphanedRetention(); - return; - } - const clock = monotonicClock(); - if (!clock.ok) { - context.fail(clock); - return; - } - if ( - deletedRows >= input.maxRows || - clock.value >= deadline - ) { - succeed("MORE", clock.value >= deadline); - return; - } - if (!isStoredRecord(cursor.value)) { - context.fail(corruptData("INDEXEDDB_WRITE")); - return; - } - scannedRows += 1; - const key = cursor.value.key; - const sidecarRequest = retention.get(key); - sidecarRequest.onerror = () => - context.requestFailed(sidecarRequest.error); - sidecarRequest.onsuccess = () => { - if ( - !isStoredRetentionRecord( - sidecarRequest.result, - key, - ) - ) { - context.fail(corruptData("INDEXEDDB_WRITE")); - return; - } - const measuredBytes = - sidecarRequest.result.measuredBytes; - const budgetRequest = governance.get( - INDEXEDDB_DATASET_BUDGET_KEY, - ); - budgetRequest.onerror = () => - context.requestFailed(budgetRequest.error); - budgetRequest.onsuccess = () => { - if ( - !isStoredDatasetBudget(budgetRequest.result) || - budgetRequest.result.usedBytes < measuredBytes - ) { - context.fail(corruptData("INDEXEDDB_WRITE")); - return; - } - const recordDelete = records.delete(key); - recordDelete.onerror = () => - context.requestFailed(recordDelete.error); - recordDelete.onsuccess = () => { - const sidecarDelete = retention.delete(key); - sidecarDelete.onerror = () => - context.requestFailed(sidecarDelete.error); - sidecarDelete.onsuccess = () => { - const storedBudgetRequest = governance.put({ - bindingKey: - INDEXEDDB_DATASET_BUDGET_KEY, - budgetVersion: 1, - usedBytes: - budgetRequest.result.usedBytes - - measuredBytes, - receiptCount: - budgetRequest.result.receiptCount, - } satisfies StoredDatasetBudget); - storedBudgetRequest.onerror = () => - context.requestFailed( - storedBudgetRequest.error, - ); - storedBudgetRequest.onsuccess = () => { - deletedRows += 1; - try { - cursor.continue(); - } catch (error) { - context.fail( - mapIndexedDbException( - error, - "INDEXEDDB_WRITE", - ), + const measuredBytes = + sidecarRequest.result.measuredBytes; + const budgetRequest = governance.get( + INDEXEDDB_DATASET_BUDGET_KEY, + ); + budgetRequest.onerror = () => + context.requestFailed(budgetRequest.error); + budgetRequest.onsuccess = () => { + if ( + !isStoredDatasetBudget(budgetRequest.result) || + budgetRequest.result.usedBytes < measuredBytes + ) { + context.fail(corruptFailure("INDEXEDDB_WRITE")); + return; + } + const recordDelete = records.delete(key); + recordDelete.onerror = () => + context.requestFailed(recordDelete.error); + recordDelete.onsuccess = () => { + const sidecarDelete = retention.delete(key); + sidecarDelete.onerror = () => + context.requestFailed(sidecarDelete.error); + sidecarDelete.onsuccess = () => { + const storedBudgetRequest = governance.put({ + bindingKey: INDEXEDDB_DATASET_BUDGET_KEY, + budgetVersion: 1, + usedBytes: + budgetRequest.result.usedBytes - + measuredBytes, + receiptCount: + budgetRequest.result.receiptCount, + } satisfies StoredDatasetBudget); + storedBudgetRequest.onerror = () => + context.requestFailed( + storedBudgetRequest.error, ); - } + storedBudgetRequest.onsuccess = () => { + deletedRows += 1; + resume({ kind: "CONTINUE" }); + }; }; }; }; }; - }; - }; + return { kind: "SUSPEND" }; + }, + () => processOrphanedRetention(), + ); }, ); } @@ -2873,12 +2717,10 @@ export function createIndexedDbRuntime( function close(): void { if (disposed) return; disposed = true; - cancelPendingOpen?.(); - cancelPendingOpen = null; - openingRequest = null; - const current = connection; - connection = null; - current?.close(); + // Idempotent in the kernel: it closes the cached connection, cancels an + // open still in flight and settles that open's waiters with CLOSED, which + // `translateFor` maps to UNAVAILABLE rather than ABORTED. + connection.close(); updateStatus({ kind: "DISPOSED" }); subscribers.clear(); } diff --git a/tests/unit/indexeddb-runtime.test.ts b/tests/unit/indexeddb-runtime.test.ts index af3b02a..792316f 100644 --- a/tests/unit/indexeddb-runtime.test.ts +++ b/tests/unit/indexeddb-runtime.test.ts @@ -152,6 +152,33 @@ function dependencies( }; } +/** + * Counts the aborts the adapter asks for. A failure code on its own cannot + * tell `fail`, which aborts, from `requestFailed`, which only records: both + * report the same code once the transaction unwinds, and only the abort + * decides whether the writes already queued can still commit. + */ +function trackAborts( + memory: MemoryIndexedDbFactory, +): Readonly<{ count: number }> { + const tracker = { count: 0 }; + let current = memory.lastTransaction; + Object.defineProperty(memory, "lastTransaction", { + configurable: true, + get: () => current, + set: (transaction: MemoryIndexedDbFactory["lastTransaction"]) => { + current = transaction; + if (!transaction) return; + const abort = transaction.abort.bind(transaction); + transaction.abort = () => { + tracker.count += 1; + abort(); + }; + }, + }); + return tracker; +} + async function waitForTransaction( memory: MemoryIndexedDbFactory, ): Promise { @@ -464,7 +491,7 @@ describe("IndexedDB runtime", () => { }); }); - it("retains native open ownership after a blocked timeout until late success", async () => { + it("reopens after a blocked deadline and ignores the abandoned request", async () => { const memory = new MemoryIndexedDbFactory(); memory.blockNextOpen(); const nativeOpen = vi.spyOn(memory.factory, "open"); @@ -489,23 +516,59 @@ describe("IndexedDB runtime", () => { ok: false, error: { code: "BLOCKED" }, }); - - const second = runtime.open(); - await Promise.resolve(); - const nativeOpenCountBeforeLateSuccess = nativeOpen.mock.calls.length; - memory.releaseBlockedOpen(); - - await expect(second).resolves.toMatchObject({ - ok: false, - error: { code: "BLOCKED" }, - }); - await Promise.resolve(); - expect(nativeOpenCountBeforeLateSuccess).toBe(1); - expect(memory.isConnectionClosed()).toBe(true); expect(runtime.getStatus()).toEqual({ kind: "CLOSED", reason: "NOT_OPENED", }); + + // The deadline ends the caller's wait, not the native request. A retry + // therefore dispatches a second open instead of answering from the + // abandoned one, which is the one behavior the connection kernel does not + // preserve: it cannot report when an open it already settled lands. + expect(await runtime.open()).toEqual({ ok: true, value: undefined }); + expect(nativeOpen).toHaveBeenCalledTimes(2); + expect(runtime.getStatus()).toEqual({ + kind: "READY", + schemaVersion: 1, + }); + + // The abandoned request lands late. Its connection is closed inside the + // kernel and must not disturb the one the retry established. + memory.releaseBlockedOpen(); + await Promise.resolve(); + expect(runtime.getStatus()).toEqual({ + kind: "READY", + schemaVersion: 1, + }); + expect(await runtime.read("late-arrival")).toEqual({ + ok: true, + value: null, + }); + }); + + it("ends an open in flight with UNAVAILABLE when the runtime closes", async () => { + const memory = new MemoryIndexedDbFactory(); + memory.blockNextOpen(); + const runtime = createIndexedDbRuntime(dependencies(memory)); + + const pending = runtime.open(); + await Promise.resolve(); + expect(runtime.getStatus()).toMatchObject({ kind: "BLOCKED" }); + runtime.close(); + + // Closing the runtime is not the caller aborting. The open is reported as + // UNAVAILABLE/REOPEN so a caller builds a new runtime rather than reading + // its own cancellation into someone else's shutdown. + expect(await pending).toEqual({ + ok: false, + error: { + code: "UNAVAILABLE", + operation: "INDEXEDDB_OPEN", + retryable: true, + recovery: "REOPEN", + }, + }); + expect(runtime.getStatus()).toEqual({ kind: "DISPOSED" }); }); it("closes immediately on versionchange and isolates listener failures", async () => { @@ -735,7 +798,24 @@ describe("IndexedDB runtime", () => { recovery: "READ_ONLY", }, }); - expect(memory.isConnectionClosed()).toBe(true); + // Rejecting the upgrade aborts the versionchange transaction, so the + // half-created schema is rolled back instead of being committed at + // version 1. Only this assertion separates a rejected upgrade from an + // upgrade that was allowed to commit and then refused at admission — + // `NotFoundError` maps to MIGRATION_FAILED either way. + expect([...memory.storeNames()]).toEqual([]); + // A failed open request never exposes a connection, so fail-closed is + // asserted through the runtime rather than through the fake's last + // database handle: no status claiming a live schema, and no read that can + // reach one. + expect(runtime.getStatus()).toEqual({ + kind: "CLOSED", + reason: "NOT_OPENED", + }); + expect(await runtime.read("any-key")).toMatchObject({ + ok: false, + error: { code: "UNAVAILABLE", recovery: "REOPEN" }, + }); }); it("derives physical identity only from opaque scope and rejects an override", () => { @@ -902,6 +982,117 @@ describe("IndexedDB runtime", () => { ).not.toContain("ttl_authority_proof_001"); }); + it("reports a batch that ran out of both budgets as time-exhausted", async () => { + const memory = new MemoryIndexedDbFactory(); + const base = dependencies(memory); + // The start of the batch and the first row are inside the deadline; the + // second row is not, and by then the row budget is spent as well. + let monotonicCalls = 0; + const runtime = createIndexedDbRuntime({ + ...base, + storagePolicy: { + ...TEST_STORAGE_POLICY, + retention: { kind: "SESSION" }, + }, + authorizeLifecycle: () => ({ + authorized: true as const, + proofToken: "budget_authority_proof_001", + }), + nowMonotonicMilliseconds: () => (monotonicCalls++ < 2 ? 0 : 5_000), + }); + expect(await runtime.open()).toMatchObject({ ok: true }); + for (const key of ["budget-one", "budget-two"]) { + expect( + await runtime.compareAndSwap({ + key, + value: { label: key, rank: 1 }, + expectedRevision: null, + idempotencyKey: `${key}-create`, + }), + ).toMatchObject({ ok: true }); + } + + // Both budgets are spent at the same row. The receipt says the deadline + // did it, because a caller that reads `budgetExhausted` as "the clock ran + // out" would otherwise schedule the next batch as if it had room. + expect( + await runtime.enforceLifecycleBatch({ + action: "SESSION_END", + maxRows: 1, + maxDurationMs: 1_000, + }), + ).toEqual({ + ok: true, + value: { + state: "MORE", + scannedRows: 1, + deletedRows: 1, + budgetExhausted: true, + }, + }); + }); + + it("aborts a sweep whose budget row cannot account for the deleted bytes", async () => { + const memory = new MemoryIndexedDbFactory(); + let epochMs = 1_000; + const runtime = createIndexedDbRuntime( + dependencies(memory, { + storagePolicy: { + ...TEST_STORAGE_POLICY, + retention: { kind: "TTL", maxAgeMs: 10 }, + }, + keyRange: + memory.keyRange as NonNullable< + IndexedDbRuntimeDependencies["keyRange"] + >, + nowEpochMilliseconds: () => epochMs, + nowMonotonicMilliseconds: () => 0, + authorizeLifecycle: () => ({ + authorized: true, + proofToken: "sweep_authority_proof_001", + }), + }), + ); + expect(await runtime.open()).toMatchObject({ ok: true }); + expect( + await runtime.compareAndSwap({ + key: "sweep-record", + value: { label: "expires", rank: 1 }, + expectedRevision: null, + idempotencyKey: "sweep-create", + }), + ).toMatchObject({ ok: true }); + + // A budget that cannot account for the record's bytes is corrupt, and the + // sweep only discovers it after both deletions have already succeeded + // inside the transaction. + memory.seed("governance", { + bindingKey: "dataset-budget", + budgetVersion: 1, + usedBytes: 0, + receiptCount: 1, + }); + const aborts = trackAborts(memory); + epochMs = 1_011; + + expect( + await runtime.enforceLifecycleBatch({ + action: "RETENTION_SWEEP", + maxRows: 10, + maxDurationMs: 1_000, + }), + ).toMatchObject({ + ok: false, + error: { code: "CORRUPT_DATA" }, + }); + // `fail` aborts. Recording the failure without aborting would report the + // same code while letting the two deletions commit against a budget that + // never moved. + expect(aborts.count).toBe(1); + expect(memory.readRaw("records", "sweep-record")).toBeDefined(); + expect(memory.readRaw("retention", "sweep-record")).toBeDefined(); + }); + it("requires lifecycle authority and only deletes sync-confirmed rows", async () => { const memory = new MemoryIndexedDbFactory(); const authorizeLifecycle = vi