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>
1302 lines
38 KiB
TypeScript
1302 lines
38 KiB
TypeScript
import { describe, expect, it, vi } from "vitest";
|
|
|
|
import type {
|
|
OpfsJournalPage,
|
|
OpfsJournalPort,
|
|
OpfsJournalTransaction,
|
|
OpfsPreparedObject,
|
|
OpfsStorageScope,
|
|
} from "../../src/application/ports/browser-file-storage/opfs-ports.ts";
|
|
import type {
|
|
BrowserDataResult,
|
|
BrowserStoragePolicy,
|
|
ByteSource,
|
|
} from "../../src/application/ports/browser-file-storage/shared.ts";
|
|
import {
|
|
createOpfsByteStoreAdapter,
|
|
type OpfsMaintenanceAuthorityConsumer,
|
|
type OpfsMaintenanceAuthorityProvider,
|
|
} from "../../src/adapters/storage/opfs/opfs-byte-store-adapter.ts";
|
|
import {
|
|
resolveOpfsRuntimePolicy,
|
|
} from "../../src/adapters/storage/opfs/opfs-policy.ts";
|
|
import type {
|
|
OpfsWorkerGateway,
|
|
PreparePhysicalObjectRequest,
|
|
} from "../../src/adapters/storage/opfs/opfs-worker-protocol.ts";
|
|
import {
|
|
writeWithSyncAccessHandle,
|
|
} from "../../src/adapters/storage/opfs/opfs-worker-runtime.ts";
|
|
import {
|
|
browserDataFailure,
|
|
browserDataSuccess,
|
|
} from "../../src/adapters/browser-file-storage/result.ts";
|
|
|
|
const storagePolicy: BrowserStoragePolicy = Object.freeze({
|
|
owner: "test-owner",
|
|
namespace: "durable-objects",
|
|
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 scope: OpfsStorageScope = Object.freeze({
|
|
namespace: "durable-objects",
|
|
authorityToken: "authority_12345678",
|
|
namespaceToken: "namespace_12345678",
|
|
partitionToken: "partition_12345678",
|
|
});
|
|
|
|
function sourceFrom(bytes: Uint8Array): ByteSource {
|
|
return {
|
|
byteLength: bytes.byteLength,
|
|
async *stream(signal: AbortSignal) {
|
|
if (signal.aborted) {
|
|
yield browserDataFailure("ABORTED", "OBJECT_READ");
|
|
return;
|
|
}
|
|
yield browserDataSuccess(Uint8Array.from(bytes));
|
|
},
|
|
};
|
|
}
|
|
|
|
function preparedFrom(
|
|
request: PreparePhysicalObjectRequest,
|
|
): OpfsPreparedObject {
|
|
return Object.freeze({
|
|
physicalSchemaVersion: 1,
|
|
descriptor: Object.freeze({
|
|
...request.descriptor,
|
|
integrity: Object.freeze({
|
|
algorithm: "SHA-256-TREE-V1",
|
|
rootDigestHex: "a".repeat(64),
|
|
chunkSizeBytes: 64 * 1024,
|
|
}),
|
|
}),
|
|
chunks: Object.freeze([
|
|
Object.freeze({
|
|
sequence: 0,
|
|
byteLength: request.descriptor.byteLength,
|
|
digestHex: "b".repeat(64),
|
|
}),
|
|
]),
|
|
});
|
|
}
|
|
|
|
function committedObject(
|
|
objectId: string,
|
|
objectPolicy: BrowserStoragePolicy,
|
|
createdAtEpochMs: number,
|
|
byteLength = 1,
|
|
): OpfsPreparedObject {
|
|
return {
|
|
physicalSchemaVersion: 1,
|
|
descriptor: {
|
|
objectId,
|
|
scope,
|
|
generation: 1,
|
|
byteLength,
|
|
mediaType: "application/octet-stream",
|
|
createdAtEpochMs,
|
|
integrity: {
|
|
algorithm: "SHA-256-TREE-V1",
|
|
rootDigestHex: "a".repeat(64),
|
|
chunkSizeBytes: 64 * 1024,
|
|
},
|
|
storagePolicy: objectPolicy,
|
|
},
|
|
chunks: [
|
|
{
|
|
sequence: 0,
|
|
byteLength,
|
|
digestHex: "b".repeat(64),
|
|
},
|
|
],
|
|
};
|
|
}
|
|
|
|
function createJournal(): OpfsJournalPort & {
|
|
readonly transactions: Map<string, OpfsJournalTransaction>;
|
|
readonly objects: Map<string, OpfsPreparedObject>;
|
|
} {
|
|
const transactions = new Map<string, OpfsJournalTransaction>();
|
|
const objects = new Map<string, OpfsPreparedObject>();
|
|
return {
|
|
transactions,
|
|
objects,
|
|
async getCommittedObject(_scope, objectId) {
|
|
return browserDataSuccess(objects.get(objectId) ?? null);
|
|
},
|
|
async begin(input) {
|
|
const current = objects.get(input.objectId);
|
|
if (
|
|
(input.expectedGeneration === null && current) ||
|
|
(input.expectedGeneration !== null &&
|
|
current?.descriptor.generation !== input.expectedGeneration)
|
|
) {
|
|
return browserDataFailure("CONFLICT", "INDEXEDDB_WRITE");
|
|
}
|
|
const transaction: OpfsJournalTransaction = {
|
|
...input,
|
|
fencingToken: "fencing_12345678",
|
|
phase: "PREPARING",
|
|
budgetReservation: {
|
|
namespace: input.scope.namespace,
|
|
reservedBytes: input.targetByteLength,
|
|
hardBudgetBytes:
|
|
input.targetStoragePolicy.hardBudgetBytes,
|
|
},
|
|
};
|
|
transactions.set(input.transactionId, transaction);
|
|
return browserDataSuccess(transaction);
|
|
},
|
|
async markFilesReady(transactionId, fencingToken, preparedObject) {
|
|
const current = transactions.get(transactionId);
|
|
if (!current || current.fencingToken !== fencingToken) {
|
|
return browserDataFailure("CONFLICT", "INDEXEDDB_WRITE");
|
|
}
|
|
const updated: OpfsJournalTransaction = {
|
|
...current,
|
|
phase: "FILES_READY",
|
|
preparedObject,
|
|
};
|
|
transactions.set(transactionId, updated);
|
|
return browserDataSuccess(updated);
|
|
},
|
|
async commitPut(transactionId, fencingToken) {
|
|
const current = transactions.get(transactionId);
|
|
if (
|
|
!current ||
|
|
current.fencingToken !== fencingToken ||
|
|
!current.preparedObject
|
|
) {
|
|
return browserDataFailure("CONFLICT", "INDEXEDDB_WRITE");
|
|
}
|
|
objects.set(current.objectId, current.preparedObject);
|
|
const updated: OpfsJournalTransaction = {
|
|
...current,
|
|
phase: "COMMITTED",
|
|
};
|
|
transactions.set(transactionId, updated);
|
|
return browserDataSuccess(updated);
|
|
},
|
|
async commitDelete(transactionId, fencingToken) {
|
|
const current = transactions.get(transactionId);
|
|
if (!current || current.fencingToken !== fencingToken) {
|
|
return browserDataFailure("CONFLICT", "INDEXEDDB_WRITE");
|
|
}
|
|
objects.delete(current.objectId);
|
|
const updated: OpfsJournalTransaction = {
|
|
...current,
|
|
phase: "COMMITTED",
|
|
};
|
|
transactions.set(transactionId, updated);
|
|
return browserDataSuccess(updated);
|
|
},
|
|
async complete(transactionId, fencingToken) {
|
|
const current = transactions.get(transactionId);
|
|
if (!current || current.fencingToken !== fencingToken) {
|
|
return browserDataFailure("CONFLICT", "INDEXEDDB_WRITE");
|
|
}
|
|
transactions.delete(transactionId);
|
|
return browserDataSuccess(undefined);
|
|
},
|
|
async rollback(transactionId) {
|
|
transactions.delete(transactionId);
|
|
return browserDataSuccess(undefined);
|
|
},
|
|
async listIncomplete(limit) {
|
|
const all = [...transactions.values()];
|
|
const page: OpfsJournalPage = {
|
|
transactions: all.slice(0, limit),
|
|
moreAvailable: all.length > limit,
|
|
};
|
|
return browserDataSuccess(page);
|
|
},
|
|
async listCommittedObjects(request) {
|
|
const values = [...objects.values()].filter(
|
|
(object) =>
|
|
object.descriptor.scope.namespaceToken ===
|
|
request.scope.namespaceToken &&
|
|
object.descriptor.scope.partitionToken ===
|
|
request.scope.partitionToken,
|
|
);
|
|
return browserDataSuccess({
|
|
objects: values.slice(0, request.limit),
|
|
nextObjectId: null,
|
|
moreAvailable: values.length > request.limit,
|
|
});
|
|
},
|
|
async isChunkReferenced(_scope, digestHex) {
|
|
return browserDataSuccess(
|
|
[...objects.values()].some((object) =>
|
|
object.chunks.some((chunk) => chunk.digestHex === digestHex),
|
|
),
|
|
);
|
|
},
|
|
};
|
|
}
|
|
|
|
function createWorker(
|
|
overrides: Partial<OpfsWorkerGateway> = {},
|
|
): OpfsWorkerGateway {
|
|
const physical = new Map<string, OpfsPreparedObject>();
|
|
const worker: OpfsWorkerGateway = {
|
|
async capabilities() {
|
|
return browserDataSuccess({
|
|
available: true,
|
|
dedicatedWorkerRequired: true,
|
|
crossContextMutationLockAvailable: true,
|
|
synchronousAccessHandleAvailable: true,
|
|
});
|
|
},
|
|
async preparePut(request) {
|
|
let count = 0;
|
|
for await (const chunk of request.source.stream(
|
|
request.signal ?? new AbortController().signal,
|
|
)) {
|
|
if (!chunk.ok) return chunk;
|
|
count += chunk.value.byteLength;
|
|
}
|
|
if (count !== request.descriptor.byteLength) {
|
|
return browserDataFailure("INTEGRITY_FAILED", "OBJECT_WRITE");
|
|
}
|
|
const prepared = preparedFrom(request);
|
|
physical.set(request.descriptor.objectId, prepared);
|
|
return browserDataSuccess(prepared);
|
|
},
|
|
async verifyObject(preparedObject) {
|
|
return browserDataSuccess(
|
|
physical.has(preparedObject.descriptor.objectId),
|
|
);
|
|
},
|
|
async openObject(preparedObject) {
|
|
return browserDataSuccess(
|
|
sourceFrom(
|
|
new Uint8Array(preparedObject.descriptor.byteLength).fill(7),
|
|
),
|
|
);
|
|
},
|
|
async removeObject(_scope, objectId) {
|
|
physical.delete(objectId);
|
|
return browserDataSuccess(undefined);
|
|
},
|
|
async cleanupTransaction() {
|
|
return browserDataSuccess({ kind: "CLEANED" as const });
|
|
},
|
|
async abortPreparedPut() {
|
|
return browserDataSuccess({ kind: "CLEANED" as const });
|
|
},
|
|
async finalizePut() {
|
|
return browserDataSuccess(undefined);
|
|
},
|
|
async listOrphanCandidates() {
|
|
return browserDataSuccess({
|
|
safeToSweep: true,
|
|
digests: [],
|
|
moreAvailable: false,
|
|
});
|
|
},
|
|
async deleteOrphanChunk() {
|
|
return browserDataSuccess({
|
|
deleted: false,
|
|
skippedUnsafe: false,
|
|
});
|
|
},
|
|
close() {},
|
|
};
|
|
return Object.assign(worker, overrides);
|
|
}
|
|
|
|
describe("OPFS byte-store coordinator", () => {
|
|
it("publishes bytes only after FILES_READY and an atomic journal commit", async () => {
|
|
const journal = createJournal();
|
|
const observations: unknown[] = [];
|
|
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),
|
|
});
|
|
|
|
const result = await adapter.objects.put({
|
|
objectId: "object_12345678",
|
|
expectedGeneration: null,
|
|
mediaType: "application/octet-stream",
|
|
source: sourceFrom(new Uint8Array([1, 2, 3])),
|
|
});
|
|
|
|
expect(result).toMatchObject({
|
|
ok: true,
|
|
value: { objectId: "object_12345678", generation: 1 },
|
|
});
|
|
expect(journal.objects.get("object_12345678")).toBeDefined();
|
|
expect(journal.transactions.size).toBe(0);
|
|
expect(JSON.stringify(observations)).not.toContain("object_12345678");
|
|
});
|
|
|
|
it("deep-snapshots scope and policy at composition against caller mutation", async () => {
|
|
const journal = createJournal();
|
|
const mutableScope = { ...scope };
|
|
const mutableRetention = {
|
|
kind: "TTL" as const,
|
|
maxAgeMs: 60_000,
|
|
};
|
|
const mutablePolicy = {
|
|
...storagePolicy,
|
|
retention: mutableRetention,
|
|
};
|
|
const adapter = createOpfsByteStoreAdapter({
|
|
journal,
|
|
worker: createWorker(),
|
|
scope: mutableScope,
|
|
storagePolicy: mutablePolicy,
|
|
policy: {
|
|
...resolveOpfsRuntimePolicy(),
|
|
chunkSizeBytes: 64 * 1024,
|
|
maxObjectBytes: 64 * 1024,
|
|
maxChunkCount: 1,
|
|
},
|
|
createTransactionId: () => "transaction_snapshot_1234",
|
|
now: () => 100,
|
|
});
|
|
|
|
mutableScope.namespace = "mutated-namespace";
|
|
mutableScope.partitionToken = "partition_mutated_1234";
|
|
mutablePolicy.namespace = "mutated-namespace";
|
|
mutableRetention.maxAgeMs = 1;
|
|
|
|
const result = await adapter.objects.put({
|
|
objectId: "object_snapshot_1234",
|
|
expectedGeneration: null,
|
|
mediaType: "application/octet-stream",
|
|
source: sourceFrom(new Uint8Array([1, 2, 3])),
|
|
});
|
|
|
|
expect(result).toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
scope,
|
|
storagePolicy: {
|
|
namespace: scope.namespace,
|
|
retention: { kind: "TTL", maxAgeMs: 60_000 },
|
|
},
|
|
},
|
|
});
|
|
expect(result.ok && Object.isFrozen(result.value.scope)).toBe(true);
|
|
expect(
|
|
result.ok && Object.isFrozen(result.value.storagePolicy.retention),
|
|
).toBe(true);
|
|
});
|
|
|
|
it("snapshots a put request at entry and dependency methods at composition", async () => {
|
|
const journal = createJournal();
|
|
const originalGetCommittedObject =
|
|
journal.getCommittedObject.bind(journal);
|
|
let markJournalEntered!: () => void;
|
|
let releaseJournal!: () => void;
|
|
const journalEntered = new Promise<void>((resolve) => {
|
|
markJournalEntered = resolve;
|
|
});
|
|
const journalRelease = new Promise<void>((resolve) => {
|
|
releaseJournal = resolve;
|
|
});
|
|
journal.getCommittedObject = async (...args) => {
|
|
markJournalEntered();
|
|
await journalRelease;
|
|
return await originalGetCommittedObject(...args);
|
|
};
|
|
|
|
const worker = createWorker();
|
|
const originalPreparePut = worker.preparePut.bind(worker);
|
|
const preparePut = vi.fn(originalPreparePut);
|
|
worker.preparePut = preparePut;
|
|
const adapter = createOpfsByteStoreAdapter({
|
|
journal,
|
|
worker,
|
|
scope,
|
|
storagePolicy,
|
|
policy: {
|
|
...resolveOpfsRuntimePolicy(),
|
|
chunkSizeBytes: 64 * 1024,
|
|
maxObjectBytes: 64 * 1024,
|
|
maxChunkCount: 1,
|
|
},
|
|
createTransactionId: () => "transaction_request_snapshot",
|
|
now: () => 100,
|
|
});
|
|
|
|
const mutableSource = {
|
|
byteLength: 3,
|
|
async *stream(_signal: AbortSignal) {
|
|
yield browserDataSuccess(new Uint8Array([1, 2, 3]));
|
|
},
|
|
};
|
|
const mutableRequest: {
|
|
objectId: string;
|
|
expectedGeneration: number | null;
|
|
mediaType: string;
|
|
source: typeof mutableSource;
|
|
} = {
|
|
objectId: "object_request_snapshot",
|
|
expectedGeneration: null,
|
|
mediaType: "application/octet-stream",
|
|
source: mutableSource,
|
|
};
|
|
|
|
const pending = adapter.objects.put(mutableRequest);
|
|
await journalEntered;
|
|
|
|
mutableRequest.objectId = "mutated_object_request";
|
|
mutableRequest.expectedGeneration = 99;
|
|
mutableRequest.mediaType = "text/html";
|
|
mutableSource.byteLength = 999;
|
|
mutableSource.stream = async function* (_signal: AbortSignal) {
|
|
yield browserDataSuccess(new Uint8Array([9]));
|
|
};
|
|
const replacementJournalMethod = vi.fn(async () =>
|
|
browserDataFailure("UNAVAILABLE", "INDEXEDDB_READ"),
|
|
);
|
|
const replacementWorkerMethod = vi.fn(async () =>
|
|
browserDataFailure("UNAVAILABLE", "OBJECT_WRITE"),
|
|
);
|
|
journal.getCommittedObject = replacementJournalMethod;
|
|
worker.preparePut = replacementWorkerMethod;
|
|
releaseJournal();
|
|
|
|
const result = await pending;
|
|
|
|
expect(result).toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
objectId: "object_request_snapshot",
|
|
byteLength: 3,
|
|
mediaType: "application/octet-stream",
|
|
},
|
|
});
|
|
expect(preparePut).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
descriptor: expect.objectContaining({
|
|
objectId: "object_request_snapshot",
|
|
byteLength: 3,
|
|
mediaType: "application/octet-stream",
|
|
}),
|
|
}),
|
|
);
|
|
expect(replacementJournalMethod).not.toHaveBeenCalled();
|
|
expect(replacementWorkerMethod).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("keeps PREPARING journal when compensating cleanup is aborted or unavailable", async () => {
|
|
for (const failing of [
|
|
{
|
|
async abortPreparedPut() {
|
|
return browserDataFailure("ABORTED", "OBJECT_RECONCILE");
|
|
},
|
|
},
|
|
{
|
|
async abortPreparedPut() {
|
|
return browserDataSuccess({ kind: "EFFECT_UNKNOWN" as const });
|
|
},
|
|
},
|
|
]) {
|
|
const journal = createJournal();
|
|
const worker = createWorker({
|
|
async preparePut() {
|
|
return browserDataFailure("QUOTA_EXCEEDED", "OBJECT_WRITE");
|
|
},
|
|
...failing,
|
|
});
|
|
const adapter = createOpfsByteStoreAdapter({
|
|
journal,
|
|
worker,
|
|
scope,
|
|
storagePolicy,
|
|
createTransactionId: () => "transaction_12345678",
|
|
now: () => 100,
|
|
});
|
|
|
|
const result = await adapter.objects.put({
|
|
objectId: "object_12345678",
|
|
expectedGeneration: null,
|
|
mediaType: "application/octet-stream",
|
|
source: sourceFrom(new Uint8Array([1])),
|
|
});
|
|
|
|
expect(result.ok).toBe(false);
|
|
// The journal row is the only durable evidence that staging bytes may
|
|
// still exist, so it survives an unconfirmed compensation.
|
|
expect(
|
|
journal.transactions.get("transaction_12345678")?.phase,
|
|
).toBe("PREPARING");
|
|
}
|
|
});
|
|
|
|
it("does not roll back journal after an unknown worker mutation effect", async () => {
|
|
const journal = createJournal();
|
|
const worker = createWorker({
|
|
async abortPreparedPut() {
|
|
return browserDataSuccess({ kind: "EFFECT_UNKNOWN" as const });
|
|
},
|
|
});
|
|
const rollback = vi.spyOn(journal, "rollback");
|
|
vi.spyOn(journal, "markFilesReady").mockResolvedValueOnce(
|
|
browserDataFailure("UNAVAILABLE", "OBJECT_WRITE"),
|
|
);
|
|
const adapter = createOpfsByteStoreAdapter({
|
|
journal,
|
|
worker,
|
|
scope,
|
|
storagePolicy,
|
|
createTransactionId: () => "transaction_12345678",
|
|
now: () => 100,
|
|
});
|
|
|
|
const result = await adapter.objects.put({
|
|
objectId: "object_12345678",
|
|
expectedGeneration: null,
|
|
mediaType: "application/octet-stream",
|
|
source: sourceFrom(new Uint8Array([1])),
|
|
});
|
|
|
|
expect(result.ok).toBe(false);
|
|
expect(rollback).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("holds the OPFS mutation lease until exact physical cleanup completes", async () => {
|
|
const journal = createJournal();
|
|
const observed: Array<Readonly<Record<string, unknown>>> = [];
|
|
const worker = createWorker({
|
|
async preparePut() {
|
|
return browserDataFailure("QUOTA_EXCEEDED", "OBJECT_WRITE");
|
|
},
|
|
async abortPreparedPut(request) {
|
|
observed.push({ ...request });
|
|
return browserDataSuccess({ kind: "CLEANED" as const });
|
|
},
|
|
});
|
|
const compensation = new AbortController();
|
|
const caller = new AbortController();
|
|
caller.abort();
|
|
const adapter = createOpfsByteStoreAdapter({
|
|
journal,
|
|
worker,
|
|
scope,
|
|
storagePolicy,
|
|
createTransactionId: () => "transaction_12345678",
|
|
createPhysicalGenerationId: () => "f".repeat(32) as never,
|
|
compensationSignal: compensation.signal,
|
|
now: () => 100,
|
|
});
|
|
|
|
await adapter.objects.put({
|
|
objectId: "object_12345678",
|
|
expectedGeneration: null,
|
|
mediaType: "application/octet-stream",
|
|
source: sourceFrom(new Uint8Array([1])),
|
|
});
|
|
|
|
expect(observed).toHaveLength(1);
|
|
expect(observed[0]).toMatchObject({
|
|
transactionId: "transaction_12345678",
|
|
physicalGenerationId: "f".repeat(32),
|
|
});
|
|
// Compensation never inherits the caller signal.
|
|
expect(observed[0]?.signal).toBe(compensation.signal);
|
|
expect(
|
|
journal.transactions.has("transaction_12345678"),
|
|
).toBe(false);
|
|
});
|
|
|
|
it("delayed stale cleanup cannot delete a reused logical generation", async () => {
|
|
const journal = createJournal();
|
|
const issued: string[] = [];
|
|
let nextToken = 0;
|
|
const worker = createWorker({
|
|
async abortPreparedPut(request) {
|
|
issued.push(request.physicalGenerationId);
|
|
return browserDataSuccess({ kind: "CLEANED" as const });
|
|
},
|
|
});
|
|
let transaction = 0;
|
|
// T1 abandons its prepared put at the same logical generation 1.
|
|
vi.spyOn(journal, "markFilesReady").mockResolvedValueOnce(
|
|
browserDataFailure("UNAVAILABLE", "OBJECT_WRITE"),
|
|
);
|
|
const adapter = createOpfsByteStoreAdapter({
|
|
journal,
|
|
worker,
|
|
scope,
|
|
storagePolicy,
|
|
createTransactionId: () => `transaction_1234567${(transaction += 1)}`,
|
|
createPhysicalGenerationId: () =>
|
|
String(nextToken += 1).padStart(32, "0") as never,
|
|
now: () => 100,
|
|
});
|
|
const first = await adapter.objects.put({
|
|
objectId: "object_12345678",
|
|
expectedGeneration: null,
|
|
mediaType: "application/octet-stream",
|
|
source: sourceFrom(new Uint8Array([1])),
|
|
});
|
|
expect(first.ok).toBe(false);
|
|
|
|
// T2 legitimately reuses logical generation 1 with a different token.
|
|
const second = await adapter.objects.put({
|
|
objectId: "object_12345678",
|
|
expectedGeneration: null,
|
|
mediaType: "application/octet-stream",
|
|
source: sourceFrom(new Uint8Array([1])),
|
|
});
|
|
expect(second.ok).toBe(true);
|
|
if (second.ok) expect(second.value.generation).toBe(1);
|
|
|
|
// The stale compensation targeted only T1's physical token.
|
|
expect(issued).toEqual([String(1).padStart(32, "0")]);
|
|
const stored = journal.objects.get("object_12345678");
|
|
expect(stored?.physicalSchemaVersion).toBe(1);
|
|
});
|
|
|
|
it("keeps a committed journal row for reconciliation when cleanup fails", async () => {
|
|
const journal = createJournal();
|
|
const worker = createWorker({
|
|
async finalizePut() {
|
|
return browserDataFailure("UNAVAILABLE", "OBJECT_RECONCILE", {
|
|
retryable: true,
|
|
recovery: "RETRY",
|
|
});
|
|
},
|
|
});
|
|
const adapter = createOpfsByteStoreAdapter({
|
|
journal,
|
|
worker,
|
|
scope,
|
|
storagePolicy,
|
|
policy: {
|
|
...resolveOpfsRuntimePolicy(),
|
|
chunkSizeBytes: 64 * 1024,
|
|
maxObjectBytes: 64 * 1024,
|
|
maxChunkCount: 1,
|
|
},
|
|
createTransactionId: () => "transaction_12345678",
|
|
now: () => 100,
|
|
});
|
|
|
|
const result = await adapter.objects.put({
|
|
objectId: "object_12345678",
|
|
expectedGeneration: null,
|
|
mediaType: "application/octet-stream",
|
|
source: sourceFrom(new Uint8Array([1])),
|
|
});
|
|
|
|
expect(result.ok).toBe(true);
|
|
expect(
|
|
journal.transactions.get("transaction_12345678")?.phase,
|
|
).toBe("COMMITTED");
|
|
});
|
|
|
|
it("closes a sync access handle even when flush fails", async () => {
|
|
const calls: string[] = [];
|
|
const fakeFileHandle = {
|
|
async createSyncAccessHandle() {
|
|
return {
|
|
truncate: (size: number) => calls.push(`truncate:${size}`),
|
|
write: (part: ArrayBufferView) => {
|
|
calls.push(`write:${part.byteLength}`);
|
|
return part.byteLength;
|
|
},
|
|
flush: () => {
|
|
calls.push("flush");
|
|
throw new Error("disk failure");
|
|
},
|
|
close: () => calls.push("close"),
|
|
};
|
|
},
|
|
};
|
|
|
|
await expect(
|
|
writeWithSyncAccessHandle(
|
|
fakeFileHandle as never,
|
|
new Uint8Array([1, 2, 3]),
|
|
),
|
|
).rejects.toThrow("disk failure");
|
|
expect(calls.at(-1)).toBe("close");
|
|
});
|
|
|
|
it("enforces TTL and logout policy within the runtime-bound partition", async () => {
|
|
const journal = createJournal();
|
|
journal.objects.set(
|
|
"expired_12345678",
|
|
committedObject(
|
|
"expired_12345678",
|
|
{
|
|
...storagePolicy,
|
|
retention: { kind: "TTL", maxAgeMs: 10 },
|
|
},
|
|
1,
|
|
7,
|
|
),
|
|
);
|
|
journal.objects.set(
|
|
"logout_12345678",
|
|
committedObject(
|
|
"logout_12345678",
|
|
{
|
|
...storagePolicy,
|
|
retention: { kind: "TTL", maxAgeMs: 10 },
|
|
},
|
|
95,
|
|
9,
|
|
),
|
|
);
|
|
const removed: string[] = [];
|
|
const adapter = createOpfsByteStoreAdapter({
|
|
journal,
|
|
worker: createWorker({
|
|
async removeObject(_scope, objectId) {
|
|
removed.push(objectId);
|
|
return browserDataSuccess(undefined);
|
|
},
|
|
}),
|
|
scope,
|
|
storagePolicy: {
|
|
...storagePolicy,
|
|
retention: { kind: "TTL", maxAgeMs: 10 },
|
|
},
|
|
policy: {
|
|
...resolveOpfsRuntimePolicy(),
|
|
chunkSizeBytes: 64 * 1024,
|
|
maxObjectBytes: 64 * 1024,
|
|
maxChunkCount: 1,
|
|
},
|
|
createTransactionId: (() => {
|
|
let sequence = 0;
|
|
return () => `transaction_policy_${++sequence}_1234`;
|
|
})(),
|
|
now: () => 100,
|
|
requestMaintenanceAuthority: async ({ reason }) =>
|
|
reason === "LOGOUT"
|
|
? {
|
|
authorized: true,
|
|
proofToken: "receipt_1234567890",
|
|
expiresAtEpochMs: 200,
|
|
}
|
|
: { authorized: false },
|
|
consumeMaintenanceAuthority: async ({
|
|
reason,
|
|
proofToken,
|
|
expiresAtEpochMs,
|
|
}) =>
|
|
reason === "LOGOUT" &&
|
|
proofToken === "receipt_1234567890" &&
|
|
expiresAtEpochMs === 200,
|
|
});
|
|
|
|
expect(
|
|
await adapter.maintenance.enforcePolicies({
|
|
reason: "TTL",
|
|
}),
|
|
).toMatchObject({
|
|
ok: true,
|
|
value: { removedObjects: 1, releasedBytes: 7 },
|
|
});
|
|
expect(removed).toEqual(["expired_12345678"]);
|
|
expect(
|
|
await adapter.maintenance.enforcePolicies({
|
|
reason: "LOGOUT",
|
|
}),
|
|
).toMatchObject({
|
|
ok: true,
|
|
value: { removedObjects: 1, releasedBytes: 9 },
|
|
});
|
|
expect(removed).toEqual([
|
|
"expired_12345678",
|
|
"logout_12345678",
|
|
]);
|
|
});
|
|
|
|
it("pressure eviction skips user-authored data and rejects invalid maintenance budgets", async () => {
|
|
const journal = createJournal();
|
|
const reconstructablePolicy: BrowserStoragePolicy = {
|
|
owner: "test-owner",
|
|
namespace: scope.namespace,
|
|
classification: "PUBLIC",
|
|
authority: "RECONSTRUCTABLE",
|
|
accountScope: "ORIGIN_SHARED",
|
|
retention: { kind: "TTL", maxAgeMs: 60_000 },
|
|
softBudgetBytes: 100,
|
|
hardBudgetBytes: 200,
|
|
evictionPriority: "RECONSTRUCTABLE",
|
|
logoutAction: "KEEP_ORIGIN_SHARED",
|
|
accountDeletionAction: "KEEP_ORIGIN_SHARED",
|
|
pressureAction: "EVICT_RECONSTRUCTABLE",
|
|
unavailableFallback: "ONLINE_ONLY",
|
|
};
|
|
journal.objects.set(
|
|
"cacheable_12345678",
|
|
committedObject(
|
|
"cacheable_12345678",
|
|
reconstructablePolicy,
|
|
1,
|
|
12,
|
|
),
|
|
);
|
|
journal.objects.set(
|
|
"authored_12345678",
|
|
committedObject(
|
|
"authored_12345678",
|
|
storagePolicy,
|
|
2,
|
|
20,
|
|
),
|
|
);
|
|
const removed: string[] = [];
|
|
const adapter = createOpfsByteStoreAdapter({
|
|
journal,
|
|
worker: createWorker({
|
|
async removeObject(_scope, objectId) {
|
|
removed.push(objectId);
|
|
return browserDataSuccess(undefined);
|
|
},
|
|
}),
|
|
scope,
|
|
storagePolicy: reconstructablePolicy,
|
|
policy: {
|
|
...resolveOpfsRuntimePolicy(),
|
|
chunkSizeBytes: 64 * 1024,
|
|
maxObjectBytes: 64 * 1024,
|
|
maxChunkCount: 1,
|
|
reconciliationBudgetMs: 10,
|
|
reconciliationBatchSize: 1,
|
|
},
|
|
createTransactionId: () => "transaction_pressure_1234",
|
|
now: () => 100,
|
|
});
|
|
|
|
expect(
|
|
await adapter.maintenance.enforcePolicies({
|
|
reason: "PRESSURE",
|
|
targetBytesToRelease: 1,
|
|
}),
|
|
).toMatchObject({
|
|
ok: true,
|
|
value: { removedObjects: 1, releasedBytes: 12 },
|
|
});
|
|
expect(removed).toEqual(["cacheable_12345678"]);
|
|
expect(journal.objects.has("authored_12345678")).toBe(true);
|
|
|
|
for (const invalid of [0, -1, Number.NaN]) {
|
|
expect(
|
|
await adapter.maintenance.reconcile({ budgetMs: invalid }),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "INVALID_INPUT" },
|
|
});
|
|
expect(
|
|
await adapter.maintenance.enforcePolicies({
|
|
reason: "TTL",
|
|
maxObjects: invalid,
|
|
}),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "INVALID_INPUT" },
|
|
});
|
|
}
|
|
expect(
|
|
await adapter.maintenance.reconcile({ budgetMs: 11 }),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "LIMIT_EXCEEDED" },
|
|
});
|
|
expect(
|
|
await adapter.maintenance.enforcePolicies({
|
|
reason: "PRESSURE",
|
|
targetBytesToRelease: 1,
|
|
maxObjects: 2,
|
|
}),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "LIMIT_EXCEEDED" },
|
|
});
|
|
});
|
|
|
|
it("runs bounded orphan GC and fails closed when staging is unreadable", async () => {
|
|
const digestHex = "c".repeat(64);
|
|
const listOrphans = vi.fn(async () =>
|
|
browserDataSuccess({
|
|
safeToSweep: true,
|
|
digests: [digestHex],
|
|
moreAvailable: false,
|
|
}),
|
|
);
|
|
const deleteOrphan = vi.fn(async () =>
|
|
browserDataSuccess({
|
|
deleted: true,
|
|
skippedUnsafe: false,
|
|
}),
|
|
);
|
|
const adapter = createOpfsByteStoreAdapter({
|
|
journal: createJournal(),
|
|
worker: createWorker({
|
|
listOrphanCandidates: listOrphans,
|
|
deleteOrphanChunk: deleteOrphan,
|
|
}),
|
|
scope,
|
|
storagePolicy,
|
|
policy: {
|
|
...resolveOpfsRuntimePolicy(),
|
|
chunkSizeBytes: 64 * 1024,
|
|
maxObjectBytes: 64 * 1024,
|
|
maxChunkCount: 1,
|
|
},
|
|
now: () => 100_000,
|
|
});
|
|
expect(await adapter.maintenance.reconcile()).toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
inspectedOrphanChunks: 1,
|
|
deletedOrphanChunks: 1,
|
|
orphanGcStatus: "COMPLETED",
|
|
},
|
|
});
|
|
expect(listOrphans).toHaveBeenCalledWith(
|
|
scope,
|
|
expect.any(Number),
|
|
expect.any(Number),
|
|
undefined,
|
|
);
|
|
expect(deleteOrphan).toHaveBeenCalledWith(
|
|
scope,
|
|
digestHex,
|
|
expect.any(Number),
|
|
undefined,
|
|
);
|
|
|
|
const failClosed = createOpfsByteStoreAdapter({
|
|
journal: createJournal(),
|
|
worker: createWorker({
|
|
async listOrphanCandidates() {
|
|
return browserDataSuccess({
|
|
safeToSweep: false,
|
|
digests: [],
|
|
moreAvailable: false,
|
|
});
|
|
},
|
|
}),
|
|
scope,
|
|
storagePolicy,
|
|
policy: {
|
|
...resolveOpfsRuntimePolicy(),
|
|
chunkSizeBytes: 64 * 1024,
|
|
maxObjectBytes: 64 * 1024,
|
|
maxChunkCount: 1,
|
|
},
|
|
now: () => 100_000,
|
|
});
|
|
expect(await failClosed.maintenance.reconcile()).toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
orphanGcStatus: "STAGING_STATE_UNREADABLE",
|
|
deletedOrphanChunks: 0,
|
|
},
|
|
});
|
|
});
|
|
|
|
it("does not scan and reports remaining work when authority consumes the maintenance budget", async () => {
|
|
const syncedPolicy: BrowserStoragePolicy = {
|
|
...storagePolicy,
|
|
authority: "SERVER",
|
|
retention: { kind: "UNTIL_SYNCED" },
|
|
evictionPriority: "SYNCED_COPY",
|
|
unavailableFallback: "READ_ONLY",
|
|
};
|
|
const journal = createJournal();
|
|
journal.objects.set(
|
|
"synced_budget_1234",
|
|
committedObject("synced_budget_1234", syncedPolicy, 1, 11),
|
|
);
|
|
const listCommittedObjects = vi.spyOn(
|
|
journal,
|
|
"listCommittedObjects",
|
|
);
|
|
const worker = createWorker();
|
|
const removeObject = vi.spyOn(worker, "removeObject");
|
|
let epochMs = 100;
|
|
const adapter = createOpfsByteStoreAdapter({
|
|
journal,
|
|
worker,
|
|
scope,
|
|
storagePolicy: syncedPolicy,
|
|
policy: {
|
|
...resolveOpfsRuntimePolicy(),
|
|
chunkSizeBytes: 64 * 1024,
|
|
maxObjectBytes: 64 * 1024,
|
|
maxChunkCount: 1,
|
|
reconciliationBudgetMs: 10,
|
|
reconciliationBatchSize: 1,
|
|
},
|
|
now: () => epochMs,
|
|
requestMaintenanceAuthority: async () => {
|
|
epochMs = 110;
|
|
return {
|
|
authorized: true,
|
|
proofToken: "budget_1234567890",
|
|
expiresAtEpochMs: 200,
|
|
};
|
|
},
|
|
consumeMaintenanceAuthority: async () => true,
|
|
});
|
|
|
|
expect(
|
|
await adapter.maintenance.enforcePolicies({
|
|
reason: "UNTIL_SYNCED",
|
|
}),
|
|
).toEqual({
|
|
ok: true,
|
|
value: {
|
|
inspectedObjects: 0,
|
|
removedObjects: 0,
|
|
releasedBytes: 0,
|
|
moreObjectsAvailable: true,
|
|
deadlineReached: true,
|
|
},
|
|
});
|
|
expect(listCommittedObjects).not.toHaveBeenCalled();
|
|
expect(removeObject).not.toHaveBeenCalled();
|
|
expect(journal.objects.has("synced_budget_1234")).toBe(true);
|
|
});
|
|
|
|
it("obtains and atomically consumes composition authority for sensitive maintenance", async () => {
|
|
const syncedPolicy: BrowserStoragePolicy = {
|
|
...storagePolicy,
|
|
authority: "SERVER",
|
|
retention: { kind: "UNTIL_SYNCED" },
|
|
evictionPriority: "SYNCED_COPY",
|
|
unavailableFallback: "READ_ONLY",
|
|
};
|
|
const createAdapter = (
|
|
authority: Readonly<{
|
|
requestMaintenanceAuthority?: OpfsMaintenanceAuthorityProvider;
|
|
consumeMaintenanceAuthority?: OpfsMaintenanceAuthorityConsumer;
|
|
}> = {},
|
|
) => {
|
|
const journal = createJournal();
|
|
journal.objects.set(
|
|
"synced_12345678",
|
|
committedObject(
|
|
"synced_12345678",
|
|
syncedPolicy,
|
|
1,
|
|
11,
|
|
),
|
|
);
|
|
const removed: string[] = [];
|
|
return {
|
|
journal,
|
|
removed,
|
|
adapter: createOpfsByteStoreAdapter({
|
|
journal,
|
|
worker: createWorker({
|
|
async removeObject(_scope, objectId) {
|
|
removed.push(objectId);
|
|
return browserDataSuccess(undefined);
|
|
},
|
|
}),
|
|
scope,
|
|
storagePolicy: syncedPolicy,
|
|
policy: {
|
|
...resolveOpfsRuntimePolicy(),
|
|
chunkSizeBytes: 64 * 1024,
|
|
maxObjectBytes: 64 * 1024,
|
|
maxChunkCount: 1,
|
|
},
|
|
createTransactionId: () => "transaction_proof_1234",
|
|
now: () => 100,
|
|
...authority,
|
|
}),
|
|
};
|
|
};
|
|
|
|
const noAuthority = createAdapter();
|
|
expect(
|
|
await noAuthority.adapter.maintenance.enforcePolicies({
|
|
reason: "UNTIL_SYNCED",
|
|
}),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "POLICY_REJECTED" },
|
|
});
|
|
expect(noAuthority.removed).toEqual([]);
|
|
expect(
|
|
await noAuthority.adapter.maintenance.enforcePolicies({
|
|
reason: "LOGOUT",
|
|
}),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "POLICY_REJECTED" },
|
|
});
|
|
expect(noAuthority.removed).toEqual([]);
|
|
|
|
const deniedConsumer = vi.fn(async () => true);
|
|
const denied = createAdapter({
|
|
requestMaintenanceAuthority: async () => ({ authorized: false }),
|
|
consumeMaintenanceAuthority: deniedConsumer,
|
|
});
|
|
expect(
|
|
await denied.adapter.maintenance.enforcePolicies({
|
|
reason: "UNTIL_SYNCED",
|
|
}),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "POLICY_REJECTED" },
|
|
});
|
|
expect(denied.removed).toEqual([]);
|
|
expect(deniedConsumer).not.toHaveBeenCalled();
|
|
|
|
let markProviderEntered!: () => void;
|
|
let releaseProvider!: () => void;
|
|
const providerEntered = new Promise<void>((resolve) => {
|
|
markProviderEntered = resolve;
|
|
});
|
|
const providerRelease = new Promise<void>((resolve) => {
|
|
releaseProvider = resolve;
|
|
});
|
|
const providerImplementation: OpfsMaintenanceAuthorityProvider =
|
|
async ({ reason }) => {
|
|
markProviderEntered();
|
|
await providerRelease;
|
|
return reason === "UNTIL_SYNCED"
|
|
? {
|
|
authorized: true,
|
|
proofToken: "receipt_1234567890",
|
|
expiresAtEpochMs: 200,
|
|
}
|
|
: { authorized: false };
|
|
};
|
|
const provider = vi.fn(providerImplementation);
|
|
const consumedProofs = new Set<string>();
|
|
const consumerImplementation: OpfsMaintenanceAuthorityConsumer =
|
|
async (request) => {
|
|
if (
|
|
request.reason !== "UNTIL_SYNCED" ||
|
|
request.scope.partitionToken !== scope.partitionToken ||
|
|
request.storagePolicy.retention.kind !== "UNTIL_SYNCED" ||
|
|
request.expiresAtEpochMs !== 200 ||
|
|
consumedProofs.has(request.proofToken)
|
|
) {
|
|
return false;
|
|
}
|
|
consumedProofs.add(request.proofToken);
|
|
return true;
|
|
};
|
|
const consumer = vi.fn(consumerImplementation);
|
|
const authorized = createAdapter({
|
|
requestMaintenanceAuthority: provider,
|
|
consumeMaintenanceAuthority: consumer,
|
|
});
|
|
const mutableMaintenanceRequest = {
|
|
reason: "UNTIL_SYNCED" as const,
|
|
maxObjects: 1,
|
|
};
|
|
const pendingMaintenance =
|
|
authorized.adapter.maintenance.enforcePolicies(
|
|
mutableMaintenanceRequest,
|
|
);
|
|
await providerEntered;
|
|
const mutationTarget = mutableMaintenanceRequest as {
|
|
reason: string;
|
|
maxObjects: number;
|
|
};
|
|
mutationTarget.reason = "ACCOUNT_DELETION";
|
|
mutationTarget.maxObjects = 999;
|
|
releaseProvider();
|
|
|
|
expect(await pendingMaintenance).toMatchObject({
|
|
ok: true,
|
|
value: { removedObjects: 1, releasedBytes: 11 },
|
|
});
|
|
expect(provider).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
reason: "UNTIL_SYNCED",
|
|
scope,
|
|
storagePolicy: syncedPolicy,
|
|
}),
|
|
);
|
|
expect(consumer).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
reason: "UNTIL_SYNCED",
|
|
proofToken: "receipt_1234567890",
|
|
expiresAtEpochMs: 200,
|
|
scope,
|
|
storagePolicy: syncedPolicy,
|
|
}),
|
|
);
|
|
expect(authorized.removed).toEqual(["synced_12345678"]);
|
|
|
|
authorized.journal.objects.set(
|
|
"synced_12345678",
|
|
committedObject("synced_12345678", syncedPolicy, 1, 11),
|
|
);
|
|
expect(
|
|
await authorized.adapter.maintenance.enforcePolicies({
|
|
reason: "UNTIL_SYNCED",
|
|
}),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "POLICY_REJECTED" },
|
|
});
|
|
expect(authorized.removed).toEqual(["synced_12345678"]);
|
|
|
|
const expiredConsumer = vi.fn(async () => true);
|
|
const expired = createAdapter({
|
|
requestMaintenanceAuthority: async () => ({
|
|
authorized: true,
|
|
proofToken: "expired_1234567890",
|
|
expiresAtEpochMs: 100,
|
|
}),
|
|
consumeMaintenanceAuthority: expiredConsumer,
|
|
});
|
|
expect(
|
|
await expired.adapter.maintenance.enforcePolicies({
|
|
reason: "UNTIL_SYNCED",
|
|
}),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "POLICY_REJECTED" },
|
|
});
|
|
expect(expiredConsumer).not.toHaveBeenCalled();
|
|
expect(expired.removed).toEqual([]);
|
|
|
|
const accountDeletionDenied = createAdapter({
|
|
requestMaintenanceAuthority: async () => ({ authorized: false }),
|
|
consumeMaintenanceAuthority: async () => true,
|
|
});
|
|
expect(
|
|
await accountDeletionDenied.adapter.maintenance.enforcePolicies({
|
|
reason: "ACCOUNT_DELETION",
|
|
}),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "POLICY_REJECTED" },
|
|
});
|
|
expect(accountDeletionDenied.removed).toEqual([]);
|
|
});
|
|
});
|