From 217c1dd52a523f0e55039f8a1c6b58f97599bfb4 Mon Sep 17 00:00:00 2001 From: DongHyeonka Date: Wed, 16 Sep 2026 19:51:09 +0900 Subject: [PATCH] refactor: move the OPFS journal onto the IndexedDB kernel, and make the fake IndexedDB enforce unique indexes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 저널은 이제 트랜잭션 안 요청 28곳이 각자 오류를 보고한다. 코드 줄은 1744 -> 1724로 줄었고 전체 줄이 1817 -> 1881로 는 것은 주석이다. 이행 전 이 파일의 주석은 6줄이었다. 사양의 OP-1 전제는 틀렸다. "중복 put의 답이 달라진다"고 봤으나 이행 전후 모두 CONFLICT / retryable:false / recovery:NONE이다. 요청 오류를 아무도 처리하지 않으면 스토어가 바로 그 에러로 abort해서 transaction.error가 request.error와 같기 때문이다. 바뀐 것은 답이 아니라 출처다. 그 과정에서 가짜 IndexedDB가 unique 인덱스를 전혀 강제하지 않는 것을 찾았다. 보강 전에는 중복 begin이 ok:true로 성공했다 — 브라우저가 거부할 상태를 테스트가 조용히 허용하고 있었다. put/add에 검사를 넣었다. unique:true 인덱스는 레포 전체에서 이 저널의 2개뿐이라 반경이 좁고, 전체 test:unit으로 파급이 없음을 확인했다. Co-Authored-By: Claude Opus 5 (1M context) --- .../storage/opfs/indexeddb-opfs-journal.ts | 1472 +++++++++-------- tests/helpers/memory-indexeddb.ts | 44 + tests/unit/indexeddb-opfs-journal.test.ts | 311 ++++ 3 files changed, 1123 insertions(+), 704 deletions(-) diff --git a/src/adapters/storage/opfs/indexeddb-opfs-journal.ts b/src/adapters/storage/opfs/indexeddb-opfs-journal.ts index 3b4996a..a578f53 100644 --- a/src/adapters/storage/opfs/indexeddb-opfs-journal.ts +++ b/src/adapters/storage/opfs/indexeddb-opfs-journal.ts @@ -18,6 +18,19 @@ 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 { + onIndexedDbRequest, + runIndexedDbTransaction, + walkIndexedDbCursor, + type IndexedDbRequestSink, + type IndexedDbTransactionContext, +} from "../../platform/indexeddb-transaction.ts"; import { mapIndexedDbException } from "../indexeddb/indexeddb-failure.ts"; import { isValidOpfsStorageScope } from "./opfs-policy.ts"; @@ -45,10 +58,10 @@ export interface IndexedDbOpfsJournal extends OpfsJournalPort { close(): void; } -type TransactionContext = Readonly<{ - succeed(value: Value): void; - fail(result: BrowserDataResult): void; -}>; +type TransactionContext = IndexedDbTransactionContext< + Value, + BrowserDataFailure +>; type StoredJournalRow = OpfsJournalTransaction & Readonly<{ logicalKey: string }>; @@ -141,17 +154,6 @@ export function createIndexedDbOpfsJournal( dependencies.createFencingToken ?? (() => globalThis.crypto.randomUUID()); const blockedTimeoutMs = dependencies.blockedTimeoutMs ?? 10_000; - const scheduler = - dependencies.scheduler ?? - Object.freeze({ - setTimeout: (callback: () => void, milliseconds: number) => - globalThis.setTimeout(callback, milliseconds), - clearTimeout: (handle: unknown) => - globalThis.clearTimeout( - handle as ReturnType, - ), - }); - if ( !SAFE_BOUNDARY_ID.test(dependencies.authorityToken) || !SAFE_DATABASE_NAME.test(databaseName) || @@ -162,9 +164,88 @@ export function createIndexedDbOpfsJournal( throw new TypeError("IndexedDB OPFS journal configuration is invalid."); } - let database: IDBDatabase | null = null; - let opening: Promise> | null = null; - let closed = false; + // 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. + const timers = snapshotAbortTimers( + dependencies.scheduler ?? { + setTimeout: (callback: () => void, milliseconds: number): unknown => + globalThis.setTimeout(callback, milliseconds), + clearTimeout: (handle: unknown): void => { + globalThis.clearTimeout( + handle as ReturnType, + ); + }, + }, + ); + + /** + * The handle owns the cached connection, the single-flight open and the + * `versionchange`/`close` invalidation this file used to wire by hand. No + * `onVersionChange` callback is passed because dropping the cached + * connection so the next call reopens — which the kernel already does — was + * this journal's entire listener body. + */ + const connection = createIndexedDbConnection({ + translate: translateFor("INDEXEDDB_OPEN"), + open: (signal) => + factory === undefined + ? Promise.resolve( + browserDataFailure("UNSUPPORTED", "INDEXEDDB_OPEN", { + recovery: "ONLINE_ONLY", + }), + ) + : openIndexedDbDatabase({ + factory, + databaseName, + version: DATABASE_VERSION, + translate: translateFor("INDEXEDDB_OPEN"), + signal, + blockedTimeoutMs, + timers, + upgrade: ({ database, oldVersion }) => { + // Version 1 is the only schema this journal has ever had, so any + // other starting point belongs to a database it does not own. + // Rejecting aborts the versionchange transaction, which is what + // the hand-written `onupgradeneeded` did. + if (oldVersion !== 0) return { kind: "REJECTED" }; + const journalStore = database.createObjectStore(JOURNAL_STORE, { + keyPath: "transactionId", + }); + journalStore.createIndex( + LOGICAL_KEY_INDEX, + "logicalKey", + { unique: true }, + ); + journalStore.createIndex( + STARTED_AT_INDEX, + "startedAtEpochMs", + { unique: false }, + ); + const objectStore = database.createObjectStore(OBJECT_STORE, { + keyPath: "logicalKey", + }); + objectStore.createIndex( + SCOPE_OBJECT_INDEX, + "scopeObjectKey", + { unique: true }, + ); + database.createObjectStore(BUDGET_STORE, { + keyPath: "budgetKey", + }); + database.createObjectStore(SCOPE_BINDING_STORE, { + keyPath: "scopeKey", + }); + database.createObjectStore(LOGICAL_SCOPE_BINDING_STORE, { + keyPath: "logicalScopeKey", + }); + database.createObjectStore(CHUNK_REFERENCE_STORE, { + keyPath: "referenceKey", + }); + return { kind: "APPLIED" }; + }, + }), + }); const journal: IndexedDbOpfsJournal = { async getCommittedObject(scope, objectId) { @@ -181,23 +262,26 @@ export function createIndexedDbOpfsJournal( "readonly", "INDEXEDDB_READ", (transaction, context) => { - const request = transaction - .objectStore(OBJECT_STORE) - .get(logicalObjectKey(scope, objectId)); - request.onsuccess = () => { - if (request.result === undefined) { - context.succeed(null); - return; - } - if ( - !isStoredObjectRow(request.result) || - !sameScope(request.result.preparedObject.descriptor.scope, scope) - ) { - context.fail(corrupt("INDEXEDDB_READ")); - return; - } - context.succeed(request.result.preparedObject); - }; + onIndexedDbRequest( + transaction + .objectStore(OBJECT_STORE) + .get(logicalObjectKey(scope, objectId)), + context, + (stored) => { + if (stored === undefined) { + context.succeed(null); + return; + } + if ( + !isStoredObjectRow(stored) || + !sameScope(stored.preparedObject.descriptor.scope, scope) + ) { + context.fail(corrupt("INDEXEDDB_READ")); + return; + } + context.succeed(stored.preparedObject); + }, + ); }, ), ); @@ -232,176 +316,189 @@ export function createIndexedDbOpfsJournal( const bindingStore = nativeTransaction.objectStore(SCOPE_BINDING_STORE); const scopeKey = storageScopeKey(input.scope); - const bindingRequest = bindingStore.get(scopeKey); - bindingRequest.onsuccess = () => { - const storedBinding = bindingRequest.result; - if ( - storedBinding !== undefined && - !isScopeBinding(storedBinding) - ) { - context.fail(corrupt("INDEXEDDB_WRITE")); - return; - } - const expectedBinding = scopeBinding( - input.scope, - input.targetStoragePolicy, - ); - const logicalBindingStore = nativeTransaction.objectStore( - LOGICAL_SCOPE_BINDING_STORE, - ); - const expectedLogicalBinding = logicalScopeBinding( - input.scope, - ); - const logicalRequest = logicalBindingStore.get( - expectedLogicalBinding.logicalScopeKey, - ); - logicalRequest.onsuccess = () => { - const storedLogicalBinding = logicalRequest.result; + onIndexedDbRequest( + bindingStore.get(scopeKey), + context, + (storedBinding) => { if ( - storedLogicalBinding !== undefined && - !isLogicalScopeBinding(storedLogicalBinding) + storedBinding !== undefined && + !isScopeBinding(storedBinding) ) { context.fail(corrupt("INDEXEDDB_WRITE")); return; } - if ( - (storedBinding && - stableJson(storedBinding) !== - stableJson(expectedBinding)) || - (storedLogicalBinding && - stableJson(storedLogicalBinding) !== - stableJson(expectedLogicalBinding)) - ) { - context.fail(policyRejected("INDEXEDDB_WRITE")); - return; - } - if ( - (storedBinding === undefined) !== - (storedLogicalBinding === undefined) - ) { - context.fail(corrupt("INDEXEDDB_WRITE")); - return; - } - if (!storedBinding) { - bindingStore.add(expectedBinding); - logicalBindingStore.add(expectedLogicalBinding); - } - }; - }; + const expectedBinding = scopeBinding( + input.scope, + input.targetStoragePolicy, + ); + const logicalBindingStore = nativeTransaction.objectStore( + LOGICAL_SCOPE_BINDING_STORE, + ); + const expectedLogicalBinding = logicalScopeBinding( + input.scope, + ); + onIndexedDbRequest( + logicalBindingStore.get( + expectedLogicalBinding.logicalScopeKey, + ), + context, + (storedLogicalBinding) => { + if ( + storedLogicalBinding !== undefined && + !isLogicalScopeBinding(storedLogicalBinding) + ) { + context.fail(corrupt("INDEXEDDB_WRITE")); + return; + } + if ( + (storedBinding && + stableJson(storedBinding) !== + stableJson(expectedBinding)) || + (storedLogicalBinding && + stableJson(storedLogicalBinding) !== + stableJson(expectedLogicalBinding)) + ) { + context.fail(policyRejected("INDEXEDDB_WRITE")); + return; + } + if ( + (storedBinding === undefined) !== + (storedLogicalBinding === undefined) + ) { + context.fail(corrupt("INDEXEDDB_WRITE")); + return; + } + if (!storedBinding) { + watchWrite(bindingStore.add(expectedBinding), context); + watchWrite( + logicalBindingStore.add(expectedLogicalBinding), + context, + ); + } + }, + ); + }, + ); const logicalKey = logicalObjectKey( input.scope, input.objectId, ); - const objectRequest = nativeTransaction - .objectStore(OBJECT_STORE) - .get(logicalKey); - objectRequest.onsuccess = () => { - const storedCurrent = objectRequest.result; - if ( - storedCurrent !== undefined && - !isStoredObjectRow(storedCurrent) - ) { - context.fail(corrupt("INDEXEDDB_WRITE")); - return; - } - const current: OpfsPreparedObject | undefined = - storedCurrent?.preparedObject; - if ( - !generationMatches( - current, - input.expectedGeneration, - ) || - input.targetGeneration !== - (current?.descriptor.generation ?? 0) + 1 - ) { - context.fail(conflict("INDEXEDDB_WRITE")); - return; - } - const currentBytes = - current?.descriptor.byteLength ?? 0; - const reservedBytes = - input.mutation === "PUT" - ? Math.max(0, input.targetByteLength - currentBytes) - : 0; - const budgetKey = storageBudgetKey(input.scope); - const budgetRequest = nativeTransaction - .objectStore(BUDGET_STORE) - .get(budgetKey); - budgetRequest.onsuccess = () => { - const budgetResult = budgetRequest.result; + onIndexedDbRequest( + nativeTransaction.objectStore(OBJECT_STORE).get(logicalKey), + context, + (storedCurrent) => { if ( - budgetResult !== undefined && - !isBudgetRow(budgetResult) + storedCurrent !== undefined && + !isStoredObjectRow(storedCurrent) ) { context.fail(corrupt("INDEXEDDB_WRITE")); return; } + const current: OpfsPreparedObject | undefined = + storedCurrent?.preparedObject; if ( - budgetResult && - (budgetResult.hardBudgetBytes !== - input.targetStoragePolicy.hardBudgetBytes || - budgetResult.namespace !== input.scope.namespace || - budgetResult.authorityToken !== - input.scope.authorityToken || - budgetResult.namespaceToken !== - input.scope.namespaceToken || - budgetResult.partitionToken !== - input.scope.partitionToken) + !generationMatches( + current, + input.expectedGeneration, + ) || + input.targetGeneration !== + (current?.descriptor.generation ?? 0) + 1 ) { - context.fail(policyRejected("INDEXEDDB_WRITE")); + context.fail(conflict("INDEXEDDB_WRITE")); return; } - if (!budgetResult && current) { - context.fail(corrupt("INDEXEDDB_WRITE")); - return; - } - const budget: StoredBudgetRow = - budgetResult ?? { - budgetKey, - namespace: input.scope.namespace, - authorityToken: input.scope.authorityToken, - namespaceToken: input.scope.namespaceToken, - partitionToken: input.scope.partitionToken, - hardBudgetBytes: - input.targetStoragePolicy.hardBudgetBytes, - committedBytes: 0, - reservedBytes: 0, - }; - if ( - budget.committedBytes + - budget.reservedBytes + - reservedBytes > - budget.hardBudgetBytes - ) { - context.fail(limitExceeded("INDEXEDDB_WRITE")); - return; - } - const nextBudget: StoredBudgetRow = Object.freeze({ - ...budget, - reservedBytes: - budget.reservedBytes + reservedBytes, - }); - nativeTransaction - .objectStore(BUDGET_STORE) - .put(nextBudget); - const row: StoredJournalRow = Object.freeze({ - ...input, - logicalKey, - fencingToken, - phase: "PREPARING", - budgetReservation: Object.freeze({ - namespace: input.scope.namespace, - reservedBytes, - hardBudgetBytes: - input.targetStoragePolicy.hardBudgetBytes, - }), - }); - nativeTransaction - .objectStore(JOURNAL_STORE) - .add(row); - context.succeed(row); - }; - }; + const currentBytes = + current?.descriptor.byteLength ?? 0; + const reservedBytes = + input.mutation === "PUT" + ? Math.max(0, input.targetByteLength - currentBytes) + : 0; + const budgetKey = storageBudgetKey(input.scope); + onIndexedDbRequest( + nativeTransaction.objectStore(BUDGET_STORE).get(budgetKey), + context, + (budgetResult) => { + if ( + budgetResult !== undefined && + !isBudgetRow(budgetResult) + ) { + context.fail(corrupt("INDEXEDDB_WRITE")); + return; + } + if ( + budgetResult && + (budgetResult.hardBudgetBytes !== + input.targetStoragePolicy.hardBudgetBytes || + budgetResult.namespace !== input.scope.namespace || + budgetResult.authorityToken !== + input.scope.authorityToken || + budgetResult.namespaceToken !== + input.scope.namespaceToken || + budgetResult.partitionToken !== + input.scope.partitionToken) + ) { + context.fail(policyRejected("INDEXEDDB_WRITE")); + return; + } + if (!budgetResult && current) { + context.fail(corrupt("INDEXEDDB_WRITE")); + return; + } + const budget: StoredBudgetRow = + budgetResult ?? { + budgetKey, + namespace: input.scope.namespace, + authorityToken: input.scope.authorityToken, + namespaceToken: input.scope.namespaceToken, + partitionToken: input.scope.partitionToken, + hardBudgetBytes: + input.targetStoragePolicy.hardBudgetBytes, + committedBytes: 0, + reservedBytes: 0, + }; + if ( + budget.committedBytes + + budget.reservedBytes + + reservedBytes > + budget.hardBudgetBytes + ) { + context.fail(limitExceeded("INDEXEDDB_WRITE")); + return; + } + const nextBudget: StoredBudgetRow = Object.freeze({ + ...budget, + reservedBytes: + budget.reservedBytes + reservedBytes, + }); + watchWrite( + nativeTransaction.objectStore(BUDGET_STORE).put(nextBudget), + context, + ); + const row: StoredJournalRow = Object.freeze({ + ...input, + logicalKey, + fencingToken, + phase: "PREPARING", + budgetReservation: Object.freeze({ + namespace: input.scope.namespace, + reservedBytes, + hardBudgetBytes: + input.targetStoragePolicy.hardBudgetBytes, + }), + }); + // OP-1. The `by-logical-key` index is unique, so a second open + // row for the same object is rejected here rather than by any + // predicate the journal evaluated: the ConstraintError belongs + // to this request, and it is now reported as such. + watchWrite( + nativeTransaction.objectStore(JOURNAL_STORE).add(row), + context, + ); + context.succeed(row); + }, + ); + }, + ); }, ), ); @@ -459,7 +556,7 @@ export function createIndexedDbOpfsJournal( phase: "FILES_READY", preparedObject, }); - store.put(updated); + watchWrite(store.put(updated), context); context.succeed(updated); }, ); @@ -493,24 +590,27 @@ export function createIndexedDbOpfsJournal( "INDEXEDDB_WRITE", (transaction, context) => { const store = transaction.objectStore(JOURNAL_STORE); - const request = store.get(transactionId); - request.onsuccess = () => { - if (request.result === undefined) { - context.succeed(undefined); - return; - } - if ( - !isStoredJournalRow(request.result) || - !isBoundScope(request.result.scope) || - request.result.fencingToken !== fencingToken || - request.result.phase !== "COMMITTED" - ) { - context.fail(conflict("INDEXEDDB_WRITE")); - return; - } - store.delete(transactionId); - context.succeed(undefined); - }; + onIndexedDbRequest( + store.get(transactionId), + context, + (stored) => { + if (stored === undefined) { + context.succeed(undefined); + return; + } + if ( + !isStoredJournalRow(stored) || + !isBoundScope(stored.scope) || + stored.fencingToken !== fencingToken || + stored.phase !== "COMMITTED" + ) { + context.fail(conflict("INDEXEDDB_WRITE")); + return; + } + watchWrite(store.delete(transactionId), context); + context.succeed(undefined); + }, + ); }, ), ); @@ -529,53 +629,56 @@ export function createIndexedDbOpfsJournal( (transaction, context) => { const journalStore = transaction.objectStore(JOURNAL_STORE); - const request = journalStore.get(transactionId); - request.onsuccess = () => { - if (request.result === undefined) { - context.succeed(undefined); - return; - } - const row = request.result; - if ( - !isStoredJournalRow(row) || - !isBoundScope(row.scope) || - row.fencingToken !== fencingToken || - row.phase === "COMMITTED" - ) { - context.fail(conflict("INDEXEDDB_WRITE")); - return; - } - const budgetStore = - transaction.objectStore(BUDGET_STORE); - const budgetRequest = budgetStore.get( - storageBudgetKey(row.scope), - ); - budgetRequest.onsuccess = () => { - if (!isBudgetRow(budgetRequest.result)) { - context.fail(corrupt("INDEXEDDB_WRITE")); + onIndexedDbRequest( + journalStore.get(transactionId), + context, + (row) => { + if (row === undefined) { + context.succeed(undefined); return; } - const budget = budgetRequest.result; if ( - budget.reservedBytes < - row.budgetReservation.reservedBytes + !isStoredJournalRow(row) || + !isBoundScope(row.scope) || + row.fencingToken !== fencingToken || + row.phase === "COMMITTED" ) { - context.fail(corrupt("INDEXEDDB_WRITE")); + context.fail(conflict("INDEXEDDB_WRITE")); return; } - putOrDeleteBudget( - budgetStore, - Object.freeze({ - ...budget, - reservedBytes: - budget.reservedBytes - - row.budgetReservation.reservedBytes, - }), + const budgetStore = + transaction.objectStore(BUDGET_STORE); + onIndexedDbRequest( + budgetStore.get(storageBudgetKey(row.scope)), + context, + (budget) => { + if (!isBudgetRow(budget)) { + context.fail(corrupt("INDEXEDDB_WRITE")); + return; + } + if ( + budget.reservedBytes < + row.budgetReservation.reservedBytes + ) { + context.fail(corrupt("INDEXEDDB_WRITE")); + return; + } + putOrDeleteBudget( + budgetStore, + Object.freeze({ + ...budget, + reservedBytes: + budget.reservedBytes - + row.budgetReservation.reservedBytes, + }), + context, + ); + watchWrite(journalStore.delete(transactionId), context); + context.succeed(undefined); + }, ); - journalStore.delete(transactionId); - context.succeed(undefined); - }; - }; + }, + ); }, ), ); @@ -597,40 +700,43 @@ export function createIndexedDbOpfsJournal( "INDEXEDDB_READ", (transaction, context) => { const rows: OpfsJournalTransaction[] = []; - const request = transaction - .objectStore(JOURNAL_STORE) - .index(STARTED_AT_INDEX) - .openCursor(); - request.onsuccess = () => { - const cursor = request.result; - if (!cursor) { + walkIndexedDbCursor({ + request: transaction + .objectStore(JOURNAL_STORE) + .index(STARTED_AT_INDEX) + .openCursor(), + sink: context, + translate: translateFor("INDEXEDDB_READ"), + // OP-4. Neither a signal nor a budget is passed: the journal + // offers no cancellation and bounds the page on `limit` alone. + visit: ({ cursor }) => { + // The row is validated before the page limit is checked, so a + // corrupt row one past the page still fails the read rather + // than being hidden behind `moreAvailable`. + if ( + !isStoredJournalRow(cursor.value) || + !isBoundScope(cursor.value.scope) + ) { + context.fail(corrupt("INDEXEDDB_READ")); + // `fail` aborts, so the pump has nowhere to go. Suspending + // without ever resuming says that without claiming a summary + // the scan never reached. + return { kind: "SUSPEND" }; + } + if (rows.length === limit) return { kind: "STOP" }; + rows.push(cursor.value); + return { kind: "CONTINUE" }; + }, + done: (summary) => { context.succeed( Object.freeze({ transactions: Object.freeze(rows), - moreAvailable: false, + moreAvailable: + summary.reason === "STOPPED" && rows.length === limit, }), ); - return; - } - if ( - !isStoredJournalRow(cursor.value) || - !isBoundScope(cursor.value.scope) - ) { - context.fail(corrupt("INDEXEDDB_READ")); - return; - } - if (rows.length === limit) { - context.succeed( - Object.freeze({ - transactions: Object.freeze(rows), - moreAvailable: true, - }), - ); - return; - } - rows.push(cursor.value); - cursor.continue(); - }; + }, + }); }, ), ); @@ -670,46 +776,47 @@ export function createIndexedDbOpfsJournal( false, ); const objects: OpfsPreparedObject[] = []; - const cursorRequest = transaction - .objectStore(OBJECT_STORE) - .index(SCOPE_OBJECT_INDEX) - .openCursor(range); - cursorRequest.onsuccess = () => { - const cursor = cursorRequest.result; - if (!cursor) { + walkIndexedDbCursor({ + request: transaction + .objectStore(OBJECT_STORE) + .index(SCOPE_OBJECT_INDEX) + .openCursor(range), + sink: context, + translate: translateFor("INDEXEDDB_READ"), + visit: ({ cursor }) => { + if ( + !isStoredObjectRow(cursor.value) || + !sameScope( + cursor.value.preparedObject.descriptor.scope, + request.scope, + ) + ) { + context.fail(corrupt("INDEXEDDB_READ")); + return { kind: "SUSPEND" }; + } + if (objects.length === request.limit) { + return { kind: "STOP" }; + } + objects.push(cursor.value.preparedObject); + return { kind: "CONTINUE" }; + }, + done: (summary) => { + // A page cut short by the limit hands back the cursor the + // caller resumes from; an exhausted scan has nothing to resume. + const more = + summary.reason === "STOPPED" && + objects.length === request.limit; context.succeed( Object.freeze({ objects: Object.freeze(objects), - nextObjectId: null, - moreAvailable: false, + nextObjectId: more + ? objects.at(-1)?.descriptor.objectId ?? null + : null, + moreAvailable: more, }), ); - return; - } - if ( - !isStoredObjectRow(cursor.value) || - !sameScope( - cursor.value.preparedObject.descriptor.scope, - request.scope, - ) - ) { - context.fail(corrupt("INDEXEDDB_READ")); - return; - } - if (objects.length === request.limit) { - context.succeed( - Object.freeze({ - objects: Object.freeze(objects), - nextObjectId: - objects.at(-1)?.descriptor.objectId ?? null, - moreAvailable: true, - }), - ); - return; - } - objects.push(cursor.value.preparedObject); - cursor.continue(); - }; + }, + }); }, ), ); @@ -731,34 +838,34 @@ export function createIndexedDbOpfsJournal( "readonly", "INDEXEDDB_READ", (transaction, context) => { - const request = transaction - .objectStore(CHUNK_REFERENCE_STORE) - .get(referenceKey); - request.onsuccess = () => { - if (request.result === undefined) { - context.succeed(false); - return; - } - if ( - !isChunkReference(request.result) || - request.result.referenceKey !== referenceKey || - request.result.scopeKey !== storageScopeKey(scope) - ) { - context.fail(corrupt("INDEXEDDB_READ")); - return; - } - context.succeed(request.result.referenceCount > 0); - }; + onIndexedDbRequest( + transaction + .objectStore(CHUNK_REFERENCE_STORE) + .get(referenceKey), + context, + (stored) => { + if (stored === undefined) { + context.succeed(false); + return; + } + if ( + !isChunkReference(stored) || + stored.referenceKey !== referenceKey || + stored.scopeKey !== storageScopeKey(scope) + ) { + context.fail(corrupt("INDEXEDDB_READ")); + return; + } + context.succeed(stored.referenceCount > 0); + }, + ); }, ), ); }, close() { - closed = true; - database?.close(); - database = null; - opening = null; + connection.close(); }, }; @@ -786,122 +893,133 @@ export function createIndexedDbOpfsJournal( (nativeTransaction, context) => { const journalStore = nativeTransaction.objectStore(JOURNAL_STORE); - const journalRequest = journalStore.get(transactionId); - journalRequest.onsuccess = () => { - const row = journalRequest.result; - if ( - !isStoredJournalRow(row) || - !isBoundScope(row.scope) || - row.fencingToken !== fencingToken || - row.mutation !== mutation - ) { - context.fail(conflict("INDEXEDDB_WRITE")); - return; - } - if (row.phase === "COMMITTED") { - context.succeed(row); - return; - } - if ( - mutation === "PUT" && - (row.phase !== "FILES_READY" || !row.preparedObject) - ) { - context.fail(conflict("INDEXEDDB_WRITE")); - return; - } - if (mutation === "DELETE" && row.phase !== "PREPARING") { - context.fail(conflict("INDEXEDDB_WRITE")); - return; - } - - const objectStore = - nativeTransaction.objectStore(OBJECT_STORE); - const objectRequest = objectStore.get(row.logicalKey); - objectRequest.onsuccess = () => { - const storedCurrent = objectRequest.result; + onIndexedDbRequest( + journalStore.get(transactionId), + context, + (row) => { if ( - storedCurrent !== undefined && - !isStoredObjectRow(storedCurrent) - ) { - context.fail(corrupt("INDEXEDDB_WRITE")); - return; - } - const current: OpfsPreparedObject | undefined = - storedCurrent?.preparedObject; - if ( - !generationMatches(current, row.expectedGeneration) + !isStoredJournalRow(row) || + !isBoundScope(row.scope) || + row.fencingToken !== fencingToken || + row.mutation !== mutation ) { context.fail(conflict("INDEXEDDB_WRITE")); return; } - const budgetStore = - nativeTransaction.objectStore(BUDGET_STORE); - const budgetRequest = budgetStore.get( - storageBudgetKey(row.scope), - ); - budgetRequest.onsuccess = () => { - if (!isBudgetRow(budgetRequest.result)) { - context.fail(corrupt("INDEXEDDB_WRITE")); - return; - } - const budget = budgetRequest.result; - const currentBytes = - current?.descriptor.byteLength ?? 0; - const nextBytes = - mutation === "PUT" ? row.targetByteLength : 0; - const nextCommitted = - budget.committedBytes - currentBytes + nextBytes; - const nextReserved = - budget.reservedBytes - - row.budgetReservation.reservedBytes; - if ( - nextCommitted < 0 || - nextReserved < 0 || - nextCommitted + nextReserved > - budget.hardBudgetBytes - ) { - context.fail(corrupt("INDEXEDDB_WRITE")); - return; - } + if (row.phase === "COMMITTED") { + context.succeed(row); + return; + } + if ( + mutation === "PUT" && + (row.phase !== "FILES_READY" || !row.preparedObject) + ) { + context.fail(conflict("INDEXEDDB_WRITE")); + return; + } + if (mutation === "DELETE" && row.phase !== "PREPARING") { + context.fail(conflict("INDEXEDDB_WRITE")); + return; + } - const referenceDeltas = chunkReferenceDeltas( - current, - mutation === "PUT" ? row.preparedObject : undefined, - ); - applyChunkReferenceDeltas( - nativeTransaction.objectStore( - CHUNK_REFERENCE_STORE, - ), - row.scope, - referenceDeltas, - context, - () => { - putOrDeleteBudget( - budgetStore, - Object.freeze({ - ...budget, - committedBytes: nextCommitted, - reservedBytes: nextReserved, - }), - ); - if (mutation === "PUT") { - objectStore.put( - storedObjectRow(row.preparedObject!), + const objectStore = + nativeTransaction.objectStore(OBJECT_STORE); + onIndexedDbRequest( + objectStore.get(row.logicalKey), + context, + (storedCurrent) => { + if ( + storedCurrent !== undefined && + !isStoredObjectRow(storedCurrent) + ) { + context.fail(corrupt("INDEXEDDB_WRITE")); + return; + } + const current: OpfsPreparedObject | undefined = + storedCurrent?.preparedObject; + if ( + !generationMatches(current, row.expectedGeneration) + ) { + context.fail(conflict("INDEXEDDB_WRITE")); + return; + } + const budgetStore = + nativeTransaction.objectStore(BUDGET_STORE); + onIndexedDbRequest( + budgetStore.get(storageBudgetKey(row.scope)), + context, + (budget) => { + if (!isBudgetRow(budget)) { + context.fail(corrupt("INDEXEDDB_WRITE")); + return; + } + const currentBytes = + current?.descriptor.byteLength ?? 0; + const nextBytes = + mutation === "PUT" ? row.targetByteLength : 0; + const nextCommitted = + budget.committedBytes - currentBytes + nextBytes; + const nextReserved = + budget.reservedBytes - + row.budgetReservation.reservedBytes; + if ( + nextCommitted < 0 || + nextReserved < 0 || + nextCommitted + nextReserved > + budget.hardBudgetBytes + ) { + context.fail(corrupt("INDEXEDDB_WRITE")); + return; + } + + const referenceDeltas = chunkReferenceDeltas( + current, + mutation === "PUT" ? row.preparedObject : undefined, ); - } else { - objectStore.delete(row.logicalKey); - } - const committed: StoredJournalRow = Object.freeze({ - ...row, - phase: "COMMITTED", - }); - journalStore.put(committed); - context.succeed(committed); - }, - ); - }; - }; - }; + applyChunkReferenceDeltas( + nativeTransaction.objectStore( + CHUNK_REFERENCE_STORE, + ), + row.scope, + referenceDeltas, + context, + () => { + putOrDeleteBudget( + budgetStore, + Object.freeze({ + ...budget, + committedBytes: nextCommitted, + reservedBytes: nextReserved, + }), + context, + ); + if (mutation === "PUT") { + watchWrite( + objectStore.put( + storedObjectRow(row.preparedObject!), + ), + context, + ); + } else { + watchWrite( + objectStore.delete(row.logicalKey), + context, + ); + } + const committed: StoredJournalRow = Object.freeze({ + ...row, + phase: "COMMITTED", + }); + watchWrite(journalStore.put(committed), context); + context.succeed(committed); + }, + ); + }, + ); + }, + ); + }, + ); }, ), ); @@ -924,18 +1042,21 @@ export function createIndexedDbOpfsJournal( "INDEXEDDB_WRITE", (transaction, context) => { const store = transaction.objectStore(JOURNAL_STORE); - const request = store.get(transactionId); - request.onsuccess = () => { - if ( - !isStoredJournalRow(request.result) || - !isBoundScope(request.result.scope) || - request.result.fencingToken !== fencingToken - ) { - context.fail(conflict("INDEXEDDB_WRITE")); - return; - } - update(request.result, context, store); - }; + onIndexedDbRequest( + store.get(transactionId), + context, + (stored) => { + if ( + !isStoredJournalRow(stored) || + !isBoundScope(stored.scope) || + stored.fencingToken !== fencingToken + ) { + context.fail(conflict("INDEXEDDB_WRITE")); + return; + } + update(stored, context, store); + }, + ); }, ), ); @@ -945,7 +1066,7 @@ export function createIndexedDbOpfsJournal( operation: BrowserDataOperation, task: (db: IDBDatabase) => Promise>, ): Promise> { - const opened = await openDatabase(); + const opened = await connection.acquire(); if (!opened.ok) return opened; try { const result = await task(opened.value); @@ -958,119 +1079,6 @@ export function createIndexedDbOpfsJournal( } } - async function openDatabase(): Promise> { - if (closed || !factory) { - return browserDataFailure("UNSUPPORTED", "INDEXEDDB_OPEN", { - recovery: "ONLINE_ONLY", - }); - } - if (database) return browserDataSuccess(database); - if (opening) return await opening; - - const currentOpening = new Promise>( - (resolve) => { - let settled = false; - let blockedTimer: unknown; - const settle = ( - result: BrowserDataResult, - ): void => { - if (settled) return; - settled = true; - if (blockedTimer !== undefined) { - scheduler.clearTimeout(blockedTimer); - } - resolve(result); - }; - let request: IDBOpenDBRequest; - try { - request = factory.open(databaseName, DATABASE_VERSION); - } catch (error) { - settle(mapIndexedDbException(error, "INDEXEDDB_OPEN")); - return; - } - request.onupgradeneeded = (event) => { - const db = request.result; - if (event.oldVersion !== 0) { - request.transaction?.abort(); - return; - } - const journalStore = db.createObjectStore(JOURNAL_STORE, { - keyPath: "transactionId", - }); - journalStore.createIndex( - LOGICAL_KEY_INDEX, - "logicalKey", - { unique: true }, - ); - journalStore.createIndex( - STARTED_AT_INDEX, - "startedAtEpochMs", - { unique: false }, - ); - const objectStore = db.createObjectStore(OBJECT_STORE, { - keyPath: "logicalKey", - }); - objectStore.createIndex( - SCOPE_OBJECT_INDEX, - "scopeObjectKey", - { unique: true }, - ); - db.createObjectStore(BUDGET_STORE, { - keyPath: "budgetKey", - }); - db.createObjectStore(SCOPE_BINDING_STORE, { - keyPath: "scopeKey", - }); - db.createObjectStore(LOGICAL_SCOPE_BINDING_STORE, { - keyPath: "logicalScopeKey", - }); - db.createObjectStore(CHUNK_REFERENCE_STORE, { - keyPath: "referenceKey", - }); - }; - request.onblocked = () => { - if (blockedTimer !== undefined) return; - blockedTimer = scheduler.setTimeout(() => { - settle( - browserDataFailure("BLOCKED", "INDEXEDDB_OPEN", { - retryable: true, - recovery: "RELOAD_OTHER_CONTEXTS", - }), - ); - }, blockedTimeoutMs); - }; - request.onerror = () => - settle( - mapIndexedDbException( - request.error, - "INDEXEDDB_OPEN", - ), - ); - request.onsuccess = () => { - if (settled || closed) { - request.result.close(); - return; - } - database = request.result; - database.onversionchange = () => { - database?.close(); - database = null; - opening = null; - }; - database.onclose = () => { - database = null; - opening = null; - }; - settle(browserDataSuccess(request.result)); - }; - }, - ).finally(() => { - if (opening === currentOpening) opening = null; - }); - opening = currentOpening; - return await currentOpening; - } - function observe( result: BrowserDataResult, operation: BrowserDataOperation, @@ -1095,6 +1103,13 @@ export function createIndexedDbOpfsJournal( } } +/** + * Every readwrite transaction is strict because the journal is the durable + * record of a two-phase OPFS mutation: a write that is only queued when the + * tab goes away would leave files on disk that no journal row claims. Reads + * pass `undefined`, which opens with no options bag at all rather than with + * `{durability:"default"}` — the form this file has always used. + */ function runTransaction( database: IDBDatabase, storeNames: readonly string[], @@ -1105,88 +1120,21 @@ function runTransaction( context: TransactionContext, ) => void, ): Promise> { - return new Promise((resolve) => { - let value: Value | undefined; - let hasValue = false; - let explicitFailure: BrowserDataResult | null = null; - let settled = false; - let transaction: IDBTransaction; - try { - transaction = - mode === "readwrite" - ? strictReadwriteTransaction(database, storeNames) - : database.transaction([...storeNames], mode); - } catch (error) { - resolve(mapIndexedDbException(error, operation)); - return; - } - const settle = (result: BrowserDataResult): void => { - if (settled) return; - settled = true; - resolve(result); - }; - const context: TransactionContext = { - succeed(nextValue) { - value = nextValue; - hasValue = true; - }, - fail(result) { - if (explicitFailure) return; - explicitFailure = result; - try { - transaction.abort(); - } catch { - settle(result); - } - }, - }; - transaction.oncomplete = () => { - if (!hasValue) { - settle( - browserDataFailure("UNAVAILABLE", operation, { - retryable: true, - recovery: "REOPEN", - }), - ); - return; - } - settle(browserDataSuccess(value as Value)); - }; - transaction.onabort = () => - settle( - explicitFailure ?? - mapIndexedDbException(transaction.error, operation), - ); - transaction.onerror = () => { - // onabort owns the single failure result. - }; - try { - run(transaction, context); - } catch (error) { - explicitFailure = mapIndexedDbException(error, operation); - try { - transaction.abort(); - } catch { - settle(explicitFailure); - } - } - }); -} - -function strictReadwriteTransaction( - database: IDBDatabase, - storeNames: readonly string[], -): IDBTransaction { - try { - return database.transaction([...storeNames], "readwrite", { - durability: "strict", - }); - } catch (error) { - if (error instanceof TypeError) { - return database.transaction([...storeNames], "readwrite"); - } - throw error; + if (mode === "versionchange") { + // The journal only ever opens readonly or readwrite transactions; a + // versionchange transaction belongs to the open request, not here. + return Promise.resolve( + browserDataFailure("INVALID_INPUT", operation), + ); } + return runIndexedDbTransaction({ + database, + stores: storeNames, + mode, + translate: translateFor(operation), + durability: mode === "readwrite" ? "strict" : undefined, + queue: run, + }); } function applyChunkReferenceDeltas( @@ -1204,40 +1152,45 @@ function applyChunkReferenceDeltas( let remaining = entries.length; for (const [digestHex, delta] of entries) { const referenceKey = chunkReferenceKey(scope, digestHex); - const request = store.get(referenceKey); - request.onsuccess = () => { - if ( - request.result !== undefined && - (!isChunkReference(request.result) || - request.result.referenceKey !== referenceKey || - request.result.scopeKey !== storageScopeKey(scope)) - ) { - context.fail(corrupt("INDEXEDDB_WRITE")); - return; - } - const current = - (request.result as StoredChunkReference | undefined) - ?.referenceCount ?? 0; - const next = current + delta; - if (!Number.isSafeInteger(next) || next < 0) { - context.fail(corrupt("INDEXEDDB_WRITE")); - return; - } - if (next === 0) { - store.delete(referenceKey); - } else { - store.put( - Object.freeze({ - referenceKey, - scopeKey: storageScopeKey(scope), - digestHex, - referenceCount: next, - }), - ); - } - remaining -= 1; - if (remaining === 0) completed(); - }; + onIndexedDbRequest( + store.get(referenceKey), + context, + (stored) => { + if ( + stored !== undefined && + (!isChunkReference(stored) || + stored.referenceKey !== referenceKey || + stored.scopeKey !== storageScopeKey(scope)) + ) { + context.fail(corrupt("INDEXEDDB_WRITE")); + return; + } + const current = + (stored as StoredChunkReference | undefined)?.referenceCount ?? 0; + const next = current + delta; + if (!Number.isSafeInteger(next) || next < 0) { + context.fail(corrupt("INDEXEDDB_WRITE")); + return; + } + if (next === 0) { + watchWrite(store.delete(referenceKey), context); + } else { + watchWrite( + store.put( + Object.freeze({ + referenceKey, + scopeKey: storageScopeKey(scope), + digestHex, + referenceCount: next, + }), + ), + context, + ); + } + remaining -= 1; + if (remaining === 0) completed(); + }, + ); } } @@ -1258,12 +1211,16 @@ function chunkReferenceDeltas( function putOrDeleteBudget( store: IDBObjectStore, budget: StoredBudgetRow, + sink: IndexedDbRequestSink, ): void { + // Separate calls rather than one over a ternary: `IDBRequest` is invariant + // in its result through the `this` type of its handlers, so the two request + // types have no useful union. if (budget.committedBytes === 0 && budget.reservedBytes === 0) { - store.delete(budget.budgetKey); - } else { - store.put(budget); + watchWrite(store.delete(budget.budgetKey), sink); + return; } + watchWrite(store.put(budget), sink); } function storedObjectRow( @@ -1784,34 +1741,141 @@ function stableJson(value: unknown): string { throw new TypeError("Journal value is not JSON-safe."); } -function conflict( - operation: BrowserDataOperation, -): BrowserDataResult { - return browserDataFailure("CONFLICT", operation, { - recovery: "REOPEN", +/** + * `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 narrows rather than + * claims. + */ +function failureOf(result: BrowserDataResult): BrowserDataFailure { + if (result.ok) { + throw new TypeError("A browser data failure was expected."); + } + return result.error; +} + +/** + * A write nobody waits on still has to report its own failure. This file wired + * no request `onerror` at all before the kernel, so a unique-index + * `ConstraintError` reached the caller only as whatever `transaction.error` + * happened to hold once the store unwound — a provenance the journal never + * chose and cannot rely on. + * + * `requestFailed` records without aborting, which is deliberate: the store + * already aborts a transaction whose request error goes unhandled, and calling + * `fail` here would report the journal as the party that rejected a write the + * index rejected. + */ +function watchWrite( + request: IDBRequest, + sink: IndexedDbRequestSink, +): void { + onIndexedDbRequest(request, sink, () => { + // A write's own success publishes nothing; the transaction's completion + // does. }); } -function corrupt( - operation: BrowserDataOperation, -): BrowserDataResult { - return browserDataFailure("CORRUPT_DATA", operation, { - recovery: "READ_ONLY", - }); +function conflict(operation: BrowserDataOperation): BrowserDataFailure { + return failureOf( + browserDataFailure("CONFLICT", operation, { recovery: "REOPEN" }), + ); +} + +function corrupt(operation: BrowserDataOperation): BrowserDataFailure { + return failureOf( + browserDataFailure("CORRUPT_DATA", operation, { recovery: "READ_ONLY" }), + ); } function policyRejected( operation: BrowserDataOperation, -): BrowserDataResult { - return browserDataFailure("POLICY_REJECTED", operation, { - recovery: "READ_ONLY", - }); +): BrowserDataFailure { + return failureOf( + browserDataFailure("POLICY_REJECTED", operation, { + recovery: "READ_ONLY", + }), + ); } function limitExceeded( operation: BrowserDataOperation, -): BrowserDataResult { - return browserDataFailure("LIMIT_EXCEEDED", operation, { - recovery: "EXPORT_REQUIRED", - }); +): BrowserDataFailure { + return failureOf( + browserDataFailure("LIMIT_EXCEEDED", operation, { + recovery: "EXPORT_REQUIRED", + }), + ); +} + +/** + * STO-01. The journal keeps `mapIndexedDbException`, which the IndexedDB + * runtime and maintenance 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 the journal labels a read and a write + * differently. The connection always translates as `INDEXEDDB_OPEN`: an open + * failure names the operation it belongs to, not the call that happened to ask + * for the connection, and that is why `withDatabase` returns it unobserved. + */ +function translateFor( + operation: BrowserDataOperation, +): IndexedDbTranslate { + return (cause) => { + switch (cause.kind) { + case "NATIVE_EXCEPTION": + return failureOf(mapIndexedDbException(cause.error, operation)); + case "BLOCKED": + case "BLOCKED_DEADLINE": + // The journal reports the same answer whether the blocked event is + // itself terminal or a deadline ran out: a `blockedTimeoutMs` of 0 is + // a valid configuration here, and it must not report a different code + // from the same standing conflict. + return failureOf( + browserDataFailure("BLOCKED", operation, { + retryable: true, + recovery: "RELOAD_OTHER_CONTEXTS", + }), + ); + case "CLOSED": + case "UNSUPPORTED": + // A closed journal and a realm without IndexedDB are the same answer + // to the caller: this store cannot serve the request at all, so work + // online instead of retrying. + return failureOf( + browserDataFailure("UNSUPPORTED", operation, { + recovery: "ONLINE_ONLY", + }), + ); + case "NO_VALUE_PRODUCED": + return failureOf( + browserDataFailure("UNAVAILABLE", operation, { + retryable: true, + recovery: "REOPEN", + }), + ); + case "UPGRADE_REJECTED": + case "CALLER_ABORT": + // Rejecting an upgrade aborts the versionchange transaction, so the + // open request fails with an `AbortError` and the caller has always + // seen ABORTED. `CALLER_ABORT` shares the arm because the journal + // exposes no cancellation — only the connection handle's own close + // cancels an open, and its waiters are told CLOSED instead. + return failureOf(browserDataFailure("ABORTED", operation)); + case "ADMISSION_REJECTED": + // Unreachable: no `admit` callback is installed. The journal validates + // stored rows per transaction rather than at open time. + return failureOf( + browserDataFailure("UNAVAILABLE", operation, { + retryable: true, + recovery: "REOPEN", + }), + ); + default: { + const exhaustive: never = cause; + return exhaustive; + } + } + }; } diff --git a/tests/helpers/memory-indexeddb.ts b/tests/helpers/memory-indexeddb.ts index b8e6979..50238e7 100644 --- a/tests/helpers/memory-indexeddb.ts +++ b/tests/helpers/memory-indexeddb.ts @@ -146,6 +146,48 @@ function compareKeys(first: IDBValidKey, second: IDBValidKey): number { return String(left) < String(right) ? -1 : 1; } +/** + * A unique index rejects a second row whose index key already belongs to + * another primary key, which is how a browser reports it: a request-level + * `ConstraintError` that then aborts the transaction. + * + * The fake enforced primary keys only, so the one `unique: true` index in the + * repository — `indexeddb-opfs-journal.ts`'s logical-key index, whose store is + * keyed by `transactionId` instead — could never fail here the way it fails in + * a browser. Every other index in `src/adapters` is non-unique, so this check + * is a no-op for them. + */ +function assertUniqueIndexes( + state: StoreState, + key: string, + value: unknown, +): void { + for (const [name, index] of state.indexes) { + if (!index.unique || typeof index.keyPath !== "string") continue; + const indexKey = readPath(value, index.keyPath); + // An absent index key keeps the row out of the index entirely, so it + // cannot collide with anything. + if (indexKey === undefined) continue; + for (const [otherKey, otherValue] of state.data) { + // Replacing a row never collides with the row it replaces. + if (otherKey === key) continue; + const otherIndexKey = readPath(otherValue, index.keyPath); + if (otherIndexKey === undefined) continue; + if ( + compareKeys( + indexKey as IDBValidKey, + otherIndexKey as IDBValidKey, + ) === 0 + ) { + throw new DOMException( + `Unique index ${name} already holds this key.`, + "ConstraintError", + ); + } + } + } +} + class FakeUpgradeTransaction { aborted = false; @@ -483,6 +525,7 @@ class FakeObjectStore { this.assertWritable(); return this.request(() => { const key = primaryKey(this.state, value); + assertUniqueIndexes(this.state, key, value); this.state.data.set(key, structuredClone(value)); return key as IDBValidKey; }); @@ -495,6 +538,7 @@ class FakeObjectStore { if (this.state.data.has(key)) { throw new DOMException("Key already exists.", "ConstraintError"); } + assertUniqueIndexes(this.state, key, value); this.state.data.set(key, structuredClone(value)); return key as IDBValidKey; }); diff --git a/tests/unit/indexeddb-opfs-journal.test.ts b/tests/unit/indexeddb-opfs-journal.test.ts index 6977bdc..14217ca 100644 --- a/tests/unit/indexeddb-opfs-journal.test.ts +++ b/tests/unit/indexeddb-opfs-journal.test.ts @@ -95,17 +95,98 @@ function prepared( }; } +/** + * Records every `transaction.abort()` the journal 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 land on the same code while only one of them stops + * the writes queued behind it. + */ +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 createHarness(factory = new MemoryIndexedDbFactory()) { let tokenSequence = 0; const journal = createIndexedDbOpfsJournal({ authorityToken, factory: factory.factory, + // `listCommittedObjects` is the only caller that needs a key range, and + // this environment has no global one. + keyRange: factory.keyRange, createFencingToken: () => `fencing_${String(++tokenSequence).padStart(8, "0")}`, }); return { factory, journal }; } +async function commitObject( + journal: ReturnType["journal"], + transactionId: string, + objectId: string, + byteLength: number, +): Promise { + const input = beginInput(transactionId, objectId, byteLength); + const begun = await journal.begin(input); + if (!begun.ok) throw new Error(`begin failed for ${objectId}`); + const token = begun.value.fencingToken; + const ready = await journal.markFilesReady( + transactionId, + token, + prepared(input), + ); + if (!ready.ok) throw new Error(`markFilesReady failed for ${objectId}`); + const committed = await journal.commitPut(transactionId, token); + if (!committed.ok) throw new Error(`commitPut failed for ${objectId}`); + const completed = await journal.complete(transactionId, token); + if (!completed.ok) throw new Error(`complete failed for ${objectId}`); +} + describe("IndexedDB OPFS journal", () => { it("fences stale writers and atomically publishes object, budget and chunk refs", async () => { const { factory, journal } = createHarness(); @@ -428,4 +509,234 @@ describe("IndexedDB OPFS journal", () => { error: { code: "CORRUPT_DATA" }, }); }); + + it("reports a unique logical-key violation as the request's own ConstraintError", async () => { + // OP-1. Two open journal rows for one object collide on the `by-logical-key` + // index, which is the only `unique: true` index in `src/adapters`. The + // violation belongs to the `add` request, not to the transaction: the + // journal never asks for the row's uniqueness, it asks the index to enforce + // it, so the answer has to be the request's `ConstraintError` rather than + // whatever the transaction happened to abort with. + const memory = new MemoryIndexedDbFactory(); + const tracker = trackAborts(memory); + let tokenSequence = 0; + const journal = createIndexedDbOpfsJournal({ + authorityToken, + factory: tracker.factory, + createFencingToken: () => + `fencing_${String(++tokenSequence).padStart(8, "0")}`, + }); + + const first = await journal.begin( + beginInput("transaction_dupe0001", "object_dupe_1234", 1), + ); + expect(first).toMatchObject({ ok: true }); + + expect( + await journal.begin( + beginInput("transaction_dupe0002", "object_dupe_1234", 1), + ), + ).toEqual({ + ok: false, + error: { + code: "CONFLICT", + operation: "INDEXEDDB_WRITE", + retryable: false, + recovery: "NONE", + }, + }); + // `mapIndexedDbException`'s ConstraintError arm carries no recovery, while + // the journal's own `conflict()` helper carries REOPEN. The NONE above is + // what proves the answer came from the native error rather than from a + // predicate the journal evaluated itself. + expect(tracker.aborts).toEqual([]); + // The losing transaction rolled back whole: no budget reservation from the + // second `begin` survives, so the partition still admits its full budget. + expect(memory.readRaw("opfs-budgets", `${scope.authorityToken}|${scope.namespaceToken}|${scope.partitionToken}`)).toMatchObject({ + reservedBytes: 1, + }); + expect( + (await journal.listIncomplete(10)) as unknown, + ).toMatchObject({ + ok: true, + value: { transactions: [{ transactionId: "transaction_dupe0001" }] }, + }); + }); + + it("lets the first explicit failure win over a later success in the same transaction", async () => { + // OP-2. `begin` runs two independent request chains: the scope-binding + // chain can fail while the object/budget chain is still queued to succeed. + // The binding chain's `get` is issued first, so its verdict always lands + // first and the later `succeed` never reaches the caller. The abort count + // is what separates "the journal rejected this" from "the store did". + const memory = new MemoryIndexedDbFactory(); + const tracker = trackAborts(memory); + let tokenSequence = 0; + const journal = createIndexedDbOpfsJournal({ + authorityToken, + factory: tracker.factory, + createFencingToken: () => + `fencing_${String(++tokenSequence).padStart(8, "0")}`, + }); + + const bound = await journal.begin( + beginInput("transaction_order001", "object_order_1234", 1), + ); + if (!bound.ok) throw new Error("binding begin failed"); + await journal.rollback( + "transaction_order001", + bound.value.fencingToken, + ); + const abortsBefore = tracker.aborts.length; + + expect( + await journal.begin({ + ...beginInput("transaction_order002", "object_order_5678", 1), + targetStoragePolicy: { ...policy, owner: "rebound-owner" }, + }), + ).toEqual({ + ok: false, + error: { + code: "POLICY_REJECTED", + operation: "INDEXEDDB_WRITE", + retryable: false, + recovery: "READ_ONLY", + }, + }); + // Exactly one abort, from the binding chain's `fail`. A second would mean + // the budget chain also reached a verdict. + expect(tracker.aborts.length - abortsBefore).toBe(1); + expect(memory.readRaw("opfs-journal", "transaction_order002")).toBeUndefined(); + }); + + it("keeps a commit failure attributable to the transaction rather than a request", async () => { + // The commit itself fails, so no request ever errors. The journal must not + // claim an abort it did not perform, and the quota code has to come from + // the transaction's own error. + const memory = new MemoryIndexedDbFactory(); + const tracker = trackAborts(memory); + let tokenSequence = 0; + const journal = createIndexedDbOpfsJournal({ + authorityToken, + factory: tracker.factory, + createFencingToken: () => + `fencing_${String(++tokenSequence).padStart(8, "0")}`, + }); + + memory.failNextWriteCommit( + new DOMException("quota", "QuotaExceededError"), + ); + expect( + await journal.begin( + beginInput("transaction_quota001", "object_quota_1234", 1), + ), + ).toEqual({ + ok: false, + error: { + code: "QUOTA_EXCEEDED", + operation: "INDEXEDDB_WRITE", + retryable: false, + recovery: "READ_ONLY", + }, + }); + expect(tracker.aborts).toEqual([]); + }); + + it("pages incomplete transactions and validates a row before the page limit", async () => { + const { factory, journal } = createHarness(); + for (const index of [1, 2, 3]) { + const id = `transaction_page000${index}`; + expect( + await journal.begin( + beginInput(id, `object_page_000${index}`, 1), + ), + ).toMatchObject({ ok: true }); + } + + expect(await journal.listIncomplete(2)).toMatchObject({ + ok: true, + value: { moreAvailable: true }, + }); + const page = await journal.listIncomplete(2); + if (!page.ok) throw new Error("listIncomplete failed"); + expect(page.value.transactions).toHaveLength(2); + + const whole = await journal.listIncomplete(10); + if (!whole.ok) throw new Error("listIncomplete failed"); + expect(whole.value.transactions).toHaveLength(3); + expect(whole.value.moreAvailable).toBe(false); + + // A corrupt row one past the requested page still fails the read. The + // journal validates a row before it checks the limit, so a page that is + // already full does not hide the next row's damage behind `moreAvailable`. + factory.seed("opfs-journal", { + transactionId: "transaction_page9999", + startedAtEpochMs: 9_000, + corrupt: true, + }); + expect(await journal.listIncomplete(3)).toMatchObject({ + ok: false, + error: { code: "CORRUPT_DATA", recovery: "READ_ONLY" }, + }); + }); + + it("pages committed objects and reports where the next page resumes", async () => { + const { journal } = createHarness(); + await commitObject(journal, "transaction_obj00001", "object_aaaa0001", 1); + await commitObject(journal, "transaction_obj00002", "object_bbbb0002", 1); + await commitObject(journal, "transaction_obj00003", "object_cccc0003", 1); + + const first = await journal.listCommittedObjects({ scope, limit: 2 }); + if (!first.ok) throw new Error("listCommittedObjects failed"); + expect(first.value.objects.map((o) => o.descriptor.objectId)).toEqual([ + "object_aaaa0001", + "object_bbbb0002", + ]); + // A page cut short by the limit names the cursor the caller resumes from. + expect(first.value.moreAvailable).toBe(true); + expect(first.value.nextObjectId).toBe("object_bbbb0002"); + + const second = await journal.listCommittedObjects({ + scope, + limit: 2, + afterObjectId: first.value.nextObjectId ?? undefined, + }); + if (!second.ok) throw new Error("listCommittedObjects failed"); + expect(second.value.objects.map((o) => o.descriptor.objectId)).toEqual([ + "object_cccc0003", + ]); + // An exhausted scan has nothing to resume from, so it reports neither. + expect(second.value.moreAvailable).toBe(false); + expect(second.value.nextObjectId).toBeNull(); + }); + + it("answers UNSUPPORTED once the journal is closed", async () => { + const { factory, journal } = createHarness(); + expect( + await journal.getCommittedObject(scope, "object_12345678"), + ).toEqual({ ok: true, value: null }); + + journal.close(); + expect(factory.isConnectionClosed()).toBe(true); + // A closed journal is not a retryable outage and not an abort: the store + // is gone for this realm, so the caller is told to work online. The + // failure names INDEXEDDB_OPEN rather than the operation that asked, + // because it belongs to the open that never happened. + expect( + await journal.getCommittedObject(scope, "object_12345678"), + ).toEqual({ + ok: false, + error: { + code: "UNSUPPORTED", + operation: "INDEXEDDB_OPEN", + retryable: false, + recovery: "ONLINE_ONLY", + }, + }); + expect( + await journal.begin( + beginInput("transaction_closed001", "object_closed_001", 1), + ), + ).toMatchObject({ ok: false, error: { code: "UNSUPPORTED" } }); + }); });