diff --git a/src/adapters/cache-storage/public-response-cache-adapter.ts b/src/adapters/cache-storage/public-response-cache-adapter.ts index 12e6118..a11a810 100644 --- a/src/adapters/cache-storage/public-response-cache-adapter.ts +++ b/src/adapters/cache-storage/public-response-cache-adapter.ts @@ -21,6 +21,7 @@ import { browserDataSuccess, mapBrowserDataException, } from "../browser-file-storage/result.ts"; +import { createAbortableOperation } from "../platform/abortable-operation.ts"; import { cacheByteBucket, cacheEntryBucket, @@ -409,16 +410,42 @@ export function createPublicResponseCacheAdapter( entryBucket: cacheEntryBucket(normalized.assets.length), }); + // NS-08. One owner covers the whole staging body. Handing the signal to + // each `Request` only asked a cooperative fetch to stop: a stream or a + // digest that ignored it kept the mutation lock and, worse, could finish + // after the abort had already ended the operation and still write both the + // asset and the marker, publishing a release nobody was waiting for. + const owner = createAbortableOperation({ signal }); + const ownedStep = async ( + task: Promise, + compensate?: (value: Value) => void, + ): Promise => { + const raced = await owner.race(task, compensate); + if (raced.kind === "REJECTED") throw raced.reason; + if (raced.kind !== "VALUE") { + throw new CacheValidationFailure("ABORTED"); + } + return raced.value; + }; + const assertOwned = (): void => { + if (owner.terminal() !== null) { + throw new CacheValidationFailure("ABORTED"); + } + }; + try { const result = await dependencies.mutationLock!.run( signal, async () => { + assertOwned(); const cacheName = releaseCacheName( policy, normalized.releaseRegistryId, normalized.manifestDigestHex, ); - const existingNames = await dependencies.cacheStorage!.keys(); + const existingNames = await ownedStep( + Promise.resolve(dependencies.cacheStorage!.keys()), + ); // STO-RR-05. A candidate this call did not create may be the one // currently serving traffic, so nothing about it is deleted before // a replacement has been fetched and verified. @@ -467,42 +494,56 @@ export function createPublicResponseCacheAdapter( } } - const cache = await dependencies.cacheStorage!.open(cacheName); + const cache = await ownedStep( + Promise.resolve(dependencies.cacheStorage!.open(cacheName)), + ); try { let totalBytes = 0; for (const asset of normalized.assets) { - if (signal?.aborted) { - throw new CacheValidationFailure("ABORTED"); - } + assertOwned(); const cacheRequest = createNativeRequest(asset); const networkRequest = createNativeRequest( asset, - signal, + owner.signal, ); - const response = - await dependencies.fetcher!(networkRequest); - const read = await readAndValidateResponse( - response, - asset, - policy, - dependencies.crypto, - signal, + const response = await ownedStep( + Promise.resolve(dependencies.fetcher!(networkRequest)), + // A response that arrives after the owner ended is released + // rather than read. + (late) => { + void late.body?.cancel().catch(() => undefined); + }, + ); + const read = await ownedStep( + readAndValidateResponse( + response, + asset, + policy, + dependencies.crypto, + owner.signal, + ), ); if (!read.ok) throw new CacheValidationFailure(read.error.code); + assertOwned(); totalBytes += read.value.bytes.byteLength; if (totalBytes > policy.maxReleaseBytes) { throw new CacheValidationFailure("LIMIT_EXCEEDED"); } - await cache.put( - cacheRequest, - new Response(Uint8Array.from(read.value.bytes), { - status: 200, - headers: read.value.headers.map( - ([name, value]) => [name, value], + await ownedStep( + Promise.resolve( + cache.put( + cacheRequest, + new Response(Uint8Array.from(read.value.bytes), { + status: 200, + headers: read.value.headers.map( + ([name, value]) => [name, value], + ), + }), ), - }), + ), ); } + assertOwned(); const marker: ReleaseMarker = Object.freeze({ schemaVersion: 1, @@ -514,10 +555,14 @@ export function createPublicResponseCacheAdapter( stagedAtEpochMs: now(), assets: normalized.assets, }); - await cache.put( - markerRequest(policy), - jsonResponse(marker), + await ownedStep( + Promise.resolve( + cache.put(markerRequest(policy), jsonResponse(marker)), + ), ); + // The marker is the activation record, so it is only a success + // once this call still owns the operation that wrote it. + assertOwned(); return browserDataSuccess(summaryFromMarker(marker)); } catch (error) { // STO-RR-05. Only a candidate this call created is removed. A @@ -552,6 +597,8 @@ export function createPublicResponseCacheAdapter( dependencies.observer, normalized.releaseRegistryId, ); + } finally { + owner.close(); } }, diff --git a/src/adapters/storage/opfs/opfs-byte-store-adapter.ts b/src/adapters/storage/opfs/opfs-byte-store-adapter.ts index 46542db..c551bbd 100644 --- a/src/adapters/storage/opfs/opfs-byte-store-adapter.ts +++ b/src/adapters/storage/opfs/opfs-byte-store-adapter.ts @@ -325,10 +325,32 @@ export function createOpfsByteStoreAdapter( request.source.byteLength!, ); } - await dependencies.journal.complete( + const completed = await dependencies.journal.complete( transactionId, begun.value.fencingToken, ); + if (!completed.ok) { + // NS-04. The payload is durable and finalized, so nothing is rolled + // back and the `COMMITTED` row stays the reconciler's authority. What + // did not happen is settling the transaction, and reporting a plain + // success for it claimed a state nobody observed while the reconcile + // backlog and its quota pressure grew unseen. + observeOpfsSafely(dependencies.observer, { + operation: "OBJECT_WRITE", + outcome: "DEGRADED", + failureCode: completed.error.code, + byteBucket: byteBucket(prepared.value.descriptor.byteLength), + }); + return { + ok: false as const, + error: Object.freeze({ + ...completed.error, + operation: "OBJECT_WRITE" as const, + retryable: true, + recovery: "RECONCILE" as const, + }), + }; + } observeOpfsSafely(dependencies.observer, { operation: "OBJECT_WRITE", outcome: "SUCCEEDED", @@ -495,17 +517,27 @@ export function createOpfsByteStoreAdapter( request.expectedGeneration, request.signal, ); - if (removed.ok) { - await dependencies.journal.complete( - transactionId, - begun.value.fencingToken, - ); - } + const completed = removed.ok + ? await dependencies.journal.complete( + transactionId, + begun.value.fencingToken, + ) + : null; // Logical deletion is already committed. Physical cleanup is retryable // maintenance and must not make the caller repeat a non-idempotent delete. + // NS-04. It is still not a settled state: an unfinished physical removal + // or an unsettled journal row is maintenance debt the reconciler owns, so + // it is observed as such instead of as a clean success. + const settled = removed.ok && completed !== null && completed.ok; + const debtCode = removed.ok + ? completed !== null && !completed.ok + ? completed.error.code + : undefined + : removed.error.code; observeOpfsSafely(dependencies.observer, { operation: "OBJECT_DELETE", - outcome: "SUCCEEDED", + outcome: settled ? "SUCCEEDED" : "DEGRADED", + ...(debtCode ? { failureCode: debtCode } : {}), }); return browserDataSuccess(undefined); }, diff --git a/src/adapters/storage/opfs/opfs-policy.ts b/src/adapters/storage/opfs/opfs-policy.ts index e02e6d5..6787661 100644 --- a/src/adapters/storage/opfs/opfs-policy.ts +++ b/src/adapters/storage/opfs/opfs-policy.ts @@ -26,7 +26,11 @@ export type OpfsRuntimePolicy = Readonly<{ export type OpfsSafeObservation = Readonly<{ operation: BrowserDataOperation; - outcome: "STARTED" | "SUCCEEDED" | "FAILED"; + /** + * NS-04. `DEGRADED` is a committed effect whose bookkeeping is unsettled: + * the payload is durable but a reconciler still owns the transaction. + */ + outcome: "STARTED" | "SUCCEEDED" | "FAILED" | "DEGRADED"; failureCode?: BrowserDataFailureCode; byteBucket?: "0" | "1B_1MiB" | "1MiB_16MiB" | "16MiB_256MiB" | "GT_256MiB"; transactionBucket?: "0" | "1_10" | "11_100" | "GT_100"; diff --git a/src/adapters/storage/opfs/opfs-worker-client.ts b/src/adapters/storage/opfs/opfs-worker-client.ts index b08a79d..47ce297 100644 --- a/src/adapters/storage/opfs/opfs-worker-client.ts +++ b/src/adapters/storage/opfs/opfs-worker-client.ts @@ -81,31 +81,44 @@ export function createOpfsWorkerGateway( const pending = new Map(); let disposed = false; - const rejectAllPending = (): void => { + const rejectAllPending = ( + code: "UNAVAILABLE" | "UNSUPPORTED" = "UNAVAILABLE", + ): void => { for (const request of pending.values()) { clearTimeout(request.timeout); request.removeAbortListener(); - request.reject(new OpfsRpcError("UNAVAILABLE")); + request.reject(new OpfsRpcError(code)); } pending.clear(); }; const onMessage = (event: MessageEvent): void => { if (disposed) return; - const requestId = ownStringField(event.data, "requestId"); - if (requestId === null) return; - const request = pending.get(requestId); + const correlation = readCorrelation(event.data); + if (correlation === IGNORE_MESSAGE) return; + if (correlation === UNREADABLE_CORRELATION) { + // NS-06. A reply whose correlation cannot even be read is a protocol + // breach on the only channel this client has. Ignoring it left every + // in-flight request to expire on the RPC timer, so the whole channel + // fails closed promptly instead. + rejectAllPending("UNSUPPORTED"); + return; + } + const request = pending.get(correlation); if (!request) return; - pending.delete(requestId); - clearTimeout(request.timeout); - request.removeAbortListener(); // STO-07 / STO-RR-03. A reply is decoded, never adopted. A different // operation, an unknown kind, an unknown failure code, an inherited or // extra field and a hostile accessor are all protocol breaches, and each // closes the request rather than leaving it to time out. // UNSUPPORTED is the closed-taxonomy code for "this runtime cannot serve // this"; no new failure code is invented. + // NS-06. The decode happens before the pending row and its timer are + // released: releasing them first meant a trap that threw inside the decoder + // left the public promise pending with nothing left to time it out. const decoded = decodeWorkerResponse(event.data, request.expectedKind); + pending.delete(correlation); + clearTimeout(request.timeout); + request.removeAbortListener(); if (decoded === null) { request.reject(new OpfsRpcError("UNSUPPORTED")); return; @@ -738,6 +751,31 @@ function ownStringField(source: unknown, key: string): string | null { return typeof value === "string" && value.length > 0 ? value : null; } +/** Not a reply at all: nothing on this channel is waiting for it. */ +const IGNORE_MESSAGE = Symbol("opfs-ignore-message"); +/** A reply whose correlation could not be read without running foreign code. */ +const UNREADABLE_CORRELATION = Symbol("opfs-unreadable-correlation"); + +function readCorrelation( + source: unknown, +): string | typeof IGNORE_MESSAGE | typeof UNREADABLE_CORRELATION { + if (source === null || typeof source !== "object") return IGNORE_MESSAGE; + let descriptor: PropertyDescriptor | undefined; + try { + descriptor = Object.getOwnPropertyDescriptor(source, "requestId"); + } catch { + return UNREADABLE_CORRELATION; + } + if (!descriptor) return IGNORE_MESSAGE; + // An accessor would have to be invoked to be read, and invoking foreign code + // to find out who a message belongs to is exactly what must not happen. + if (!("value" in descriptor)) return UNREADABLE_CORRELATION; + const value = descriptor.value; + return typeof value === "string" && value.length > 0 && value.length <= 128 + ? value + : UNREADABLE_CORRELATION; +} + function hasOnlyOwnDataKeys( source: object, allowed: ReadonlySet, @@ -758,6 +796,20 @@ function hasOnlyOwnDataKeys( function decodeWorkerResponse( value: unknown, expectedKind: OpfsWorkerRequest["kind"], +): OpfsWorkerResponse | null { + // NS-06. Total by construction: every reflection operation below can be + // trapped, and a decoder that throws would strand the request it was + // decoding rather than closing it. + try { + return decodeWorkerResponseUnguarded(value, expectedKind); + } catch { + return null; + } +} + +function decodeWorkerResponseUnguarded( + value: unknown, + expectedKind: OpfsWorkerRequest["kind"], ): OpfsWorkerResponse | null { if (value === null || typeof value !== "object") return null; if (!hasOnlyOwnDataKeys(value, WORKER_RESPONSE_KEYS)) return null; diff --git a/src/adapters/storage/opfs/opfs-worker-runtime.ts b/src/adapters/storage/opfs/opfs-worker-runtime.ts index 77dca06..50b93e3 100644 --- a/src/adapters/storage/opfs/opfs-worker-runtime.ts +++ b/src/adapters/storage/opfs/opfs-worker-runtime.ts @@ -157,12 +157,21 @@ export async function startBrowserOpfsDedicatedWorker( host.addEventListener("message", (event) => { if (!hasRequestId(event.data)) return; const requestData = event.data; + // NS-05. The envelope's correlation is captured once, up front. Answering a + // bootstrap failure with a default `CAPABILITIES` kind made the client see + // an expected-kind mismatch and overwrite the real cause — a `BLOCKED` or + // `QUOTA_EXCEEDED` outage was reported to operators as `UNSUPPORTED`. + const correlation = requestCorrelation(requestData); void runtimePromise .then((runtime) => runtime.handleRequest(requestData)) .then((response) => postWorkerResponse(host, response)) .catch((error: unknown) => { host.postMessage( - failure(requestData.requestId, mapRuntimeFailure(error)), + failure( + correlation.requestId, + mapRuntimeFailure(error), + correlation.kind, + ), ); }); }); @@ -1871,6 +1880,31 @@ function hasRequestId( ); } +/** + * NS-05. Reads a request envelope's correlation exactly once, through its own + * data descriptors, so a reply can name the request it answers even when the + * runtime that would have handled it never came up. An envelope whose kind + * cannot be read stays a protocol-level failure rather than borrowing an + * unrelated kind. + */ +function requestCorrelation( + value: Readonly<{ requestId: string }>, +): Readonly<{ requestId: string; kind: OpfsWorkerRequest["kind"] }> { + let kind: unknown; + try { + kind = Object.getOwnPropertyDescriptor(value, "kind")?.value; + } catch { + kind = undefined; + } + return { + requestId: value.requestId, + kind: + typeof kind === "string" && WORKER_REQUEST_KINDS.has(kind) + ? (kind as OpfsWorkerRequest["kind"]) + : "CAPABILITIES", + }; +} + function isWorkerRequest(value: unknown): value is OpfsWorkerRequest { return Boolean( hasRequestId(value) && diff --git a/tests/unit/opfs-byte-store.test.ts b/tests/unit/opfs-byte-store.test.ts index d5e29e1..731d7d5 100644 --- a/tests/unit/opfs-byte-store.test.ts +++ b/tests/unit/opfs-byte-store.test.ts @@ -352,6 +352,123 @@ describe("OPFS byte-store coordinator", () => { expect(JSON.stringify(observations)).not.toContain("object_12345678"); }); + /** + * NS-04. Awaiting `journal.complete` without checking it reported a settled + * transaction that nobody settled: the caller saw plain success and success + * telemetry while the durable row stayed `COMMITTED`, so the reconcile + * backlog grew invisibly. + */ + it("does not report a settled write when the journal cannot complete it", async () => { + const journal = createJournal(); + journal.complete = async () => + browserDataFailure("UNAVAILABLE", "INDEXEDDB_WRITE", { + retryable: true, + recovery: "RECONCILE", + }); + const observations: { operation: string; outcome: string }[] = []; + const adapter = createOpfsByteStoreAdapter({ + journal, + worker: createWorker(), + scope, + storagePolicy, + policy: { + ...resolveOpfsRuntimePolicy(), + chunkSizeBytes: 64 * 1024, + maxObjectBytes: 64 * 1024, + maxChunkCount: 1, + }, + createTransactionId: () => "transaction_12345678", + now: () => 100, + observer: (event) => + observations.push(event as { operation: string; outcome: string }), + }); + + const result = await adapter.objects.put({ + objectId: "object_12345678", + expectedGeneration: null, + mediaType: "application/octet-stream", + source: sourceFrom(new Uint8Array([1, 2, 3])), + }); + + expect(result.ok).toBe(false); + if (!result.ok) { + // The payload is durable, so recovery is reconciliation, not a rewrite. + expect(result.error.recovery).toBe("RECONCILE"); + expect(result.error.operation).toBe("OBJECT_WRITE"); + } + expect( + observations.some( + (event) => + event.operation === "OBJECT_WRITE" && event.outcome === "SUCCEEDED", + ), + ).toBe(false); + // The committed row stays the reconciler's authority. + expect(journal.transactions.get("transaction_12345678")?.phase).toBe( + "COMMITTED", + ); + expect(journal.objects.get("object_12345678")).toBeDefined(); + }); + + it("records maintenance debt when a committed delete cannot be completed", async () => { + const journal = createJournal(); + const observations: { operation: string; outcome: string }[] = []; + const policy = { + ...resolveOpfsRuntimePolicy(), + chunkSizeBytes: 64 * 1024, + maxObjectBytes: 64 * 1024, + maxChunkCount: 1, + }; + const writer = createOpfsByteStoreAdapter({ + journal, + worker: createWorker(), + scope, + storagePolicy, + policy, + createTransactionId: () => "transaction_12345678", + now: () => 100, + }); + await writer.objects.put({ + objectId: "object_12345678", + expectedGeneration: null, + mediaType: "application/octet-stream", + source: sourceFrom(new Uint8Array([1, 2, 3])), + }); + + // The journal can commit the deletion but cannot settle the transaction. + const remover = createOpfsByteStoreAdapter({ + journal: { + ...journal, + complete: async () => + browserDataFailure("UNAVAILABLE", "INDEXEDDB_WRITE", { + retryable: true, + recovery: "RECONCILE", + }), + }, + worker: createWorker(), + scope, + storagePolicy, + policy, + createTransactionId: () => "transaction_87654321", + now: () => 200, + observer: (event) => + observations.push(event as { operation: string; outcome: string }), + }); + + // Logical deletion is already committed, so the caller must not be asked to + // repeat a non-idempotent delete — but it is not a settled success either. + const result = await remover.objects.remove({ + objectId: "object_12345678", + expectedGeneration: 1, + }); + + expect(result.ok).toBe(true); + const deleteOutcomes = observations + .filter((event) => event.operation === "OBJECT_DELETE") + .map((event) => event.outcome); + expect(deleteOutcomes.at(-1)).toBe("DEGRADED"); + expect(deleteOutcomes).not.toContain("SUCCEEDED"); + }); + it("deep-snapshots scope and policy at composition against caller mutation", async () => { const journal = createJournal(); const mutableScope = { ...scope }; diff --git a/tests/unit/opfs-worker-runtime.test.ts b/tests/unit/opfs-worker-runtime.test.ts index 0d44ebe..eb419e2 100644 --- a/tests/unit/opfs-worker-runtime.test.ts +++ b/tests/unit/opfs-worker-runtime.test.ts @@ -22,7 +22,9 @@ import type { } from "../../src/adapters/storage/opfs/opfs-worker-protocol.ts"; import { createOpfsWorkerRuntime, + startBrowserOpfsDedicatedWorker, type OpfsMutationLeaseManager, + type OpfsWorkerMessageHost, } from "../../src/adapters/storage/opfs/opfs-worker-runtime.ts"; const scopeA: OpfsStorageScope = Object.freeze({ @@ -1117,6 +1119,144 @@ describe("STO-RR-03 worker responses are decoded against closed sets", () => { }); } + /** + * NS-06. Correlation was read separately and the pending row and its timer + * were removed before the reply was decoded. A trap that threw inside the + * decoder therefore left the public promise pending forever, and a throwing + * `requestId` getter produced a timeout instead of a prompt protocol failure. + */ + const uncorrelatableReplies: readonly (readonly [ + string, + (request: OpfsWorkerRequest) => unknown, + ])[] = [ + [ + "throwing requestId getter", + (request) => + Object.defineProperty( + { + protocolVersion: OPFS_WORKER_PROTOCOL_VERSION, + kind: request.kind, + ok: true, + value: {}, + }, + "requestId", + { + enumerable: true, + get: () => { + throw new TypeError("hostile requestId getter"); + }, + }, + ), + ], + [ + "throwing ownKeys trap", + (request) => + new Proxy( + { + requestId: request.requestId, + protocolVersion: OPFS_WORKER_PROTOCOL_VERSION, + kind: request.kind, + ok: true, + value: {}, + }, + { + ownKeys() { + throw new TypeError("hostile ownKeys trap"); + }, + }, + ), + ], + [ + "descriptor trap that throws after correlation", + (request) => { + let reads = 0; + return new Proxy( + { + requestId: request.requestId, + protocolVersion: OPFS_WORKER_PROTOCOL_VERSION, + kind: request.kind, + ok: true, + value: {}, + }, + { + getOwnPropertyDescriptor(target, key) { + reads += 1; + if (reads > 1) { + throw new TypeError("stateful descriptor trap"); + } + return Reflect.getOwnPropertyDescriptor(target, key); + }, + }, + ); + }, + ], + [ + "symbol-keyed field", + (request) => ({ + requestId: request.requestId, + protocolVersion: OPFS_WORKER_PROTOCOL_VERSION, + kind: request.kind, + ok: true, + value: {}, + [Symbol.for("injected")]: true, + }), + ], + [ + "non-enumerable own field", + (request) => + Object.defineProperty( + { + requestId: request.requestId, + protocolVersion: OPFS_WORKER_PROTOCOL_VERSION, + kind: request.kind, + ok: true, + value: {}, + }, + "injected", + { enumerable: false, value: true }, + ), + ], + ]; + + for (const [label, reply] of uncorrelatableReplies) { + it(`closes a ${label} promptly as UNSUPPORTED`, async () => { + const gateway = createOpfsWorkerGateway({ + worker: respondingWorker(reply), + policy: { ...runtimePolicy, rpcTimeoutMs: 60_000 }, + createRequestId: () => `request_uncorr_${label.replace(/\W/gu, "")}`, + }); + // The RPC timeout is far beyond the test budget, so a pass here means the + // reply itself closed the request rather than the timer. + const result = await withTimeout(gateway.capabilities(), label); + expect(result.ok).toBe(false); + expect(result.ok ? null : result.error.code).toBe("UNSUPPORTED"); + }); + } + + it("keeps serving requests after a malformed reply", async () => { + let replies = 0; + const gateway = createOpfsWorkerGateway({ + worker: respondingWorker((request) => { + replies += 1; + if (replies === 1) return { requestId: request.requestId }; + return { + requestId: request.requestId, + protocolVersion: OPFS_WORKER_PROTOCOL_VERSION, + kind: request.kind, + ok: false, + failure: { code: "QUOTA_EXCEEDED", retryable: true }, + }; + }), + policy: { ...runtimePolicy, rpcTimeoutMs: 60_000 }, + createRequestId: () => `request_sequence_${replies}`, + }); + + const first = await withTimeout(gateway.capabilities(), "first"); + expect(first.ok ? null : first.error.code).toBe("UNSUPPORTED"); + const second = await withTimeout(gateway.capabilities(), "second"); + expect(second.ok ? null : second.error.code).toBe("QUOTA_EXCEEDED"); + }); + it("still admits a well-formed closed failure", async () => { const gateway = createOpfsWorkerGateway({ worker: respondingWorker((request) => ({ @@ -1134,3 +1274,83 @@ describe("STO-RR-03 worker responses are decoded against closed sets", () => { expect(result.ok ? null : result.error.code).toBe("QUOTA_EXCEEDED"); }); }); + +/** + * NS-05. When the runtime never came up, the message host answered with a + * default `CAPABILITIES` kind. The gateway saw that as an expected-kind + * mismatch and replaced the real cause — `BLOCKED`, `QUOTA_EXCEEDED` — with a + * generic `UNSUPPORTED` protocol breach, so the outage was misreported. + */ +describe("NS-05 a bootstrap failure answers the request it belongs to", () => { + function hostFor(): Readonly<{ + host: OpfsWorkerMessageHost; + posted: OpfsWorkerResponse[]; + deliver(message: unknown): void; + }> { + const listeners: ((event: MessageEvent) => void)[] = []; + const posted: OpfsWorkerResponse[] = []; + return { + host: { + addEventListener(_type, listener) { + listeners.push(listener); + }, + postMessage(message) { + posted.push(message); + }, + }, + posted, + deliver(message: unknown) { + for (const listener of listeners) { + listener({ data: message } as MessageEvent); + } + }, + }; + } + + const requestKinds = [ + "CAPABILITIES", + "BEGIN_PUT", + "APPEND_CHUNK", + "FINISH_PUT", + "ABORT_PUT", + "VERIFY_OBJECT", + "READ_CHUNK", + "REMOVE_OBJECT", + "CLEANUP_TRANSACTION", + "FINALIZE_PUT", + "LIST_ORPHAN_CANDIDATES", + "DELETE_ORPHAN_CHUNK", + ] as const; + + for (const kind of requestKinds) { + it(`preserves the ${kind} correlation when bootstrap fails`, async () => { + const { host, posted, deliver } = hostFor(); + const start = startBrowserOpfsDedicatedWorker(host, { + storageManager: { + getDirectory: () => + Promise.reject( + new DOMException("blocked", "SecurityError"), + ), + } as unknown as StorageManager, + crypto: globalThis.crypto, + }); + // The bootstrap rejection must not escape the worker entry point either. + await expect(start).rejects.toBeInstanceOf(Error); + + deliver({ + requestId: "request_bootstrap_failure", + protocolVersion: OPFS_WORKER_PROTOCOL_VERSION, + kind, + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(posted).toHaveLength(1); + expect(posted[0]).toMatchObject({ + requestId: "request_bootstrap_failure", + protocolVersion: OPFS_WORKER_PROTOCOL_VERSION, + kind, + ok: false, + }); + }); + } +}); diff --git a/tests/unit/public-response-cache.test.ts b/tests/unit/public-response-cache.test.ts index 0578cfb..5f97fd1 100644 --- a/tests/unit/public-response-cache.test.ts +++ b/tests/unit/public-response-cache.test.ts @@ -528,6 +528,150 @@ describe("public response Cache Storage adapter", () => { await expect(cacheStorage.keys()).resolves.toEqual([]); }); + /** + * NS-08. Handing the signal to each `Request` only asked a cooperative fetch + * to stop. A stream that ignored it held the mutation lock forever, and work + * that finished after the abort still wrote its asset and its marker. + */ + it("does not wait for a non-cooperative fetch after the caller aborts", async () => { + const cacheStorage = new MemoryCacheStorage(); + const policy = createDefaultPublicCachePolicy( + "https://assets.example.test", + ); + const bytes = new Uint8Array([1, 2, 3, 4]); + const asset: PublicCacheAsset = { + absoluteUrl: "https://assets.example.test/never-settles.js", + expectedByteLength: bytes.byteLength, + expectedContentType: "application/javascript", + integrity: { algorithm: "SHA-256", digestHex: await digestHex(bytes) }, + }; + const manifest = await manifestFor("never-settles", [asset], policy); + const fetchStarted = deferred(); + let locksHeld = 0; + const adapter = createPublicResponseCacheAdapter({ + cacheStorage: cacheStorage as unknown as CacheStorage, + crypto: globalThis.crypto, + mutationLock: { + async run(signal, operation) { + locksHeld += 1; + try { + return await operation(); + } finally { + locksHeld -= 1; + } + }, + }, + policy, + fetcher: () => { + fetchStarted.resolve(undefined); + // Ignores the signal entirely. + return new Promise(() => {}); + }, + }); + const controller = new AbortController(); + + const staging = adapter.admin.stageRelease(manifest, { + signal: controller.signal, + }); + await fetchStarted.promise; + controller.abort(); + + await expect(staging).resolves.toMatchObject({ + ok: false, + error: { code: "ABORTED" }, + }); + expect(locksHeld).toBe(0); + await expect(cacheStorage.keys()).resolves.toEqual([]); + }); + + it("writes neither asset nor marker when a digest completes after the abort", async () => { + const cacheStorage = new MemoryCacheStorage(); + const policy = createDefaultPublicCachePolicy( + "https://assets.example.test", + ); + const bytes = new Uint8Array([9, 9, 9, 9]); + const asset: PublicCacheAsset = { + absoluteUrl: "https://assets.example.test/late-digest.js", + expectedByteLength: bytes.byteLength, + expectedContentType: "application/javascript", + integrity: { algorithm: "SHA-256", digestHex: await digestHex(bytes) }, + }; + const manifest = await manifestFor("late-digest", [asset], policy); + const digestStarted = deferred(); + const releaseDigest = deferred(); + let delayFirstDigest = true; + const realCrypto = globalThis.crypto; + const delayedCrypto = { + subtle: { + async digest( + algorithm: AlgorithmIdentifier, + data: BufferSource, + ): Promise { + if (delayFirstDigest) { + delayFirstDigest = false; + digestStarted.resolve(undefined); + await releaseDigest.promise; + } + return await realCrypto.subtle.digest(algorithm, data); + }, + }, + } as unknown as Crypto; + let puts = 0; + const adapter = createPublicResponseCacheAdapter({ + cacheStorage: new Proxy(cacheStorage, { + get(target, key, receiver) { + if (key === "open") { + return async (name: string) => { + const cache = await target.open(name); + return new Proxy(cache, { + get(cacheTarget, cacheKey, cacheReceiver) { + if (cacheKey === "put") { + return async (...args: readonly unknown[]) => { + puts += 1; + return await ( + cacheTarget.put as ( + ...values: readonly unknown[] + ) => Promise + )(...args); + }; + } + return Reflect.get(cacheTarget, cacheKey, cacheReceiver); + }, + }); + }; + } + return Reflect.get(target, key, receiver); + }, + }) as unknown as CacheStorage, + crypto: delayedCrypto, + mutationLock: immediateLock, + policy, + fetcher: async () => + new Response(bytes, { + headers: { + "cache-control": "public, max-age=60", + "content-type": "application/javascript", + }, + }), + }); + const controller = new AbortController(); + + const staging = adapter.admin.stageRelease(manifest, { + signal: controller.signal, + }); + await digestStarted.promise; + controller.abort(); + releaseDigest.resolve(undefined); + + await expect(staging).resolves.toMatchObject({ + ok: false, + error: { code: "ABORTED" }, + }); + // Neither the asset nor the activation marker may be written by work the + // abort already disowned. + expect(puts).toBe(0); + }); + it("keeps the original activation signal while waiting for the mutation lock", async () => { const cacheStorage = new MemoryCacheStorage(); const policy = createDefaultPublicCachePolicy(