From 16f5f0af4d85ca583d045ceb5ed6d646fcaf7761 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 21 Oct 2026 03:02:55 +0200 Subject: [PATCH] feat(durable): IndexedDbStorage, a browser storage backed by IndexedDB (@earendil-works/pi-durable/storage/indexeddb) --- package-lock.json | 11 + packages/durable/CHANGELOG.md | 1 + packages/durable/README.md | 5 + packages/durable/package.json | 7 + packages/durable/src/storage/indexeddb.ts | 1208 +++++++++++++++++ .../durable/test/indexeddb-storage.test.ts | 282 ++++ 6 files changed, 1514 insertions(+) create mode 100644 packages/durable/src/storage/indexeddb.ts create mode 100644 packages/durable/test/indexeddb-storage.test.ts diff --git a/package-lock.json b/package-lock.json index 36864aff4..51984d649 100644 --- a/package-lock.json +++ b/package-lock.json @@ -3615,6 +3615,16 @@ "integrity": "sha512-fjquC59cD7CyW6urNXK0FBufkZcoiGG80wTuPujX590cB5Ttln20E2UB4S/WARVqhXffZl2LNgS+gQdPIIim/g==", "license": "MIT" }, + "node_modules/fake-indexeddb": { + "version": "6.2.6", + "resolved": "https://registry.npmjs.org/fake-indexeddb/-/fake-indexeddb-6.2.6.tgz", + "integrity": "sha512-ihhmr0KoET51jT8R4YhbUYuLLvr1qqO6INISygtN16i8loR8kAxtAUr4+EKHTSMjL+lggaAn4OC7DRqzrNgnuQ==", + "dev": true, + "license": "Apache-2.0", + "engines": { + "node": ">=18" + } + }, "node_modules/fast-deep-equal": { "version": "3.1.3", "resolved": "https://registry.npmjs.org/fast-deep-equal/-/fast-deep-equal-3.1.3.tgz", @@ -6363,6 +6373,7 @@ }, "devDependencies": { "@cloudflare/vitest-plugin": "1.4.4", + "fake-indexeddb": "6.2.6", "shx": "0.3.4", "vitest": "4.1.11" }, diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 1bb24cccd..8d0ff1fc8 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -13,6 +13,7 @@ ### Added +- Added `IndexedDbStorage` (`@earendil-works/pi-durable/storage/indexeddb`), a browser storage backed by IndexedDB. - `HarnessOptions.telemetry` emits the content-free `DURABLE_TELEMETRY_SCHEMA`: public submission admission spans, task phase spans (including extension tasks), committed entry and submission events, running phase-change checkpoints, and nested-tool replay events. Optional propagation stores opaque carriers on submissions and tasks so queued inputs and reopened tasks retain their trace; `TaskOptions.trace` accepts a carrier override or `"root"` (absence inherits the creator's carrier); creation-time `links` connect inputs sharing a run, while steer admissions link to the running trace. Internal admissions emit events without a submit span. Optional metrics record phase duration and live task counts. Failed phases record fixed error-status categories (`task_faulted`, `task_failed`, `task_orphaned`) and diagnostics capped at 500 characters; aborted phases keep status `ok`. Every `onReport` report also emits a fixed-field `pi.durable.report` event on its current span (dropped by the NOOP span when there is none), plus a `pi.durable.reports` counter by type; report telemetry contains no error payload and leaves the callback unchanged. Clock failures report without a phase context. Generation and compaction forward their phase context to pi-ai requests. - `settings.compaction.modelOverrides` sets compaction thresholds per model, keyed by exact `"provider/modelId"`, so one harness can serve small- and large-window models; the threshold check, overflow recovery, and the compaction task use the entry of the model they act for ([#10942](https://github.com/earendil-works/pi/issues/10942)). - Durable sleeps: a sleeping task holds no invocation, survives restarts, and resumes at its checkpoint once the Harness clock reaches `until`. diff --git a/packages/durable/README.md b/packages/durable/README.md index f43e67564..b6c28f997 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -759,10 +759,15 @@ const usage = await harness.usage(context); // { models: { "openai/gpt-6-sol": U | Backend | Import | Notes | |---|---|---| | Memory | `MemoryStorage` from the package root | Nothing is persisted. | +| IndexedDB | `IndexedDbStorage.open(name)` from `@earendil-works/pi-durable/storage/indexeddb` | Browser-only storage in one IndexedDB database. | | SQLite | `openNodeSqliteStorage(file)` from `@earendil-works/pi-durable/storage/sqlite/node` | One database file. WAL mode with `synchronous = NORMAL`: commits survive process crashes; the newest may be lost on power or host failure. | | JSONL | `openNodeJsonlStorage(directory, context)` from `@earendil-works/pi-durable/storage/jsonl/node` | Append-only files in one directory. Pass `{ fsync: true }` to flush before each commit marker. Keeps the whole history in memory too, so memory grows with it; use SQLite for long-running agents. | | Cloudflare Durable Object | `openDurableObjectSqliteStorage(ctx.storage)` from `@earendil-works/pi-durable/storage/sqlite/cloudflare` | The SQLite storage of one SQLite-backed Durable Object; commits are its storage transactions. | +In a browser, open IndexedDB storage with `await IndexedDbStorage.open(name)`. One owner must hold the database at a +time; use a [Web Lock](https://developer.mozilla.org/docs/Web/API/Web_Locks_API) around its lifetime rather than opening +it in competing tabs or workers. Call `await navigator.storage.persist()` when durable retention matters. + SQLite storage can namespace its tables and indexes when sharing a database: pass `{ prefix: "pi_" }` to `SqliteStorage.open(database, options)`, `openNodeSqliteStorage(path, options)`, or `openDurableObjectSqliteStorage(ctx.storage, options)`. Storage renders the prefixed names into its SQL, so adapters diff --git a/packages/durable/package.json b/packages/durable/package.json index 5823f4240..c15e96ad4 100644 --- a/packages/durable/package.json +++ b/packages/durable/package.json @@ -66,6 +66,12 @@ "import": "./dist/storage/memory.js", "default": "./dist/storage/memory.js" }, + "./storage/indexeddb": { + "source": "./src/storage/indexeddb.ts", + "types": "./dist/storage/indexeddb.d.ts", + "import": "./dist/storage/indexeddb.js", + "default": "./dist/storage/indexeddb.js" + }, "./storage/jsonl": { "source": "./src/storage/jsonl/index.ts", "types": "./dist/storage/jsonl/index.d.ts", @@ -154,6 +160,7 @@ }, "devDependencies": { "@cloudflare/vitest-plugin": "1.4.4", + "fake-indexeddb": "6.2.6", "shx": "0.3.4", "vitest": "4.1.11" } diff --git a/packages/durable/src/storage/indexeddb.ts b/packages/durable/src/storage/indexeddb.ts new file mode 100644 index 000000000..6b2121444 --- /dev/null +++ b/packages/durable/src/storage/indexeddb.ts @@ -0,0 +1,1208 @@ +/// + +import type { Context, JsonValue } from "@earendil-works/chord"; +import { apply, type Op } from "@earendil-works/chord/delta"; +import { StorageRequestError } from "../errors.ts"; +import { idFromNumber, seqFromNumber } from "../ids.ts"; +import type { + ConversationId, + ConversationQuery, + ConversationRecord, + Cursor, + DocumentAddress, + DocumentContent, + DocumentCreate, + DocumentId, + DocumentPoint, + DocumentQuery, + DocumentRecord, + EntryId, + EntryQuery, + EntryRecord, + Id, + JsonObject, + Page, + ScanOrder, + Seq, + Storage, + StorageWrite, + StoredDocument, + SubmissionId, + SubmissionQuery, + SubmissionRecord, + TaskId, + TaskQuery, + TaskRecord, +} from "../types.ts"; +import { nextCursor, type ScanStart, scanStart } from "./scan.ts"; + +const DATABASE_VERSION = 1; +const MAX_KEY = Number.MAX_SAFE_INTEGER; +const META_KEY = "state"; + +const STORES = { + meta: "meta", + recordIds: "record_ids", + conversations: "conversations", + entries: "entries", + tasks: "tasks", + submissions: "submissions", + documents: "documents", + documentRevisions: "document_revisions", +} as const; + +const ALL_STORES = Object.values(STORES); + +const INDEXES = { + conversationsByOwnerConversation: "conversations_by_owner_conversation", + conversationsByOwnerTask: "conversations_by_owner_task", + entriesByConversation: "entries_by_conversation", + entryHeadsByConversation: "entry_heads_by_conversation", + tasksByStatus: "tasks_by_status", + tasksByConversation: "tasks_by_conversation", + tasksByKind: "tasks_by_kind", + tasksByAbortRequested: "tasks_by_abort_requested", + tasksByBackground: "tasks_by_background", + submissionsByRequest: "submissions_by_request", + submissionsByConversation: "submissions_by_conversation", + submissionsByStatus: "submissions_by_status", + documentsByAddress: "documents_by_address", + documentsByScope: "documents_by_scope", + documentsByScopeKind: "documents_by_scope_kind", + documentRevisionsByKind: "document_revisions_by_kind", +} as const; + +type StoredTask = TaskRecord; +type TableName = "conversation" | "entry" | "task" | "submission" | "document"; +type DocumentAction = { + create?: DocumentCreate; + copy?: Extract["source"]; + content?: DocumentContent; + retire: boolean; +}; +type ScopeColumns = { + readonly scopeKind: DocumentRecord["scope"]["kind"]; + readonly ownerId: number; +}; +type MetaRow = { readonly key: typeof META_KEY; readonly nextId: number; readonly nextSeq: number }; +type RecordIdRow = { readonly id: number; readonly recordType: TableName }; +type ConversationRow = { + readonly id: ConversationId; + readonly ownerConversationId?: ConversationId; + readonly ownerTaskId?: TaskId; + readonly record: ConversationRecord; +}; +type EntryRow = { + readonly id: EntryId; + readonly conversationId: ConversationId; + readonly headConversationId?: ConversationId; + readonly commitSeq: Seq; + readonly record: EntryRecord; +}; +type TaskRow = { + readonly id: TaskId; + readonly conversationId: ConversationId; + readonly kind: string; + readonly status: StoredTask["state"]["status"]; + readonly abortRequested: 0 | 1; + readonly background: 0 | 1; + readonly record: StoredTask; +}; +type SubmissionRow = { + readonly id: SubmissionId; + readonly conversationId: ConversationId; + readonly requestId?: string; + readonly status: SubmissionRecord["status"]; + readonly record: SubmissionRecord; +}; +type DocumentRow = { + readonly id: DocumentId; + readonly kind: string; + readonly family: 0 | 1; + readonly keyValue: string; + readonly scopeKind: DocumentRecord["scope"]["kind"]; + readonly ownerId: number; + readonly createdAt: Seq; + readonly retiredAt?: Seq; + readonly record: DocumentRecord; +}; +type RevisionRow = { + readonly documentId: DocumentId; + readonly seq: Seq; + readonly kind: DocumentContent["kind"]; + readonly version: number; + readonly content: JsonObject | readonly Op[]; +}; + +type IndexedDbStorageOptions = { + readonly indexedDB?: IDBFactory; +}; + +const requestResult = (request: IDBRequest): Promise => + new Promise((resolve, reject) => { + request.addEventListener("success", () => resolve(request.result), { once: true }); + request.addEventListener("error", () => reject(request.error ?? new Error("IndexedDB request failed")), { + once: true, + }); + }); + +const transactionResult = (transaction: IDBTransaction): Promise => + new Promise((resolve, reject) => { + transaction.addEventListener("complete", () => resolve(), { once: true }); + transaction.addEventListener( + "abort", + () => reject(transaction.error ?? new Error("IndexedDB transaction was aborted")), + { once: true }, + ); + transaction.addEventListener( + "error", + () => reject(transaction.error ?? new Error("IndexedDB transaction failed")), + { once: true }, + ); + }); + +const keyRanges = (): typeof IDBKeyRange => { + if (typeof IDBKeyRange === "undefined") throw new Error("IndexedDB key ranges are not available"); + return IDBKeyRange; +}; + +const cursorValues = ( + source: IDBObjectStore | IDBIndex, + range: IDBKeyRange | IDBValidKey | null, + direction: IDBCursorDirection, + limit: number, + accept: (value: T) => boolean = () => true, +): Promise => + new Promise((resolve, reject) => { + const values: T[] = []; + const request = source.openCursor(range, direction); + request.addEventListener("error", () => reject(request.error ?? new Error("IndexedDB cursor failed")), { + once: true, + }); + request.addEventListener("success", () => { + const cursor = request.result; + if (cursor === null) { + resolve(values); + return; + } + const value = cursor.value as T; + if (accept(value)) values.push(value); + if (values.length >= limit) { + resolve(values); + return; + } + cursor.continue(); + }); + }); + +const firstCursorValue = ( + source: IDBObjectStore | IDBIndex, + range: IDBKeyRange | IDBValidKey | null, + direction: IDBCursorDirection, + accept: (value: T) => boolean = () => true, +): Promise => + new Promise((resolve, reject) => { + const request = source.openCursor(range, direction); + request.addEventListener("error", () => reject(request.error ?? new Error("IndexedDB cursor failed")), { + once: true, + }); + request.addEventListener("success", () => { + const cursor = request.result; + if (cursor === null) { + resolve(undefined); + return; + } + const value = cursor.value as T; + if (accept(value)) { + resolve(value); + return; + } + cursor.continue(); + }); + }); + +const page = }>( + values: readonly T[], + limit: number, + order: ScanOrder = "ascending", +): Page => { + const items = values.slice(0, limit); + if (values.length <= limit) return { items }; + return { items, next: nextCursor(items.at(-1)!.id, order) }; +}; + +const cursorId = >(cursor: Cursor | undefined): I | undefined => { + const after = cursor?.after; + if (after === undefined) return undefined; + if (typeof after !== "number" || !Number.isSafeInteger(after)) { + throw new StorageRequestError("Invalid storage cursor"); + } + return idFromNumber(after); +}; + +const primaryRange = (start: ScanStart): IDBKeyRange | null => { + if (start.after === undefined) return null; + return start.order === "ascending" + ? keyRanges().lowerBound(start.after, true) + : keyRanges().upperBound(start.after, true); +}; + +const compoundIdRange = ( + prefix: readonly IDBValidKey[], + minimum: number, + maximum: number, + minimumOpen = false, + maximumOpen = false, +): IDBKeyRange => keyRanges().bound([...prefix, minimum], [...prefix, maximum], minimumOpen, maximumOpen); + +const scanCompoundRows = ( + index: IDBIndex, + prefix: readonly IDBValidKey[], + start: ScanStart, + limit: number, + accept: (value: T) => boolean, +): Promise => { + const minimum = start.order === "ascending" && start.after !== undefined ? start.after : 0; + const maximum = start.order === "descending" && start.after !== undefined ? start.after : MAX_KEY; + return cursorValues( + index, + compoundIdRange( + prefix, + minimum, + maximum, + start.order === "ascending" && start.after !== undefined, + start.order === "descending" && start.after !== undefined, + ), + start.order === "ascending" ? "next" : "prev", + limit, + accept, + ); +}; + +const scopeColumns = (scope: DocumentRecord["scope"]): ScopeColumns => { + switch (scope.kind) { + case "session": + return { scopeKind: "session", ownerId: 0 }; + case "conversation": + return { scopeKind: "conversation", ownerId: scope.conversationId }; + case "task": + return { scopeKind: "task", ownerId: scope.taskId }; + } +}; + +const addressParts = (address: DocumentAddress | DocumentCreate | DocumentRecord) => { + const scope = scopeColumns(address.scope); + return { + kind: address.kind, + ...scope, + family: address.key === undefined ? (0 as const) : (1 as const), + keyValue: address.key ?? "", + }; +}; + +const addressPrefix = (address: DocumentAddress | DocumentCreate | DocumentRecord): readonly IDBValidKey[] => { + const parts = addressParts(address); + return [parts.kind, parts.scopeKind, parts.ownerId, parts.family, parts.keyValue]; +}; + +const addressKey = (address: DocumentAddress | DocumentCreate | DocumentRecord): string => + JSON.stringify(addressPrefix(address)); + +const isAliveAt = (record: DocumentRecord, at: DocumentPoint): boolean => { + if (at === "current") return record.retiredAt === undefined; + return record.createdAt <= at && (record.retiredAt === undefined || at < record.retiredAt); +}; + +const isCurrentOnly = (record: DocumentRecord): boolean => + record.scope.kind !== "conversation" || record.history === "latest"; + +const writeId = (write: StorageWrite): Id | undefined => { + switch (write.type) { + case "conversation": + case "entry": + case "task": + case "submission": + return write.value.id; + case "document.create": + case "document.copy": + return write.record.id; + case "document.change": + case "document.retire": + return undefined; + } +}; + +function createSchema(database: IDBDatabase): void { + const meta = database.createObjectStore(STORES.meta, { keyPath: "key" }); + meta.add({ key: META_KEY, nextId: 2, nextSeq: 1 } satisfies MetaRow); + database.createObjectStore(STORES.recordIds, { keyPath: "id" }); + + const conversations = database.createObjectStore(STORES.conversations, { keyPath: "id" }); + conversations.createIndex(INDEXES.conversationsByOwnerConversation, ["ownerConversationId", "id"]); + conversations.createIndex(INDEXES.conversationsByOwnerTask, ["ownerTaskId", "id"]); + + const entries = database.createObjectStore(STORES.entries, { keyPath: "id" }); + entries.createIndex(INDEXES.entriesByConversation, ["conversationId", "id"]); + entries.createIndex(INDEXES.entryHeadsByConversation, ["headConversationId", "id"]); + + const tasks = database.createObjectStore(STORES.tasks, { keyPath: "id" }); + tasks.createIndex(INDEXES.tasksByStatus, ["status", "id"]); + tasks.createIndex(INDEXES.tasksByConversation, ["conversationId", "id"]); + tasks.createIndex(INDEXES.tasksByKind, ["kind", "id"]); + tasks.createIndex(INDEXES.tasksByAbortRequested, ["abortRequested", "id"]); + tasks.createIndex(INDEXES.tasksByBackground, ["background", "id"]); + + const submissions = database.createObjectStore(STORES.submissions, { keyPath: "id" }); + submissions.createIndex(INDEXES.submissionsByRequest, ["conversationId", "requestId"]); + submissions.createIndex(INDEXES.submissionsByConversation, ["conversationId", "id"]); + submissions.createIndex(INDEXES.submissionsByStatus, ["status", "id"]); + + const documents = database.createObjectStore(STORES.documents, { keyPath: "id" }); + documents.createIndex(INDEXES.documentsByAddress, [ + "kind", + "scopeKind", + "ownerId", + "family", + "keyValue", + "createdAt", + "id", + ]); + documents.createIndex(INDEXES.documentsByScope, ["scopeKind", "ownerId", "id"]); + documents.createIndex(INDEXES.documentsByScopeKind, ["scopeKind", "ownerId", "kind", "id"]); + + const revisions = database.createObjectStore(STORES.documentRevisions, { keyPath: ["documentId", "seq"] }); + revisions.createIndex(INDEXES.documentRevisionsByKind, ["documentId", "kind", "seq"]); +} + +/** Browser IndexedDB implementation of the durable storage contract. One owner must serialize access to each database. */ +export class IndexedDbStorage implements Storage { + private readonly database: IDBDatabase; + private closed = false; + private closing: Promise | undefined; + private activeOperations = 0; + private operationsDrained: (() => void) | undefined; + + private constructor(database: IDBDatabase) { + this.database = database; + } + + /** Open or create a browser-only IndexedDB database. */ + static async open(name: string, options: IndexedDbStorageOptions = {}): Promise { + const factory = options.indexedDB ?? globalThis.indexedDB; + if (factory === undefined) throw new Error("IndexedDB is not available"); + const request = factory.open(name, DATABASE_VERSION); + request.addEventListener("upgradeneeded", (event) => { + if ((event as IDBVersionChangeEvent).oldVersion === 0) createSchema(request.result); + }); + const database = await requestResult(request); + try { + const transaction = database.transaction(STORES.meta, "readonly"); + const done = transactionResult(transaction); + const metadata = await requestResult( + transaction.objectStore(STORES.meta).get(META_KEY) as IDBRequest, + ); + await done; + if (metadata === undefined) throw new Error("Durable IndexedDB metadata is missing"); + return new IndexedDbStorage(database); + } catch (error) { + database.close(); + throw error; + } + } + + /** Delete a closed database. Open connections in this or another owner can block deletion. */ + static async delete(name: string, options: IndexedDbStorageOptions = {}): Promise { + const factory = options.indexedDB ?? globalThis.indexedDB; + if (factory === undefined) throw new Error("IndexedDB is not available"); + await requestResult(factory.deleteDatabase(name)); + } + + async commit(writes: readonly StorageWrite[], _context: Context): Promise { + this.assertOpen(); + const documentActions = this.prepareDocumentActions(writes); + return this.transaction(ALL_STORES, "readwrite", async (transaction) => { + const metaStore = transaction.objectStore(STORES.meta); + const metadata = await requestResult(metaStore.get(META_KEY) as IDBRequest); + if (metadata === undefined) throw new Error("Durable IndexedDB metadata is missing"); + const seq = seqFromNumber(metadata.nextSeq); + await this.checkGlobalIds(transaction, writes); + await this.checkDocumentActions(transaction, documentActions); + for (const write of writes) await this.applyTableWrite(transaction, write, seq); + await this.applyDocumentActions(transaction, documentActions, seq); + let nextId = metadata.nextId; + for (const write of writes) { + const id = writeId(write); + if (id !== undefined) nextId = Math.max(nextId, id + 1); + } + await requestResult(metaStore.put({ key: META_KEY, nextId, nextSeq: seq + 1 } satisfies MetaRow)); + return seq; + }); + } + + async mintId>(): Promise { + this.assertOpen(); + return this.transaction(STORES.meta, "readwrite", async (transaction) => { + const store = transaction.objectStore(STORES.meta); + const metadata = await requestResult(store.get(META_KEY) as IDBRequest); + if (metadata === undefined) throw new Error("Durable IndexedDB metadata is missing"); + if (!Number.isSafeInteger(metadata.nextId)) throw new Error("ID space is exhausted"); + await requestResult(store.put({ ...metadata, nextId: metadata.nextId + 1 })); + return idFromNumber(metadata.nextId); + }); + } + + conversation(id: ConversationId, _context: Context): Promise { + return this.transaction( + STORES.conversations, + "readonly", + async (transaction) => + ( + await requestResult( + transaction.objectStore(STORES.conversations).get(id) as IDBRequest, + ) + )?.record, + ); + } + + async scanConversations( + query: ConversationQuery, + limit: number, + cursor: Cursor | undefined, + _context: Context, + ): Promise> { + const start = scanStart(query.order, cursor, "ascending"); + return this.transaction(STORES.conversations, "readonly", async (transaction) => { + const store = transaction.objectStore(STORES.conversations); + const accept = (row: ConversationRow) => + query.ownerConversationId === undefined || row.ownerConversationId === query.ownerConversationId; + let rows: ConversationRow[]; + if (query.ownerTaskId !== undefined) { + rows = await scanCompoundRows( + store.index(INDEXES.conversationsByOwnerTask), + [query.ownerTaskId], + start, + limit + 1, + accept, + ); + } else if (query.ownerConversationId !== undefined) { + rows = await scanCompoundRows( + store.index(INDEXES.conversationsByOwnerConversation), + [query.ownerConversationId], + start, + limit + 1, + accept, + ); + } else { + rows = await cursorValues( + store, + primaryRange(start), + start.order === "ascending" ? "next" : "prev", + limit + 1, + accept, + ); + } + return page( + rows.map((row) => row.record), + limit, + start.order, + ); + }); + } + + entry(id: EntryId, context: Context): Promise<{ readonly entry: EntryRecord; readonly commitSeq: Seq } | undefined>; + entry( + conversationId: ConversationId, + id: EntryId, + context: Context, + ): Promise<{ readonly entry: EntryRecord; readonly commitSeq: Seq } | undefined>; + entry( + idOrConversationId: EntryId | ConversationId, + idOrContext: EntryId | Context, + context?: Context, + ): Promise<{ readonly entry: EntryRecord; readonly commitSeq: Seq } | undefined> { + return this.transaction([STORES.conversations, STORES.entries], "readonly", async (transaction) => { + const id = + context === undefined + ? idFromNumber(idOrConversationId) + : typeof idOrContext === "number" + ? idFromNumber(idOrContext) + : undefined; + if (id === undefined) throw new StorageRequestError("Storage.entry() requires an entry ID"); + let conversation: ConversationRecord | undefined; + if (context !== undefined) { + const conversationId = idFromNumber(idOrConversationId); + conversation = await this.readConversation(transaction, conversationId); + if (conversation === undefined) throw new StorageRequestError(`Unknown conversation: ${conversationId}`); + } + const row = await requestResult( + transaction.objectStore(STORES.entries).get(id) as IDBRequest, + ); + if (row === undefined) return undefined; + if (conversation !== undefined) { + let upperEntryId = MAX_KEY; + while (conversation.id !== row.conversationId) { + if (conversation.parent === undefined) return undefined; + upperEntryId = Math.min(upperEntryId, conversation.parent.at); + conversation = (await this.readConversation(transaction, conversation.parent.conversationId))!; + } + if (row.id > upperEntryId) return undefined; + } + return { entry: row.record, commitSeq: row.commitSeq }; + }); + } + + findLatestHeadMarker( + conversationId: ConversationId, + atOrBeforeEntryId: EntryId | undefined, + _context: Context, + ): Promise<(EntryRecord & { readonly head: EntryId }) | undefined> { + return this.transaction([STORES.conversations, STORES.entries], "readonly", async (transaction) => { + let conversation = await this.readConversation(transaction, conversationId); + if (conversation === undefined) throw new StorageRequestError(`Unknown conversation: ${conversationId}`); + let upper = atOrBeforeEntryId ?? MAX_KEY; + const index = transaction.objectStore(STORES.entries).index(INDEXES.entryHeadsByConversation); + while (true) { + const row = await firstCursorValue(index, compoundIdRange([conversation.id], 0, upper), "prev"); + if (row !== undefined) return row.record as EntryRecord & { readonly head: EntryId }; + if (conversation.parent === undefined) return undefined; + upper = Math.min(upper, conversation.parent.at); + conversation = (await this.readConversation(transaction, conversation.parent.conversationId))!; + } + }); + } + + async scanEntries( + query: EntryQuery, + limit: number, + cursor: Cursor | undefined, + _context: Context, + ): Promise> { + const start = scanStart(query.order, cursor, "descending"); + return this.transaction([STORES.conversations, STORES.entries], "readonly", async (transaction) => { + const values = + start.order === "ascending" + ? await this.readEntriesAscending(transaction, query, limit, start.after) + : await this.readEntriesDescending(transaction, query, limit, start.after); + return page(values, limit, start.order); + }); + } + + task(id: TaskId, _context: Context): Promise { + return this.transaction( + STORES.tasks, + "readonly", + async (transaction) => + (await requestResult(transaction.objectStore(STORES.tasks).get(id) as IDBRequest)) + ?.record, + ); + } + + async scanTasks( + query: TaskQuery, + limit: number, + cursor: Cursor | undefined, + _context: Context, + ): Promise> { + const start = scanStart(query.order, cursor, "ascending"); + return this.transaction(STORES.tasks, "readonly", async (transaction) => { + const store = transaction.objectStore(STORES.tasks); + const accept = (row: TaskRow) => + (query.conversationId === undefined || row.conversationId === query.conversationId) && + (query.kind === undefined || row.kind === query.kind) && + (query.status === undefined || row.status === query.status) && + (query.abortRequested === undefined || row.abortRequested === (query.abortRequested ? 1 : 0)) && + (query.background === undefined || row.background === (query.background ? 1 : 0)); + let source: IDBObjectStore | IDBIndex = store; + let prefix: readonly IDBValidKey[] | undefined; + if (query.status !== undefined) { + source = store.index(INDEXES.tasksByStatus); + prefix = [query.status]; + } else if (query.conversationId !== undefined) { + source = store.index(INDEXES.tasksByConversation); + prefix = [query.conversationId]; + } else if (query.kind !== undefined) { + source = store.index(INDEXES.tasksByKind); + prefix = [query.kind]; + } else if (query.abortRequested !== undefined) { + source = store.index(INDEXES.tasksByAbortRequested); + prefix = [query.abortRequested ? 1 : 0]; + } else if (query.background !== undefined) { + source = store.index(INDEXES.tasksByBackground); + prefix = [query.background ? 1 : 0]; + } + const rows = + prefix === undefined + ? await cursorValues( + source, + primaryRange(start), + start.order === "ascending" ? "next" : "prev", + limit + 1, + accept, + ) + : await scanCompoundRows(source as IDBIndex, prefix, start, limit + 1, accept); + return page( + rows.map((row) => row.record), + limit, + start.order, + ); + }); + } + + submission(id: SubmissionId, _context: Context): Promise { + return this.transaction( + STORES.submissions, + "readonly", + async (transaction) => + ( + await requestResult( + transaction.objectStore(STORES.submissions).get(id) as IDBRequest, + ) + )?.record, + ); + } + + async scanSubmissions( + query: SubmissionQuery, + limit: number, + cursor: Cursor | undefined, + _context: Context, + ): Promise> { + const start = scanStart(query.order, cursor, "ascending"); + return this.transaction(STORES.submissions, "readonly", async (transaction) => { + const store = transaction.objectStore(STORES.submissions); + const accept = (row: SubmissionRow) => + (query.conversationId === undefined || row.conversationId === query.conversationId) && + (query.status === undefined || row.status === query.status); + let rows: SubmissionRow[]; + if (query.status !== undefined) { + rows = await scanCompoundRows( + store.index(INDEXES.submissionsByStatus), + [query.status], + start, + limit + 1, + accept, + ); + } else if (query.conversationId !== undefined) { + rows = await scanCompoundRows( + store.index(INDEXES.submissionsByConversation), + [query.conversationId], + start, + limit + 1, + accept, + ); + } else { + rows = await cursorValues( + store, + primaryRange(start), + start.order === "ascending" ? "next" : "prev", + limit + 1, + accept, + ); + } + return page( + rows.map((row) => row.record), + limit, + start.order, + ); + }); + } + + submissionByRequest( + conversationId: ConversationId, + requestId: string, + _context: Context, + ): Promise { + return this.transaction(STORES.submissions, "readonly", async (transaction) => { + const row = await requestResult( + transaction + .objectStore(STORES.submissions) + .index(INDEXES.submissionsByRequest) + .get([conversationId, requestId]) as IDBRequest, + ); + return row?.record; + }); + } + + findDocument(address: DocumentAddress, at: DocumentPoint, _context: Context): Promise { + return this.transaction(STORES.documents, "readonly", async (transaction) => { + const row = await firstCursorValue( + transaction.objectStore(STORES.documents).index(INDEXES.documentsByAddress), + keyRanges().bound([...addressPrefix(address), 0, 0], [...addressPrefix(address), MAX_KEY, MAX_KEY]), + "prev", + (value) => isAliveAt(value.record, at), + ); + return row?.record; + }); + } + + document(id: DocumentId, at: DocumentPoint, _context: Context): Promise { + return this.transaction([STORES.documents, STORES.documentRevisions], "readonly", (transaction) => + this.materializeDocument(transaction, id, at), + ); + } + + async scanDocuments( + query: DocumentQuery, + limit: number, + cursor: Cursor | undefined, + _context: Context, + ): Promise> { + const after = cursorId(cursor); + return this.transaction(STORES.documents, "readonly", async (transaction) => { + const scope = scopeColumns(query.scope); + const store = transaction.objectStore(STORES.documents); + const index = + query.kind === undefined + ? store.index(INDEXES.documentsByScope) + : store.index(INDEXES.documentsByScopeKind); + const prefix: readonly IDBValidKey[] = + query.kind === undefined ? [scope.scopeKind, scope.ownerId] : [scope.scopeKind, scope.ownerId, query.kind]; + const rows = await cursorValues( + index, + compoundIdRange(prefix, after ?? 0, MAX_KEY, after !== undefined), + "next", + limit + 1, + (row) => isAliveAt(row.record, query.at), + ); + return page( + rows.map((row) => row.record), + limit, + ); + }); + } + + close(_context: Context): Promise { + if (this.closing === undefined) { + this.closed = true; + this.closing = this.closeDatabase(); + } + return this.closing; + } + + private async closeDatabase(): Promise { + if (this.activeOperations > 0) { + await new Promise((resolve) => { + this.operationsDrained = resolve; + }); + } + this.database.close(); + } + + private async transaction( + stores: string | readonly string[], + mode: IDBTransactionMode, + operation: (transaction: IDBTransaction) => Promise, + ): Promise { + this.assertOpen(); + this.activeOperations++; + try { + const transaction = this.database.transaction(stores, mode); + const done = transactionResult(transaction); + try { + const value = await operation(transaction); + await done; + return value; + } catch (error) { + try { + transaction.abort(); + } catch { + // It may already have aborted or committed. + } + try { + await done; + } catch { + // Preserve the operation failure. + } + throw error; + } + } finally { + if (--this.activeOperations === 0) this.operationsDrained?.(); + } + } + + private async readConversation( + transaction: IDBTransaction, + id: ConversationId, + ): Promise { + return ( + await requestResult( + transaction.objectStore(STORES.conversations).get(id) as IDBRequest, + ) + )?.record; + } + + private async readEntriesDescending( + transaction: IDBTransaction, + query: EntryQuery, + limit: number, + after: number | undefined, + ): Promise { + let conversation = await this.readConversation(transaction, query.conversationId); + if (conversation === undefined) throw new StorageRequestError(`Unknown conversation: ${query.conversationId}`); + let upper = query.maxEntryId ?? MAX_KEY; + if (after !== undefined) upper = Math.min(upper, after - 1); + const minimum = query.minEntryId ?? 0; + const values: EntryRecord[] = []; + const index = transaction.objectStore(STORES.entries).index(INDEXES.entriesByConversation); + while (true) { + if (upper >= minimum) { + const rows = await cursorValues( + index, + compoundIdRange([conversation.id], minimum, upper), + "prev", + limit + 1 - values.length, + ); + values.push(...rows.map((row) => row.record)); + } + if (values.length > limit || conversation.parent === undefined) break; + upper = Math.min(upper, conversation.parent.at); + if (upper < minimum) break; + conversation = (await this.readConversation(transaction, conversation.parent.conversationId))!; + } + return values; + } + + private async readEntriesAscending( + transaction: IDBTransaction, + query: EntryQuery, + limit: number, + after: number | undefined, + ): Promise { + const segments: { readonly conversationId: ConversationId; readonly upper: number }[] = []; + let conversation = await this.readConversation(transaction, query.conversationId); + if (conversation === undefined) throw new StorageRequestError(`Unknown conversation: ${query.conversationId}`); + let upper = query.maxEntryId ?? MAX_KEY; + const requestedMinimum = query.minEntryId ?? 0; + while (true) { + segments.push({ conversationId: conversation.id, upper }); + if (conversation.parent === undefined) break; + upper = Math.min(upper, conversation.parent.at); + if (upper < requestedMinimum) break; + conversation = (await this.readConversation(transaction, conversation.parent.conversationId))!; + } + const minimum = after === undefined ? requestedMinimum : Math.max(requestedMinimum, after + 1); + const values: EntryRecord[] = []; + const index = transaction.objectStore(STORES.entries).index(INDEXES.entriesByConversation); + for (const segment of segments.reverse()) { + if (segment.upper < minimum) continue; + const rows = await cursorValues( + index, + compoundIdRange([segment.conversationId], minimum, segment.upper), + "next", + limit + 1 - values.length, + ); + values.push(...rows.map((row) => row.record)); + if (values.length > limit) break; + } + return values; + } + + private async materializeDocument( + transaction: IDBTransaction, + id: DocumentId, + at: DocumentPoint, + ): Promise { + const record = ( + await requestResult(transaction.objectStore(STORES.documents).get(id) as IDBRequest) + )?.record; + if (record === undefined) return undefined; + if (at !== "current" && isCurrentOnly(record)) { + throw new StorageRequestError(`Document ${id} does not retain historical content`); + } + if (!isAliveAt(record, at)) return undefined; + const upper = at === "current" ? MAX_KEY : at; + const revisions = transaction.objectStore(STORES.documentRevisions); + const base = await firstCursorValue( + revisions.index(INDEXES.documentRevisionsByKind), + keyRanges().bound([id, "base", 0], [id, "base", upper]), + "prev", + ); + if (base === undefined || base.kind !== "base") throw new Error(`Document ${id} is missing a required base`); + let value = base.content as JsonObject; + const tail = + base.seq >= upper + ? [] + : await cursorValues( + revisions, + keyRanges().bound([id, base.seq], [id, upper], true), + "next", + MAX_KEY, + ); + 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, revision.content as readonly Op[]) as JsonObject; + } + return { record, version: base.version, value, deltasSinceBase: tail.length }; + } + + private async checkGlobalIds(transaction: IDBTransaction, writes: readonly StorageWrite[]): Promise { + const store = transaction.objectStore(STORES.recordIds); + const claimed = new Map, TableName>(); + for (const write of writes) { + if (write.type === "document.change" || write.type === "document.retire") continue; + 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 = (await requestResult(store.get(id) as IDBRequest))?.recordType; + const earlier = claimed.get(id); + if (table === "conversation" || table === "entry" || table === "document") { + if (existing !== undefined) throw new Error(`ID ${id} already belongs to ${existing}`); + if (earlier !== undefined) throw new Error(`ID ${id} is written more than once`); + } else { + if (existing !== undefined && existing !== table) + throw new Error(`ID ${id} already belongs to ${existing}`); + if (earlier !== undefined && earlier !== table) throw new Error(`ID ${id} is written as two record types`); + } + claimed.set(id, table); + } + } + + private prepareDocumentActions(writes: readonly StorageWrite[]): Map { + const actions = new Map(); + for (const write of writes) { + 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.type === "document.copy" ? write.record.id : write.id; + let action = actions.get(id); + if (action === undefined) { + action = { retire: false }; + actions.set(id, action); + } + switch (write.type) { + case "document.create": + 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 || action.copy !== undefined) { + throw new Error(`Document ${id} has more than one content command`); + } + action.content = write.content; + break; + case "document.retire": + if (action.retire) throw new Error(`Document ${id} is retired more than once`); + action.retire = true; + break; + } + } + return actions; + } + + private async checkDocumentActions( + transaction: IDBTransaction, + actions: ReadonlyMap, + ): Promise { + const documents = transaction.objectStore(STORES.documents); + const revisions = transaction.objectStore(STORES.documentRevisions); + const liveCounts = new Map(); + for (const [id, action] of actions) { + if (action.copy !== undefined && actions.has(action.copy.id)) { + throw new Error(`Document copy ${id} source is changed in the copy batch`); + } + const existing = (await requestResult(documents.get(id) as IDBRequest))?.record; + if (action.create === undefined && existing === undefined) throw new Error(`Unknown document: ${id}`); + if (action.create !== undefined && existing !== undefined) throw new Error(`Document ${id} already exists`); + if (existing?.retiredAt !== undefined) throw new Error(`Document ${id} is retired`); + if (action.content?.kind === "delta") { + const previous = await firstCursorValue(revisions, compoundIdRange([id], 0, MAX_KEY), "prev"); + if (previous === undefined) throw new Error(`Document ${id} delta has no base`); + if (previous.version !== action.content.version) { + throw new Error(`Document ${id} version transition requires a base`); + } + } + const record = action.create ?? existing!; + const key = addressKey(record); + let live = liveCounts.get(key); + if (live === undefined) live = (await this.currentDocumentId(transaction, record)) === undefined ? 0 : 1; + if (action.retire && existing !== undefined) live--; + if (action.create !== undefined && !action.retire) live++; + liveCounts.set(key, live); + } + for (const live of liveCounts.values()) { + if (live > 1) throw new Error("Document address already has a current incarnation"); + } + } + + private async currentDocumentId( + transaction: IDBTransaction, + address: DocumentAddress | DocumentCreate | DocumentRecord, + ): Promise { + const row = await firstCursorValue( + transaction.objectStore(STORES.documents).index(INDEXES.documentsByAddress), + keyRanges().bound([...addressPrefix(address), 0, 0], [...addressPrefix(address), MAX_KEY, MAX_KEY]), + "prev", + (value) => value.retiredAt === undefined, + ); + return row?.id; + } + + private async claimId(transaction: IDBTransaction, id: Id, recordType: TableName): Promise { + await requestResult(transaction.objectStore(STORES.recordIds).put({ id, recordType } satisfies RecordIdRow)); + } + + private async applyTableWrite(transaction: IDBTransaction, write: StorageWrite, seq: Seq): Promise { + switch (write.type) { + case "conversation": + await this.claimId(transaction, write.value.id, "conversation"); + await requestResult( + transaction.objectStore(STORES.conversations).add({ + id: write.value.id, + ...(write.value.owner === undefined + ? {} + : { + ownerConversationId: write.value.owner.conversationId, + ownerTaskId: write.value.owner.taskId, + }), + record: write.value, + } satisfies ConversationRow), + ); + break; + case "entry": + await this.claimId(transaction, write.value.id, "entry"); + await requestResult( + transaction.objectStore(STORES.entries).add({ + id: write.value.id, + conversationId: write.value.conversationId, + ...(write.value.head === undefined ? {} : { headConversationId: write.value.conversationId }), + commitSeq: seq, + record: write.value, + } satisfies EntryRow), + ); + break; + case "task": + await this.claimId(transaction, write.value.id, "task"); + await requestResult( + transaction.objectStore(STORES.tasks).put({ + id: write.value.id, + conversationId: write.value.conversationId, + kind: write.value.kind, + status: write.value.state.status, + abortRequested: write.value.abortRequested ? 1 : 0, + background: write.value.background ? 1 : 0, + record: write.value, + } satisfies TaskRow), + ); + break; + case "submission": + await this.claimId(transaction, write.value.id, "submission"); + await requestResult( + transaction.objectStore(STORES.submissions).put({ + id: write.value.id, + conversationId: write.value.conversationId, + ...(write.value.requestId === undefined ? {} : { requestId: write.value.requestId }), + status: write.value.status, + record: write.value, + } satisfies SubmissionRow), + ); + break; + case "document.create": + case "document.copy": + case "document.change": + case "document.retire": + break; + } + } + + private async applyDocumentActions( + transaction: IDBTransaction, + actions: ReadonlyMap, + seq: Seq, + ): Promise { + const documents = transaction.objectStore(STORES.documents); + const revisions = transaction.objectStore(STORES.documentRevisions); + for (const [id, action] of actions) { + let content = action.content; + if (action.copy !== undefined) { + try { + const stored = await this.materializeDocument(transaction, 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) { + throw new Error(`Document copy ${id} was rejected`, { cause: error }); + } + } + let record: DocumentRecord; + if (action.create !== undefined) { + record = { + ...action.create, + createdAt: seq, + ...(action.retire ? { retiredAt: seq } : {}), + }; + const parts = addressParts(record); + await this.claimId(transaction, id, "document"); + await requestResult( + documents.add({ + id, + ...parts, + createdAt: seq, + ...(action.retire ? { retiredAt: seq } : {}), + record, + } satisfies DocumentRow), + ); + } else { + record = (await requestResult(documents.get(id) as IDBRequest)).record; + } + + if (content !== undefined) { + if (content.kind === "base" && isCurrentOnly(record)) { + await requestResult(revisions.delete(compoundIdRange([id], 0, MAX_KEY))); + } + await requestResult( + revisions.add({ + documentId: id, + seq, + kind: content.kind, + version: content.version, + content: content.kind === "base" ? content.value : content.ops, + } satisfies RevisionRow), + ); + } + + if (action.retire) { + if (action.create === undefined) { + record = { ...record, retiredAt: seq }; + const row = (await requestResult(documents.get(id) as IDBRequest))!; + await requestResult(documents.put({ ...row, retiredAt: seq, record } satisfies DocumentRow)); + } + if (isCurrentOnly(record)) { + await requestResult(revisions.delete(compoundIdRange([id], 0, MAX_KEY))); + } + } + } + } + + private assertOpen(): void { + if (this.closed) throw new Error("IndexedDbStorage is closed"); + } +} + +export type { IndexedDbStorageOptions }; diff --git a/packages/durable/test/indexeddb-storage.test.ts b/packages/durable/test/indexeddb-storage.test.ts new file mode 100644 index 000000000..40109deaf --- /dev/null +++ b/packages/durable/test/indexeddb-storage.test.ts @@ -0,0 +1,282 @@ +/// + +import type { Context, JsonValue } from "@earendil-works/chord"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { registerStorageConformance } from "@earendil-works/pi-durable/testing"; +import { IDBKeyRange as FakeIDBKeyRange, IDBFactory } from "fake-indexeddb"; +import { describe, expect, it } from "vitest"; +import { idFromNumber } from "../src/ids.ts"; +import { IndexedDbStorage } from "../src/storage/indexeddb.ts"; +import type { + ConversationId, + DocumentCreate, + DocumentId, + EntryId, + Id, + Seq, + Storage, + StorageWrite, + SubmissionId, + TaskId, + TaskRecord, +} from "../src/types.ts"; +import { ROOT_CONVERSATION_ID } from "../src/types.ts"; + +Object.defineProperty(globalThis, "IDBKeyRange", { configurable: true, value: FakeIDBKeyRange }); + +const context = BACKGROUND_CONTEXT; +type StoredTask = TaskRecord; + +class ReopeningStorage implements Storage { + private current: IndexedDbStorage; + private readonly name: string; + private readonly indexedDB: IDBFactory; + private closed = false; + + constructor(current: IndexedDbStorage, name: string, indexedDB: IDBFactory) { + this.current = current; + this.name = name; + this.indexedDB = indexedDB; + } + + async commit(writes: readonly StorageWrite[], commitContext: Context): Promise { + if (this.closed) return this.current.commit(writes, commitContext); + try { + return await this.current.commit(writes, commitContext); + } finally { + await this.current.close(context); + this.current = await IndexedDbStorage.open(this.name, { indexedDB: this.indexedDB }); + } + } + + mintId>(): Promise { + return this.current.mintId(); + } + conversation: Storage["conversation"] = (id, readContext) => this.current.conversation(id, readContext); + scanConversations: Storage["scanConversations"] = (query, limit, cursor, readContext) => + this.current.scanConversations(query, limit, cursor, readContext); + entry(id: EntryId, readContext: Context): ReturnType; + entry(conversationId: ConversationId, id: EntryId, readContext: Context): ReturnType; + entry(idOrConversationId: EntryId | ConversationId, idOrContext: EntryId | Context, readContext?: Context) { + if (readContext === undefined) { + return this.current.entry(idFromNumber(idOrConversationId), idOrContext as Context); + } + if (typeof idOrContext !== "number") throw new TypeError("Storage.entry() requires an entry ID"); + return this.current.entry( + idFromNumber(idOrConversationId), + idFromNumber(idOrContext), + readContext, + ); + } + findLatestHeadMarker: Storage["findLatestHeadMarker"] = (conversationId, at, readContext) => + this.current.findLatestHeadMarker(conversationId, at, readContext); + scanEntries: Storage["scanEntries"] = (query, limit, cursor, readContext) => + this.current.scanEntries(query, limit, cursor, readContext); + task: Storage["task"] = (id, readContext) => this.current.task(id, readContext); + scanTasks: Storage["scanTasks"] = (query, limit, cursor, readContext) => + this.current.scanTasks(query, limit, cursor, readContext); + submission: Storage["submission"] = (id, readContext) => this.current.submission(id, readContext); + scanSubmissions: Storage["scanSubmissions"] = (query, limit, cursor, readContext) => + this.current.scanSubmissions(query, limit, cursor, readContext); + submissionByRequest: Storage["submissionByRequest"] = (conversationId, requestId, readContext) => + this.current.submissionByRequest(conversationId, requestId, readContext); + findDocument: Storage["findDocument"] = (address, at, readContext) => + this.current.findDocument(address, at, readContext); + document: Storage["document"] = (id, at, readContext) => this.current.document(id, at, readContext); + scanDocuments: Storage["scanDocuments"] = (query, limit, cursor, readContext) => + this.current.scanDocuments(query, limit, cursor, readContext); + + async close(closeContext: Context): Promise { + if (this.closed) return; + this.closed = true; + await this.current.close(closeContext); + } +} + +async function withStorage(use: (storage: Storage) => Promise): Promise { + const indexedDB = new IDBFactory(); + const storage = await IndexedDbStorage.open("conformance", { indexedDB }); + try { + await use(storage); + } finally { + await storage.close(context); + } +} + +registerStorageConformance({ describe, expect, it }, "IndexedDbStorage", withStorage); + +registerStorageConformance({ describe, expect, it }, "IndexedDbStorage across reopen", async (use) => { + const indexedDB = new IDBFactory(); + const name = "reopening-conformance"; + const created = await IndexedDbStorage.open(name, { indexedDB }); + await created.close(context); + const storage = new ReopeningStorage(await IndexedDbStorage.open(name, { indexedDB }), name, indexedDB); + try { + await use(storage); + } finally { + await storage.close(context); + } +}); + +function pendingTask(id: TaskId, conversationId = ROOT_CONVERSATION_ID): StoredTask { + return { + id, + conversationId, + kind: "test.task", + version: 1, + input: null, + state: { status: "pending", checkpoint: { phase: "ready" } }, + background: false, + abortRequested: false, + }; +} + +async function createRoot(storage: Storage): Promise { + await storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context); +} + +it("persists minted IDs across reopen", async () => { + const indexedDB = new IDBFactory(); + const storage = await IndexedDbStorage.open("mint-reopen", { indexedDB }); + expect(await storage.mintId()).toBe(2); + await storage.close(context); + const reopened = await IndexedDbStorage.open("mint-reopen", { indexedDB }); + try { + expect(await reopened.mintId()).toBe(3); + } finally { + await reopened.close(context); + } +}); + +it("rolls back table writes when a later document copy fails", async () => { + const indexedDB = new IDBFactory(); + const storage = await IndexedDbStorage.open("atomic-copy", { indexedDB }); + try { + await createRoot(storage); + const sourceId = await storage.mintId(); + const source = { + id: sourceId, + kind: "source", + scope: { kind: "conversation", conversationId: ROOT_CONVERSATION_ID }, + history: "rewindable", + fork: "asOf", + } satisfies DocumentCreate; + await storage.commit( + [{ type: "document.create", record: source, content: { kind: "base", version: 1, value: { count: 1 } } }], + context, + ); + const taskId = await storage.mintId>(); + const copyId = await storage.mintId(); + await expect( + storage.commit( + [ + { type: "task", value: pendingTask(taskId) }, + { + type: "document.copy", + record: { ...source, id: copyId, kind: "mismatch" }, + source: { id: sourceId, at: "current" }, + }, + ], + context, + ), + ).rejects.toThrow("was rejected"); + expect(await storage.task(taskId, context)).toBeUndefined(); + expect(await storage.document(copyId, "current", context)).toBeUndefined(); + } finally { + await storage.close(context); + } +}); + +it("scans entries and head markers through fork ancestry", async () => { + const indexedDB = new IDBFactory(); + const storage = await IndexedDbStorage.open("fork-ancestry", { indexedDB }); + try { + await createRoot(storage); + const rootFirst = await storage.mintId(); + const rootMarker = await storage.mintId(); + await storage.commit( + [ + { type: "entry", value: { id: rootFirst, conversationId: ROOT_CONVERSATION_ID, kind: "first" } }, + { + type: "entry", + value: { + id: rootMarker, + conversationId: ROOT_CONVERSATION_ID, + kind: "marker", + head: rootFirst, + }, + }, + ], + context, + ); + const childId = await storage.mintId(); + await storage.commit( + [ + { + type: "conversation", + value: { id: childId, parent: { conversationId: ROOT_CONVERSATION_ID, at: rootMarker } }, + }, + ], + context, + ); + const childEntry = await storage.mintId(); + await storage.commit( + [{ type: "entry", value: { id: childEntry, conversationId: childId, kind: "child" } }], + context, + ); + expect( + (await storage.scanEntries({ conversationId: childId }, 10, undefined, context)).items.map(({ id }) => id), + ).toEqual([childEntry, rootMarker, rootFirst]); + expect((await storage.findLatestHeadMarker(childId, undefined, context))?.id).toBe(rootMarker); + } finally { + await storage.close(context); + } +}); + +it("keeps one readwrite transaction active through many mixed writes", async () => { + const indexedDB = new IDBFactory(); + const storage = await IndexedDbStorage.open("large-commit", { indexedDB }); + try { + await createRoot(storage); + const writes: StorageWrite[] = []; + let lastEntryId: EntryId | undefined; + let lastTaskId: TaskId | undefined; + let lastSubmissionId: SubmissionId | undefined; + let lastDocumentId: DocumentId | undefined; + for (let index = 0; index < 40; index++) { + lastEntryId = await storage.mintId(); + lastTaskId = await storage.mintId>(); + lastSubmissionId = await storage.mintId(); + lastDocumentId = await storage.mintId(); + writes.push( + { + type: "entry", + value: { id: lastEntryId, conversationId: ROOT_CONVERSATION_ID, kind: "bulk", data: { index } }, + }, + { type: "task", value: pendingTask(lastTaskId) }, + { + type: "submission", + value: { + id: lastSubmissionId, + conversationId: ROOT_CONVERSATION_ID, + requestId: `bulk-${index}`, + type: "input", + status: "queued", + }, + }, + { + type: "document.create", + record: { id: lastDocumentId, kind: "bulk", key: String(index), scope: { kind: "session" } }, + content: { kind: "base", version: 1, value: { index } }, + }, + ); + } + await storage.commit(writes, context); + expect((await storage.entry(lastEntryId!, context))?.entry.data).toEqual({ index: 39 }); + expect(await storage.task(lastTaskId!, context)).toEqual(pendingTask(lastTaskId!)); + expect((await storage.submission(lastSubmissionId!, context))?.requestId).toBe("bulk-39"); + expect((await storage.document(lastDocumentId!, "current", context))?.value).toEqual({ index: 39 }); + } finally { + await storage.close(context); + } +});