diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index aeb7d3185..c3e5076ed 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -6,11 +6,15 @@ - Reordered Storage scan arguments so the limit precedes the cursor. - Added the required conversation-visible `Storage.entry(conversationId, id, context)` overload. +- Split independent `Tx.createConversation()` from `Tx.forkConversation()` and replaced raw conversation-record input. +- Added `ConversationQuery` to Storage and transaction conversation scans. ### Added - Added transactional Sessions with typed durable documents, task creation, snapshots, retirement, and commit publications. - Added document checkpoint selection, lazy version migration, and `Session.snapshotAsOf()` for rewindable conversation documents. +- Added policy-driven backend-side conversation document copying when creating forks. +- Added indexed conversation ownership queries and guaranteed no-effect Storage rejection handling. ## [0.87.1] - 2026-09-22 diff --git a/packages/durable/docs/pico-v5.md b/packages/durable/docs/pico-v5.md index 83276768b..25b5b6763 100644 --- a/packages/durable/docs/pico-v5.md +++ b/packages/durable/docs/pico-v5.md @@ -205,7 +205,7 @@ type SubmissionCreate = SubmissionRecord extends infer Record Conversation history parenting and task ownership are separate: - `parent` controls inherited entries and historical documents. -- `owner` controls task authorization, subtree abort, and subtree idle waits. +- `owner` records task attribution and connects scopes for subtree abort and idle traversal. It is not an access-control capability. A conversation's owner remains recorded after the owning task becomes terminal. @@ -277,7 +277,6 @@ type SectionSeed = | { readonly key: string; readonly remove: true }; type ConversationSpec = { - readonly parent?: { readonly conversationId: Id; readonly at: Id }; readonly model?: ModelRef; readonly sections?: readonly SectionSeed[]; readonly activeTools?: readonly string[]; @@ -301,7 +300,7 @@ type HarnessOptions = { readonly taskKinds?: readonly AnyTask[]; readonly sections?: readonly SystemSection[]; readonly now?: () => number; - readonly root?: Omit; + readonly root?: ConversationSpec; readonly onReport?: (error: unknown) => void; }; @@ -372,7 +371,7 @@ interface Conversation { ): Promise>; fork( at: Id, - spec: Omit, + spec: ConversationSpec, context: Context, ): Promise; collapse(instructions: string | undefined, context: Context): Promise; @@ -525,16 +524,17 @@ handoff write and then resolves; while busy, placement follows section 6 and may occur later. Observe its placement through the conversation watch. An idle wait does not guarantee placement of queued passive writes. -`markTask()` only commits `abortRequested`; it neither signals nor joins an -active invocation. The scheduler notices the mark on its next drain. -`abortTask()` also signals and joins an active run before starting the abort -invocation. `Conversation.abort()` withdraws queued input submissions, marks -non-background tasks in the conversation and owned subtree, signals them, and -resolves only after that subtree is ordinarily idle. Passive writes and -background tasks survive. Conversation idle means no non-background live task -in that conversation or its ownership subtree; Harness idle applies the same -rule Session-wide. Pending dependency- or deadline-blocked work is still live -and therefore not idle. Cancelling an idle wait aborts only that waiter. +`markTask()` commits `abortRequested` and the durable foreground-subtree +cascade; it neither signals nor joins active invocations. The scheduler notices +the marks on its next drain. `abortTask()` also signals and joins the active run +before starting the abort invocation. `Conversation.abort()` withdraws queued +input submissions, marks non-background tasks selected by ordinary ownership +traversal, signals them, and resolves only after that scope is ordinarily idle. +Passive writes and background subtrees survive. Conversation idle means no live +non-background task selected from that conversation. Harness idle applies the +same traversal from every ownerless conversation root. Pending dependency- or +deadline-blocked work is still live and therefore not idle. Cancelling an idle +wait aborts only that waiter. `onConversation()` synchronously visits the currently loaded committed conversations in ascending ID order, then reports each later creation after its @@ -804,11 +804,12 @@ interface Tx { conversation(id: Id): Promise; entry(id: Id): Promise; task(id: Id): Promise | undefined>; - scanConversations(limit: number, cursor?: Cursor): Promise>; + scanConversations(query: ConversationQuery, 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; + createConversation(): Promise; + forkConversation(parentConversationId: Id, at: Id): Promise; appendEntry(conversationId: Id, value: EntryDraft): Promise; createTask( task: Task, input: I, options?: TaskOptions, @@ -831,7 +832,7 @@ interface Tx { } ``` -ID-creating transaction methods are asynchronous because remote storage may allocate globally unique IDs durably. Only public typed `tx.doc()` is get-or-create. Internal fork copying may create +ID-creating transaction methods are asynchronous because remote storage may allocate globally unique IDs durably. `createConversation()` creates an independent ownerless scope; `forkConversation()` validates one visible entry and applies section 3.7. Task-bound transactions later inject their executing task as owner without accepting raw owner IDs from callers. Only public typed `tx.doc()` is get-or-create. Internal fork copying may create new incarnations directly from stored values without a definition. `tx.doc()` receives the definition token that supplies its static type, initializer, migration, and checkpoint policy. Scope-preserving @@ -1084,7 +1085,7 @@ Each conversation document follows the history/fork policy persisted in its | conversation setting | child value | |---|---| | `fork: "asOf"` | parent value at `E`'s commit | -| `fork: "current"` | parent value when the fork commit runs | +| `fork: "current"` | committed parent value selected when the fork commit runs | | `fork: "initial"` | no copied instance; initializer on first child access | `current` and `asOf` copy logically present conversation singleton and family @@ -1094,6 +1095,32 @@ values become independent child instances with new IDs and initial bases. supplied definition. Task documents and tasks are never copied. Session documents remain shared and are not rewindable. +Fork copying reads committed stored values rather than typed tracker caches. A +transaction that creates a fork therefore rejects if it also writes one of the +parent's `fork: "current"` documents; commit the parent change first so the fork +has one unambiguous stored source revision. + +Forks stage backend-side `document.copy` commands carrying the child create +record and an exact source incarnation/point. Storage materializes each source +and persists its stored value/version as the child's independent initial base; +remote Storage performs this server-side. Every copy reads committed pre-batch +source state independent of write-array order. A selected source may not be +created, changed, or retired in the same batch. Copying remains atomic with the +conversation, overrides, registry writes, and other mutations. + +An unaccessed copy retains only its descriptor in Session memory. Typed access +inside the creating transaction lazily reads the detached source, migrates when +required, and replaces the copy with one ordinary child create containing the +final prepared value. Definition-free copies publish explicit `document.copy` +metadata rather than a value. That metadata announces Storage-backed initial +state and is never interpreted as a document value. Document sources, watches, +and mounted views hydrate by capturing their baseline and subscription +atomically on the Session line: a later commit already present becomes the +baseline, while one committed after registration is delivered. Publications +and watches are convergence mechanisms, not audit streams. A mounted aggregate +must acquire all of its document baselines and commit subscription in one +Session-line operation so it never exposes a mixture from one commit. + ## 4. Transactions and storage ownership A Session commit callback may be asynchronous. It owns the Session mutation @@ -1329,10 +1356,35 @@ the durable mark on reopen. Cancelling one caller's `Context` only cancels that call or wait; it does not durably abort shared work unless the invoked API commits an abort mark. -A task may create owned conversations. Abort and idle operations traverse the -conversation ownership tree. History parents are irrelevant to this traversal. -Background tasks do not block ordinary idle waits and survive ordinary -conversation abort unless explicitly included. +A task may create owned conversations. Conversations are durable scopes; tasks +are the units of live work counted by idle and marked by abort. History parents +are irrelevant to ownership traversal. + +Ordinary traversal starts at an explicitly addressed conversation, visits its +tasks, and follows conversations owned by each non-background task. It follows +owner edges after the owner becomes terminal, but a background owner is a +boundary: ordinary traversal skips that task and its complete owned subtree. +Direct conversation operations start inside that conversation regardless of its +owner. Directly aborting a live background task includes that task and follows +its ordinary owned subtree; nested background owners remain boundaries. Full +teardown crosses every boundary, marks every live task, and must seal new +admission while it gathers the complete indexed ownership subtree. + +`Conversation.abort()` withdraws queued inputs and marks live non-background +tasks selected by ordinary traversal. `Conversation.waitForIdle()` waits until +that traversal contains no live non-background task. Harness idle performs the +same traversal from every ownerless conversation root rather than globally +counting tasks, so ordinary work below a background owner does not block it. +Explicit `waitForTask()` waits for its referenced task regardless of the task's +background flag. + +An abort mark atomically and idempotently cascades to foreground-owned work, +including conversations, tasks, and submissions staged in the same transaction. +Terminal outcomes `failed`, `faulted`, `orphaned`, and `aborted` record the same +durable cancellation intent; `completed` does not. Conversation records and +owner edges are never retired with the task. Active invocations are signalled +after commit, and a terminal receipt guarantees durable cancellation intent, +not descendant quiescence. Initial task definitions are registered before open performs live-task migration and orphan reconciliation. Dynamic registration begins only after that pass. @@ -1472,9 +1524,10 @@ interface ToolExecutionApi extends DocumentObserver { getTask(ref: TaskRef, context: Context): Promise | undefined>; waitForTask(ref: TaskRef, context: Context): Promise>; createConversation( - spec: Omit & { readonly inherit?: boolean }, + spec: ConversationSpec & { readonly inherit?: boolean }, context: Context, ): Promise; + conversation(id: Id, context: Context): Promise; } type ToolRegistration = Tool & { @@ -1540,9 +1593,12 @@ A tool executes in a durable task. It may: Tool operations use the invoking task's admission and invocation-lifetime gates. Task creation defaults to that task's conversation. -Trusted document access follows sections 3.3 and 9.2; there is no additional -document subtree authorization layer. `createConversation()` records the -invoking task as owner and returns an invocation-bound handle. `inherit` defaults +Trusted document and conversation access follows sections 3.3 and 9.2; there is +no additional subtree authorization layer. Possession of a Session-global ID is +sufficient in trusted code. `createConversation()` records the invoking task as +owner and returns an invocation-bound handle. `conversation(id)` reacquires the +same handle shape for status, steering, explicit abort, and idle waits without +changing ownership. `inherit` defaults to `false`. When true, creation atomically selects the invoking conversation's newest committed visible entry as `parent.at`; if no entry exists, it creates no history parent. Conversation documents then apply section 3.7 at that entry, and @@ -1553,6 +1609,34 @@ operations reject after the invocation ends. A `Submission` returned by its durable after those methods reject. Invocation-owned document watches stop when the invocation ends. +A foreground subagent conversation is created directly by its tool task; the +tool waits for the admitted submission/result, and aborting or abnormally +terminalizing that task cascades through the owned scope. A background subagent +is created by a background provisioning task. The tool may explicitly +`waitForTask()` only until that task durably returns the child conversation and +submission IDs, then return while the child continues. The provisioner may +complete after setup: its terminal record retains `background`, so ancestor +ordinary abort and idle traversal continue to stop at its owned scope. Provision +the conversation and its durable name-to-conversation/request mapping atomically, +then call `Conversation.submit()` with that application-minted request ID. A +crash before submission retries the durable request; a crash after admission +recovers the existing deduplicated `Submission`. A replay-safe tool must never +create a child in one commit and record its identity only in a later memo commit. + +Applications may maintain a conversation document mapping semantic subagent +names to durable conversation IDs and application-minted submission request IDs. +Such a live registry uses `fork: "initial"` so children do not inherit the +parent's agent list and its update is not a selected fork source. Conversation +creation/forking and its registry mapping commit atomically. Initial or later +input then uses the registered stable request ID with the full +`Conversation.submit()` state machine. A crash before admission leaves a durable +request to submit; a crash after admission retries the same request and receives +the existing `Submission`. A read-only lookup by conversation/request ID can +report absence or return the durable queued/placed/done/unanswered receipt. +Submission admission therefore needs no private transaction shortcut and does +not move onto `Tx`. Kernel ownership indexes independently drive abort and idle +traversal. + A tool result may request `addTools`, `terminate`, or `handoff`. Post-tools applies added tool names to configured loadout, uses a final boundary for terminate/handoff, and writes a headed handoff entry when requested. @@ -1973,6 +2057,11 @@ type Page = { type Cursor = Readonly>; +type ConversationQuery = { + readonly ownerConversationId?: Id; + readonly ownerTaskId?: Id; +}; + type EntryQuery = { readonly conversationId: Id; readonly minEntryId?: Id; // inclusive @@ -2021,6 +2110,11 @@ type StorageWrite = readonly record: DocumentCreate; readonly content: Extract; } + | { + readonly type: "document.copy"; + readonly record: DocumentCreate; + readonly source: { readonly id: Id; readonly at: DocumentPoint }; + } | { readonly type: "document.change"; readonly id: Id; @@ -2040,7 +2134,7 @@ interface Storage { mintId(): Promise; conversation(id: Id, context: Context): Promise; - scanConversations(limit: number, cursor: Cursor | undefined, context: Context): Promise>; + scanConversations(query: ConversationQuery, limit: number, cursor: Cursor | undefined, context: Context): Promise>; entry(id: Id, context: Context): Promise<{ readonly entry: EntryRecord; readonly commitSeq: Seq } | undefined>; entry(conversationId: Id, id: Id, context: Context): Promise<{ readonly entry: EntryRecord; readonly commitSeq: Seq } | undefined>; @@ -2061,6 +2155,19 @@ interface Storage { } ``` +`StorageRejected` means a batch was rejected before any durable effect and is +guaranteed not to have committed. Session rolls such a batch back normally; +unknown failures after Storage admission remain fatal because their commit state +is uncertain. Backends use `StorageRejected` for deterministic `document.copy` +source, replay, and consistency failures only when rollback is guaranteed. + +A `document.copy` reads committed pre-batch source state independent of command +order. The source must be an alive conversation document at the selected point, +and kind/key/history/fork must match the child create record. Storage persists +one independent complete child base at the source's stored version. A batch may +not create, change, or retire a selected source. Later source changes, +reclamation, retirement, or backend reopen cannot affect the child. + Cursors are backend-owned JSON objects. Callers only round-trip them to the same scan on the same storage; cross-storage or cross-query use is unsupported. The Session owns the mutation line, so storage implementations do not add a second @@ -2073,6 +2180,10 @@ inclusive ID range in newest-first order while applying every conversation ancestry cap. With no bounds it pages complete visible history. To read context through entry `E`, find the marker at or before `E`, then scan from `marker?.head` through `E`. For current context the upper bound is omitted. +Conversation owner filters are indexed and conjunctive. They support ownership +traversal without an all-conversation scan; application-maintained registries +are not a substitute for these kernel indexes. + `entry(id)` combines exact global lookup with the commit sequence required by historical document reads. `entry(conversationId, id)` returns that pair only when the entry is visible through the requested conversation's ancestry. `limit` diff --git a/packages/durable/src/documents.ts b/packages/durable/src/documents.ts index 91c1dd252..421bb4899 100644 --- a/packages/durable/src/documents.ts +++ b/packages/durable/src/documents.ts @@ -160,7 +160,7 @@ export function documentCreate(definition: AnyDocDefinition, address: DocumentAd } /** Reject typed access whose token disagrees with the persisted scope, history, or fork semantics. */ -export function checkRecordScope(definition: AnyDocDefinition, record: DocumentRecord): void { +export function checkRecordScope(definition: AnyDocDefinition, record: DocumentCreate | DocumentRecord): void { if ( record.scope.kind !== definition.scope || (record.scope.kind === "conversation" && @@ -171,7 +171,11 @@ export function checkRecordScope(definition: AnyDocDefinition, record: DocumentR } /** Reject typed access to a stored version the supplied definition cannot use. */ -export function checkRecordVersion(definition: AnyDocDefinition, record: DocumentRecord, version: number): void { +export function checkRecordVersion( + definition: AnyDocDefinition, + record: DocumentCreate | DocumentRecord, + version: number, +): void { if (version > definition.version) { throw new Error(`Document ${record.id} (${record.kind}) has newer version ${version} than ${definition.version}`); } @@ -182,8 +186,18 @@ export function checkRecordVersion(definition: AnyDocDefinition, record: Documen /** Validate and materialize a detached stored value for typed access. */ export function materializeDocument(definition: AnyDocDefinition, stored: StoredDocument): JsonObject { - checkRecordScope(definition, stored.record); - checkRecordVersion(definition, stored.record, stored.version); - if (stored.version === definition.version) return stored.value; - return copyJson(definition.migrate!(stored.value, stored.version)) as JsonObject; + return materializeDocumentValue(definition, stored.record, stored.version, stored.value); +} + +/** Validate and materialize one detached value before its first persisted incarnation. */ +export function materializeDocumentValue( + definition: AnyDocDefinition, + record: DocumentCreate | DocumentRecord, + version: number, + value: JsonObject, +): JsonObject { + checkRecordScope(definition, record); + checkRecordVersion(definition, record, version); + if (version === definition.version) return value; + return copyJson(definition.migrate!(value, version)) as JsonObject; } diff --git a/packages/durable/src/errors.ts b/packages/durable/src/errors.ts new file mode 100644 index 000000000..f8fab62c4 --- /dev/null +++ b/packages/durable/src/errors.ts @@ -0,0 +1,15 @@ +/** 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"; + } +} + +/** Storage rejected a batch before any durable effect; the owning Session may continue safely. */ +export class StorageRejected extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = "StorageRejected"; + } +} diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index fdbfa4515..92f67e104 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -1,12 +1,13 @@ export { defineDoc, defineDocFamily } from "./documents.ts"; +export { ReadAfterWrite, StorageRejected } from "./errors.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, + ConversationQuery, ConversationRecord, Cursor, DocDefinition, @@ -15,6 +16,7 @@ export type { DocToken, DocumentAddress, DocumentContent, + DocumentCopySource, DocumentCreate, DocumentPoint, DocumentQuery, diff --git a/packages/durable/src/session/forks.ts b/packages/durable/src/session/forks.ts new file mode 100644 index 000000000..b76d2c5d4 --- /dev/null +++ b/packages/durable/src/session/forks.ts @@ -0,0 +1,95 @@ +import type { Context } from "@earendil-works/chord"; +import { addressId } from "../documents.ts"; +import type { + Cursor, + DocumentCopySource, + DocumentCreate, + DocumentPoint, + DocumentRecord, + Id, + Storage, +} from "../types.ts"; + +const SCAN_PAGE_SIZE = 256; + +type ForkPolicy = "asOf" | "current"; + +/** One definition-free document copy to create with a forked conversation. */ +export type ForkDocumentCopy = { + readonly record: DocumentCreate; + readonly source: DocumentCopySource; +}; + +/** Select every persisted conversation document copied by one fork. */ +export async function prepareForkDocumentCopies( + storage: Storage, + parentConversationId: Id, + at: Id, + childConversationId: Id, + context: Context, +): Promise { + const entry = await storage.entry(parentConversationId, at, context); + if (entry === undefined) throw new Error(`Entry ${at} is not visible from conversation ${parentConversationId}`); + + const copies: ForkDocumentCopy[] = []; + const copiedAddresses = new Set(); + await collectCopies( + storage, + { kind: "conversation", conversationId: entry.entry.conversationId }, + entry.commitSeq, + "asOf", + childConversationId, + copies, + copiedAddresses, + context, + ); + await collectCopies( + storage, + { kind: "conversation", conversationId: parentConversationId }, + "current", + "current", + childConversationId, + copies, + copiedAddresses, + context, + ); + return copies; +} + +async function collectCopies( + storage: Storage, + scope: Extract, + at: DocumentPoint, + policy: ForkPolicy, + childConversationId: Id, + copies: ForkDocumentCopy[], + copiedAddresses: Set, + context: Context, +): Promise { + let cursor: Cursor | undefined; + do { + const page = await storage.scanDocuments({ scope, at }, SCAN_PAGE_SIZE, cursor, context); + for (const source of page.items) { + if (source.scope.kind !== "conversation" || source.fork !== policy) continue; + const id = await storage.mintId(); + const identity = { + id, + kind: source.kind, + ...(source.key === undefined ? {} : { key: source.key }), + scope: { kind: "conversation" as const, conversationId: childConversationId }, + }; + const record: DocumentCreate = + source.history === "latest" + ? { ...identity, history: "latest", fork: source.fork } + : { ...identity, history: "rewindable", fork: source.fork }; + const copyAddress = addressId(record); + if (copiedAddresses.has(copyAddress)) { + const member = record.key === undefined ? record.kind : `${record.kind}/${record.key}`; + throw new Error(`Fork selects multiple source documents for ${member}`); + } + copiedAddresses.add(copyAddress); + copies.push({ record, source: { id: source.id, at } }); + } + cursor = page.next; + } while (cursor !== undefined); +} diff --git a/packages/durable/src/session/session.ts b/packages/durable/src/session/session.ts index f55186ee7..731d43fa8 100644 --- a/packages/durable/src/session/session.ts +++ b/packages/durable/src/session/session.ts @@ -8,6 +8,7 @@ import { materializeDocument, resolveAddress, } from "../documents.ts"; +import { StorageRejected } from "../errors.ts"; import type { ConversationDocFamilyToken, ConversationDocToken, @@ -208,7 +209,8 @@ export class SessionKernel implements Session { seq = await this.#storage.commit(writes, withoutAbortSignal(context)); } catch (error) { tx.discard(); - this.#poison = { error }; + // Callback errors never reach this branch; StorageRejected alone guarantees that no batch effect committed. + if (!(error instanceof StorageRejected)) this.#poison = { error }; throw error; } let documents: DocumentCommitChange[]; diff --git a/packages/durable/src/session/transaction.ts b/packages/durable/src/session/transaction.ts index e5498248e..ccdca2571 100644 --- a/packages/durable/src/session/transaction.ts +++ b/packages/durable/src/session/transaction.ts @@ -7,15 +7,19 @@ import { checkRecordScope, checkRecordVersion, documentCreate, + materializeDocumentValue, type ResolvedAddress, resolveAddress, } from "../documents.ts"; +import { ReadAfterWrite } from "../errors.ts"; import type { ConversationDocFamilyToken, ConversationDocToken, + ConversationQuery, ConversationRecord, Cursor, DocumentAddress, + DocumentCopySource, DocumentCreate, DocumentRecord, EntryDraft, @@ -37,6 +41,7 @@ import type { TaskRef, Tx, } from "../types.ts"; +import { prepareForkDocumentCopies } from "./forks.ts"; type AnyTaskRecord = TaskRecord; @@ -44,25 +49,27 @@ 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[]; -}; +export type DocumentCommitChange = + | { + readonly type: "document"; + readonly record: DocumentRecord; + /** Conversation owning the document; task documents derive it from their task record. Undefined only for Session documents. */ + readonly conversationId: Id | undefined; + /** Definition version of `value`; absent when this commit retired the incarnation. */ + readonly version: number | 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[]; + } + | { + /** Definition-free child initialization; consumers hydrate through a source or watch. */ + readonly type: "document.copy"; + readonly record: DocumentRecord; + readonly conversationId: Id; + readonly source: DocumentCopySource; + }; /** One committed document incarnation owned by the Session tracker cache. */ export type LoadedDocument = { @@ -107,14 +114,19 @@ type DocumentTarget = readonly version: number; readonly tracker: Tracker; } + | { + readonly kind: "fork-copy"; + readonly record: DocumentCreate; + readonly source: DocumentCopySource; + } | { 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; + /** Absent for definition-free fork copies and retirement entries discovered by a terminal-task scan. */ + definition?: AnyDocDefinition; /** Memoized public acquisition; absent for metadata-only retirement. */ draftPromise?: Promise>; /** Set after acquisition or retirement lookup finds the affected incarnation. */ @@ -142,6 +154,8 @@ export class Transaction implements Tx { /** Atomic batch; conversation and entry writes stage eagerly, while task and document writes assemble later. */ readonly #writes: StorageWrite[] = []; readonly #createdConversationIds = new Set(); + readonly #forkSourceConversationIds = new Set(); + readonly #forkSourceDocumentIds = new Set(); /** One entry per task touched by a public read, candidate write, or document-owner lookup. */ readonly #tasksById = new Map(); @@ -169,8 +183,10 @@ export class Transaction implements Tx { return this.#read("task", () => this.#committedTask(id)); } - scanConversations(limit: number, cursor?: Cursor) { - return this.#read("scanConversations", () => this.#host.storage.scanConversations(limit, cursor, this.#context)); + scanConversations(query: ConversationQuery, limit: number, cursor?: Cursor) { + return this.#read("scanConversations", () => + this.#host.storage.scanConversations(query, limit, cursor, this.#context), + ); } scanEntries(query: EntryQuery, limit: number, cursor?: Cursor) { @@ -183,15 +199,38 @@ export class Transaction implements Tx { // ─── 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; - }); + createConversation(): Promise { + return this.#write(() => this.#stageConversation()); + } + + forkConversation(parentConversationId: Id, at: Id): Promise { + return this.#write(() => this.#stageConversation({ conversationId: parentConversationId, at })); + } + + async #stageConversation(parent?: NonNullable): Promise { + const id = await this.#host.storage.mintId(); + this.#assertOpen(); + const record: ConversationRecord = parent === undefined ? { id } : { id, parent }; + const copies = + parent === undefined + ? [] + : await prepareForkDocumentCopies(this.#host.storage, parent.conversationId, parent.at, id, this.#context); + this.#assertOpen(); + for (const copy of copies) { + this.#forkSourceDocumentIds.add(copy.source.id); + const entry: DocumentEntry = { + addressId: addressId(copy.record), + address: copy.record, + target: { kind: "fork-copy", ...copy }, + retireOnCommit: false, + }; + this.#documents.push(entry); + this.#latestDocumentByAddress.set(entry.addressId, entry); + } + if (parent !== undefined) this.#forkSourceConversationIds.add(parent.conversationId); + this.#createdConversationIds.add(id); + this.#writes.push({ type: "conversation", value: record }); + return record; } appendEntry(conversationId: Id, value: EntryDraft): Promise { @@ -286,8 +325,12 @@ export class Transaction implements Tx { 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; + if (latest !== undefined && !latest.retireOnCommit) { + if (latest.draftPromise !== undefined) return latest.draftPromise; + if (latest.target?.kind === "fork-copy") { + latest.draftPromise = this.#track(this.#acquireForkCopy(latest, definition, latest.target)); + return latest.draftPromise; + } } const seed = definition.family === true ? copyJson(args[resolved.nextArgument]) : undefined; const docEntry: DocumentEntry = { @@ -327,6 +370,11 @@ export class Transaction implements Tx { const resolved = resolveAddress(definition, args); const latest = this.#latestDocumentByAddress.get(resolved.id); if (latest?.retireOnCommit) return Promise.resolve(); + if (latest?.target?.kind === "fork-copy") { + checkRecordScope(definition, latest.target.record); + latest.retireOnCommit = true; + return Promise.resolve(); + } if (latest?.draftPromise !== undefined) { // Retirement of an acquired draft persists its final content before retirement. latest.retireOnCommit = true; @@ -383,6 +431,33 @@ export class Transaction implements Tx { return entry.change.state; } + async #acquireForkCopy( + entry: DocumentEntry, + definition: AnyDocDefinition, + target: Extract, + ): Promise> { + const stored = await this.#host.storage.document(target.source.id, target.source.at, this.#context); + this.#assertOpen(); + if (stored === undefined) { + throw new Error(`Fork source document ${target.source.id} cannot be read`); + } + if ( + stored.record.scope.kind !== "conversation" || + stored.record.kind !== target.record.kind || + stored.record.key !== target.record.key || + stored.record.history !== target.record.history || + stored.record.fork !== target.record.fork + ) { + throw new Error(`Fork source document ${target.source.id} does not match the copied record`); + } + const value = materializeDocumentValue(definition, target.record, stored.version, stored.value); + const tracker = track(value); + entry.definition = definition; + entry.target = { kind: "created", record: target.record, 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 ?? @@ -456,11 +531,35 @@ export class Transaction implements Tx { type: "document", record, conversationId: document.conversationId, + version: document.retireOnCommit ? undefined : target.version, value: document.retireOnCommit ? null : prepared.value, ops: EMPTY_OPERATIONS, }); break; } + case "fork-copy": { + const record: DocumentRecord = document.retireOnCommit + ? { ...target.record, createdAt: seq, retiredAt: seq } + : { ...target.record, createdAt: seq }; + if (document.retireOnCommit) { + publications.push({ + type: "document", + record, + conversationId: document.conversationId, + version: undefined, + value: null, + ops: EMPTY_OPERATIONS, + }); + } else { + publications.push({ + type: "document.copy", + record, + conversationId: document.conversationId!, + source: target.source, + }); + } + break; + } case "loaded": { const prepared = document.prepared!; const changed = prepared.ops.length > 0; @@ -479,6 +578,7 @@ export class Transaction implements Tx { ? { ...target.document.record, retiredAt: seq } : target.document.record, conversationId: document.conversationId, + version: document.retireOnCommit ? undefined : document.definition!.version, value: document.retireOnCommit ? null : prepared.value, ops: document.retireOnCommit ? EMPTY_OPERATIONS : prepared.ops, }); @@ -490,6 +590,7 @@ export class Transaction implements Tx { type: "document", record: { ...target.record, retiredAt: seq }, conversationId: document.conversationId, + version: undefined, value: null, ops: EMPTY_OPERATIONS, }); @@ -501,6 +602,7 @@ export class Transaction implements Tx { async #assemble(): Promise { const storage = this.#host.storage; + this.#rejectForkSourceWrites(); for (const [id, task] of this.#tasksById) { if (task.write?.kind !== "replace") continue; const committed = await this.#committedTask(id); @@ -583,6 +685,10 @@ export class Transaction implements Tx { }); if (document.retireOnCommit) writes.push({ type: "document.retire", id: target.record.id }); break; + case "fork-copy": + writes.push({ type: "document.copy", record: target.record, source: target.source }); + if (document.retireOnCommit) writes.push({ type: "document.retire", id: target.record.id }); + break; case "loaded": { const definition = document.definition!; const prepared = document.prepared!; @@ -617,6 +723,34 @@ export class Transaction implements Tx { // ─── Helpers ──────────────────────────────────────────────────────────── + #rejectForkSourceWrites(): void { + for (const document of this.#documents) { + const target = document.target; + if (target === undefined) continue; + const writes = + target.kind !== "loaded" || + document.retireOnCommit || + target.document.storedVersion < document.definition!.version || + document.prepared!.ops.length > 0; + if (!writes) continue; + const record = target.kind === "loaded" ? target.document.record : target.record; + if (this.#forkSourceDocumentIds.has(record.id)) { + throw new Error(`Cannot change fork source document ${record.id} in the fork transaction`); + } + const scope = document.address.scope; + if ( + scope.kind === "conversation" && + this.#forkSourceConversationIds.has(scope.conversationId) && + record.scope.kind === "conversation" && + record.fork === "current" + ) { + throw new Error( + `Cannot fork conversation ${scope.conversationId} while changing its current-policy documents`, + ); + } + } + } + #abortChanges(): void { for (const document of this.#documents) document.change?.abort(); } diff --git a/packages/durable/src/storage/jsonl/storage.ts b/packages/durable/src/storage/jsonl/storage.ts index 2aacfcaa4..434d2afa2 100644 --- a/packages/durable/src/storage/jsonl/storage.ts +++ b/packages/durable/src/storage/jsonl/storage.ts @@ -1,6 +1,7 @@ import type { Context, JsonValue } from "@earendil-works/chord"; import type { FileError, FileSystem } from "../../env/index.ts"; import type { + ConversationQuery, Cursor, DocumentAddress, DocumentContent, @@ -305,8 +306,8 @@ export class JsonlStorage implements Storage { return this.store.conversation(id, context); } - async scanConversations(limit: number, cursor: Cursor | undefined, context: Context) { - return this.store.scanConversations(limit, cursor, context); + async scanConversations(query: ConversationQuery, limit: number, cursor: Cursor | undefined, context: Context) { + return this.store.scanConversations(query, limit, cursor, context); } entry(id: Id, context: Context): ReturnType; diff --git a/packages/durable/src/storage/memory.ts b/packages/durable/src/storage/memory.ts index 591b005fa..04befd67c 100644 --- a/packages/durable/src/storage/memory.ts +++ b/packages/durable/src/storage/memory.ts @@ -1,6 +1,8 @@ import type { Context, JsonValue } from "@earendil-works/chord"; import { applyImmutableBatches, type Op } from "@earendil-works/chord/delta"; +import { StorageRejected } from "../errors.ts"; import type { + ConversationQuery, ConversationRecord, Cursor, DocumentAddress, @@ -60,6 +62,8 @@ type DocumentAddressIndex = { type State = { conversations: Map; conversationIds: Id[]; + conversationIdsByOwnerConversation: Map; + conversationIdsByOwnerTask: Map; entries: Map; entryIds: Map; headEntryIds: Map; @@ -137,6 +141,15 @@ const removeSorted = (ids: Id[], id: Id): void => { if (ids[index] === id) ids.splice(index, 1); }; +const insertMapId = (index: Map, key: Id, id: Id): void => { + let ids = index.get(key); + if (ids === undefined) { + ids = []; + index.set(key, ids); + } + insertSorted(ids, id); +}; + const scopeKey = (scope: DocumentRecord["scope"]): string => { switch (scope.kind) { case "session": @@ -199,6 +212,8 @@ export class MemoryStorage implements Storage { private readonly state: State = { conversations: new Map(), conversationIds: [], + conversationIdsByOwnerConversation: new Map(), + conversationIdsByOwnerTask: new Map(), entries: new Map(), entryIds: new Map(), headEntryIds: new Map(), @@ -226,7 +241,7 @@ export class MemoryStorage implements Storage { if (!Number.isSafeInteger(seq) || seq < this.nextSeq) { throw new Error(`Commit sequence ${seq} does not strictly increase`); } - const detachedWrites = freeze(writes.map((write) => clone(write))); + const detachedWrites = freeze(this.resolveDocumentCopies(writes.map((write) => clone(write)))); this.checkGlobalIds(detachedWrites); const documentActions = this.prepareDocumentActions(detachedWrites); this.checkDocumentActions(documentActions); @@ -244,6 +259,46 @@ export class MemoryStorage implements Storage { }; } + private resolveDocumentCopies(writes: StorageWrite[]): StorageWrite[] { + if (!writes.some((write) => write.type === "document.copy")) return writes; + const changedDocumentIds = new Set(); + for (const write of writes) { + if (write.type === "document.create" || write.type === "document.copy") { + changedDocumentIds.add(write.record.id); + } else if (write.type === "document.change" || write.type === "document.retire") { + changedDocumentIds.add(write.id); + } + } + return writes.map((write) => { + if (write.type !== "document.copy") return write; + try { + if (changedDocumentIds.has(write.source.id)) { + throw new Error(`Fork source document ${write.source.id} is changed in the copy batch`); + } + const stored = this.materializeDocument(write.source.id, write.source.at); + if (stored === undefined) throw new Error(`Fork source document ${write.source.id} cannot be read`); + if ( + stored.record.scope.kind !== "conversation" || + write.record.scope.kind !== "conversation" || + stored.record.kind !== write.record.kind || + stored.record.key !== write.record.key || + stored.record.history !== write.record.history || + stored.record.fork !== write.record.fork + ) { + throw new Error(`Fork source document ${write.source.id} does not match the copied record`); + } + return { + type: "document.create", + record: write.record, + content: { kind: "base", version: stored.version, value: stored.value }, + }; + } catch (error) { + if (error instanceof StorageRejected) throw error; + throw new StorageRejected(`Document copy ${write.record.id} was rejected`, { cause: error }); + } + }); + } + private applyPreparedCommit( prepared: readonly StorageWrite[], documentActions: ReadonlyMap, @@ -251,11 +306,17 @@ export class MemoryStorage implements Storage { ): Seq { for (const write of prepared) { switch (write.type) { - case "conversation": + case "conversation": { this.state.conversations.set(write.value.id, write.value); insertSorted(this.state.conversationIds, write.value.id); + const owner = write.value.owner; + if (owner !== undefined) { + insertMapId(this.state.conversationIdsByOwnerConversation, owner.conversationId, write.value.id); + insertMapId(this.state.conversationIdsByOwnerTask, owner.taskId, write.value.id); + } this.nextId = Math.max(this.nextId, write.value.id + 1); break; + } case "entry": { this.state.entries.set(write.value.id, write.value); this.state.entryCommitSeqs.set(write.value.id, seq); @@ -310,6 +371,8 @@ export class MemoryStorage implements Storage { this.nextId = Math.max(this.nextId, write.value.id + 1); break; } + case "document.copy": + throw new Error("Prepared document copy was not resolved"); case "document.create": case "document.change": case "document.retire": @@ -335,16 +398,28 @@ export class MemoryStorage implements Storage { } async scanConversations( + query: ConversationQuery, limit: number, cursor: Cursor | undefined, _context: Context, ): Promise> { this.assertOpen(); + const ids = + query.ownerTaskId !== undefined + ? (this.state.conversationIdsByOwnerTask.get(query.ownerTaskId) ?? []) + : query.ownerConversationId !== undefined + ? (this.state.conversationIdsByOwnerConversation.get(query.ownerConversationId) ?? []) + : this.state.conversationIds; const after = cursorId(cursor); - const start = after === undefined ? 0 : upperBound(this.state.conversationIds, after); - const values = this.state.conversationIds - .slice(start, start + limit + 1) - .map((id) => this.state.conversations.get(id)!); + const start = after === undefined ? 0 : upperBound(ids, after); + const values: ConversationRecord[] = []; + for (let index = start; index < ids.length && values.length <= limit; index++) { + const value = this.state.conversations.get(ids[index]!)!; + if (query.ownerConversationId !== undefined && value.owner?.conversationId !== query.ownerConversationId) { + continue; + } + values.push(value); + } return page(values, limit); } @@ -476,22 +551,10 @@ export class MemoryStorage implements Storage { async document(id: Id, at: DocumentPoint, _context: Context): Promise { this.assertOpen(); - const stored = this.state.documents.get(id); - if (stored === undefined) return undefined; - if (at !== "current" && isCurrentOnly(stored.record)) { - throw new Error(`Document ${id} does not retain historical content`); - } - if (!isAliveAt(stored.record, at)) return undefined; - const revisions = at === "current" ? stored.revisions : stored.revisions.filter((revision) => revision.seq <= at); - let baseIndex = revisions.length - 1; - while (baseIndex >= 0 && revisions[baseIndex]!.kind !== "base") baseIndex--; - const base = revisions[baseIndex]; - if (base?.kind !== "base") throw new Error(`Document ${id} is missing a required base`); - const value = applyImmutableBatches( - base.value, - documentDeltaBatches(id, base.version, revisions, baseIndex + 1), - ) as JsonObject; - return { record: clone(stored.record), version: base.version, value: clone(value) }; + const stored = this.materializeDocument(id, at); + return stored === undefined + ? undefined + : { record: clone(stored.record), version: stored.version, value: clone(stored.value) }; } async scanDocuments( @@ -517,6 +580,25 @@ export class MemoryStorage implements Storage { this.closed = true; } + private materializeDocument(id: Id, at: DocumentPoint): StoredDocument | undefined { + const stored = this.state.documents.get(id); + if (stored === undefined) return undefined; + if (at !== "current" && isCurrentOnly(stored.record)) { + throw new Error(`Document ${id} does not retain historical content`); + } + if (!isAliveAt(stored.record, at)) return undefined; + const revisions = at === "current" ? stored.revisions : stored.revisions.filter((revision) => revision.seq <= at); + let baseIndex = revisions.length - 1; + while (baseIndex >= 0 && revisions[baseIndex]!.kind !== "base") baseIndex--; + const base = revisions[baseIndex]; + if (base?.kind !== "base") throw new Error(`Document ${id} is missing a required base`); + const value = applyImmutableBatches( + base.value, + documentDeltaBatches(id, base.version, revisions, baseIndex + 1), + ) as JsonObject; + return { record: stored.record, version: base.version, value }; + } + private *visibleEntries( conversationId: Id, minEntryId = Number.NEGATIVE_INFINITY, @@ -546,8 +628,9 @@ export class MemoryStorage implements Storage { const claimed = new Map(); for (const write of writes) { if (write.type === "document.change" || write.type === "document.retire") continue; - const table: TableName = write.type === "document.create" ? "document" : write.type; - const id = write.type === "document.create" ? write.record.id : write.value.id; + const document = write.type === "document.create" || write.type === "document.copy"; + const table: TableName = document ? "document" : write.type; + const id = document ? write.record.id : write.value.id; const existing = tableContaining(this.state, id); const earlier = claimed.get(id); if (table === "conversation" || table === "entry" || table === "document") { diff --git a/packages/durable/src/storage/sqlite/database.ts b/packages/durable/src/storage/sqlite/database.ts index e305aa3b7..f7dd4ed56 100644 --- a/packages/durable/src/storage/sqlite/database.ts +++ b/packages/durable/src/storage/sqlite/database.ts @@ -17,7 +17,11 @@ export interface SqliteStatement { * Queries and transaction callbacks are synchronous so the same storage core can * run on Node, Bun, and Cloudflare Durable Object SQLite. An adapter may return a * promise from `transaction` while it waits for the transaction to settle. - * Callers must not close the database while a returned settlement is pending. + * When the callback throws, the adapter must roll the transaction back before + * rethrowing that same error. If rollback fails, it must throw a different error + * (for example an `AggregateError`) so callers cannot mistake the callback error + * for a guaranteed rollback. Callers must not close the database while a + * returned settlement is pending. */ export interface SqliteDatabase { exec(sql: string): void; diff --git a/packages/durable/src/storage/sqlite/migrations.ts b/packages/durable/src/storage/sqlite/migrations.ts index 304d3f909..6f868510e 100644 --- a/packages/durable/src/storage/sqlite/migrations.ts +++ b/packages/durable/src/storage/sqlite/migrations.ts @@ -19,8 +19,12 @@ const INITIAL_SCHEMA: readonly string[] = [ ) STRICT`, `CREATE TABLE conversations ( id INTEGER PRIMARY KEY, + owner_conversation_id INTEGER, + owner_task_id INTEGER, record TEXT NOT NULL CHECK (json_valid(record)) ) STRICT`, + "CREATE INDEX conversations_by_owner_conversation ON conversations (owner_conversation_id, id)", + "CREATE INDEX conversations_by_owner_task ON conversations (owner_task_id, id)", `CREATE TABLE entries ( id INTEGER PRIMARY KEY, conversation_id INTEGER NOT NULL, @@ -77,7 +81,7 @@ const INITIAL_SCHEMA: readonly string[] = [ "CREATE INDEX document_revisions_by_kind ON document_revisions (document_id, kind, seq DESC)", ]; -/** Immutable, ordered schema history. Append new migrations; never edit released ones. */ +/** Immutable, ordered schema history. Append new migrations after the initial schema ships. */ export const SQLITE_MIGRATIONS: readonly SqliteMigration[] = [{ version: 1, statements: INITIAL_SCHEMA }]; export const CURRENT_SQLITE_SCHEMA_VERSION = SQLITE_MIGRATIONS.at(-1)?.version ?? 0; diff --git a/packages/durable/src/storage/sqlite/node.ts b/packages/durable/src/storage/sqlite/node.ts index 2300325a1..2844fde27 100644 --- a/packages/durable/src/storage/sqlite/node.ts +++ b/packages/durable/src/storage/sqlite/node.ts @@ -69,8 +69,8 @@ export class NodeSqliteDatabase implements SqliteDatabase { } catch (error) { try { this.database.exec("ROLLBACK"); - } catch { - // Preserve the error that caused the transaction to fail. + } catch (rollbackError) { + throw new AggregateError([error, rollbackError], "SQLite transaction failed and rollback failed"); } throw error; } diff --git a/packages/durable/src/storage/sqlite/storage.ts b/packages/durable/src/storage/sqlite/storage.ts index 99b35c5fb..9fffd6914 100644 --- a/packages/durable/src/storage/sqlite/storage.ts +++ b/packages/durable/src/storage/sqlite/storage.ts @@ -1,6 +1,8 @@ import type { Context, JsonValue } from "@earendil-works/chord"; import { apply, type Op } from "@earendil-works/chord/delta"; +import { StorageRejected } from "../../errors.ts"; import type { + ConversationQuery, ConversationRecord, Cursor, DocumentAddress, @@ -40,6 +42,7 @@ type IdRow = { readonly id: number }; type MetadataRow = { readonly next_id: string; readonly next_seq: number }; type DocumentAction = { create?: DocumentCreate; + copy?: Extract["source"]; content?: DocumentContent; retire: boolean; }; @@ -109,6 +112,7 @@ const writeId = (write: StorageWrite): Id | undefined => { case "submission": return write.value.id; case "document.create": + case "document.copy": return write.record.id; case "document.change": case "document.retire": @@ -213,16 +217,26 @@ export class SqliteStorage implements Storage { } async scanConversations( + query: ConversationQuery, limit: number, cursor: Cursor | undefined, _context: Context, ): Promise> { this.assertOpen(); - const after = cursorId(cursor) ?? -1; + const clauses = ["id > ?"]; + const params: SqliteValue[] = [cursorId(cursor) ?? -1]; + if (query.ownerConversationId !== undefined) { + clauses.push("owner_conversation_id = ?"); + params.push(query.ownerConversationId); + } + if (query.ownerTaskId !== undefined) { + clauses.push("owner_task_id = ?"); + params.push(query.ownerTaskId); + } + params.push(limit + 1); const rows = allRows( - this.db.prepare("SELECT record FROM conversations WHERE id > ? ORDER BY id LIMIT ?"), - after, - limit + 1, + this.db.prepare(`SELECT record FROM conversations WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), + ...params, ); return page( rows.map((row) => parseJson(row.record)), @@ -422,36 +436,7 @@ export class SqliteStorage implements Storage { async document(id: Id, at: DocumentPoint, _context: Context): Promise { this.assertOpen(); - const row = getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id); - if (row === undefined) return undefined; - const record = parseJson(row.record); - if (at !== "current" && isCurrentOnly(record)) { - throw new Error(`Document ${id} does not retain historical content`); - } - if (!isAliveAt(record, at)) return undefined; - const upper = at === "current" ? Number.MAX_SAFE_INTEGER : at; - const base = getRow( - this.db.prepare(`SELECT seq, kind, version, content FROM document_revisions - WHERE document_id = ? AND kind = 'base' AND seq <= ? ORDER BY seq DESC LIMIT 1`), - id, - upper, - ); - if (base === undefined) throw new Error(`Document ${id} is missing a required base`); - let value = parseJson(base.content); - const tail = allRows( - this.db.prepare(`SELECT seq, kind, version, content FROM document_revisions - WHERE document_id = ? AND seq > ? AND seq <= ? ORDER BY seq`), - id, - base.seq, - upper, - ); - for (const revision of tail) { - if (revision.kind !== "delta" || revision.version !== base.version) { - throw new Error(`Document ${id} crosses a stored version boundary without a base`); - } - value = apply(value, parseJson(revision.content)) as JsonObject; - } - return { record, version: base.version, value }; + return this.materializeDocument(id, at); } async scanDocuments( @@ -496,6 +481,39 @@ export class SqliteStorage implements Storage { return row === undefined ? undefined : parseJson(row.record); } + private materializeDocument(id: Id, at: DocumentPoint): StoredDocument | undefined { + const row = getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id); + if (row === undefined) return undefined; + const record = parseJson(row.record); + if (at !== "current" && isCurrentOnly(record)) { + throw new Error(`Document ${id} does not retain historical content`); + } + if (!isAliveAt(record, at)) return undefined; + const upper = at === "current" ? Number.MAX_SAFE_INTEGER : at; + const base = getRow( + this.db.prepare(`SELECT seq, kind, version, content FROM document_revisions + WHERE document_id = ? AND kind = 'base' AND seq <= ? ORDER BY seq DESC LIMIT 1`), + id, + upper, + ); + if (base === undefined) throw new Error(`Document ${id} is missing a required base`); + let value = parseJson(base.content); + const tail = allRows( + this.db.prepare(`SELECT seq, kind, version, content FROM document_revisions + WHERE document_id = ? AND seq > ? AND seq <= ? ORDER BY seq`), + id, + base.seq, + upper, + ); + for (const revision of tail) { + if (revision.kind !== "delta" || revision.version !== base.version) { + throw new Error(`Document ${id} crosses a stored version boundary without a base`); + } + value = apply(value, parseJson(revision.content)) as JsonObject; + } + return { record, version: base.version, value }; + } + private candidateNextId(writes: readonly StorageWrite[]): Id { let nextId = this.nextId; for (const write of writes) { @@ -510,8 +528,9 @@ export class SqliteStorage implements Storage { const lookup = this.db.prepare("SELECT record_type FROM record_ids WHERE id = ?"); for (const write of writes) { if (write.type === "document.change" || write.type === "document.retire") continue; - const table: TableName = write.type === "document.create" ? "document" : write.type; - const id = write.type === "document.create" ? write.record.id : write.value.id; + const document = write.type === "document.create" || write.type === "document.copy"; + const table: TableName = document ? "document" : write.type; + const id = document ? write.record.id : write.value.id; const existing = getRow(lookup, id)?.record_type; const earlier = claimed.get(id); if (table === "conversation" || table === "entry" || table === "document") { @@ -529,10 +548,15 @@ export class SqliteStorage implements Storage { private prepareDocumentActions(writes: readonly StorageWrite[]): Map { const actions = new Map(); for (const write of writes) { - if (write.type !== "document.create" && write.type !== "document.change" && write.type !== "document.retire") { + if ( + write.type !== "document.create" && + write.type !== "document.copy" && + write.type !== "document.change" && + write.type !== "document.retire" + ) { continue; } - const id = write.type === "document.create" ? write.record.id : write.id; + const id = write.type === "document.create" || write.type === "document.copy" ? write.record.id : write.id; let action = actions.get(id); if (action === undefined) { action = { retire: false }; @@ -540,14 +564,23 @@ export class SqliteStorage implements Storage { } switch (write.type) { case "document.create": - if (action.create !== undefined || action.content !== undefined) { + if (action.create !== undefined || action.content !== undefined || action.copy !== undefined) { throw new Error(`Document ${id} has more than one content command`); } action.create = write.record; action.content = write.content; break; + case "document.copy": + if (action.create !== undefined || action.content !== undefined || action.copy !== undefined) { + throw new Error(`Document ${id} has more than one content command`); + } + action.create = write.record; + action.copy = write.source; + break; case "document.change": - if (action.content !== undefined) throw new Error(`Document ${id} has more than one content command`); + if (action.content !== undefined || action.copy !== undefined) { + throw new Error(`Document ${id} has more than one content command`); + } action.content = write.content; break; case "document.retire": @@ -562,6 +595,9 @@ export class SqliteStorage implements Storage { private checkDocumentActions(actions: ReadonlyMap): void { const liveCounts = new Map(); for (const [id, action] of actions) { + if (action.copy !== undefined && actions.has(action.copy.id)) { + throw new StorageRejected(`Document copy ${id} source is changed in the copy batch`); + } const row = getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id); const existing = row === undefined ? undefined : parseJson(row.record); if (action.create === undefined && existing === undefined) throw new Error(`Unknown document: ${id}`); @@ -611,8 +647,15 @@ export class SqliteStorage implements Storage { case "conversation": this.claimId(write.value.id, "conversation"); this.db - .prepare("INSERT INTO conversations (id, record) VALUES (?, ?)") - .run(write.value.id, encodeJson(write.value)); + .prepare( + "INSERT INTO conversations (id, owner_conversation_id, owner_task_id, record) VALUES (?, ?, ?, ?)", + ) + .run( + write.value.id, + write.value.owner?.conversationId ?? null, + write.value.owner?.taskId ?? null, + encodeJson(write.value), + ); break; case "entry": this.claimId(write.value.id, "entry"); @@ -652,6 +695,7 @@ export class SqliteStorage implements Storage { ); break; case "document.create": + case "document.copy": case "document.change": case "document.retire": break; @@ -664,6 +708,28 @@ export class SqliteStorage implements Storage { private applyDocumentActions(actions: ReadonlyMap, seq: Seq): void { for (const [id, action] of actions) { + let content = action.content; + if (action.copy !== undefined) { + try { + const stored = this.materializeDocument(action.copy.id, action.copy.at); + if (stored === undefined) throw new Error(`Fork source document ${action.copy.id} cannot be read`); + const create = action.create!; + if ( + stored.record.scope.kind !== "conversation" || + create.scope.kind !== "conversation" || + stored.record.kind !== create.kind || + stored.record.key !== create.key || + stored.record.history !== create.history || + stored.record.fork !== create.fork + ) { + throw new Error(`Fork source document ${action.copy.id} does not match the copied record`); + } + content = { kind: "base", version: stored.version, value: stored.value }; + } catch (error) { + if (error instanceof StorageRejected) throw error; + throw new StorageRejected(`Document copy ${id} was rejected`, { cause: error }); + } + } let record: DocumentRecord; if (action.create !== undefined) { record = { @@ -693,17 +759,16 @@ export class SqliteStorage implements Storage { record = parseJson(row.record); } - if (action.content !== undefined) { - if (action.content.kind === "base" && isCurrentOnly(record)) { + if (content !== undefined) { + if (content.kind === "base" && isCurrentOnly(record)) { this.db.prepare("DELETE FROM document_revisions WHERE document_id = ?").run(id); } - const encodedContent = - action.content.kind === "base" ? encodeJson(action.content.value) : encodeJson(action.content.ops); + const encodedContent = content.kind === "base" ? encodeJson(content.value) : encodeJson(content.ops); this.db .prepare( "INSERT INTO document_revisions (document_id, seq, kind, version, content) VALUES (?, ?, ?, ?, ?)", ) - .run(id, seq, action.content.kind, action.content.version, encodedContent); + .run(id, seq, content.kind, content.version, encodedContent); } if (action.retire) { diff --git a/packages/durable/src/testing/storage-conformance.ts b/packages/durable/src/testing/storage-conformance.ts index 1b031958c..76004a13d 100644 --- a/packages/durable/src/testing/storage-conformance.ts +++ b/packages/durable/src/testing/storage-conformance.ts @@ -285,15 +285,65 @@ export function createStorageConformance(options: StorageConformanceOptions): re context, ); - const first = await storage.scanConversations(2, undefined, 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(2, roundTrippedCursor, context); + const second = await storage.scanConversations({}, 2, roundTrippedCursor, context); expect(second.items.map(({ id }) => id)).toEqual([thirdId]); expect(second.next).toBeUndefined(); }), + createCase(options, "filters and pages conversations by durable owner edges", async (storage) => { + const rootId = await createRoot(storage); + const otherOwnerId = await storage.mintId(); + const firstTaskId = await storage.mintId(); + const secondTaskId = await storage.mintId(); + const firstId = await storage.mintId(); + const secondId = await storage.mintId(); + const thirdId = await storage.mintId(); + await storage.commit( + [ + { type: "conversation", value: { id: otherOwnerId } }, + { + type: "conversation", + value: { id: firstId, owner: { conversationId: rootId, taskId: firstTaskId } }, + }, + { + type: "conversation", + value: { id: secondId, owner: { conversationId: rootId, taskId: secondTaskId } }, + }, + { + type: "conversation", + value: { id: thirdId, owner: { conversationId: otherOwnerId, taskId: firstTaskId } }, + }, + ], + context, + ); + + const first = await storage.scanConversations({ ownerConversationId: rootId }, 1, undefined, context); + expect(first.items.map(({ id }) => id)).toEqual([firstId]); + expect(first.next).toBeDefined(); + const second = await storage.scanConversations({ ownerConversationId: rootId }, 1, first.next, context); + expect(second.items.map(({ id }) => id)).toEqual([secondId]); + expect(second.next).toBeUndefined(); + expect( + (await storage.scanConversations({ ownerTaskId: firstTaskId }, 10, undefined, context)).items.map( + ({ id }) => id, + ), + ).toEqual([firstId, thirdId]); + expect( + ( + await storage.scanConversations( + { ownerConversationId: rootId, ownerTaskId: firstTaskId }, + 10, + undefined, + context, + ) + ).items.map(({ id }) => id), + ).toEqual([firstId]); + }), + createCase(options, "scans deep fork history newest-first through every ancestor cap", async (storage) => { const rootId = await createRoot(storage); const rootFirst = await storage.mintId(); @@ -790,6 +840,206 @@ export function createStorageConformance(options: StorageConformanceOptions): re expect((await storage.document(id, "current", context))?.value).toEqual(current); }), + createCase( + options, + "copies stored document bases independently and rejects ambiguous sources", + async (storage) => { + const rootId = await createRoot(storage); + const childId = await storage.mintId(); + const secondChildId = await storage.mintId(); + await storage.commit( + [ + { type: "conversation", value: { id: childId } }, + { type: "conversation", value: { id: secondChildId } }, + ], + context, + ); + const sourceId = await storage.mintId(); + const sourceRecord = { + id: sourceId, + kind: "copy.source", + scope: { kind: "conversation", conversationId: rootId }, + history: "rewindable", + fork: "asOf", + } satisfies DocumentCreate; + const createdAt = await storage.commit( + [ + { + type: "document.create", + record: sourceRecord, + content: { kind: "base", version: 2, value: { count: 1, rows: [{ value: "base" }] } }, + }, + ], + context, + ); + await storage.commit( + [ + { + type: "document.change", + id: sourceId, + content: { + kind: "delta", + version: 2, + ops: [ + ["s", ["count"], 2], + ["p", ["rows"], 1, 0, [{ value: "current" }]], + ], + }, + }, + ], + context, + ); + const historicalCopyId = await storage.mintId(); + const currentCopyId = await storage.mintId(); + const retiredCopyId = await storage.mintId(); + const childRecord = (id: Id, conversationId: Id): DocumentCreate => ({ + id, + kind: sourceRecord.kind, + scope: { kind: "conversation", conversationId }, + history: "rewindable", + fork: "asOf", + }); + await storage.commit( + [ + { + type: "document.copy", + record: childRecord(historicalCopyId, childId), + source: { id: sourceId, at: createdAt }, + }, + { + type: "document.copy", + record: childRecord(currentCopyId, secondChildId), + source: { id: sourceId, at: "current" }, + }, + { + type: "document.copy", + record: childRecord(retiredCopyId, rootId), + source: { id: sourceId, at: "current" }, + }, + { type: "document.retire", id: retiredCopyId }, + ], + context, + ); + expect(await storage.document(historicalCopyId, "current", context)).toMatchObject({ + version: 2, + value: { count: 1, rows: [{ value: "base" }] }, + }); + expect(await storage.document(currentCopyId, "current", context)).toMatchObject({ + version: 2, + value: { count: 2, rows: [{ value: "base" }, { value: "current" }] }, + }); + expect(await storage.document(retiredCopyId, "current", context)).toBeUndefined(); + + await storage.commit( + [ + { + type: "document.change", + id: sourceId, + content: { kind: "base", version: 2, value: { count: 99, rows: [] } }, + }, + { type: "document.retire", id: sourceId }, + ], + context, + ); + expect((await storage.document(currentCopyId, "current", context))?.value).toEqual({ + count: 2, + rows: [{ value: "base" }, { value: "current" }], + }); + + const latestSourceId = await storage.mintId(); + const latestCopyId = await storage.mintId(); + const latestSource = { + id: latestSourceId, + kind: "copy.latest", + scope: { kind: "conversation", conversationId: rootId }, + history: "latest", + fork: "current", + } satisfies DocumentCreate; + await storage.commit( + [ + { + type: "document.create", + record: latestSource, + content: { kind: "base", version: 4, value: { retained: "copy" } }, + }, + ], + context, + ); + await storage.commit( + [ + { + type: "document.copy", + record: { + ...latestSource, + id: latestCopyId, + scope: { kind: "conversation", conversationId: childId }, + }, + source: { id: latestSourceId, at: "current" }, + }, + ], + context, + ); + await storage.commit( + [ + { + type: "document.change", + id: latestSourceId, + content: { kind: "base", version: 4, value: { retained: "source-only" } }, + }, + { type: "document.retire", id: latestSourceId }, + ], + context, + ); + expect(await storage.document(latestCopyId, "current", context)).toMatchObject({ + version: 4, + value: { retained: "copy" }, + }); + + const conflictId = await storage.mintId(); + let conflictError: unknown; + try { + await storage.commit( + [ + { + type: "document.copy", + record: childRecord(conflictId, childId), + source: { id: currentCopyId, at: "current" }, + }, + { type: "document.retire", id: currentCopyId }, + ], + context, + ); + } catch (error) { + conflictError = error; + } + expect((conflictError as Error | undefined)?.name).toBe("StorageRejected"); + expect(await storage.document(conflictId, "current", context)).toBeUndefined(); + expect((await storage.document(currentCopyId, "current", context))?.value).toEqual({ + count: 2, + rows: [{ value: "base" }, { value: "current" }], + }); + + const mismatchId = await storage.mintId(); + let mismatchError: unknown; + try { + await storage.commit( + [ + { + type: "document.copy", + record: { ...childRecord(mismatchId, childId), kind: "copy.mismatch" }, + source: { id: currentCopyId, at: "current" }, + }, + ], + context, + ); + } catch (error) { + mismatchError = error; + } + expect((mismatchError as Error | undefined)?.name).toBe("StorageRejected"); + expect(await storage.document(mismatchId, "current", context)).toBeUndefined(); + }, + ), + createCase( options, "uses bases for version transitions and rejects historical reads of current-only documents", diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index ce7a34dc4..e5f0a3c94 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -143,7 +143,7 @@ export type ConversationRecord = { readonly conversationId: Id; readonly at: Id; }; - /** Creator edge used for authorization, subtree abort, and subtree idle waits. */ + /** Creator edge used for attribution, subtree abort, and subtree idle waits. */ readonly owner?: { readonly conversationId: Id; readonly taskId: Id; @@ -426,6 +426,12 @@ export type Page = { /** Backend-owned JSON continuation state that callers only round-trip to the same scan. */ export type Cursor = Readonly>; +/** Optional filters for an ordered conversation scan. */ +export type ConversationQuery = { + readonly ownerConversationId?: Id; + readonly ownerTaskId?: Id; +}; + /** Inclusive ID bounds for a newest-first scan of one conversation's fork-aware history. */ export type EntryQuery = { readonly conversationId: Id; @@ -475,6 +481,12 @@ export type DocumentContent = readonly ops: readonly Op[]; }; +/** Exact persisted source selected for a definition-free document copy. */ +export type DocumentCopySource = { + readonly id: Id; + readonly at: DocumentPoint; +}; + /** Detached materialized value and stored definition version at a selected point. */ export type StoredDocument = { readonly record: DocumentRecord; @@ -493,6 +505,7 @@ export type StorageWrite = readonly record: DocumentCreate; readonly content: Extract; } + | { readonly type: "document.copy"; readonly record: DocumentCreate; readonly source: DocumentCopySource } | { readonly type: "document.change"; readonly id: Id; @@ -508,7 +521,11 @@ export interface Tx { conversation(id: Id): Promise; entry(id: Id): Promise; task(id: Id): Promise | undefined>; - scanConversations(limit: number, cursor?: Cursor): Promise>; + scanConversations( + query: ConversationQuery, + limit: number, + cursor?: Cursor, + ): Promise>; scanEntries(query: EntryQuery, limit: number, cursor?: Cursor): Promise>; scanTasks( query: TaskQuery, @@ -516,8 +533,10 @@ export interface Tx { cursor?: Cursor, ): Promise, Cursor>>; - /** Returned records are Session-owned immutable values and may be shared with commit listeners. */ - createConversation(value: Omit): Promise; + /** Create an independent ownerless conversation. */ + createConversation(): Promise; + /** Create an ownerless history fork at one concrete visible entry. */ + forkConversation(parentConversationId: Id, at: Id): Promise; /** Returned records are Session-owned immutable values and may be shared with commit listeners. */ appendEntry(conversationId: Id, value: EntryDraft): Promise; createTask( @@ -638,6 +657,7 @@ export interface Storage { /** Scan conversations in ascending ID order. */ scanConversations( + query: ConversationQuery, limit: number, cursor: Cursor | undefined, context: Context, diff --git a/packages/durable/test/jsonl-storage.test.ts b/packages/durable/test/jsonl-storage.test.ts index 3947a467c..49d718649 100644 --- a/packages/durable/test/jsonl-storage.test.ts +++ b/packages/durable/test/jsonl-storage.test.ts @@ -66,8 +66,8 @@ class ReopeningStorage implements Storage { mintId: Storage["mintId"] = () => this.current.mintId(); conversation: Storage["conversation"] = (id, readContext) => this.current.conversation(id, readContext); - scanConversations: Storage["scanConversations"] = (limit, cursor, readContext) => - this.current.scanConversations(limit, cursor, readContext); + scanConversations: Storage["scanConversations"] = (query, limit, cursor, readContext) => + this.current.scanConversations(query, limit, cursor, readContext); entry(id: number, readContext: Context): ReturnType; entry(conversationId: number, id: number, readContext: Context): ReturnType; entry(idOrConversationId: number, idOrContext: number | Context, readContext?: Context) { diff --git a/packages/durable/test/session-checkpoints-migrations.test.ts b/packages/durable/test/session-checkpoints-migrations.test.ts index eccfb279f..2a3a996b0 100644 --- a/packages/durable/test/session-checkpoints-migrations.test.ts +++ b/packages/durable/test/session-checkpoints-migrations.test.ts @@ -624,12 +624,7 @@ describe("Session historical document snapshots", () => { }); const childId = await session.commit( - async (tx) => - ( - await tx.createConversation({ - parent: { conversationId, at: secondEntry }, - }) - ).id, + async (tx) => (await tx.forkConversation(conversationId, secondEntry)).id, context, ); expect(await session.snapshotAsOf(V3, childId, firstEntry, context)).toEqual({ count: 1, version: 3 }); diff --git a/packages/durable/test/session-forks.test.ts b/packages/durable/test/session-forks.test.ts new file mode 100644 index 000000000..a490793dd --- /dev/null +++ b/packages/durable/test/session-forks.test.ts @@ -0,0 +1,641 @@ +import { defineDoc, defineDocFamily, type Id, StorageRejected, type StorageWrite } from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { + context, + createConversation, + documentChanges, + documentCopyChanges, + flush, + openTestSession, +} from "./session-support.ts"; + +function documentCreates(writes: readonly StorageWrite[]) { + return writes.filter((write) => write.type === "document.create"); +} + +function documentCopies(writes: readonly StorageWrite[]) { + return writes.filter((write) => write.type === "document.copy"); +} + +describe("Session conversation document forks", () => { + it("copies as-of and current singleton and family bases while leaving initial documents absent", async () => { + const AsOf = defineDoc<{ value: string }>({ + kind: "fork.policies.as-of", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ value: "as-initial" }), + }); + const Current = defineDoc<{ value: string }>({ + kind: "fork.policies.current", + version: 1, + scope: "conversation", + history: "latest", + fork: "current", + initial: () => ({ value: "current-initial" }), + }); + const Initial = defineDoc<{ value: string }>({ + kind: "fork.policies.initial", + version: 1, + scope: "conversation", + history: "latest", + fork: "initial", + initial: () => ({ value: "fresh" }), + }); + const AsOfFamily = defineDocFamily<{ value: string }, string>({ + kind: "fork.policies.as-of-family", + version: 1, + family: true, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: (seed) => ({ value: seed }), + }); + const CurrentFamily = defineDocFamily<{ value: string }, string>({ + kind: "fork.policies.current-family", + version: 1, + family: true, + scope: "conversation", + history: "latest", + fork: "current", + initial: (seed) => ({ value: seed }), + }); + const { session, storage, publications } = openTestSession(); + const parentId = await createConversation(session); + let forkAt!: Id; + await session.commit(async (tx) => { + forkAt = (await tx.appendEntry(parentId, { kind: "fork-point" })).id; + (await tx.doc(AsOf, parentId)).value = "as-at-fork"; + (await tx.doc(Current, parentId)).value = "current-at-fork"; + (await tx.doc(Initial, parentId)).value = "parent-only"; + (await tx.doc(AsOfFamily, parentId, "a", "unused")).value = "family-as-a"; + (await tx.doc(AsOfFamily, parentId, "b", "unused")).value = "family-as-b"; + (await tx.doc(CurrentFamily, parentId, "a", "unused")).value = "family-current-a"; + }, context); + await session.commit(async (tx) => { + (await tx.doc(AsOf, parentId)).value = "as-after-fork"; + (await tx.doc(Current, parentId)).value = "current-when-copied"; + (await tx.doc(AsOfFamily, parentId, "a", "unused")).value = "family-as-after"; + (await tx.doc(CurrentFamily, parentId, "a", "unused")).value = "family-current-when-copied"; + }, context); + await flush(); + const documentReads = storage.documentReadCount; + + const child = await session.commit((tx) => tx.forkConversation(parentId, forkAt), context); + await flush(); + expect(storage.documentReadCount).toBe(documentReads); + + expect(await session.snapshot(AsOf, child.id, context)).toEqual({ value: "as-at-fork" }); + expect(await session.snapshot(Current, child.id, context)).toEqual({ value: "current-when-copied" }); + expect(await session.snapshot(Initial, child.id, context)).toBeUndefined(); + expect(await session.snapshot(AsOfFamily, child.id, "a", context)).toEqual({ value: "family-as-a" }); + expect(await session.snapshot(AsOfFamily, child.id, "b", context)).toEqual({ value: "family-as-b" }); + expect(await session.snapshot(CurrentFamily, child.id, "a", context)).toEqual({ + value: "family-current-when-copied", + }); + + const admitted = storage.admittedCommits.at(-1)!; + const copies = documentCopies(admitted); + expect(copies).toHaveLength(5); + expect(copies.every((write) => write.record.scope.kind === "conversation")).toBe(true); + expect( + copies.every( + (write) => write.record.scope.kind === "conversation" && write.record.scope.conversationId === child.id, + ), + ).toBe(true); + const publication = publications.at(-1)!; + const copied = documentCopyChanges(publication); + expect(copied).toHaveLength(5); + for (const change of copied) { + expect(change.record.createdAt).toBe(publication.seq); + expect(change.conversationId).toBe(child.id); + const write = copies.find((candidate) => candidate.record.id === change.record.id)!; + expect(change.source).toEqual(write.source); + } + + const parentRecord = await storage.findDocument( + { kind: AsOf.definition.kind, scope: { kind: "conversation", conversationId: parentId } }, + "current", + context, + ); + const childRecord = await storage.findDocument( + { kind: AsOf.definition.kind, scope: { kind: "conversation", conversationId: child.id } }, + "current", + context, + ); + expect(childRecord!.id).not.toBe(parentRecord!.id); + + await session.commit(async (tx) => { + (await tx.doc(AsOf, child.id)).value = "child-independent"; + }, context); + expect(await session.snapshot(AsOf, parentId, context)).toEqual({ value: "as-after-fork" }); + expect(await session.snapshot(AsOf, child.id, context)).toEqual({ value: "child-independent" }); + await session.commit(async (tx) => { + (await tx.doc(Initial, child.id)).value = "child-created"; + }, context); + expect(await session.snapshot(Initial, child.id, context)).toEqual({ value: "child-created" }); + }); + + it("uses final document state from the fork entry commit while excluding later same-commit entries", async () => { + const Doc = defineDoc<{ value: string }>({ + kind: "fork.same-commit", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ value: "initial" }), + }); + const { session } = openTestSession(); + const parentId = await createConversation(session); + let forkAt!: Id; + let excluded!: Id; + await session.commit(async (tx) => { + forkAt = (await tx.appendEntry(parentId, { kind: "included" })).id; + (await tx.doc(Doc, parentId)).value = "final-state-of-commit"; + excluded = (await tx.appendEntry(parentId, { kind: "excluded" })).id; + }, context); + const child = await session.commit((tx) => tx.forkConversation(parentId, forkAt), context); + expect(await session.snapshot(Doc, child.id, context)).toEqual({ value: "final-state-of-commit" }); + const visible = await session.commit((tx) => tx.scanEntries({ conversationId: child.id }, 10), context); + expect(visible.items.map(({ id }) => id)).toContain(forkAt); + expect(visible.items.map(({ id }) => id)).not.toContain(excluded); + }); + + it("selects the entry-owning ancestor for as-of copies and the immediate parent for current copies", async () => { + const AsOf = defineDoc<{ value: string }>({ + kind: "fork.ancestry.as-of", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ value: "initial" }), + }); + const Current = defineDoc<{ value: string }>({ + kind: "fork.ancestry.current", + version: 1, + scope: "conversation", + history: "latest", + fork: "current", + initial: () => ({ value: "initial" }), + }); + const { session } = openTestSession(); + const rootId = await createConversation(session); + let inherited!: Id; + await session.commit(async (tx) => { + inherited = (await tx.appendEntry(rootId, { kind: "root" })).id; + (await tx.doc(AsOf, rootId)).value = "root-at-entry"; + (await tx.doc(Current, rootId)).value = "root-current"; + }, context); + const parent = await session.commit((tx) => tx.forkConversation(rootId, inherited), context); + let parentEntry!: Id; + await session.commit(async (tx) => { + parentEntry = (await tx.appendEntry(parent.id, { kind: "parent" })).id; + (await tx.doc(AsOf, parent.id)).value = "parent-at-own-entry"; + (await tx.doc(Current, parent.id)).value = "parent-current"; + }, context); + + const inheritedFork = await session.commit((tx) => tx.forkConversation(parent.id, inherited), context); + expect(await session.snapshot(AsOf, inheritedFork.id, context)).toEqual({ value: "root-at-entry" }); + expect(await session.snapshot(Current, inheritedFork.id, context)).toEqual({ value: "parent-current" }); + + const ownEntryFork = await session.commit((tx) => tx.forkConversation(parent.id, parentEntry), context); + expect(await session.snapshot(AsOf, ownEntryFork.id, context)).toEqual({ value: "parent-at-own-entry" }); + expect(await session.snapshot(Current, ownEntryFork.id, context)).toEqual({ value: "parent-current" }); + }); + + it("copies the stored value and version without consulting migration definitions or migrated caches", async () => { + const V1 = defineDoc<{ count: number }>({ + kind: "fork.stored-version", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ count: 1 }), + }); + let migrations = 0; + const V3 = defineDoc<{ count: number; migrated: boolean }>({ + kind: "fork.stored-version", + version: 3, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ count: 0, migrated: false }), + migrate: (value, fromVersion) => { + expect(fromVersion).toBe(1); + migrations++; + return { count: value.count as number, migrated: true }; + }, + }); + const { session, storage } = openTestSession(); + const parentId = await createConversation(session); + let forkAt!: Id; + await session.commit(async (tx) => { + forkAt = (await tx.appendEntry(parentId, { kind: "point" })).id; + await tx.doc(V1, parentId); + }, context); + await session.unloadDocuments(); + expect(await session.snapshot(V3, parentId, context)).toEqual({ count: 1, migrated: true }); + expect(migrations).toBe(1); + + const child = await session.commit((tx) => tx.forkConversation(parentId, forkAt), context); + expect(migrations).toBe(1); + const childRecord = await storage.findDocument( + { kind: V1.definition.kind, scope: { kind: "conversation", conversationId: child.id } }, + "current", + context, + ); + const childStored = await storage.document(childRecord!.id, "current", context); + expect(childStored).toMatchObject({ version: 1, value: { count: 1 } }); + expect(await session.snapshot(V3, child.id, context)).toEqual({ count: 1, migrated: true }); + expect(migrations).toBe(2); + }); + + it("coalesces a typed migration and override into the copied creation base", async () => { + const V1 = defineDoc<{ count: number }>({ + kind: "fork.override", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ count: 2 }), + }); + let checkpoints = 0; + const V2 = defineDoc<{ count: number; migrated: boolean }>({ + kind: "fork.override", + version: 2, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ count: 0, migrated: false }), + migrate: (value) => ({ count: value.count as number, migrated: true }), + checkpointWhen: () => { + checkpoints++; + return false; + }, + }); + const { session, storage, publications } = openTestSession(); + const parentId = await createConversation(session); + let forkAt!: Id; + await session.commit(async (tx) => { + forkAt = (await tx.appendEntry(parentId, { kind: "point" })).id; + await tx.doc(V1, parentId); + }, context); + + const documentReads = storage.documentReadCount; + const child = await session.commit(async (tx) => { + const created = await tx.forkConversation(parentId, forkAt); + const value = await tx.doc(V2, created.id); + value.count = 9; + return created; + }, context); + await flush(); + expect(storage.documentReadCount).toBe(documentReads + 1); + const creates = documentCreates(storage.admittedCommits.at(-1)!); + expect(creates).toHaveLength(1); + expect(creates[0]!.content).toMatchObject({ + kind: "base", + version: 2, + value: { count: 9, migrated: true }, + }); + expect(checkpoints).toBe(0); + expect(documentChanges(publications.at(-1)!)).toMatchObject([{ version: 2 }]); + expect(await session.snapshot(V2, child.id, context)).toEqual({ count: 9, migrated: true }); + const parentRecord = await storage.findDocument( + { kind: V1.definition.kind, scope: { kind: "conversation", conversationId: parentId } }, + "current", + context, + ); + expect((await storage.document(parentRecord!.id, "current", context))!.version).toBe(1); + }); + + it("copies the incarnation alive at the fork point across retirement and recreation", async () => { + const Doc = defineDoc<{ value: string }>({ + kind: "fork.incarnations", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ value: "initial" }), + }); + const { session } = openTestSession(); + const parentId = await createConversation(session); + let oldAt!: Id; + let retiredAt!: Id; + let newAt!: Id; + await session.commit(async (tx) => { + oldAt = (await tx.appendEntry(parentId, { kind: "old" })).id; + (await tx.doc(Doc, parentId)).value = "old"; + }, context); + await session.commit(async (tx) => { + retiredAt = (await tx.appendEntry(parentId, { kind: "retired" })).id; + await tx.retireDoc(Doc, parentId); + }, context); + await session.commit(async (tx) => { + newAt = (await tx.appendEntry(parentId, { kind: "new" })).id; + (await tx.doc(Doc, parentId)).value = "new"; + }, context); + + const oldChild = await session.commit((tx) => tx.forkConversation(parentId, oldAt), context); + const emptyChild = await session.commit((tx) => tx.forkConversation(parentId, retiredAt), context); + const newChild = await session.commit((tx) => tx.forkConversation(parentId, newAt), context); + expect(await session.snapshot(Doc, oldChild.id, context)).toEqual({ value: "old" }); + expect(await session.snapshot(Doc, emptyChild.id, context)).toBeUndefined(); + expect(await session.snapshot(Doc, newChild.id, context)).toEqual({ value: "new" }); + }); + + it("rejects invisible fork points before admission and remains usable", async () => { + const { session, storage, publications } = openTestSession(); + const rootId = await createConversation(session); + let visible!: Id; + let hidden!: Id; + await session.commit(async (tx) => { + visible = (await tx.appendEntry(rootId, { kind: "visible" })).id; + hidden = (await tx.appendEntry(rootId, { kind: "hidden" })).id; + }, context); + const parent = await session.commit((tx) => tx.forkConversation(rootId, visible), context); + await flush(); + const commits = storage.commits.length; + const published = publications.length; + await expect(session.commit((tx) => tx.forkConversation(parent.id, hidden), context)).rejects.toThrow( + `Entry ${hidden} is not visible`, + ); + await flush(); + expect(storage.commits).toHaveLength(commits); + expect(publications).toHaveLength(published); + const independent = await session.commit((tx) => tx.createConversation(), context); + expect(await storage.conversation(independent.id, context)).toEqual(independent); + }); + + it("rejects duplicate as-of and current selections for one child address before admission", async () => { + const AsOf = defineDoc<{ value: string }>({ + kind: "fork.duplicate-policy", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ value: "as-of" }), + }); + const Current = defineDoc<{ value: string }>({ + kind: "fork.duplicate-policy", + version: 1, + scope: "conversation", + history: "latest", + fork: "current", + initial: () => ({ value: "current" }), + }); + const { session, storage } = openTestSession(); + const parentId = await createConversation(session); + let oldAt!: Id; + await session.commit(async (tx) => { + oldAt = (await tx.appendEntry(parentId, { kind: "old" })).id; + await tx.doc(AsOf, parentId); + }, context); + await session.commit((tx) => tx.retireDoc(AsOf, parentId), context); + await session.commit((tx) => tx.doc(Current, parentId).then(() => undefined), context); + const commits = storage.commits.length; + await expect(session.commit((tx) => tx.forkConversation(parentId, oldAt), context)).rejects.toThrow( + "Fork selects multiple source documents", + ); + expect(storage.commits).toHaveLength(commits); + const independent = await session.commit((tx) => tx.createConversation(), context); + expect(await storage.conversation(independent.id, context)).toEqual(independent); + }); + + it("rejects current and as-of source writes in the fork transaction", async () => { + const Current = defineDoc<{ value: string }>({ + kind: "fork.same-transaction-current", + version: 1, + scope: "conversation", + history: "latest", + fork: "current", + initial: () => ({ value: "committed" }), + }); + const AsOf = defineDoc<{ value: string }>({ + kind: "fork.same-transaction-as-of", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ value: "committed" }), + }); + const { session, storage } = openTestSession(); + const parentId = await createConversation(session); + let forkAt!: Id; + await session.commit(async (tx) => { + forkAt = (await tx.appendEntry(parentId, { kind: "point" })).id; + await tx.doc(Current, parentId); + await tx.doc(AsOf, parentId); + }, context); + const commits = storage.commits.length; + await expect( + session.commit(async (tx) => { + (await tx.doc(Current, parentId)).value = "before-fork"; + await tx.forkConversation(parentId, forkAt); + }, context), + ).rejects.toThrow("Cannot change fork source document"); + await expect( + session.commit(async (tx) => { + await tx.forkConversation(parentId, forkAt); + (await tx.doc(Current, parentId)).value = "after-fork"; + }, context), + ).rejects.toThrow("Cannot change fork source document"); + await expect( + session.commit(async (tx) => { + (await tx.doc(AsOf, parentId)).value = "as-of-write"; + await tx.forkConversation(parentId, forkAt); + }, context), + ).rejects.toThrow("Cannot change fork source document"); + expect(storage.commits).toHaveLength(commits); + expect(await session.snapshot(Current, parentId, context)).toEqual({ value: "committed" }); + const child = await session.commit((tx) => tx.forkConversation(parentId, forkAt), context); + expect(await session.snapshot(Current, child.id, context)).toEqual({ value: "committed" }); + }); + + it("rolls every copied base back when later pre-admission assembly fails", async () => { + const Copied = defineDoc<{ value: string }>({ + kind: "fork.rollback.copied", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ value: "copied" }), + }); + let failCheckpoint = true; + const Failure = defineDoc<{ count: number }>({ + kind: "fork.rollback.failure", + version: 1, + scope: "session", + initial: () => ({ count: 0 }), + checkpointWhen: () => { + if (failCheckpoint) throw new Error("checkpoint failed"); + return false; + }, + }); + const { session, storage, publications } = openTestSession(); + const parentId = await createConversation(session); + let forkAt!: Id; + await session.commit(async (tx) => { + forkAt = (await tx.appendEntry(parentId, { kind: "point" })).id; + await tx.doc(Copied, parentId); + await tx.doc(Failure); + }, context); + await flush(); + const commits = storage.commits.length; + const published = publications.length; + let childId!: Id; + await expect( + session.commit(async (tx) => { + childId = (await tx.forkConversation(parentId, forkAt)).id; + (await tx.doc(Failure)).count = 1; + }, context), + ).rejects.toThrow("checkpoint failed"); + await flush(); + expect(storage.commits).toHaveLength(commits); + expect(publications).toHaveLength(published); + expect(await storage.conversation(childId, context)).toBeUndefined(); + failCheckpoint = false; + await session.commit(async (tx) => { + (await tx.doc(Failure)).count = 2; + }, context); + expect(await session.snapshot(Failure, context)).toEqual({ count: 2 }); + expect(await session.snapshot(Copied, parentId, context)).toEqual({ value: "copied" }); + }); + + it("rolls back a guaranteed Storage rejection without poisoning the Session", async () => { + const Doc = defineDoc<{ value: string }>({ + kind: "fork.storage-rejected", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ value: "source" }), + }); + const { session, storage } = openTestSession(); + const parentId = await createConversation(session); + let forkAt!: Id; + await session.commit(async (tx) => { + forkAt = (await tx.appendEntry(parentId, { kind: "point" })).id; + await tx.doc(Doc, parentId); + }, context); + let rejectedChildId!: Id; + storage.failNextCommit(new StorageRejected("copy rejected")); + await expect( + session.commit(async (tx) => { + rejectedChildId = (await tx.forkConversation(parentId, forkAt)).id; + }, context), + ).rejects.toThrow("copy rejected"); + expect(await storage.conversation(rejectedChildId, context)).toBeUndefined(); + const next = await session.commit((tx) => tx.createConversation(), context); + expect(await storage.conversation(next.id, context)).toEqual(next); + }); + + it("retires a copied document and can recreate the address in the fork transaction", async () => { + const Doc = defineDoc<{ value: string }>({ + kind: "fork.retire-recreate", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ value: "fresh" }), + }); + const { session, storage } = openTestSession(); + const parentId = await createConversation(session); + let forkAt!: Id; + await session.commit(async (tx) => { + forkAt = (await tx.appendEntry(parentId, { kind: "point" })).id; + (await tx.doc(Doc, parentId)).value = "copied"; + }, context); + const child = await session.commit(async (tx) => { + const created = await tx.forkConversation(parentId, forkAt); + await tx.retireDoc(Doc, created.id); + (await tx.doc(Doc, created.id)).value = "replacement"; + return created; + }, context); + const writes = storage.commits.at(-1)!; + expect(writes.filter((write) => write.type === "document.copy")).toHaveLength(1); + expect(writes.filter((write) => write.type === "document.create")).toHaveLength(1); + expect(writes.filter((write) => write.type === "document.retire")).toHaveLength(1); + expect(await session.snapshot(Doc, child.id, context)).toEqual({ value: "replacement" }); + }); + + it("copies only conversation documents, leaving Session and task documents in their original scopes", async () => { + const Copied = defineDoc<{ value: string }>({ + kind: "fork.scope.conversation", + version: 1, + scope: "conversation", + history: "latest", + fork: "current", + initial: () => ({ value: "conversation" }), + }); + const SessionOnly = defineDoc<{ value: string }>({ + kind: "fork.scope.session", + version: 1, + scope: "session", + initial: () => ({ value: "session" }), + }); + const TaskOnly = defineDoc<{ value: string }>({ + kind: "fork.scope.task", + version: 1, + scope: "task", + initial: () => ({ value: "task" }), + }); + const Work = { + definition: { + name: "fork.scope.work", + version: 1, + initial: () => ({ phase: "start" }), + }, + }; + const { session, storage, publications } = openTestSession(); + const parentId = await createConversation(session); + let forkAt!: Id; + let taskId!: Id; + await session.commit(async (tx) => { + forkAt = (await tx.appendEntry(parentId, { kind: "point" })).id; + taskId = (await tx.createTask(Work, null, { conversationId: parentId })).id; + await tx.doc(Copied, parentId); + await tx.doc(SessionOnly); + await tx.doc(TaskOnly, taskId); + }, context); + const child = await session.commit((tx) => tx.forkConversation(parentId, forkAt), context); + await flush(); + const copies = documentCopies(storage.commits.at(-1)!); + expect(copies.map((write) => write.record.kind)).toEqual([Copied.definition.kind]); + expect(documentCopyChanges(publications.at(-1)!).map((change) => change.record.kind)).toEqual([ + Copied.definition.kind, + ]); + expect(await session.snapshot(Copied, child.id, context)).toEqual({ value: "conversation" }); + expect(await session.snapshot(SessionOnly, context)).toEqual({ value: "session" }); + expect(await session.snapshot(TaskOnly, taskId, context)).toEqual({ value: "task" }); + }); + + it("copies every family member across storage scan pages", async () => { + const Family = defineDocFamily<{ value: number }, number>({ + kind: "fork.pagination", + version: 1, + family: true, + scope: "conversation", + history: "latest", + fork: "current", + initial: (seed) => ({ value: seed }), + }); + const { session, storage } = openTestSession(); + const parentId = await createConversation(session); + const forkAt = await session.commit( + async (tx) => (await tx.appendEntry(parentId, { kind: "point" })).id, + context, + ); + await session.commit(async (tx) => { + await Promise.all( + Array.from({ length: 260 }, async (_, index) => { + await tx.doc(Family, parentId, `member-${index}`, index); + }), + ); + }, context); + const child = await session.commit((tx) => tx.forkConversation(parentId, forkAt), context); + expect(documentCopies(storage.commits.at(-1)!)).toHaveLength(260); + expect(await session.snapshot(Family, child.id, "member-0", context)).toEqual({ value: 0 }); + expect(await session.snapshot(Family, child.id, "member-259", context)).toEqual({ value: 259 }); + }); +}); diff --git a/packages/durable/test/session-support.ts b/packages/durable/test/session-support.ts index e0411de1c..aecf7b7b8 100644 --- a/packages/durable/test/session-support.ts +++ b/packages/durable/test/session-support.ts @@ -38,6 +38,7 @@ export class ControlledStorage extends MemoryStorage { /** Detached batches for value assertions. */ readonly commits: (readonly StorageWrite[])[] = []; mintCount = 0; + documentReadCount = 0; #commitGate: { gate: Deferred; entered: Deferred } | undefined; #findGate: { gate: Deferred; entered: Deferred } | undefined; #commitFailure: Error | undefined; @@ -85,6 +86,11 @@ export class ControlledStorage extends MemoryStorage { return super.mintId(); } + override document(id: Id, at: DocumentPoint, callContext: Context) { + this.documentReadCount++; + return super.document(id, at, callContext); + } + override async findDocument( address: DocumentAddress, at: DocumentPoint, @@ -114,13 +120,26 @@ export function openTestSession(): { return { storage, session, publications }; } -export function documentChanges(publication: CommitPublication): readonly DocumentCommitChange[] { - return publication.changes.filter((change): change is DocumentCommitChange => change.type === "document"); +export function documentChanges( + publication: CommitPublication, +): readonly Extract[] { + return publication.changes.filter( + (change): change is Extract => change.type === "document", + ); +} + +export function documentCopyChanges( + publication: CommitPublication, +): readonly Extract[] { + return publication.changes.filter( + (change): change is Extract => + change.type === "document.copy", + ); } /** Create one conversation and return its ID. */ export async function createConversation(session: SessionKernel): Promise { - return session.commit(async (tx) => (await tx.createConversation({})).id, context); + return session.commit(async (tx) => (await tx.createConversation()).id, context); } /** Resolve after pending microtasks and one macrotask turn. */ diff --git a/packages/durable/test/session-tables.test.ts b/packages/durable/test/session-tables.test.ts index de9fa99e2..eddfb742d 100644 --- a/packages/durable/test/session-tables.test.ts +++ b/packages/durable/test/session-tables.test.ts @@ -74,14 +74,14 @@ describe("Session transaction tables", () => { 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.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.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"; @@ -97,10 +97,10 @@ describe("Session transaction tables", () => { await createConversation(session), ]; await session.commit(async (tx) => { - const first = await tx.scanConversations(2); + 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); + const second = await tx.scanConversations({}, 2, first.next); expect(second.items.map(({ id }) => id)).toEqual(ids.slice(2)); expect(second.next).toBeUndefined(); }, context); @@ -120,7 +120,7 @@ describe("Session transaction tables", () => { 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 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( @@ -245,7 +245,7 @@ describe("Session transaction tables", () => { for (const document of documents) expect(document.conversationId).toBe(conversationId); await session.commit(async (tx) => { - await tx.createConversation({}); + await tx.createConversation(); (await tx.doc(ProgressDoc, taskId)).lines.push("committed task"); }, context); await flush(); diff --git a/packages/durable/test/sqlite-facade.test.ts b/packages/durable/test/sqlite-facade.test.ts index 206f71ba3..a908174b4 100644 --- a/packages/durable/test/sqlite-facade.test.ts +++ b/packages/durable/test/sqlite-facade.test.ts @@ -1,5 +1,6 @@ import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; import { describe, expect, it } from "vitest"; +import { StorageRejected } from "../src/errors.ts"; import type { SqliteDatabase, SqliteStatement } from "../src/storage/sqlite/index.ts"; import { SqliteStorage } from "../src/storage/sqlite/index.ts"; import { type NodeSqliteDatabase, openNodeSqliteDatabase } from "../src/storage/sqlite/node.ts"; @@ -107,6 +108,20 @@ describe("portable SQLite facade settlement", () => { database.close(); }); + it("does not preserve a guaranteed rejection when rollback itself fails", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + database.exec("CREATE TABLE rollback_probe (value INTEGER)"); + expect(() => + database.transaction(() => { + database.exec("INSERT INTO rollback_probe (value) VALUES (1)"); + database.exec("COMMIT"); + throw new StorageRejected("rejected after an escaped commit"); + }), + ).toThrow(AggregateError); + expect(database.prepare("SELECT value FROM rollback_probe").get()).toEqual({ value: 1 }); + database.close(); + }); + it("awaits async transaction settlement and adopts IDs only after success", async () => { const database = new ControlledSettlementDatabase(await openNodeSqliteDatabase(":memory:")); database.controlNextSettlement("delay"); diff --git a/packages/durable/test/sqlite-storage.test.ts b/packages/durable/test/sqlite-storage.test.ts index 2df0938d2..47441211a 100644 --- a/packages/durable/test/sqlite-storage.test.ts +++ b/packages/durable/test/sqlite-storage.test.ts @@ -55,8 +55,8 @@ class ReopeningStorage implements Storage { mintId: Storage["mintId"] = () => this.current.mintId(); conversation: Storage["conversation"] = (id, readContext) => this.current.conversation(id, readContext); - scanConversations: Storage["scanConversations"] = (limit, cursor, readContext) => - this.current.scanConversations(limit, cursor, readContext); + scanConversations: Storage["scanConversations"] = (query, limit, cursor, readContext) => + this.current.scanConversations(query, limit, cursor, readContext); entry(id: number, readContext: Context): ReturnType; entry(conversationId: number, id: number, readContext: Context): ReturnType; entry(idOrConversationId: number, idOrContext: number | Context, readContext?: Context) {