From b45597504eeaba1f11a9920a1d1048c361ed4b8e Mon Sep 17 00:00:00 2001 From: Christian Klotz Date: Wed, 23 Sep 2026 23:21:15 +0200 Subject: [PATCH] feat(durable): export scoped storage conformance suite (#9977) * feat(durable): export scoped storage conformance suite * docs(durable): clarify storage conformance example --- packages/durable/README.md | 20 + packages/durable/package.json | 5 + packages/durable/src/testing/assertions.ts | 39 ++ packages/durable/src/testing/index.ts | 9 + packages/durable/src/testing/runner.ts | 24 + .../testing}/storage-conformance.ts | 643 ++++++++++-------- packages/durable/src/testing/types.ts | 22 + packages/durable/test/jsonl-storage.test.ts | 15 +- packages/durable/test/memory-storage.test.ts | 6 +- packages/durable/test/sqlite-storage.test.ts | 17 +- tsconfig.json | 1 + 11 files changed, 487 insertions(+), 314 deletions(-) create mode 100644 packages/durable/src/testing/assertions.ts create mode 100644 packages/durable/src/testing/index.ts create mode 100644 packages/durable/src/testing/runner.ts rename packages/durable/{test => src/testing}/storage-conformance.ts (67%) create mode 100644 packages/durable/src/testing/types.ts diff --git a/packages/durable/README.md b/packages/durable/README.md index 8a44feaf5..01308f30c 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -31,6 +31,26 @@ The portable SQLite core, minimal database facade, and ordered schema migrations The Node adapter uses WAL mode with `synchronous = NORMAL` and checkpoints the WAL on close. Acknowledged commits survive process crashes, but the newest commits may be lost after a power or host failure. One `SqliteStorage` owner must serialize writes to a database file; cross-process ID allocation is not supported. +## Storage conformance + +Storage adapters can register the runner-independent conformance cases through the testing entry. The convenience adapter accepts Vitest/Jest-compatible runner functions without importing either package: + +```ts +import { registerStorageConformance } from "@earendil-works/pi-durable/testing"; +import { describe, expect, it } from "vitest"; + +registerStorageConformance({ describe, expect, it }, "Some Custom Storage", async (use) => { + const storage = await openCustomStorage(); + try { + await use(storage); + } finally { + await closeStorage(storage); + } +}); +``` + +The provider must call and await `use` exactly once with isolated storage. Other runners can use `createStorageConformance` with their own `StorageConformanceAssertions` implementation. + ## Storage benchmarks From this package directory: diff --git a/packages/durable/package.json b/packages/durable/package.json index d3004dc60..7e439987f 100644 --- a/packages/durable/package.json +++ b/packages/durable/package.json @@ -46,6 +46,11 @@ "types": "./dist/storage/sqlite/node.d.ts", "import": "./dist/storage/sqlite/node.js" }, + "./testing": { + "source": "./src/testing/index.ts", + "types": "./dist/testing/index.d.ts", + "import": "./dist/testing/index.js" + }, "./package.json": "./package.json" }, "sideEffects": false, diff --git a/packages/durable/src/testing/assertions.ts b/packages/durable/src/testing/assertions.ts new file mode 100644 index 000000000..a450554b6 --- /dev/null +++ b/packages/durable/src/testing/assertions.ts @@ -0,0 +1,39 @@ +import type { StorageConformanceAssertions } from "./types.ts"; + +export type ExpectLike = (actual: unknown, message?: string) => unknown; + +type ExpectResult = { + toBe(expected: unknown): unknown; + toBeGreaterThan(expected: number): unknown; + toEqual(expected: unknown): unknown; + toMatchObject(expected: unknown): unknown; + toBeTruthy(): unknown; + readonly rejects: { + toThrow(expected?: string | RegExp): unknown; + }; +}; + +/** Adapts a Vitest/Jest-compatible `expect` function without importing either runner. */ +export function createExpectAssertions(expect: ExpectLike): StorageConformanceAssertions { + const result = (actual: unknown, message?: string): ExpectResult => expect(actual, message) as ExpectResult; + return { + ok(value, message) { + result(value, message).toBeTruthy(); + }, + strictEqual(actual, expected) { + result(actual).toBe(expected); + }, + deepEqual(actual, expected) { + result(actual).toEqual(expected); + }, + partialDeepEqual(actual, expected) { + result(actual).toMatchObject(expected); + }, + greaterThan(actual, expected) { + result(actual).toBeGreaterThan(expected); + }, + async rejects(operation, messageIncludes) { + await result(operation).rejects.toThrow(messageIncludes); + }, + }; +} diff --git a/packages/durable/src/testing/index.ts b/packages/durable/src/testing/index.ts new file mode 100644 index 000000000..98332bb2e --- /dev/null +++ b/packages/durable/src/testing/index.ts @@ -0,0 +1,9 @@ +export { createExpectAssertions, type ExpectLike } from "./assertions.ts"; +export { registerStorageConformance, type StorageConformanceRunner } from "./runner.ts"; +export { createStorageConformance } from "./storage-conformance.ts"; +export type { + StorageConformanceAssertions, + StorageConformanceCase, + StorageConformanceOptions, + StorageConformanceProvider, +} from "./types.ts"; diff --git a/packages/durable/src/testing/runner.ts b/packages/durable/src/testing/runner.ts new file mode 100644 index 000000000..8f63519a5 --- /dev/null +++ b/packages/durable/src/testing/runner.ts @@ -0,0 +1,24 @@ +import { createExpectAssertions, type ExpectLike } from "./assertions.ts"; +import { createStorageConformance } from "./storage-conformance.ts"; +import type { StorageConformanceProvider } from "./types.ts"; + +export interface StorageConformanceRunner { + readonly describe: (name: string, suite: () => void) => unknown; + readonly expect: ExpectLike; + readonly it: (name: string, test: () => Promise) => unknown; +} + +/** Registers the runner-independent cases with a Vitest/Jest-compatible test runner. */ +export function registerStorageConformance( + runner: StorageConformanceRunner, + name: string, + withStorage: StorageConformanceProvider, +): void { + const cases = createStorageConformance({ + assertions: createExpectAssertions(runner.expect), + withStorage, + }); + runner.describe(name, () => { + for (const testCase of cases) runner.it(testCase.name, testCase.run); + }); +} diff --git a/packages/durable/test/storage-conformance.ts b/packages/durable/src/testing/storage-conformance.ts similarity index 67% rename from packages/durable/test/storage-conformance.ts rename to packages/durable/src/testing/storage-conformance.ts index 8e84550f1..56d49b650 100644 --- a/packages/durable/test/storage-conformance.ts +++ b/packages/durable/src/testing/storage-conformance.ts @@ -1,7 +1,6 @@ import type { JsonValue } from "@earendil-works/chord"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; import type { Op } from "@earendil-works/chord/delta"; -import { describe, expect, it } from "vitest"; import { type DocumentCreate, type EntryRecord, @@ -12,7 +11,8 @@ import { type StorageWrite, type SubmissionRecord, type TaskRecord, -} from "../src/types.ts"; +} from "../types.ts"; +import type { StorageConformanceAssertions, StorageConformanceCase, StorageConformanceOptions } from "./types.ts"; const context = BACKGROUND_CONTEXT; type StoredTask = TaskRecord; @@ -40,79 +40,118 @@ function entry(id: Id, conversationId: Id, kind = "message", extra: Partial Storage | Promise): void { - describe(name, () => { - it("reserves ID 1 for the immutable root conversation", async () => { - const storage = await createStorage(); +type ConformanceTest = (storage: Storage) => Promise; + +type AssertionResult = { + toBe(expected: unknown): void; + toBeDefined(): void; + toBeGreaterThan(expected: number): void; + toBeUndefined(): void; + toEqual(expected: unknown): void; + toHaveLength(expected: number): void; + toMatchObject(expected: unknown): void; + readonly rejects: { toThrow(messageIncludes: string): Promise }; + readonly resolves: { toBe(expected: unknown): Promise }; +}; + +function assertionFacade(assertions: StorageConformanceAssertions) { + return (actual: unknown): AssertionResult => ({ + toBe: (expected) => assertions.strictEqual(actual, expected), + toBeDefined: () => assertions.ok(actual !== undefined, "Expected value to be defined"), + toBeGreaterThan: (expected) => assertions.greaterThan(actual as number, expected), + toBeUndefined: () => assertions.strictEqual(actual, undefined), + toEqual: (expected) => assertions.deepEqual(actual, expected), + toHaveLength: (expected) => assertions.strictEqual((actual as { readonly length: unknown }).length, expected), + toMatchObject: (expected) => assertions.partialDeepEqual(actual, expected), + rejects: { + toThrow: (messageIncludes) => assertions.rejects(Promise.resolve(actual), messageIncludes), + }, + resolves: { + toBe: async (expected) => assertions.strictEqual(await Promise.resolve(actual), expected), + }, + }); +} + +function createCase(options: StorageConformanceOptions, name: string, test: ConformanceTest): StorageConformanceCase { + return { name, run: () => options.withStorage(test) }; +} + +/** Creates runner-independent cases. `withStorage` must call and await its callback exactly once per case. */ +export function createStorageConformance(options: StorageConformanceOptions): readonly StorageConformanceCase[] { + const expect = assertionFacade(options.assertions); + return [ + createCase(options, "reserves ID 1 for the immutable root conversation", async (storage) => { expect(await storage.mintId()).toBe(2); await expect(createRoot(storage)).resolves.toBe(ROOT_CONVERSATION_ID); expect(await storage.conversation(ROOT_CONVERSATION_ID, context)).toEqual({ id: ROOT_CONVERSATION_ID }); await expect( storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context), ).rejects.toThrow(`ID ${ROOT_CONVERSATION_ID} already belongs to conversation`); - }); + }), - it("commits mixed table writes atomically and rolls all of them back on failure", async () => { - const storage = await createStorage(); - const rootId = await createRoot(storage); - const entryId = await storage.mintId(); - const taskId = await storage.mintId(); - const submissionId = await storage.mintId(); - const task = pendingTask(taskId, rootId); - const input: SubmissionRecord = { - id: submissionId, - conversationId: rootId, - requestId: "request-1", - type: "input", - status: "placed", - entry: entryId, - }; - const initialSeq = await storage.commit( - [ - { type: "entry", value: entry(entryId, rootId, "user", { data: { text: "hello" } }) }, - { type: "task", value: task }, - { type: "submission", value: input }, - ], - context, - ); - - expect(await storage.entry(entryId, context)).toEqual({ - entry: entry(entryId, rootId, "user", { data: { text: "hello" } }), - commitSeq: initialSeq, - }); - expect(await storage.task(taskId, context)).toEqual(task); - expect(await storage.submission(submissionId, context)).toEqual(input); - - const transientEntryId = await storage.mintId(); - const runningTask: StoredTask = { - ...task, - state: { status: "running", checkpoint: { phase: "effect" } }, - }; - const doneInput: SubmissionRecord = { ...input, status: "done", answer: transientEntryId }; - await expect( - storage.commit( + createCase( + options, + "commits mixed table writes atomically and rolls all of them back on failure", + async (storage) => { + const rootId = await createRoot(storage); + const entryId = await storage.mintId(); + const taskId = await storage.mintId(); + const submissionId = await storage.mintId(); + const task = pendingTask(taskId, rootId); + const input: SubmissionRecord = { + id: submissionId, + conversationId: rootId, + requestId: "request-1", + type: "input", + status: "placed", + entry: entryId, + }; + const initialSeq = await storage.commit( [ - { type: "task", value: runningTask }, - { type: "submission", value: doneInput }, - { type: "entry", value: entry(transientEntryId, rootId, "assistant") }, - { type: "conversation", value: { id: rootId } }, + { type: "entry", value: entry(entryId, rootId, "user", { data: { text: "hello" } }) }, + { type: "task", value: task }, + { type: "submission", value: input }, ], context, - ), - ).rejects.toThrow(`ID ${rootId} already belongs to conversation`); + ); - expect(await storage.task(taskId, context)).toEqual(task); - expect(await storage.submission(submissionId, context)).toEqual(input); - expect(await storage.entry(transientEntryId, context)).toBeUndefined(); - const afterRollbackSeq = await storage.commit( - [{ type: "entry", value: entry(await storage.mintId(), rootId, "after-rollback") }], - context, - ); - expect(afterRollbackSeq).toBeGreaterThan(initialSeq); - }); + expect(await storage.entry(entryId, context)).toEqual({ + entry: entry(entryId, rootId, "user", { data: { text: "hello" } }), + commitSeq: initialSeq, + }); + expect(await storage.task(taskId, context)).toEqual(task); + expect(await storage.submission(submissionId, context)).toEqual(input); - it("detaches retained writes and every returned record", async () => { - const storage = await createStorage(); + const transientEntryId = await storage.mintId(); + const runningTask: StoredTask = { + ...task, + state: { status: "running", checkpoint: { phase: "effect" } }, + }; + const doneInput: SubmissionRecord = { ...input, status: "done", answer: transientEntryId }; + await expect( + storage.commit( + [ + { type: "task", value: runningTask }, + { type: "submission", value: doneInput }, + { type: "entry", value: entry(transientEntryId, rootId, "assistant") }, + { type: "conversation", value: { id: rootId } }, + ], + context, + ), + ).rejects.toThrow(`ID ${rootId} already belongs to conversation`); + + expect(await storage.task(taskId, context)).toEqual(task); + expect(await storage.submission(submissionId, context)).toEqual(input); + expect(await storage.entry(transientEntryId, context)).toBeUndefined(); + const afterRollbackSeq = await storage.commit( + [{ type: "entry", value: entry(await storage.mintId(), rootId, "after-rollback") }], + context, + ); + expect(afterRollbackSeq).toBeGreaterThan(initialSeq); + }, + ), + + createCase(options, "detaches retained writes and every returned record", async (storage) => { const rootId = await createRoot(storage); const entryId = await storage.mintId(); const taskId = await storage.mintId(); @@ -167,10 +206,9 @@ export function registerStorageConformance(name: string, createStorage: () => St checkpoint: { phase: "ready", nested: { count: 1 } }, }); expect((await storage.submission(submissionId, context))?.detail).toEqual({ codes: ["initial"] }); - }); + }), - it("detaches prototype-like JSON keys without changing object prototypes", async () => { - const storage = await createStorage(); + createCase(options, "detaches prototype-like JSON keys without changing object prototypes", async (storage) => { const rootId = await createRoot(storage); const entryId = await storage.mintId(); const data = JSON.parse( @@ -193,10 +231,9 @@ export function registerStorageConformance(name: string, createStorage: () => St expect(Reflect.get(secondRead, "__proto__")).toEqual({ polluted: false }); expect(Reflect.get(secondRead, "constructor")).toEqual({ label: "stored" }); expect(Reflect.get(secondRead, "toString")).toBe("value"); - }); + }), - it("indexes entries committed out of ID order", async () => { - const storage = await createStorage(); + createCase(options, "indexes entries committed out of ID order", async (storage) => { const rootId = await createRoot(storage); await storage.commit( [ @@ -211,10 +248,9 @@ export function registerStorageConformance(name: string, createStorage: () => St (await storage.scanEntries({ conversationId: rootId }, undefined, 10, context)).items.map(({ id }) => id), ).toEqual([30, 20, 10]); expect((await storage.findLatestHeadMarker(rootId, undefined, context))?.id).toBe(20); - }); + }), - it("continues an entry cursor below its last item after a newer commit", async () => { - const storage = await createStorage(); + createCase(options, "continues an entry cursor below its last item after a newer commit", async (storage) => { const rootId = await createRoot(storage); const oldestId = await storage.mintId(); const middleId = await storage.mintId(); @@ -235,10 +271,9 @@ export function registerStorageConformance(name: string, createStorage: () => St const second = await storage.scanEntries({ conversationId: rootId }, first.next, 2, context); expect(second.items.map(({ id }) => id)).toEqual([oldestId]); expect(second.next).toBeUndefined(); - }); + }), - it("paginates conversations by opaque cursor in ascending ID order", async () => { - const storage = await createStorage(); + createCase(options, "paginates conversations by opaque cursor in ascending ID order", async (storage) => { const rootId = await createRoot(storage); const secondId = await storage.mintId(); const thirdId = await storage.mintId(); @@ -257,10 +292,9 @@ export function registerStorageConformance(name: string, createStorage: () => St const second = await storage.scanConversations(roundTrippedCursor, 2, context); expect(second.items.map(({ id }) => id)).toEqual([thirdId]); expect(second.next).toBeUndefined(); - }); + }), - it("scans deep fork history newest-first through every ancestor cap", async () => { - const storage = await createStorage(); + createCase(options, "scans deep fork history newest-first through every ancestor cap", async (storage) => { const rootId = await createRoot(storage); const rootFirst = await storage.mintId(); const rootForkPoint = await storage.mintId(); @@ -381,10 +415,9 @@ export function registerStorageConformance(name: string, createStorage: () => St await expect(storage.scanEntries({ conversationId: 999_999 }, undefined, 10, context)).rejects.toThrow( "Unknown conversation", ); - }); + }), - it("replaces complete task records and pages filtered task scans", async () => { - const storage = await createStorage(); + createCase(options, "replaces complete task records and pages filtered task scans", async (storage) => { const rootId = await createRoot(storage); const firstId = await storage.mintId(); const secondId = await storage.mintId(); @@ -434,56 +467,58 @@ export function registerStorageConformance(name: string, createStorage: () => St expect( (await storage.scanTasks({ background: true }, undefined, 10, context)).items.map(({ id }) => id), ).toEqual([secondId]); - }); + }), - it("indexes request IDs per conversation and replaces complete submission records", async () => { - const storage = await createStorage(); - const rootId = await createRoot(storage); - const secondConversationId = await storage.mintId(); - await storage.commit([{ type: "conversation", value: { id: secondConversationId } }], context); - const firstId = await storage.mintId(); - const secondId = await storage.mintId(); - const otherConversationId = await storage.mintId(); - const first: SubmissionRecord = { - id: firstId, - conversationId: rootId, - requestId: "same", - type: "input", - status: "queued", - }; - const second: SubmissionRecord = { - id: secondId, - conversationId: rootId, - requestId: "other", - type: "input", - status: "queued", - }; - const otherConversation: SubmissionRecord = { - id: otherConversationId, - conversationId: secondConversationId, - requestId: "same", - type: "input", - status: "queued", - }; - await storage.commit( - [ - { type: "submission", value: first }, - { type: "submission", value: second }, - { type: "submission", value: otherConversation }, - ], - context, - ); - expect(await storage.submissionByRequest(rootId, "same", context)).toEqual(first); - expect(await storage.submissionByRequest(secondConversationId, "same", context)).toEqual(otherConversation); + createCase( + options, + "indexes request IDs per conversation and replaces complete submission records", + async (storage) => { + const rootId = await createRoot(storage); + const secondConversationId = await storage.mintId(); + await storage.commit([{ type: "conversation", value: { id: secondConversationId } }], context); + const firstId = await storage.mintId(); + const secondId = await storage.mintId(); + const otherConversationId = await storage.mintId(); + const first: SubmissionRecord = { + id: firstId, + conversationId: rootId, + requestId: "same", + type: "input", + status: "queued", + }; + const second: SubmissionRecord = { + id: secondId, + conversationId: rootId, + requestId: "other", + type: "input", + status: "queued", + }; + const otherConversation: SubmissionRecord = { + id: otherConversationId, + conversationId: secondConversationId, + requestId: "same", + type: "input", + status: "queued", + }; + await storage.commit( + [ + { type: "submission", value: first }, + { type: "submission", value: second }, + { type: "submission", value: otherConversation }, + ], + context, + ); + expect(await storage.submissionByRequest(rootId, "same", context)).toEqual(first); + expect(await storage.submissionByRequest(secondConversationId, "same", context)).toEqual(otherConversation); - const placedSecond: SubmissionRecord = { ...second, status: "placed", entry: await storage.mintId() }; - await storage.commit([{ type: "submission", value: placedSecond }], context); - expect(await storage.submission(secondId, context)).toEqual(placedSecond); - expect(await storage.submissionByRequest(rootId, "other", context)).toEqual(placedSecond); - }); + const placedSecond: SubmissionRecord = { ...second, status: "placed", entry: await storage.mintId() }; + await storage.commit([{ type: "submission", value: placedSecond }], context); + expect(await storage.submission(secondId, context)).toEqual(placedSecond); + expect(await storage.submissionByRequest(rootId, "other", context)).toEqual(placedSecond); + }, + ), - it("stores passive write submissions without input-only lifecycle states", async () => { - const storage = await createStorage(); + createCase(options, "stores passive write submissions without input-only lifecycle states", async (storage) => { const rootId = await createRoot(storage); const doneId = await storage.mintId(); const failedId = await storage.mintId(); @@ -527,10 +562,9 @@ export function registerStorageConformance(name: string, createStorage: () => St expect(await storage.submissionByRequest(rootId, "passive-done", context)).toEqual(done); expect(await storage.submission(failedId, context)).toEqual(unanswered); expect(await storage.submissionByRequest(rootId, "passive-failed", context)).toEqual(unanswered); - }); + }), - it("reconstructs rewindable documents and preserves half-open incarnations", async () => { - const storage = await createStorage(); + createCase(options, "reconstructs rewindable documents and preserves half-open incarnations", async (storage) => { const rootId = await createRoot(storage); const firstId = await storage.mintId(); const firstRecord = { @@ -542,7 +576,13 @@ export function registerStorageConformance(name: string, createStorage: () => St } satisfies DocumentCreate; const initial: JsonObject = { items: ["a"], nested: { count: 1 } }; const createdAt = await storage.commit( - [{ type: "document.create", record: firstRecord, content: { kind: "base", version: 1, value: initial } }], + [ + { + type: "document.create", + record: firstRecord, + content: { kind: "base", version: 1, value: initial }, + }, + ], context, ); const appended = ["b"]; @@ -641,45 +681,49 @@ export function registerStorageConformance(name: string, createStorage: () => St ).toEqual([secondId]); expect(await storage.document(firstId, retiredAt, context)).toBeUndefined(); expect((await storage.document(secondId, "current", context))?.value).toEqual({ items: ["new"] }); - }); + }), - it("uses bases for version transitions and rejects historical reads of current-only documents", async () => { - const storage = await createStorage(); - await createRoot(storage); - const id = await storage.mintId(); - const record = { - id, - kind: "session.settings", - scope: { kind: "session" }, - } satisfies DocumentCreate; - await storage.commit( - [{ type: "document.create", record, content: { kind: "base", version: 1, value: { count: 1 } } }], - context, - ); - await storage.commit( - [{ type: "document.change", id, content: { kind: "delta", version: 1, ops: [["s", ["count"], 2]] } }], - context, - ); - const migratedAt = await storage.commit( - [{ type: "document.change", id, content: { kind: "base", version: 2, value: { count: 3 } } }], - context, - ); - expect(await storage.document(id, "current", context)).toMatchObject({ version: 2, value: { count: 3 } }); - await expect(storage.document(id, migratedAt, context)).rejects.toThrow("does not retain historical content"); - - await expect( - storage.commit( - [{ type: "document.change", id, content: { kind: "delta", version: 1, ops: [["s", ["count"], 4]] } }], + createCase( + options, + "uses bases for version transitions and rejects historical reads of current-only documents", + async (storage) => { + await createRoot(storage); + const id = await storage.mintId(); + const record = { + id, + kind: "session.settings", + scope: { kind: "session" }, + } satisfies DocumentCreate; + await storage.commit( + [{ type: "document.create", record, content: { kind: "base", version: 1, value: { count: 1 } } }], context, - ), - ).rejects.toThrow("version transition requires a base"); - expect((await storage.document(id, "current", context))?.value).toEqual({ count: 3 }); - await storage.commit([{ type: "document.retire", id }], context); - expect(await storage.document(id, "current", context)).toBeUndefined(); - }); + ); + await storage.commit( + [{ type: "document.change", id, content: { kind: "delta", version: 1, ops: [["s", ["count"], 2]] } }], + context, + ); + const migratedAt = await storage.commit( + [{ type: "document.change", id, content: { kind: "base", version: 2, value: { count: 3 } } }], + context, + ); + expect(await storage.document(id, "current", context)).toMatchObject({ version: 2, value: { count: 3 } }); + await expect(storage.document(id, migratedAt, context)).rejects.toThrow( + "does not retain historical content", + ); - it("indexes logical addresses and exact-scope scans independently", async () => { - const storage = await createStorage(); + await expect( + storage.commit( + [{ type: "document.change", id, content: { kind: "delta", version: 1, ops: [["s", ["count"], 4]] } }], + context, + ), + ).rejects.toThrow("version transition requires a base"); + expect((await storage.document(id, "current", context))?.value).toEqual({ count: 3 }); + await storage.commit([{ type: "document.retire", id }], context); + expect(await storage.document(id, "current", context)).toBeUndefined(); + }, + ), + + createCase(options, "indexes logical addresses and exact-scope scans independently", async (storage) => { const rootId = await createRoot(storage); const firstId = await storage.mintId(); const secondId = await storage.mintId(); @@ -816,147 +860,152 @@ export function registerStorageConformance(name: string, createStorage: () => St await expect(storage.document(taskSingletonId, createdAt, context)).rejects.toThrow( "does not retain historical content", ); - }); + }), - it("keeps document lifecycle failures atomic and gives create-plus-retire an empty lifetime", async () => { - const storage = await createStorage(); - const rootId = await createRoot(storage); - const firstId = await storage.mintId(); - const secondId = await storage.mintId(); - const record = { - id: firstId, - kind: "singleton", - scope: { kind: "session" }, - } satisfies DocumentCreate; - await storage.commit( - [{ type: "document.create", record, content: { kind: "base", version: 1, value: { value: 1 } } }], - context, - ); - await expect( - storage.commit( + createCase( + options, + "keeps document lifecycle failures atomic and gives create-plus-retire an empty lifetime", + async (storage) => { + const rootId = await createRoot(storage); + const firstId = await storage.mintId(); + const secondId = await storage.mintId(); + const record = { + id: firstId, + kind: "singleton", + scope: { kind: "session" }, + } satisfies DocumentCreate; + await storage.commit( + [{ type: "document.create", record, content: { kind: "base", version: 1, value: { value: 1 } } }], + context, + ); + await expect( + storage.commit( + [ + { + type: "document.create", + record: { ...record, id: secondId }, + content: { kind: "base", version: 1, value: { value: 2 } }, + }, + { type: "document.change", id: firstId, content: { kind: "delta", version: 1, ops: [] } }, + ], + context, + ), + ).rejects.toThrow("already has a current incarnation"); + expect((await storage.document(firstId, "current", context))?.value).toEqual({ value: 1 }); + expect(await storage.document(secondId, "current", context)).toBeUndefined(); + + const emptyId = await storage.mintId(); + const emptyAt = await storage.commit( [ { type: "document.create", - record: { ...record, id: secondId }, - content: { kind: "base", version: 1, value: { value: 2 } }, + record: { + id: emptyId, + kind: record.kind, + key: "empty", + scope: { kind: "conversation", conversationId: rootId }, + history: "rewindable", + fork: "initial", + }, + content: { kind: "base", version: 1, value: {} }, }, - { type: "document.change", id: firstId, content: { kind: "delta", version: 1, ops: [] } }, + { type: "document.retire", id: emptyId }, ], context, - ), - ).rejects.toThrow("already has a current incarnation"); - expect((await storage.document(firstId, "current", context))?.value).toEqual({ value: 1 }); - expect(await storage.document(secondId, "current", context)).toBeUndefined(); - - const emptyId = await storage.mintId(); - const emptyAt = await storage.commit( - [ - { - type: "document.create", - record: { - id: emptyId, + ); + expect(await storage.document(emptyId, "current", context)).toBeUndefined(); + expect(await storage.document(emptyId, emptyAt, context)).toBeUndefined(); + expect( + await storage.findDocument( + { kind: record.kind, - key: "empty", scope: { kind: "conversation", conversationId: rootId }, - history: "rewindable", - fork: "initial", + key: "empty", }, - content: { kind: "base", version: 1, value: {} }, - }, - { type: "document.retire", id: emptyId }, - ], - context, - ); - expect(await storage.document(emptyId, "current", context)).toBeUndefined(); - expect(await storage.document(emptyId, emptyAt, context)).toBeUndefined(); - expect( - await storage.findDocument( - { - kind: record.kind, - scope: { kind: "conversation", conversationId: rootId }, - key: "empty", - }, - emptyAt, + emptyAt, + context, + ), + ).toBeUndefined(); + }, + ), + + createCase( + options, + "rolls back record tables and secondary indexes when a document command fails", + async (storage) => { + const rootId = await createRoot(storage); + const taskId = await storage.mintId(); + const submissionId = await storage.mintId(); + const documentId = await storage.mintId(); + const task = pendingTask(taskId, rootId); + const submission: SubmissionRecord = { + id: submissionId, + conversationId: rootId, + requestId: "atomic", + type: "input", + status: "queued", + }; + const record = { + id: documentId, + kind: "atomic", + scope: { kind: "session" }, + } satisfies DocumentCreate; + const baselineSeq = await storage.commit( + [ + { type: "task", value: task }, + { type: "submission", value: submission }, + { type: "document.create", record, content: { kind: "base", version: 1, value: { count: 1 } } }, + ], context, - ), - ).toBeUndefined(); - }); + ); - it("rolls back record tables and secondary indexes when a document command fails", async () => { - const storage = await createStorage(); - const rootId = await createRoot(storage); - const taskId = await storage.mintId(); - const submissionId = await storage.mintId(); - const documentId = await storage.mintId(); - const task = pendingTask(taskId, rootId); - const submission: SubmissionRecord = { - id: submissionId, - conversationId: rootId, - requestId: "atomic", - type: "input", - status: "queued", - }; - const record = { - id: documentId, - kind: "atomic", - scope: { kind: "session" }, - } satisfies DocumentCreate; - const baselineSeq = await storage.commit( - [ - { type: "task", value: task }, - { type: "submission", value: submission }, - { type: "document.create", record, content: { kind: "base", version: 1, value: { count: 1 } } }, - ], - context, - ); + const entryId = await storage.mintId(); + const conflictingDocumentId = await storage.mintId(); + await expect( + storage.commit( + [ + { + type: "task", + value: { ...task, state: { status: "running", checkpoint: { phase: "effect" } } }, + }, + { + type: "submission", + value: { ...submission, status: "unanswered", reason: "failed" }, + }, + { type: "entry", value: entry(entryId, rootId, "transient") }, + { + type: "document.create", + record: { ...record, id: conflictingDocumentId }, + content: { kind: "base", version: 1, value: { count: 2 } }, + }, + ], + context, + ), + ).rejects.toThrow("already has a current incarnation"); - const entryId = await storage.mintId(); - const conflictingDocumentId = await storage.mintId(); - await expect( - storage.commit( + expect(await storage.task(taskId, context)).toEqual(task); + expect((await storage.scanTasks({ status: "pending" }, undefined, 10, context)).items).toEqual([task]); + expect(await storage.submissionByRequest(rootId, "atomic", context)).toEqual(submission); + expect(await storage.entry(entryId, context)).toBeUndefined(); + expect(await storage.document(conflictingDocumentId, "current", context)).toBeUndefined(); + expect( + (await storage.findDocument({ kind: record.kind, scope: record.scope }, "current", context))?.id, + ).toBe(documentId); + const afterRollbackSeq = await storage.commit( [ { - type: "task", - value: { ...task, state: { status: "running", checkpoint: { phase: "effect" } } }, - }, - { - type: "submission", - value: { ...submission, status: "unanswered", reason: "failed" }, - }, - { type: "entry", value: entry(entryId, rootId, "transient") }, - { - type: "document.create", - record: { ...record, id: conflictingDocumentId }, - content: { kind: "base", version: 1, value: { count: 2 } }, + type: "document.change", + id: documentId, + content: { kind: "delta", version: 1, ops: [["s", ["count"], 3]] }, }, ], context, - ), - ).rejects.toThrow("already has a current incarnation"); + ); + expect(afterRollbackSeq).toBeGreaterThan(baselineSeq); + }, + ), - expect(await storage.task(taskId, context)).toEqual(task); - expect((await storage.scanTasks({ status: "pending" }, undefined, 10, context)).items).toEqual([task]); - expect(await storage.submissionByRequest(rootId, "atomic", context)).toEqual(submission); - expect(await storage.entry(entryId, context)).toBeUndefined(); - expect(await storage.document(conflictingDocumentId, "current", context)).toBeUndefined(); - expect((await storage.findDocument({ kind: record.kind, scope: record.scope }, "current", context))?.id).toBe( - documentId, - ); - const afterRollbackSeq = await storage.commit( - [ - { - type: "document.change", - id: documentId, - content: { kind: "delta", version: 1, ops: [["s", ["count"], 3]] }, - }, - ], - context, - ); - expect(afterRollbackSeq).toBeGreaterThan(baselineSeq); - }); - - it("keeps indexed string identities lossless", async () => { - const storage = await createStorage(); + createCase(options, "keeps indexed string identities lossless", async (storage) => { const rootId = await createRoot(storage); const first = "\ud800"; const second = "\ud801"; @@ -1056,10 +1105,9 @@ export function registerStorageConformance(name: string, createStorage: () => St ) ).items.map(({ id }) => id), ).toEqual([firstKindDocumentId]); - }); + }), - it("keeps one global record ID namespace and rejects exhausted ID minting", async () => { - const storage = await createStorage(); + createCase(options, "keeps one global record ID namespace and rejects exhausted ID minting", async (storage) => { const rootId = await createRoot(storage); const explicitEntryId = 100; await storage.commit([{ type: "entry", value: entry(explicitEntryId, rootId) }], context); @@ -1071,15 +1119,14 @@ export function registerStorageConformance(name: string, createStorage: () => St await storage.commit([{ type: "entry", value: entry(Number.MAX_SAFE_INTEGER, rootId, "last-id") }], context); await expect(storage.mintId()).rejects.toThrow("ID space is exhausted"); await expect(storage.mintId()).rejects.toThrow("ID space is exhausted"); - }); + }), - it("rejects every operation after close", async () => { - const storage = await createStorage(); + createCase(options, "rejects every operation after close", async (storage) => { await createRoot(storage); await storage.close(context); await expect(storage.conversation(ROOT_CONVERSATION_ID, context)).rejects.toThrow("closed"); await expect(storage.commit([] satisfies StorageWrite[], context)).rejects.toThrow("closed"); await expect(storage.mintId()).rejects.toThrow("closed"); - }); - }); + }), + ]; } diff --git a/packages/durable/src/testing/types.ts b/packages/durable/src/testing/types.ts new file mode 100644 index 000000000..a23ffa0db --- /dev/null +++ b/packages/durable/src/testing/types.ts @@ -0,0 +1,22 @@ +import type { Storage } from "../types.ts"; + +export interface StorageConformanceAssertions { + ok(value: unknown, message?: string): void; + strictEqual(actual: unknown, expected: unknown): void; + deepEqual(actual: unknown, expected: unknown): void; + partialDeepEqual(actual: unknown, expected: unknown): void; + greaterThan(actual: number, expected: number): void; + rejects(operation: Promise, messageIncludes: string): Promise; +} + +export type StorageConformanceProvider = (use: (storage: Storage) => Promise) => Promise; + +export interface StorageConformanceOptions { + readonly assertions: StorageConformanceAssertions; + readonly withStorage: StorageConformanceProvider; +} + +export interface StorageConformanceCase { + readonly name: string; + run(): Promise; +} diff --git a/packages/durable/test/jsonl-storage.test.ts b/packages/durable/test/jsonl-storage.test.ts index 0de533b4d..f420311cd 100644 --- a/packages/durable/test/jsonl-storage.test.ts +++ b/packages/durable/test/jsonl-storage.test.ts @@ -3,14 +3,14 @@ import { tmpdir } from "node:os"; import { basename, join } from "node:path"; import type { Context, JsonValue } from "@earendil-works/chord"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; -import { afterEach, describe, expect, it, onTestFinished } from "vitest"; +import { registerStorageConformance } from "@earendil-works/pi-durable/testing"; +import { afterEach, describe, expect, it } from "vitest"; import { err, FileError, type FileSystem, type Result } from "../src/env/index.ts"; import { NodeExecutionEnv } from "../src/env/node.ts"; import { JsonlStorage } from "../src/storage/jsonl/index.ts"; import { openNodeJsonlStorage } from "../src/storage/jsonl/node.ts"; import type { DocumentCreate, Seq, Storage, StorageWrite, TaskRecord } from "../src/types.ts"; import { ROOT_CONVERSATION_ID } from "../src/types.ts"; -import { registerStorageConformance } from "./storage-conformance.ts"; const context = BACKGROUND_CONTEXT; type StoredTask = TaskRecord; @@ -92,14 +92,17 @@ class ReopeningStorage implements Storage { } } -registerStorageConformance("Pico JsonlStorage conformance", createStorage); +registerStorageConformance({ describe, expect, it }, "JsonlStorage", async (use) => use(await createStorage())); -registerStorageConformance("Pico JsonlStorage conformance across reopen", async () => { +registerStorageConformance({ describe, expect, it }, "JsonlStorage across reopen", async (use) => { const directory = await tempDirectory("pi-durable-jsonl-conformance-"); const current = await JsonlStorage.open(directory, new NodeExecutionEnv({ cwd: directory }), context); const storage = new ReopeningStorage(current, directory); - onTestFinished(() => storage.close(context)); - return storage; + try { + await use(storage); + } finally { + await storage.close(context); + } }); function pendingTask(id: number, phase = "ready"): StoredTask { diff --git a/packages/durable/test/memory-storage.test.ts b/packages/durable/test/memory-storage.test.ts index 633109e84..ee9622316 100644 --- a/packages/durable/test/memory-storage.test.ts +++ b/packages/durable/test/memory-storage.test.ts @@ -1,10 +1,10 @@ import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; -import { expect, it } from "vitest"; +import { registerStorageConformance } from "@earendil-works/pi-durable/testing"; +import { describe, expect, it } from "vitest"; import { MemoryStorage } from "../src/storage/memory.ts"; import { ROOT_CONVERSATION_ID } from "../src/types.ts"; -import { registerStorageConformance } from "./storage-conformance.ts"; -registerStorageConformance("Pico MemoryStorage", () => new MemoryStorage()); +registerStorageConformance({ describe, expect, it }, "MemoryStorage", (use) => use(new MemoryStorage())); it("does not expose retained state through a prepared commit", async () => { const storage = new MemoryStorage(); diff --git a/packages/durable/test/sqlite-storage.test.ts b/packages/durable/test/sqlite-storage.test.ts index 562572cbb..baaebc155 100644 --- a/packages/durable/test/sqlite-storage.test.ts +++ b/packages/durable/test/sqlite-storage.test.ts @@ -4,12 +4,12 @@ import { join } from "node:path"; import { DatabaseSync } from "node:sqlite"; import type { JsonValue } from "@earendil-works/chord"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; -import { afterEach, describe, expect, it, onTestFinished } from "vitest"; +import { registerStorageConformance } from "@earendil-works/pi-durable/testing"; +import { afterEach, describe, expect, it } from "vitest"; import type { SqliteStorage } from "../src/storage/sqlite/index.ts"; import { type NodeSqliteStorageOptions, openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; import type { DocumentCreate, EntryRecord, Seq, Storage, StorageWrite, TaskRecord } from "../src/types.ts"; import { ROOT_CONVERSATION_ID } from "../src/types.ts"; -import { registerStorageConformance } from "./storage-conformance.ts"; const context = BACKGROUND_CONTEXT; const openStorages = new Set(); @@ -81,19 +81,22 @@ class ReopeningStorage implements Storage { } } -registerStorageConformance("Pico SqliteStorage conformance", async () => (await createSqliteStorage()).storage); +registerStorageConformance({ describe, expect, it }, "SqliteStorage", async (use) => + use((await createSqliteStorage()).storage), +); -registerStorageConformance("Pico SqliteStorage conformance across reopen", async () => { +registerStorageConformance({ describe, expect, it }, "SqliteStorage across reopen", async (use) => { const directory = await mkdtemp(join(tmpdir(), "pi-durable-sqlite-conformance-")); const path = join(directory, "storage.sqlite"); const created = await openNodeSqliteStorage(path); await created.close(context); const storage = new ReopeningStorage(await openNodeSqliteStorage(path), path); - onTestFinished(async () => { + try { + await use(storage); + } finally { await storage.close(context); await rm(directory, { recursive: true, force: true }); - }); - return storage; + } }); function entry(id: number, conversationId: number, data?: JsonValue): EntryRecord { diff --git a/tsconfig.json b/tsconfig.json index da12bbfd7..31d343d92 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -18,6 +18,7 @@ "@earendil-works/pi-ai/*": ["./packages/ai/src/*.ts", "./packages/ai/src/providers/*.ts"], "@earendil-works/pi-ai/dist/*": ["./packages/ai/src/*"], "@earendil-works/pi-durable": ["./packages/durable/src/index.ts"], + "@earendil-works/pi-durable/testing": ["./packages/durable/src/testing/index.ts"], "@earendil-works/pi-durable/*": ["./packages/durable/src/*"], "@earendil-works/pi-agent-core": ["./packages/agent/src/index.ts"], "@earendil-works/pi-agent-core/harness/session/testing": [