donghyeon.kang/clean-architecture-frontend-template
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

refactor: move IndexedDB maintenance onto the kernel

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
DongHyeonka
2026-09-16 19:35:22 +09:00
co-authored by Claude Opus 5
parent bb6080bb1c
commit 3366a81f0f
3 changed files with 494 additions and 453 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/check-adapter-inventory.ts
+2 -1
View File
@@ -198,7 +198,8 @@ async function main(): Promise<void> {
//
// 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"],
src/adapters/storage/indexeddb/indexeddb-maintenance.ts
+321 -449
View File
@@ -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<WireValue> = Readonly<{
measuredBytes: number | undefined;
}>;
type TransactionContext<Value> = Readonly<{
succeed(value: Value): void;
fail(result: BrowserDataResult<never>): 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<never> {
});
}
function unsupported(): BrowserDataResult<never> {
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<never>): 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<WireValue>(
}
}
/**
* 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<BrowserDataFailure> = (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<WireValue>(
): Promise<BrowserDataResult<IDBDatabase>> {
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<BrowserDataResult<IDBDatabase>>((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<IDBDatabase>) => {
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<BrowserDataFailure>({
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<Value>(
database: IDBDatabase,
stores: readonly string[],
@@ -610,97 +607,20 @@ export function createIndexedDbMaintenance<WireValue>(
signal: AbortSignal | undefined,
queue: (
transaction: IDBTransaction,
context: TransactionContext<Value>,
context: IndexedDbTransactionContext<Value, BrowserDataFailure>,
) => void,
): Promise<BrowserDataResult<Value>> {
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<BrowserDataResult<Value>>((resolve) => {
let candidate: BrowserDataResult<Value> | undefined;
let requestError: unknown;
let callerAborted = false;
let settled = false;
const finish = (result: BrowserDataResult<Value>) => {
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<Value> = 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<Value, BrowserDataFailure>({
database,
stores,
mode,
translate,
signal,
durability:
mode === "readonly"
? dependencies.durability?.read ?? "default"
: dependencies.durability?.write ?? "strict",
queue,
});
}
@@ -732,7 +652,7 @@ export function createIndexedDbMaintenance<WireValue>(
dependencies.checkpointKey,
)
) {
context.fail(migrationFailed());
context.fail(migrationFailure());
return;
}
const persisted = request.result;
@@ -777,85 +697,69 @@ export function createIndexedDbMaintenance<WireValue>(
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<BrowserDataFailure> = {
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<BrowserDataFailure>({
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<WireValue>(
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<WireValue>(
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<WireValue>(
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<WireValue>(
preparedRecord.source.key,
)
) {
context.fail(migrationFailed());
context.fail(migrationFailure());
return;
}
if (
@@ -1123,7 +1023,7 @@ export function createIndexedDbMaintenance<WireValue>(
!preparedRecord.needsMigration ||
preparedRecord.measuredBytes === undefined
) {
context.fail(migrationFailed());
context.fail(migrationFailure());
return;
}
const measuredBytes = preparedRecord.measuredBytes;
@@ -1137,7 +1037,7 @@ export function createIndexedDbMaintenance<WireValue>(
live.key,
)
) {
context.fail(migrationFailed());
context.fail(migrationFailure());
return;
}
const previousSidecar = sidecarRequest.result;
@@ -1152,7 +1052,7 @@ export function createIndexedDbMaintenance<WireValue>(
budgetRequest.result.usedBytes <
previousSidecar.measuredBytes
) {
context.fail(migrationFailed());
context.fail(migrationFailure());
return;
}
const usedBytes =
@@ -1165,7 +1065,7 @@ export function createIndexedDbMaintenance<WireValue>(
usedBytes >
storagePolicySnapshot.hardBudgetBytes
) {
context.fail(migrationFailed());
context.fail(migrationFailure());
return;
}
const writeRequest = records.put({
@@ -1221,7 +1121,7 @@ export function createIndexedDbMaintenance<WireValue>(
dependencies.checkpointKey,
)
) {
context.fail(migrationFailed());
context.fail(migrationFailure());
return;
}
liveCheckpoint = checkpointRequest.result;
@@ -1412,134 +1312,106 @@ export function createIndexedDbMaintenance<WireValue>(
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<BrowserDataFailure> = {
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<undefined>;
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<BrowserDataFailure>({
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<undefined>;
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(
tests/unit/indexeddb-maintenance.test.ts
+171 -3
View File
@@ -205,6 +205,7 @@ function createMaintenance(
now?: () => number;
nowEpochMilliseconds?: () => number;
observe?: (event: IndexedDbObservation) => void;
factory?: IDBFactory;
}> = {},
) {
return createIndexedDbMaintenance<CurrentPayload>({
@@ -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<object>();
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<IDBDatabase["transaction"]>) => {
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(