fix: report OPFS completion honestly and bound public cache staging
A durable write whose journal transaction could not be completed returned plain success with `SUCCEEDED` telemetry. The payload was committed but the transaction stayed `COMMITTED`, so the reconcile backlog and its quota pressure grew while every caller was told the write had settled. That is now a `RECONCILE` failure with the effect certainty preserved, and an unfinished delete is observed `DEGRADED` rather than clean. The worker seam lost causes in both directions. A bootstrap failure answered every request with kind `CAPABILITIES`, so the gateway read a kind mismatch and replaced the real `BLOCKED` or `QUOTA_EXCEEDED` with a generic `UNSUPPORTED`; the envelope's correlation is now captured once at the listener. On the client, the pending row and its timer were released before the reply was decoded, so a trap that threw inside the decoder left the public promise pending with nothing left to time it out, and a throwing `requestId` getter produced a timeout instead of a prompt protocol failure. Public cache staging handed its signal to each `Request` and called that ownership. A fetch that ignored it held the mutation lock forever, and a digest that finished after the abort still wrote both the asset and the activation marker — publishing a release nobody was waiting for. One terminal owner now covers the whole staging body and every await re-checks it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
DongHyeonka
co-authored by
Claude Opus 5
Claude Opus 5
parent
df18349682
commit
632b230c82
@@ -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 };
|
||||
|
||||
@@ -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<unknown>) => 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<unknown>);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
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,
|
||||
});
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
@@ -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<void>();
|
||||
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<Response>(() => {});
|
||||
},
|
||||
});
|
||||
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<void>();
|
||||
const releaseDigest = deferred<void>();
|
||||
let delayFirstDigest = true;
|
||||
const realCrypto = globalThis.crypto;
|
||||
const delayedCrypto = {
|
||||
subtle: {
|
||||
async digest(
|
||||
algorithm: AlgorithmIdentifier,
|
||||
data: BufferSource,
|
||||
): Promise<ArrayBuffer> {
|
||||
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<void>
|
||||
)(...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(
|
||||
|
||||
Reference in New Issue
Block a user