From ff03ebbbffcbb6a3678a5d0c06a172162bc72959 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 20:41:49 +0000 Subject: [PATCH 1/2] Check the room as a history under concurrent clients, and state what the record promises docs/durability.md is the contract: what a delivery, a read and a lease promise when the process, the storage, the wire or a model fails, what the room does not promise yet, what a host must do, and where each claim is proved. test/support/history.ts records every host action as an invocation and an outcome: ok landed, fail was refused with nothing landed, and info is an outcome the client cannot tell. The checker reads the history, the record and the storage together and reports every guarantee that broke: an acknowledged delivery on the record once, one in doubt at most once, a refused one never; every read a prefix, forward only, holding the client's own deliveries; every seq on the storage once; one attempt at a wake at a time; nothing running, pending or owed after the drain. test/consistency.test.ts runs two people and the host at once under a nemesis that crashes the run, fails the storage, faults the wire and jumps the clock, over 25 seeds by default and 200 widened. test/split.test.ts pins the split the design forbids as a history: in memory a seq lands twice and an acknowledged delivery is off the record; on JSONL, Pi refuses to load the file and no run can open the name again. Backlog item 35 says what a fence takes; item 36 names the clocks and the storage faults no test moves yet. The history found one weakness. A claim whose confirmation was lost left the seat with no lease in hand and the row live for the whole expiry. The seat asks a claim and a release again once: a claim of an id the room already runs is a renewal, so the row lands once. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01SJjjR3S6iwqEbDXsRDJgiq --- CLAUDE.md | 31 +- README.md | 7 +- docs/agent.md | 4 +- docs/durability.md | 163 +++++++++ docs/toolchain.md | 14 +- package.json | 2 +- packages/ambion/src/seat/seat.ts | 38 ++- packages/ambion/test/consistency.test.ts | 368 +++++++++++++++++++++ packages/ambion/test/lease.test.ts | 9 +- packages/ambion/test/split.test.ts | 206 ++++++++++++ packages/ambion/test/support/history.ts | 243 ++++++++++++++ packages/ambion/test/support/invariants.ts | 4 +- planning/backlog.md | 36 ++ 13 files changed, 1089 insertions(+), 36 deletions(-) create mode 100644 docs/durability.md create mode 100644 packages/ambion/test/consistency.test.ts create mode 100644 packages/ambion/test/split.test.ts create mode 100644 packages/ambion/test/support/history.ts diff --git a/CLAUDE.md b/CLAUDE.md index 5572c5ad..d9e6296e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -12,20 +12,21 @@ the room goes quiet. pnpm workspace, Node >= 22.19, ESM only, TypeScript. -| Path | What | -| ------------------- | -------------------------------------------------------------------------------------- | -| `packages/ambion` | The runtime. One file per concern, in layers Biome holds; `session.ts` composes them | -| `packages/cli` | The `ambion` binary | -| `docs/agent.md` | Design contract for the core — read before changing the runtime | -| `docs/exchange.md` | Design contract for the exchange, the room's unit of work — read with `agent.md` | -| `docs/presence.md` | Design contract for presence and visits — read with `agent.md` | -| `docs/assistant.md` | Design contract for the assistant, the room's counterpart to the people in it | -| `docs/workspace.md` | Design contract for the workspace an agent's tools reach into — read with `agent.md` | -| `docs/roster.md` | Design contract for a roster that changes while the room runs — read with `agent.md` | -| `docs/toolchain.md` | Build, CI, release — read before changing `.github/`, `scripts/`, root configs | -| `examples/site` | Runnable example | -| `demos/` | One dated report per merged change — regenerate on the branch, then leave it | -| `planning/` | `backlog.md` holds the debt in what is built and the work deferred; `next.md` the five | +| Path | What | +| -------------------- | ----------------------------------------------------------------------------------------- | +| `packages/ambion` | The runtime. One file per concern, in layers Biome holds; `session.ts` composes them | +| `packages/cli` | The `ambion` binary | +| `docs/agent.md` | Design contract for the core — read before changing the runtime | +| `docs/exchange.md` | Design contract for the exchange, the room's unit of work — read with `agent.md` | +| `docs/presence.md` | Design contract for presence and visits — read with `agent.md` | +| `docs/assistant.md` | Design contract for the assistant, the room's counterpart to the people in it | +| `docs/workspace.md` | Design contract for the workspace an agent's tools reach into — read with `agent.md` | +| `docs/roster.md` | Design contract for a roster that changes while the room runs — read with `agent.md` | +| `docs/durability.md` | What the record promises under failure, and how the tiers prove it — read with `agent.md` | +| `docs/toolchain.md` | Build, CI, release — read before changing `.github/`, `scripts/`, root configs | +| `examples/site` | Runnable example | +| `demos/` | One dated report per merged change — regenerate on the branch, then leave it | +| `planning/` | `backlog.md` holds the debt in what is built and the work deferred; `next.md` the five | ## Thesis @@ -59,7 +60,7 @@ pnpm install pnpm check # build, typecheck, lint, test — the gate CI runs pnpm format # biome --write, then prettier --write pnpm test:live # the room on a real model; needs _API_KEY and costs money -pnpm chaos # the crash sweep on both storages, the handover at every write, the kill at every third write, 200 seeds of the walk +pnpm chaos # the sweeps on both storages, the handover at every write, the kill at every third write, 200 seeds of the walk and the history ``` Run `pnpm format` and `pnpm check` before every push. CI runs the same gate. diff --git a/README.md b/README.md index 8ac15456..16eb1751 100644 --- a/README.md +++ b/README.md @@ -260,9 +260,10 @@ npm install @ambionframework/ambion The design is specified in [`docs/agent.md`](docs/agent.md), [`docs/exchange.md`](docs/exchange.md), [`docs/presence.md`](docs/presence.md), -[`docs/assistant.md`](docs/assistant.md), [`docs/roster.md`](docs/roster.md), and -[`docs/workspace.md`](docs/workspace.md). Build and contribution instructions -are in [`CONTRIBUTING.md`](CONTRIBUTING.md). +[`docs/assistant.md`](docs/assistant.md), [`docs/roster.md`](docs/roster.md), +[`docs/workspace.md`](docs/workspace.md), and [`docs/durability.md`](docs/durability.md), +which says what the record promises when something fails. Build and +contribution instructions are in [`CONTRIBUTING.md`](CONTRIBUTING.md). ## License diff --git a/docs/agent.md b/docs/agent.md index 43d40ea4..b5ca17c6 100644 --- a/docs/agent.md +++ b/docs/agent.md @@ -323,7 +323,9 @@ pending again when the lease expired or failed, so a run that dies while the seat works loses nothing: the seat is woken for the message after the backoff. A wake to a seat at rest is on the message, so a wake lost on the way is sent again after the resend window, and the seat side runs a -wake sent twice once. When a pass ends, +wake sent twice once. A claim or a release the seat never heard back on +is asked again once: a claim of an id the room already runs is a +renewal, so the row lands once whichever call reached it. When a pass ends, the activation renews its lease, and the renewal says how far the record reaches. An activation that heard less than that reads the room again through a fresh view. diff --git a/docs/durability.md b/docs/durability.md new file mode 100644 index 00000000..890dd795 --- /dev/null +++ b/docs/durability.md @@ -0,0 +1,163 @@ +# Durability + +What the room promises about its record when the process, the storage, +the wire or a model fails, and how the promise is proved. This page is +the contract the chaos tier and the history checker hold the room to. +Read it with [`agent.md`](agent.md) §5, which names the mechanisms, and +[`toolchain.md`](toolchain.md) §8, which says how the tiers run. + +## 1. The log is the truth + +**One record, one writer, one order.** A room's record is one append-only +log in a Pi session. Every message takes the next seq, and every row the +room writes beside the messages carries `after`, the last seq when it +landed. The fold reads the log from the start and rebuilds the room from +it: the roster, the people, the open exchange, every lease, every wake +still pending and every summary still owed. Nothing the room holds in +memory outlives what the log says. + +**Durable means the storage's append resolved.** Pi's in-memory repository +holds the record for the life of the process. Pi's JSONL repository +writes every entry to a file and calls no `fsync`. A storage that lies +about an append breaks every promise below. + +**One run per name.** One process holds a name at a time. A second live +run over the same log is the one fault the room does not survive yet: +§5 says what happens. + +## 2. What a delivery promises + +**Acknowledged: on the record once.** `deliver()` resolves once the +write is confirmed. The message is on the record, on the stream, and the +seats it reaches are woken for it. It stays on the record for the life +of the log. + +**Refused: nowhere.** A delivery the room refuses rejects `deliver()` +before anything lands: the visit is over, the room is stopped, or the +recipient is not in the room. Nothing is on the record, nothing is on +the stream, and nobody woke. + +**In doubt: at most once.** A write the storage failed, or a process that +died with the write in flight, leaves the host without an answer. The +message is on the record or it is not. The room reads the storage back +before its next write, so a message that landed is on the record before +anything lands on top of it. `deliver({ key })` names the delivery: a +host that never learned whether a delivery landed delivers it again +under the same key, and the key lands once. + +## 3. What a read promises + +**A prefix, in order.** `messages()` returns the record from seq 1 to +the last seq the run has confirmed or read back, with no gap. A read +never shows a message before the write that carries it is confirmed. + +**Forward only.** Two reads by one host over one run never move +backwards. A resumed run replays the whole log first, so a read after a +resume holds everything the run before it confirmed. + +**Your own writes.** A read after an acknowledged delivery holds that +delivery. + +**The stream is the push side.** A listener learns nothing the pulls +cannot tell it. One `message` event per message, in record order, from +the first message the run saw. + +## 4. What a lease promises + +**One attempt at a time.** A wake names one message and one seat. The +seat claims a lease under the wake's id and renews it while it works. +The next attempt claims only after the last one ended. + +**A lease answers what it heard.** The log says which messages a lease +heard: the ones it was at work for, and the ones its view held because +it was claimed after them. A lease answers them while it runs and once +it stood down, through the seq its last renewal confirmed. + +**A lease that came to nothing answers nothing.** A lease that expired +or failed leaves every message it heard pending again, whatever it said. +Its words stay on the record, and the seat reads them at the next +attempt. The failure is one attempt. The room wakes the seat again after +the backoff, and at the cap it stops. `runtime.retry` holds the policy: +three attempts, thirty seconds apart, by default. + +**A claim or a release asked twice lands once.** A claim of an id the +room already runs is a renewal. The seat asks again once when it never +heard back, and a release lost twice leaves the room to expire the lease +on its side. + +## 5. What the room does not promise + +**Two live hosts over one log.** A host that is paused, and a second host +that resumes the name while it is paused, both write from their own last +seq once the first comes back. In memory, a seq is on the storage twice +and a delivery the first host acknowledged is off the record the second +host reads. On JSONL, Pi refuses to load the file, and no run can open +the name again. `split.test.ts` pins both. A fence is deferred: +`planning/backlog.md` item 35 says what it takes. + +**A storage that tears.** A partial line at the end of a JSONL file, a +lost `fsync`, or entries the storage reorders are not exercised. The +room reads what the storage returns. + +**Clocks that disagree.** Every host in the tests shares one clock. A +resumed host whose clock runs ahead of the last run's expires its leases +early, and one that runs behind holds them past their time. No test +moves two clocks apart. + +**A model that repeats itself.** A seat woken again reads its own earlier +words on the record. The room hands them to the model, and the model +decides what to add. A scripted model stands down; a real one is held +by its instructions. + +## 6. What a host must do + +- Retry a delivery it never heard back on under the same key. +- Resume a name after the process that ran it died, with + `resumeSession(name, { runtime })`, and let the first reconcile expire + what the dead run held. +- Run one host per name. Evict a room with `runtime.evict(name)` before + another host takes it, and never continue a host that was paused past + its leases. +- Read `messages()` after a resume for what the stream did not carry. + +## 7. How it is proved + +The scripted tier runs every proof in process, on a fake clock and a +scripted model, so a run is deterministic and a sweep is exhaustive. + +| Proof | Test | What it holds | +| ---------------------- | --------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | +| A crash at every write | `chaos.test.ts` | One scenario, crashed before and after every append it takes, resumed and retried under the same key, ends with the same record every time | +| A handover under load | `hosts.test.ts` | The same sweep with a model that fails and a seat whose say wakes a peer: the second host wakes the failed seat again, and every answer lands once | +| A kill from outside | `chaos.test.ts` | The scenario in a child process on JSONL, killed with `SIGKILL` at a write, resumed over the directory | +| The random walk | `property.test.ts` | Twenty seeded steps of visits, deliveries, seat changes, clock jumps, wire faults, disk faults and crashes; the invariants hold | +| The history | `consistency.test.ts` | Two people and the host act at once under a nemesis; every action is an invocation and an outcome; §2 to §4 are checked against the record | +| The split | `split.test.ts` | A paused host comes back after a takeover, in process and as a process under `SIGSTOP`; what §5 says happens, happens | + +**The history checker** lives in +[`test/support/history.ts`](../packages/ambion/test/support/history.ts). +Every host action is recorded as `invoke`, then `ok`, `fail` or `info`: +`ok` landed, `fail` was refused with nothing landed, and `info` is an +outcome the client cannot tell, because the storage, the wire or the +process failed under it. The checker reads the history, the record and +the storage together and reports every guarantee that broke: + +- an acknowledged delivery is on the record once, one in doubt at most + once, a refused one never; +- every read is a prefix of the record, a client's reads move forward, + and they hold what the client delivered; +- every seq on the storage names one message; +- one attempt at a wake runs at a time; +- once the room drains, nothing runs, nothing is pending, nothing is + owed. + +**The nemesis** crashes the run and resumes it in a fresh runtime, fails +the next write before or after it lands, drops, repeats and delays +requests on the wire, and jumps the clock the way a paused process sees +it. Every error the room reports under it is bounded: the leases a +resumed run inherited, the failures the cast injects, the requests the +nemesis dropped, and the leases live across a jump past the expiry. + +`pnpm test` runs 25 seeds of the history and the walk; `pnpm chaos` +runs 200 of each, the sweep on JSONL too, the handover at every write, +and the kill at every third write. diff --git a/docs/toolchain.md b/docs/toolchain.md index 0fb70131..47a8672a 100644 --- a/docs/toolchain.md +++ b/docs/toolchain.md @@ -352,10 +352,20 @@ harness in - **The random walk** (`property.test.ts`) loses and repeats requests on the wire, fails a write before or after it lands, and crashes the room up to three times. +- **The history** (`consistency.test.ts`). Two people and the host act + at once against whichever run holds the room, under a nemesis that + crashes the run, fails the storage, faults the wire and jumps the + clock. Every action is recorded as an invocation and an outcome, and + the checker in `test/support/history.ts` holds the history to the + record: what [`durability.md`](durability.md) §2 to §4 promise. +- **The split** (`split.test.ts`). A paused host comes back after a + second host resumed the name, in process and as a process under + `SIGSTOP`. The tests pin what [`durability.md`](durability.md) §5 + says happens, and turn when a fence lands. `AMBION_CHAOS=all` widens the sweep to JSONL, the handover to every -write, and the kill to every third write; `pnpm chaos` runs all three -widened, with 200 seeds of the walk. +write, and the kill to every third write; `pnpm chaos` runs all of them +widened, with 200 seeds of the walk and of the history. `pnpm test:live` runs the tier. Two configurations keep the tiers apart: `vitest.config.ts` excludes `test/live` from `pnpm test`, and diff --git a/package.json b/package.json index aed60f73..6c2977bc 100644 --- a/package.json +++ b/package.json @@ -13,7 +13,7 @@ "build": "turbo build", "test": "turbo test", "test:live": "pnpm --filter @ambionframework/ambion run test:live", - "chaos": "AMBION_CHAOS=all AMBION_SEEDS=200 pnpm --filter @ambionframework/ambion exec vitest run test/chaos.test.ts test/hosts.test.ts test/property.test.ts", + "chaos": "AMBION_CHAOS=all AMBION_SEEDS=200 pnpm --filter @ambionframework/ambion exec vitest run test/chaos.test.ts test/hosts.test.ts test/property.test.ts test/consistency.test.ts test/split.test.ts", "check": "turbo run build check:types && pnpm run check:lint && turbo run test", "check:lint": "biome lint . --error-on-warnings && knip", "check:format": "prettier . --cache --check", diff --git a/packages/ambion/src/seat/seat.ts b/packages/ambion/src/seat/seat.ts index e65e85f4..38b32b65 100644 --- a/packages/ambion/src/seat/seat.ts +++ b/packages/ambion/src/seat/seat.ts @@ -172,14 +172,22 @@ export class SeatActor implements SeatPort { await this.next(); } - /** The lease, or nothing: the room refused it, or the claim never came back. */ + /** + * The lease, or nothing: the room refused it, or the claim never came + * back twice. A claim the seat never heard back on is asked again once: + * a claim of an id the room already runs is a renewal, so the row lands + * once whichever call reached it. + */ private async claim(id: string): Promise<{ expiry: number } | undefined> { - try { - const claimed = await this.room.lease({ activation: id, phase: 'running' }); - return 'stale' in claimed ? undefined : claimed.ok; - } catch { - return undefined; + for (let attempt = 0; attempt < 2; attempt += 1) { + try { + const claimed = await this.room.lease({ activation: id, phase: 'running' }); + return 'stale' in claimed ? undefined : claimed.ok; + } catch { + // The claim never came back: asked again, once. + } } + return undefined; } /** The wake that queued, to its end. */ @@ -189,12 +197,20 @@ export class SeatActor implements SeatPort { if (queued !== undefined) await this.take(queued); } - /** The lease is released, however the activation went. A room that is gone answers stale, and that is fine. */ + /** + * The lease is released, however the activation went. A release the seat + * never heard back on is asked again once; a lease that ended answers + * stale, and that is fine. A release lost twice leaves the room to end + * the lease on its side. + */ private async release(id: string, activation: Activation): Promise { - try { - await this.room.lease({ activation: id, phase: 'ended', reason: activation.reason }); - } catch { - // The release never reached the room. The room ends the lease on its side. + for (let attempt = 0; attempt < 2; attempt += 1) { + try { + await this.room.lease({ activation: id, phase: 'ended', reason: activation.reason }); + return; + } catch { + // The release never came back: asked again, once. + } } } diff --git a/packages/ambion/test/consistency.test.ts b/packages/ambion/test/consistency.test.ts new file mode 100644 index 00000000..e1b1d677 --- /dev/null +++ b/packages/ambion/test/consistency.test.ts @@ -0,0 +1,368 @@ +/** + * The room under concurrent clients and a nemesis, checked as a history. + * Two people and the host act at once against whichever run holds the + * room, while the nemesis crashes the run, fails the storage, and loses, + * repeats and delays requests on the wire. Every action is recorded as + * an invocation and an outcome, and the checks in `support/history.ts` + * hold the history to the record the room ends with. + * + * `AMBION_SEEDS` widens the walk; the seed prints on failure. + */ +import { describe, expect, it } from 'vitest'; +import { + createRuntime, + inProcessTransport, + type Runtime, + resumeSession, + type Session, + type SessionEvent, + startSession, + stopSession, + type Visit, + visitSession, +} from '../src/index.ts'; +import { foldRoom } from '../src/room/fold.ts'; +import { agents, assistant, colleague, priya, product, sam, troubled } from './support/cast.ts'; +import { liveLeases } from './support/chaos.ts'; +import { type FakeClock, fakeClock } from './support/clock.ts'; +import { History, violations } from './support/history.ts'; +import { invariants } from './support/invariants.ts'; +import { roomName, rowsOf } from './support/room.ts'; +import { scripted } from './support/scripted.ts'; +import { type FailMode, memory, tappedOpener } from './support/storage.ts'; +import { type Fault, faultyTransport, type Operation, serializing } from './support/transport.ts'; + +function mulberry32(seed: number): () => number { + let a = seed >>> 0; + return () => { + a = (a + 0x6d2b79f5) >>> 0; + let t = a; + t = Math.imul(t ^ (t >>> 15), t | 1); + t ^= t + Math.imul(t ^ (t >>> 7), t | 61); + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; +} + +const OPERATIONS: Operation[] = ['wake', 'view', 'commit', 'lease']; +const RETRY = { attempts: 3, backoff: (attempt: number) => attempt * 30_000 }; + +/** One room over one storage, and the run that holds it now. */ +class Cluster { + readonly clock: FakeClock = fakeClock(); + readonly history = new History(this.clock); + readonly faults: Fault[] = []; + readonly cast = troubled(); + /** Which run holds the room: a visit taken on an earlier run is over. */ + epoch = 0; + events: SessionEvent[] = []; + inherited = { activations: 0, exchange: false }; + runtime!: Runtime; + session!: Session; + private disk: FailMode = false; + private failedBefore = 0; + /** Requests the nemesis dropped, counted when taken: every one may fail an activation, and that is one error. */ + private dropped = 0; + private droppedBefore = 0; + /** Leases live when time jumped past the whole expiry: every one expires, and that is one error. */ + private jumped = 0; + private jumpedBefore = 0; + private readonly sessions; + + constructor( + readonly name: string, + readonly opened: Awaited>, + readonly random: () => number, + ) { + this.sessions = tappedOpener(opened.sessions, (id, _n, phase) => { + if (id !== name || this.disk !== phase) return; + this.disk = false; + throw new Error('the disk is full'); + }); + } + + pick(items: readonly T[]): T { + return items[Math.floor(this.random() * items.length)] as T; + } + + private host(): Runtime { + return createRuntime({ + sessions: this.sessions, + clock: this.clock, + agents, + transport: serializing(faultyTransport(inProcessTransport(), this.faults, this.clock)), + }); + } + + async start(): Promise { + this.runtime = this.host(); + this.session = startSession({ + name: this.name, + runtime: this.runtime, + assistant, + agents: [product, colleague], + streamFn: scripted(this.cast.script), + }); + this.watch(); + await this.session.messages(); + } + + private watch(): void { + this.events = []; + this.failedBefore = this.cast.failures(); + this.droppedBefore = this.dropped; + this.jumpedBefore = this.jumped; + this.session.subscribe((event) => this.events.push(event)); + } + + /** The run dies and a fresh host resumes the name over the same log. */ + async crash(): Promise { + this.runtime.evict(this.name); + this.epoch += 1; + const activations = await liveLeases(this.opened.sessions, this.name, this.clock.now()); + this.runtime = this.host(); + this.session = await resumeSession(this.name, { + runtime: this.runtime, + streamFn: scripted(this.cast.script), + }); + this.inherited = { activations, exchange: this.session.exchange() !== undefined }; + this.watch(); + } + + /** The storage fails the next write: it never lands, or it lands and the confirmation is lost. */ + failDisk(): string { + this.disk = this.pick(['before', 'after'] as const); + return this.disk; + } + + failWire(): string { + const kind = this.pick(['drop', 'duplicate', 'delay'] as const); + const on = this.pick(OPERATIONS); + const taken = () => { + if (kind === 'drop') this.dropped += 1; + return true; + }; + this.faults.push({ on, kind, match: taken, ...(kind === 'delay' ? { ms: 2_000 } : {}) }); + return `${kind} ${on}`; + } + + /** Time jumps, the way a paused process sees it: a lease live across a jump past its expiry ends. */ + async advance(ms: number): Promise { + if (ms >= this.runtime.wake.expiry) { + this.jumped += await liveLeases(this.opened.sessions, this.name, this.clock.now()); + } + await this.clock.advance(ms); + } + + /** Time moves until nothing is live: every lease expires, every backoff passes, every draft is due. */ + async drain(): Promise { + this.faults.length = 0; + this.disk = false; + // in steps under the expiry, so an activation in flight renews across them + for (let i = 0; i < 14; i += 1) await this.advance(31_000); + await within(this.session.quiet(), 10_000, 'quiet after the drain'); + } + + async check(): Promise { + const errorsAllowed = + this.inherited.activations + + (this.cast.failures() - this.failedBefore) + + (this.dropped - this.droppedBefore) + + (this.jumped - this.jumpedBefore); + await invariants(this.session, this.events, { + allowErrors: errorsAllowed, + sessions: this.opened.sessions, + inherited: this.inherited.activations, + inheritedExchange: this.inherited.exchange, + }); + const rows = await rowsOf(this.opened.sessions, this.name); + const entries = rows.flatMap((row) => { + const type = row.type.slice('ambion/'.length); + if (type === 'message') return [{ type, message: row.data } as never]; + if (type === 'lease') return [{ type, lease: row.data } as never]; + if (type === 'close') return [{ type, close: row.data } as never]; + if (type === 'composition') return [{ type, composition: row.data } as never]; + return []; + }); + const state = foldRoom(entries, RETRY); + expect( + violations(this.history, { record: await this.session.messages(), rows, state }), + ).toEqual([]); + } +} + +/** The promise, or an error naming what did not happen within `ms`. */ +function within(promise: Promise, ms: number, what: string): Promise { + let timer: ReturnType | undefined; + const deadline = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`'${what}' did not finish within ${ms} ms.`)), ms); + }); + return Promise.race([promise, deadline]).finally(() => clearTimeout(timer)); +} + +const yields = () => new Promise((resolve) => setImmediate(resolve)); + +/** A person: visits, asks, reads, leaves, and retries a delivery it never heard back on. */ +class Person { + private visit: { handle: Visit; epoch: number } | undefined; + private deliveries = 0; + + constructor( + private readonly cluster: Cluster, + readonly definition: typeof priya, + ) {} + + get name(): string { + return this.definition.name; + } + + async step(): Promise { + const { cluster } = this; + const op = cluster.pick(['visit', 'deliver', 'deliver', 'read', 'leave'] as const); + if (op === 'visit') return this.arrive(); + if (op === 'read') return this.read(); + if (op === 'leave') return this.leave(); + return this.deliver(); + } + + private current(): Visit | undefined { + if (this.visit === undefined || this.visit.epoch !== this.cluster.epoch) return undefined; + return this.visit.handle; + } + + private async arrive(): Promise { + if (this.current() !== undefined) return; + const { cluster } = this; + const epoch = cluster.epoch; + const handle = await cluster.history.run(this.name, 'visit', undefined, () => + visitSession(cluster.session, this.definition), + ); + if (handle !== undefined) this.visit = { handle, epoch }; + } + + private async deliver(): Promise { + const visit = this.current(); + if (visit === undefined) return this.arrive(); + const key = `${this.name}-${++this.deliveries}`; + const { history } = this.cluster; + const landed = await history.run(this.name, 'deliver', key, () => + visit.deliver({ text: `${key}?`, key }).then(() => true), + ); + // A delivery the person never heard back on is delivered again under the same key. + if (landed === undefined) { + await this.arrive(); + const again = this.current(); + if (again === undefined) return; + await history.run(this.name, 'deliver', key, () => + again.deliver({ text: `${key}?`, key }).then(() => true), + ); + } + } + + private async read(): Promise { + const { cluster } = this; + await cluster.history.run( + this.name, + 'read', + undefined, + () => cluster.session.messages(), + (record) => record.map((m) => ({ seq: m.seq, key: m.key })), + ); + } + + private async leave(): Promise { + const visit = this.current(); + if (visit === undefined) return; + this.visit = undefined; + await this.cluster.history.run(this.name, 'leave', undefined, () => visit.leave()); + } +} + +/** The host: seats and unseats, reads, moves the clock, and is the nemesis. */ +class Host { + constructor(readonly cluster: Cluster) {} + + async step(): Promise { + const { cluster } = this; + const op = cluster.pick([ + 'seat', + 'unseat', + 'read', + 'advance', + 'advance', + 'wire', + 'disk', + 'crash', + ] as const); + if (op === 'seat') { + await cluster.history.run('host', 'seat', undefined, () => cluster.session.seat(colleague)); + } else if (op === 'unseat') { + await cluster.history.run('host', 'unseat', undefined, () => + cluster.session.unseat(colleague), + ); + } else if (op === 'read') { + await cluster.history.run( + 'host', + 'read', + undefined, + () => cluster.session.messages(), + (record) => record.map((m) => ({ seq: m.seq, key: m.key })), + ); + } else if (op === 'advance') { + const ms = Math.floor(cluster.random() * 70_000); + await cluster.history.run('host', `advance ${ms}`, undefined, () => cluster.advance(ms)); + } else if (op === 'wire') { + await cluster.history.run('host', 'wire', undefined, async () => cluster.failWire()); + } else if (op === 'disk') { + await cluster.history.run('host', 'disk', undefined, async () => cluster.failDisk()); + } else { + await cluster.history.run('host', 'crash', undefined, () => cluster.crash()); + } + } +} + +const seeds = Number(process.env.AMBION_SEEDS ?? 25); +const STEPS = 12; + +describe('the room under concurrent clients and a nemesis', () => { + it.each(Array.from({ length: seeds }, (_, i) => i + 1))( + 'keeps every guarantee on seed %i', + async (seed) => { + const opened = await memory.open(); + const cluster = new Cluster(roomName(`consistency-${seed}`), opened, mulberry32(seed)); + try { + await cluster.start(); + const clients = [new Person(cluster, priya), new Person(cluster, sam), new Host(cluster)]; + await within( + Promise.all( + clients.map(async (client) => { + for (let i = 0; i < STEPS; i += 1) { + await client.step(); + await yields(); + } + }), + ), + 20_000, + 'the clients', + ); + await cluster.drain(); + await cluster.check(); + await stopSession(cluster.session); + } catch (error) { + const detail = error instanceof Error ? (error.stack ?? error.message) : String(error); + const rows = await rowsOf(opened.sessions, cluster.name); + const errors = cluster.events.flatMap((e) => + e.type === 'error' ? [`${e.agent}: ${e.error.message}`] : [], + ); + throw new Error( + `seed ${seed} failed:\n${cluster.history.describe()}\nerrors on the last run: ${errors.join('; ')} (inherited ${cluster.inherited.activations})\nrows:\n ${rows + .map((r) => `${r.type.slice(7)} ${JSON.stringify(r.data)}`) + .join('\n ')}\n\n${detail}`, + { cause: error }, + ); + } finally { + await opened.dispose(); + } + }, + 40_000, + ); +}); diff --git a/packages/ambion/test/lease.test.ts b/packages/ambion/test/lease.test.ts index 14c7dd1f..31a484dc 100644 --- a/packages/ambion/test/lease.test.ts +++ b/packages/ambion/test/lease.test.ts @@ -105,9 +105,14 @@ describe('a lease', () => { expect((await session.messages()).filter(isSpoken)).toHaveLength(2); }); - it('expires a lease whose release was lost, and answers the late release stale', async () => { + it('expires a lease whose release was lost twice, and answers the late release stale', async () => { + // a release the seat never heard back on is asked again once, so both are lost + const ended = (l: unknown) => (l as { phase: string }).phase === 'ended'; const { session, clock } = open( - [{ on: 'lease', kind: 'drop', match: (l) => (l as { phase: string }).phase === 'ended' }], + [ + { on: 'lease', kind: 'drop', match: ended }, + { on: 'lease', kind: 'drop', match: ended }, + ], (_c, _a, call) => (call === 1 ? speak('hi') : quiet()), ); const events = collect(session); diff --git a/packages/ambion/test/split.test.ts b/packages/ambion/test/split.test.ts new file mode 100644 index 00000000..412f1a5a --- /dev/null +++ b/packages/ambion/test/split.test.ts @@ -0,0 +1,206 @@ +/** + * The split the design forbids, as a history: two live hosts over one + * log. The first host is paused, in this process by holding its writes + * and in a process of its own with SIGSTOP, a second host resumes the + * name, and the first comes back and keeps writing. Nothing fences the + * first host out yet, and these tests pin what the storage ends up with: + * in memory, a seq on the storage twice and a delivery the first host + * acknowledged that the record lacks; on JSONL, a file Pi refuses to + * load, so no run can open the name again. They turn when a fence lands. + */ +import { spawn } from 'node:child_process'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +import { + createRuntime, + inProcessTransport, + resumeSession, + type Session, + startSession, + stopSession, + visitSession, +} from '../src/index.ts'; +import type { LogEntry } from '../src/log/log.ts'; +import { foldRoom } from '../src/room/fold.ts'; +import { + agents, + assistant, + colleague, + priya, + product, + questions, + sam, + script, + TIMING, +} from './support/cast.ts'; +import { idle } from './support/chaos.ts'; +import { type FakeClock, fakeClock } from './support/clock.ts'; +import { History, violations } from './support/history.ts'; +import { roomName, rowsOf } from './support/room.ts'; +import { scripted } from './support/scripted.ts'; +import { gatedOpener, jsonlSessions, memory } from './support/storage.ts'; +import { serializing } from './support/transport.ts'; + +const RETRY = { attempts: 3, backoff: (attempt: number) => attempt * 30_000 }; + +/** The rows as the fold reads them. */ +function entriesOf(rows: { type: string; data: unknown }[]): LogEntry[] { + return rows.flatMap((row) => { + const type = row.type.slice('ambion/'.length); + if (type === 'message') return [{ type, message: row.data } as LogEntry]; + if (type === 'lease') return [{ type, lease: row.data } as LogEntry]; + if (type === 'close') return [{ type, close: row.data } as LogEntry]; + if (type === 'composition') return [{ type, composition: row.data } as LogEntry]; + return []; + }); +} + +describe('a split: two live hosts over one log', () => { + it('a paused host that comes back writes a seq twice, and its delivery is off the record', async () => { + const opened = await memory.open(); + const clock = fakeClock(); + const history = new History(clock); + // the first host's writes are held while it is paused; the second host's are not + let gate: Promise | undefined; + let release = () => {}; + const first = createRuntime({ + sessions: gatedOpener(opened.sessions, () => gate), + clock, + agents, + transport: serializing(inProcessTransport()), + }); + const second = createRuntime({ + sessions: opened.sessions, + clock, + agents, + transport: serializing(inProcessTransport()), + }); + const name = roomName('split-pause'); + const room = startSession({ + name, + runtime: first, + assistant, + agents: [product, colleague], + streamFn: scripted(script), + }); + const hers = await visitSession(room, priya); + await history.run('priya', 'deliver', 'q1', () => hers.deliver({ text: 'First?', key: 'q1' })); + await room.quiet(); + // paused: a delivery on the first host is in flight and held + gate = new Promise((resolve) => { + release = resolve; + }); + const held = history.run('priya', 'deliver', 'q2', () => + hers.deliver({ text: 'Second?', key: 'q2' }), + ); + // the second host takes the name and serves a question + const taken = await resumeSession(name, { runtime: second, streamFn: scripted(script) }); + const his = await visitSession(taken, sam); + await history.run('sam', 'deliver', 'q3', () => his.deliver({ text: 'Third?', key: 'q3' })); + await taken.quiet(); + // the first host comes back: its held write lands + release(); + await held; + await room.quiet(); + await history.run( + 'sam', + 'read', + undefined, + () => taken.messages(), + (record) => record.map((m) => ({ seq: m.seq, key: m.key })), + ); + try { + const rows = await rowsOf(opened.sessions, name); + const found = violations(history, { + record: await taken.messages(), + rows, + state: foldRoom(entriesOf(rows), RETRY), + }); + expect(found).toContainEqual(expect.stringMatching(/^seq \d+ is on the storage 2 times$/)); + expect(found).toContainEqual('delivery q2 acknowledged, on the record 0 times'); + } finally { + await stopSession(room); + await stopSession(taken); + await opened.dispose(); + } + }); + + const child = fileURLToPath(new URL('./support/child.ts', import.meta.url)); + + /** Run the child until its log takes `at` appends, then stop it where it stands. */ + function stopAt(dir: string, name: string, at: number) { + const args = ['--experimental-transform-types', '--no-warnings', child, dir, name, '40']; + const process_ = spawn(process.execPath, args, { stdio: ['ignore', 'pipe', 'inherit'] }); + const exited = new Promise((resolve) => process_.on('exit', () => resolve())); + const stopped = new Promise((resolve, reject) => { + let last = 0; + let buffer = ''; + process_.stdout.on('data', (chunk: Buffer) => { + buffer += chunk.toString(); + const lines = buffer.split('\n'); + buffer = lines.pop() ?? ''; + for (const line of lines) { + const reported = /^write (\d+)$/.exec(line); + if (reported) last = Number(reported[1]); + if (last >= at) { + process_.kill('SIGSTOP'); + resolve(last); + } + } + }); + process_.on('error', reject); + process_.on('exit', () => reject(new Error('the child ended before the stop'))); + }); + return { + stopped, + continue: () => process_.kill('SIGCONT'), + kill: () => process_.kill('SIGKILL'), + exited, + }; + } + + async function quietNow(session: Session, clock: FakeClock): Promise { + for (let round = 0; round < 12; round += 1) { + const settled = await Promise.race([ + session.quiet().then(() => true), + new Promise((resolve) => setTimeout(() => resolve(false), 300)), + ]); + if (settled && idle(session)) return; + await clock.advance(2_000); + } + throw new Error('the room never went quiet'); + } + + it('a process stopped mid-activation and continued after a takeover leaves a file no run can open', async () => { + const dir = await mkdtemp(join(tmpdir(), 'ambion-split-')); + const name = 'split'; + const paused = stopAt(dir, name, 6); + try { + await paused.stopped; + const sessions = jsonlSessions(dir); + const clock = fakeClock(Date.now()); + const runtime = createRuntime({ sessions, agents, clock, ...TIMING }); + const session = await resumeSession(name, { runtime, streamFn: scripted(script) }); + await quietNow(session, clock); + const [, second] = questions; + if (second === undefined) throw new Error('cast'); + const his = await visitSession(session, sam); + await his.deliver({ text: second.text, key: second.key }); + await quietNow(session, clock); + // the stopped process continues where it stood, and its writes land beside the + // second host's: Pi's JSONL storage refuses the file from then on, so no run can + // ever open the name again + paused.continue(); + await Promise.race([paused.exited, new Promise((resolve) => setTimeout(resolve, 3_000))]); + paused.kill(); + await expect(rowsOf(jsonlSessions(dir), name)).rejects.toThrow(/non-consecutive seq/); + await stopSession(session); + } finally { + paused.kill(); + await rm(dir, { recursive: true, force: true, maxRetries: 5, retryDelay: 20 }); + } + }, 60_000); +}); diff --git a/packages/ambion/test/support/history.ts b/packages/ambion/test/support/history.ts new file mode 100644 index 00000000..3fb1a7cc --- /dev/null +++ b/packages/ambion/test/support/history.ts @@ -0,0 +1,243 @@ +/** + * A history of what the clients of a room asked for and what they heard + * back, and the checks that hold it to the record. Every host action is + * one invocation, then one of three outcomes: `ok`, the action landed; + * `fail`, the room refused it and nothing landed; `info`, the client never + * learned, because the storage, the wire or the process failed under it. + * + * The checks are the guarantees `docs/durability.md` states, read off the + * history and the log together: an acknowledged delivery is on the record + * once, a delivery in doubt is on it at most once, a refused one never, + * every read is a prefix of the record, a client's reads move forward and + * hold what it delivered, every seq on the storage is one message, one + * attempt at a wake runs at a time, and nothing is pending once the room + * drains. + */ +import type { Clock, LeaseRow, Message, Seq } from '../../src/index.ts'; +import type { RoomState } from '../../src/room/fold.ts'; +import { activationId, parseId } from '../../src/room/lease.ts'; + +export type Outcome = 'ok' | 'fail' | 'info'; + +export interface Entry { + index: number; + client: string; + op: string; + /** The delivery this entry is about, for a `deliver`. */ + key?: string; + phase: 'invoke' | Outcome; + /** The invocation this outcome answers, for an outcome. */ + of?: number; + at: number; + /** What an `ok` read saw: the seqs and keys of the record, in order. */ + seen?: { seq: Seq; key: string | undefined }[]; + error?: string; +} + +/** The room said no, and nothing landed. Anything else is an outcome the client cannot tell. */ +const DEFINITIVE = + /visit has ended|is stopped|not in this session|one name names one participant|is not seated|already running|has no composition|not in the runtime's catalog|already in this session/; + +export class History { + readonly entries: Entry[] = []; + + constructor(private readonly clock: Clock) {} + + /** One host action, recorded: its invocation, then what the client heard back. */ + async run( + client: string, + op: string, + key: string | undefined, + action: () => Promise, + seen?: (value: T) => Entry['seen'], + ): Promise { + const invoke = this.push({ + client, + op, + phase: 'invoke', + ...(key === undefined ? {} : { key }), + }); + try { + const value = await action(); + this.push({ + client, + op, + phase: 'ok', + of: invoke.index, + ...(key === undefined ? {} : { key }), + ...(seen === undefined ? {} : { seen: seen(value) }), + ...(typeof value === 'string' ? { error: value } : {}), + }); + return value; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + this.push({ + client, + op, + phase: DEFINITIVE.test(message) ? 'fail' : 'info', + of: invoke.index, + ...(key === undefined ? {} : { key }), + error: message, + }); + return undefined; + } + } + + private push(entry: Omit): Entry { + const full: Entry = { ...entry, index: this.entries.length, at: this.clock.now() }; + this.entries.push(full); + return full; + } + + /** The history, one line per entry, for a failure message. */ + describe(): string { + return this.entries + .map((e) => { + const key = e.key === undefined ? '' : ` ${e.key}`; + const seen = e.seen === undefined ? '' : ` [${e.seen.map((s) => s.seq).join(' ')}]`; + const error = e.error === undefined ? '' : ` (${e.error})`; + const at = new Date(e.at).toISOString().slice(11, 23); + return `#${e.index} ${at} ${e.client} ${e.op}${key} ${e.phase}${seen}${error}`; + }) + .join('\n'); + } +} + +export interface Checked { + /** The record the room holds at the end. */ + record: readonly Message[]; + /** Every row on the storage, in append order. */ + rows: readonly { type: string; data: unknown }[]; + /** The fold at the end, after the drain. */ + state: RoomState; +} + +/** Every guarantee the history and the record break, as one line each. Empty when all hold. */ +export function violations(history: History, checked: Checked): string[] { + const found: string[] = []; + found.push(...deliveries(history, checked.record)); + found.push(...reads(history, checked.record)); + found.push(...seqs(checked.rows)); + found.push(...exclusion(checked.rows)); + found.push(...drained(checked.state)); + return found; +} + +/** An acknowledged delivery is on the record once; one in doubt at most once; a refused one never. */ +function deliveries(history: History, record: readonly Message[]): string[] { + const found: string[] = []; + const outcomes = new Map>(); + for (const entry of history.entries) { + if (entry.op !== 'deliver' || entry.phase === 'invoke' || entry.key === undefined) continue; + outcomes.set(entry.key, new Set([...(outcomes.get(entry.key) ?? []), entry.phase])); + } + for (const [key, seen] of outcomes) { + const landed = record.filter((m) => m.key === key).length; + if (seen.has('ok') && landed !== 1) + found.push(`delivery ${key} acknowledged, on the record ${landed} times`); + if (!seen.has('ok') && seen.has('info') && landed > 1) + found.push(`delivery ${key} in doubt, on the record ${landed} times`); + if (!seen.has('ok') && !seen.has('info') && landed !== 0) + found.push(`delivery ${key} refused, on the record ${landed} times`); + } + return found; +} + +/** Every read is a prefix of the record, a client's reads move forward, and they hold what it delivered. */ +function reads(history: History, record: readonly Message[]): string[] { + const found: string[] = []; + const last = new Map(); + const delivered = new Map(); + for (const entry of history.entries) { + if (entry.phase !== 'ok') continue; + if (entry.op === 'deliver' && entry.key !== undefined) { + const own = delivered.get(entry.client) ?? []; + delivered.set(entry.client, [...own, { key: entry.key, index: entry.index }]); + } + if (entry.op !== 'read' || entry.seen === undefined) continue; + found.push(...oneRead(entry, entry.seen, record, last.get(entry.client) ?? 0, delivered)); + last.set(entry.client, entry.seen.at(-1)?.seq ?? 0); + } + return found; +} + +/** What one read breaks: the prefix, the client's forward motion, or the client's own deliveries. */ +function oneRead( + entry: Entry, + seen: NonNullable, + record: readonly Message[], + lastSeen: number, + delivered: ReadonlyMap, +): string[] { + const found: string[] = []; + const who = `read #${entry.index} by ${entry.client}`; + const cut = prefixBreak(seen, record); + if (cut !== undefined) found.push(`${who} is not a prefix of the record at position ${cut}`); + if ((seen.at(-1)?.seq ?? 0) < lastSeen) found.push(`${who} moved backwards`); + const invoked = entry.of ?? entry.index; + for (const own of delivered.get(entry.client) ?? []) { + if (own.index < invoked && !seen.some((s) => s.key === own.key)) { + found.push(`${who} lacks its own delivery ${own.key}`); + } + } + return found; +} + +/** The first position where what a read saw differs from the record, or nothing for a prefix. */ +function prefixBreak( + seen: NonNullable, + record: readonly Message[], +): number | undefined { + for (const [i, item] of seen.entries()) { + const actual = record[i]; + if (actual?.seq !== item.seq || actual.key !== item.key) return i; + } + return undefined; +} + +/** Every seq on the storage names one message. */ +function seqs(rows: Checked['rows']): string[] { + const seen = new Map(); + for (const row of rows) { + if (row.type !== 'ambion/message') continue; + const seq = (row.data as { seq: number }).seq; + seen.set(seq, (seen.get(seq) ?? 0) + 1); + } + return [...seen] + .filter(([, n]) => n > 1) + .map(([seq, n]) => `seq ${seq} is on the storage ${n} times`); +} + +/** One attempt at a wake runs at a time: the next claims only after the last ended. */ +function exclusion(rows: Checked['rows']): string[] { + const found: string[] = []; + const running = new Map(); + for (const row of rows) { + if (row.type !== 'ambion/lease') continue; + const lease = row.data as LeaseRow; + const parsed = parseId(lease.id); + if (parsed?.kind !== 'wake') continue; + const wake = activationId(parsed.seq, parsed.seat); + const held = running.get(wake); + if (lease.phase === 'running') { + if (held !== undefined && held !== lease.id) + found.push(`${lease.id} claimed while ${held} ran`); + running.set(wake, lease.id); + } else if (held === lease.id) { + running.delete(wake); + } + } + return found; +} + +/** Once the room drained, nothing runs, nothing is pending, and nothing is owed. */ +function drained(state: RoomState): string[] { + const found: string[] = []; + for (const lease of state.leases.values()) { + if (lease.phase === 'running') found.push(`${lease.id} still runs after the drain`); + } + for (const wake of state.pending) found.push(`${wake.id} still pending after the drain`); + for (const owed of state.owed) + found.push(`a summary for ${owed.person} still owed after the drain`); + return found; +} diff --git a/packages/ambion/test/support/invariants.ts b/packages/ambion/test/support/invariants.ts index abdfa77f..9b97fceb 100644 --- a/packages/ambion/test/support/invariants.ts +++ b/packages/ambion/test/support/invariants.ts @@ -44,7 +44,9 @@ export async function invariants( // Every author is a name the room seated, admitted, or was composed with. const names = new Set(session.seats().map((seat) => seat.name)); for (const message of messages) { - if (message.kind === 'arrived' || message.kind === 'seated') names.add(message.from); + if (message.kind === 'arrived' || message.kind === 'seated' || message.kind === 'unseated') { + names.add(message.from); + } } for (const message of messages) { expect(names).toContain(message.from); diff --git a/planning/backlog.md b/planning/backlog.md index 8abe2782..87ffe3be 100644 --- a/planning/backlog.md +++ b/planning/backlog.md @@ -624,3 +624,39 @@ chaos sweep holds every answer to exactly once again, and `restart.test.ts` pins the seat woken again on the next run. Item 29 stands for the cap: at three attempts the wake is dropped, and nothing says so. + +### 35. Two live hosts over one log corrupt it + +**What.** A host that is paused past its leases, and a second host that +resumes the name while it is paused, both write from their own last seq +once the first comes back. In memory, a seq is on the storage twice and +a delivery the first host acknowledged is off the record the second host +reads. On JSONL, Pi's storage refuses to load the file, and no run can +open the name again. `split.test.ts` pins both; +[`docs/durability.md`](../docs/durability.md) §5 states it. + +**Where.** `RoomLog` in `log/log.ts` assigns seqs from its own cache; +nothing on the storage refuses a second writer. + +**Fix.** A run epoch. Every run writes its composition row first, and the +row is the fence: the fold voids every later entry from an earlier run, +and a run that reads past its cursor and finds a later composition +evicts itself. A superseded host loses the writes it acknowledged after +the fence; a storage with a conditional append (SQLite, a Durable +Object) refuses them instead, and loses nothing. Pi's JSONL storage +needs the Pi seq to stay consecutive, so the fence there is the +conditional append or nothing. + +### 36. Every host in the tests shares one clock + +**What.** A resumed host whose clock runs ahead of the last run's expires +the leases it inherited early; one that runs behind holds them past their +time. No test moves two clocks apart, and no test tears a JSONL tail or +loses an `fsync`. + +**Where.** `test/consistency.test.ts`, `test/split.test.ts`; +[`docs/durability.md`](../docs/durability.md) §5. + +**Fix.** A clock per host in the history harness, with a drift the +nemesis picks, and a storage fault that truncates the last line of the +file before a resume. From f0b19ef7f4905dd9bd8995534781a06f7beac3d8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 21:01:24 +0000 Subject: [PATCH 2/2] Hold every run to its error bound, check reads against every acknowledged delivery, and stop the child once An independent review found three defects in the harness. The error bound covered the run that held the room at the end. A run that died took its errors with it. Every run is now checked against its own bound when it dies, and the bound counts only the dropped room calls: a dropped wake is sent again and fails nothing. A read was checked against the client's own deliveries. It is now checked against every delivery acknowledged before the read was asked. One attempt at a time now covers drafts as well as wakes. The child under SIGSTOP was stopped again on every line it printed after the continue. It is stopped once. The docs say what the tier covers: the clients take turns at every await, and the nemesis acts between two actions. A claim asked twice starts one activation and lands a renewal row. The retry policy is stated as the runtime holds it. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01SJjjR3S6iwqEbDXsRDJgiq --- docs/agent.md | 2 +- docs/durability.md | 96 +++++++++++++----------- docs/toolchain.md | 4 +- packages/ambion/src/seat/seat.ts | 4 +- packages/ambion/test/consistency.test.ts | 32 ++++++-- packages/ambion/test/split.test.ts | 35 +++++---- packages/ambion/test/support/history.ts | 39 +++++----- 7 files changed, 125 insertions(+), 87 deletions(-) diff --git a/docs/agent.md b/docs/agent.md index b5ca17c6..11bce853 100644 --- a/docs/agent.md +++ b/docs/agent.md @@ -325,7 +325,7 @@ backoff. A wake to a seat at rest is on the message, so a wake lost on the way is sent again after the resend window, and the seat side runs a wake sent twice once. A claim or a release the seat never heard back on is asked again once: a claim of an id the room already runs is a -renewal, so the row lands once whichever call reached it. When a pass ends, +renewal, and a release of a lease that ended is answered stale. When a pass ends, the activation renews its lease, and the renewal says how far the record reaches. An activation that heard less than that reads the room again through a fresh view. diff --git a/docs/durability.md b/docs/durability.md index 890dd795..fa49439c 100644 --- a/docs/durability.md +++ b/docs/durability.md @@ -2,7 +2,7 @@ What the room promises about its record when the process, the storage, the wire or a model fails, and how the promise is proved. This page is -the contract the chaos tier and the history checker hold the room to. +the contract. The chaos tier and the history checker hold the room to it. Read it with [`agent.md`](agent.md) §5, which names the mechanisms, and [`toolchain.md`](toolchain.md) §8, which says how the tiers run. @@ -12,28 +12,29 @@ Read it with [`agent.md`](agent.md) §5, which names the mechanisms, and log in a Pi session. Every message takes the next seq, and every row the room writes beside the messages carries `after`, the last seq when it landed. The fold reads the log from the start and rebuilds the room from -it: the roster, the people, the open exchange, every lease, every wake -still pending and every summary still owed. Nothing the room holds in -memory outlives what the log says. +it. That is the roster, the people, the open exchange, every lease, every +wake still pending and every summary still owed. Nothing the room holds +in memory outlives what the log says. **Durable means the storage's append resolved.** Pi's in-memory repository holds the record for the life of the process. Pi's JSONL repository writes every entry to a file and calls no `fsync`. A storage that lies about an append breaks every promise below. -**One run per name.** One process holds a name at a time. A second live -run over the same log is the one fault the room does not survive yet: -§5 says what happens. +**One run per name, per runtime.** `startSession` and `resumeSession` +refuse a name the runtime already runs. Nothing refuses a second runtime, +in this process or another. A second live run over the same log is the +one fault the room does not survive yet: §5 says what happens. ## 2. What a delivery promises **Acknowledged: on the record once.** `deliver()` resolves once the -write is confirmed. The message is on the record, on the stream, and the -seats it reaches are woken for it. It stays on the record for the life -of the log. +write is confirmed. The message is on the record and on the stream, and +the wake is sent to every seat it reaches. It stays on the record for +the life of the log. **Refused: nowhere.** A delivery the room refuses rejects `deliver()` -before anything lands: the visit is over, the room is stopped, or the +before anything lands. The visit is over, the room is stopped, or the recipient is not in the room. Nothing is on the record, nothing is on the stream, and nobody woke. @@ -41,7 +42,7 @@ the stream, and nobody woke. died with the write in flight, leaves the host without an answer. The message is on the record or it is not. The room reads the storage back before its next write, so a message that landed is on the record before -anything lands on top of it. `deliver({ key })` names the delivery: a +anything lands on top of it. `deliver({ key })` names the delivery. A host that never learned whether a delivery landed delivers it again under the same key, and the key lands once. @@ -69,27 +70,29 @@ seat claims a lease under the wake's id and renews it while it works. The next attempt claims only after the last one ended. **A lease answers what it heard.** The log says which messages a lease -heard: the ones it was at work for, and the ones its view held because -it was claimed after them. A lease answers them while it runs and once -it stood down, through the seq its last renewal confirmed. +heard. They are the ones it was at work for, and the ones its view held +because it was claimed after them. A lease answers them while it runs +and once it stood down, through the seq its last renewal confirmed. **A lease that came to nothing answers nothing.** A lease that expired or failed leaves every message it heard pending again, whatever it said. Its words stay on the record, and the seat reads them at the next attempt. The failure is one attempt. The room wakes the seat again after the backoff, and at the cap it stops. `runtime.retry` holds the policy: -three attempts, thirty seconds apart, by default. +three attempts by default, thirty seconds after the first failure and +sixty after the second. -**A claim or a release asked twice lands once.** A claim of an id the -room already runs is a renewal. The seat asks again once when it never -heard back, and a release lost twice leaves the room to expire the lease -on its side. +**A claim asked twice starts one activation.** A claim of an id the room +already runs is a renewal, and lands as a renewal row. The seat asks +again once when it never heard back. A release asked twice ends the +lease once: the second call is answered stale. A release lost twice +leaves the room to expire the lease on its side. ## 5. What the room does not promise -**Two live hosts over one log.** A host that is paused, and a second host -that resumes the name while it is paused, both write from their own last -seq once the first comes back. In memory, a seq is on the storage twice +**Two live hosts over one log.** A host is paused, and a second host +resumes the name while it is paused. Once the first comes back, both +write from their own last seq. In memory, a seq is on the storage twice, and a delivery the first host acknowledged is off the record the second host reads. On JSONL, Pi refuses to load the file, and no run can open the name again. `split.test.ts` pins both. A fence is deferred: @@ -113,8 +116,8 @@ by its instructions. - Retry a delivery it never heard back on under the same key. - Resume a name after the process that ran it died, with - `resumeSession(name, { runtime })`, and let the first reconcile expire - what the dead run held. + `resumeSession(name, { runtime })`. The first reconcile expires what + the dead run held. - Run one host per name. Evict a room with `runtime.evict(name)` before another host takes it, and never continue a host that was paused past its leases. @@ -122,8 +125,10 @@ by its instructions. ## 7. How it is proved -The scripted tier runs every proof in process, on a fake clock and a -scripted model, so a run is deterministic and a sweep is exhaustive. +The scripted tier runs on a fake clock and a scripted model, so a run is +deterministic and a sweep is exhaustive. Two proofs leave the process: +the kill from outside and the `SIGSTOP` half of the split run a child on +the system clock. | Proof | Test | What it holds | | ---------------------- | --------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | @@ -131,13 +136,13 @@ scripted model, so a run is deterministic and a sweep is exhaustive. | A handover under load | `hosts.test.ts` | The same sweep with a model that fails and a seat whose say wakes a peer: the second host wakes the failed seat again, and every answer lands once | | A kill from outside | `chaos.test.ts` | The scenario in a child process on JSONL, killed with `SIGKILL` at a write, resumed over the directory | | The random walk | `property.test.ts` | Twenty seeded steps of visits, deliveries, seat changes, clock jumps, wire faults, disk faults and crashes; the invariants hold | -| The history | `consistency.test.ts` | Two people and the host act at once under a nemesis; every action is an invocation and an outcome; §2 to §4 are checked against the record | +| The history | `consistency.test.ts` | Two people and the host take turns under a nemesis; every action is an invocation and an outcome; §2 to §4 are checked against the record | | The split | `split.test.ts` | A paused host comes back after a takeover, in process and as a process under `SIGSTOP`; what §5 says happens, happens | **The history checker** lives in [`test/support/history.ts`](../packages/ambion/test/support/history.ts). -Every host action is recorded as `invoke`, then `ok`, `fail` or `info`: -`ok` landed, `fail` was refused with nothing landed, and `info` is an +Every host action is recorded as `invoke`, then `ok`, `fail` or `info`. +`ok` landed. `fail` was refused, with nothing landed. `info` is an outcome the client cannot tell, because the storage, the wire or the process failed under it. The checker reads the history, the record and the storage together and reports every guarantee that broke: @@ -145,19 +150,26 @@ the storage together and reports every guarantee that broke: - an acknowledged delivery is on the record once, one in doubt at most once, a refused one never; - every read is a prefix of the record, a client's reads move forward, - and they hold what the client delivered; + and every read holds every delivery acknowledged before it was asked; - every seq on the storage names one message; -- one attempt at a wake runs at a time; +- one attempt at a wake or a draft runs at a time; - once the room drains, nothing runs, nothing is pending, nothing is owed. -**The nemesis** crashes the run and resumes it in a fresh runtime, fails -the next write before or after it lands, drops, repeats and delays -requests on the wire, and jumps the clock the way a paused process sees -it. Every error the room reports under it is bounded: the leases a -resumed run inherited, the failures the cast injects, the requests the -nemesis dropped, and the leases live across a jump past the expiry. - -`pnpm test` runs 25 seeds of the history and the walk; `pnpm chaos` -runs 200 of each, the sweep on JSONL too, the handover at every write, -and the kill at every third write. +**The clients take turns.** The people and the host interleave at every +await, and the nemesis acts between two actions. A crash does not cut an +action in flight yet, so a delivery in doubt comes from the storage +alone in this tier. The sweeps in `chaos.test.ts` and `hosts.test.ts` +are where a crash lands inside an append. + +**The nemesis** crashes the run and resumes it in a fresh runtime. It +fails the next write before or after it lands. It drops, repeats and +delays requests on the wire. It jumps the clock the way a paused process +sees it. Every run is held to a bound on the errors it reports, checked +when the run dies and at the end. The bound is the leases the run +inherited, the failures the cast injects, the room calls the nemesis +dropped, and the leases live across a jump past the expiry. + +`pnpm test` runs 25 seeds of the history and the walk. `pnpm chaos` runs +200 of each, the sweep on JSONL too, the handover at every write, and +the kill at every third write. diff --git a/docs/toolchain.md b/docs/toolchain.md index 47a8672a..52323865 100644 --- a/docs/toolchain.md +++ b/docs/toolchain.md @@ -352,8 +352,8 @@ harness in - **The random walk** (`property.test.ts`) loses and repeats requests on the wire, fails a write before or after it lands, and crashes the room up to three times. -- **The history** (`consistency.test.ts`). Two people and the host act - at once against whichever run holds the room, under a nemesis that +- **The history** (`consistency.test.ts`). Two people and the host take + turns against whichever run holds the room, under a nemesis that crashes the run, fails the storage, faults the wire and jumps the clock. Every action is recorded as an invocation and an outcome, and the checker in `test/support/history.ts` holds the history to the diff --git a/packages/ambion/src/seat/seat.ts b/packages/ambion/src/seat/seat.ts index 38b32b65..f6715fc6 100644 --- a/packages/ambion/src/seat/seat.ts +++ b/packages/ambion/src/seat/seat.ts @@ -175,8 +175,8 @@ export class SeatActor implements SeatPort { /** * The lease, or nothing: the room refused it, or the claim never came * back twice. A claim the seat never heard back on is asked again once: - * a claim of an id the room already runs is a renewal, so the row lands - * once whichever call reached it. + * a claim of an id the room already runs is a renewal, so one activation + * starts whichever call reached the room first. */ private async claim(id: string): Promise<{ expiry: number } | undefined> { for (let attempt = 0; attempt < 2; attempt += 1) { diff --git a/packages/ambion/test/consistency.test.ts b/packages/ambion/test/consistency.test.ts index e1b1d677..4b746cdf 100644 --- a/packages/ambion/test/consistency.test.ts +++ b/packages/ambion/test/consistency.test.ts @@ -60,7 +60,7 @@ class Cluster { session!: Session; private disk: FailMode = false; private failedBefore = 0; - /** Requests the nemesis dropped, counted when taken: every one may fail an activation, and that is one error. */ + /** Room calls the nemesis dropped, counted when taken: every one fails an activation, and that is one error. A dropped wake is sent again and fails nothing. */ private dropped = 0; private droppedBefore = 0; /** Leases live when time jumped past the whole expiry: every one expires, and that is one error. */ @@ -114,8 +114,29 @@ class Cluster { this.session.subscribe((event) => this.events.push(event)); } + /** The errors a run may carry: what it inherited, the cast's failures, the drops and the jumps it saw. */ + private allowance(): number { + return ( + this.inherited.activations + + (this.cast.failures() - this.failedBefore) + + (this.dropped - this.droppedBefore) + + (this.jumped - this.jumpedBefore) + ); + } + + /** Every run is held to its own bound: a run that dies is checked before the next one starts. */ + private bounded(): void { + const errors = this.events.flatMap((e) => + e.type === 'error' ? [`${e.agent}: ${e.error.message}`] : [], + ); + expect(errors.length, `errors on a run: ${errors.join('; ')}`).toBeLessThanOrEqual( + this.allowance(), + ); + } + /** The run dies and a fresh host resumes the name over the same log. */ async crash(): Promise { + this.bounded(); this.runtime.evict(this.name); this.epoch += 1; const activations = await liveLeases(this.opened.sessions, this.name, this.clock.now()); @@ -138,7 +159,7 @@ class Cluster { const kind = this.pick(['drop', 'duplicate', 'delay'] as const); const on = this.pick(OPERATIONS); const taken = () => { - if (kind === 'drop') this.dropped += 1; + if (kind === 'drop' && on !== 'wake') this.dropped += 1; return true; }; this.faults.push({ on, kind, match: taken, ...(kind === 'delay' ? { ms: 2_000 } : {}) }); @@ -163,13 +184,8 @@ class Cluster { } async check(): Promise { - const errorsAllowed = - this.inherited.activations + - (this.cast.failures() - this.failedBefore) + - (this.dropped - this.droppedBefore) + - (this.jumped - this.jumpedBefore); await invariants(this.session, this.events, { - allowErrors: errorsAllowed, + allowErrors: this.allowance(), sessions: this.opened.sessions, inherited: this.inherited.activations, inheritedExchange: this.inherited.exchange, diff --git a/packages/ambion/test/split.test.ts b/packages/ambion/test/split.test.ts index 412f1a5a..c7cb5e0b 100644 --- a/packages/ambion/test/split.test.ts +++ b/packages/ambion/test/split.test.ts @@ -130,26 +130,33 @@ describe('a split: two live hosts over one log', () => { const child = fileURLToPath(new URL('./support/child.ts', import.meta.url)); + /** Every `write N` line the child prints, as it prints it. */ + function writes(stdout: NodeJS.ReadableStream, report: (last: number) => void): void { + let buffer = ''; + stdout.on('data', (chunk: Buffer) => { + buffer += chunk.toString(); + const lines = buffer.split('\n'); + buffer = lines.pop() ?? ''; + for (const line of lines) { + const reported = /^write (\d+)$/.exec(line); + if (reported) report(Number(reported[1])); + } + }); + } + /** Run the child until its log takes `at` appends, then stop it where it stands. */ function stopAt(dir: string, name: string, at: number) { const args = ['--experimental-transform-types', '--no-warnings', child, dir, name, '40']; const process_ = spawn(process.execPath, args, { stdio: ['ignore', 'pipe', 'inherit'] }); const exited = new Promise((resolve) => process_.on('exit', () => resolve())); const stopped = new Promise((resolve, reject) => { - let last = 0; - let buffer = ''; - process_.stdout.on('data', (chunk: Buffer) => { - buffer += chunk.toString(); - const lines = buffer.split('\n'); - buffer = lines.pop() ?? ''; - for (const line of lines) { - const reported = /^write (\d+)$/.exec(line); - if (reported) last = Number(reported[1]); - if (last >= at) { - process_.kill('SIGSTOP'); - resolve(last); - } - } + let sent = false; + // stopped once, where it stands; what it writes after the continue is its own + writes(process_.stdout, (last) => { + if (last < at || sent) return; + sent = true; + process_.kill('SIGSTOP'); + resolve(last); }); process_.on('error', reject); process_.on('exit', () => reject(new Error('the child ended before the stop'))); diff --git a/packages/ambion/test/support/history.ts b/packages/ambion/test/support/history.ts index 3fb1a7cc..28ce7268 100644 --- a/packages/ambion/test/support/history.ts +++ b/packages/ambion/test/support/history.ts @@ -9,9 +9,9 @@ * history and the log together: an acknowledged delivery is on the record * once, a delivery in doubt is on it at most once, a refused one never, * every read is a prefix of the record, a client's reads move forward and - * hold what it delivered, every seq on the storage is one message, one - * attempt at a wake runs at a time, and nothing is pending once the room - * drains. + * hold every delivery acknowledged before them, every seq on the storage + * is one message, one attempt at a wake or a draft runs at a time, and + * nothing is pending once the room drains. */ import type { Clock, LeaseRow, Message, Seq } from '../../src/index.ts'; import type { RoomState } from '../../src/room/fold.ts'; @@ -143,31 +143,33 @@ function deliveries(history: History, record: readonly Message[]): string[] { return found; } -/** Every read is a prefix of the record, a client's reads move forward, and they hold what it delivered. */ +/** + * Every read is a prefix of the record, a client's reads move forward, and + * every read holds every delivery acknowledged before the read was asked. + */ function reads(history: History, record: readonly Message[]): string[] { const found: string[] = []; const last = new Map(); - const delivered = new Map(); + const acknowledged: { key: string; index: number }[] = []; for (const entry of history.entries) { if (entry.phase !== 'ok') continue; if (entry.op === 'deliver' && entry.key !== undefined) { - const own = delivered.get(entry.client) ?? []; - delivered.set(entry.client, [...own, { key: entry.key, index: entry.index }]); + acknowledged.push({ key: entry.key, index: entry.index }); } if (entry.op !== 'read' || entry.seen === undefined) continue; - found.push(...oneRead(entry, entry.seen, record, last.get(entry.client) ?? 0, delivered)); + found.push(...oneRead(entry, entry.seen, record, last.get(entry.client) ?? 0, acknowledged)); last.set(entry.client, entry.seen.at(-1)?.seq ?? 0); } return found; } -/** What one read breaks: the prefix, the client's forward motion, or the client's own deliveries. */ +/** What one read breaks: the prefix, the client's forward motion, or a delivery acknowledged before it. */ function oneRead( entry: Entry, seen: NonNullable, record: readonly Message[], lastSeen: number, - delivered: ReadonlyMap, + acknowledged: readonly { key: string; index: number }[], ): string[] { const found: string[] = []; const who = `read #${entry.index} by ${entry.client}`; @@ -175,9 +177,9 @@ function oneRead( if (cut !== undefined) found.push(`${who} is not a prefix of the record at position ${cut}`); if ((seen.at(-1)?.seq ?? 0) < lastSeen) found.push(`${who} moved backwards`); const invoked = entry.of ?? entry.index; - for (const own of delivered.get(entry.client) ?? []) { + for (const own of acknowledged) { if (own.index < invoked && !seen.some((s) => s.key === own.key)) { - found.push(`${who} lacks its own delivery ${own.key}`); + found.push(`${who} lacks delivery ${own.key}, acknowledged before it was asked`); } } return found; @@ -208,7 +210,7 @@ function seqs(rows: Checked['rows']): string[] { .map(([seq, n]) => `seq ${seq} is on the storage ${n} times`); } -/** One attempt at a wake runs at a time: the next claims only after the last ended. */ +/** One attempt at a wake or a draft runs at a time: the next claims only after the last ended. */ function exclusion(rows: Checked['rows']): string[] { const found: string[] = []; const running = new Map(); @@ -216,15 +218,16 @@ function exclusion(rows: Checked['rows']): string[] { if (row.type !== 'ambion/lease') continue; const lease = row.data as LeaseRow; const parsed = parseId(lease.id); - if (parsed?.kind !== 'wake') continue; - const wake = activationId(parsed.seq, parsed.seat); - const held = running.get(wake); + if (parsed === undefined) continue; + const attempt = + parsed.kind === 'wake' ? activationId(parsed.seq, parsed.seat) : `close:${parsed.through}`; + const held = running.get(attempt); if (lease.phase === 'running') { if (held !== undefined && held !== lease.id) found.push(`${lease.id} claimed while ${held} ran`); - running.set(wake, lease.id); + running.set(attempt, lease.id); } else if (held === lease.id) { - running.delete(wake); + running.delete(attempt); } } return found;