sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(durable): add conversation document forks

Browse source
This commit is contained in:
Mario Zechner 2026-09-25 16:17:23 +02:00
parent 8930b9ec0b
commit 507d7649e9
23 changed files with 1644 additions and 170 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/durable/CHANGELOG.md
View file

@ -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

165
packages/durable/docs/pico-v5.md
View file

@ -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<JsonValue>[];
readonly now?: () => number;
readonly root?: Omit<ConversationSpec, "parent">;
readonly root?: ConversationSpec;
readonly onReport?: (error: unknown) => void;
};
@ -372,7 +371,7 @@ interface Conversation {
): Promise<Page<EntryRecord, Cursor>>;
fork(
at: Id,
spec: Omit<ConversationSpec, "parent">,
spec: ConversationSpec,
context: Context,
): Promise<Conversation>;
collapse(instructions: string | undefined, context: Context): Promise<Id>;
@ -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<ConversationRecord | undefined>;
entry(id: Id): Promise<EntryRecord | undefined>;
task(id: Id): Promise<TaskRecord<JsonValue, JsonValue, JsonValue> | undefined>;
scanConversations(limit: number, cursor?: Cursor): Promise<Page<ConversationRecord, Cursor>>;
scanConversations(query: ConversationQuery, limit: number, cursor?: Cursor): Promise<Page<ConversationRecord, Cursor>>;
scanEntries(query: EntryQuery, limit: number, cursor?: Cursor): Promise<Page<EntryRecord, Cursor>>;
scanTasks(query: TaskQuery, limit: number, cursor?: Cursor): Promise<Page<TaskRecord<JsonValue, JsonValue, JsonValue>, Cursor>>;
createConversation(value: Omit<ConversationRecord, "id">): Promise<ConversationRecord>;
createConversation(): Promise<ConversationRecord>;
forkConversation(parentConversationId: Id, at: Id): Promise<ConversationRecord>;
appendEntry(conversationId: Id, value: EntryDraft): Promise<EntryRecord>;
createTask<I, S extends { phase: string }, R, H extends object>(
task: Task<I, S, R, H>, 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<R>(ref: TaskRef<R>, context: Context): Promise<TaskRecord<JsonValue, JsonValue, R> | undefined>;
waitForTask<R>(ref: TaskRef<R>, context: Context): Promise<SettledTask<R>>;
createConversation(
spec: Omit<ConversationSpec, "parent"> & { readonly inherit?: boolean },
spec: ConversationSpec & { readonly inherit?: boolean },
context: Context,
): Promise<OwnedConversation>;
conversation(id: Id, context: Context): Promise<OwnedConversation | undefined>;
}
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<T, C> = {
type Cursor = Readonly<Record<string, JsonValue>>;
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<DocumentContent, { kind: "base" }>;
}
| {
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<Id>;
conversation(id: Id, context: Context): Promise<ConversationRecord | undefined>;
scanConversations(limit: number, cursor: Cursor | undefined, context: Context): Promise<Page<ConversationRecord, Cursor>>;
scanConversations(query: ConversationQuery, limit: number, cursor: Cursor | undefined, context: Context): Promise<Page<ConversationRecord, Cursor>>;
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`

26
packages/durable/src/documents.ts
View file

@ -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;
}

15
packages/durable/src/errors.ts Normal file
View file

@ -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";
}
}

4
packages/durable/src/index.ts
View file

@ -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,

95
packages/durable/src/session/forks.ts Normal file
View file

@ -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<readonly ForkDocumentCopy[]> {
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<string>();
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<DocumentRecord["scope"], { readonly kind: "conversation" }>,
at: DocumentPoint,
policy: ForkPolicy,
childConversationId: Id,
copies: ForkDocumentCopy[],
copiedAddresses: Set<string>,
context: Context,
): Promise<void> {
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);
}

4
packages/durable/src/session/session.ts
View file

@ -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[];

200
packages/durable/src/session/transaction.ts
View file

@ -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<JsonValue, JsonValue, JsonValue>;
@ -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<JsonObject>;
}
| {
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<Draft<JsonObject>>;
/** 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<Id>();
readonly #forkSourceConversationIds = new Set<Id>();
readonly #forkSourceDocumentIds = new Set<Id>();
/** One entry per task touched by a public read, candidate write, or document-owner lookup. */
readonly #tasksById = new Map<Id, TransactionTask>();
@ -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<ConversationRecord, "id">): Promise<ConversationRecord> {
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<ConversationRecord> {
return this.#write(() => this.#stageConversation());
}
forkConversation(parentConversationId: Id, at: Id): Promise<ConversationRecord> {
return this.#write(() => this.#stageConversation({ conversationId: parentConversationId, at }));
}
async #stageConversation(parent?: NonNullable<ConversationRecord["parent"]>): Promise<ConversationRecord> {
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<EntryRecord> {
@ -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<DocumentTarget, { readonly kind: "fork-copy" }>,
): Promise<Draft<JsonObject>> {
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<void> {
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<StorageWrite[]> {
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();
}

5
packages/durable/src/storage/jsonl/storage.ts
View file

@ -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<Storage["entry"]>;

131
packages/durable/src/storage/memory.ts
View file

@ -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<Id, ConversationRecord>;
conversationIds: Id[];
conversationIdsByOwnerConversation: Map<Id, Id[]>;
conversationIdsByOwnerTask: Map<Id, Id[]>;
entries: Map<Id, EntryRecord>;
entryIds: Map<Id, Id[]>;
headEntryIds: Map<Id, Id[]>;
@ -137,6 +141,15 @@ const removeSorted = (ids: Id[], id: Id): void => {
if (ids[index] === id) ids.splice(index, 1);
};
const insertMapId = (index: Map<Id, Id[]>, 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<Id>();
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<Id, DocumentAction>,
@ -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<Page<ConversationRecord, Cursor>> {
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<StoredDocument | undefined> {
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<Id, TableName>();
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") {

6
packages/durable/src/storage/sqlite/database.ts
View file

@ -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;

6
packages/durable/src/storage/sqlite/migrations.ts
View file

@ -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;

4
packages/durable/src/storage/sqlite/node.ts
View file

@ -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;
}

159
packages/durable/src/storage/sqlite/storage.ts
View file

@ -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<StorageWrite, { readonly type: "document.copy" }>["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<Page<ConversationRecord, Cursor>> {
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<JsonRow>(
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<ConversationRecord>(row.record)),
@ -422,36 +436,7 @@ export class SqliteStorage implements Storage {
async document(id: Id, at: DocumentPoint, _context: Context): Promise<StoredDocument | undefined> {
this.assertOpen();
const row = getRow<JsonRow>(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id);
if (row === undefined) return undefined;
const record = parseJson<DocumentRecord>(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<RevisionRow>(
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<JsonObject>(base.content);
const tail = allRows<RevisionRow>(
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<readonly Op[]>(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<ConversationRecord>(row.record);
}
private materializeDocument(id: Id, at: DocumentPoint): StoredDocument | undefined {
const row = getRow<JsonRow>(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id);
if (row === undefined) return undefined;
const record = parseJson<DocumentRecord>(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<RevisionRow>(
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<JsonObject>(base.content);
const tail = allRows<RevisionRow>(
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<readonly Op[]>(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<RecordIdRow>(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<Id, DocumentAction> {
const actions = new Map<Id, DocumentAction>();
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<Id, DocumentAction>): void {
const liveCounts = new Map<string, number>();
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<JsonRow>(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id);
const existing = row === undefined ? undefined : parseJson<DocumentRecord>(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<Id, DocumentAction>, 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<DocumentRecord>(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) {

254
packages/durable/src/testing/storage-conformance.ts
View file

@ -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<typeof first.next>;
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",

28
packages/durable/src/types.ts
View file

@ -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<T, C> = {
/** Backend-owned JSON continuation state that callers only round-trip to the same scan. */
export type Cursor = Readonly<Record<string, JsonValue>>;
/** 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<DocumentContent, { readonly kind: "base" }>;
}
| { 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<ConversationRecord | undefined>;
entry(id: Id): Promise<EntryRecord | undefined>;
task(id: Id): Promise<TaskRecord<JsonValue, JsonValue, JsonValue> | undefined>;
scanConversations(limit: number, cursor?: Cursor): Promise<Page<ConversationRecord, Cursor>>;
scanConversations(
query: ConversationQuery,
limit: number,
cursor?: Cursor,
): Promise<Page<ConversationRecord, Cursor>>;
scanEntries(query: EntryQuery, limit: number, cursor?: Cursor): Promise<Page<EntryRecord, Cursor>>;
scanTasks(
query: TaskQuery,
@ -516,8 +533,10 @@ export interface Tx {
cursor?: Cursor,
): Promise<Page<TaskRecord<JsonValue, JsonValue, JsonValue>, Cursor>>;
/** Returned records are Session-owned immutable values and may be shared with commit listeners. */
createConversation(value: Omit<ConversationRecord, "id">): Promise<ConversationRecord>;
/** Create an independent ownerless conversation. */
createConversation(): Promise<ConversationRecord>;
/** Create an ownerless history fork at one concrete visible entry. */
forkConversation(parentConversationId: Id, at: Id): Promise<ConversationRecord>;
/** Returned records are Session-owned immutable values and may be shared with commit listeners. */
appendEntry(conversationId: Id, value: EntryDraft): Promise<EntryRecord>;
createTask<I, S extends { phase: string }, R, H extends object>(
@ -638,6 +657,7 @@ export interface Storage {
/** Scan conversations in ascending ID order. */
scanConversations(
query: ConversationQuery,
limit: number,
cursor: Cursor | undefined,
context: Context,

4
packages/durable/test/jsonl-storage.test.ts
View file

@ -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<Storage["entry"]>;
entry(conversationId: number, id: number, readContext: Context): ReturnType<Storage["entry"]>;
entry(idOrConversationId: number, idOrContext: number | Context, readContext?: Context) {

7
packages/durable/test/session-checkpoints-migrations.test.ts
View file

@ -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 });

641
packages/durable/test/session-forks.test.ts Normal file
View file

@ -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 });
});
});

25
packages/durable/test/session-support.ts
View file

@ -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<DocumentCommitChange, { readonly type: "document" }>[] {
return publication.changes.filter(
(change): change is Extract<DocumentCommitChange, { readonly type: "document" }> => change.type === "document",
);
}
export function documentCopyChanges(
publication: CommitPublication,
): readonly Extract<DocumentCommitChange, { readonly type: "document.copy" }>[] {
return publication.changes.filter(
(change): change is Extract<DocumentCommitChange, { readonly type: "document.copy" }> =>
change.type === "document.copy",
);
}
/** Create one conversation and return its ID. */
export async function createConversation(session: SessionKernel): Promise<Id> {
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. */

12
packages/durable/test/session-tables.test.ts
View file

@ -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();

15
packages/durable/test/sqlite-facade.test.ts
View file

@ -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");

4
packages/durable/test/sqlite-storage.test.ts
View file

@ -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<Storage["entry"]>;
entry(conversationId: number, id: number, readContext: Context): ReturnType<Storage["entry"]>;
entry(idOrConversationId: number, idOrContext: number | Context, readContext?: Context) {