mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
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.
This commit is contained in:
Mario Zechner
parent
fabf916fcf
commit
8889e623d5
8 changed files with 814 additions and 57 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
|
|
|
|||
|
|
@ -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`)
|
||||
|
||||
|
|
|
|||
|
|
@ -130,6 +130,8 @@ export class QueryRefresh {
|
|||
}
|
||||
|
||||
async #rerun(step: Step): Promise<readonly StoreWrite[]> {
|
||||
// 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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<ArchivePart>;
|
||||
/**
|
||||
* 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<void>;
|
||||
}
|
||||
|
|
@ -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<string, TableRecord> }
|
||||
|
|
@ -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<void> = 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<number> {
|
||||
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<number>();
|
||||
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<number> {
|
||||
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<void> {
|
||||
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<number>): Promise<void> {
|
||||
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<void> {
|
||||
// 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<void> {
|
||||
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<GroupWritten> {
|
||||
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<number> {
|
||||
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<Series> {
|
||||
): Promise<Written> {
|
||||
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<void> {
|
||||
async #writeCommit(
|
||||
writes: readonly StagedWrite[],
|
||||
executor: SqliteExecutor,
|
||||
progress: Progress,
|
||||
): Promise<Step | undefined> {
|
||||
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<Series> {
|
||||
/** Ends a transaction's writes with the metadata once, after its last commit that wrote; returns what memory takes. */
|
||||
async #closeWrites(progress: Progress): Promise<Written> {
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -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 }]);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<number> | 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<void>();
|
||||
const solo = Promise.withResolvers<void>();
|
||||
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<void>();
|
||||
const solo = Promise.withResolvers<void>();
|
||||
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<void>();
|
||||
const solo = Promise.withResolvers<void>();
|
||||
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" });
|
||||
});
|
||||
});
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue