From 3366a81f0fe532289beb9effd0b7e4a2cbb9f32b Mon Sep 17 00:00:00 2001 From: DongHyeonka Date: Wed, 16 Sep 2026 19:35:22 +0900 Subject: [PATCH] refactor: move IndexedDB maintenance onto the kernel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1558 -> 1430줄. 손수 abort 리스너가 0이 되어 래칫을 23에서 22로 조인다. 보존한 동작 셋: - blockedTimeoutMs를 넘기지 않는다. blocked 이벤트는 유지보수에게 종단이다 — 배치는 opt-in 백그라운드 패스라 다른 컨텍스트가 사라지길 기대하며 데드라인만큼 매다는 비용이 BLOCKED를 보고하고 재시도시키는 것보다 크다 - upgrade 콜백을 생략한다. 커널은 생략을 "어떤 upgrade든 예상 밖"으로 읽고 거절하는데, 그게 유지보수가 늘 해온 동작이다. 빈 APPLIED를 넣으면 잘못된 스키마로 열린다 - onversionchange는 admit 안, 바인딩 검증 앞에 등록한다 계획이 blocked 회귀 테스트로 지목한 indexeddb-maintenance.test.ts:289-290은 실제로는 onblocked 경로가 아니라 drain 검사였다. MT의 진짜 blocked 경로에는 테스트가 없었으므로 새로 썼다. Co-Authored-By: Claude Opus 5 (1M context) --- scripts/check-adapter-inventory.ts | 3 +- .../indexeddb/indexeddb-maintenance.ts | 770 ++++++++---------- tests/unit/indexeddb-maintenance.test.ts | 174 +++- 3 files changed, 494 insertions(+), 453 deletions(-) diff --git a/scripts/check-adapter-inventory.ts b/scripts/check-adapter-inventory.ts index 331f20b..eac6727 100644 --- a/scripts/check-adapter-inventory.ts +++ b/scripts/check-adapter-inventory.ts @@ -198,7 +198,8 @@ async function main(): Promise { // // 24 → 23: `browser-transfer/resumable-upload/indexeddb-checkpoint-store.ts`가 // IndexedDB 커널로 옮겨가면서 자기 abort 리스너를 지웠다. - const HAND_ROLLED_ABORT_CEILING = 23; + // 23 → 22: `storage/indexeddb/indexeddb-maintenance.ts`가 같은 이유로 지웠다. + const HAND_ROLLED_ABORT_CEILING = 22; const handRolledScan = spawnSync( "git", ["grep", "-l", 'addEventListener("abort"', "--", "src/adapters"], diff --git a/src/adapters/storage/indexeddb/indexeddb-maintenance.ts b/src/adapters/storage/indexeddb/indexeddb-maintenance.ts index 2667384..deada7a 100644 --- a/src/adapters/storage/indexeddb/indexeddb-maintenance.ts +++ b/src/adapters/storage/indexeddb/indexeddb-maintenance.ts @@ -5,6 +5,7 @@ import type { IndexedDbReceiptPruneBatchReceipt, } from "../../../application/ports/browser-file-storage/indexeddb-port.ts"; import type { + BrowserDataFailure, BrowserDataResult, } from "../../../application/ports/browser-file-storage/shared.ts"; import { @@ -12,6 +13,17 @@ import { browserDataFailure, browserDataSuccess, } from "../../browser-file-storage/result.ts"; +import { + openIndexedDbDatabase, + type IndexedDbTranslate, +} from "../../platform/indexeddb-connection.ts"; +import { + openIndexedDbTransaction, + runIndexedDbTransaction, + walkIndexedDbCursor, + type IndexedDbBudget, + type IndexedDbTransactionContext, +} from "../../platform/indexeddb-transaction.ts"; import { mapIndexedDbException } from "./indexeddb-failure.ts"; import { createIndexedDbDatasetBinding, @@ -94,12 +106,6 @@ type PreparedRecord = Readonly<{ measuredBytes: number | undefined; }>; -type TransactionContext = Readonly<{ - succeed(value: Value): void; - fail(result: BrowserDataResult): void; - requestFailed(error: unknown): void; -}>; - const SAFE_IDENTIFIER = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/u; const OPAQUE_SHA256_FINGERPRINT = /^[a-f0-9]{64}$/u; const MAX_BATCH_ROWS = 500; @@ -261,6 +267,41 @@ function unavailable(): BrowserDataResult { }); } +function unsupported(): BrowserDataResult { + return browserDataFailure("UNSUPPORTED", "INDEXEDDB_MIGRATE", { + recovery: "ONLINE_ONLY", + }); +} + +/** + * `browserDataFailure` and `mapIndexedDbException` build a `Result`, while the + * kernel's `translate` and `context.fail` want the failure on its own. Both + * 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 invalidInputFailure(): BrowserDataFailure { + return failureOf(invalidInput()); +} + +function migrationFailure(): BrowserDataFailure { + return failureOf(migrationFailed()); +} + +function unsupportedFailure(): BrowserDataFailure { + return failureOf(unsupported()); +} + +function mappedFailure(error: unknown): BrowserDataFailure { + return failureOf(mapIndexedDbException(error, "INDEXEDDB_MIGRATE")); +} + function defaultNow(): number { return typeof globalThis.performance === "undefined" ? Date.now() @@ -420,6 +461,59 @@ export function createIndexedDbMaintenance( } } + /** + * STO-06. Maintenance keeps `mapIndexedDbException`, which the runtime 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. + * + * One translator covers the whole adapter because the operation label is the + * constant `INDEXEDDB_MIGRATE` here — the runtime needs one per operation. + */ + const translate: IndexedDbTranslate = (cause) => { + switch (cause.kind) { + case "NATIVE_EXCEPTION": + return mappedFailure(cause.error); + case "BLOCKED": + case "BLOCKED_DEADLINE": + return failureOf( + browserDataFailure("BLOCKED", "INDEXEDDB_MIGRATE", { + retryable: true, + recovery: "RELOAD_OTHER_CONTEXTS", + }), + ); + case "CALLER_ABORT": + return failureOf( + browserDataFailure("ABORTED", "INDEXEDDB_MIGRATE"), + ); + case "CLOSED": + case "NO_VALUE_PRODUCED": + // A transaction that committed without producing a value is retried, + // not reconciled. `CLOSED` cannot reach here — maintenance opens a + // connection per batch instead of holding a cached handle — and is + // mapped alongside it so the union stays total. + return failureOf(unavailable()); + case "UPGRADE_REJECTED": + // Maintenance owns no schema. Any upgrade means the database is not + // the one this batch was configured against. + return migrationFailure(); + case "ADMISSION_REJECTED": + return cause.detail === "POLICY" + ? failureOf( + browserDataFailure("POLICY_REJECTED", "INDEXEDDB_MIGRATE", { + recovery: storagePolicySnapshot.unavailableFallback, + }), + ) + : migrationFailure(); + case "UNSUPPORTED": + return unsupportedFailure(); + default: { + const exhaustive: never = cause; + return exhaustive; + } + } + }; + function validBatchInput( input: IndexedDbMaintenanceBatchInput, ): boolean { @@ -436,173 +530,76 @@ export function createIndexedDbMaintenance( ): Promise> { const cancelled = abortedResult(signal, "INDEXEDDB_MIGRATE"); if (cancelled) return Promise.resolve(cancelled); - if (!factory) { - return Promise.resolve( - browserDataFailure("UNSUPPORTED", "INDEXEDDB_MIGRATE", { - recovery: "ONLINE_ONLY", - }), - ); - } + if (!factory) return Promise.resolve(unsupported()); - return new Promise>((resolve) => { - let request: IDBOpenDBRequest; - try { - request = factory.open( - databaseName, - dependencies.schemaVersion, - ); - } catch (error) { - resolve(mapIndexedDbException(error, "INDEXEDDB_MIGRATE")); - return; - } - - let settled = false; - let unexpectedUpgrade = false; - const finish = (result: BrowserDataResult) => { - if (settled) return; - settled = true; - signal?.removeEventListener("abort", onAbort); - resolve(result); - }; - function onAbort(): void { - try { - request.transaction?.abort(); - } catch { - // A pending non-upgrade open request cannot be cancelled. - } - finish(browserDataFailure("ABORTED", "INDEXEDDB_MIGRATE")); - } - signal?.addEventListener("abort", onAbort, { once: true }); - - request.onupgradeneeded = () => { - unexpectedUpgrade = true; - try { - request.transaction?.abort(); - } catch { - // The error handler below owns the closed failure result. - } - }; - request.onblocked = () => { - finish( - browserDataFailure("BLOCKED", "INDEXEDDB_MIGRATE", { - retryable: true, - recovery: "RELOAD_OTHER_CONTEXTS", - }), - ); - }; - request.onerror = () => { - finish( - unexpectedUpgrade - ? migrationFailed() - : mapIndexedDbException( - request.error, - "INDEXEDDB_MIGRATE", - ), - ); - }; - request.onsuccess = () => { - const database = request.result; - if (settled) { - database.close(); - return; - } - if ( - !database.objectStoreNames.contains( - dependencies.recordStore, - ) || - !database.objectStoreNames.contains( - dependencies.governanceStore, - ) || - !database.objectStoreNames.contains( - dependencies.retentionStore, - ) || - !database.objectStoreNames.contains( - dependencies.checkpointStore, - ) || - !database.objectStoreNames.contains( - dependencies.idempotencyStore, - ) - ) { - database.close(); - finish(migrationFailed()); - return; + return openIndexedDbDatabase({ + factory, + databaseName, + version: dependencies.schemaVersion, + translate, + signal, + // No `blockedTimeoutMs`, and therefore no `timers`. The blocked event is + // terminal for maintenance: a batch is an opt-in background pass, so + // hanging it for a deadline in the hope another context goes away costs + // more than reporting BLOCKED and letting the caller retry. + // + // No `upgrade` either. The kernel reads an omitted callback as "any + // upgrade is unexpected" and rejects it, which is what maintenance has + // always done: it migrates records under a schema somebody else owns and + // must never create or change one. + admit: async (database) => { + for (const store of [ + dependencies.recordStore, + dependencies.governanceStore, + dependencies.retentionStore, + dependencies.checkpointStore, + dependencies.idempotencyStore, + ]) { + if (!database.objectStoreNames.contains(store)) { + return { kind: "REJECT", detail: "STORE" }; + } } try { - const transaction = database.transaction( - dependencies.idempotencyStore, + openIndexedDbTransaction( + database, + [dependencies.idempotencyStore], "readonly", - ); - transaction + ) .objectStore(dependencies.idempotencyStore) .index(dependencies.idempotencyExpiryIndex); } catch { - database.close(); - finish(migrationFailed()); - return; + return { kind: "REJECT", detail: "INDEX" }; } + // Maintenance never holds the connection past a batch, so the listener + // belongs here rather than on a cached handle: `admit` runs on the + // successful open path only, which is exactly the window the batch + // owns the connection for. database.onversionchange = () => database.close(); - void (async () => { - const binding = await verifyIndexedDbDatasetBinding( - database, - dependencies.governanceStore, - expectedBinding, - signal, - ); - if (!binding.ok) { - database.close(); - if (!settled) { - finish( - binding.reason === "ABORTED" - ? browserDataFailure( - "ABORTED", - "INDEXEDDB_MIGRATE", - ) - : binding.reason === "NATIVE_ERROR" - ? mapIndexedDbException( - binding.error, - "INDEXEDDB_MIGRATE", - ) - : browserDataFailure( - "POLICY_REJECTED", - "INDEXEDDB_MIGRATE", - { - recovery: - storagePolicySnapshot.unavailableFallback, - }, - ), - ); - } - return; - } - if (settled) { - database.close(); - return; - } - finish(browserDataSuccess(database)); - })(); - }; + const binding = await verifyIndexedDbDatasetBinding( + database, + dependencies.governanceStore, + expectedBinding, + signal, + ); + 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" }; + }, }); } - function createTransaction( - database: IDBDatabase, - stores: readonly string[], - mode: "readonly" | "readwrite", - ): IDBTransaction { - const durability = - mode === "readonly" - ? dependencies.durability?.read ?? "default" - : dependencies.durability?.write ?? "strict"; - try { - return database.transaction([...stores], mode, { durability }); - } catch (error) { - if (error instanceof TypeError) { - return database.transaction([...stores], mode); - } - throw error; - } - } - + /** + * Durability is named explicitly on both sides because maintenance rewrites + * records: a write that is only queued when the tab goes away would leave a + * checkpoint claiming rows that were never stored. The kernel falls back to + * the no-options form on an engine that rejects the bag. + */ function runTransaction( database: IDBDatabase, stores: readonly string[], @@ -610,97 +607,20 @@ export function createIndexedDbMaintenance( signal: AbortSignal | undefined, queue: ( transaction: IDBTransaction, - context: TransactionContext, + context: IndexedDbTransactionContext, ) => void, ): Promise> { - const cancelled = abortedResult(signal, "INDEXEDDB_MIGRATE"); - if (cancelled) return Promise.resolve(cancelled); - - let transaction: IDBTransaction; - try { - transaction = createTransaction(database, stores, mode); - } catch (error) { - return Promise.resolve( - mapIndexedDbException(error, "INDEXEDDB_MIGRATE"), - ); - } - - 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); - }; - const abortTransaction = () => { - try { - transaction.abort(); - } catch { - // Completion or a prior abort already owns the result. - } - }; - function onAbort(): void { - callerAborted = true; - abortTransaction(); - } - - transaction.oncomplete = () => { - finish(candidate ?? unavailable()); - }; - transaction.onerror = () => { - requestError ??= transaction.error; - }; - transaction.onabort = () => { - if (callerAborted) { - finish( - browserDataFailure("ABORTED", "INDEXEDDB_MIGRATE"), - ); - return; - } - if (candidate && !candidate.ok) { - finish(candidate); - return; - } - finish( - mapIndexedDbException( - requestError ?? transaction.error, - "INDEXEDDB_MIGRATE", - ), - ); - }; - signal?.addEventListener("abort", onAbort, { once: true }); - - const context: TransactionContext = Object.freeze({ - succeed(value) { - if (!candidate) candidate = browserDataSuccess(value); - }, - fail(result) { - if (!candidate) candidate = result; - abortTransaction(); - }, - requestFailed(error) { - requestError ??= error; - if (!candidate) { - candidate = mapIndexedDbException( - error, - "INDEXEDDB_MIGRATE", - ); - } - }, - }); - - try { - queue(transaction, context); - } catch (error) { - context.fail( - mapIndexedDbException(error, "INDEXEDDB_MIGRATE"), - ); - } + return runIndexedDbTransaction({ + database, + stores, + mode, + translate, + signal, + durability: + mode === "readonly" + ? dependencies.durability?.read ?? "default" + : dependencies.durability?.write ?? "strict", + queue, }); } @@ -732,7 +652,7 @@ export function createIndexedDbMaintenance( dependencies.checkpointKey, ) ) { - context.fail(migrationFailed()); + context.fail(migrationFailure()); return; } const persisted = request.result; @@ -777,85 +697,69 @@ export function createIndexedDbMaintenance( true, ); } catch { - context.fail(invalidInput()); + context.fail(invalidInputFailure()); return; } if (checkpoint.effective.lastKey !== null && !query) { - context.fail( - browserDataFailure( - "UNSUPPORTED", - "INDEXEDDB_MIGRATE", - { recovery: "ONLINE_ONLY" }, - ), - ); + context.fail(unsupportedFailure()); return; } const rows: ScannedRecord[] = []; - const request = transaction - .objectStore(dependencies.recordStore) - .openCursor(query); - request.onerror = () => context.requestFailed(request.error); - request.onsuccess = () => { - if (input.signal?.aborted) { - try { - transaction.abort(); - } catch { - // The transaction event decides the abort/complete race. + // The scan bounds on rows it read, so the counter the kernel offers is + // the right one here. The time check comes first so a batch that ran + // out of both answers `budgetExhausted` the way it does today. + const budget: IndexedDbBudget = { + admit: (scannedRows) => { + const currentTime = clock(); + if (!currentTime.ok) return currentTime; + if (currentTime.value >= deadline) { + return browserDataSuccess("TIME_BUDGET" as const); } - return; - } - const cursor = request.result; - if (!cursor) { - context.succeed({ - rows: Object.freeze(rows), - reachedEnd: true, - budgetExhausted: false, - }); - return; - } - const currentTime = clock(); - if (!currentTime.ok) { - context.fail(currentTime); - return; - } - if ( - rows.length >= input.maxRows || - currentTime.value >= deadline - ) { - context.succeed({ - rows: Object.freeze(rows), - reachedEnd: false, - budgetExhausted: - currentTime.value >= deadline, - }); - return; - } - if (!isStoredRecord(cursor.value)) { - context.fail(migrationFailed()); - return; - } - if ( - cursor.value.codecVersion > - dependencies.migrationPolicy.targetCodecVersion - ) { - context.fail(migrationFailed()); - return; - } - rows.push({ - key: cursor.value.key, - codecVersion: cursor.value.codecVersion, - revision: cursor.value.revision, - payload: cursor.value.payload, - }); - try { - cursor.continue(); - } catch (error) { - context.fail( - mapIndexedDbException(error, "INDEXEDDB_MIGRATE"), + return browserDataSuccess( + scannedRows >= input.maxRows + ? ("ROW_BUDGET" as const) + : ("CONTINUE" as const), ); - } + }, }; + walkIndexedDbCursor({ + request: transaction + .objectStore(dependencies.recordStore) + .openCursor(query), + sink: context, + translate, + budget, + signal: input.signal, + visit: ({ cursor }) => { + if ( + !isStoredRecord(cursor.value) || + cursor.value.codecVersion > + dependencies.migrationPolicy.targetCodecVersion + ) { + context.fail(migrationFailure()); + // `fail` aborts, so the walk has nowhere to go. Suspending + // without ever resuming says that without asking the pump for + // another row or claiming a summary the scan never reached. + return { kind: "SUSPEND" }; + } + rows.push({ + key: cursor.value.key, + codecVersion: cursor.value.codecVersion, + revision: cursor.value.revision, + payload: cursor.value.payload, + }); + return { kind: "CONTINUE" }; + }, + done: (summary) => { + if (summary.reason === "ABORTED") return; + context.succeed({ + rows: Object.freeze(rows), + reachedEnd: summary.reason === "EXHAUSTED", + budgetExhausted: summary.reason === "TIME_BUDGET", + }); + }, + }); }, ); } @@ -1023,9 +927,7 @@ export function createIndexedDbMaintenance( try { request = checkpoints.put(stored); } catch (error) { - context.fail( - mapIndexedDbException(error, "INDEXEDDB_MIGRATE"), - ); + context.fail(mappedFailure(error)); return; } request.onerror = () => @@ -1057,7 +959,7 @@ export function createIndexedDbMaintenance( const currentTime = clock(); if (!currentTime.ok) { // A broken clock aborts rather than committing an unbounded batch. - context.fail(currentTime); + context.fail(currentTime.error); return; } if (currentTime.value >= deadline) { @@ -1069,9 +971,7 @@ export function createIndexedDbMaintenance( try { request = records.get(preparedRecord.source.key); } catch (error) { - context.fail( - mapIndexedDbException(error, "INDEXEDDB_MIGRATE"), - ); + context.fail(mappedFailure(error)); return; } request.onerror = () => @@ -1091,7 +991,7 @@ export function createIndexedDbMaintenance( preparedRecord.source.key, ) ) { - context.fail(migrationFailed()); + context.fail(migrationFailure()); return; } if ( @@ -1123,7 +1023,7 @@ export function createIndexedDbMaintenance( !preparedRecord.needsMigration || preparedRecord.measuredBytes === undefined ) { - context.fail(migrationFailed()); + context.fail(migrationFailure()); return; } const measuredBytes = preparedRecord.measuredBytes; @@ -1137,7 +1037,7 @@ export function createIndexedDbMaintenance( live.key, ) ) { - context.fail(migrationFailed()); + context.fail(migrationFailure()); return; } const previousSidecar = sidecarRequest.result; @@ -1152,7 +1052,7 @@ export function createIndexedDbMaintenance( budgetRequest.result.usedBytes < previousSidecar.measuredBytes ) { - context.fail(migrationFailed()); + context.fail(migrationFailure()); return; } const usedBytes = @@ -1165,7 +1065,7 @@ export function createIndexedDbMaintenance( usedBytes > storagePolicySnapshot.hardBudgetBytes ) { - context.fail(migrationFailed()); + context.fail(migrationFailure()); return; } const writeRequest = records.put({ @@ -1221,7 +1121,7 @@ export function createIndexedDbMaintenance( dependencies.checkpointKey, ) ) { - context.fail(migrationFailed()); + context.fail(migrationFailure()); return; } liveCheckpoint = checkpointRequest.result; @@ -1412,134 +1312,106 @@ export function createIndexedDbMaintenance( let range: IDBKeyRange; try { if (!keyRange) { - context.fail( - browserDataFailure( - "UNSUPPORTED", - "INDEXEDDB_MIGRATE", - { recovery: "ONLINE_ONLY" }, - ), - ); + context.fail(unsupportedFailure()); return; } range = keyRange.upperBound(cutoff.value); } catch { - context.fail(invalidInput()); + context.fail(invalidInputFailure()); return; } - const request = store - .index(dependencies.idempotencyExpiryIndex) - .openCursor(range); - let scannedRows = 0; let deletedRows = 0; - - const succeed = ( - state: "MORE" | "COMPLETE", - budgetExhausted: boolean, - ) => { - context.succeed( - Object.freeze({ - state, - scannedRows, - deletedRows, - budgetExhausted, - }), - ); - }; - - request.onerror = () => - context.requestFailed(request.error); - request.onsuccess = () => { - if (input.signal?.aborted) { - try { - transaction.abort(); - } catch { - // The transaction event owns the completion race. + // The prune bounds on rows it deleted, not on rows it read, so the + // counter stays here while the kernel counts scanned rows for the + // receipt. A row past the cutoff is corrupt data, not a stop + // condition, so the walk never reports STOPPED. + const budget: IndexedDbBudget = { + admit: () => { + const currentTime = clock(); + if (!currentTime.ok) return currentTime; + if (currentTime.value >= deadline) { + return browserDataSuccess("TIME_BUDGET" as const); } - return; - } - const cursor = request.result; - if (!cursor) { - succeed("COMPLETE", false); - return; - } - const currentTime = clock(); - if (!currentTime.ok) { - context.fail(currentTime); - return; - } - if ( - deletedRows >= input.maxRows || - currentTime.value >= deadline - ) { - succeed( - "MORE", - currentTime.value >= deadline, + return browserDataSuccess( + deletedRows >= input.maxRows + ? ("ROW_BUDGET" as const) + : ("CONTINUE" as const), ); - return; - } - if ( - !isStoredReceipt(cursor.value) || - cursor.value.idempotencyKey !== - String(cursor.primaryKey) || - cursor.value.expiresAtEpochMs > cutoff.value - ) { - context.fail(migrationFailed()); - return; - } - scannedRows += 1; - let deleteRequest: IDBRequest; - try { - deleteRequest = store.delete( - cursor.primaryKey, - ); - } catch (error) { - context.fail( - mapIndexedDbException( - error, - "INDEXEDDB_MIGRATE", - ), - ); - return; - } - deleteRequest.onerror = () => - context.requestFailed(deleteRequest.error); - deleteRequest.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(migrationFailed()); - return; - } - const budgetWrite = governance.put({ - ...budgetRequest.result, - receiptCount: - budgetRequest.result.receiptCount - 1, - } satisfies StoredDatasetBudget); - budgetWrite.onerror = () => - context.requestFailed(budgetWrite.error); - budgetWrite.onsuccess = () => { - deletedRows += 1; - try { - cursor.continue(); - } catch (error) { - context.fail( - mapIndexedDbException( - error, - "INDEXEDDB_MIGRATE", - ), - ); + }, + }; + walkIndexedDbCursor({ + request: store + .index(dependencies.idempotencyExpiryIndex) + .openCursor(range), + sink: context, + translate, + budget, + signal: input.signal, + visit: ({ cursor, resume }) => { + if ( + !isStoredReceipt(cursor.value) || + cursor.value.idempotencyKey !== + String(cursor.primaryKey) || + cursor.value.expiresAtEpochMs > cutoff.value + ) { + context.fail(migrationFailure()); + return { kind: "SUSPEND" }; + } + let deleteRequest: IDBRequest; + try { + deleteRequest = store.delete(cursor.primaryKey); + } catch (error) { + context.fail(mappedFailure(error)); + return { kind: "SUSPEND" }; + } + deleteRequest.onerror = () => + context.requestFailed(deleteRequest.error); + deleteRequest.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(migrationFailure()); + return; } + const budgetWrite = governance.put({ + ...budgetRequest.result, + receiptCount: + budgetRequest.result.receiptCount - 1, + } satisfies StoredDatasetBudget); + budgetWrite.onerror = () => + context.requestFailed(budgetWrite.error); + budgetWrite.onsuccess = () => { + deletedRows += 1; + resume({ kind: "CONTINUE" }); + }; }; }; - }; - }; + // The receipt is only counted once its whole delete chain + // commits, so the cursor stays parked until `resume`. + return { kind: "SUSPEND" }; + }, + done: (summary) => { + if (summary.reason === "ABORTED") return; + context.succeed( + Object.freeze({ + state: + summary.reason === "EXHAUSTED" + ? ("COMPLETE" as const) + : ("MORE" as const), + scannedRows: summary.scannedRows, + deletedRows, + budgetExhausted: summary.reason === "TIME_BUDGET", + }), + ); + }, + }); }, ); return observeResult( diff --git a/tests/unit/indexeddb-maintenance.test.ts b/tests/unit/indexeddb-maintenance.test.ts index 248efd1..3fe8ea2 100644 --- a/tests/unit/indexeddb-maintenance.test.ts +++ b/tests/unit/indexeddb-maintenance.test.ts @@ -205,6 +205,7 @@ function createMaintenance( now?: () => number; nowEpochMilliseconds?: () => number; observe?: (event: IndexedDbObservation) => void; + factory?: IDBFactory; }> = {}, ) { return createIndexedDbMaintenance({ @@ -225,6 +226,63 @@ function createMaintenance( }); } +/** + * Records every `transaction.abort()` the adapter itself calls, with the mode + * of the transaction it aborted. + * + * A failure code alone cannot tell `fail` (record and abort) from + * `requestFailed` (record only): the fake aborts a transaction on a request + * error by itself, so both end at the same code while only one of them stops + * the writes queued behind it. Swapping the two is exactly the mistake this + * counter makes visible. + */ +function trackAborts(memory: MemoryIndexedDbFactory): Readonly<{ + factory: IDBFactory; + aborts: readonly string[]; +}> { + const aborts: string[] = []; + const wrapped = new WeakSet(); + const wrapDatabase = (database: IDBDatabase): IDBDatabase => { + if (wrapped.has(database)) return database; + wrapped.add(database); + const openTransaction = database.transaction.bind(database); + Object.defineProperty(database, "transaction", { + configurable: true, + value: (...args: Parameters) => { + const transaction = openTransaction(...args); + const abort = transaction.abort.bind(transaction); + Object.defineProperty(transaction, "abort", { + configurable: true, + value: () => { + aborts.push(transaction.mode); + abort(); + }, + }); + return transaction; + }, + }); + return database; + }; + const factory = { + cmp: (first: IDBValidKey, second: IDBValidKey) => + memory.factory.cmp(first, second), + open: (name: string, version?: number) => { + const request = memory.factory.open(name, version); + let stored: IDBDatabase | undefined = request.result; + if (stored) wrapDatabase(stored); + Object.defineProperty(request, "result", { + configurable: true, + get: () => stored, + set: (value: IDBDatabase | undefined) => { + stored = value ? wrapDatabase(value) : value; + }, + }); + return request; + }, + } as unknown as IDBFactory; + return Object.freeze({ factory, aborts }); +} + function seedReceipt( memory: MemoryIndexedDbFactory, idempotencyKey: string, @@ -299,6 +357,101 @@ describe("IndexedDB bounded codec maintenance", () => { }); }); + it("reports a blocked open immediately instead of waiting out a deadline", async () => { + const memory = new MemoryIndexedDbFactory(); + await prepareSchema(memory); + seedLegacy(memory, "blocked", "waiting"); + const maintenance = createMaintenance(memory, defaultPolicy()); + memory.blockNextOpen(); + + // MT-1. The blocked open is never released here. A blocked deadline could + // only settle from a timer, and no amount of microtask draining reaches a + // timer, so a batch that has already landed proves the blocked event + // itself is terminal. Give maintenance a blocked timeout and this test + // stops finishing instead of failing on a code. + let landed = false; + const pending = maintenance + .migrateCodecBatch({ maxRows: 10, maxDurationMs: 10_000 }) + .then((result) => { + landed = true; + return result; + }); + for (let tick = 0; tick < 50; tick += 1) await Promise.resolve(); + expect(landed).toBe(true); + + expect(await pending).toEqual({ + ok: false, + error: { + code: "BLOCKED", + operation: "INDEXEDDB_MIGRATE", + retryable: true, + recovery: "RELOAD_OTHER_CONTEXTS", + }, + }); + // The connection the caller already gave up on is closed, not leaked. + memory.releaseBlockedOpen(); + expect(memory.isConnectionClosed()).toBe(true); + expect(memory.readRaw("records", "blocked")).toMatchObject({ + codecVersion: 1, + }); + }); + + it("rejects an unexpected schema upgrade without creating the schema", async () => { + const memory = new MemoryIndexedDbFactory(); + // No prepareSchema: the database does not exist yet, so opening at the + // exact schema version is an upgrade. + const maintenance = createMaintenance(memory, defaultPolicy()); + + expect( + await maintenance.migrateCodecBatch({ + maxRows: 10, + maxDurationMs: 10_000, + }), + ).toEqual({ + ok: false, + error: { + code: "MIGRATION_FAILED", + operation: "INDEXEDDB_MIGRATE", + retryable: false, + recovery: "READ_ONLY", + }, + }); + // MT-2. Maintenance owns no schema, so a rejected upgrade must leave the + // database untouched rather than open it at the right version with the + // wrong contents. Applying the upgrade instead would fail with the same + // code here and only surface later, as a database nobody can use. + expect(memory.hasStore("records")).toBe(false); + expect(memory.hasStore("governance")).toBe(false); + await prepareSchema(memory); + expect(memory.hasStore("records")).toBe(true); + }); + + it("closes the maintenance connection when another context needs a version change", async () => { + const memory = new MemoryIndexedDbFactory(); + await prepareSchema(memory); + seedReceipt(memory, "versionchange-receipt", 100); + const maintenance = createMaintenance(memory, defaultPolicy(), { + nowEpochMilliseconds: () => 200, + }); + memory.clearLastTransaction(); + memory.pauseTransactions(); + + const pending = maintenance.pruneExpiredReceipts({ + maxRows: 10, + maxDurationMs: 10_000, + }); + await waitForWriteTransaction(memory); + expect(memory.isConnectionClosed()).toBe(false); + + // MT-3. The listener is registered on the successful open path only, so it + // covers exactly the window in which a batch holds the connection. + memory.triggerVersionChange(2); + expect(memory.isConnectionClosed()).toBe(true); + + memory.resumeTransactions(); + await pending; + }); + it("resumes from a durable checkpoint and completes in bounded row batches", async () => { const memory = new MemoryIndexedDbFactory(); await prepareSchema(memory); @@ -667,7 +820,10 @@ describe("IndexedDB bounded codec maintenance", () => { ...defaultPolicy(), measureStoredBytes: () => 2_000_000, }; - const maintenance = createMaintenance(memory, policy); + const tracker = trackAborts(memory); + const maintenance = createMaintenance(memory, policy, { + factory: tracker.factory, + }); expect( await maintenance.migrateCodecBatch({ @@ -681,6 +837,10 @@ describe("IndexedDB bounded codec maintenance", () => { recovery: "READ_ONLY", }, }); + // The budget verdict aborts the commit transaction rather than only + // recording a failure, so the writes already queued behind it cannot + // commit. Recording without aborting would end at the same failure code. + expect(tracker.aborts).toEqual(["readwrite"]); expect(memory.readRaw("records", "oversized")).toMatchObject({ codecVersion: 1, }); @@ -749,10 +909,11 @@ describe("IndexedDB bounded codec maintenance", () => { const memory = new MemoryIndexedDbFactory(); await prepareSchema(memory); seedReceipt(memory, "quota-receipt", 100); + const tracker = trackAborts(memory); const maintenance = createMaintenance( memory, defaultPolicy(), - { nowEpochMilliseconds: () => 200 }, + { nowEpochMilliseconds: () => 200, factory: tracker.factory }, ); memory.failNextWriteCommit( new DOMException("private receipt", "QuotaExceededError"), @@ -767,6 +928,9 @@ describe("IndexedDB bounded codec maintenance", () => { ok: false, error: { code: "QUOTA_EXCEEDED" }, }); + // The commit is what failed, so the adapter never aborts anything itself; + // the quota code has to come from the transaction's own error. + expect(tracker.aborts).toEqual([]); expect( memory.readRaw("receipts", "quota-receipt"), ).toBeDefined(); @@ -776,10 +940,11 @@ describe("IndexedDB bounded codec maintenance", () => { const memory = new MemoryIndexedDbFactory(); await prepareSchema(memory); seedReceipt(memory, "abort-receipt", 100); + const tracker = trackAborts(memory); const maintenance = createMaintenance( memory, defaultPolicy(), - { nowEpochMilliseconds: () => 200 }, + { nowEpochMilliseconds: () => 200, factory: tracker.factory }, ); memory.clearLastTransaction(); memory.pauseTransactions(); @@ -796,6 +961,9 @@ describe("IndexedDB bounded codec maintenance", () => { ok: false, error: { code: "ABORTED" }, }); + // Exactly one abort: the caller's signal. A second one would mean the + // cursor pump and the transaction both claim the abort. + expect(tracker.aborts).toEqual(["readwrite"]); memory.resumeTransactions(); await Promise.resolve(); expect(