DongHyeonkaandClaude Opus 5
6a8281a941
STO-RR-01. finalizePut re-acquired the origin mutation lease it was already holding. A Web Lock is not reentrant, so an ordinary PUT stopped for good at FINALIZE; it now calls the locked cleanup directly. A strict non-reentrant fake lease manager pins one acquire and one release per finalization. The adapter no longer reports a failed finalization as a plain write success either: the journal row stays COMMITTED for reconciliation, but the caller is told the write did not settle. STO-RR-02. A failure raised while serving a validated request now carries that request's kind. Defaulting every catch to CAPABILITIES made the client's own expected-kind check reject genuine quota, integrity and abort failures as protocol breaches and report them as UNSUPPORTED. Only an envelope the runtime could not read still answers at protocol level. STO-RR-03. The worker client decodes a response instead of adopting it: exact own-data descriptors, the negotiated protocol version, the exact awaited kind, a code inside the closed BrowserDataFailure set and a boolean retryable. An accessor, a proxy trap, an inherited or extra field and an unknown code all close the call as UNSUPPORTED rather than leaving it to time out. STO-RR-04. A marker read that fails transiently is unknown, not damaged, so it no longer deletes the candidate that may be serving traffic. Only a confirmed corrupt or missing marker enters the repair path. STO-RR-05. Staging never deletes a candidate it did not create. A repair replaces exact entries in place, so a failed fetch leaves every healthy asset and the active release usable; a candidate this call created is still removed on failure. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1137 lines
34 KiB
TypeScript
1137 lines
34 KiB
TypeScript
import { describe, expect, it } from "vitest";
|
|
|
|
import type {
|
|
OpfsPreparedObject,
|
|
OpfsPhysicalGenerationId,
|
|
OpfsStorageScope,
|
|
} from "../../src/application/ports/browser-file-storage/opfs-ports.ts";
|
|
import type {
|
|
BrowserStoragePolicy,
|
|
} from "../../src/application/ports/browser-file-storage/shared.ts";
|
|
import {
|
|
resolveOpfsRuntimePolicy,
|
|
} from "../../src/adapters/storage/opfs/opfs-policy.ts";
|
|
import {
|
|
createOpfsWorkerGateway,
|
|
type OpfsWorkerLike,
|
|
} from "../../src/adapters/storage/opfs/opfs-worker-client.ts";
|
|
import { OPFS_WORKER_PROTOCOL_VERSION } from "../../src/adapters/storage/opfs/opfs-worker-protocol.ts";
|
|
import type {
|
|
OpfsWorkerRequest,
|
|
OpfsWorkerResponse,
|
|
} from "../../src/adapters/storage/opfs/opfs-worker-protocol.ts";
|
|
import {
|
|
createOpfsWorkerRuntime,
|
|
type OpfsMutationLeaseManager,
|
|
} from "../../src/adapters/storage/opfs/opfs-worker-runtime.ts";
|
|
|
|
const scopeA: OpfsStorageScope = Object.freeze({
|
|
namespace: "durable-objects",
|
|
authorityToken: "authority_12345678",
|
|
namespaceToken: "namespace_12345678",
|
|
partitionToken: "partition_12345678",
|
|
});
|
|
const scopeB: OpfsStorageScope = Object.freeze({
|
|
...scopeA,
|
|
authorityToken: "authority_87654321",
|
|
});
|
|
const storagePolicy: BrowserStoragePolicy = Object.freeze({
|
|
owner: "test-owner",
|
|
namespace: scopeA.namespace,
|
|
classification: "PERSONAL",
|
|
authority: "LOCAL_FIRST",
|
|
accountScope: "OPAQUE_PARTITION",
|
|
retention: Object.freeze({ kind: "EXPLICIT_DELETE" }),
|
|
softBudgetBytes: 1024 * 1024,
|
|
hardBudgetBytes: 2 * 1024 * 1024,
|
|
evictionPriority: "USER_AUTHORED",
|
|
logoutAction: "EXPORT_THEN_PURGE",
|
|
accountDeletionAction: "PURGE_PARTITION",
|
|
pressureAction: "RETAIN",
|
|
unavailableFallback: "EXPORT_REQUIRED",
|
|
});
|
|
const runtimePolicy = resolveOpfsRuntimePolicy({
|
|
chunkSizeBytes: 64 * 1024,
|
|
maxObjectBytes: 64 * 1024,
|
|
maxChunkCount: 1,
|
|
orphanGracePeriodMs: 60_000,
|
|
});
|
|
|
|
type MemoryNode = MemoryDirectory | MemoryFile;
|
|
|
|
class MemoryFile {
|
|
readonly kind = "file";
|
|
bytes = new Uint8Array();
|
|
lastModified = 0;
|
|
}
|
|
|
|
class MemoryDirectory {
|
|
readonly kind = "directory";
|
|
readonly children = new Map<string, MemoryNode>();
|
|
|
|
async getDirectoryHandle(
|
|
name: string,
|
|
options: FileSystemGetDirectoryOptions = {},
|
|
): Promise<FileSystemDirectoryHandle> {
|
|
const current = this.children.get(name);
|
|
if (current instanceof MemoryDirectory) {
|
|
return current as unknown as FileSystemDirectoryHandle;
|
|
}
|
|
if (current || !options.create) throw notFound();
|
|
const created = new MemoryDirectory();
|
|
this.children.set(name, created);
|
|
return created as unknown as FileSystemDirectoryHandle;
|
|
}
|
|
|
|
async getFileHandle(
|
|
name: string,
|
|
options: FileSystemGetFileOptions = {},
|
|
): Promise<FileSystemFileHandle> {
|
|
const current = this.children.get(name);
|
|
if (current instanceof MemoryFile) {
|
|
return fileHandle(current);
|
|
}
|
|
if (current || !options.create) throw notFound();
|
|
const created = new MemoryFile();
|
|
this.children.set(name, created);
|
|
return fileHandle(created);
|
|
}
|
|
|
|
async removeEntry(
|
|
name: string,
|
|
options: FileSystemRemoveOptions = {},
|
|
): Promise<void> {
|
|
const current = this.children.get(name);
|
|
if (!current) throw notFound();
|
|
if (
|
|
current instanceof MemoryDirectory &&
|
|
current.children.size > 0 &&
|
|
!options.recursive
|
|
) {
|
|
throw new DOMException("Directory is not empty.", "InvalidModificationError");
|
|
}
|
|
this.children.delete(name);
|
|
}
|
|
|
|
async *entries(): AsyncIterableIterator<
|
|
[string, FileSystemDirectoryHandle | FileSystemFileHandle]
|
|
> {
|
|
for (const [name, node] of this.children) {
|
|
yield [
|
|
name,
|
|
node instanceof MemoryDirectory
|
|
? (node as unknown as FileSystemDirectoryHandle)
|
|
: fileHandle(node),
|
|
];
|
|
}
|
|
}
|
|
|
|
has(path: readonly string[]): boolean {
|
|
let current: MemoryNode = this;
|
|
for (const segment of path) {
|
|
if (!(current instanceof MemoryDirectory)) return false;
|
|
const next = current.children.get(segment);
|
|
if (!next) return false;
|
|
current = next;
|
|
}
|
|
return true;
|
|
}
|
|
}
|
|
|
|
function fileHandle(file: MemoryFile): FileSystemFileHandle {
|
|
return {
|
|
kind: "file",
|
|
name: "memory",
|
|
async getFile() {
|
|
const blob = new Blob([Uint8Array.from(file.bytes)]);
|
|
Object.defineProperty(blob, "lastModified", {
|
|
value: file.lastModified,
|
|
});
|
|
return blob as File;
|
|
},
|
|
async createWritable() {
|
|
let pending = Uint8Array.from(file.bytes);
|
|
return {
|
|
async write(data: FileSystemWriteChunkType) {
|
|
if (!(data instanceof Uint8Array)) {
|
|
throw new TypeError("The test writer accepts Uint8Array only.");
|
|
}
|
|
pending = Uint8Array.from(data);
|
|
},
|
|
async close() {
|
|
file.bytes = pending;
|
|
file.lastModified = Date.now();
|
|
},
|
|
async abort() {},
|
|
} as unknown as FileSystemWritableFileStream;
|
|
},
|
|
} as FileSystemFileHandle;
|
|
}
|
|
|
|
function notFound(): DOMException {
|
|
return new DOMException("Entry was not found.", "NotFoundError");
|
|
}
|
|
|
|
const PHYSICAL_GENERATION_A = "a".repeat(32) as OpfsPhysicalGenerationId;
|
|
|
|
function beginRequest(
|
|
requestId: string,
|
|
transactionId: string,
|
|
scope: OpfsStorageScope,
|
|
physicalGenerationId: OpfsPhysicalGenerationId = PHYSICAL_GENERATION_A,
|
|
): OpfsWorkerRequest {
|
|
return {
|
|
requestId,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "BEGIN_PUT",
|
|
transactionId,
|
|
scope,
|
|
objectId: "object_12345678",
|
|
generation: 1,
|
|
physicalGenerationId,
|
|
declaredByteLength: 1,
|
|
mediaType: "application/octet-stream",
|
|
createdAtEpochMs: 100,
|
|
storagePolicy,
|
|
chunkSizeBytes: runtimePolicy.chunkSizeBytes,
|
|
};
|
|
}
|
|
|
|
function immediateLeases(releases: { count: number }): OpfsMutationLeaseManager {
|
|
return {
|
|
async acquire() {
|
|
let released = false;
|
|
return {
|
|
release() {
|
|
if (released) return;
|
|
released = true;
|
|
releases.count += 1;
|
|
},
|
|
};
|
|
},
|
|
};
|
|
}
|
|
|
|
function expectFailureCode(
|
|
response: OpfsWorkerResponse | null,
|
|
code: string,
|
|
): void {
|
|
expect(response).toMatchObject({ ok: false, failure: { code } });
|
|
}
|
|
|
|
function preparedValue(
|
|
response: OpfsWorkerResponse | null,
|
|
): OpfsPreparedObject {
|
|
if (
|
|
!response?.ok ||
|
|
!("value" in response) ||
|
|
!response.value ||
|
|
typeof response.value !== "object" ||
|
|
response.value instanceof ArrayBuffer ||
|
|
!("descriptor" in response.value)
|
|
) {
|
|
throw new Error("Expected a prepared OPFS object.");
|
|
}
|
|
return response.value;
|
|
}
|
|
|
|
describe("OPFS dedicated worker runtime", () => {
|
|
it("cancels an exact BEGIN while waiting for a lock and releases a late lease", async () => {
|
|
const root = new MemoryDirectory();
|
|
let resolveLease:
|
|
| ((lease: { release(): void }) => void)
|
|
| undefined;
|
|
let acquireSignal: AbortSignal | undefined;
|
|
let releases = 0;
|
|
const leaseManager: OpfsMutationLeaseManager = {
|
|
acquire(signal) {
|
|
acquireSignal = signal;
|
|
return new Promise((resolve) => {
|
|
resolveLease = resolve;
|
|
});
|
|
},
|
|
};
|
|
const runtime = createOpfsWorkerRuntime({
|
|
root: root as unknown as FileSystemDirectoryHandle,
|
|
crypto: globalThis.crypto,
|
|
policy: runtimePolicy,
|
|
leaseManager,
|
|
dedicatedWorker: true,
|
|
supportsSynchronousAccessHandles: false,
|
|
});
|
|
|
|
const beginning = runtime.handleRequest(
|
|
beginRequest(
|
|
"request_begin_1234",
|
|
"transaction_12345678",
|
|
scopeA,
|
|
),
|
|
);
|
|
await Promise.resolve();
|
|
const aborted = await runtime.handleRequest({
|
|
requestId: "request_abort_1234",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "ABORT_PUT",
|
|
scope: scopeA,
|
|
transactionId: "transaction_12345678",
|
|
});
|
|
expect(aborted).toMatchObject({ ok: true });
|
|
expect(acquireSignal?.aborted).toBe(true);
|
|
|
|
resolveLease?.({
|
|
release() {
|
|
releases += 1;
|
|
},
|
|
});
|
|
expectFailureCode(await beginning, "ABORTED");
|
|
expect(releases).toBe(1);
|
|
expect(root.children.size).toBe(0);
|
|
});
|
|
|
|
it("serializes APPEND against ABORT so no receipt or generation is resurrected", async () => {
|
|
const root = new MemoryDirectory();
|
|
const releases = { count: 0 };
|
|
let resolveDigest:
|
|
| ((digest: ArrayBuffer) => void)
|
|
| undefined;
|
|
const delayedCrypto = {
|
|
subtle: {
|
|
digest: () =>
|
|
new Promise<ArrayBuffer>((resolve) => {
|
|
resolveDigest = resolve;
|
|
}),
|
|
},
|
|
} as unknown as Crypto;
|
|
const runtime = createOpfsWorkerRuntime({
|
|
root: root as unknown as FileSystemDirectoryHandle,
|
|
crypto: delayedCrypto,
|
|
policy: runtimePolicy,
|
|
leaseManager: immediateLeases(releases),
|
|
dedicatedWorker: true,
|
|
supportsSynchronousAccessHandles: false,
|
|
});
|
|
expect(
|
|
await runtime.handleRequest(
|
|
beginRequest(
|
|
"request_begin_5678",
|
|
"transaction_56785678",
|
|
scopeA,
|
|
),
|
|
),
|
|
).toMatchObject({ ok: true });
|
|
|
|
const appending = runtime.handleRequest({
|
|
requestId: "request_append_5678",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "APPEND_CHUNK",
|
|
scope: scopeA,
|
|
transactionId: "transaction_56785678",
|
|
sequence: 0,
|
|
bytes: new Uint8Array([7]).buffer,
|
|
});
|
|
await Promise.resolve();
|
|
const aborting = runtime.handleRequest({
|
|
requestId: "request_abort_5678",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "ABORT_PUT",
|
|
scope: scopeA,
|
|
transactionId: "transaction_56785678",
|
|
});
|
|
resolveDigest?.(new Uint8Array(32).buffer);
|
|
|
|
expectFailureCode(await appending, "ABORTED");
|
|
expect(await aborting).toMatchObject({ ok: true });
|
|
expect(releases.count).toBe(1);
|
|
expect(
|
|
root.has([
|
|
"authorities",
|
|
scopeA.authorityToken,
|
|
scopeA.namespaceToken,
|
|
scopeA.partitionToken,
|
|
"staging",
|
|
"transaction_56785678",
|
|
]),
|
|
).toBe(false);
|
|
});
|
|
|
|
it("isolates physical chunks and cancellation tombstones by opaque authority", async () => {
|
|
const root = new MemoryDirectory();
|
|
const releases = { count: 0 };
|
|
const runtime = createOpfsWorkerRuntime({
|
|
root: root as unknown as FileSystemDirectoryHandle,
|
|
crypto: globalThis.crypto,
|
|
policy: runtimePolicy,
|
|
leaseManager: immediateLeases(releases),
|
|
dedicatedWorker: true,
|
|
supportsSynchronousAccessHandles: false,
|
|
});
|
|
const transactionId = "transaction_99999999";
|
|
for (const [index, targetScope] of [scopeA, scopeB].entries()) {
|
|
expect(
|
|
await runtime.handleRequest(
|
|
beginRequest(
|
|
`request_begin_iso_${index}`,
|
|
transactionId,
|
|
targetScope,
|
|
),
|
|
),
|
|
).toMatchObject({ ok: true });
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: `request_append_iso_${index}`,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "APPEND_CHUNK",
|
|
scope: targetScope,
|
|
transactionId,
|
|
sequence: 0,
|
|
bytes: new Uint8Array([42]).buffer,
|
|
}),
|
|
).toMatchObject({ ok: true });
|
|
}
|
|
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: "request_abort_iso_a",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "ABORT_PUT",
|
|
scope: scopeA,
|
|
transactionId,
|
|
}),
|
|
).toMatchObject({ ok: true });
|
|
const finished = await runtime.handleRequest({
|
|
requestId: "request_finish_iso_b",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "FINISH_PUT",
|
|
scope: scopeB,
|
|
transactionId,
|
|
});
|
|
expect(finished).toMatchObject({ ok: true });
|
|
const preparedObject = preparedValue(finished);
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: "request_verify_iso_b",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "VERIFY_OBJECT",
|
|
preparedObject,
|
|
}),
|
|
).toEqual({
|
|
requestId: "request_verify_iso_b",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: expect.any(String),
|
|
ok: true,
|
|
value: true,
|
|
});
|
|
});
|
|
|
|
it("garbage-collects orphan chunks only inside the requested physical authority", async () => {
|
|
const root = new MemoryDirectory();
|
|
const releases = { count: 0 };
|
|
const runtime = createOpfsWorkerRuntime({
|
|
root: root as unknown as FileSystemDirectoryHandle,
|
|
crypto: globalThis.crypto,
|
|
policy: runtimePolicy,
|
|
leaseManager: immediateLeases(releases),
|
|
dedicatedWorker: true,
|
|
supportsSynchronousAccessHandles: false,
|
|
});
|
|
const preparedByScope = new Map<string, OpfsPreparedObject>();
|
|
for (const [index, targetScope] of [scopeA, scopeB].entries()) {
|
|
const transactionId = `transaction_gc_${index}_1234`;
|
|
await runtime.handleRequest(
|
|
beginRequest(
|
|
`request_gc_begin_${index}`,
|
|
transactionId,
|
|
targetScope,
|
|
),
|
|
);
|
|
await runtime.handleRequest({
|
|
requestId: `request_gc_append_${index}`,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "APPEND_CHUNK",
|
|
scope: targetScope,
|
|
transactionId,
|
|
sequence: 0,
|
|
bytes: new Uint8Array([99]).buffer,
|
|
});
|
|
const object = preparedValue(
|
|
await runtime.handleRequest({
|
|
requestId: `request_gc_finish_${index}`,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "FINISH_PUT",
|
|
scope: targetScope,
|
|
transactionId,
|
|
}),
|
|
);
|
|
preparedByScope.set(targetScope.authorityToken, object);
|
|
await runtime.handleRequest({
|
|
requestId: `request_gc_finalize_${index}`,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "FINALIZE_PUT",
|
|
transactionId,
|
|
preparedObject: object,
|
|
});
|
|
}
|
|
const digestHex = preparedByScope.get(
|
|
scopeA.authorityToken,
|
|
)!.chunks[0]!.digestHex;
|
|
const cutoff = Date.now() + 10_000;
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: "request_gc_list_a",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "LIST_ORPHAN_CANDIDATES",
|
|
scope: scopeA,
|
|
olderThanEpochMs: cutoff,
|
|
maxEntries: 10,
|
|
}),
|
|
).toMatchObject({ ok: true, value: { digests: [digestHex] } });
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: "request_gc_delete_a",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "DELETE_ORPHAN_CHUNK",
|
|
scope: scopeA,
|
|
digestHex,
|
|
olderThanEpochMs: cutoff,
|
|
}),
|
|
).toMatchObject({ ok: true, value: { deleted: true } });
|
|
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: "request_gc_verify_a",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "VERIFY_OBJECT",
|
|
preparedObject: preparedByScope.get(scopeA.authorityToken)!,
|
|
}),
|
|
).toEqual({
|
|
requestId: "request_gc_verify_a",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: expect.any(String),
|
|
ok: true,
|
|
value: false,
|
|
});
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: "request_gc_verify_b",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "VERIFY_OBJECT",
|
|
preparedObject: preparedByScope.get(scopeB.authorityToken)!,
|
|
}),
|
|
).toEqual({
|
|
requestId: "request_gc_verify_b",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: expect.any(String),
|
|
ok: true,
|
|
value: true,
|
|
});
|
|
});
|
|
|
|
it("reports every required capability and fails closed when the lock is absent", async () => {
|
|
const runtime = createOpfsWorkerRuntime({
|
|
root: new MemoryDirectory() as unknown as FileSystemDirectoryHandle,
|
|
crypto: globalThis.crypto,
|
|
policy: runtimePolicy,
|
|
leaseManager: null,
|
|
dedicatedWorker: true,
|
|
supportsSynchronousAccessHandles: true,
|
|
});
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: "request_caps_1234",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "CAPABILITIES",
|
|
}),
|
|
).toEqual({
|
|
requestId: "request_caps_1234",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: expect.any(String),
|
|
ok: true,
|
|
value: {
|
|
available: false,
|
|
dedicatedWorkerRequired: true,
|
|
crossContextMutationLockAvailable: false,
|
|
synchronousAccessHandleAvailable: true,
|
|
},
|
|
});
|
|
expectFailureCode(
|
|
await runtime.handleRequest(
|
|
beginRequest(
|
|
"request_begin_caps",
|
|
"transaction_caps_1234",
|
|
scopeA,
|
|
),
|
|
),
|
|
"UNSUPPORTED",
|
|
);
|
|
});
|
|
});
|
|
|
|
describe("OPFS worker client lifecycle", () => {
|
|
it("rejects a colliding concurrent request id without replacing the first RPC", async () => {
|
|
let listener: ((event: MessageEvent<unknown>) => void) | undefined;
|
|
const posted: OpfsWorkerRequest[] = [];
|
|
const worker: OpfsWorkerLike = {
|
|
postMessage(message) {
|
|
posted.push(message);
|
|
},
|
|
addEventListener(_type, next) {
|
|
listener = next;
|
|
},
|
|
removeEventListener() {
|
|
listener = undefined;
|
|
},
|
|
};
|
|
const gateway = createOpfsWorkerGateway({
|
|
worker,
|
|
policy: runtimePolicy,
|
|
createRequestId: () => "request_collision_1234",
|
|
});
|
|
|
|
const first = gateway.capabilities();
|
|
const second = gateway.capabilities();
|
|
expect(posted).toHaveLength(1);
|
|
|
|
listener?.({
|
|
data: {
|
|
requestId: "request_collision_1234",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "CAPABILITIES",
|
|
ok: true,
|
|
value: {
|
|
available: true,
|
|
dedicatedWorkerRequired: true,
|
|
crossContextMutationLockAvailable: true,
|
|
synchronousAccessHandleAvailable: true,
|
|
},
|
|
},
|
|
} as MessageEvent<unknown>);
|
|
const secondResult = await second;
|
|
gateway.close();
|
|
await expect(first).resolves.toMatchObject({ ok: true });
|
|
expect(secondResult).toMatchObject({
|
|
ok: false,
|
|
error: { code: "UNAVAILABLE" },
|
|
});
|
|
});
|
|
|
|
it("rejects pending RPCs immediately when the worker reports a fatal event", async () => {
|
|
let failureListener: ((event: Event) => void) | undefined;
|
|
const worker: OpfsWorkerLike = {
|
|
postMessage() {},
|
|
addEventListener() {},
|
|
removeEventListener() {},
|
|
addFailureEventListener(listener) {
|
|
failureListener = listener;
|
|
},
|
|
removeFailureEventListener() {
|
|
failureListener = undefined;
|
|
},
|
|
};
|
|
const gateway = createOpfsWorkerGateway({
|
|
worker,
|
|
policy: runtimePolicy,
|
|
createRequestId: () => "request_crash_1234",
|
|
});
|
|
const pending = gateway.capabilities();
|
|
|
|
expect(failureListener).toBeTypeOf("function");
|
|
if (!failureListener) {
|
|
gateway.close();
|
|
return;
|
|
}
|
|
failureListener(new Event("error"));
|
|
await expect(pending).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { code: "UNAVAILABLE" },
|
|
});
|
|
await expect(gateway.capabilities()).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { code: "UNAVAILABLE" },
|
|
});
|
|
});
|
|
|
|
it("scopes the open signal to verification and leaves the acquired source readable", async () => {
|
|
let listener:
|
|
| ((event: MessageEvent<unknown>) => void)
|
|
| undefined;
|
|
const worker: OpfsWorkerLike = {
|
|
postMessage(message) {
|
|
const response: OpfsWorkerResponse =
|
|
message.kind === "VERIFY_OBJECT"
|
|
? {
|
|
requestId: message.requestId,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: message.kind,
|
|
ok: true,
|
|
value: true,
|
|
}
|
|
: {
|
|
requestId: message.requestId,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: message.kind,
|
|
ok: true,
|
|
value: new Uint8Array([4, 2]).buffer,
|
|
};
|
|
queueMicrotask(() =>
|
|
listener?.({ data: response } as MessageEvent<unknown>),
|
|
);
|
|
},
|
|
addEventListener(_type, nextListener) {
|
|
listener = nextListener;
|
|
},
|
|
removeEventListener() {
|
|
listener = undefined;
|
|
},
|
|
};
|
|
let requestSequence = 0;
|
|
const gateway = createOpfsWorkerGateway({
|
|
worker,
|
|
policy: runtimePolicy,
|
|
createRequestId: () => `request_open_${++requestSequence}_1234`,
|
|
});
|
|
const preparedObject: OpfsPreparedObject = {
|
|
physicalSchemaVersion: 1,
|
|
descriptor: {
|
|
objectId: "object_open_12345678",
|
|
scope: scopeA,
|
|
generation: 1,
|
|
byteLength: 2,
|
|
mediaType: "application/octet-stream",
|
|
createdAtEpochMs: 100,
|
|
integrity: {
|
|
algorithm: "SHA-256-TREE-V1",
|
|
rootDigestHex: "a".repeat(64),
|
|
chunkSizeBytes: runtimePolicy.chunkSizeBytes,
|
|
},
|
|
storagePolicy,
|
|
},
|
|
chunks: [
|
|
{
|
|
sequence: 0,
|
|
byteLength: 2,
|
|
digestHex: "b".repeat(64),
|
|
},
|
|
],
|
|
};
|
|
const openController = new AbortController();
|
|
const opened = await gateway.openObject(
|
|
preparedObject,
|
|
openController.signal,
|
|
);
|
|
expect(opened.ok).toBe(true);
|
|
openController.abort();
|
|
|
|
const chunks: number[][] = [];
|
|
if (opened.ok) {
|
|
for await (const result of opened.value.stream(
|
|
new AbortController().signal,
|
|
)) {
|
|
expect(result.ok).toBe(true);
|
|
if (result.ok) chunks.push([...result.value]);
|
|
}
|
|
}
|
|
expect(chunks).toEqual([[4, 2]]);
|
|
gateway.close();
|
|
});
|
|
|
|
it("removes its listener and rejects all pending RPCs before Worker termination", async () => {
|
|
const listeners = new Set<(event: MessageEvent<unknown>) => void>();
|
|
const worker: OpfsWorkerLike = {
|
|
postMessage() {},
|
|
addEventListener(_type, listener) {
|
|
listeners.add(listener);
|
|
},
|
|
removeEventListener(_type, listener) {
|
|
listeners.delete(listener);
|
|
},
|
|
};
|
|
const gateway = createOpfsWorkerGateway({
|
|
worker,
|
|
policy: runtimePolicy,
|
|
createRequestId: () => "request_pending_1234",
|
|
});
|
|
const pending = gateway.capabilities();
|
|
expect(listeners.size).toBe(1);
|
|
gateway.close();
|
|
expect(listeners.size).toBe(0);
|
|
await expect(pending).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { code: "UNAVAILABLE" },
|
|
});
|
|
await expect(gateway.capabilities()).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { code: "UNAVAILABLE" },
|
|
});
|
|
});
|
|
});
|
|
|
|
/**
|
|
* STO-RR-01. A Web Lock is not reentrant. Any path that re-acquires the origin
|
|
* mutation lease while already holding it stops making progress forever, and a
|
|
* lock the runtime waits on cannot be observed by a fake that hands out an
|
|
* unlimited number of leases.
|
|
*/
|
|
function strictNonReentrantLeases(
|
|
counters: { acquires: number; releases: number },
|
|
): OpfsMutationLeaseManager {
|
|
let held = false;
|
|
return {
|
|
async acquire() {
|
|
if (held) {
|
|
// A second holder waits for the first to release. Nothing here ever
|
|
// does, which is exactly what a deadlock looks like.
|
|
return await new Promise<never>(() => {});
|
|
}
|
|
held = true;
|
|
counters.acquires += 1;
|
|
let released = false;
|
|
return {
|
|
release() {
|
|
if (released) return;
|
|
released = true;
|
|
held = false;
|
|
counters.releases += 1;
|
|
},
|
|
};
|
|
},
|
|
};
|
|
}
|
|
|
|
function withTimeout<Value>(
|
|
operation: Promise<Value>,
|
|
label: string,
|
|
ms = 200,
|
|
): Promise<Value> {
|
|
return Promise.race([
|
|
operation,
|
|
new Promise<never>((_resolve, reject) => {
|
|
setTimeout(() => reject(new Error(`${label} did not settle`)), ms);
|
|
}),
|
|
]);
|
|
}
|
|
|
|
describe("STO-RR-01 OPFS finalization under a non-reentrant lock", () => {
|
|
async function completedPut(
|
|
runtime: ReturnType<typeof createOpfsWorkerRuntime>,
|
|
transactionId: string,
|
|
): Promise<OpfsPreparedObject> {
|
|
expect(
|
|
await runtime.handleRequest(
|
|
beginRequest(`request_begin_${transactionId}`, transactionId, scopeA),
|
|
),
|
|
).toMatchObject({ ok: true });
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: `request_append_${transactionId}`,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "APPEND_CHUNK",
|
|
scope: scopeA,
|
|
transactionId,
|
|
sequence: 0,
|
|
bytes: new Uint8Array([9]).buffer,
|
|
}),
|
|
).toMatchObject({ ok: true });
|
|
const finished = await runtime.handleRequest({
|
|
requestId: `request_finish_${transactionId}`,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "FINISH_PUT",
|
|
scope: scopeA,
|
|
transactionId,
|
|
});
|
|
expect(finished).toMatchObject({ ok: true });
|
|
return preparedValue(finished);
|
|
}
|
|
|
|
it("finalizes a normal PUT with exactly one lock acquisition", async () => {
|
|
const root = new MemoryDirectory();
|
|
const counters = { acquires: 0, releases: 0 };
|
|
const runtime = createOpfsWorkerRuntime({
|
|
root: root as unknown as FileSystemDirectoryHandle,
|
|
crypto: globalThis.crypto,
|
|
policy: runtimePolicy,
|
|
leaseManager: strictNonReentrantLeases(counters),
|
|
dedicatedWorker: true,
|
|
supportsSynchronousAccessHandles: false,
|
|
});
|
|
const transactionId = "transaction_final_0001";
|
|
const prepared = await completedPut(runtime, transactionId);
|
|
const before = counters.acquires;
|
|
|
|
const finalized = await withTimeout(
|
|
runtime.handleRequest({
|
|
requestId: "request_finalize_0001",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "FINALIZE_PUT",
|
|
transactionId,
|
|
preparedObject: prepared,
|
|
}),
|
|
"FINALIZE_PUT",
|
|
);
|
|
|
|
expect(finalized).toMatchObject({ ok: true });
|
|
expect(counters.acquires - before).toBe(1);
|
|
expect(counters.acquires).toBe(counters.releases);
|
|
expect(
|
|
root.has([
|
|
"authorities",
|
|
scopeA.authorityToken,
|
|
scopeA.namespaceToken,
|
|
scopeA.partitionToken,
|
|
"staging",
|
|
transactionId,
|
|
]),
|
|
).toBe(false);
|
|
});
|
|
|
|
it("leaves the lock free for the next mutation after a finalized PUT", async () => {
|
|
const root = new MemoryDirectory();
|
|
const counters = { acquires: 0, releases: 0 };
|
|
const runtime = createOpfsWorkerRuntime({
|
|
root: root as unknown as FileSystemDirectoryHandle,
|
|
crypto: globalThis.crypto,
|
|
policy: runtimePolicy,
|
|
leaseManager: strictNonReentrantLeases(counters),
|
|
dedicatedWorker: true,
|
|
supportsSynchronousAccessHandles: false,
|
|
});
|
|
const first = "transaction_final_0002";
|
|
const prepared = await completedPut(runtime, first);
|
|
await withTimeout(
|
|
runtime.handleRequest({
|
|
requestId: "request_finalize_0002",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "FINALIZE_PUT",
|
|
transactionId: first,
|
|
preparedObject: prepared,
|
|
}),
|
|
"first FINALIZE_PUT",
|
|
);
|
|
|
|
const second = "transaction_final_0003";
|
|
await expect(
|
|
withTimeout(completedPut(runtime, second), "second PUT"),
|
|
).resolves.toMatchObject({ descriptor: { objectId: "object_12345678" } });
|
|
expect(counters.acquires).toBe(counters.releases);
|
|
});
|
|
});
|
|
|
|
/**
|
|
* STO-RR-02. A failure raised while serving a validated request must answer
|
|
* that request. Defaulting the response kind to `CAPABILITIES` made the client's
|
|
* own expected-kind check reject it as a protocol breach, so a quota or
|
|
* integrity failure reached the caller as `UNSUPPORTED`.
|
|
*/
|
|
describe("STO-RR-02 worker failure responses echo the request kind", () => {
|
|
const failingRoot = {
|
|
async getDirectoryHandle(): Promise<FileSystemDirectoryHandle> {
|
|
throw new DOMException("Out of room", "QuotaExceededError");
|
|
},
|
|
async getFileHandle(): Promise<FileSystemFileHandle> {
|
|
throw new DOMException("Out of room", "QuotaExceededError");
|
|
},
|
|
async removeEntry(): Promise<void> {
|
|
throw new DOMException("Out of room", "QuotaExceededError");
|
|
},
|
|
async *entries(): AsyncIterableIterator<never> {},
|
|
} as unknown as FileSystemDirectoryHandle;
|
|
|
|
it("keeps the validated kind on every failure path", async () => {
|
|
const counters = { acquires: 0, releases: 0 };
|
|
const runtime = createOpfsWorkerRuntime({
|
|
root: failingRoot,
|
|
crypto: globalThis.crypto,
|
|
policy: runtimePolicy,
|
|
leaseManager: strictNonReentrantLeases(counters),
|
|
dedicatedWorker: true,
|
|
supportsSynchronousAccessHandles: false,
|
|
});
|
|
|
|
const requests: readonly OpfsWorkerRequest[] = [
|
|
beginRequest("request_kind_begin", "transaction_kind_0001", scopeA),
|
|
{
|
|
requestId: "request_kind_remove",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "REMOVE_OBJECT",
|
|
scope: scopeA,
|
|
objectId: "object_12345678",
|
|
generation: 1,
|
|
},
|
|
{
|
|
requestId: "request_kind_cleanup",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "CLEANUP_TRANSACTION",
|
|
scope: scopeA,
|
|
transactionId: "transaction_kind_0001",
|
|
},
|
|
{
|
|
requestId: "request_kind_orphans",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "LIST_ORPHAN_CANDIDATES",
|
|
scope: scopeA,
|
|
olderThanEpochMs: 1,
|
|
maxEntries: 1,
|
|
},
|
|
];
|
|
|
|
for (const request of requests) {
|
|
const response = await runtime.handleRequest(request);
|
|
expect(response).toMatchObject({
|
|
ok: false,
|
|
kind: request.kind,
|
|
requestId: request.requestId,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
});
|
|
}
|
|
});
|
|
|
|
it("still reports a protocol-level failure for an unreadable envelope", async () => {
|
|
const counters = { acquires: 0, releases: 0 };
|
|
const runtime = createOpfsWorkerRuntime({
|
|
root: failingRoot,
|
|
crypto: globalThis.crypto,
|
|
policy: runtimePolicy,
|
|
leaseManager: strictNonReentrantLeases(counters),
|
|
dedicatedWorker: true,
|
|
supportsSynchronousAccessHandles: false,
|
|
});
|
|
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: "request_kind_broken",
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "NOT_A_KIND",
|
|
}),
|
|
).toMatchObject({ ok: false, kind: "CAPABILITIES" });
|
|
});
|
|
});
|
|
|
|
/**
|
|
* STO-RR-03. The client decoder is the trust boundary for anything a worker
|
|
* says. A `code` that is merely a string lets an arbitrary value escape the
|
|
* closed `BrowserDataFailure` taxonomy into application code.
|
|
*/
|
|
describe("STO-RR-03 worker responses are decoded against closed sets", () => {
|
|
function respondingWorker(
|
|
reply: (request: OpfsWorkerRequest) => unknown,
|
|
): OpfsWorkerLike {
|
|
const listeners = new Set<(event: MessageEvent<unknown>) => void>();
|
|
return {
|
|
postMessage(message: unknown) {
|
|
const response = reply(message as OpfsWorkerRequest);
|
|
queueMicrotask(() => {
|
|
for (const listener of listeners) {
|
|
listener({ data: response } as MessageEvent<unknown>);
|
|
}
|
|
});
|
|
},
|
|
addEventListener(_type: "message", listener: (event: MessageEvent<unknown>) => void) {
|
|
listeners.add(listener);
|
|
},
|
|
removeEventListener(_type: "message", listener: (event: MessageEvent<unknown>) => void) {
|
|
listeners.delete(listener);
|
|
},
|
|
} as unknown as OpfsWorkerLike;
|
|
}
|
|
|
|
const hostileReplies: readonly (readonly [string, (request: OpfsWorkerRequest) => unknown])[] = [
|
|
[
|
|
"unknown failure code",
|
|
(request) => ({
|
|
requestId: request.requestId,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: request.kind,
|
|
ok: false,
|
|
failure: { code: "EVIL", retryable: false },
|
|
}),
|
|
],
|
|
[
|
|
"unknown request kind",
|
|
(request) => ({
|
|
requestId: request.requestId,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: "NOT_A_KIND",
|
|
ok: false,
|
|
failure: { code: "UNAVAILABLE", retryable: false },
|
|
}),
|
|
],
|
|
[
|
|
"non-boolean retryable",
|
|
(request) => ({
|
|
requestId: request.requestId,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: request.kind,
|
|
ok: false,
|
|
failure: { code: "UNAVAILABLE", retryable: "yes" },
|
|
}),
|
|
],
|
|
[
|
|
"inherited failure fields",
|
|
(request) => ({
|
|
requestId: request.requestId,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: request.kind,
|
|
ok: false,
|
|
failure: Object.create({ code: "UNAVAILABLE", retryable: false }) as object,
|
|
}),
|
|
],
|
|
[
|
|
"throwing getter",
|
|
(request) => {
|
|
const response: Record<string, unknown> = {
|
|
requestId: request.requestId,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
ok: false,
|
|
failure: { code: "UNAVAILABLE", retryable: false },
|
|
};
|
|
Object.defineProperty(response, "kind", {
|
|
enumerable: true,
|
|
get: () => {
|
|
throw new TypeError("hostile getter");
|
|
},
|
|
});
|
|
return response;
|
|
},
|
|
],
|
|
[
|
|
"extra own field",
|
|
(request) => ({
|
|
requestId: request.requestId,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: request.kind,
|
|
ok: false,
|
|
failure: { code: "UNAVAILABLE", retryable: false, injected: 1 },
|
|
}),
|
|
],
|
|
];
|
|
|
|
for (const [label, reply] of hostileReplies) {
|
|
it(`closes a ${label} as UNSUPPORTED without rejecting`, async () => {
|
|
const gateway = createOpfsWorkerGateway({
|
|
worker: respondingWorker(reply),
|
|
policy: runtimePolicy,
|
|
createRequestId: () => `request_hostile_${label.replace(/\W/gu, "")}`,
|
|
});
|
|
const result = await withTimeout(gateway.capabilities(), label);
|
|
expect(result.ok).toBe(false);
|
|
expect(result.ok ? null : result.error.code).toBe("UNSUPPORTED");
|
|
});
|
|
}
|
|
|
|
it("still admits a well-formed closed failure", async () => {
|
|
const gateway = createOpfsWorkerGateway({
|
|
worker: respondingWorker((request) => ({
|
|
requestId: request.requestId,
|
|
protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
|
|
kind: request.kind,
|
|
ok: false,
|
|
failure: { code: "QUOTA_EXCEEDED", retryable: true },
|
|
})),
|
|
policy: runtimePolicy,
|
|
createRequestId: () => "request_wellformed_1234",
|
|
});
|
|
const result = await withTimeout(gateway.capabilities(), "well-formed");
|
|
expect(result.ok).toBe(false);
|
|
expect(result.ok ? null : result.error.code).toBe("QUOTA_EXCEEDED");
|
|
});
|
|
});
|