From 8889e623d5f1b02a81723c44f3c7a72fe9e9dde1 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 4 Nov 2026 19:53:19 +0000 Subject: [PATCH] feat(chord): group commit in the store (decision S18) Commits queued behind a running commit step share the next step's backend transaction: each keeps its own step, seq, outbox commit and adoption in queue order, the metadata is written once. A refusal rolls the group back and its admitted commits run again alone; series, exports, replays and empty commits end a group; backend failures while closing are latched; any throw after COMMIT stops the store with memory at the durable group. --- packages/chord/CHANGELOG.md | 1 + packages/chord/docs/decisions.md | 15 + packages/chord/docs/store/book.md | 5 + packages/chord/src/store/README.md | 3 +- packages/chord/src/store/live-query.ts | 2 + packages/chord/src/store/store.ts | 304 +++++++++-- packages/chord/test/store/live-query.test.ts | 23 +- packages/chord/test/store/store.test.ts | 518 ++++++++++++++++++- 8 files changed, 814 insertions(+), 57 deletions(-) diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index 7bd128e6c..8ed3adde0 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -34,6 +34,7 @@ - Chord v2 plugins import each other by relative path within a root (copied into the importer's script) and across repositories by package name through the package's `exports`, pinned in `dependencies`; slice 10's imports by plugin name (`@app/notes/contract`) are removed, and a relative path leaving the root is a build error. - Chord v2's `host.reload` starts the objects' fates dependents first: removals, then the other fates in the reverse of the order the objects went live (the report keeps its order). A pure dependent reloaded with its provider (a view over a store in one plugin) binds the provider while it is still live and swaps instead of parking until the provider's new instance is live: its contributions are replaced in one batch and its holds over links never show `loading`. A dependent still parks when its new code needs the provider's new version, when its provider went live after it, or when it needs something not live. - Chord v2 keeps calls over a link across a `loading` unbind (decision "Calls in flight survive a `loading` unbind"): a hold whose object restarts, whose swap parks or whose live failure retries no longer answers its calls in flight `unavailable`; each is answered by the instance it already runs on or by the new instance once it is live (or by its failure, `readyTimeout`, its end), and calls a consumer sent by the old names before it saw the `loading` wait the same way instead of `not_ready`. A `failed` unbind (a removed type, a dropped version, a swap parked on a failed dependency) still rejects them. +- Chord v2's store writes the commits queued behind a running commit step in one backend transaction (group commit, decision S18): each keeps its own seq, outbox commit and adoption in queue order, the metadata is written once; a refusal in a group rolls it back and runs its commits one by one, so only the refusing commit is refused. ### Fixed diff --git a/packages/chord/docs/decisions.md b/packages/chord/docs/decisions.md index 1bc4f0fb9..5b4ae73a2 100644 --- a/packages/chord/docs/decisions.md +++ b/packages/chord/docs/decisions.md @@ -793,6 +793,21 @@ validation and copies the v2 kernel added around Delta. order, `tx.update` per row, chunks of N rows per commit; table version `migrating` → `ready`; a crash resumes the scan). Indexes are local and created after `ready` (**superseded in part** by "Index definitions are recorded": the index changes after `ready` are recorded). +- **S18, group commit (owner, 2026-11-04):** the commits queued behind a running commit step form one group, which the + next turn of the commit queue admits at its entry and writes in one backend transaction: each commit keeps its own + step, seq and outbox commit, the metadata is written once, and after `COMMIT` memory follows commit by commit, each + adopted when memory stands at it, in queue order, before the next turn (S1 as for lone steps). Fewer transactions are + always good, and a remote SQLite backend would pay a round trip per statement. A series, an export or a batch apply + queued after an open group ends it (queue order stays); an empty commit never joins one. A refusal in a group (one + commit's own writes throw, D4) rolls the group back whole and its admitted commits run again one by one, not admitted + again, so only the refusing commit is refused; their live queries run again too (`chord.query.runs` counts both + attempts, the statements only the stored one). The ids the rolled-back attempt took stay unused (D3: the allocator + is shared with Durable's `mintId`, never rewound; so within a group's worth of ids of `Number.MAX_SAFE_INTEGER` a + refused group can exhaust the id space a little earlier, accepted). A group that would pass the end of the seq range + runs its commits one by one, so the ones that fit are stored. A backend failure while the store closes (a commit's, + a read's, a maintenance write's) still refuses the admitted commits after it. No savepoints (two statements per + commit, and Durable Objects refuse them) and no check/write split of Chord's writes (their counted conditional writes + are the checks). - **Blobs and derived data (owner, 2026-10-21):** blobs (artifact versions, uploads, sent files) are app layer: files in the session directory named by the hash of their content, never changed, written (tmp, fsync, rename) before the row or document that names them; any process may write one. Export copies `blobs/` with the store snapshot; import diff --git a/packages/chord/docs/store/book.md b/packages/chord/docs/store/book.md index 42036ec68..f5ab11961 100644 --- a/packages/chord/docs/store/book.md +++ b/packages/chord/docs/store/book.md @@ -207,6 +207,11 @@ d -> q: commit q => q: seq 43: write, outbox, adopt Durable ::: +When steps queue up, the queue writes them together (**group commit**, S18): the commits waiting behind a running step +are admitted at the next turn and written in one backend transaction, each still with its own seq, outbox commit and +adoption, in queue order. A refusal inside a group rolls the group back and runs its commits one by one, so it stays +its own commit's. + Durable's seq is the store's seq. Durable sees gaps (other owners' commits in between), and that is harmless: every use of a seq in Durable compares (`created_at <= at`, `seq > base`); nothing computes `seq + 1` to find a neighbour (checked in the storages and the Session). diff --git a/packages/chord/src/store/README.md b/packages/chord/src/store/README.md index adaaa5b12..1868de835 100644 --- a/packages/chord/src/store/README.md +++ b/packages/chord/src/store/README.md @@ -2,7 +2,8 @@ The store (slice 6a): one per host, opened by the app over a backend (SQLite: `chord/sqlite/node`, Durable Objects' storage), holding the host's tables, persisted documents and outbox; one commit queue serializing only the commit -step; eager table migrations; export and import. +step, writing the commits queued behind a running step in one backend transaction (group commit, S18); eager table +migrations; export and import. ## For apps (`index.ts`) diff --git a/packages/chord/src/store/live-query.ts b/packages/chord/src/store/live-query.ts index 777f4f4a9..1d40574a1 100644 --- a/packages/chord/src/store/live-query.ts +++ b/packages/chord/src/store/live-query.ts @@ -130,6 +130,8 @@ export class QueryRefresh { } async #rerun(step: Step): Promise { + // A grouped commit that rolled back applies again alone (store.ts): only the last application's runs count. + this.#runs.length = 0; for (const query of this.#queries) this.#runs.push(await query.rerun(step, this.#context)); return NO_WRITES; } diff --git a/packages/chord/src/store/store.ts b/packages/chord/src/store/store.ts index 8350d0dcf..290f53e8d 100644 --- a/packages/chord/src/store/store.ts +++ b/packages/chord/src/store/store.ts @@ -1,8 +1,9 @@ -// The store (slice 6a §2, §3, slice 6b §1): opened over a backend, one commit queue that serializes only the commit step -// (the seal fence's admission check, `BEGIN`, the writes in their order, the outbox commit, `lastSeq` and the -// allocator's value, `COMMIT`, then the owner's adoption), the id allocator, the claims of its tables and documents, -// reads guarded as Durable's `GuardedStorage` guards them, Durable's Storage semantics for a refused step (D4), and the -// stop after an unknown outcome. +// The store (slice 6a §2, §3, slice 6b §1): opened over a backend, one commit queue that serializes only the commit +// step (the seal fence's admission check, `BEGIN`, the writes in their order, the outbox commit, `lastSeq` and the +// allocator's value, `COMMIT`, then the owner's adoption), group commit (S18: the commits queued behind a running step +// share the next step's backend transaction), the id allocator, the claims of its tables and documents, reads guarded +// as Durable's `GuardedStorage` guards them, Durable's Storage semantics for a refused step (D4), and the stop after an +// unknown outcome. import { awaitWithContext, type Context } from "../context/index.ts"; import { ChordError } from "../data/index.ts"; @@ -33,8 +34,9 @@ export interface Store { */ export(context: Context): AsyncIterableIterator; /** - * Closes the store: waits for a running commit step, rejects queued commits and reads `released`, closes the - * backend. Idempotent, also after `storage_failed`; `context` ends only this caller's wait (the close goes on). + * Closes the store: waits for a running commit step (a group's, its commits' solo runs included), rejects queued + * commits and reads `released`, closes the backend. Idempotent, also after `storage_failed`; `context` ends only + * this caller's wait (the close goes on). */ close(context: Context): Promise; } @@ -105,11 +107,11 @@ export type SeriesCommit = { export type SeriesResult = { readonly stored: number; readonly refused?: unknown }; /** - * A series as its transaction leaves it: each stored commit's seq and created documents; once a commit wrote, the last - * seq, the allocator's value after it and the table records the series changed; the commits that wrote and the - * statements run. + * The commits of one backend transaction (a series or a group) as it leaves them: each stored commit's seq and created + * documents; once a commit wrote, the last seq, the allocator's value after it and the table records they changed; the + * commits that wrote and the statements run. */ -type Series = { +type Written = { readonly stored: readonly { readonly seq: number; readonly created: readonly number[] }[]; readonly last: | { readonly seq: number; readonly nextId: number; readonly tables: ReadonlyMap } @@ -119,11 +121,11 @@ type Series = { }; /** - * A series while its transaction runs: each stored commit's seq and created documents, its last step that wrote, the - * allocator's value after it, the table records it changed (a copy of the catalog, made once one changed) and the - * statements run. + * The commits of one backend transaction (a series or a group) while it runs: each stored commit's seq and created + * documents, the last step that wrote, the allocator's value after it, the table records they changed (a copy of the + * catalog, made once one changed) and the statements run. */ -type SeriesProgress = { +type Progress = { readonly stored: { readonly seq: number; readonly created: readonly number[] }[]; last: Step | undefined; nextId: number; @@ -137,6 +139,21 @@ export type Attachment = { readonly batch: Batch; }; +/** A group as its transaction leaves it: each commit's step and the allocator's value after it, in order; the whole. */ +type GroupWritten = { + readonly steps: readonly { readonly step: Step; readonly nextId: number }[]; + readonly written: Written; +}; + +/** A commit waiting in the commit queue for its group's turn: what `commit` was given, and its caller's promise. */ +type QueuedCommit = { + readonly writes: readonly StagedWrite[]; + readonly fence: CommitFence; + readonly context: Context; + readonly resolve: (seq: number) => void; + readonly reject: (error: unknown) => void; +}; + /** Where a replayed batch leaves the store: its last commit's seq and allocator value, the tables it changed. */ export type Replayed = { readonly lastSeq: number; @@ -168,10 +185,17 @@ export class ChordStore implements Store { readonly claims = new Claims(); #outbox: Outbox | undefined; #state: StoreState = { kind: "open" }; + /** + * The backend error that stopped the store, also when it happened while the store was closing (the state stays + * `closed` then): an admitted commit still running after it is refused. + */ + #failure: { readonly error: unknown } | undefined; /** Set once a batch's recorded count did not reproduce (S17): no batch applies any more. */ #diverged = false; /** The commit queue's tail: each step starts after the one before it settled. */ #tail: Promise = Promise.resolve(); + /** The group the next commit joins: queued last, its turn not yet come; `undefined` once anything queued after it. */ + #open: QueuedCommit[] | undefined; /** Reads and maintenance writes underway, which `close` waits for. */ #underway = 0; /** Resolves `close`'s wait once no read is underway. */ @@ -246,6 +270,7 @@ export class ChordStore implements Store { held.resolve(); }; this.#held.add(end); + this.#open = undefined; const taken = this.#tail.then(() => { // A store closed meanwhile (also before the export's first `next()`) holds nothing: the export ends. if (this.#state.kind === "closed") { @@ -296,12 +321,18 @@ export class ChordStore implements Store { } /** - * Commits `writes` as one seq when the commit queue reaches it: `fence.admit()` at the step's entry, the writes in - * their order (retirements, creations, the rest), the outbox commit, `lastSeq` and the allocator's value, `COMMIT`, - * then `fence.adopt` before the next step starts. Resolves with the seq. A refused admission rejects `released`. A - * step the backend rolled back after the step's own code threw (a refusal, D4) rejects with that error unchanged and - * the store goes on; any other failure (`BEGIN`, `COMMIT`, `ROLLBACK`, a throwing `adopt`) is an unknown outcome - * that stops the store (`storage_failed`). `context` counts the step (no abort once queued). + * Commits `writes` as one seq when the commit queue reaches it: `fence.admit()` at the entry of its step (its + * group's), the writes in their order (retirements, creations, the rest), the outbox commit, `lastSeq` and the + * allocator's value, `COMMIT`, then `fence.adopt` before the next step starts. Resolves with the seq. A refused + * admission rejects `released`. A step the backend rolled back after the step's own code threw (a refusal, D4) + * rejects with that error unchanged and the store goes on; any other failure (`BEGIN`, `COMMIT`, `ROLLBACK`, a + * throwing `adopt` or metrics adapter) is an unknown outcome that stops the store (`storage_failed`). `context` + * counts the step (no abort once queued). + * + * Group commit: the commits queued while a step runs form one group, which the next turn admits and writes in one + * backend transaction (each commit its own seq, outbox commit and adoption, in queue order; the metadata once), so a + * busy store costs a transaction per group, not per commit. If one of them refuses, the group rolls back and its + * commits run again one by one as above: the refusal stays its own commit's. */ commit(writes: readonly StagedWrite[], fence: CommitFence, context: Context): Promise { try { @@ -309,9 +340,23 @@ export class ChordStore implements Store { } catch (error) { return Promise.reject(error); } - const step = this.#tail.then(() => this.#step(writes, fence, context)); - this.#tail = step.then(NOTHING, NOTHING); - return step; + const { promise, resolve, reject } = Promise.withResolvers(); + const queued: QueuedCommit = { writes, fence, context, resolve, reject }; + // An empty commit takes no transaction: it neither joins nor opens a group. + if (writes.length > 0 && this.#open !== undefined) { + this.#open.push(queued); + return promise; + } + const group = [queued]; + // What a turn throws past its own handling (a host's `failed` listener throwing in `#fail`) rejects its commits, + // as it rejected a lone step; rejecting a settled one does nothing. The queue goes on either way. + this.#tail = this.#tail + .then(() => this.#group(group)) + .catch((error: unknown) => { + for (const each of group) each.reject(error); + }); + this.#open = writes.length > 0 ? group : undefined; + return promise; } /** @@ -334,6 +379,7 @@ export class ChordStore implements Store { } catch (error) { return Promise.reject(error); } + this.#open = undefined; const series = this.#tail.then(() => this.#series(commits, receive, context)); this.#tail = series.then(NOTHING, NOTHING); return series; @@ -365,8 +411,10 @@ export class ChordStore implements Store { // A read throwing before its first await is the backend's too: inside the `try`. result = await run(); } catch (error) { - this.assertUsable(); const signal = context.abortSignal; + // Latched before the closed check: a backend failure while closing still refuses later admitted commits. + if (!signal?.aborted) this.#failure ??= { error }; + this.assertUsable(); if (signal?.aborted) throw new ChordError({ code: "aborted", message: "store: read aborted" }, { cause: signal.reason }); throw this.#fail(error); @@ -392,6 +440,8 @@ export class ChordStore implements Store { try { await this.#db.transaction(write); } catch (error) { + // Latched before the closed check, as for a read. + this.#failure ??= { error }; this.assertUsable(); throw this.#fail(error); } finally { @@ -416,11 +466,142 @@ export class ChordStore implements Store { await this.#db.close(); } - async #step(writes: readonly StagedWrite[], fence: CommitFence, context: Context): Promise { - this.assertUsable(); - if (!fence.admit()) { - throw new ChordError({ code: "released", reason: "discarded", message: "store: the commit's attempt ended" }); + /** + * A group's turn (it takes no more commits from here): its commits admitted together at its entry, one as its own + * step, several written in one backend transaction (`#grouped`). Settles every commit's promise. + */ + async #group(group: readonly QueuedCommit[]): Promise { + if (this.#open === group) this.#open = undefined; + try { + this.assertUsable(); + } catch (error) { + for (const queued of group) queued.reject(error); + return; } + // Admitted together at the group's entry, as one step admits its commit: each runs to its end from here. + const admitted = group.filter((queued) => { + try { + if (queued.fence.admit()) return true; + queued.reject(discarded()); + } catch (error) { + // As a lone step: a throwing admission rejects its commit with that error, the queue goes on. + queued.reject(error); + } + return false; + }); + if (admitted.length === 1) { + const [queued] = admitted; + return this.#alone(queued!, () => this.#admittedStep(queued!.writes, queued!.fence, queued!.context)); + } + if (admitted.length > 1) await this.#grouped(admitted); + } + + /** Settles `queued`'s promise with `step`, its own step. */ + async #alone(queued: QueuedCommit, step: () => Promise): Promise { + try { + queued.resolve(await step()); + } catch (error) { + queued.reject(error); + } + } + + /** + * Admitted commits in one backend transaction, each on its own step and seq, the metadata once; after `COMMIT`, + * memory, the counts and each commit's adoption in order. When a commit's own writes throw (a refusal, D4), the + * transaction rolls back whole and every commit runs again alone, admitted already: only the ones whose own writes + * throw are refused, as without grouping. Any other failure is an unknown outcome that stops the store. + */ + async #grouped(admitted: readonly QueuedCommit[]): Promise { + // Near the end of the seq range, alone: the commits that still fit are stored, the first past it stops the store. + if (this.#lastSeq + admitted.length > Number.MAX_SAFE_INTEGER) { + await this.#eachAlone(admitted); + return; + } + let thrown: unknown = NONE; + let done: GroupWritten; + try { + done = await this.#db.transaction(async (executor) => { + try { + return await this.#writeGroup(admitted, executor); + } catch (error) { + thrown = error; + throw error; + } + }); + } catch (error) { + if (error !== thrown) { + const failed = this.#fail(error); + for (const queued of admitted) queued.reject(failed); + return; + } + await this.#eachAlone(admitted); + return; + } + this.#adoptGroup(admitted, done); + } + + /** Runs admitted commits one by one, each its own step, in order. */ + async #eachAlone(admitted: readonly QueuedCommit[]): Promise { + for (const queued of admitted) { + await this.#alone(queued, () => this.#admittedStep(queued.writes, queued.fence, queued.context)); + } + } + + /** The admitted commits' writes in their backend transaction, each on its own step, then the metadata once. */ + async #writeGroup(admitted: readonly QueuedCommit[], executor: SqliteExecutor): Promise { + const progress: Progress = { + stored: [], + last: undefined, + nextId: this.#ids.value, + changed: undefined, + statements: 0, + }; + const steps: { step: Step; nextId: number }[] = []; + for (const queued of admitted) { + // A grouped commit has writes: `commit` never groups an empty one. + const step = (await this.#writeCommit(queued.writes, executor, progress))!; + steps.push({ step, nextId: progress.nextId }); + } + return { steps, written: await this.#closeWrites(progress) }; + } + + /** + * After a group's `COMMIT`, commit by commit as after a lone step: memory follows it, it counts through its own + * context (the metadata statement ran on the last one's step), it is adopted. Anything that throws here (an adoption, + * a metrics adapter) is an unknown outcome: memory takes the whole group (it is durable), the store stops, and that + * commit and the ones after it reject `storage_failed`, unadopted. + */ + #adoptGroup(admitted: readonly QueuedCommit[], { steps, written }: GroupWritten): void { + let index = 0; + try { + for (; index < admitted.length; index++) { + const queued = admitted[index]!; + const { step, nextId } = steps[index]!; + this.#lastSeq = step.seq; + this.#committedNextId = nextId; + for (const [name, record] of step.tables) this.#catalog.set(name, record); + const metrics = chordMetrics(queued.context.telemetry.metrics); + metrics.counter("chord.store.commits").add(1); + metrics.counter("chord.store.statements").add(step.statements); + queued.fence.adopt(step.seq, step.created, step.rows); + queued.resolve(step.seq); + } + } catch (error) { + const last = written.last!; + this.#lastSeq = last.seq; + this.#committedNextId = last.nextId; + for (const [name, record] of last.tables) this.#catalog.set(name, record); + const failed = this.#fail(error); + for (const rest of admitted.slice(index)) rest.reject(failed); + } + } + + /** + * An admitted commit's step (`commit`'s semantics after admission; `fence` only adopts): it runs to its end, also + * when the store closes meanwhile (close waits for it); only a backend failure before it, closing or not, refuses it. + */ + async #admittedStep(writes: readonly StagedWrite[], fence: CommitFence, context: Context): Promise { + if (this.#failure !== undefined) throw storageFailed(this.#failure.error); if (writes.length === 0) { // An empty commit takes no seq; a throwing `adopt` stops the store as for any commit. try { @@ -457,10 +638,12 @@ export class ChordStore implements Store { this.#lastSeq = seq; this.#committedNextId = nextId; for (const [name, record] of step.tables) this.#catalog.set(name, record); - const metrics = chordMetrics(context.telemetry.metrics); - metrics.counter("chord.store.commits").add(1); - metrics.counter("chord.store.statements").add(step.statements); + // Anything that throws from here (a metrics adapter, the adoption) leaves the commit durable and unadopted: an + // unknown outcome that stops the store. try { + const metrics = chordMetrics(context.telemetry.metrics); + metrics.counter("chord.store.commits").add(1); + metrics.counter("chord.store.statements").add(step.statements); fence.adopt(seq, step.created, step.rows); } catch (error) { throw this.#fail(error); @@ -470,7 +653,7 @@ export class ChordStore implements Store { /** * A series inside one backend transaction: per commit a step at the next seq, its check (a throw refuses it, unless - * one of its statements failed), then its writes; then, after `COMMIT`, memory and the receipts as after `#step`. + * one of its statements failed), then its writes; then, after `COMMIT`, memory and the receipts as after a lone step. */ async #series( commits: readonly SeriesCommit[], @@ -487,7 +670,7 @@ export class ChordStore implements Store { } let thrown: unknown = NONE; let refused: { readonly error: unknown } | undefined; - let done: Series; + let done: Written; try { done = await this.#db.transaction(async (executor) => { try { @@ -511,10 +694,11 @@ export class ChordStore implements Store { this.#committedNextId = last.nextId; for (const [name, record] of last.tables) this.#catalog.set(name, record); } - const metrics = chordMetrics(context.telemetry.metrics); - metrics.counter("chord.store.commits").add(written); - metrics.counter("chord.store.statements").add(statements); + // As after a lone step: a throw from here (a metrics adapter, a receipt) stops the store. try { + const metrics = chordMetrics(context.telemetry.metrics); + metrics.counter("chord.store.commits").add(written); + metrics.counter("chord.store.statements").add(statements); for (const [index, { seq, created }] of stored.entries()) receive(index, seq, created); } catch (error) { throw this.#fail(error); @@ -530,10 +714,10 @@ export class ChordStore implements Store { executor: SqliteExecutor, commits: readonly SeriesCommit[], refuse: (error: unknown) => void, - ): Promise { + ): Promise { const calls = { failed: false }; const watched = watchCalls(executor, calls); - const progress: SeriesProgress = { + const progress: Progress = { stored: [], last: undefined, nextId: this.#ids.value, @@ -546,9 +730,9 @@ export class ChordStore implements Store { refuse(refusal.error); break; } - await this.#writeCommit(commit, executor, progress); + await this.#writeCommit(commit.writes, executor, progress); } - return this.#closeSeries(progress); + return this.#closeWrites(progress); } /** @@ -559,7 +743,7 @@ export class ChordStore implements Store { commit: SeriesCommit, watched: SqliteExecutor, calls: { failed: boolean }, - progress: SeriesProgress, + progress: Progress, ): Promise<{ readonly error: unknown } | undefined> { const step = new Step(watched, this.#nextSeq(progress), this.#ids, progress.changed ?? this.#catalog); calls.failed = false; @@ -575,17 +759,22 @@ export class ChordStore implements Store { } /** - * Writes `commit` on its own step at the series' next seq (`#write`), unwatched, and keeps the table records it - * changed in a copy of the catalog, made only once one changed; a commit without writes takes no seq. + * Writes a commit's `writes` on its own step at the series' next seq (`#write`), unwatched, and keeps the table + * records it changed in a copy of the catalog, made only once one changed; resolves with the step. A commit without + * writes takes no seq and no step. */ - async #writeCommit(commit: SeriesCommit, executor: SqliteExecutor, progress: SeriesProgress): Promise { + async #writeCommit( + writes: readonly StagedWrite[], + executor: SqliteExecutor, + progress: Progress, + ): Promise { const seq = this.#nextSeq(progress); - if (commit.writes.length === 0) { + if (writes.length === 0) { progress.stored.push({ seq: seq - 1, created: [] }); - return; + return undefined; } const step = new Step(executor, seq, this.#ids, progress.changed ?? this.#catalog); - progress.nextId = await this.#write(step, commit.writes); + progress.nextId = await this.#write(step, writes); progress.statements += step.statements; if (step.tables.size > 0) { progress.changed ??= new Map(this.#catalog); @@ -593,15 +782,16 @@ export class ChordStore implements Store { } progress.last = step; progress.stored.push({ seq, created: step.created }); + return step; } /** The seq the series' next commit takes. */ - #nextSeq(progress: SeriesProgress): number { + #nextSeq(progress: Progress): number { return (progress.last?.seq ?? this.#lastSeq) + 1; } - /** Ends the series' writes with the metadata, once, after its last commit that wrote; returns what memory takes. */ - async #closeSeries(progress: SeriesProgress): Promise { + /** Ends a transaction's writes with the metadata once, after its last commit that wrote; returns what memory takes. */ + async #closeWrites(progress: Progress): Promise { const { stored, last, nextId, changed, statements } = progress; if (last === undefined) return { stored, last: undefined, written: 0, statements }; await writeMeta(last, nextId); @@ -658,6 +848,7 @@ export class ChordStore implements Store { } catch (error) { return Promise.reject(error); } + this.#open = undefined; const turn = this.#tail.then(() => this.#replay(apply, isDivergence)); this.#tail = turn.then(NOTHING, NOTHING); return turn; @@ -723,8 +914,12 @@ export class ChordStore implements Store { throw new ChordError({ code: "released", reason: "closed", message: "store: closed" }); } - /** Stops the store after `error` (once; its host is told once); returns the error the caller rejects with. */ + /** + * Stops the store after `error` (once; its host is told once), latching it also when the store is closing; returns + * the error the caller rejects with. + */ #fail(error: unknown): ChordError { + this.#failure ??= { error }; if (this.#state.kind === "open") { this.#state = { kind: "failed", error }; this.#attachment?.failed(error); @@ -760,6 +955,11 @@ function watchCalls(executor: SqliteExecutor, calls: { failed: boolean }): Sqlit }; } +/** The `released` error of a commit its fence refused at the step's entry: its attempt ended. */ +function discarded(): ChordError { + return new ChordError({ code: "released", reason: "discarded", message: "store: the commit's attempt ended" }); +} + /** The `unavailable/storage_failed` error of a stopped store, its backend error as the cause. */ function storageFailed(error: unknown): ChordError { return new ChordError( diff --git a/packages/chord/test/store/live-query.test.ts b/packages/chord/test/store/live-query.test.ts index 7ba6c2687..c58b18f26 100644 --- a/packages/chord/test/store/live-query.test.ts +++ b/packages/chord/test/store/live-query.test.ts @@ -1,7 +1,8 @@ // Tests of live queries (decision "Live queries over an object's own tables"): `self.query(table, q, limit)` runs // before its object goes live, again in the commit step of every commit of the object that writes its table, and // publishes the diff in that commit's batch; a commit writing another table runs none; a reload swaps the object; -// the limit is required; a failing run is a failed member; an ended attempt's query is collected. +// the limit is required; a failing run is a failed member; an ended attempt's query is collected; a refresh applied +// again adopts only its last runs. import { describe, expect, it } from "vitest"; import { BACKGROUND_CONTEXT as context } from "../../src/context/index.ts"; @@ -10,8 +11,9 @@ import type { ChordError } from "../../src/data/index.ts"; import { createHost, defineFacet, type FacetHost, type HostError, type StoreSelf } from "../../src/host/index.ts"; import { eq } from "../../src/store/expression.ts"; import { json } from "../../src/store/json-column.ts"; -import { LiveQuery } from "../../src/store/live-query.ts"; +import { LiveQuery, QueryRefresh } from "../../src/store/live-query.ts"; import { query } from "../../src/store/query.ts"; +import type { Step } from "../../src/store/step.ts"; import type { ChordStore } from "../../src/store/store.ts"; import { defineTable } from "../../src/store/table.ts"; import { @@ -333,3 +335,20 @@ describe("memory", () => { await store.close(context); }); }); + +describe("a refresh applied again", () => { + it("adopts its last application's runs only: a grouped commit that rolled back applies again alone", async () => { + const adopted: unknown[] = []; + let runs = 0; + // Stubs: the refresh only calls a query's rerun and adopt, and hands the step through. + const stub = { + rerun: async () => ({ kind: "rows", rows: ++runs }), + adopt: (run: unknown) => void adopted.push(run), + } as unknown as LiveQuery; + const refresh = new QueryRefresh([stub], context); + await refresh.write.apply({} as Step); + await refresh.write.apply({} as Step); + refresh.adopt(); + expect(adopted).toEqual([{ kind: "rows", rows: 2 }]); + }); +}); diff --git a/packages/chord/test/store/store.test.ts b/packages/chord/test/store/store.test.ts index a7519f516..764c1620b 100644 --- a/packages/chord/test/store/store.test.ts +++ b/packages/chord/test/store/store.test.ts @@ -1,11 +1,12 @@ // The store (slice 6a §2, §3, slice 6b §1): opening, the commit step, refusals and unknown outcomes (D4), the id -// allocator (D3), one host per store (rule 7), close. +// allocator (D3), one host per store (rule 7), close, a commit series, group commit. import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { NOOP_TELEMETRY_CONTEXT, NOOP_TELEMETRY_METRICS } from "@earendil-works/pi-telemetry"; import { afterEach, describe, expect, it } from "vitest"; -import { BACKGROUND_CONTEXT as context, withCancel } from "../../src/context/index.ts"; +import { BACKGROUND_CONTEXT as context, withCancel, withTelemetry } from "../../src/context/index.ts"; import { int, str } from "../../src/contracts/index.ts"; import { isChordError } from "../../src/data/index.ts"; import { openNodeSqliteDatabase } from "../../src/sqlite/node.ts"; @@ -577,3 +578,516 @@ describe("a commit series", () => { expect(await rejection(store.read(async () => 1, context))).toMatchObject({ reason: "storage_failed" }); }); }); + +describe("group commit", () => { + /** A store on `:memory:` counting its backend transactions. */ + async function countingStore(options?: { readonly outbox?: boolean }) { + const db = await openNodeSqliteDatabase(":memory:"); + const transaction = db.transaction.bind(db); + let transactions = 0; + db.transaction = (callback) => { + transactions++; + return transaction(callback); + }; + const store = (await openStore(sqliteBackend(db), context, options)) as ChordStore; + // Opening's own transactions (the schema) do not count. + const opened = transactions; + return { store, transactions: () => transactions - opened }; + } + + /** A fence recording `name`'s admission and adoption into `events`. */ + const traced = (events: string[], name: string, admit = true) => ({ + admit: () => { + events.push(`admit ${name}`); + return admit; + }, + adopt: (seq: number, ids: readonly number[]) => void events.push(`adopt ${name} ${seq} ${ids}`), + }); + + const create = (key: string) => createDocument({ ...ADDRESS, key }, BASE, "{}"); + + it("writes the commits queued behind a running step in one transaction, one seq, outbox commit and adoption each", async () => { + const { store, transactions } = await countingStore({ outbox: true }); + const events: string[] = []; + const held = holdNextCommit(store); + const running = store.commit([create("a")], traced(events, "a"), context); + await held.entered; + const queued = ["b", "c", "d"].map((key) => store.commit([create(key)], traced(events, key), context)); + // Admitted at the group's turn, not when queued. + expect(events).toEqual(["admit a"]); + held.release(); + expect(await Promise.all([running, ...queued])).toEqual([1, 2, 3, 4]); + expect(events).toEqual([ + "admit a", + "adopt a 1 2", + "admit b", + "admit c", + "admit d", + "adopt b 2 3", + "adopt c 3 4", + "adopt d 4 5", + ]); + expect(transactions()).toBe(2); + expect([store.lastSeq, store.nextId]).toEqual([4, 6]); + const outbox = await store.outbox!.read(0, 10, context); + expect(outbox.map((commit) => [commit.seq, commit.nextId])).toEqual([ + [1, 3], + [2, 4], + [3, 5], + [4, 6], + ]); + const meta = await store.read( + (executor) => executor.get<{ last_seq: number; next_id: number }>("SELECT last_seq, next_id FROM chord_meta"), + context, + ); + expect(meta).toEqual({ last_seq: 4, next_id: 6 }); + }); + + it("writes the metadata once per group: three statements per delta with its outbox commit, plus one (cost guard)", async () => { + const store = await memoryStore({ outbox: true }); + const id = await created(store); + const held = holdNextCommit(store); + const running = store.commit([create("held")], recording().fence, context); + await held.entered; + const telemetry = await metered(); + const deltas = [1, 2, 3].map((n) => { + const ops = [["s", ["n"], n]] as const; + return store.commit( + [changeDocument(id, n, { kind: "delta", version: 1, ops }, JSON.stringify(ops))], + recording().fence, + telemetry.context, + ); + }); + held.release(); + await Promise.all([running, ...deltas]); + expect(telemetry.counts()).toEqual({ "chord.store.commits": 3, "chord.store.statements": 3 * 3 + 1 }); + }); + + it("rolls a group back when one commit's writes throw, then runs each alone: only that one is refused", async () => { + const { store, transactions } = await countingStore(); + const told: unknown[] = []; + store.attach({ failed: (error) => told.push(error), batch: new Batch(() => {}, undefined) }); + const events: string[] = []; + const held = holdNextCommit(store); + const running = store.commit([create("a")], traced(events, "a"), context); + await held.entered; + const before = store.commit([create("b")], traced(events, "b"), context); + const broken = store.commit([create("x"), BROKEN], traced(events, "x"), context); + const after = store.commit([create("c")], traced(events, "c"), context); + held.release(); + expect(await running).toBe(1); + expect(await before).toBe(2); + expect(String(await rejection(broken))).toContain("no such table"); + expect(await after).toBe(3); + // Admitted once, at the group's entry; the solo runs do not admit again. The rolled back ids stay unused (D3). + expect(events).toEqual(["admit a", "adopt a 1 2", "admit b", "admit x", "admit c", "adopt b 2 5", "adopt c 3 7"]); + // The held step, the rolled back group, then b, x (refused) and c alone. + expect([transactions(), told, store.lastSeq]).toEqual([5, [], 3]); + const keys = await store.read( + (executor) => executor.all<{ key: string }>("SELECT key FROM chord_documents ORDER BY id"), + context, + ); + expect(keys.map((row) => row.key)).toEqual(["a", "b", "c"]); + }); + + it("rejects a commit its fence refuses at the group's entry released, writing nothing; the others commit", async () => { + const store = await memoryStore(); + const events: string[] = []; + const held = holdNextCommit(store); + const running = store.commit([create("a")], traced(events, "a"), context); + await held.entered; + const first = store.commit([create("b")], traced(events, "b"), context); + const cut = store.commit([create("x")], traced(events, "x", false), context); + const last = store.commit([create("c")], traced(events, "c"), context); + held.release(); + expect(await Promise.all([running, first, last])).toEqual([1, 2, 3]); + expect(await rejection(cut)).toMatchObject({ code: "released", reason: "discarded" }); + expect(events.slice(2)).toEqual(["admit b", "admit x", "admit c", "adopt b 2 3", "adopt c 3 4"]); + }); + + it("keeps queue order: a series, an export or an empty commit queued after a group ends it", async () => { + const store = await memoryStore(); + const events: string[] = []; + const held = holdNextCommit(store); + const running = store.commit([create("a")], traced(events, "a"), context); + await held.entered; + const grouped = store.commit([create("b")], traced(events, "b"), context); + const series = store.commitSeries( + [{ check: async () => {}, writes: [create("s")] }], + (_index, seq) => events.push(`receive s ${seq}`), + context, + ); + const afterSeries = store.commit([create("c")], traced(events, "c"), context); + const empty = store.commit([], traced(events, "empty"), context); + const afterEmpty = store.commit([create("d")], traced(events, "d"), context); + held.release(); + await Promise.all([running, grouped, series, afterSeries, empty, afterEmpty]); + expect(events).toEqual([ + "admit a", + "adopt a 1 2", + "admit b", + "adopt b 2 3", + "receive s 3", + "admit c", + "adopt c 4 5", + "admit empty", + "adopt empty 4 ", + "admit d", + "adopt d 5 6", + ]); + }); + + it("runs a commit issued while a group runs at the next turn: a running group takes no more commits", async () => { + const store = await memoryStore(); + const events: string[] = []; + const held = holdNextCommit(store); + const running = store.commit([create("a")], traced(events, "a"), context); + await held.entered; + let late: Promise | undefined; + const issuing = { + admit: () => true, + adopt: (seq: number) => { + events.push(`adopt b ${seq}`); + late = store.commit([create("late")], traced(events, "late"), context); + }, + }; + const grouped = [ + store.commit([create("b")], issuing, context), + store.commit([create("c")], traced(events, "c"), context), + ]; + held.release(); + await Promise.all([running, ...grouped]); + expect(await late).toBe(4); + expect(events).toEqual([ + "admit a", + "adopt a 1 2", + "admit c", + "adopt b 2", + "adopt c 3 4", + "admit late", + "adopt late 4 5", + ]); + }); + + it("lets a later commit of a group see the table records an earlier one changed", async () => { + const store = await memoryStore(); + const { version, record: columns, key, fingerprint } = Edits; + const table = stageTable({ kind: "table", table: TABLE, version, state: "ready", columns, key, fingerprint }); + const seen: (number | undefined)[] = []; + const sees: StagedWrite = { + order: "write", + apply: async (step) => { + seen.push(step.table(TABLE)?.version); + return []; + }, + }; + const held = holdNextCommit(store); + const running = store.commit([create("a")], recording().fence, context); + await held.entered; + const commits = [ + store.commit([table], recording().fence, context), + store.commit([sees], recording().fence, context), + ]; + held.release(); + await Promise.all([running, ...commits]); + expect(seen).toEqual([1]); + expect(store.table(TABLE)?.version).toBe(1); + }); + + it.each([ + ["BEGIN IMMEDIATE", undefined], + ["COMMIT", undefined], + ["ROLLBACK", BROKEN], + ] as const)( + "a failed %s of a group is an unknown outcome: every commit rejects storage_failed, the host told once", + async (at, write) => { + const { store, fail } = await faultyStore(); + const told: unknown[] = []; + store.attach({ failed: (error) => told.push(error), batch: new Batch(() => {}, undefined) }); + const held = holdNextCommit(store); + // Armed after the held step's `COMMIT`, before the group's turn: the group's statement fails. + const running = store.commit([create("a")], { admit: () => true, adopt: () => fail(at) }, context); + await held.entered; + const group = [ + store.commit([create("b")], recording().fence, context), + store.commit([write ?? create("c")], recording().fence, context), + ]; + held.release(); + expect(await running).toBe(1); + for (const commit of group) { + expect(await rejection(commit)).toMatchObject({ code: "unavailable", reason: "storage_failed" }); + } + expect([told.length, store.lastSeq]).toEqual([1, 1]); + }, + ); + + it("stops the store when an adoption in a group throws: it and the ones after it reject, the ones before stand", async () => { + const store = await memoryStore(); + const held = holdNextCommit(store); + const running = store.commit([create("a")], recording().fence, context); + await held.entered; + const throwing = { + admit: () => true, + adopt: () => { + throw new Error("adopt threw"); + }, + }; + const first = store.commit([create("b")], recording().fence, context); + const second = store.commit([create("c")], throwing, context); + const third = store.commit([create("d")], recording().fence, context); + held.release(); + expect(await Promise.all([running, first])).toEqual([1, 2]); + expect(await rejection(second)).toMatchObject({ reason: "storage_failed" }); + expect(await rejection(third)).toMatchObject({ reason: "storage_failed" }); + expect(await rejection(store.read(async () => 1, context))).toMatchObject({ reason: "storage_failed" }); + }); + + it("finishes the admitted commits of a group when the store closes during their solo runs; queued ones are released", async () => { + const store = await memoryStore(); + const held = holdNextCommit(store); + const running = store.commit([create("a")], recording().fence, context); + await held.entered; + // Its second application (the solo run after the group's rollback) waits for the close to begin. + const gate = Promise.withResolvers(); + const solo = Promise.withResolvers(); + let applications = 0; + const slow: StagedWrite = { + order: "write", + apply: async () => { + if (++applications === 2) { + solo.resolve(); + await gate.promise; + } + return []; + }, + }; + const admitted = store.commit([create("b"), slow], recording().fence, context); + const broken = store.commit([BROKEN], recording().fence, context); + const later = store.commit([create("c")], recording().fence, context); + held.release(); + await solo.promise; + const queued = store.commit([create("q")], recording().fence, context); + const closed = store.close(context); + gate.resolve(); + expect(await running).toBe(1); + expect(await admitted).toBe(2); + expect(String(await rejection(broken))).toContain("no such table"); + expect(await later).toBe(3); + expect(await rejection(queued)).toMatchObject({ code: "released" }); + await closed; + }); + + it("refuses the admitted commits after a backend failure while the store closes", async () => { + const { store, fail } = await faultyStore(); + const held = holdNextCommit(store); + const running = store.commit([create("a")], recording().fence, context); + await held.entered; + // Its second application is its solo run after the group's rollback: the close and the failure land there. + const gate = Promise.withResolvers(); + const solo = Promise.withResolvers(); + let applications = 0; + const slow: StagedWrite = { + order: "write", + apply: async () => { + if (++applications === 2) { + solo.resolve(); + await gate.promise; + } + return []; + }, + }; + const first = store.commit([create("b"), slow], recording().fence, context); + const broken = store.commit([BROKEN], recording().fence, context); + const last = store.commit([create("c")], recording().fence, context); + held.release(); + expect(await running).toBe(1); + await solo.promise; + fail("COMMIT"); + const closed = store.close(context); + gate.resolve(); + for (const commit of [first, broken, last]) { + expect(await rejection(commit)).toMatchObject({ code: "unavailable", reason: "storage_failed" }); + } + await closed; + expect(store.lastSeq).toBe(1); + }); + + it("runs a group that would pass the end of the seq range alone: the commits that fit are stored", async () => { + const [manifest, ...rest] = await exported(await memoryStore()); + const parts = [{ ...manifest!, lastSeq: Number.MAX_SAFE_INTEGER - 2 }, ...rest]; + const store = (await importStore( + sqliteBackend(await openNodeSqliteDatabase(":memory:")), + parts, + context, + )) as ChordStore; + const held = holdNextCommit(store); + const running = store.commit([create("a")], recording().fence, context); + await held.entered; + const fits = store.commit([create("b")], recording().fence, context); + const past = store.commit([create("c")], recording().fence, context); + held.release(); + expect(await Promise.all([running, fits])).toEqual([Number.MAX_SAFE_INTEGER - 1, Number.MAX_SAFE_INTEGER]); + expect(await rejection(past)).toMatchObject({ code: "unavailable", reason: "storage_failed" }); + }); + + it("rejects a grouped commit whose admission throws with that error; the others commit, the queue goes on", async () => { + const store = await memoryStore(); + const held = holdNextCommit(store); + const running = store.commit([create("a")], recording().fence, context); + await held.entered; + const thrown = new Error("admit threw"); + const throwing = { + admit: () => { + throw thrown; + }, + adopt: () => {}, + }; + const refused = store.commit([create("x")], throwing, context); + const admitted = store.commit([create("b")], recording().fence, context); + held.release(); + expect(await rejection(refused)).toBe(thrown); + expect(await Promise.all([running, admitted])).toEqual([1, 2]); + expect(await store.commit([create("c")], recording().fence, context)).toBe(3); + }); + + it("refuses the admitted commits after a maintenance write fails while the store closes", async () => { + const { store, fail } = await faultyStore(); + const held = holdNextCommit(store); + const running = store.commit([create("a")], recording().fence, context); + await held.entered; + const gate = Promise.withResolvers(); + const solo = Promise.withResolvers(); + let applications = 0; + const slow: StagedWrite = { + order: "write", + apply: async () => { + if (++applications === 2) { + solo.resolve(); + await gate.promise; + } + return []; + }, + }; + const first = store.commit([create("b"), slow], recording().fence, context); + const broken = store.commit([BROKEN], recording().fence, context); + const last = store.commit([create("c")], recording().fence, context); + held.release(); + expect(await running).toBe(1); + await solo.promise; + // Queued behind the solo run's transaction; its COMMIT fails once the store is closing. + const maintenance = store.maintain(async (executor) => { + await executor.run("SELECT 1"); + fail("COMMIT"); + }); + const closed = store.close(context); + gate.resolve(); + expect(await first).toBe(2); + expect(await rejection(maintenance)).toMatchObject({ code: "released" }); + await rejection(broken); + expect(await rejection(last)).toMatchObject({ code: "unavailable", reason: "storage_failed" }); + await closed; + expect(store.lastSeq).toBe(2); + }); + + it("rejects a group's commits with what its turn throws after COMMIT, and the queue goes on", async () => { + const store = await memoryStore(); + const thrown = new Error("listener threw"); + store.attach({ + failed: () => { + throw thrown; + }, + batch: new Batch(() => {}, undefined), + }); + const throwing = { + admit: () => true, + adopt: () => { + throw new Error("adopt threw"); + }, + }; + const held = holdNextCommit(store); + const running = store.commit([create("a")], recording().fence, context); + await held.entered; + const group = [ + store.commit([create("b")], throwing, context), + store.commit([create("c")], recording().fence, context), + ]; + held.release(); + expect(await running).toBe(1); + for (const commit of group) expect(await rejection(commit)).toBe(thrown); + expect(await rejection(store.commit([create("d")], recording().fence, context))).toMatchObject({ + reason: "storage_failed", + }); + }); + + it("advances memory commit by commit while a group adopts, as lone steps do", async () => { + const store = await memoryStore(); + const seen: [number, number, number][] = []; + const fence = { + admit: () => true, + adopt: (seq: number) => void seen.push([seq, store.lastSeq, store.nextId]), + }; + const held = holdNextCommit(store); + const running = store.commit([create("a")], recording().fence, context); + await held.entered; + const group = [store.commit([create("b")], fence, context), store.commit([create("c")], fence, context)]; + held.release(); + await Promise.all([running, ...group]); + expect(seen).toEqual([ + [2, 2, 4], + [3, 3, 5], + ]); + }); + it("stops the store when a metrics adapter throws after a group's COMMIT: memory takes the whole group, no seq repeats", async () => { + const store = await memoryStore(); + const throwing = withTelemetry( + { + ...NOOP_TELEMETRY_CONTEXT, + metrics: { + ...NOOP_TELEMETRY_METRICS, + counter: () => ({ + add: () => { + throw new Error("metrics threw"); + }, + }), + }, + }, + context, + ); + const held = holdNextCommit(store); + const running = store.commit([create("a")], recording().fence, context); + await held.entered; + const group = [ + store.commit([create("b")], recording().fence, throwing), + store.commit([create("c")], recording().fence, context), + ]; + held.release(); + expect(await running).toBe(1); + for (const commit of group) expect(await rejection(commit)).toMatchObject({ reason: "storage_failed" }); + expect(store.lastSeq).toBe(3); + expect(await rejection(store.commit([create("d")], recording().fence, context))).toMatchObject({ + reason: "storage_failed", + }); + }); + + it("stops the store when a metrics adapter throws after a lone commit's COMMIT too", async () => { + const store = await memoryStore(); + const throwing = withTelemetry( + { + ...NOOP_TELEMETRY_CONTEXT, + metrics: { + ...NOOP_TELEMETRY_METRICS, + counter: () => ({ + add: () => { + throw new Error("metrics threw"); + }, + }), + }, + }, + context, + ); + const { adopted, fence } = recording(); + expect(await rejection(store.commit([create("a")], fence, throwing))).toMatchObject({ reason: "storage_failed" }); + expect([adopted, store.lastSeq]).toEqual([[], 1]); + expect(await rejection(store.read(async () => 1, context))).toMatchObject({ reason: "storage_failed" }); + }); +});