From 19a0361be89bf78ccf9bbaed9a496d6484759f67 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Thu, 24 Sep 2026 22:23:36 +0200 Subject: [PATCH] feat(durable): add transactional sessions and documents Add typed document definitions, atomic Session transactions, complete commit publication, task/document lifecycle handling, and paginated Tx scans. Share Chord's strict JSON copier across draft placements and durable ownership boundaries. --- packages/chord/src/delta/tracker.ts | 52 +- packages/chord/src/index.ts | 2 +- packages/chord/src/json.ts | 80 ++- packages/chord/test/json.test.ts | 53 +- packages/durable/CHANGELOG.md | 8 + packages/durable/docs/pico-v5-handoff.md | 8 +- packages/durable/docs/pico-v5.md | 15 +- packages/durable/src/documents.ts | 177 +++++ packages/durable/src/index.ts | 25 + packages/durable/src/session/publications.ts | 16 + packages/durable/src/session/session.ts | 239 +++++++ packages/durable/src/session/transaction.ts | 677 ++++++++++++++++++ packages/durable/src/storage/jsonl/storage.ts | 22 +- packages/durable/src/storage/memory.ts | 11 +- .../durable/src/storage/sqlite/storage.ts | 11 +- .../durable/src/testing/storage-benchmark.ts | 6 +- .../src/testing/storage-conformance.ts | 54 +- packages/durable/src/types.ts | 233 +++++- packages/durable/test/jsonl-storage.test.ts | 16 +- .../durable/test/session-definitions.test.ts | 111 +++ .../durable/test/session-documents.test.ts | 651 +++++++++++++++++ packages/durable/test/session-support.ts | 129 ++++ packages/durable/test/session-tables.test.ts | 328 +++++++++ packages/durable/test/sqlite-storage.test.ts | 16 +- 24 files changed, 2797 insertions(+), 143 deletions(-) create mode 100644 packages/durable/src/documents.ts create mode 100644 packages/durable/src/session/publications.ts create mode 100644 packages/durable/src/session/session.ts create mode 100644 packages/durable/src/session/transaction.ts create mode 100644 packages/durable/test/session-definitions.test.ts create mode 100644 packages/durable/test/session-documents.test.ts create mode 100644 packages/durable/test/session-support.ts create mode 100644 packages/durable/test/session-tables.test.ts diff --git a/packages/chord/src/delta/tracker.ts b/packages/chord/src/delta/tracker.ts index 0c4182071..1d2dd4627 100644 --- a/packages/chord/src/delta/tracker.ts +++ b/packages/chord/src/delta/tracker.ts @@ -1,3 +1,4 @@ +import { copyJson } from "../json.ts"; import type { JsonValue } from "../types.ts"; import { applyImmutable } from "./apply-immutable-batch.ts"; import { applyImmutableTrusted } from "./apply-immutable-trusted.ts"; @@ -1540,52 +1541,7 @@ function singletonPiece(piece: Piece, sourceIndex: number): Piece { function clonePlacement(value: unknown): Stored { const proxyNode = isContainer(value) ? (Reflect.get(value, NODE) as OverlayNode | undefined) : undefined; - return proxyNode === undefined ? cloneJson(value) : clonePlacementNode(proxyNode); -} - -function cloneJson(value: unknown, ancestors?: Set): Stored { - if (!isContainer(value)) return strictPrimitive(value); - const active = ancestors ?? new Set(); - if (active.has(value)) throw new TypeError("Draft placements cannot contain cycles"); - active.add(value); - try { - if (Array.isArray(value)) { - if (Object.getPrototypeOf(value) !== Array.prototype || Reflect.ownKeys(value).length !== value.length + 1) { - throw new TypeError("Draft placements must contain dense plain arrays"); - } - const result = new Array(value.length); - for (let index = 0; index < value.length; index++) { - const descriptor = Object.getOwnPropertyDescriptor(value, String(index)); - if (descriptor === undefined || !descriptor.enumerable || !("value" in descriptor)) { - throw new TypeError("Draft placements must contain enumerable indexed data properties"); - } - defineData(result, String(index), cloneJson(descriptor.value, active)); - } - return result; - } - const prototype = Object.getPrototypeOf(value); - if (prototype !== Object.prototype && prototype !== null) { - throw new TypeError("Draft placements must contain plain objects or arrays"); - } - const result = Object.create(prototype) as Record; - for (const key of Reflect.ownKeys(value)) { - if (typeof key === "symbol") throw new TypeError("Draft placements cannot contain symbol properties"); - const descriptor = Object.getOwnPropertyDescriptor(value, key); - if (descriptor === undefined || !descriptor.enumerable || !("value" in descriptor)) { - throw new TypeError("Draft placements must contain enumerable data properties"); - } - defineData(result, key, cloneJson(descriptor.value, active)); - } - return result; - } finally { - active.delete(value); - } -} - -function strictPrimitive(value: unknown): Primitive { - if (value === null || typeof value === "string" || typeof value === "boolean") return value; - if (typeof value === "number" && Number.isFinite(value)) return value; - throw new TypeError("Draft placements must contain strict JSON values"); + return proxyNode === undefined ? copyJson(value) : clonePlacementNode(proxyNode); } function cloneNode(node: OverlayNode): Container { @@ -1633,9 +1589,9 @@ function clonePlacementNode(node: OverlayNode): Container { } function clonePlacementStored(value: Stored, context: OverlayContext): Stored { - if (!isContainer(value)) return strictPrimitive(value); + if (!isContainer(value)) return copyJson(value); const node = context.rawNodes!.get(value); - return node === undefined ? cloneJson(value) : clonePlacementNode(node); + return node === undefined ? copyJson(value) : clonePlacementNode(node); } function defineData(target: object, key: PropertyKey, value: JsonValue): void { diff --git a/packages/chord/src/index.ts b/packages/chord/src/index.ts index fe156d642..4eca4457d 100644 --- a/packages/chord/src/index.ts +++ b/packages/chord/src/index.ts @@ -11,7 +11,7 @@ export { replicatedState, } from "./api.ts"; export type { Draft } from "./delta/index.ts"; -export { isJsonValue } from "./json.ts"; +export { type CopyJsonOptions, copyJson, isJsonValue } from "./json.ts"; export { isRemoteServiceErrorCode, REMOTE_SERVICE_ERROR_CODES, diff --git a/packages/chord/src/json.ts b/packages/chord/src/json.ts index e932aa6ab..4747dddd4 100644 --- a/packages/chord/src/json.ts +++ b/packages/chord/src/json.ts @@ -1,15 +1,85 @@ import type { JsonValue } from "./types.ts"; +const DATA_DESCRIPTOR: PropertyDescriptor = { + value: undefined, + writable: true, + enumerable: true, + configurable: true, +}; + +export type CopyJsonOptions = { + /** Omit undefined object properties while preserving strict array semantics. */ + readonly omitUndefinedProperties?: boolean; +}; + +/** Copy a value into an alias-free strict-JSON tree owned by the caller. */ +export function copyJson(value: unknown, options?: CopyJsonOptions): JsonValue { + return copy(value, undefined, options?.omitUndefinedProperties === true); +} + +function copy(value: unknown, ancestors: Set | undefined, omitUndefinedProperties: boolean): JsonValue { + if (value === null || typeof value === "string" || typeof value === "boolean") return value; + if (typeof value === "number") { + if (Number.isFinite(value)) return value; + throw new TypeError("Value contains a non-finite number and is not strict JSON"); + } + if (typeof value !== "object") + throw new TypeError(`Value contains a non-JSON ${typeof value}; expected strict JSON`); + const active = ancestors ?? new Set(); + if (active.has(value)) throw new TypeError("Value contains cycles and is not strict JSON"); + active.add(value); + try { + if (Array.isArray(value)) { + // Indices plus `length` only: extra, symbol, and missing keys all change the count. + if (Object.getPrototypeOf(value) !== Array.prototype || Reflect.ownKeys(value).length !== value.length + 1) { + throw new TypeError("Value must contain strict JSON dense plain arrays"); + } + const result = new Array(value.length); + for (let index = 0; index < value.length; index++) { + const descriptor = Object.getOwnPropertyDescriptor(value, index); + if (descriptor === undefined || !descriptor.enumerable || !("value" in descriptor)) { + throw new TypeError("Value must contain strict JSON enumerable indexed data properties"); + } + defineData(result, String(index), copy(descriptor.value, active, omitUndefinedProperties)); + } + return result; + } + const prototype = Object.getPrototypeOf(value); + if (prototype !== Object.prototype && prototype !== null) { + throw new TypeError("Value must contain strict JSON plain objects or arrays"); + } + const result = Object.create(prototype) as Record; + for (const key of Reflect.ownKeys(value)) { + if (typeof key === "symbol") throw new TypeError("Value contains a symbol key and is not strict JSON"); + const descriptor = Object.getOwnPropertyDescriptor(value, key)!; + if (!descriptor.enumerable || !("value" in descriptor)) { + throw new TypeError("Value must contain strict JSON enumerable data properties"); + } + if (descriptor.value === undefined && omitUndefinedProperties) continue; + defineData(result, key, copy(descriptor.value, active, omitUndefinedProperties)); + } + return result; + } finally { + active.delete(value); + } +} + +function defineData(target: object, key: PropertyKey, value: JsonValue): void { + DATA_DESCRIPTOR.value = value; + Object.defineProperty(target, key, DATA_DESCRIPTOR); + DATA_DESCRIPTOR.value = undefined; +} + /** Return whether a value is finite strict JSON with plain objects and no cycles. */ export function isJsonValue(value: unknown): value is JsonValue { - return check(value, new Set(), 0); + return check(value, new Set()); } -function check(value: unknown, ancestors: Set, depth: number): boolean { - if (depth > 512) return false; +function check(value: unknown, ancestors: Set): boolean { if (value === null || typeof value === "string" || typeof value === "boolean") return true; if (typeof value === "number") return Number.isFinite(value); if (Array.isArray(value)) { + if (Object.getPrototypeOf(value) !== Array.prototype) return false; const keys = Reflect.ownKeys(value); if (keys.length !== value.length + 1 || keys.some((key) => typeof key !== "string")) return false; if (ancestors.has(value)) return false; @@ -21,7 +91,7 @@ function check(value: unknown, ancestors: Set, depth: number): boolean { descriptor === undefined || !descriptor.enumerable || !("value" in descriptor) || - !check(descriptor.value, ancestors, depth + 1) + !check(descriptor.value, ancestors) ) { return false; } @@ -39,7 +109,7 @@ function check(value: unknown, ancestors: Set, depth: number): boolean { ancestors.add(value); try { for (const descriptor of Object.values(Object.getOwnPropertyDescriptors(value))) { - if (!descriptor.enumerable || !("value" in descriptor) || !check(descriptor.value, ancestors, depth + 1)) { + if (!descriptor.enumerable || !("value" in descriptor) || !check(descriptor.value, ancestors)) { return false; } } diff --git a/packages/chord/test/json.test.ts b/packages/chord/test/json.test.ts index 04424a5da..9e7080cb2 100644 --- a/packages/chord/test/json.test.ts +++ b/packages/chord/test/json.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "vitest"; -import { isJsonValue } from "../src/index.ts"; +import { copyJson, isJsonValue } from "../src/index.ts"; describe("isJsonValue", () => { test("checks strict JSON without normalizing it", () => { @@ -12,3 +12,54 @@ describe("isJsonValue", () => { expect(isJsonValue(cyclic)).toBe(false); }); }); + +describe("copyJson", () => { + test("copies strict JSON without retaining aliases", () => { + const shared = { value: 1 }; + const input = { left: shared, right: shared }; + const copied = copyJson(input) as typeof input; + expect(copied).toEqual(input); + expect(copied).not.toBe(input); + expect(copied.left).not.toBe(shared); + expect(copied.right).not.toBe(shared); + expect(copied.left).not.toBe(copied.right); + }); + + test("optionally omits undefined object properties without normalizing arrays", () => { + const input = { kept: 1, omitted: undefined, nested: { omitted: undefined, kept: true } }; + expect(() => copyJson(input)).toThrow(/strict JSON/); + expect(copyJson(input, { omitUndefinedProperties: true })).toEqual({ kept: 1, nested: { kept: true } }); + expect(() => copyJson([undefined], { omitUndefinedProperties: true })).toThrow(/strict JSON/); + }); + + test("preserves null prototypes and own __proto__ data properties", () => { + const input = Object.create(null) as Record; + Object.defineProperty(input, "__proto__", { + value: { safe: true }, + enumerable: true, + writable: true, + configurable: true, + }); + const copied = copyJson(input) as Record; + expect(Object.getPrototypeOf(copied)).toBeNull(); + expect(Object.hasOwn(copied, "__proto__")).toBe(true); + expect(copied.__proto__).toEqual({ safe: true }); + }); + + test("rejects cycles and non-strict container properties", () => { + const cyclic: { self?: unknown } = {}; + cyclic.self = cyclic; + const sparse: unknown[] = []; + sparse[1] = 1; + const extra = Object.assign([1], { extra: 2 }); + class ArraySubclass extends Array {} + const subclass = new ArraySubclass(1); + const accessor = Object.defineProperty({}, "value", { enumerable: true, get: () => 1 }); + const hidden = Object.defineProperty({}, "value", { enumerable: false, value: 1 }); + const symbol = { [Symbol("value")]: 1 }; + for (const invalid of [cyclic, sparse, extra, subclass, accessor, hidden, symbol]) { + expect(isJsonValue(invalid)).toBe(false); + expect(() => copyJson(invalid)).toThrow(/strict JSON/); + } + }); +}); diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 59682dec2..80e4ca78d 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -2,6 +2,14 @@ ## [Unreleased] +### Breaking Changes + +- Reordered Storage scan arguments so the limit precedes the cursor. + +### Added + +- Added transactional Sessions with typed durable documents, task creation, snapshots, retirement, and commit publications. + ## [0.87.1] - 2026-09-22 ## [0.87.0] - 2026-09-21 diff --git a/packages/durable/docs/pico-v5-handoff.md b/packages/durable/docs/pico-v5-handoff.md index 885a4c106..b6b347a49 100644 --- a/packages/durable/docs/pico-v5-handoff.md +++ b/packages/durable/docs/pico-v5-handoff.md @@ -11,7 +11,7 @@ facades, membranes, document routing, view projection, events, or clone chains. - Obsolete `pico` and `pico4` prototypes were removed. - `pico3` remains. -- Packages 1–5 are implemented in `packages/durable`; later Pico5 runtime packages remain. +- Packages 1–7 are implemented in `packages/durable`; later Pico5 runtime packages remain. ## 1. Records, cursors, and memory tables @@ -113,8 +113,10 @@ Experimental variants under other Delta directories are not Pico APIs. Implement these packages as one milestone. Keep the implementation layers separate, but do not build a temporary untyped document-acquisition seam. -Implement the generic `Tx` table surface these tests require: table reads, -`ReadAfterWrite`, ID-creating writes, and full task replacement. Semantic +Implement the generic `Tx` table surface these tests require: exact table reads, +paginated conversation/entry/task scans, `ReadAfterWrite`, ID-creating writes, +and full task replacement. Scans expose the Storage cursor and caller-selected +limit; they never hide an unbounded full scan. Semantic conversation, entry, task, and scheduler behavior remains in Packages 13–17. Keep one Astra-immutable tracker per loaded document. Its trusted immutable diff --git a/packages/durable/docs/pico-v5.md b/packages/durable/docs/pico-v5.md index 5e213a4d7..6796be810 100644 --- a/packages/durable/docs/pico-v5.md +++ b/packages/durable/docs/pico-v5.md @@ -366,8 +366,8 @@ interface Conversation { context(context: Context): Promise; entries( query: Omit, - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, context: Context, ): Promise>; fork( @@ -804,8 +804,9 @@ interface Tx { conversation(id: Id): Promise; entry(id: Id): Promise; task(id: Id): Promise | undefined>; - scanEntries(query: EntryQuery): Promise; - scanTasks(query: TaskQuery): Promise[]>; + scanConversations(limit: number, cursor?: Cursor): Promise>; + scanEntries(query: EntryQuery, limit: number, cursor?: Cursor): Promise>; + scanTasks(query: TaskQuery, limit: number, cursor?: Cursor): Promise, Cursor>>; createConversation(value: Omit): Promise; appendEntry(conversationId: Id, value: EntryDraft): Promise; @@ -2039,21 +2040,21 @@ interface Storage { mintId(): Promise; conversation(id: Id, context: Context): Promise; - scanConversations(cursor: Cursor | undefined, limit: number, context: Context): Promise>; + scanConversations(limit: number, cursor: Cursor | undefined, context: Context): Promise>; entry(id: Id, context: Context): Promise<{ readonly entry: EntryRecord; readonly commitSeq: Seq } | undefined>; findLatestHeadMarker(conversationId: Id, atOrBeforeEntryId: Id | undefined, context: Context): Promise<(EntryRecord & { readonly head: Id }) | undefined>; - scanEntries(query: EntryQuery, cursor: Cursor | undefined, limit: number, context: Context): Promise>; + scanEntries(query: EntryQuery, limit: number, cursor: Cursor | undefined, context: Context): Promise>; task(id: Id, context: Context): Promise | undefined>; - scanTasks(query: TaskQuery, cursor: Cursor | undefined, limit: number, context: Context): Promise, Cursor>>; + scanTasks(query: TaskQuery, limit: number, cursor: Cursor | undefined, context: Context): Promise, Cursor>>; submission(id: Id, context: Context): Promise; submissionByRequest(conversationId: Id, requestId: string, context: Context): Promise; findDocument(address: DocumentAddress, at: DocumentPoint, context: Context): Promise; document(id: Id, at: DocumentPoint, context: Context): Promise; - scanDocuments(query: DocumentQuery, cursor: Cursor | undefined, limit: number, context: Context): Promise>; + scanDocuments(query: DocumentQuery, limit: number, cursor: Cursor | undefined, context: Context): Promise>; close(context: Context): Promise; } diff --git a/packages/durable/src/documents.ts b/packages/durable/src/documents.ts new file mode 100644 index 000000000..6bb13689b --- /dev/null +++ b/packages/durable/src/documents.ts @@ -0,0 +1,177 @@ +import type { JsonValue } from "@earendil-works/chord"; +import type { + CommonDocDefinition, + ConversationDocFamilyToken, + ConversationDocToken, + DocDefinition, + DocFamilyDefinition, + DocFamilyToken, + DocToken, + DocumentAddress, + DocumentCreate, + DocumentRecord, + DocumentSemantics, + Id, + JsonObject, + LatestConversationSemantics, + RewindableConversationDocFamilyToken, + RewindableConversationDocToken, + RewindableConversationSemantics, + SessionDocFamilyToken, + SessionDocToken, + TaskDocFamilyToken, + TaskDocToken, +} from "./types.ts"; + +type FamilyInput = Omit, "initial"> & { + readonly family: true; + initial(seed: I): T; +}; + +/** Define a Session-scoped singleton document. */ +export function defineDoc( + definition: CommonDocDefinition & { readonly scope: "session" }, +): SessionDocToken; +/** Define a latest-only conversation singleton document. */ +export function defineDoc( + definition: CommonDocDefinition & LatestConversationSemantics, +): ConversationDocToken; +/** Define a rewindable conversation singleton document. */ +export function defineDoc( + definition: CommonDocDefinition & RewindableConversationSemantics, +): RewindableConversationDocToken; +/** Define a task-scoped singleton document. */ +export function defineDoc( + definition: CommonDocDefinition & { readonly scope: "task" }, +): TaskDocToken; +export function defineDoc(definition: DocDefinition): DocToken> { + validateDefinition(definition); + return { definition }; +} + +/** Define a Session-scoped document family. */ +export function defineDocFamily( + definition: FamilyInput & { readonly scope: "session" }, +): SessionDocFamilyToken; +/** Define a latest-only conversation document family. */ +export function defineDocFamily( + definition: FamilyInput & LatestConversationSemantics, +): ConversationDocFamilyToken; +/** Define a rewindable conversation document family. */ +export function defineDocFamily( + definition: FamilyInput & RewindableConversationSemantics, +): RewindableConversationDocFamilyToken; +/** Define a task-scoped document family. */ +export function defineDocFamily( + definition: FamilyInput & { readonly scope: "task" }, +): TaskDocFamilyToken; +export function defineDocFamily( + definition: DocFamilyDefinition, +): DocFamilyToken> { + validateDefinition(definition); + return { definition }; +} + +/** Erased definition shape used by the Session after overload resolution. */ +export type AnyDocDefinition = DocumentSemantics & { + readonly kind: string; + readonly version: number; + readonly family?: true; + initial(seed?: JsonValue): JsonObject; +}; + +/** Erased singleton or family token. */ +export type AnyDocToken = { readonly definition: AnyDocDefinition }; + +function validateDefinition(definition: AnyDocDefinition): void { + if (!Number.isSafeInteger(definition.version) || definition.version < 1) { + throw new TypeError(`Document ${definition.kind} version must be a positive integer`); + } +} + +/** Logical address plus its string identity for maps. */ +export type ResolvedAddress = { + readonly address: DocumentAddress; + readonly id: string; + readonly nextArgument: number; +}; + +/** Resolve an overloaded argument list and return the index after the owner and family key. */ +export function resolveAddress(definition: AnyDocDefinition, args: readonly unknown[]): ResolvedAddress { + let index = 0; + let scope: DocumentAddress["scope"]; + switch (definition.scope) { + case "session": + scope = { kind: "session" }; + break; + case "conversation": + scope = { kind: "conversation", conversationId: ownerId(args[index++], definition) }; + break; + case "task": + scope = { kind: "task", taskId: ownerId(args[index++], definition) }; + break; + } + const key = definition.family === true ? (args[index++] as string) : undefined; + const address: DocumentAddress = + key === undefined ? { kind: definition.kind, scope } : { kind: definition.kind, scope, key }; + return { address, id: addressId(address), nextArgument: index }; +} + +function ownerId(value: unknown, definition: AnyDocDefinition): Id { + if (!Number.isSafeInteger(value as number)) { + throw new TypeError(`Document ${definition.kind} requires a ${definition.scope} ID`); + } + return value as Id; +} + +/** Stable string identity of one logical address. */ +export function addressId(address: DocumentAddress): string { + const owner = + address.scope.kind === "session" + ? null + : address.scope.kind === "conversation" + ? address.scope.conversationId + : address.scope.taskId; + return JSON.stringify([address.kind, address.scope.kind, owner, address.key ?? null]); +} + +/** Build the storage create record for a new incarnation at an address. */ +export function documentCreate(definition: AnyDocDefinition, address: DocumentAddress, id: Id): DocumentCreate { + const key = address.key === undefined ? {} : { key: address.key }; + switch (address.scope.kind) { + case "session": + return { id, kind: address.kind, ...key, scope: address.scope }; + case "task": + return { id, kind: address.kind, ...key, scope: address.scope }; + case "conversation": + return { + id, + kind: address.kind, + ...key, + scope: address.scope, + history: definition.history!, + fork: definition.fork!, + } as DocumentCreate; + } +} + +/** Reject typed access whose token disagrees with the persisted scope, history, or fork semantics. */ +export function checkRecordScope(definition: AnyDocDefinition, record: DocumentRecord): void { + if ( + record.scope.kind !== definition.scope || + (record.scope.kind === "conversation" && + (record.history !== definition.history || record.fork !== definition.fork)) + ) { + throw new TypeError(`Document ${record.id} (${record.kind}) does not match the supplied definition semantics`); + } +} + +/** Reject typed access to a stored version the token cannot use without migration. */ +export function checkRecordVersion(definition: AnyDocDefinition, record: DocumentRecord, version: number): void { + if (version > definition.version) { + throw new Error(`Document ${record.id} (${record.kind}) has newer version ${version} than ${definition.version}`); + } + if (version < definition.version) { + throw new Error(`Document ${record.id} (${record.kind}) requires migration from version ${version}`); + } +} diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index 1c5a43fff..fdbfa4515 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -1,30 +1,55 @@ +export { defineDoc, defineDocFamily } from "./documents.ts"; +export { createSession } from "./session/session.ts"; +export { ReadAfterWrite } from "./session/transaction.ts"; export { MemoryStorage } from "./storage/memory.ts"; export type { + CommonDocDefinition, ContextEdit, + ConversationDocFamilyToken, + ConversationDocToken, ConversationRecord, Cursor, + DocDefinition, + DocFamilyDefinition, + DocFamilyToken, + DocToken, DocumentAddress, DocumentContent, DocumentCreate, DocumentPoint, DocumentQuery, DocumentRecord, + DocumentSemantics, EntryDraft, EntryQuery, EntryRecord, Id, JsonObject, + LatestConversationSemantics, Page, + RewindableConversationDocFamilyToken, + RewindableConversationDocToken, + RewindableConversationSemantics, Seq, + Session, + SessionDocFamilyToken, + SessionDocToken, Storage, StorageWrite, StoredDocument, SubmissionCreate, SubmissionRecord, + Task, + TaskDefinition, + TaskDocFamilyToken, + TaskDocToken, + TaskOptions, TaskOutcome, TaskOutcomeError, TaskQuery, TaskRecord, + TaskRef, TaskState, + Tx, } from "./types.ts"; export { ROOT_CONVERSATION_ID } from "./types.ts"; diff --git a/packages/durable/src/session/publications.ts b/packages/durable/src/session/publications.ts new file mode 100644 index 000000000..3df8d78e7 --- /dev/null +++ b/packages/durable/src/session/publications.ts @@ -0,0 +1,16 @@ +import type { Seq, StorageWrite } from "../types.ts"; +import type { DocumentCommitChange } from "./transaction.ts"; + +/** Complete table record committed without another publication copy. */ +export type TableCommitChange = Extract< + StorageWrite, + { readonly type: "conversation" | "entry" | "task" | "submission" } +>; + +export type CommitChange = TableCommitChange | DocumentCommitChange; + +/** Every immutable change from one successful Session commit. Change order is unspecified. */ +export type CommitPublication = { + readonly seq: Seq; + readonly changes: readonly CommitChange[]; +}; diff --git a/packages/durable/src/session/session.ts b/packages/durable/src/session/session.ts new file mode 100644 index 000000000..62215376f --- /dev/null +++ b/packages/durable/src/session/session.ts @@ -0,0 +1,239 @@ +import type { Context, JsonValue } from "@earendil-works/chord"; +import { awaitWithContext, withoutAbortSignal } from "@earendil-works/chord/context"; +import { track } from "@earendil-works/chord/delta"; +import { type AnyDocToken, checkRecordScope, checkRecordVersion, resolveAddress } from "../documents.ts"; +import type { + ConversationDocFamilyToken, + ConversationDocToken, + DocumentAddress, + Id, + JsonObject, + Session, + SessionDocFamilyToken, + SessionDocToken, + Storage, + StorageWrite, + TaskDocFamilyToken, + TaskDocToken, + Tx, +} from "../types.ts"; +import type { CommitChange, CommitPublication } from "./publications.ts"; +import { type DocumentCommitChange, type LoadedDocument, Transaction, type TransactionHost } from "./transaction.ts"; + +/** Open a Session kernel over one storage backend. */ +export function createSession(storage: Storage): Session { + return new SessionKernel(storage); +} + +/** + * Session kernel: one mutation line, the loaded document tracker cache, and committed publication. + * + * Only committed state is observable. Every commit callback, preparation, Storage settlement, adoption, and + * publication enqueue runs while the line is held; listeners run later. + */ +export class SessionKernel implements Session { + readonly #storage: Storage; + readonly #documents = new Map(); + readonly #listeners = new Set<(publication: CommitPublication, context: Context) => void>(); + readonly #host: TransactionHost; + #tail: Promise = Promise.resolve(); + #closing: Promise | undefined; + #poison: { readonly error: unknown } | undefined; + + constructor(storage: Storage) { + this.#storage = storage; + this.#host = { + storage, + cached: (id) => this.#documents.get(id), + load: (addressId, address, context) => this.#load(addressId, address, context), + install: (document) => { + this.#documents.set(document.addressId, document); + }, + evict: (id, recordId) => { + if (this.#documents.get(id)?.record.id === recordId) this.#documents.delete(id); + }, + }; + } + + commit(change: (tx: Tx) => T | Promise, context: Context): Promise { + try { + this.#assertUsable(); + } catch (error) { + return Promise.reject(error); + } + return this.#enqueue(() => this.#runCommit(change, context)); + } + + snapshot(token: SessionDocToken, context: Context): Promise | undefined>; + snapshot( + token: ConversationDocToken, + conversationId: Id, + context: Context, + ): Promise | undefined>; + snapshot( + token: TaskDocToken, + taskId: Id, + context: Context, + ): Promise | undefined>; + snapshot( + token: SessionDocFamilyToken, + key: string, + context: Context, + ): Promise | undefined>; + snapshot( + token: ConversationDocFamilyToken, + conversationId: Id, + key: string, + context: Context, + ): Promise | undefined>; + snapshot( + token: TaskDocFamilyToken, + taskId: Id, + key: string, + context: Context, + ): Promise | undefined>; + async snapshot(token: AnyDocToken, ...args: readonly unknown[]): Promise { + this.#assertUsable(); + const definition = token.definition; + const resolved = resolveAddress(definition, args); + const context = args[resolved.nextArgument] as Context; + const loaded = + this.#documents.get(resolved.id) ?? + (await this.#enqueue(async () => { + this.#assertHealthy(); + return this.#load(resolved.id, resolved.address, context); + })); + if (loaded === undefined) return undefined; + checkRecordScope(definition, loaded.record); + checkRecordVersion(definition, loaded.record, loaded.version); + return loaded.tracker.value; + } + + close(context: Context): Promise { + if (this.#closing === undefined) { + const cleanup = withoutAbortSignal(context); + this.#closing = this.#enqueue(async () => { + this.#documents.clear(); + await this.#storage.close(cleanup); + }); + } + return awaitWithContext(this.#closing, context); + } + + /** Register an internal commit listener. Callbacks run later in commit order and must not throw. */ + subscribeCommits(listener: (publication: CommitPublication, context: Context) => void): () => void { + this.#listeners.add(listener); + return () => { + this.#listeners.delete(listener); + }; + } + + /** Drop every loaded tracker on the mutation line; later access cold-loads from Storage. */ + unloadDocuments(): Promise { + return this.#enqueue(async () => { + this.#documents.clear(); + }); + } + + async #runCommit(change: (tx: Tx) => T | Promise, context: Context): Promise { + this.#assertHealthy(); + context.abortSignal?.throwIfAborted(); + const tx = new Transaction(this.#host, context); + let result: T; + try { + result = await change(tx); + } catch (error) { + await tx.settleFailure(); + throw error; + } + const writes = await tx.settleSuccess(); + if (writes.length === 0) { + tx.discard(); + return result; + } + let seq: number; + try { + // Once admitted, caller cancellation does not interrupt Storage settlement. + seq = await this.#storage.commit(writes, withoutAbortSignal(context)); + } catch (error) { + tx.discard(); + this.#poison = { error }; + throw error; + } + let documents: DocumentCommitChange[]; + try { + documents = tx.adopt(seq); + } catch (error) { + // Storage already committed; a failed adoption leaves memory behind durable state. + this.#poison = { error }; + throw error; + } + this.#publish(seq, writes, documents, context); + return result; + } + + #publish( + seq: number, + writes: readonly StorageWrite[], + documents: readonly DocumentCommitChange[], + context: Context, + ): void { + if (this.#listeners.size === 0) return; + const changes: CommitChange[] = []; + for (const write of writes) { + switch (write.type) { + case "conversation": + case "entry": + case "task": + case "submission": + changes.push(write); + } + } + for (const document of documents) changes.push(document); + const publication: CommitPublication = { seq, changes }; + const listeners = [...this.#listeners]; + queueMicrotask(() => { + for (const listener of listeners) listener(publication, context); + }); + } + + async #load(addressId: string, address: DocumentAddress, context: Context): Promise { + const cached = this.#documents.get(addressId); + if (cached !== undefined) return cached; + const record = await this.#storage.findDocument(address, "current", context); + if (record === undefined) return undefined; + const stored = await this.#storage.document(record.id, "current", context); + if (stored === undefined) throw new Error(`Current document ${record.id} (${record.kind}) cannot be read`); + // Storage returns detached strict JSON, which the tracker owns without another copy. + const loaded: LoadedDocument = { + addressId, + record: stored.record, + version: stored.version, + tracker: track(stored.value), + }; + this.#documents.set(addressId, loaded); + return loaded; + } + + #enqueue(job: () => Promise): Promise { + const run = this.#tail.then(job); + this.#tail = run.then( + () => undefined, + () => undefined, + ); + return run; + } + + #assertUsable(): void { + if (this.#closing !== undefined) throw new Error("Session is closed"); + this.#assertHealthy(); + } + + #assertHealthy(): void { + if (this.#poison !== undefined) { + throw new Error("Session is poisoned by a failed commit after storage admission; reopen it", { + cause: this.#poison.error, + }); + } + } +} diff --git a/packages/durable/src/session/transaction.ts b/packages/durable/src/session/transaction.ts new file mode 100644 index 000000000..5f10b72b9 --- /dev/null +++ b/packages/durable/src/session/transaction.ts @@ -0,0 +1,677 @@ +import { type Context, copyJson, type Draft, type JsonValue } from "@earendil-works/chord"; +import { type Change, type Op, type Prepared, type Tracker, track } from "@earendil-works/chord/delta"; +import { + type AnyDocDefinition, + type AnyDocToken, + addressId, + checkRecordScope, + checkRecordVersion, + documentCreate, + type ResolvedAddress, + resolveAddress, +} from "../documents.ts"; +import type { + ConversationDocFamilyToken, + ConversationDocToken, + ConversationRecord, + Cursor, + DocumentAddress, + DocumentCreate, + DocumentRecord, + EntryDraft, + EntryQuery, + EntryRecord, + Id, + JsonObject, + Seq, + SessionDocFamilyToken, + SessionDocToken, + Storage, + StorageWrite, + Task, + TaskDocFamilyToken, + TaskDocToken, + TaskOptions, + TaskQuery, + TaskRecord, + TaskRef, + Tx, +} from "../types.ts"; + +type AnyTaskRecord = TaskRecord; + +const INTERNAL_SCAN_PAGE_SIZE = 256; +const EMPTY_OPERATIONS: readonly Op[] = []; +const TABLE_JSON_COPY_OPTIONS = { omitUndefinedProperties: true } as const; + +/** A transaction read a table after its first table write. Read every required row before writing. */ +export class ReadAfterWrite extends Error { + constructor(method: string) { + super(`Tx.${method}() cannot read tables after the first table write`); + this.name = "ReadAfterWrite"; + } +} + +/** Committed change of one document incarnation. */ +export type DocumentCommitChange = { + readonly type: "document"; + readonly record: DocumentRecord; + /** Conversation owning the document; task documents derive it from their task record. */ + readonly conversationId: Id | undefined; + /** Exact adopted immutable revision, or `null` when this commit retired the incarnation. */ + readonly value: JsonObject | null; + /** Exact adopted operations for an ordinary update; empty for creation and retirement. */ + readonly ops: readonly Op[]; +}; + +/** One committed document incarnation owned by the Session tracker cache. */ +export type LoadedDocument = { + readonly addressId: string; + readonly record: DocumentRecord; + /** Stored definition version of the tracked value. */ + readonly version: number; + readonly tracker: Tracker; +}; + +/** Session services used by a transaction while it holds the mutation line. */ +export interface TransactionHost { + readonly storage: Storage; + /** Return the cached current incarnation without loading. */ + cached(addressId: string): LoadedDocument | undefined; + /** Return the cached current incarnation, cold-loading it on the mutation line when necessary. */ + load(addressId: string, address: DocumentAddress, context: Context): Promise; + /** Install a newly committed incarnation. */ + install(document: LoadedDocument): void; + /** Remove a retired incarnation if it is still the cached occupant of its address. */ + evict(addressId: string, recordId: Id): void; +} + +/** Committed and candidate state for one task touched by this transaction. */ +type TransactionTask = { + committedRead?: Promise; + write?: { readonly kind: "create" | "replace"; readonly record: AnyTaskRecord }; + publicationConversationId?: Id; +}; + +/** Storage/cache provenance of one staged document incarnation. */ +type DocumentTarget = + | { readonly kind: "loaded"; readonly document: LoadedDocument } + | { + readonly kind: "created"; + readonly record: DocumentCreate; + readonly version: number; + readonly tracker: Tracker; + } + | { readonly kind: "retire-only"; readonly record: DocumentRecord }; + +/** One document incarnation acquired, created, or retired by this transaction. */ +type DocumentEntry = { + readonly addressId: string; + readonly address: DocumentAddress; + /** Absent only for retirement entries discovered by a terminal-task scan. */ + readonly definition?: AnyDocDefinition; + /** Memoized public acquisition; absent for metadata-only retirement. */ + draftPromise?: Promise>; + /** Set after acquisition or retirement lookup finds the affected incarnation. */ + target?: DocumentTarget; + change?: Change; + prepared?: Prepared; + retireOnCommit: boolean; + /** Resolved before Storage admission so adoption performs no reads. */ + conversationId?: Id; +}; + +/** + * Transaction for one Session commit callback. + * + * Every asynchronous operation is tracked so callback settlement can reject and drain unfinished work. Session calls + * one settlement method, then either discards prepared changes or adopts them once after Storage succeeds. + */ +export class Transaction implements Tx { + readonly #host: TransactionHost; + readonly #context: Context; + readonly #pendingOperations = new Set>(); + #sealed = false; + #hasTableWrite = false; + + /** Atomic batch; conversation and entry writes stage eagerly, while task and document writes assemble later. */ + readonly #writes: StorageWrite[] = []; + readonly #createdConversationIds = new Set(); + /** One entry per task touched by a public read, candidate write, or document-owner lookup. */ + readonly #tasksById = new Map(); + + /** Every document acquisition or retirement marker in staging order. */ + readonly #documents: DocumentEntry[] = []; + /** Latest transaction-local incarnation or retirement marker at each logical address. */ + readonly #latestDocumentByAddress = new Map(); + + constructor(host: TransactionHost, context: Context) { + this.#host = host; + this.#context = context; + } + + // ─── Table reads ──────────────────────────────────────────────────────── + + conversation(id: Id): Promise { + return this.#read("conversation", () => this.#host.storage.conversation(id, this.#context)); + } + + entry(id: Id): Promise { + return this.#read("entry", async () => (await this.#host.storage.entry(id, this.#context))?.entry); + } + + task(id: Id): Promise { + return this.#read("task", () => this.#committedTask(id)); + } + + scanConversations(limit: number, cursor?: Cursor) { + return this.#read("scanConversations", () => this.#host.storage.scanConversations(limit, cursor, this.#context)); + } + + scanEntries(query: EntryQuery, limit: number, cursor?: Cursor) { + return this.#read("scanEntries", () => this.#host.storage.scanEntries(query, limit, cursor, this.#context)); + } + + scanTasks(query: TaskQuery, limit: number, cursor?: Cursor) { + return this.#read("scanTasks", () => this.#host.storage.scanTasks(query, limit, cursor, this.#context)); + } + + // ─── Table writes ─────────────────────────────────────────────────────── + + createConversation(value: Omit): Promise { + return this.#write(async () => { + const id = await this.#host.storage.mintId(); + this.#assertOpen(); + const record = copyJson({ ...value, id }, TABLE_JSON_COPY_OPTIONS) as ConversationRecord; + this.#createdConversationIds.add(id); + this.#writes.push({ type: "conversation", value: record }); + return record; + }); + } + + appendEntry(conversationId: Id, value: EntryDraft): Promise { + return this.#write(async () => { + await this.#requireConversation(conversationId); + this.#assertOpen(); + const id = await this.#host.storage.mintId(); + this.#assertOpen(); + const { head, ...rest } = value; + const record = copyJson( + head === undefined + ? { ...rest, id, conversationId } + : { ...rest, id, conversationId, head: head === "self" ? id : head }, + TABLE_JSON_COPY_OPTIONS, + ) as unknown as EntryRecord; + this.#writes.push({ type: "entry", value: record }); + return record; + }); + } + + createTask( + task: Task, + input: I, + options?: TaskOptions, + ): Promise> { + return this.#write(async () => { + const conversationId = options?.conversationId; + if (conversationId === undefined) throw new TypeError("Tx.createTask() requires options.conversationId"); + await this.#requireConversation(conversationId); + this.#assertOpen(); + const definition = task.definition; + const checkpoint = definition.initial(input); + const id = await this.#host.storage.mintId(); + this.#assertOpen(); + const record = copyJson( + { + id, + conversationId, + kind: definition.name, + version: definition.version, + input, + after: options?.after ?? [], + background: options?.background ?? false, + abortRequested: false, + state: { status: "pending", checkpoint }, + }, + TABLE_JSON_COPY_OPTIONS, + ) as unknown as AnyTaskRecord; + this.#tasksById.set(id, { write: { kind: "create", record } }); + return { id }; + }); + } + + setTask(value: AnyTaskRecord): void { + this.#assertOpen(); + this.#hasTableWrite = true; + const task = this.#taskEntry(value.id); + if (task.write?.record.state.status === "terminal") { + throw new Error(`Task ${value.id} already has a terminal candidate`); + } + task.write = { + kind: task.write?.kind === "create" ? "create" : "replace", + record: copyJson(value, TABLE_JSON_COPY_OPTIONS) as unknown as AnyTaskRecord, + }; + } + + // ─── Documents ────────────────────────────────────────────────────────── + doc(token: SessionDocToken): Promise>; + doc(token: ConversationDocToken, conversationId: Id): Promise>; + doc(token: TaskDocToken, taskId: Id): Promise>; + doc( + token: SessionDocFamilyToken, + key: string, + seed: I, + ): Promise>; + doc( + token: ConversationDocFamilyToken, + conversationId: Id, + key: string, + seed: I, + ): Promise>; + doc( + token: TaskDocFamilyToken, + taskId: Id, + key: string, + seed: I, + ): Promise>; + doc(token: AnyDocToken, ...args: readonly unknown[]): Promise> { + try { + this.#assertOpen(); + const definition = token.definition; + const resolved = resolveAddress(definition, args); + this.#assertTaskDocumentsOpen(resolved); + const latest = this.#latestDocumentByAddress.get(resolved.id); + if (latest !== undefined && !latest.retireOnCommit && latest.draftPromise !== undefined) { + return latest.draftPromise; + } + const seed = definition.family === true ? copyJson(args[resolved.nextArgument]) : undefined; + const docEntry: DocumentEntry = { + addressId: resolved.id, + address: resolved.address, + definition, + retireOnCommit: false, + }; + this.#documents.push(docEntry); + this.#latestDocumentByAddress.set(docEntry.addressId, docEntry); + // Capture retirement before awaiting so a pending old acquisition and its replacement stay distinct. + docEntry.draftPromise = this.#track(this.#acquire(docEntry, seed, latest?.retireOnCommit === true)); + return docEntry.draftPromise; + } catch (error) { + return Promise.reject(error); + } + } + + retireDoc(token: SessionDocToken): Promise; + retireDoc(token: ConversationDocToken, conversationId: Id): Promise; + retireDoc(token: TaskDocToken, taskId: Id): Promise; + retireDoc(token: SessionDocFamilyToken, key: string): Promise; + retireDoc( + token: ConversationDocFamilyToken, + conversationId: Id, + key: string, + ): Promise; + retireDoc( + token: TaskDocFamilyToken, + taskId: Id, + key: string, + ): Promise; + retireDoc(token: AnyDocToken, ...args: readonly unknown[]): Promise { + try { + this.#assertOpen(); + const definition = token.definition; + const resolved = resolveAddress(definition, args); + const latest = this.#latestDocumentByAddress.get(resolved.id); + if (latest?.retireOnCommit) return Promise.resolve(); + if (latest?.draftPromise !== undefined) { + // Retirement of an acquired draft persists its final content before retirement. + latest.retireOnCommit = true; + return this.#track(latest.draftPromise.then(() => undefined)); + } + const entry: DocumentEntry = { + addressId: resolved.id, + address: resolved.address, + definition, + retireOnCommit: true, + }; + this.#documents.push(entry); + this.#latestDocumentByAddress.set(entry.addressId, entry); + return this.#track(this.#findRetirement(entry)); + } catch (error) { + return Promise.reject(error); + } + } + + async #acquire(entry: DocumentEntry, seed: JsonValue | undefined, skipLoad: boolean): Promise> { + const definition = entry.definition!; + const loaded = skipLoad ? undefined : await this.#host.load(entry.addressId, entry.address, this.#context); + this.#assertOpen(); + if (loaded !== undefined) { + checkRecordScope(definition, loaded.record); + checkRecordVersion(definition, loaded.record, loaded.version); + entry.target = { kind: "loaded", document: loaded }; + entry.change = loaded.tracker.beginChange(); + return entry.change.state; + } + const scope = entry.address.scope; + if (scope.kind === "conversation") await this.#requireConversation(scope.conversationId); + if (scope.kind === "task") { + const task = await this.#currentTask(scope.taskId); + if (task === undefined) throw new Error(`Task ${scope.taskId} does not exist`); + if (task.state.status === "terminal") throw new Error(`Task ${scope.taskId} is terminal`); + } + this.#assertOpen(); + const value = copyJson( + definition.family === true ? definition.initial(seed) : definition.initial(), + ) as JsonObject; + const id = await this.#host.storage.mintId(); + this.#assertOpen(); + const tracker = track(value); + entry.target = { + kind: "created", + record: documentCreate(definition, entry.address, id), + version: definition.version, + tracker, + }; + entry.change = tracker.beginChange(); + return entry.change.state; + } + + async #findRetirement(entry: DocumentEntry): Promise { + const record = + this.#host.cached(entry.addressId)?.record ?? + (await this.#host.storage.findDocument(entry.address, "current", this.#context)); + this.#assertOpen(); + if (record === undefined) return; + checkRecordScope(entry.definition!, record); + entry.target = { kind: "retire-only", record }; + } + + // ─── Settlement ───────────────────────────────────────────────────────── + + /** Seal after callback failure: abort every change and observe every pending operation. */ + async settleFailure(): Promise { + this.#sealed = true; + this.#abortChanges(); + await this.#drain(); + } + + /** + * Seal after callback success, prepare every open change, and assemble the atomic batch. + * Any failure aborts every change before Storage admission. + */ + async settleSuccess(): Promise { + this.#sealed = true; + if (this.#pendingOperations.size > 0) { + this.#abortChanges(); + await this.#drain(); + throw new Error("Session commit callback settled before its pending Tx operations"); + } + try { + // Synchronously prepare every open change; this revokes every draft. + for (const document of this.#documents) { + if (document.change !== undefined) document.prepared = document.change.prepare(); + } + return await this.#assemble(); + } catch (error) { + this.#abortChanges(); + throw error; + } + } + + /** Abort every prepared change after Storage failure or when no write is required. */ + discard(): void { + this.#abortChanges(); + } + + /** Adopt every prepared change by pointer swap after Storage success and describe the publication. */ + adopt(seq: Seq): DocumentCommitChange[] { + const publications: DocumentCommitChange[] = []; + for (const document of this.#documents) { + const target = document.target; + if (target === undefined) continue; + switch (target.kind) { + case "created": { + const prepared = document.prepared!; + const record: DocumentRecord = document.retireOnCommit + ? { ...target.record, createdAt: seq, retiredAt: seq } + : { ...target.record, createdAt: seq }; + if (document.retireOnCommit) prepared.abort(); + else { + target.tracker.adopt(prepared); + this.#host.install({ + addressId: document.addressId, + record, + version: target.version, + tracker: target.tracker, + }); + } + publications.push({ + type: "document", + record, + conversationId: document.conversationId, + value: document.retireOnCommit ? null : prepared.value, + ops: EMPTY_OPERATIONS, + }); + break; + } + case "loaded": { + const prepared = document.prepared!; + const changed = prepared.ops.length > 0; + if (changed) target.document.tracker.adopt(prepared); + else prepared.abort(); + if (!document.retireOnCommit && !changed) break; + if (document.retireOnCommit) { + this.#host.evict(target.document.addressId, target.document.record.id); + } + publications.push({ + type: "document", + record: document.retireOnCommit + ? { ...target.document.record, retiredAt: seq } + : target.document.record, + conversationId: document.conversationId, + value: document.retireOnCommit ? null : prepared.value, + ops: document.retireOnCommit ? EMPTY_OPERATIONS : prepared.ops, + }); + break; + } + case "retire-only": + this.#host.evict(document.addressId, target.record.id); + publications.push({ + type: "document", + record: { ...target.record, retiredAt: seq }, + conversationId: document.conversationId, + value: null, + ops: EMPTY_OPERATIONS, + }); + break; + } + } + return publications; + } + + async #assemble(): Promise { + const storage = this.#host.storage; + for (const [id, task] of this.#tasksById) { + if (task.write?.kind !== "replace") continue; + const committed = await this.#committedTask(id); + if (committed === undefined) throw new Error(`Task ${id} does not exist`); + if (committed.state.status === "terminal") throw new Error(`Task ${id} is already terminal`); + } + + // Terminal settlement retires every task document, including ones created by this transaction. + let terminalTaskIds: Set | undefined; + for (const task of this.#tasksById.values()) { + const candidate = task.write?.record; + if (candidate?.state.status !== "terminal") continue; + terminalTaskIds ??= new Set(); + terminalTaskIds.add(candidate.id); + } + if (terminalTaskIds !== undefined) { + const targetedDocumentIds = new Set(); + for (const document of this.#documents) { + const scope = document.address.scope; + if (scope.kind !== "task" || !terminalTaskIds.has(scope.taskId)) continue; + document.retireOnCommit = true; + const target = document.target; + if (target?.kind === "loaded") targetedDocumentIds.add(target.document.record.id); + if (target?.kind === "created" || target?.kind === "retire-only") targetedDocumentIds.add(target.record.id); + } + for (const taskId of terminalTaskIds) { + if (this.#tasksById.get(taskId)?.write?.kind === "create") continue; + let cursor: Cursor | undefined; + do { + const page = await storage.scanDocuments( + { scope: { kind: "task", taskId }, at: "current" }, + INTERNAL_SCAN_PAGE_SIZE, + cursor, + this.#context, + ); + for (const record of page.items) { + if (targetedDocumentIds.has(record.id)) continue; + this.#documents.push({ + addressId: addressId(record), + address: record, + target: { kind: "retire-only", record }, + retireOnCommit: true, + }); + targetedDocumentIds.add(record.id); + } + cursor = page.next; + } while (cursor !== undefined); + } + } + + // Resolve task-document publication ownership before Storage admission so adoption remains synchronous. + for (const document of this.#documents) { + const target = document.target; + if (target === undefined) continue; + if (target.kind === "loaded" && !document.retireOnCommit && document.prepared!.ops.length === 0) continue; + const scope = document.address.scope; + if (scope.kind === "conversation") document.conversationId = scope.conversationId; + if (scope.kind !== "task") continue; + const task = this.#taskEntry(scope.taskId); + if (task.publicationConversationId === undefined) { + const current = await this.#currentTask(scope.taskId); + if (current !== undefined) task.publicationConversationId = current.conversationId; + } + document.conversationId = task.publicationConversationId; + } + + const writes = this.#writes; + for (const task of this.#tasksById.values()) { + if (task.write !== undefined) writes.push({ type: "task", value: task.write.record }); + } + for (const document of this.#documents) { + const target = document.target; + if (target === undefined) continue; + switch (target.kind) { + case "created": + writes.push({ + type: "document.create", + record: target.record, + content: { version: target.version, kind: "base", value: document.prepared!.value }, + }); + if (document.retireOnCommit) writes.push({ type: "document.retire", id: target.record.id }); + break; + case "loaded": + if (document.prepared!.ops.length > 0) { + writes.push({ + type: "document.change", + id: target.document.record.id, + content: { + version: target.document.version, + kind: "delta", + ops: document.prepared!.ops, + }, + }); + } + if (document.retireOnCommit) { + writes.push({ type: "document.retire", id: target.document.record.id }); + } + break; + case "retire-only": + writes.push({ type: "document.retire", id: target.record.id }); + break; + } + } + return writes; + } + + // ─── Helpers ──────────────────────────────────────────────────────────── + + #abortChanges(): void { + for (const document of this.#documents) document.change?.abort(); + } + + async #drain(): Promise { + await Promise.allSettled(this.#pendingOperations); + } + + #assertOpen(): void { + if (this.#sealed) throw new Error("Transaction has settled"); + } + + #assertTaskDocumentsOpen(resolved: ResolvedAddress): void { + const scope = resolved.address.scope; + if (scope.kind === "task" && this.#tasksById.get(scope.taskId)?.write?.record.state.status === "terminal") { + throw new Error(`Task ${scope.taskId} is terminal`); + } + } + + /** Register an operation so callback settlement can reject and drain it. */ + #track(operation: Promise): Promise { + this.#pendingOperations.add(operation); + const settle = (): void => { + this.#pendingOperations.delete(operation); + }; + operation.then(settle, settle); + return operation; + } + + #read(method: string, read: () => Promise): Promise { + try { + this.#assertOpen(); + if (this.#hasTableWrite) throw new ReadAfterWrite(method); + return this.#track(read()); + } catch (error) { + return Promise.reject(error); + } + } + + #write(write: () => Promise): Promise { + try { + this.#assertOpen(); + this.#hasTableWrite = true; + return this.#track(write()); + } catch (error) { + return Promise.reject(error); + } + } + + async #requireConversation(id: Id): Promise { + if (this.#createdConversationIds.has(id)) return; + if ((await this.#host.storage.conversation(id, this.#context)) === undefined) { + throw new Error(`Conversation ${id} does not exist`); + } + } + + #taskEntry(id: Id): TransactionTask { + let task = this.#tasksById.get(id); + if (task === undefined) { + task = {}; + this.#tasksById.set(id, task); + } + return task; + } + + /** Latest candidate task record, falling back to committed state; not a caller table read. */ + async #currentTask(id: Id): Promise { + return this.#tasksById.get(id)?.write?.record ?? (await this.#committedTask(id)); + } + + #committedTask(id: Id): Promise { + const task = this.#taskEntry(id); + task.committedRead ??= this.#host.storage.task(id, this.#context); + return task.committedRead; + } +} diff --git a/packages/durable/src/storage/jsonl/storage.ts b/packages/durable/src/storage/jsonl/storage.ts index 36b62851b..fc22cdc41 100644 --- a/packages/durable/src/storage/jsonl/storage.ts +++ b/packages/durable/src/storage/jsonl/storage.ts @@ -100,11 +100,7 @@ const isCurrentOnly = (record: DocumentCreate): boolean => const sidecarKey = (file: string, seq: Seq, ordinal: number): string => JSON.stringify([file, seq, ordinal]); -const jsonLine = (value: unknown): string => { - const encoded = JSON.stringify(value); - if (encoded === undefined) throw new TypeError("JSONL record is not serializable"); - return `${encoded}\n`; -}; +const jsonLine = (value: MainMarker | SidecarRecord): string => `${JSON.stringify(value)}\n`; const errorFromFile = (action: string, error: FileError): Error => new Error(`JSONL ${action} failed: ${error.message}`, { cause: error }); @@ -309,8 +305,8 @@ export class JsonlStorage implements Storage { return this.store.conversation(id, context); } - async scanConversations(cursor: Cursor | undefined, limit: number, context: Context) { - return this.store.scanConversations(cursor, limit, context); + async scanConversations(limit: number, cursor: Cursor | undefined, context: Context) { + return this.store.scanConversations(limit, cursor, context); } async entry(id: Id, context: Context) { @@ -321,16 +317,16 @@ export class JsonlStorage implements Storage { return this.store.findLatestHeadMarker(conversationId, atOrBeforeEntryId, context); } - async scanEntries(query: EntryQuery, cursor: Cursor | undefined, limit: number, context: Context) { - return this.store.scanEntries(query, cursor, limit, context); + async scanEntries(query: EntryQuery, limit: number, cursor: Cursor | undefined, context: Context) { + return this.store.scanEntries(query, limit, cursor, context); } async task(id: Id, context: Context) { return this.store.task(id, context); } - async scanTasks(query: TaskQuery, cursor: Cursor | undefined, limit: number, context: Context) { - return this.store.scanTasks(query, cursor, limit, context); + async scanTasks(query: TaskQuery, limit: number, cursor: Cursor | undefined, context: Context) { + return this.store.scanTasks(query, limit, cursor, context); } async submission(id: Id, context: Context) { @@ -349,8 +345,8 @@ export class JsonlStorage implements Storage { return this.store.document(id, at, context); } - async scanDocuments(query: DocumentQuery, cursor: Cursor | undefined, limit: number, context: Context) { - return this.store.scanDocuments(query, cursor, limit, context); + async scanDocuments(query: DocumentQuery, limit: number, cursor: Cursor | undefined, context: Context) { + return this.store.scanDocuments(query, limit, cursor, context); } async close(context: Context): Promise { diff --git a/packages/durable/src/storage/memory.ts b/packages/durable/src/storage/memory.ts index c31cbf3e8..feba4a886 100644 --- a/packages/durable/src/storage/memory.ts +++ b/packages/durable/src/storage/memory.ts @@ -320,8 +320,8 @@ export class MemoryStorage implements Storage { } async scanConversations( - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, _context: Context, ): Promise> { this.assertOpen(); @@ -370,8 +370,8 @@ export class MemoryStorage implements Storage { async scanEntries( query: EntryQuery, - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, _context: Context, ): Promise> { this.assertOpen(); @@ -394,8 +394,8 @@ export class MemoryStorage implements Storage { async scanTasks( query: TaskQuery, - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, _context: Context, ): Promise> { this.assertOpen(); @@ -474,8 +474,8 @@ export class MemoryStorage implements Storage { async scanDocuments( query: DocumentQuery, - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, _context: Context, ): Promise> { this.assertOpen(); @@ -581,9 +581,6 @@ export class MemoryStorage implements Storage { if (action.create === undefined && existing === undefined) throw new Error(`Unknown document: ${id}`); if (action.create !== undefined && existing !== undefined) throw new Error(`Document ${id} already exists`); if (existing?.record.retiredAt !== undefined) throw new Error(`Document ${id} is retired`); - if (action.create !== undefined && action.content?.kind !== "base") { - throw new Error(`Document ${id} creation requires a base`); - } const previous = existing?.revisions.at(-1); if (action.content?.kind === "delta") { if (previous === undefined) throw new Error(`Document ${id} delta has no base`); diff --git a/packages/durable/src/storage/sqlite/storage.ts b/packages/durable/src/storage/sqlite/storage.ts index cd656a2fe..d6c6fd4db 100644 --- a/packages/durable/src/storage/sqlite/storage.ts +++ b/packages/durable/src/storage/sqlite/storage.ts @@ -213,8 +213,8 @@ export class SqliteStorage implements Storage { } async scanConversations( - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, _context: Context, ): Promise> { this.assertOpen(); @@ -273,8 +273,8 @@ export class SqliteStorage implements Storage { async scanEntries( query: EntryQuery, - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, _context: Context, ): Promise> { this.assertOpen(); @@ -317,8 +317,8 @@ export class SqliteStorage implements Storage { async scanTasks( query: TaskQuery, - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, _context: Context, ): Promise> { this.assertOpen(); @@ -433,8 +433,8 @@ export class SqliteStorage implements Storage { async scanDocuments( query: DocumentQuery, - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, _context: Context, ): Promise> { this.assertOpen(); @@ -544,9 +544,6 @@ export class SqliteStorage implements Storage { if (action.create === undefined && existing === undefined) throw new Error(`Unknown document: ${id}`); if (action.create !== undefined && existing !== undefined) throw new Error(`Document ${id} already exists`); if (existing?.retiredAt !== undefined) throw new Error(`Document ${id} is retired`); - if (action.create !== undefined && action.content?.kind !== "base") { - throw new Error(`Document ${id} creation requires a base`); - } if (action.content?.kind === "delta") { const previous = getRow<{ readonly version: number }>( this.db.prepare( diff --git a/packages/durable/src/testing/storage-benchmark.ts b/packages/durable/src/testing/storage-benchmark.ts index 92904e924..b6e462ca0 100644 --- a/packages/durable/src/testing/storage-benchmark.ts +++ b/packages/durable/src/testing/storage-benchmark.ts @@ -280,7 +280,7 @@ export const STORAGE_READ_BENCHMARKS: readonly StorageReadBenchmark[] = [ name: "entry page scan (100)", async run(storage) { return ( - await storage.scanEntries({ conversationId: ROOT_CONVERSATION_ID }, undefined, 100, BACKGROUND_CONTEXT) + await storage.scanEntries({ conversationId: ROOT_CONVERSATION_ID }, 100, undefined, BACKGROUND_CONTEXT) ).items.length; }, expected: () => 100, @@ -291,8 +291,8 @@ export const STORAGE_READ_BENCHMARKS: readonly StorageReadBenchmark[] = [ return ( await storage.scanTasks( { kind: "benchmark.filtered", status: "pending", background: true }, - undefined, 50, + undefined, BACKGROUND_CONTEXT, ) ).items.length; @@ -349,8 +349,8 @@ export const STORAGE_READ_BENCHMARKS: readonly StorageReadBenchmark[] = [ return ( await storage.scanEntries( { conversationId: dataset.deepestConversationId }, - undefined, 100, + undefined, BACKGROUND_CONTEXT, ) ).items.length; diff --git a/packages/durable/src/testing/storage-conformance.ts b/packages/durable/src/testing/storage-conformance.ts index 56d49b650..cc7d651b2 100644 --- a/packages/durable/src/testing/storage-conformance.ts +++ b/packages/durable/src/testing/storage-conformance.ts @@ -245,7 +245,7 @@ export function createStorageConformance(options: StorageConformanceOptions): re ); expect( - (await storage.scanEntries({ conversationId: rootId }, undefined, 10, context)).items.map(({ id }) => id), + (await storage.scanEntries({ conversationId: rootId }, 10, undefined, context)).items.map(({ id }) => id), ).toEqual([30, 20, 10]); expect((await storage.findLatestHeadMarker(rootId, undefined, context))?.id).toBe(20); }), @@ -264,11 +264,11 @@ export function createStorageConformance(options: StorageConformanceOptions): re context, ); - const first = await storage.scanEntries({ conversationId: rootId }, undefined, 2, context); + const first = await storage.scanEntries({ conversationId: rootId }, 2, undefined, context); expect(first.items.map(({ id }) => id)).toEqual([newestId, middleId]); const appendedId = await storage.mintId(); await storage.commit([{ type: "entry", value: entry(appendedId, rootId) }], context); - const second = await storage.scanEntries({ conversationId: rootId }, first.next, 2, context); + const second = await storage.scanEntries({ conversationId: rootId }, 2, first.next, context); expect(second.items.map(({ id }) => id)).toEqual([oldestId]); expect(second.next).toBeUndefined(); }), @@ -285,11 +285,11 @@ export function createStorageConformance(options: StorageConformanceOptions): re context, ); - const first = await storage.scanConversations(undefined, 2, context); + const first = await storage.scanConversations(2, undefined, context); expect(first.items.map(({ id }) => id)).toEqual([rootId, secondId]); expect(first.next).toBeDefined(); const roundTrippedCursor = JSON.parse(JSON.stringify(first.next)) as NonNullable; - const second = await storage.scanConversations(roundTrippedCursor, 2, context); + const second = await storage.scanConversations(2, roundTrippedCursor, context); expect(second.items.map(({ id }) => id)).toEqual([thirdId]); expect(second.next).toBeUndefined(); }), @@ -356,11 +356,11 @@ export function createStorageConformance(options: StorageConformanceOptions): re const childExcludedLater = await storage.mintId(); await storage.commit([{ type: "entry", value: entry(childExcludedLater, childId) }], context); - const first = await storage.scanEntries({ conversationId: grandchildId }, undefined, 2, context); + const first = await storage.scanEntries({ conversationId: grandchildId }, 2, undefined, context); expect(first.items.map(({ id }) => id)).toEqual([grandchildTail, grandchildHead]); - const second = await storage.scanEntries({ conversationId: grandchildId }, first.next, 2, context); + const second = await storage.scanEntries({ conversationId: grandchildId }, 2, first.next, context); expect(second.items.map(({ id }) => id)).toEqual([childForkPoint, rootForkPoint]); - const third = await storage.scanEntries({ conversationId: grandchildId }, second.next, 2, context); + const third = await storage.scanEntries({ conversationId: grandchildId }, 2, second.next, context); expect(third.items.map(({ id }) => id)).toEqual([rootFirst]); expect(third.next).toBeUndefined(); @@ -374,16 +374,16 @@ export function createStorageConformance(options: StorageConformanceOptions): re const activeFirst = await storage.scanEntries( { conversationId: grandchildId, minEntryId: currentMarker?.head }, - undefined, 1, + undefined, context, ); expect(activeFirst.items.map(({ id }) => id)).toEqual([grandchildTail]); expect(activeFirst.next).toBeDefined(); const activeSecond = await storage.scanEntries( { conversationId: grandchildId, minEntryId: currentMarker?.head }, - activeFirst.next, 1, + activeFirst.next, context, ); expect(activeSecond.items.map(({ id }) => id)).toEqual([grandchildHead]); @@ -397,8 +397,8 @@ export function createStorageConformance(options: StorageConformanceOptions): re minEntryId: historicalMarker?.head, maxEntryId: childForkPoint, }, - undefined, 10, + undefined, context, ) ).items.map(({ id }) => id), @@ -412,7 +412,7 @@ export function createStorageConformance(options: StorageConformanceOptions): re expect((await storage.entry(grandchildHead, context))?.commitSeq).toBe(grandchildEntriesSeq); expect((await storage.entry(grandchildTail, context))?.commitSeq).toBe(grandchildEntriesSeq); expect(await storage.entry(999_999, context)).toBeUndefined(); - await expect(storage.scanEntries({ conversationId: 999_999 }, undefined, 10, context)).rejects.toThrow( + await expect(storage.scanEntries({ conversationId: 999_999 }, 10, undefined, context)).rejects.toThrow( "Unknown conversation", ); }), @@ -455,17 +455,17 @@ export function createStorageConformance(options: StorageConformanceOptions): re await storage.commit([{ type: "task", value: terminal }], context); expect(await storage.task(firstId, context)).toEqual(terminal); - const pendingPage = await storage.scanTasks({ status: "pending" }, undefined, 1, context); + const pendingPage = await storage.scanTasks({ status: "pending" }, 1, undefined, context); expect(pendingPage.items.map(({ id }) => id)).toEqual([secondId]); expect(pendingPage.next).toBeDefined(); expect( - (await storage.scanTasks({ status: "pending" }, pendingPage.next, 1, context)).items.map(({ id }) => id), + (await storage.scanTasks({ status: "pending" }, 1, pendingPage.next, context)).items.map(({ id }) => id), ).toEqual([thirdId]); expect( - (await storage.scanTasks({ status: "terminal", abortRequested: true }, undefined, 10, context)).items, + (await storage.scanTasks({ status: "terminal", abortRequested: true }, 10, undefined, context)).items, ).toEqual([terminal]); expect( - (await storage.scanTasks({ background: true }, undefined, 10, context)).items.map(({ id }) => id), + (await storage.scanTasks({ background: true }, 10, undefined, context)).items.map(({ id }) => id), ).toEqual([secondId]); }), @@ -671,12 +671,12 @@ export function createStorageConformance(options: StorageConformanceOptions): re }); expect( ( - await storage.scanDocuments({ scope: firstRecord.scope, at: changedAt }, undefined, 10, context) + await storage.scanDocuments({ scope: firstRecord.scope, at: changedAt }, 10, undefined, context) ).items.map(({ id }) => id), ).toEqual([firstId]); expect( ( - await storage.scanDocuments({ scope: firstRecord.scope, at: retiredAt }, undefined, 10, context) + await storage.scanDocuments({ scope: firstRecord.scope, at: retiredAt }, 10, undefined, context) ).items.map(({ id }) => id), ).toEqual([secondId]); expect(await storage.document(firstId, retiredAt, context)).toBeUndefined(); @@ -809,18 +809,18 @@ export function createStorageConformance(options: StorageConformanceOptions): re )?.id, ).toBe(firstId); expect( - (await storage.scanDocuments({ scope: { kind: "session" }, at: "current" }, undefined, 1, context)).items, + (await storage.scanDocuments({ scope: { kind: "session" }, at: "current" }, 1, undefined, context)).items, ).toHaveLength(1); const first = await storage.scanDocuments( { scope: { kind: "session" }, at: "current" }, - undefined, 1, + undefined, context, ); const second = await storage.scanDocuments( { scope: { kind: "session" }, at: "current" }, - first.next, 1, + first.next, context, ); expect([...first.items, ...second.items].map(({ id }) => id)).toEqual([firstId, secondId]); @@ -828,8 +828,8 @@ export function createStorageConformance(options: StorageConformanceOptions): re ( await storage.scanDocuments( { scope: { kind: "conversation", conversationId: rootId }, at: "current" }, - undefined, 10, + undefined, context, ) ).items.map(({ id }) => id), @@ -851,8 +851,8 @@ export function createStorageConformance(options: StorageConformanceOptions): re ( await storage.scanDocuments( { scope: { kind: "task", taskId }, at: "current", kind: "task.cache" }, - undefined, 10, + undefined, context, ) ).items.map(({ id }) => id), @@ -984,7 +984,7 @@ export function createStorageConformance(options: StorageConformanceOptions): re ).rejects.toThrow("already has a current incarnation"); expect(await storage.task(taskId, context)).toEqual(task); - expect((await storage.scanTasks({ status: "pending" }, undefined, 10, context)).items).toEqual([task]); + expect((await storage.scanTasks({ status: "pending" }, 10, undefined, 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(); @@ -1065,10 +1065,10 @@ export function createStorageConformance(options: StorageConformanceOptions): re context, ); - expect((await storage.scanTasks({ kind: first }, undefined, 10, context)).items.map(({ id }) => id)).toEqual([ + expect((await storage.scanTasks({ kind: first }, 10, undefined, context)).items.map(({ id }) => id)).toEqual([ firstTaskId, ]); - expect((await storage.scanTasks({ kind: second }, undefined, 10, context)).items.map(({ id }) => id)).toEqual([ + expect((await storage.scanTasks({ kind: second }, 10, undefined, context)).items.map(({ id }) => id)).toEqual([ secondTaskId, ]); expect((await storage.task(firstTaskId, context))?.kind).toBe(first); @@ -1099,8 +1099,8 @@ export function createStorageConformance(options: StorageConformanceOptions): re ( await storage.scanDocuments( { scope: { kind: "session" }, at: "current", kind: first }, - undefined, 10, + undefined, context, ) ).items.map(({ id }) => id), diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index 0005f8713..d41d14bba 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -1,4 +1,4 @@ -import type { Context, JsonValue } from "@earendil-works/chord"; +import type { Context, Draft, JsonValue } from "@earendil-works/chord"; import type { Op } from "@earendil-works/chord/delta"; import type { Message } from "@earendil-works/pi-ai"; @@ -14,6 +14,127 @@ export type Seq = number; /** The root conversation always uses this reserved ID. */ export const ROOT_CONVERSATION_ID: Id = 1; +/** Conversation document that retains only its current state. */ +export type LatestConversationSemantics = { + readonly scope: "conversation"; + readonly history: "latest"; + readonly fork: "current" | "initial"; +}; + +/** Conversation document whose history remains addressable for as-of reads. */ +export type RewindableConversationSemantics = { + readonly scope: "conversation"; + readonly history: "rewindable"; + readonly fork: "asOf" | "current" | "initial"; +}; + +/** Ownership and lifetime of a document; only conversation documents declare history and fork behavior. */ +export type DocumentSemantics = + | { readonly scope: "session"; readonly history?: never; readonly fork?: never } + | LatestConversationSemantics + | RewindableConversationSemantics + | { readonly scope: "task"; readonly history?: never; readonly fork?: never }; + +/** Definition fields shared by singleton documents and document families. */ +export type CommonDocDefinition = { + /** Stable persisted kind; part of the public protocol. */ + readonly kind: string; + /** Positive integer version of the stored value shape. */ + readonly version: number; + initial(): T; + migrate?(value: JsonObject, fromVersion: number): T; + checkpointWhen?(value: Readonly, ops: readonly Op[]): boolean; +}; + +/** Singleton document definition. */ +export type DocDefinition = CommonDocDefinition & DocumentSemantics; + +/** Keyed document family definition; `initial(seed)` runs only when a member is absent. */ +export type DocFamilyDefinition = Omit, "initial"> & + DocumentSemantics & { + readonly family: true; + initial(seed: I): T; + }; + +declare const docType: unique symbol; + +/** Typed singleton document token passed explicitly to typed access. */ +export interface DocToken> { + readonly definition: D; + readonly [docType]?: T; +} + +/** Typed document family token passed explicitly to typed access. */ +export interface DocFamilyToken> { + readonly definition: D; + readonly [docType]?: T; +} + +export type SessionDocToken = DocToken & { readonly scope: "session" }>; +export type ConversationDocToken = DocToken< + T, + CommonDocDefinition & (LatestConversationSemantics | RewindableConversationSemantics) +>; +export type RewindableConversationDocToken = DocToken< + T, + CommonDocDefinition & RewindableConversationSemantics +>; +export type TaskDocToken = DocToken & { readonly scope: "task" }>; + +export type SessionDocFamilyToken = DocFamilyToken< + T, + I, + DocFamilyDefinition & { readonly scope: "session" } +>; +export type ConversationDocFamilyToken = DocFamilyToken< + T, + I, + DocFamilyDefinition & (LatestConversationSemantics | RewindableConversationSemantics) +>; +export type RewindableConversationDocFamilyToken = DocFamilyToken< + T, + I, + DocFamilyDefinition & RewindableConversationSemantics +>; +export type TaskDocFamilyToken = DocFamilyToken< + T, + I, + DocFamilyDefinition & { readonly scope: "task" } +>; + +declare const taskResultType: unique symbol; + +/** Task definition fields currently supported by Session task creation. */ +export type TaskDefinition = { + /** Registered task kind persisted in `TaskRecord.kind`. */ + readonly name: string; + /** Definition version persisted with live input and checkpoints. */ + readonly version: number; + /** First durable checkpoint for a newly created task. */ + initial(input: I): S; + readonly hooks?: H; + /** Type-only result marker until phase handlers commit typed outcomes. */ + readonly [taskResultType]?: R; +}; + +/** Typed executable task definition. */ +export interface Task { + readonly definition: TaskDefinition; +} + +/** Task ID carrying its result type for typed waits. */ +export type TaskRef = { readonly id: Id; readonly [taskResultType]?: R }; + +/** Creation options for a durable task. */ +export type TaskOptions = { + /** Owning conversation; required for Session commits that are not bound to a conversation. */ + readonly conversationId?: Id; + /** Tasks that must be terminal before ordinary execution may begin. */ + readonly after?: readonly Id[]; + /** Excluded from ordinary idle waits and conversation aborts. */ + readonly background?: boolean; +}; + /** Immutable identity, history ancestry, and task ownership of a transcript scope. */ export type ConversationRecord = { readonly id: Id; @@ -379,6 +500,108 @@ export type StorageWrite = } | { readonly type: "document.retire"; readonly id: Id }; +/** + * Transaction surface of one Session commit callback. + * Table reads and creation results are trusted immutable values and may be shared with internal commit state. + */ +export interface Tx { + conversation(id: Id): Promise; + entry(id: Id): Promise; + task(id: Id): Promise | undefined>; + scanConversations(limit: number, cursor?: Cursor): Promise>; + scanEntries(query: EntryQuery, limit: number, cursor?: Cursor): Promise>; + scanTasks( + query: TaskQuery, + limit: number, + cursor?: Cursor, + ): Promise, Cursor>>; + + /** Returned records are Session-owned immutable values and may be shared with commit listeners. */ + createConversation(value: Omit): Promise; + /** Returned records are Session-owned immutable values and may be shared with commit listeners. */ + appendEntry(conversationId: Id, value: EntryDraft): Promise; + createTask( + task: Task, + input: I, + options?: TaskOptions, + ): Promise>; + /** Replace one task record completely. */ + setTask(value: TaskRecord): void; + + doc(token: SessionDocToken): Promise>; + doc(token: ConversationDocToken, conversationId: Id): Promise>; + doc(token: TaskDocToken, taskId: Id): Promise>; + doc( + token: SessionDocFamilyToken, + key: string, + seed: I, + ): Promise>; + doc( + token: ConversationDocFamilyToken, + conversationId: Id, + key: string, + seed: I, + ): Promise>; + doc( + token: TaskDocFamilyToken, + taskId: Id, + key: string, + seed: I, + ): Promise>; + + retireDoc(token: SessionDocToken): Promise; + retireDoc(token: ConversationDocToken, conversationId: Id): Promise; + retireDoc(token: TaskDocToken, taskId: Id): Promise; + retireDoc(token: SessionDocFamilyToken, key: string): Promise; + retireDoc( + token: ConversationDocFamilyToken, + conversationId: Id, + key: string, + ): Promise; + retireDoc( + token: TaskDocFamilyToken, + taskId: Id, + key: string, + ): Promise; +} + +/** Owner of one mutation line, its records, and its tracked documents. */ +export interface Session { + /** Run one atomic transaction on the Session mutation line. */ + commit(change: (tx: Tx) => T | Promise, context: Context): Promise; + /** Seal admission, settle admitted commits, then close storage. */ + close(context: Context): Promise; + + snapshot(token: SessionDocToken, context: Context): Promise | undefined>; + snapshot( + token: ConversationDocToken, + conversationId: Id, + context: Context, + ): Promise | undefined>; + snapshot( + token: TaskDocToken, + taskId: Id, + context: Context, + ): Promise | undefined>; + snapshot( + token: SessionDocFamilyToken, + key: string, + context: Context, + ): Promise | undefined>; + snapshot( + token: ConversationDocFamilyToken, + conversationId: Id, + key: string, + context: Context, + ): Promise | undefined>; + snapshot( + token: TaskDocFamilyToken, + taskId: Id, + key: string, + context: Context, + ): Promise | undefined>; +} + /** * Atomic persistence boundary for Session records. * @@ -401,8 +624,8 @@ export interface Storage { /** Scan conversations in ascending ID order. */ scanConversations( - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, context: Context, ): Promise>; @@ -422,8 +645,8 @@ export interface Storage { /** Scan the inclusive visible range newest-first, returning at most `limit` entries. */ scanEntries( query: EntryQuery, - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, context: Context, ): Promise>; @@ -433,8 +656,8 @@ export interface Storage { /** Scan task records matching every supplied filter. */ scanTasks( query: TaskQuery, - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, context: Context, ): Promise, Cursor>>; @@ -453,8 +676,8 @@ export interface Storage { /** Scan incarnations alive in one exact scope at the selected point. */ scanDocuments( query: DocumentQuery, - cursor: Cursor | undefined, limit: number, + cursor: Cursor | undefined, context: Context, ): Promise>; diff --git a/packages/durable/test/jsonl-storage.test.ts b/packages/durable/test/jsonl-storage.test.ts index f420311cd..13d8a74b3 100644 --- a/packages/durable/test/jsonl-storage.test.ts +++ b/packages/durable/test/jsonl-storage.test.ts @@ -66,24 +66,24 @@ class ReopeningStorage implements Storage { mintId: Storage["mintId"] = () => this.current.mintId(); conversation: Storage["conversation"] = (id, readContext) => this.current.conversation(id, readContext); - scanConversations: Storage["scanConversations"] = (cursor, limit, readContext) => - this.current.scanConversations(cursor, limit, readContext); + scanConversations: Storage["scanConversations"] = (limit, cursor, readContext) => + this.current.scanConversations(limit, cursor, readContext); entry: Storage["entry"] = (id, readContext) => this.current.entry(id, readContext); findLatestHeadMarker: Storage["findLatestHeadMarker"] = (conversationId, at, readContext) => this.current.findLatestHeadMarker(conversationId, at, readContext); - scanEntries: Storage["scanEntries"] = (query, cursor, limit, readContext) => - this.current.scanEntries(query, cursor, limit, readContext); + scanEntries: Storage["scanEntries"] = (query, limit, cursor, readContext) => + this.current.scanEntries(query, limit, cursor, readContext); task: Storage["task"] = (id, readContext) => this.current.task(id, readContext); - scanTasks: Storage["scanTasks"] = (query, cursor, limit, readContext) => - this.current.scanTasks(query, cursor, limit, readContext); + scanTasks: Storage["scanTasks"] = (query, limit, cursor, readContext) => + this.current.scanTasks(query, limit, cursor, readContext); submission: Storage["submission"] = (id, readContext) => this.current.submission(id, readContext); submissionByRequest: Storage["submissionByRequest"] = (conversationId, requestId, readContext) => this.current.submissionByRequest(conversationId, requestId, readContext); findDocument: Storage["findDocument"] = (address, at, readContext) => this.current.findDocument(address, at, readContext); document: Storage["document"] = (id, at, readContext) => this.current.document(id, at, readContext); - scanDocuments: Storage["scanDocuments"] = (query, cursor, limit, readContext) => - this.current.scanDocuments(query, cursor, limit, readContext); + scanDocuments: Storage["scanDocuments"] = (query, limit, cursor, readContext) => + this.current.scanDocuments(query, limit, cursor, readContext); async close(closeContext: Context): Promise { if (this.closed) return; diff --git a/packages/durable/test/session-definitions.test.ts b/packages/durable/test/session-definitions.test.ts new file mode 100644 index 000000000..249bbd0a8 --- /dev/null +++ b/packages/durable/test/session-definitions.test.ts @@ -0,0 +1,111 @@ +import type { Draft } from "@earendil-works/chord"; +import { + createSession, + defineDoc, + defineDocFamily, + type Id, + MemoryStorage, + type Session, + type Tx, +} from "@earendil-works/pi-durable"; +import { describe, expect, expectTypeOf, it } from "vitest"; +import { context } from "./session-support.ts"; + +type State = { value: number }; +const initial = (): State => ({ value: 0 }); + +const SessionDoc = defineDoc({ kind: "t.session", version: 1, scope: "session", initial }); +const LatestDoc = defineDoc({ + kind: "t.latest", + version: 1, + scope: "conversation", + history: "latest", + fork: "current", + initial, +}); +const RewindableDoc = defineDoc({ + kind: "t.rewindable", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial, +}); +const TaskDoc = defineDoc({ kind: "t.task", version: 1, scope: "task", initial }); +const SessionFamily = defineDocFamily({ + kind: "t.session-family", + version: 1, + family: true, + scope: "session", + initial: (seed) => ({ value: seed }), +}); +const ConversationFamily = defineDocFamily({ + kind: "t.conversation-family", + version: 1, + family: true, + scope: "conversation", + history: "rewindable", + fork: "initial", + initial: (seed) => ({ value: seed }), +}); +const TaskFamily = defineDocFamily({ + kind: "t.task-family", + version: 1, + family: true, + scope: "task", + initial: (seed) => ({ value: seed }), +}); + +describe("document definitions", () => { + it("validates persisted version semantics", () => { + expect(() => defineDoc({ kind: "k", version: 0, scope: "session", initial })).toThrow("positive integer"); + expect(() => defineDoc({ kind: "k", version: 1.5, scope: "session", initial })).toThrow( + "positive integer", + ); + expect(defineDoc({ kind: "", version: 1, scope: "session", initial }).definition.kind).toBe(""); + expect(SessionDoc.definition.kind).toBe("t.session"); + }); + + it("types every owner, key, and seed overload", async () => { + const session: Session = createSession(new MemoryStorage()); + const check = async (tx: Tx, conversationId: Id, taskId: Id): Promise => { + expectTypeOf(await tx.doc(SessionDoc)).toEqualTypeOf>(); + expectTypeOf(await tx.doc(LatestDoc, conversationId)).toEqualTypeOf>(); + expectTypeOf(await tx.doc(RewindableDoc, conversationId)).toEqualTypeOf>(); + expectTypeOf(await tx.doc(TaskDoc, taskId)).toEqualTypeOf>(); + expectTypeOf(await tx.doc(SessionFamily, "k", 1)).toEqualTypeOf>(); + expectTypeOf(await tx.doc(ConversationFamily, conversationId, "k", 1)).toEqualTypeOf>(); + expectTypeOf(await tx.doc(TaskFamily, taskId, "k", 1)).toEqualTypeOf>(); + await tx.retireDoc(SessionDoc); + await tx.retireDoc(LatestDoc, conversationId); + await tx.retireDoc(TaskDoc, taskId); + await tx.retireDoc(SessionFamily, "k"); + await tx.retireDoc(ConversationFamily, conversationId, "k"); + await tx.retireDoc(TaskFamily, taskId, "k"); + + // @ts-expect-error Session documents take no owner + await tx.doc(SessionDoc, conversationId); + // @ts-expect-error conversation documents require a conversation ID + await tx.doc(LatestDoc); + // @ts-expect-error family access requires a creation seed + await tx.doc(SessionFamily, "k"); + // @ts-expect-error family seeds are typed + await tx.doc(SessionFamily, "k", "seed"); + // @ts-expect-error family retirement takes no seed + await tx.retireDoc(TaskFamily, taskId, "k", 1); + }; + expect(check).toBeTypeOf("function"); + + expectTypeOf(await session.snapshot(SessionDoc, context)).toEqualTypeOf | undefined>(); + expectTypeOf(await session.snapshot(LatestDoc, 1, context)).toEqualTypeOf | undefined>(); + expectTypeOf(await session.snapshot(TaskDoc, 1, context)).toEqualTypeOf | undefined>(); + expectTypeOf(await session.snapshot(SessionFamily, "k", context)).toEqualTypeOf | undefined>(); + expectTypeOf(await session.snapshot(ConversationFamily, 1, "k", context)).toEqualTypeOf< + Readonly | undefined + >(); + expectTypeOf(await session.snapshot(TaskFamily, 1, "k", context)).toEqualTypeOf | undefined>(); + // @ts-expect-error snapshots never take a creation seed + await session.snapshot(SessionFamily, "k", 1, context).catch(() => undefined); + await session.close(context); + }); +}); diff --git a/packages/durable/test/session-documents.test.ts b/packages/durable/test/session-documents.test.ts new file mode 100644 index 000000000..249bdf3fe --- /dev/null +++ b/packages/durable/test/session-documents.test.ts @@ -0,0 +1,651 @@ +import type { Draft } from "@earendil-works/chord"; +import { defineDoc, defineDocFamily, type Id, type JsonObject } from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { context, createConversation, documentChanges, flush, openTestSession } from "./session-support.ts"; + +type Live = { message?: string; items: string[]; nested: { count: number }; other: { label: string } }; + +let liveInitCount = 0; +const LiveDoc = defineDoc({ + kind: "test.live", + version: 1, + scope: "conversation", + history: "latest", + fork: "initial", + initial: () => { + liveInitCount++; + return { items: [], nested: { count: 0 }, other: { label: "x" } }; + }, +}); + +const RewindableLiveDoc = defineDoc({ + kind: "test.live", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ items: [], nested: { count: 0 }, other: { label: "x" } }), +}); + +const LiveDocV2 = defineDoc({ + kind: "test.live", + version: 2, + scope: "conversation", + history: "latest", + fork: "initial", + initial: () => ({ items: [], nested: { count: 0 }, other: { label: "x" } }), +}); + +type Counter = { count: number }; +const CounterDoc = defineDoc({ + kind: "test.counter", + version: 1, + scope: "session", + initial: () => ({ count: 0 }), +}); + +type Member = { seed: string; hits: number }; +const seeds: string[] = []; +const MemberDoc = defineDocFamily({ + kind: "test.member", + version: 1, + family: true, + scope: "session", + initial: (seed) => { + seeds.push(seed); + return { seed, hits: 0 }; + }, +}); + +async function setupLive(): Promise & { readonly conversationId: Id }> { + const harness = openTestSession(); + const conversationId = await createConversation(harness.session); + await harness.session.commit(async (tx) => { + const live = await tx.doc(LiveDoc, conversationId); + live.items.push("a", "b"); + }, context); + return { ...harness, conversationId }; +} + +describe("Session document transactions", () => { + it("creates an initial base on first access and adopts it after Storage success", async () => { + const { session, storage, publications } = openTestSession(); + const conversationId = await createConversation(session); + await session.commit(async (tx) => { + const live = await tx.doc(LiveDoc, conversationId); + live.message = "hello"; + }, context); + const writes = storage.commits.at(-1)!; + expect(writes).toHaveLength(1); + expect(writes[0]).toMatchObject({ + type: "document.create", + record: { + kind: "test.live", + scope: { kind: "conversation", conversationId }, + history: "latest", + fork: "initial", + }, + content: { kind: "base", version: 1, value: { message: "hello", items: [], nested: { count: 0 } } }, + }); + const snapshot = await session.snapshot(LiveDoc, conversationId, context); + expect(snapshot).toEqual({ message: "hello", items: [], nested: { count: 0 }, other: { label: "x" } }); + await flush(); + const publication = publications.at(-1)!; + const published = documentChanges(publication)[0]!; + expect(published.record.createdAt).toBe(publication.seq); + expect(published.value).toBe(snapshot); + expect(published.conversationId).toBe(conversationId); + const create = storage.admittedCommits.at(-1)!.find((write) => write.type === "document.create")!; + expect(create.content.value).toBe(published.value); + }); + + it("never creates on snapshot and returns undefined when absent", async () => { + const { session, storage } = openTestSession(); + const conversationId = await createConversation(session); + const before = storage.commits.length; + expect(await session.snapshot(LiveDoc, conversationId, context)).toBeUndefined(); + expect(await session.snapshot(CounterDoc, context)).toBeUndefined(); + expect(await session.snapshot(MemberDoc, "k", context)).toBeUndefined(); + expect(storage.commits.length).toBe(before); + expect(storage.mintCount).toBe(1); + }); + + it("returns shared immutable snapshots and keeps prior revisions stable", async () => { + const { session, conversationId } = await setupLive(); + const first = (await session.snapshot(LiveDoc, conversationId, context))!; + expect(await session.snapshot(LiveDoc, conversationId, context)).toBe(first); + await session.commit(async (tx) => { + const live = await tx.doc(LiveDoc, conversationId); + live.nested.count = 1; + }, context); + const second = (await session.snapshot(LiveDoc, conversationId, context))!; + expect(second).not.toBe(first); + expect(first.nested.count).toBe(0); + expect(second.nested.count).toBe(1); + // Unchanged subtrees are structurally shared between immutable revisions. + expect(second.items).toBe(first.items); + expect(second.other).toBe(first.other); + }); + + it("adopts by pointer swap and shares operation payloads with the published revision", async () => { + const { session, storage, publications, conversationId } = await setupLive(); + await session.commit(async (tx) => { + const live = await tx.doc(LiveDoc, conversationId); + live.other = { label: "y" }; + live.items.push("c"); + }, context); + await flush(); + const snapshot = (await session.snapshot(LiveDoc, conversationId, context))!; + const published = documentChanges(publications.at(-1)!)[0]!; + expect(published.value).toBe(snapshot); + const admitted = storage.admittedCommits.at(-1)!.find((write) => write.type === "document.change")!; + expect(admitted.content.kind).toBe("delta"); + if (admitted.content.kind !== "delta") throw new Error("Expected delta"); + expect(published.ops).toBe(admitted.content.ops); + const set = published.ops.find((op) => op[0] === "s"); + expect(set).toEqual(["s", ["other"], { label: "y" }]); + // Trusted immutability: the Session makes no second copy of operation payloads. + expect(set![2]).toBe(snapshot.other); + }); + + it("copies assigned values per placement", async () => { + const { session, conversationId } = await setupLive(); + const value = { label: "shared" }; + await session.commit(async (tx) => { + const live = await tx.doc(LiveDoc, conversationId); + live.other = value; + (live as Draft & { copy?: { label: string } }).copy = value; + value.label = "mutated"; + }, context); + const snapshot = (await session.snapshot(LiveDoc, conversationId, context)) as Live & { copy: { label: string } }; + expect(snapshot.other).toEqual({ label: "shared" }); + expect(snapshot.copy).toEqual({ label: "shared" }); + expect(snapshot.copy).not.toBe(snapshot.other); + expect(snapshot.other).not.toBe(value); + }); + + it("suppresses writes and publications for empty batches", async () => { + const { session, storage, publications, conversationId } = await setupLive(); + await flush(); + const commits = storage.commits.length; + const published = publications.length; + await session.commit(async (tx) => { + const live = await tx.doc(LiveDoc, conversationId); + live.nested.count = 0; + live.items.push("z"); + live.items.pop(); + }, context); + await flush(); + expect(storage.commits.length).toBe(commits); + expect(publications.length).toBe(published); + }); + + it("writes and publishes replayable nonempty structural no-ops", async () => { + const { session, storage, publications, conversationId } = await setupLive(); + const before = (await session.snapshot(LiveDoc, conversationId, context))!; + await session.commit(async (tx) => { + const live = await tx.doc(LiveDoc, conversationId); + const first = live.items.shift()!; + live.items.unshift(first); + }, context); + await flush(); + const writes = storage.commits.at(-1)!; + expect(writes[0]).toMatchObject({ type: "document.change", content: { kind: "delta" } }); + const after = (await session.snapshot(LiveDoc, conversationId, context))!; + expect(after).toEqual(before); + expect(after).not.toBe(before); + expect(documentChanges(publications.at(-1)!)[0]!.ops.length).toBeGreaterThan(0); + }); + + it("revokes escaped drafts when the callback settles", async () => { + const { session, conversationId } = await setupLive(); + let escaped: Draft | undefined; + let items: Draft | undefined; + await session.commit(async (tx) => { + escaped = await tx.doc(LiveDoc, conversationId); + items = escaped.items; + escaped.message = "inside"; + }, context); + expect(() => escaped!.message).toThrow(); + expect(() => { + escaped!.message = "outside"; + }).toThrow(); + expect(() => items!.length).toThrow(); + expect(() => items!.push("outside")).toThrow(); + expect((await session.snapshot(LiveDoc, conversationId, context))!.message).toBe("inside"); + + const returned = await session.commit((tx) => tx.doc(LiveDoc, conversationId), context); + expect(() => returned.message).toThrow(); + }); + + it("aborts every change when the callback fails", async () => { + const { session, storage, conversationId } = await setupLive(); + const before = (await session.snapshot(LiveDoc, conversationId, context))!; + const commits = storage.commits.length; + await expect( + session.commit(async (tx) => { + const live = await tx.doc(LiveDoc, conversationId); + const counter = await tx.doc(CounterDoc); + live.message = "lost"; + counter.count = 5; + throw new Error("callback failed"); + }, context), + ).rejects.toThrow("callback failed"); + expect(storage.commits.length).toBe(commits); + expect(await session.snapshot(LiveDoc, conversationId, context)).toBe(before); + expect(await session.snapshot(CounterDoc, context)).toBeUndefined(); + await session.commit(async (tx) => { + (await tx.doc(LiveDoc, conversationId)).message = "kept"; + }, context); + expect((await session.snapshot(LiveDoc, conversationId, context))!.message).toBe("kept"); + }); + + it("memoizes concurrent duplicate acquisition and initializes once", async () => { + const { session, storage } = openTestSession(); + const conversationId = await createConversation(session); + const initCount = liveInitCount; + const mints = storage.mintCount; + await session.commit(async (tx) => { + const [first, second] = await Promise.all([tx.doc(LiveDoc, conversationId), tx.doc(LiveDoc, conversationId)]); + expect(first).toBe(second); + expect(await tx.doc(LiveDoc, conversationId)).toBe(first); + first.message = "once"; + }, context); + expect(liveInitCount).toBe(initCount + 1); + expect(storage.mintCount).toBe(mints + 1); + expect(storage.commits.at(-1)!.filter((write) => write.type === "document.create")).toHaveLength(1); + }); + + it("uses the first family seed and ignores seeds for existing members", async () => { + const { session } = openTestSession(); + seeds.length = 0; + await session.commit(async (tx) => { + const first = await tx.doc(MemberDoc, "k", "first"); + const second = await tx.doc(MemberDoc, "k", "second"); + expect(second).toBe(first); + first.hits++; + }, context); + await session.commit(async (tx) => { + (await tx.doc(MemberDoc, "k", "third")).hits++; + (await tx.doc(MemberDoc, "other", "fourth")).hits++; + }, context); + expect(seeds).toEqual(["first", "fourth"]); + expect(await session.snapshot(MemberDoc, "k", context)).toEqual({ seed: "first", hits: 2 }); + expect(await session.snapshot(MemberDoc, "other", context)).toEqual({ seed: "fourth", hits: 1 }); + }); + + it("rejects a callback that succeeds with a pending acquisition and drains it", async () => { + const { session, storage, conversationId } = await setupLive(); + await session.unloadDocuments(); + const gate = storage.holdFindDocument(); + const commits = storage.commits.length; + let pending: Promise | undefined; + const commit = session.commit((tx) => { + pending = tx.doc(LiveDoc, conversationId); + }, context); + await gate.entered; + let settled = false; + void commit.catch(() => { + settled = true; + }); + await flush(); + // The line stays held until the late acquisition settles. + expect(settled).toBe(false); + gate.release(); + await expect(commit).rejects.toThrow("pending Tx operations"); + await expect(pending).rejects.toThrow("Transaction has settled"); + expect(storage.commits.length).toBe(commits); + await session.commit(async (tx) => { + (await tx.doc(LiveDoc, conversationId)).message = "after"; + }, context); + expect((await session.snapshot(LiveDoc, conversationId, context))!.message).toBe("after"); + }); + + it("does not initialize or mint for an absent acquisition that finishes after settlement", async () => { + const { session, storage } = openTestSession(); + let initialized = 0; + const LateDoc = defineDoc({ + kind: "test.late", + version: 1, + scope: "session", + initial: () => { + initialized++; + return { count: 0 }; + }, + }); + const gate = storage.holdFindDocument(); + const mints = storage.mintCount; + let pending: Promise | undefined; + const commit = session.commit((tx) => { + pending = tx.doc(LateDoc); + }, context); + await gate.entered; + gate.release(); + await expect(commit).rejects.toThrow("pending Tx operations"); + await expect(pending).rejects.toThrow("Transaction has settled"); + expect(initialized).toBe(0); + expect(storage.mintCount).toBe(mints); + }); + + it("rejects with the callback error when it fails with a pending acquisition", async () => { + const { session, storage, conversationId } = await setupLive(); + await session.unloadDocuments(); + const gate = storage.holdFindDocument(); + let pending: Promise | undefined; + const commit = session.commit((tx) => { + pending = tx.doc(LiveDoc, conversationId); + throw new Error("callback failed"); + }, context); + await gate.entered; + gate.release(); + await expect(commit).rejects.toThrow("callback failed"); + await expect(pending).rejects.toThrow("Transaction has settled"); + }); + + it("rejects Tx use after the callback settles", async () => { + const { session, conversationId } = await setupLive(); + let captured: Parameters[0]>[0] | undefined; + await session.commit((tx) => { + captured = tx; + }, context); + await expect(captured!.doc(LiveDoc, conversationId)).rejects.toThrow("Transaction has settled"); + await expect(captured!.conversation(conversationId)).rejects.toThrow("Transaction has settled"); + expect(() => captured!.setTask({} as never)).toThrow("Transaction has settled"); + }); + + it("rejects tokens whose semantics or version disagree with the stored incarnation", async () => { + const { session, conversationId } = await setupLive(); + await expect(session.snapshot(RewindableLiveDoc, conversationId, context)).rejects.toThrow( + "does not match the supplied definition semantics", + ); + await expect( + session.commit(async (tx) => { + await tx.doc(RewindableLiveDoc, conversationId); + }, context), + ).rejects.toThrow("does not match the supplied definition semantics"); + await expect( + session.commit(async (tx) => { + await tx.doc(LiveDocV2, conversationId); + }, context), + ).rejects.toThrow("requires migration from version 1"); + }); + + it("rejects non-JSON initializer values and draft placements before Storage admission", async () => { + const { session, storage } = openTestSession(); + const conversationId = await createConversation(session); + const DateDoc = defineDoc({ + kind: "test.date", + version: 1, + scope: "session", + initial: () => ({ at: new Date() }) as unknown as JsonObject, + }); + await expect( + session.commit(async (tx) => { + await tx.doc(DateDoc); + }, context), + ).rejects.toThrow("strict JSON"); + const commits = storage.commits.length; + await expect( + session.commit(async (tx) => { + const live = await tx.doc(LiveDoc, conversationId); + live.items.push(undefined as unknown as string); + }, context), + ).rejects.toThrow("strict JSON"); + expect(storage.commits.length).toBe(commits); + expect(await session.snapshot(LiveDoc, conversationId, context)).toBeUndefined(); + }); + + it("rolls back prepared documents when batch assembly fails", async () => { + const { session, storage, conversationId } = await setupLive(); + await session.commit(async (tx) => { + (await tx.doc(CounterDoc)).count = 1; + }, context); + const live = await session.snapshot(LiveDoc, conversationId, context); + const counter = await session.snapshot(CounterDoc, context); + const commits = storage.commits.length; + await expect( + session.commit(async (tx) => { + (await tx.doc(LiveDoc, conversationId)).message = "lost"; + (await tx.doc(CounterDoc)).count = 2; + // Replacing a missing task fails during assembly, after every change was prepared. + tx.setTask({ + id: 999, + conversationId, + kind: "missing", + version: 1, + input: null, + after: [], + background: false, + abortRequested: false, + state: { status: "pending", checkpoint: { phase: "start" } }, + }); + }, context), + ).rejects.toThrow("Task 999 does not exist"); + expect(storage.commits.length).toBe(commits); + expect(await session.snapshot(LiveDoc, conversationId, context)).toBe(live); + expect(await session.snapshot(CounterDoc, context)).toBe(counter); + await session.commit(async (tx) => { + (await tx.doc(LiveDoc, conversationId)).message = "next"; + (await tx.doc(CounterDoc)).count = 3; + }, context); + expect((await session.snapshot(CounterDoc, context))!.count).toBe(3); + }); + + it("poisons the Session after an uncertain Storage failure and publishes nothing", async () => { + const { session, storage, publications, conversationId } = await setupLive(); + await flush(); + const before = (await session.snapshot(LiveDoc, conversationId, context))!; + const published = publications.length; + storage.failNextCommit(new Error("disk vanished")); + await expect( + session.commit(async (tx) => { + (await tx.doc(LiveDoc, conversationId)).message = "uncertain"; + }, context), + ).rejects.toThrow("disk vanished"); + await flush(); + expect(publications.length).toBe(published); + expect(before.message).toBeUndefined(); + await expect(session.snapshot(LiveDoc, conversationId, context)).rejects.toThrow("poisoned"); + await expect(session.commit(() => undefined, context)).rejects.toThrow("poisoned"); + await session.close(context); + }); + + it("keeps the previous revision unchanged through Storage settlement", async () => { + const { session, storage, conversationId } = await setupLive(); + const before = (await session.snapshot(LiveDoc, conversationId, context))!; + const copy = structuredClone(before); + const gate = storage.holdCommits(); + const commit = session.commit(async (tx) => { + const live = await tx.doc(LiveDoc, conversationId); + live.items.push("c"); + live.nested.count = 9; + }, context); + await gate.entered; + expect(await session.snapshot(LiveDoc, conversationId, context)).toBe(before); + expect(before).toEqual(copy); + gate.release(); + await commit; + const after = (await session.snapshot(LiveDoc, conversationId, context))!; + expect(after.items).toEqual(["a", "b", "c"]); + expect(before).toEqual(copy); + }); + + it("retires documents and creates a new incarnation at the same address", async () => { + const { session, storage, publications, conversationId } = await setupLive(); + await flush(); + const oldId = documentChanges(publications.at(-1)!)[0]!.record.id; + let replacement: Draft | undefined; + await session.commit(async (tx) => { + const live = await tx.doc(LiveDoc, conversationId); + live.message = "final"; + await tx.retireDoc(LiveDoc, conversationId); + replacement = await tx.doc(LiveDoc, conversationId); + expect(replacement).not.toBe(live); + replacement.message = "new"; + }, context); + const writes = storage.commits.at(-1)!; + expect(writes).toHaveLength(3); + expect(writes).toContainEqual(expect.objectContaining({ type: "document.change", id: oldId })); + expect(writes).toContainEqual({ type: "document.retire", id: oldId }); + expect(writes).toContainEqual( + expect.objectContaining({ type: "document.create", record: expect.objectContaining({ kind: "test.live" }) }), + ); + await flush(); + const publication = publications.at(-1)!; + const [retired, created] = documentChanges(publication); + expect(retired).toMatchObject({ record: { id: oldId }, value: null, ops: [] }); + expect(retired!.record.retiredAt).toBe(publication.seq); + expect(created!.record.id).not.toBe(oldId); + expect(created!.record.createdAt).toBe(publication.seq); + expect(created).toMatchObject({ value: { message: "new", items: [] }, ops: [] }); + expect(await session.snapshot(LiveDoc, conversationId, context)).toBe(created!.value); + + await session.commit((tx) => tx.retireDoc(LiveDoc, conversationId), context); + expect(await session.snapshot(LiveDoc, conversationId, context)).toBeUndefined(); + await session.unloadDocuments(); + expect(await session.snapshot(LiveDoc, conversationId, context)).toBeUndefined(); + // Retiring an absent address is a no-op. + const commits = storage.commits.length; + await session.commit((tx) => tx.retireDoc(LiveDoc, conversationId), context); + expect(storage.commits.length).toBe(commits); + }); + + it("retires without acquisition and recreates both existing and absent addresses", async () => { + const { session, storage, publications, conversationId } = await setupLive(); + await flush(); + const oldId = documentChanges(publications.at(-1)!)[0]!.record.id; + await session.unloadDocuments(); + await session.commit(async (tx) => { + const retired = tx.retireDoc(LiveDoc, conversationId); + const replacement = tx.doc(LiveDoc, conversationId); + const [live] = await Promise.all([replacement, retired]); + live.message = "replacement"; + }, context); + const existingWrites = storage.commits.at(-1)!; + expect(existingWrites).toHaveLength(2); + expect(existingWrites).toContainEqual({ type: "document.retire", id: oldId }); + expect(existingWrites).toContainEqual( + expect.objectContaining({ type: "document.create", record: expect.objectContaining({ kind: "test.live" }) }), + ); + + await session.commit(async (tx) => { + const retired = tx.retireDoc(MemberDoc, "absent"); + const replacement = tx.doc(MemberDoc, "absent", "seed"); + const [member] = await Promise.all([replacement, retired]); + member.hits = 1; + }, context); + const absentWrites = storage.commits.at(-1)!; + expect(absentWrites.filter((write) => write.type === "document.retire")).toHaveLength(0); + expect(absentWrites.filter((write) => write.type === "document.create")).toHaveLength(1); + expect(await session.snapshot(MemberDoc, "absent", context)).toEqual({ seed: "seed", hits: 1 }); + }); + + it("retires the existing incarnation when retirement races a pending acquisition", async () => { + const { session, storage, publications, conversationId } = await setupLive(); + await flush(); + const oldId = documentChanges(publications.at(-1)!)[0]!.record.id; + await session.commit(async (tx) => { + const acquired = tx.doc(LiveDoc, conversationId); + const retired = tx.retireDoc(LiveDoc, conversationId); + (await acquired).message = "final"; + await retired; + }, context); + const firstWrites = storage.commits.at(-1)!; + expect(firstWrites).toHaveLength(2); + expect(firstWrites).toContainEqual({ + type: "document.change", + id: oldId, + content: expect.objectContaining({ kind: "delta" }), + }); + expect(firstWrites).toContainEqual({ type: "document.retire", id: oldId }); + expect(await session.snapshot(LiveDoc, conversationId, context)).toBeUndefined(); + + await session.commit(async (tx) => { + (await tx.doc(LiveDoc, conversationId)).message = "second"; + }, context); + await flush(); + const secondId = documentChanges(publications.at(-1)!)[0]!.record.id; + await session.commit(async (tx) => { + const acquired = tx.doc(LiveDoc, conversationId); + const retired = tx.retireDoc(LiveDoc, conversationId); + const recreated = tx.doc(LiveDoc, conversationId); + const [old, fresh] = await Promise.all([acquired, recreated, retired]); + expect(fresh).not.toBe(old); + fresh!.message = "third"; + }, context); + const writes = storage.commits.at(-1)!; + expect(writes).toHaveLength(2); + expect(writes).toContainEqual({ type: "document.retire", id: secondId }); + expect(writes).toContainEqual( + expect.objectContaining({ type: "document.create", record: expect.objectContaining({ kind: "test.live" }) }), + ); + expect((await session.snapshot(LiveDoc, conversationId, context))!.message).toBe("third"); + }); + + it("reloads an unloaded document from Storage", async () => { + const { session, conversationId } = await setupLive(); + const loaded = (await session.snapshot(LiveDoc, conversationId, context))!; + await session.unloadDocuments(); + const reloaded = (await session.snapshot(LiveDoc, conversationId, context))!; + expect(reloaded).not.toBe(loaded); + expect(reloaded).toEqual(loaded); + await session.commit(async (tx) => { + (await tx.doc(LiveDoc, conversationId)).items.push("c"); + }, context); + await session.unloadDocuments(); + expect((await session.snapshot(LiveDoc, conversationId, context))!.items).toEqual(["a", "b", "c"]); + }); + + it("delivers publications off the mutation line so listeners can start a nested commit", async () => { + const { session, publications, conversationId } = await setupLive(); + await flush(); + const published = publications.length; + let listenerContext: typeof context | undefined; + const nested = new Promise((resolve, reject) => { + const unsubscribe = session.subscribeCommits((_publication, deliveredContext) => { + unsubscribe(); + listenerContext = deliveredContext; + void session + .commit(async (tx) => { + (await tx.doc(CounterDoc)).count = 1; + }, context) + .then(resolve, reject); + }); + }); + const result = await session.commit(async (tx) => { + (await tx.doc(LiveDoc, conversationId)).message = "m"; + return "done"; + }, context); + expect(result).toBe("done"); + await nested; + expect(listenerContext).toBe(context); + expect(await session.snapshot(CounterDoc, context)).toEqual({ count: 1 }); + await flush(); + expect(publications.length).toBe(published + 2); + }); + + it("settles admitted commits before close and rejects later admission", async () => { + const { session, storage, conversationId } = await setupLive(); + const gate = storage.holdCommits(); + const commit = session.commit(async (tx) => { + (await tx.doc(LiveDoc, conversationId)).message = "admitted"; + }, context); + await gate.entered; + const queued = session.commit(async (tx) => { + (await tx.doc(LiveDoc, conversationId)).message = "queued"; + }, context); + const admittedSnapshot = session.snapshot(MemberDoc, "absent", context); + const closed = session.close(context); + await expect(session.commit(() => undefined, context)).rejects.toThrow("closed"); + await expect(session.snapshot(LiveDoc, conversationId, context)).rejects.toThrow("closed"); + gate.release(); + await commit; + await queued; + expect(await admittedSnapshot).toBeUndefined(); + await closed; + const stored = await storage + .findDocument({ kind: "test.live", scope: { kind: "conversation", conversationId } }, "current", context) + .catch((error: unknown) => error); + expect(stored).toBeInstanceOf(Error); + }); +}); diff --git a/packages/durable/test/session-support.ts b/packages/durable/test/session-support.ts new file mode 100644 index 000000000..e0411de1c --- /dev/null +++ b/packages/durable/test/session-support.ts @@ -0,0 +1,129 @@ +import type { Context } from "@earendil-works/chord"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { + type DocumentAddress, + type DocumentPoint, + type DocumentRecord, + type Id, + MemoryStorage, + type Seq, + type StorageWrite, +} from "@earendil-works/pi-durable"; +import type { CommitPublication } from "../src/session/publications.ts"; +import { SessionKernel } from "../src/session/session.ts"; +import type { DocumentCommitChange } from "../src/session/transaction.ts"; + +export const context: Context = BACKGROUND_CONTEXT; + +type Deferred = { readonly promise: Promise; readonly resolve: () => void }; + +function deferred(): Deferred { + let resolve!: () => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +/** A gate that holds calls until released and reports when the first held call arrives. */ +export type Gate = { + readonly entered: Promise; + release(): void; +}; + +/** Memory storage with observable commits, held calls, and injected commit failures. */ +export class ControlledStorage extends MemoryStorage { + /** Exact borrowed batches admitted by Session. */ + readonly admittedCommits: (readonly StorageWrite[])[] = []; + /** Detached batches for value assertions. */ + readonly commits: (readonly StorageWrite[])[] = []; + mintCount = 0; + #commitGate: { gate: Deferred; entered: Deferred } | undefined; + #findGate: { gate: Deferred; entered: Deferred } | undefined; + #commitFailure: Error | undefined; + + holdCommits(): Gate { + const held = { gate: deferred(), entered: deferred() }; + this.#commitGate = held; + return { entered: held.entered.promise, release: () => this.#release("commit", held) }; + } + + holdFindDocument(): Gate { + const held = { gate: deferred(), entered: deferred() }; + this.#findGate = held; + return { entered: held.entered.promise, release: () => this.#release("find", held) }; + } + + failNextCommit(error: Error): void { + this.#commitFailure = error; + } + + #release(kind: "commit" | "find", held: { gate: Deferred }): void { + if (kind === "commit" && this.#commitGate === held) this.#commitGate = undefined; + if (kind === "find" && this.#findGate === held) this.#findGate = undefined; + held.gate.resolve(); + } + + override async commit(writes: readonly StorageWrite[], commitContext: Context): Promise { + this.admittedCommits.push(writes); + this.commits.push(structuredClone(writes)); + const held = this.#commitGate; + if (held !== undefined) { + held.entered.resolve(); + await held.gate.promise; + } + const failure = this.#commitFailure; + if (failure !== undefined) { + this.#commitFailure = undefined; + throw failure; + } + return super.commit(writes, commitContext); + } + + override mintId(): Promise { + this.mintCount++; + return super.mintId(); + } + + override async findDocument( + address: DocumentAddress, + at: DocumentPoint, + callContext: Context, + ): Promise { + const held = this.#findGate; + if (held !== undefined) { + held.entered.resolve(); + await held.gate.promise; + } + return super.findDocument(address, at, callContext); + } +} + +/** Session kernel plus its controlled storage and every committed publication. */ +export function openTestSession(): { + readonly storage: ControlledStorage; + readonly session: SessionKernel; + readonly publications: CommitPublication[]; +} { + const storage = new ControlledStorage(); + const session = new SessionKernel(storage); + const publications: CommitPublication[] = []; + session.subscribeCommits((publication) => { + publications.push(publication); + }); + return { storage, session, publications }; +} + +export function documentChanges(publication: CommitPublication): readonly DocumentCommitChange[] { + return publication.changes.filter((change): change is DocumentCommitChange => change.type === "document"); +} + +/** Create one conversation and return its ID. */ +export async function createConversation(session: SessionKernel): Promise { + return session.commit(async (tx) => (await tx.createConversation({})).id, context); +} + +/** Resolve after pending microtasks and one macrotask turn. */ +export function flush(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +} diff --git a/packages/durable/test/session-tables.test.ts b/packages/durable/test/session-tables.test.ts new file mode 100644 index 000000000..de9fa99e2 --- /dev/null +++ b/packages/durable/test/session-tables.test.ts @@ -0,0 +1,328 @@ +import type { JsonValue } from "@earendil-works/chord"; +import { + defineDoc, + defineDocFamily, + type Id, + ReadAfterWrite, + type Task, + type TaskRecord, +} from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { context, createConversation, documentChanges, flush, openTestSession } from "./session-support.ts"; + +type Checkpoint = { phase: "start" } | { phase: "next"; step: number }; +const WorkTask: Task<{ path: string }, Checkpoint, { ok: boolean }, object> = { + definition: { name: "test.work", version: 1, initial: () => ({ phase: "start" }) }, +}; + +type Progress = { lines: string[] }; +const ProgressDoc = defineDoc({ + kind: "test.progress", + version: 1, + scope: "task", + initial: () => ({ lines: [] }), +}); +const StepDoc = defineDocFamily({ + kind: "test.step", + version: 1, + family: true, + scope: "task", + initial: () => ({ lines: [] }), +}); + +type Notes = { text: string }; +const NotesDoc = defineDoc({ + kind: "test.notes", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ text: "" }), +}); + +type AnyTask = TaskRecord; + +function terminal(task: AnyTask): AnyTask { + return { + id: task.id, + conversationId: task.conversationId, + kind: task.kind, + version: task.version, + input: task.input, + after: task.after, + background: task.background, + abortRequested: task.abortRequested, + state: { status: "terminal", outcome: { status: "completed", result: { ok: true } } }, + }; +} + +async function createTask( + session: ReturnType["session"], + conversationId: Id, + withDocument = false, +): Promise { + return session.commit(async (tx) => { + const ref = await tx.createTask(WorkTask, { path: "a" }, { conversationId }); + if (withDocument) (await tx.doc(ProgressDoc, ref.id)).lines.push("started"); + return ref.id; + }, context); +} + +describe("Session transaction tables", () => { + it("allows table reads only before the first table write", async () => { + const { session } = openTestSession(); + const conversationId = await createConversation(session); + await session.commit(async (tx) => { + expect(await tx.conversation(conversationId)).toEqual({ id: conversationId }); + expect(await tx.scanConversations(1)).toEqual({ items: [{ id: conversationId }] }); + expect(await tx.scanTasks({ conversationId }, 10)).toEqual({ items: [] }); + expect(await tx.scanEntries({ conversationId }, 10)).toEqual({ items: [] }); + await tx.appendEntry(conversationId, { kind: "note" }); + await expect(tx.conversation(conversationId)).rejects.toBeInstanceOf(ReadAfterWrite); + await expect(tx.task(1)).rejects.toThrow("Tx.task() cannot read tables after the first table write"); + await expect(tx.entry(1)).rejects.toBeInstanceOf(ReadAfterWrite); + await expect(tx.scanConversations(10)).rejects.toBeInstanceOf(ReadAfterWrite); + await expect(tx.scanEntries({ conversationId }, 10)).rejects.toBeInstanceOf(ReadAfterWrite); + // Document access remains available after table writes. + (await tx.doc(NotesDoc, conversationId)).text = "after write"; + }, context); + expect(await session.snapshot(NotesDoc, conversationId, context)).toEqual({ text: "after write" }); + }); + + it("passes caller-selected limits and cursors through table scans", async () => { + const { session } = openTestSession(); + const ids = [ + await createConversation(session), + await createConversation(session), + await createConversation(session), + ]; + await session.commit(async (tx) => { + const first = await tx.scanConversations(2); + expect(first.items.map(({ id }) => id)).toEqual(ids.slice(0, 2)); + expect(first.next).toBeDefined(); + const second = await tx.scanConversations(2, first.next); + expect(second.items.map(({ id }) => id)).toEqual(ids.slice(2)); + expect(second.next).toBeUndefined(); + }, context); + }); + + it("treats synchronous setTask as the first table write", async () => { + const { session } = openTestSession(); + const conversationId = await createConversation(session); + const taskId = await createTask(session, conversationId); + await session.commit(async (tx) => { + const task = (await tx.task(taskId))!; + tx.setTask(task); + await expect(tx.task(taskId)).rejects.toBeInstanceOf(ReadAfterWrite); + }, context); + }); + + it("creates conversations, entries, and tasks with minted IDs", async () => { + const { session, storage, publications } = openTestSession(); + const created = await session.commit(async (tx) => { + const conversation = await tx.createConversation({}); + const first = await tx.appendEntry(conversation.id, { kind: "note", data: "one" }); + const headed = await tx.appendEntry(conversation.id, { kind: "summary", head: "self" }); + const task = await tx.createTask( + WorkTask, + { path: "x" }, + { conversationId: conversation.id, background: true }, + ); + return { conversation, first, headed, task }; + }, context); + const ids = [created.conversation.id, created.first.id, created.headed.id, created.task.id]; + expect(new Set(ids).size).toBe(4); + expect(created.headed.head).toBe(created.headed.id); + expect(created.first).toEqual({ + id: created.first.id, + conversationId: created.conversation.id, + kind: "note", + data: "one", + }); + expect((await storage.entry(created.headed.id, context))!.entry).toEqual(created.headed); + expect(await storage.task(created.task.id, context)).toEqual({ + id: created.task.id, + conversationId: created.conversation.id, + kind: "test.work", + version: 1, + input: { path: "x" }, + after: [], + background: true, + abortRequested: false, + state: { status: "pending", checkpoint: { phase: "start" } }, + }); + await flush(); + const changes = publications.at(-1)!.changes; + expect(changes).toHaveLength(4); + const admitted = storage.admittedCommits.at(-1)!; + for (const change of changes) expect(admitted.some((write) => write === change)).toBe(true); + expect(changes.map((change) => change.type)).toEqual( + expect.arrayContaining(["conversation", "entry", "entry", "task"]), + ); + expect(changes.find((change) => change.type === "conversation")!.value).toBe(created.conversation); + const entries = changes.filter((change) => change.type === "entry"); + expect(entries.map((change) => change.value)).toContain(created.first); + expect(entries.map((change) => change.value)).toContain(created.headed); + await expect(session.commit((tx) => tx.createTask(WorkTask, { path: "x" }), context)).rejects.toThrow( + "requires options.conversationId", + ); + await expect(session.commit((tx) => tx.appendEntry(12345, { kind: "note" }), context)).rejects.toThrow( + "Conversation 12345 does not exist", + ); + }); + + it("takes ownership of table JSON and rejects non-strict values", async () => { + const { session, storage } = openTestSession(); + const conversationId = await createConversation(session); + const payload = { nested: { value: 1 } }; + const entry = await session.commit(async (tx) => { + const created = await tx.appendEntry(conversationId, { kind: "data", data: payload }); + payload.nested.value = 2; + return created; + }, context); + expect((await storage.entry(entry.id, context))!.entry.data).toEqual({ nested: { value: 1 } }); + const omitted = await session.commit( + (tx) => tx.appendEntry(conversationId, { kind: "omitted", data: undefined }), + context, + ); + expect(Object.hasOwn(omitted, "data")).toBe(false); + expect(Object.hasOwn((await storage.entry(omitted.id, context))!.entry, "data")).toBe(false); + + const commits = storage.commits.length; + await expect( + session.commit( + (tx) => tx.appendEntry(conversationId, { kind: "invalid", data: Number.NaN }).then(() => undefined), + context, + ), + ).rejects.toThrow("strict JSON"); + expect(storage.commits.length).toBe(commits); + }); + + it("replaces task records completely", async () => { + const { session, storage } = openTestSession(); + const conversationId = await createConversation(session); + const taskId = await createTask(session, conversationId); + await session.commit(async (tx) => { + const task = (await tx.task(taskId))!; + tx.setTask({ + id: task.id, + conversationId: task.conversationId, + kind: task.kind, + version: task.version, + input: task.input, + after: task.after, + background: task.background, + abortRequested: task.abortRequested, + state: { status: "running", checkpoint: { phase: "next", step: 2 } }, + memos: { choice: "b" }, + }); + }, context); + expect(await storage.task(taskId, context)).toMatchObject({ + state: { status: "running", checkpoint: { phase: "next", step: 2 } }, + memos: { choice: "b" }, + }); + }); + + it("creates a task and then its document in one transaction without ReadAfterWrite", async () => { + const { session, publications } = openTestSession(); + const conversationId = await createConversation(session); + const taskId = await session.commit(async (tx) => { + const ref = await tx.createTask(WorkTask, { path: "a" }, { conversationId }); + // Validation uses the candidate task record, not a caller table read. + (await tx.doc(ProgressDoc, ref.id)).lines.push("created"); + (await tx.doc(StepDoc, ref.id, "one", null)).lines.push("step"); + return ref.id; + }, context); + expect(await session.snapshot(ProgressDoc, taskId, context)).toEqual({ lines: ["created"] }); + await flush(); + const publication = publications.at(-1)!; + const documents = documentChanges(publication); + expect(publication.changes).toContainEqual( + expect.objectContaining({ type: "task", value: expect.objectContaining({ id: taskId }) }), + ); + expect(documents).toHaveLength(2); + // Task documents derive their conversation from the task record. + for (const document of documents) expect(document.conversationId).toBe(conversationId); + + await session.commit(async (tx) => { + await tx.createConversation({}); + (await tx.doc(ProgressDoc, taskId)).lines.push("committed task"); + }, context); + await flush(); + expect(documentChanges(publications.at(-1)!)[0]!.conversationId).toBe(conversationId); + }); + + it("rejects task documents after a terminal candidate", async () => { + const { session } = openTestSession(); + const conversationId = await createConversation(session); + const taskId = await createTask(session, conversationId, true); + await session.commit(async (tx) => { + const task = (await tx.task(taskId))!; + const progress = await tx.doc(ProgressDoc, taskId); + tx.setTask(terminal(task)); + await expect(tx.doc(ProgressDoc, taskId)).rejects.toThrow(`Task ${taskId} is terminal`); + await expect(tx.doc(StepDoc, taskId, "late", null)).rejects.toThrow(`Task ${taskId} is terminal`); + expect(() => tx.setTask(task)).toThrow("terminal candidate"); + progress.lines.push("final"); + }, context); + await expect(session.commit((tx) => tx.doc(ProgressDoc, taskId).then(() => undefined), context)).rejects.toThrow( + `Task ${taskId} is terminal`, + ); + }); + + it("retires task documents at terminal settlement, including documents created in the same transaction", async () => { + const { session, storage, publications } = openTestSession(); + const conversationId = await createConversation(session); + const taskId = await createTask(session, conversationId, true); + await session.commit(async (tx) => { + (await tx.doc(StepDoc, taskId, "committed", null)).lines.push("x"); + }, context); + await flush(); + const published = publications.length; + await session.commit(async (tx) => { + const task = (await tx.task(taskId))!; + (await tx.doc(StepDoc, taskId, "new", null)).lines.push("created then retired"); + tx.setTask(terminal(task)); + }, context); + const writes = storage.commits.at(-1)!; + expect(writes).toHaveLength(5); + expect(writes.filter((write) => write.type === "task")).toHaveLength(1); + const creations = writes.filter((write) => write.type === "document.create"); + const retirements = writes.filter((write) => write.type === "document.retire"); + expect(creations).toHaveLength(1); + expect(retirements).toHaveLength(3); + expect(retirements.map((write) => write.id)).toContain(creations[0]!.record.id); + await flush(); + expect(publications.length).toBe(published + 1); + const publication = publications.at(-1)!; + const documents = documentChanges(publication); + expect(publication.changes.filter((change) => change.type === "task")).toHaveLength(1); + expect(documents.map((document) => document.value)).toEqual([null, null, null]); + for (const document of documents) expect(document.ops).toEqual([]); + for (const document of documents) expect(document.conversationId).toBe(conversationId); + expect(await session.snapshot(ProgressDoc, taskId, context)).toBeUndefined(); + expect(await session.snapshot(StepDoc, taskId, "committed", context)).toBeUndefined(); + const alive = await storage.scanDocuments( + { scope: { kind: "task", taskId }, at: "current" }, + 10, + undefined, + context, + ); + expect(alive.items).toEqual([]); + await expect( + session.commit(async (tx) => { + tx.setTask(terminal((await tx.task(taskId))!)); + }, context), + ).rejects.toThrow(`Task ${taskId} is already terminal`); + }); + + it("validates document owners", async () => { + const { session } = openTestSession(); + await expect(session.commit((tx) => tx.doc(ProgressDoc, 4242).then(() => undefined), context)).rejects.toThrow( + "Task 4242 does not exist", + ); + await expect(session.commit((tx) => tx.doc(NotesDoc, 4242).then(() => undefined), context)).rejects.toThrow( + "Conversation 4242 does not exist", + ); + }); +}); diff --git a/packages/durable/test/sqlite-storage.test.ts b/packages/durable/test/sqlite-storage.test.ts index baaebc155..73b45d0fa 100644 --- a/packages/durable/test/sqlite-storage.test.ts +++ b/packages/durable/test/sqlite-storage.test.ts @@ -55,24 +55,24 @@ class ReopeningStorage implements Storage { mintId: Storage["mintId"] = () => this.current.mintId(); conversation: Storage["conversation"] = (id, readContext) => this.current.conversation(id, readContext); - scanConversations: Storage["scanConversations"] = (cursor, limit, readContext) => - this.current.scanConversations(cursor, limit, readContext); + scanConversations: Storage["scanConversations"] = (limit, cursor, readContext) => + this.current.scanConversations(limit, cursor, readContext); entry: Storage["entry"] = (id, readContext) => this.current.entry(id, readContext); findLatestHeadMarker: Storage["findLatestHeadMarker"] = (conversationId, at, readContext) => this.current.findLatestHeadMarker(conversationId, at, readContext); - scanEntries: Storage["scanEntries"] = (query, cursor, limit, readContext) => - this.current.scanEntries(query, cursor, limit, readContext); + scanEntries: Storage["scanEntries"] = (query, limit, cursor, readContext) => + this.current.scanEntries(query, limit, cursor, readContext); task: Storage["task"] = (id, readContext) => this.current.task(id, readContext); - scanTasks: Storage["scanTasks"] = (query, cursor, limit, readContext) => - this.current.scanTasks(query, cursor, limit, readContext); + scanTasks: Storage["scanTasks"] = (query, limit, cursor, readContext) => + this.current.scanTasks(query, limit, cursor, readContext); submission: Storage["submission"] = (id, readContext) => this.current.submission(id, readContext); submissionByRequest: Storage["submissionByRequest"] = (conversationId, requestId, readContext) => this.current.submissionByRequest(conversationId, requestId, readContext); findDocument: Storage["findDocument"] = (address, at, readContext) => this.current.findDocument(address, at, readContext); document: Storage["document"] = (id, at, readContext) => this.current.document(id, at, readContext); - scanDocuments: Storage["scanDocuments"] = (query, cursor, limit, readContext) => - this.current.scanDocuments(query, cursor, limit, readContext); + scanDocuments: Storage["scanDocuments"] = (query, limit, cursor, readContext) => + this.current.scanDocuments(query, limit, cursor, readContext); async close(closeContext: Parameters[0]): Promise { if (this.closed) return;