432 lines
12 KiB
TypeScript
432 lines
12 KiB
TypeScript
import { describe, expect, it } from "vitest";
|
|
|
|
import type {
|
|
BeginOpfsJournalTransaction,
|
|
OpfsPreparedObject,
|
|
OpfsStorageScope,
|
|
} from "../../src/application/ports/browser-file-storage/opfs-ports.ts";
|
|
import type {
|
|
BrowserStoragePolicy,
|
|
} from "../../src/application/ports/browser-file-storage/shared.ts";
|
|
import {
|
|
createIndexedDbOpfsJournal,
|
|
opfsJournalDatabaseName,
|
|
} from "../../src/adapters/storage/opfs/indexeddb-opfs-journal.ts";
|
|
import { MemoryIndexedDbFactory } from "../helpers/memory-indexeddb.ts";
|
|
|
|
const authorityToken = "authority_12345678";
|
|
const scope: OpfsStorageScope = Object.freeze({
|
|
namespace: "durable-objects",
|
|
authorityToken,
|
|
namespaceToken: "namespace_12345678",
|
|
partitionToken: "partition_12345678",
|
|
});
|
|
const otherPartition: OpfsStorageScope = Object.freeze({
|
|
...scope,
|
|
partitionToken: "partition_87654321",
|
|
});
|
|
const policy: BrowserStoragePolicy = Object.freeze({
|
|
owner: "test-owner",
|
|
namespace: scope.namespace,
|
|
classification: "PERSONAL",
|
|
authority: "LOCAL_FIRST",
|
|
accountScope: "OPAQUE_PARTITION",
|
|
retention: Object.freeze({ kind: "EXPLICIT_DELETE" }),
|
|
softBudgetBytes: 8,
|
|
hardBudgetBytes: 10,
|
|
evictionPriority: "USER_AUTHORED",
|
|
logoutAction: "EXPORT_THEN_PURGE",
|
|
accountDeletionAction: "PURGE_PARTITION",
|
|
pressureAction: "RETAIN",
|
|
unavailableFallback: "EXPORT_REQUIRED",
|
|
});
|
|
|
|
function beginInput(
|
|
transactionId: string,
|
|
objectId: string,
|
|
byteLength: number,
|
|
targetScope = scope,
|
|
): BeginOpfsJournalTransaction {
|
|
return {
|
|
transactionId,
|
|
mutation: "PUT",
|
|
scope: targetScope,
|
|
objectId,
|
|
expectedGeneration: null,
|
|
targetGeneration: 1,
|
|
targetByteLength: byteLength,
|
|
targetStoragePolicy: {
|
|
...policy,
|
|
namespace: targetScope.namespace,
|
|
},
|
|
startedAtEpochMs: 100,
|
|
};
|
|
}
|
|
|
|
function prepared(
|
|
input: BeginOpfsJournalTransaction,
|
|
): OpfsPreparedObject {
|
|
return {
|
|
physicalSchemaVersion: 1,
|
|
descriptor: {
|
|
objectId: input.objectId,
|
|
scope: input.scope,
|
|
generation: input.targetGeneration,
|
|
byteLength: input.targetByteLength,
|
|
mediaType: "application/octet-stream",
|
|
createdAtEpochMs: input.startedAtEpochMs,
|
|
integrity: {
|
|
algorithm: "SHA-256-TREE-V1",
|
|
rootDigestHex: "a".repeat(64),
|
|
chunkSizeBytes: 64 * 1024,
|
|
},
|
|
storagePolicy: input.targetStoragePolicy,
|
|
},
|
|
chunks:
|
|
input.targetByteLength === 0
|
|
? []
|
|
: [
|
|
{
|
|
sequence: 0,
|
|
byteLength: input.targetByteLength,
|
|
digestHex: "b".repeat(64),
|
|
},
|
|
],
|
|
};
|
|
}
|
|
|
|
function createHarness(factory = new MemoryIndexedDbFactory()) {
|
|
let tokenSequence = 0;
|
|
const journal = createIndexedDbOpfsJournal({
|
|
authorityToken,
|
|
factory: factory.factory,
|
|
createFencingToken: () =>
|
|
`fencing_${String(++tokenSequence).padStart(8, "0")}`,
|
|
});
|
|
return { factory, journal };
|
|
}
|
|
|
|
describe("IndexedDB OPFS journal", () => {
|
|
it("fences stale writers and atomically publishes object, budget and chunk refs", async () => {
|
|
const { factory, journal } = createHarness();
|
|
const input = beginInput(
|
|
"transaction_12345678",
|
|
"object_12345678",
|
|
3,
|
|
);
|
|
const begun = await journal.begin(input);
|
|
expect(begun).toMatchObject({ ok: true });
|
|
if (!begun.ok) throw new Error("begin failed");
|
|
|
|
expect(
|
|
await journal.markFilesReady(
|
|
input.transactionId,
|
|
"fencing_stale_1234",
|
|
prepared(input),
|
|
),
|
|
).toMatchObject({ ok: false, error: { code: "CONFLICT" } });
|
|
expect(
|
|
await journal.markFilesReady(
|
|
input.transactionId,
|
|
begun.value.fencingToken,
|
|
prepared(input),
|
|
),
|
|
).toMatchObject({ ok: true, value: { phase: "FILES_READY" } });
|
|
|
|
factory.failNextWriteCommit(
|
|
new DOMException("quota", "QuotaExceededError"),
|
|
);
|
|
expect(
|
|
await journal.commitPut(
|
|
input.transactionId,
|
|
begun.value.fencingToken,
|
|
),
|
|
).toMatchObject({ ok: false, error: { code: "QUOTA_EXCEEDED" } });
|
|
expect(
|
|
await journal.getCommittedObject(scope, input.objectId),
|
|
).toEqual({ ok: true, value: null });
|
|
expect(
|
|
await journal.isChunkReferenced(scope, "b".repeat(64)),
|
|
).toEqual({ ok: true, value: false });
|
|
|
|
expect(
|
|
await journal.commitPut(
|
|
input.transactionId,
|
|
begun.value.fencingToken,
|
|
),
|
|
).toMatchObject({ ok: true, value: { phase: "COMMITTED" } });
|
|
expect(
|
|
await journal.getCommittedObject(scope, input.objectId),
|
|
).toMatchObject({
|
|
ok: true,
|
|
value: { descriptor: { byteLength: 3, generation: 1 } },
|
|
});
|
|
expect(
|
|
await journal.isChunkReferenced(scope, "b".repeat(64)),
|
|
).toEqual({ ok: true, value: true });
|
|
expect(
|
|
await journal.isChunkReferenced(otherPartition, "b".repeat(64)),
|
|
).toEqual({ ok: true, value: false });
|
|
});
|
|
|
|
it("reserves the hard budget in the same transaction and releases it on rollback", async () => {
|
|
const { journal } = createHarness();
|
|
const first = await journal.begin(
|
|
beginInput("transaction_11111111", "object_11111111", 6),
|
|
);
|
|
expect(first).toMatchObject({ ok: true });
|
|
if (!first.ok) throw new Error("begin failed");
|
|
|
|
expect(
|
|
await journal.begin(
|
|
beginInput("transaction_22222222", "object_22222222", 5),
|
|
),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "LIMIT_EXCEEDED" },
|
|
});
|
|
expect(
|
|
await journal.begin(
|
|
beginInput(
|
|
"transaction_33333333",
|
|
"object_33333333",
|
|
10,
|
|
otherPartition,
|
|
),
|
|
),
|
|
).toMatchObject({ ok: true });
|
|
|
|
expect(
|
|
await journal.rollback(
|
|
"transaction_11111111",
|
|
first.value.fencingToken,
|
|
),
|
|
).toEqual({ ok: true, value: undefined });
|
|
expect(
|
|
await journal.begin(
|
|
beginInput("transaction_44444444", "object_44444444", 5),
|
|
),
|
|
).toMatchObject({ ok: true });
|
|
});
|
|
|
|
it("commits and completes a DELETE row without requiring a prepared object", async () => {
|
|
const { journal } = createHarness();
|
|
const putInput = beginInput(
|
|
"transaction_55555555",
|
|
"object_55555555",
|
|
2,
|
|
);
|
|
const put = await journal.begin(putInput);
|
|
if (!put.ok) throw new Error("put begin failed");
|
|
await journal.markFilesReady(
|
|
putInput.transactionId,
|
|
put.value.fencingToken,
|
|
prepared(putInput),
|
|
);
|
|
await journal.commitPut(
|
|
putInput.transactionId,
|
|
put.value.fencingToken,
|
|
);
|
|
await journal.complete(
|
|
putInput.transactionId,
|
|
put.value.fencingToken,
|
|
);
|
|
|
|
const deletion = await journal.begin({
|
|
transactionId: "transaction_66666666",
|
|
mutation: "DELETE",
|
|
scope,
|
|
objectId: putInput.objectId,
|
|
expectedGeneration: 1,
|
|
targetGeneration: 2,
|
|
targetByteLength: 0,
|
|
targetStoragePolicy: policy,
|
|
startedAtEpochMs: 200,
|
|
});
|
|
if (!deletion.ok) throw new Error("delete begin failed");
|
|
expect(
|
|
await journal.commitDelete(
|
|
"transaction_66666666",
|
|
deletion.value.fencingToken,
|
|
),
|
|
).toMatchObject({
|
|
ok: true,
|
|
value: { mutation: "DELETE", phase: "COMMITTED" },
|
|
});
|
|
expect(
|
|
await journal.complete(
|
|
"transaction_66666666",
|
|
deletion.value.fencingToken,
|
|
),
|
|
).toEqual({ ok: true, value: undefined });
|
|
expect(
|
|
await journal.getCommittedObject(scope, putInput.objectId),
|
|
).toEqual({ ok: true, value: null });
|
|
});
|
|
|
|
it("binds one opaque physical authority to one deterministic database", async () => {
|
|
expect(opfsJournalDatabaseName(authorityToken)).toBe(
|
|
`ca-frontend-opfs-metadata-v1:${authorityToken}`,
|
|
);
|
|
expect(() =>
|
|
createIndexedDbOpfsJournal({
|
|
authorityToken,
|
|
databaseName: "unrelated-database",
|
|
factory: new MemoryIndexedDbFactory().factory,
|
|
}),
|
|
).toThrow();
|
|
|
|
const { journal } = createHarness();
|
|
const binding = await journal.begin(
|
|
beginInput(
|
|
"transaction_binding_1234",
|
|
"object_binding_12345678",
|
|
1,
|
|
),
|
|
);
|
|
if (!binding.ok) throw new Error("scope binding failed");
|
|
await journal.rollback(
|
|
"transaction_binding_1234",
|
|
binding.value.fencingToken,
|
|
);
|
|
|
|
const remappedScope: OpfsStorageScope = {
|
|
...scope,
|
|
namespace: "different-namespace",
|
|
};
|
|
expect(
|
|
await journal.begin(
|
|
beginInput(
|
|
"transaction_remap_1234",
|
|
"object_remap_12345678",
|
|
1,
|
|
remappedScope,
|
|
),
|
|
),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "POLICY_REJECTED" },
|
|
});
|
|
const retokenedScope: OpfsStorageScope = {
|
|
...scope,
|
|
namespaceToken: "namespace_87654321",
|
|
};
|
|
expect(
|
|
await journal.begin(
|
|
beginInput(
|
|
"transaction_retoken_1234",
|
|
"object_retoken_12345678",
|
|
1,
|
|
retokenedScope,
|
|
),
|
|
),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "POLICY_REJECTED" },
|
|
});
|
|
expect(
|
|
await journal.begin({
|
|
...beginInput(
|
|
"transaction_policy_1234",
|
|
"object_policy_12345678",
|
|
1,
|
|
),
|
|
targetStoragePolicy: {
|
|
...policy,
|
|
owner: "different-owner",
|
|
},
|
|
}),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "POLICY_REJECTED" },
|
|
});
|
|
|
|
const foreignScope: OpfsStorageScope = {
|
|
...scope,
|
|
authorityToken: "authority_87654321",
|
|
};
|
|
expect(
|
|
await journal.getCommittedObject(
|
|
foreignScope,
|
|
"object_12345678",
|
|
),
|
|
).toMatchObject({ ok: false, error: { code: "INVALID_INPUT" } });
|
|
expect(
|
|
await journal.begin(
|
|
beginInput(
|
|
"transaction_77777777",
|
|
"object_77777777",
|
|
1,
|
|
foreignScope,
|
|
),
|
|
),
|
|
).toMatchObject({ ok: false, error: { code: "INVALID_INPUT" } });
|
|
});
|
|
|
|
it("maps synchronous open failures and blocked upgrades to Result failures", async () => {
|
|
const throwingFactory = {
|
|
open() {
|
|
throw new DOMException("denied", "SecurityError");
|
|
},
|
|
} as unknown as IDBFactory;
|
|
const throwingJournal = createIndexedDbOpfsJournal({
|
|
authorityToken,
|
|
factory: throwingFactory,
|
|
});
|
|
await expect(
|
|
throwingJournal.getCommittedObject(scope, "object_12345678"),
|
|
).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { code: "PERMISSION_DENIED" },
|
|
});
|
|
|
|
const factory = new MemoryIndexedDbFactory();
|
|
factory.blockNextOpen();
|
|
const callbacks: Array<() => void> = [];
|
|
const blockedJournal = createIndexedDbOpfsJournal({
|
|
authorityToken,
|
|
factory: factory.factory,
|
|
blockedTimeoutMs: 1,
|
|
scheduler: {
|
|
setTimeout(callback) {
|
|
callbacks.push(callback);
|
|
return callback;
|
|
},
|
|
clearTimeout() {},
|
|
},
|
|
});
|
|
const opening = blockedJournal.getCommittedObject(
|
|
scope,
|
|
"object_12345678",
|
|
);
|
|
await Promise.resolve();
|
|
callbacks.forEach((callback) => callback());
|
|
await expect(opening).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { code: "BLOCKED" },
|
|
});
|
|
});
|
|
|
|
it("reopens after versionchange and rejects corrupt persisted journal rows", async () => {
|
|
const { factory, journal } = createHarness();
|
|
expect(
|
|
await journal.getCommittedObject(scope, "object_12345678"),
|
|
).toEqual({ ok: true, value: null });
|
|
factory.triggerVersionChange(2);
|
|
expect(factory.isConnectionClosed()).toBe(true);
|
|
expect(
|
|
await journal.getCommittedObject(scope, "object_12345678"),
|
|
).toEqual({ ok: true, value: null });
|
|
|
|
factory.seed("opfs-journal", {
|
|
transactionId: "transaction_88888888",
|
|
startedAtEpochMs: 1,
|
|
corrupt: true,
|
|
});
|
|
expect(await journal.listIncomplete(10)).toMatchObject({
|
|
ok: false,
|
|
error: { code: "CORRUPT_DATA" },
|
|
});
|
|
});
|
|
});
|