DongHyeonkaandClaude Opus 5
618da9abf5
Repair the compensating half of the OPFS put saga. The coordinator now owns a single abortPreparedPut() driven by a composition-owned bounded signal instead of the caller's already aborted one, and the worker client no longer issues a duplicate fire-and-forget abort. Journal rows and budget reservations are released only after the physical effect is confirmed CLEANED or ALREADY_CLEAN; a timeout, malformed response or EFFECT_UNKNOWN keeps PREPARING/FILES_READY and returns OBJECT_RECONCILE. New writes carry a transaction-unique physicalGenerationId through the staging receipt, manifest path and prepared object, so a late compensation deletes only its own transaction's directory even when a newer transaction legitimately reuses the same logical generation. v1 paths, receipts and prepared objects stay readable through the rollback window. Abort and cleanup hold the origin mutation lease through physical deletion and staging removal. A transaction that never reached staging returns ALREADY_CLEAN without waiting for the lease, which would otherwise deadlock against the BEGIN it is cancelling. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
736 lines
21 KiB
TypeScript
736 lines
21 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 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,
|
|
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",
|
|
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",
|
|
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",
|
|
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}`,
|
|
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",
|
|
kind: "ABORT_PUT",
|
|
scope: scopeA,
|
|
transactionId,
|
|
}),
|
|
).toMatchObject({ ok: true });
|
|
const finished = await runtime.handleRequest({
|
|
requestId: "request_finish_iso_b",
|
|
kind: "FINISH_PUT",
|
|
scope: scopeB,
|
|
transactionId,
|
|
});
|
|
expect(finished).toMatchObject({ ok: true });
|
|
const preparedObject = preparedValue(finished);
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: "request_verify_iso_b",
|
|
kind: "VERIFY_OBJECT",
|
|
preparedObject,
|
|
}),
|
|
).toEqual({
|
|
requestId: "request_verify_iso_b",
|
|
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}`,
|
|
kind: "APPEND_CHUNK",
|
|
scope: targetScope,
|
|
transactionId,
|
|
sequence: 0,
|
|
bytes: new Uint8Array([99]).buffer,
|
|
});
|
|
const object = preparedValue(
|
|
await runtime.handleRequest({
|
|
requestId: `request_gc_finish_${index}`,
|
|
kind: "FINISH_PUT",
|
|
scope: targetScope,
|
|
transactionId,
|
|
}),
|
|
);
|
|
preparedByScope.set(targetScope.authorityToken, object);
|
|
await runtime.handleRequest({
|
|
requestId: `request_gc_finalize_${index}`,
|
|
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",
|
|
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",
|
|
kind: "DELETE_ORPHAN_CHUNK",
|
|
scope: scopeA,
|
|
digestHex,
|
|
olderThanEpochMs: cutoff,
|
|
}),
|
|
).toMatchObject({ ok: true, value: { deleted: true } });
|
|
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: "request_gc_verify_a",
|
|
kind: "VERIFY_OBJECT",
|
|
preparedObject: preparedByScope.get(scopeA.authorityToken)!,
|
|
}),
|
|
).toEqual({
|
|
requestId: "request_gc_verify_a",
|
|
ok: true,
|
|
value: false,
|
|
});
|
|
expect(
|
|
await runtime.handleRequest({
|
|
requestId: "request_gc_verify_b",
|
|
kind: "VERIFY_OBJECT",
|
|
preparedObject: preparedByScope.get(scopeB.authorityToken)!,
|
|
}),
|
|
).toEqual({
|
|
requestId: "request_gc_verify_b",
|
|
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",
|
|
kind: "CAPABILITIES",
|
|
}),
|
|
).toEqual({
|
|
requestId: "request_caps_1234",
|
|
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",
|
|
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,
|
|
ok: true,
|
|
value: true,
|
|
}
|
|
: {
|
|
requestId: message.requestId,
|
|
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" },
|
|
});
|
|
});
|
|
});
|