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

fix: bound migration commits and version the OPFS worker protocol

Browse Source 
STO-06: the IndexedDB codec migration commit chain runs entirely inside
IndexedDB callbacks, so up to maxRows records could keep executing past the
caller's cooperative deadline. The monotonic budget is now re-checked before
each record's first write; a started record still completes atomically, the
checkpoint advances only to the last safe key, and a clock failure aborts the
transaction rather than committing an unbounded batch.

STO-07: every OPFS worker request and response now carries
OPFS_WORKER_PROTOCOL_VERSION = 2, responses echo their request kind, and the
client validates the envelope and failure shape strictly while remembering the
expected kind per pending request. A page/worker release mismatch or a reply for
a different operation closes as UNSUPPORTED instead of being decoded as a value
of the wrong shape. UNSUPPORTED is used deliberately: the closed browser-data
taxonomy has no INCOMPATIBLE code and none was invented.

SW-10 and the OPFS/Web Push V2 wire rollouts remain deferred: they are
expand/dual-read/drain/contract deployments across releases rather than a single
in-repo change. The ledger records them as DEFERRED_TO_MIGRATION.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
DongHyeonka
2026-08-14 00:41:12 +09:00
co-authored by Claude Opus 5
parent fce8e046ea
commit f6098242be
7 changed files with 205 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapters/storage/indexeddb/indexeddb-maintenance.ts
+18
View File
@@ -972,6 +972,7 @@ export function createIndexedDbMaintenance<WireValue>(
prepared: readonly PreparedRecord<WireValue>[],
scan: ScanBatch,
budgetExhausted: boolean,
deadline: number,
signal: AbortSignal | undefined,
): Promise<
BrowserDataResult<IndexedDbMaintenanceBatchReceipt>
@@ -1048,6 +1049,22 @@ export function createIndexedDbMaintenance<WireValue>(
finishWithCheckpoint();
return;
}
// STO-06. The commit chain runs entirely inside IndexedDB callbacks,
// so without this check up to `maxRows` records keep executing past
// the caller's invocation deadline. The budget is only ever checked
// before a record's first write, so a started record still finishes
// atomically and the checkpoint stays exactly at the last safe key.
const currentTime = clock();
if (!currentTime.ok) {
// A broken clock aborts rather than committing an unbounded batch.
context.fail(currentTime);
return;
}
if (currentTime.value >= deadline) {
budgetExhausted = true;
finishWithCheckpoint();
return;
}
let request: IDBRequest<unknown>;
try {
request = records.get(preparedRecord.source.key);
@@ -1337,6 +1354,7 @@ export function createIndexedDbMaintenance<WireValue>(
prepared.value.records,
scan.value,
prepared.value.budgetExhausted,
deadline,
input.signal,
);
return observeResult(
src/adapters/storage/opfs/opfs-worker-client.ts
+48 -7
View File
@@ -14,6 +14,9 @@ import {
browserDataSuccess,
} from "../../browser-file-storage/result.ts";
import type { OpfsRuntimePolicy } from "./opfs-policy.ts";
import {
OPFS_WORKER_PROTOCOL_VERSION,
} from "./opfs-worker-protocol.ts";
import type {
AbortPreparedPutRequest,
OpfsWorkerGateway,
@@ -51,6 +54,7 @@ export type OwnedOpfsWorkerClient = Readonly<{
}>;
type PendingRequest = Readonly<{
expectedKind: OpfsWorkerRequest["kind"];
resolve: (response: OpfsWorkerResponse) => void;
reject: (error: OpfsRpcError) => void;
timeout: ReturnType<typeof setTimeout>;
@@ -93,6 +97,14 @@ export function createOpfsWorkerGateway(
pending.delete(event.data.requestId);
clearTimeout(request.timeout);
request.removeAbortListener();
if (event.data.kind !== request.expectedKind) {
// STO-07. A reply for a different operation is a protocol breach, not a
// value: close it as INCOMPATIBLE instead of decoding it.
// UNSUPPORTED is the closed-taxonomy code for "this runtime
// cannot serve this"; no new failure code is invented.
request.reject(new OpfsRpcError("UNSUPPORTED"));
return;
}
request.resolve(event.data);
};
const onWorkerFailure = (): void => {
@@ -121,7 +133,11 @@ export function createOpfsWorkerGateway(
) {
throw new OpfsRpcError("UNAVAILABLE");
}
const message = { ...request, requestId } as OpfsWorkerRequest;
const message = {
...request,
requestId,
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
} as OpfsWorkerRequest;
return await new Promise<OpfsWorkerResponse>((resolve, reject) => {
const abort = (): void => {
@@ -141,6 +157,9 @@ export function createOpfsWorkerGateway(
reject(new OpfsRpcError("UNAVAILABLE"));
}, dependencies.policy.rpcTimeoutMs);
pending.set(requestId, {
// STO-07. The expected kind is stored so a reply for a different
// operation can never satisfy this request.
expectedKind: request.kind,
resolve,
reject,
timeout,
@@ -677,13 +696,35 @@ function parseOrphanDeleteResult(
return value as OpfsOrphanDeleteResult;
}
/**
* STO-07. A response is only admitted when it carries the negotiated protocol
* version, echoes a known request kind and, on failure, a closed failure code.
* Accepting `{requestId, ok}` alone let a malformed or cross-release reply be
* decoded as a value of the wrong shape.
*/
function isWorkerResponse(value: unknown): value is OpfsWorkerResponse {
if (
!value ||
typeof value !== "object" ||
!("requestId" in value) ||
typeof value.requestId !== "string" ||
!("ok" in value) ||
typeof value.ok !== "boolean" ||
!("protocolVersion" in value) ||
value.protocolVersion !== OPFS_WORKER_PROTOCOL_VERSION ||
!("kind" in value) ||
typeof value.kind !== "string"
) {
return false;
}
if (value.ok) return true;
const failure = (value as { failure?: unknown }).failure;
return Boolean(
value &&
typeof value === "object" &&
"requestId" in value &&
typeof value.requestId === "string" &&
"ok" in value &&
typeof value.ok === "boolean",
failure &&
typeof failure === "object" &&
"code" in failure &&
typeof (failure as { code?: unknown }).code === "string" &&
"retryable" in failure &&
typeof (failure as { retryable?: unknown }).retryable === "boolean",
);
}
src/adapters/storage/opfs/opfs-worker-protocol.ts
+29 -1
View File
@@ -14,13 +14,27 @@ import type {
TransferProgress,
} from "../../../application/ports/browser-file-storage/shared.ts";
/**
* STO-07. Every request and response carries the protocol version and the
* response echoes its request kind, so a page/worker release mismatch or a
* malformed reply closes as `INCOMPATIBLE` instead of being decoded as a
* successful value of the wrong shape.
*/
export const OPFS_WORKER_PROTOCOL_VERSION = 2 as const;
export type OpfsWorkerRequestEnvelope = Readonly<{
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
}>;
export type OpfsWorkerRequest =
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "CAPABILITIES";
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "BEGIN_PUT";
transactionId: string;
scope: OpfsStorageScope;
@@ -36,6 +50,7 @@ export type OpfsWorkerRequest =
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "APPEND_CHUNK";
scope: OpfsStorageScope;
transactionId: string;
@@ -44,12 +59,14 @@ export type OpfsWorkerRequest =
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "FINISH_PUT";
scope: OpfsStorageScope;
transactionId: string;
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "ABORT_PUT";
scope: OpfsStorageScope;
transactionId: string;
@@ -57,17 +74,20 @@ export type OpfsWorkerRequest =
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "VERIFY_OBJECT";
preparedObject: OpfsPreparedObject;
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "READ_CHUNK";
preparedObject: OpfsPreparedObject;
sequence: number;
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "REMOVE_OBJECT";
scope: OpfsStorageScope;
objectId: string;
@@ -75,6 +95,7 @@ export type OpfsWorkerRequest =
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "CLEANUP_TRANSACTION";
scope: OpfsStorageScope;
transactionId: string;
@@ -87,12 +108,14 @@ export type OpfsWorkerRequest =
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "FINALIZE_PUT";
transactionId: string;
preparedObject: OpfsPreparedObject;
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "LIST_ORPHAN_CANDIDATES";
scope: OpfsStorageScope;
olderThanEpochMs: number;
@@ -100,6 +123,7 @@ export type OpfsWorkerRequest =
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: "DELETE_ORPHAN_CHUNK";
scope: OpfsStorageScope;
digestHex: string;
@@ -109,7 +133,7 @@ export type OpfsWorkerRequest =
export type OpfsWorkerRequestBody =
OpfsWorkerRequest extends infer Request
? Request extends OpfsWorkerRequest
? Omit<Request, "requestId">
? Omit<Request, "requestId" | "protocolVersion">
: never
: never;
@@ -132,6 +156,8 @@ export type OpfsOrphanDeleteResult = Readonly<{
export type OpfsWorkerResponse =
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: OpfsWorkerRequest["kind"];
ok: true;
value?:
| OpfsCapabilities
@@ -144,6 +170,8 @@ export type OpfsWorkerResponse =
}>
| Readonly<{
requestId: string;
protocolVersion: typeof OPFS_WORKER_PROTOCOL_VERSION;
kind: OpfsWorkerRequest["kind"];
ok: false;
failure: OpfsWorkerFailure;
}>;
src/adapters/storage/opfs/opfs-worker-runtime.ts
+45 -10
View File
@@ -15,6 +15,9 @@ import {
isValidOpfsStorageScope,
type OpfsRuntimePolicy,
} from "./opfs-policy.ts";
import {
OPFS_WORKER_PROTOCOL_VERSION,
} from "./opfs-worker-protocol.ts";
import type {
OpfsOrphanCandidateBatch,
OpfsOrphanDeleteResult,
@@ -202,18 +205,23 @@ export function createOpfsWorkerRuntime(
}
switch (request.kind) {
case "CAPABILITIES":
return success(request.requestId, capabilities);
return success(request.requestId, request.kind, capabilities);
case "BEGIN_PUT":
await beginPut(request);
return success(request.requestId);
return success(request.requestId, request.kind);
case "APPEND_CHUNK":
await appendChunk(request);
return success(request.requestId);
return success(request.requestId, request.kind);
case "FINISH_PUT":
return success(request.requestId, await finishPut(request));
return success(
request.requestId,
request.kind,
await finishPut(request),
);
case "ABORT_PUT":
return success(
request.requestId,
request.kind,
await abortPut(
request.scope,
request.transactionId,
@@ -223,11 +231,13 @@ export function createOpfsWorkerRuntime(
case "VERIFY_OBJECT":
return success(
request.requestId,
request.kind,
await verifyObject(request.preparedObject),
);
case "READ_CHUNK":
return success(
request.requestId,
request.kind,
await readVerifiedChunk(
request.preparedObject,
request.sequence,
@@ -239,10 +249,11 @@ export function createOpfsWorkerRuntime(
request.objectId,
request.generation,
);
return success(request.requestId);
return success(request.requestId, request.kind);
case "CLEANUP_TRANSACTION":
return success(
request.requestId,
request.kind,
await cleanupTransaction(
request.scope,
request.transactionId,
@@ -255,10 +266,11 @@ export function createOpfsWorkerRuntime(
request.transactionId,
request.preparedObject,
);
return success(request.requestId);
return success(request.requestId, request.kind);
case "LIST_ORPHAN_CANDIDATES":
return success(
request.requestId,
request.kind,
await listOrphanCandidates(
request.scope,
request.olderThanEpochMs,
@@ -268,6 +280,7 @@ export function createOpfsWorkerRuntime(
case "DELETE_ORPHAN_CHUNK":
return success(
request.requestId,
request.kind,
await deleteOrphanChunk(
request.scope,
request.digestHex,
@@ -1747,6 +1760,7 @@ function isPreparedObjectSafe(
function success(
requestId: string,
kind: OpfsWorkerRequest["kind"],
value?: OpfsWorkerResponse extends infer _Response
?
| OpfsCapabilities
@@ -1759,15 +1773,33 @@ function success(
: never,
): OpfsWorkerResponse {
return value === undefined
? { requestId, ok: true }
: { requestId, ok: true, value };
? {
requestId,
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
kind,
ok: true,
}
: {
requestId,
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
kind,
ok: true,
value,
};
}
function failure(
requestId: string,
workerFailure: OpfsWorkerFailure,
kind: OpfsWorkerRequest["kind"] = "CAPABILITIES",
): OpfsWorkerResponse {
return { requestId, ok: false, failure: workerFailure };
return {
requestId,
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
kind,
ok: false,
failure: workerFailure,
};
}
function mapRuntimeFailure(error: unknown): OpfsWorkerFailure {
@@ -1829,7 +1861,10 @@ function isWorkerRequest(value: unknown): value is OpfsWorkerRequest {
hasRequestId(value) &&
"kind" in value &&
typeof value.kind === "string" &&
WORKER_REQUEST_KINDS.has(value.kind),
WORKER_REQUEST_KINDS.has(value.kind) &&
// STO-07. A page from another release must not be served.
"protocolVersion" in value &&
value.protocolVersion === OPFS_WORKER_PROTOCOL_VERSION,
);
}