API index and search · Build metadata
Source snapshot
packages/room/src/shell.ts
1// The Node test shell (RINDLE-REALTIME-DESIGN.md §11, P0/P1): a plain process running2// the SAME wasm build the Durable Object shell will run, wired to real sockets.3//4// rindled ──ws (normalized protocol)──▶ WasmRoom (base store) ──serving──▶ subscribers5// upstream leg (§3) downstream leg (§4)6//7// Both legs speak the one protocol. Upstream, the shell is a client of rindled's public8// ws plane: `init` → `subscribe {queryId, leaseToken}` → `nhello` → seq-0 `nbatch` →9// live tail; the lease is minted on rindled's private control plane (`POST /materialize`)10// — the room fetches its own footprint lease at boot, the same shape as the DO shell's11// boot callback (§10.1). Downstream, the shell serves the `@rindle/remote` wire verbatim,12// gated by **self-authorizing signed lease tokens** (§10.1): `subscribe {queryId,13// leaseToken}` verifies the token (signature, doc, expiry, revocation) and materializes14// the approved AST it carries on presentation — no `/materialize` control call, no named15// registry, ASTs never composed by clients. Identical queries share one pipeline16// (QueryKey dedup in the wasm room); each subscriber gets its own epoch/seq envelope.17//18// Lease lifecycle (§4.1): tokens are short-lived; the shell drops any subscription whose19// lease passes `exp` unrenewed (a `queryError`, then detach — renewal is20// re-authorization through the API server, which is a fresh subscribe with a fresh21// token). The private control plane's `POST /revoke {userId}` terminates a user's22// subscriptions and sockets immediately and refuses pre-revocation tokens (`iat` <23// revocation) — the synchronous layer; the TTL is the backstop.24//25// The write path (§5.1, P2 — no write-behind yet): a connection `init`s its clientID,26// then pushes `{t:"pushMutation", envelope:{clientID, mid, name, args}}` frames. Per27// mutation: dedup/gap-check → run the registered mutator against a staged wasm tx →28// commit to the shared head → fan the data `nbatch`es out IMMEDIATELY (step 2: fanout29// is never gated on durability) → append the envelope to the journal under a ≤~5ms30// group commit → on append, ACK: advance the author's `_rindle_client_mutations` row31// and fan the lmid frame (step 4: the ack a client observes survives a process crash32// by construction — §8.1). A failed/unknown mutator is a silent reject that still33// consumes the mid (the ledger advances with no effects; the author's prediction snaps34// back — `rindle-replica`'s contract). On boot, the journal replays by RE-INVOKING the35// mutators against the fresh base (§3.3), then acks everything replayed.36//37// Release protocol: every downstream `nbatch` is stamped with the room's `head_cv`,38// and after each drain the shell sends the connection-level `{t:"progress",39// frame:{cvMin}}` the optimistic client releases on. One §1.3.1-coherence rule: the40// AUTHOR's cvMin holds below its oldest un-acked mutation's apply-cv until the ack41// lands (data released before its lmid advance would make the client re-invoke a42// confirmed mutation on top of its own effect); everyone else releases at head_cv.43//44// Backpressure (§9, T8): each downstream socket has a bounded send budget. A stalled45// client that overflows it is TERMINATED with a gap — the normalized protocol's46// re-subscribe repair — so one slow client never holds frames in room memory.47//48// Failure posture (§3.4): the room is an INCARNATION. Any upstream violation — a frame49// that fails to decode, a seq gap, a poisoned apply, an upstream `queryError`, the50// socket dropping — kills the incarnation: the wasm store is freed, every downstream51// socket is closed (subscriptions die with the incarnation; reconnecting clients52// re-subscribe against the next one), and the shell re-subscribes upstream. On the same53// socket that is a `subscribe` re-send (rindled bumps the epoch and re-snapshots); on a54// fresh socket it is a new subscription. There is no patch-up path, by design.5556import { createServer, type Server } from "node:http";57import { randomUUID } from "node:crypto";5859import { WebSocket, WebSocketServer, type RawData } from "ws";6061import { initRoomWasm, WasmRoom } from "./wasm.ts";62import { verifyRoomToken, RoomTokenError, scopeSpecsHash } from "./token.ts";63import {64 assertSyncMutatorReturn,65 isEnvironmentShortfall,66 mutationTx,67 type RoomMutator,68 type TableShape,69} from "./mutation-tx.ts";70import {71 journalEntryOutcome,72 memoryJournal,73 type RoomJournal,74 type RoomJournalEntry,75} from "./journal.ts";76import type { RoomAuthority } from "./authority.ts";777879/** The upstream half of the shell's config: where rindled lives and what to follow. */80export interface UpstreamOptions {81 /** rindled's public subscription plane, e.g. `ws://127.0.0.1:7601`. */82 wsUrl: string;83 /** rindled's private control plane, e.g. `http://127.0.0.1:7600` (lease minting). */84 controlUrl: string;85 /** Bearer token for the control plane (required unless rindled runs unauthenticated). */86 authToken?: string;87 /** The document footprint — a wire `Ast` (§3.1). What this room follows and serves from. */88 footprintAst: unknown;89 /** Lease TTL passed to `/materialize` (rindled's default when omitted). */90 leaseTtlMs?: number;91 /** The `init` clientID on the upstream socket (diagnostic identity). */92 clientId?: string;93}9495/** The downstream half: how clients are authorized and served (§4/§10.1). */96export interface DownstreamOptions {97 /** This room's document id — lease tokens for any other doc are refused. */98 docId: string;99 /** Token key ring: `kid` → shared secret (the API server signs with the same ring). */100 tokenKeys: Record<string, string>;101 /** Idle grace before an unsubscribed query's pipeline is reclaimed (default 30s). */102 idleTtlMs?: number;103 /** Lease-expiry / idle-sweep cadence (default 1s). */104 sweepIntervalMs?: number;105 /** How long a revocation keeps refusing pre-revocation tokens (default 30min — set106 * it ≥ the longest token TTL the API server mints). */107 revocationWindowMs?: number;108 /** The room's private control plane (`POST /revoke`, `GET /stats`). Omit to run109 * without one (no revocation surface). */110 control?: { authToken: string; port?: number };111 /** The write plane (§5.1). Omit to run read-only (writes are refused). */112 writes?: WritesOptions;113 /** Per-socket downstream send budget in bytes (§9; default 4 MiB). A socket whose114 * queued bytes would exceed it is terminated with a gap — re-subscribing is the115 * repair. */116 sendBudgetBytes?: number;117}118119/** One table's §3.3 scope spec (H-iv-b): an element of the api-server's120 * `RoomBootResponse.scopes`, passed VERBATIM to the wasm room's `enableWritesV2`.121 * Structurally identical to `@rindle/api-server`'s `RoomScopeSpec` (declared locally —122 * the shell must not depend on the api-server package; TS structural typing keeps the123 * two in lockstep at the host's threading site). */124export interface RoomScopeSpec {125 table: string;126 /** The footprint's row-local predicate for this table (a wire `Condition`) — drives127 * the commit gate's absent-read proof. Absent ⇒ absent reads on this table always128 * deopt (fail closed). */129 footprintWhere?: unknown;130 writable:131 | { kind: "none" }132 | { kind: "predicate"; where?: unknown; joinKeyCols: string[] };133}134135/** The §5.1 write plane: the room's own mutator registry over its owned tables. */136export interface WritesOptions {137 /** The server registry (§4.2): named mutators run against the shared head. PLAIN138 * synchronous `(tx, args, ctx)` functions only — a `shared(...)` GENERATOR registry139 * does NOT register verbatim (nothing drives it here; the shell rejects a mutator140 * that returns a generator/promise — see `assertSyncMutatorReturn`). */141 mutators: Record<string, RoomMutator>;142 /** §3.2's owned set — the only tables mutators may write. Must all be in the143 * upstream footprint; followed tables and the ledger are never writable. */144 ownedTables: string[];145 /** The §3.3 per-table scope specs from the boot wire (`RoomBootResponse.scopes` —146 * H-iv-b). Present ⇒ the write plane enables in v2 GATED mode (`enableWritesV2`):147 * staged writes validate against the writable predicates, join-key edits refuse,148 * absent reads must prove against `footprintWhere`, context (`kind:"none"`) tables149 * become txGet-READABLE, and a violating commit returns a structured DEOPT instead150 * of applying. The scopes' writable tables must be a SUBSET of {@link ownedTables}151 * (the host's own declaration) — a wider server scope throws at construction, so a152 * self-hoster's owned set can never be extended from the wire. Absent ⇒ the v1153 * table-granular write plane, byte-identical to before. */154 scopes?: RoomScopeSpec[];155 /** The durable sidecar an ack means (§8.1). Defaults to `memoryJournal()` — the156 * "survives nothing beyond the process" class; hosts bring their own. */157 journal?: RoomJournal;158 /** The journal group-commit window (default 5ms): mutations arriving within it159 * share one append, and their acks ride one ledger commit. */160 groupCommitMs?: number;161 /** The write authority (§5.3.1) — the API server's `/apply-row-change-txn` host162 * (or the P3 gate's mock). Omit to run journal-only (P2 semantics: nothing is163 * ever durable upstream). With an authority, the shell claims a placement epoch164 * at boot, probes durable lmids before replay, and write-behinds on the flush165 * cadence (§5.3). */166 authority?: RoomAuthority;167 /** The flush debounce (§5.3; default 250ms, within the design's ≤1s budget). */168 flushDebounceMs?: number;169 /** Flush immediately once this many keys are dirty (default 512). */170 flushDirtyMax?: number;171}172173export interface RoomShellOptions {174 upstream: UpstreamOptions;175 downstream: DownstreamOptions;176 /** Downstream ws port (default 0 = ephemeral; bound on 127.0.0.1). */177 port?: number;178 /** Diagnostic sink (default: silent). */179 log?: (line: string) => void;180}181182export interface RoomShell {183 /** The bound downstream port. */184 readonly port: number;185 /** The bound control-plane port (0 when no control plane was configured). */186 readonly controlPort: number;187 /** Resolves when the CURRENT incarnation is live (seq-0 snapshot applied);188 * immediately if it already is. */189 awaitLive(): Promise<void>;190 /** The last-applied upstream commit version, if live. */191 cv(): number | undefined;192 /** The upstream subscription epoch of the current incarnation, if any. */193 upstreamEpoch(): number | undefined;194 /** This incarnation's downstream bootId (rotates on every re-subscribe). */195 bootId(): string;196 /** Fire the write-behind flush now (instead of the debounce) and await its197 * settlement — deterministic flushing for tests and drain-before-close. No-op198 * without an authority. */199 flushNow(): Promise<void>;200 close(): Promise<void>;201}202203const UPSTREAM_QID = 1;204/** Delay before re-subscribing after a violation — keeps a persistent violation from205 * becoming a hot loop while staying far below human-perceptible recovery time. */206const RESUBSCRIBE_DELAY_MS = 250;207const RECONNECT_MAX_MS = 5_000;208const DEFAULT_IDLE_TTL_MS = 30_000;209const DEFAULT_SWEEP_INTERVAL_MS = 1_000;210const DEFAULT_REVOCATION_WINDOW_MS = 30 * 60_000;211const DEFAULT_GROUP_COMMIT_MS = 5;212const DEFAULT_SEND_BUDGET_BYTES = 4 * 1024 * 1024;213const DEFAULT_FLUSH_DEBOUNCE_MS = 250;214const DEFAULT_FLUSH_DIRTY_MAX = 512;215/** §4.2 drain-before-downgrade iteration cap: a healthy room quiesces in a handful of flushes; a216 * runaway (a straggler push per flush, or a wedged CAS loop) fails LOUD rather than spinning. */217const DRAIN_MAX_ITERATIONS = 100;218/** Flush-retry backoff bounds (network-class failures only; same journaled bytes). */219const FLUSH_RETRY_MIN_MS = 100;220const FLUSH_RETRY_MAX_MS = 2_000;221/** The reserved lmid system query — subscribed BY NAME even in lease mode; identity222 * comes from the connection's `init`, never from args. */223const LMID_QUERY_NAME = "_rindle/clientLmid";224225/** Mint the room's upstream footprint lease on rindled's control plane (§4: the room226 * fetches its own lease — on the DO shell this is the boot callback's job). */227async function mintLease(up: UpstreamOptions): Promise<string> {228 const res = await fetch(new URL("/materialize", up.controlUrl), {229 method: "POST",230 headers: {231 "content-type": "application/json",232 ...(up.authToken ? { authorization: `Bearer ${up.authToken}` } : {}),233 },234 body: JSON.stringify({235 ast: up.footprintAst,236 ...(up.leaseTtlMs !== undefined ? { leaseTtlMs: up.leaseTtlMs } : {}),237 }),238 });239 if (!res.ok) {240 throw new Error(`upstream /materialize failed: ${res.status} ${await res.text()}`);241 }242 const out = (await res.json()) as { leaseToken?: string };243 if (typeof out.leaseToken !== "string") {244 throw new Error("upstream /materialize returned no leaseToken");245 }246 return out.leaseToken;247}248249function send(ws: WebSocket, frame: unknown): void {250 if (ws.readyState === WebSocket.OPEN) {251 ws.send(JSON.stringify(frame));252 }253}254255/** One downstream subscription's routing + lease state (keyed by its wasm subKey). */256interface SubMeta {257 ws: WebSocket;258 conn: ConnState;259 clientQid: number;260 /** The token's subject — the §4.1 revocation key. */261 user: string;262 /** The token's expiry — enforced by the sweep (drop at `exp` unrenewed). */263 exp: number;264}265266interface ConnQuery {267 subKey: string;268 epoch: number;269}270271interface ConnState {272 id: number;273 ws: WebSocket;274 queries: Map<number, ConnQuery>;275 /** Serializes message handling per connection (token verify is async; a client's276 * subscribe/unsubscribe order must hold). */277 busy: Promise<void>;278 /** The `init` identity — what the lmid subscribe and pushMutation key on. An279 * idempotency key, NOT an identity (it is client-supplied). */280 clientID: string | null;281 /** The connection's AUTHENTICATED subject (managed-writes §3.1): the `sub` of the282 * first verified lease token presented on this connection. Shell-stamped, one283 * principal per connection, and the gate `pushMutation` requires. */284 sub: string | null;285}286287/** One mutation applied to the head but not yet acked: the author's release point288 * holds below `applyCv` until the journal append lands (§1.3.1 coherence — see the289 * module docs' release-protocol note). */290interface UnackedMutation {291 mid: number;292 applyCv: number;293}294295/** A recorded NON-APPLIED verdict (H-iv-b): what the `mutationOutcome` frame carries,296 * and what a re-sent mid whose `beginMutation` dedups is re-answered with. `name`/297 * `args` are kept for DEOPT verdicts only — the frame must be self-contained (a298 * client that already retired the entry re-invokes from the frame). */299interface RecordedOutcome {300 kind: "deopt" | "rejected";301 reason?: string;302 name?: string;303 args?: unknown;304}305306/** Per-client retention cap for {@link RecordedOutcome}s. The map only holds307 * NON-APPLIED mids, and a client only re-sends a mid while its ledger lmid trails it308 * — a contiguously-advancing window bounded by the client's in-flight backlog, far309 * below this cap. Past it the oldest records evict (insertion order): a re-send of an310 * evicted mid degrades to today's silence — the mid still dedups and the ledger still311 * covers it, only the outcome re-answer is lost. */312const MAX_RECORDED_OUTCOMES_PER_CLIENT = 512;313314class Shell implements RoomShell {315 private readonly opts: RoomShellOptions;316 private readonly log: (line: string) => void;317 private readonly wss: WebSocketServer;318 private control: Server | null = null;319320 private room: WasmRoom | null = null;321 private live = false;322 private incarnationBootId = randomUUID();323 private liveWaiters: Array<() => void> = [];324325 private upstream: WebSocket | null = null;326 private leaseToken = "";327 private subscribeInFlight = false;328 private reconnectDelayMs = RESUBSCRIBE_DELAY_MS;329 private closed = false;330 private nextConnId = 1;331 private sweepTimer: ReturnType<typeof setInterval> | null = null;332333 /** subKey (what the wasm room routes by) → where its frames go + lease state. */334 private readonly subs = new Map<string, SubMeta>();335 /** Every open downstream connection — revocation must reach a BOUND conn even when336 * it holds no live subscription (it is still write-capable). */337 private readonly conns = new Set<ConnState>();338 /** userId → when they were revoked (refuses tokens with `iat` ≤ this; pruned after339 * the revocation window). */340 private readonly revoked = new Map<string, number>();341 /** Downstream subscribes queued while no incarnation is live. */342 private pendingSubs: Array<{ ws: WebSocket; conn: ConnState; msg: SubscribeMsg }> = [];343344 // ------------------------------ write-plane state ------------------------------345 /** The {@link scopeSpecsHash} of the boot-wire scopes THIS shell armed its §3.3 gate346 * with — `undefined` in v1 (ungated) mode. Compared against each lease token's347 * `scopesHash` to flag scope skew (a profile edited under this live room), which348 * otherwise manifests only as an undiagnosable deopt loop. */349 private readonly armedScopesHash: string | undefined;350 /** Scope-skew hash pairs already logged (`lease→armed`), so the diagnostic fires ONCE351 * per distinct skew, not once per subscribe. */352 private readonly loggedScopeSkew = new Set<string>();353 /** The journal (write plane only). One per shell — it outlives incarnations; that354 * is the point (§3.3: pending is replayed from it on every re-subscribe). */355 private journal: RoomJournal | null = null;356 /** Positional table shapes from the upstream hello (keyed layer of MutationTx). */357 private tableShapes = new Map<string, TableShape>();358 /** Mutations queued while no incarnation is live (drained after pendingSubs). */359 private pendingMutes: Array<{ ws: WebSocket; conn: ConnState; envelope: PushEnvelope }> = [];360 /** clientID → its applied-but-unacked mutations, oldest first (release holdback). */361 private readonly unacked = new Map<string, UnackedMutation[]>();362 /** clientID → its recorded NON-APPLIED outcomes (mid → verdict), insertion-ordered363 * and capped per client ({@link MAX_RECORDED_OUTCOMES_PER_CLIENT}). Seeded from the364 * journal replay each incarnation (`finishBoot` — so it reflects what THIS365 * incarnation's replay produced), appended live, and cleared with the incarnation. */366 private readonly outcomes = new Map<string, Map<number, RecordedOutcome>>();367 /** The group-commit window: entries awaiting the next journal append. */368 private ackQueue: RoomJournalEntry[] = [];369 private ackTimer: ReturnType<typeof setTimeout> | null = null;370 /** Serializes journal appends (one in flight; acks apply in append order). */371 private ackChain: Promise<void> = Promise.resolve();372373 // ------------------------------ flush state (§5.3) ------------------------------374 /** The placement epoch (§2.5), claimed once per shell process at start. 0 = no375 * authority configured. */376 private placementEpoch = 0;377 /** The next flush-stream seq — also the (zero-padded) wire offset. Seeded from the378 * journal so offset strings stay monotone across restarts sharing one journal. */379 private flushSeq = 1;380 private flushTimer: ReturnType<typeof setTimeout> | null = null;381 /** One flush settlement in flight at a time (the wasm room guards too). */382 private flushBusy = false;383 /** The settlement chain `flushNow()` awaits. */384 private flushChain: Promise<void> = Promise.resolve();385 private flushesConfirmed = 0;386 /** The last flush seq that COMMITTED at the authority (§4.2/§5.4 `flush_ok`): the value the387 * `/drain` control reports as `finalFlushSeq`, the fence a downgraded client's ghost waits on388 * (`_rindle_room_watermark(doc) ≥ finalFlushSeq`, and the watermark row's `flush_seq` IS this389 * offset — consumer.rs). Seeded from the journal at boot (a re-booted room that already flushed390 * reports its journaled max, not 0); 0 for a never-flushed room. */391 private lastCommittedFlushSeq = 0;392 /** Fenced at the authority (§2.5): this room is superseded — terminal. */393 private moved = false;394395 constructor(opts: RoomShellOptions) {396 this.opts = opts;397 this.log = opts.log ?? (() => {});398 if (opts.downstream.writes) {399 if (opts.downstream.writes.ownedTables.length === 0) {400 throw new Error("writes.ownedTables must name at least one table");401 }402 // H-iv-b: the boot-wire scopes may only ever NARROW the host's owned set, never403 // extend it — a wider server scope would let mutators write tables this404 // deployment never declared writable (self-hoster semantics preserved). Loud at405 // construction, exactly like the empty-owned check above.406 const scopes = opts.downstream.writes.scopes;407 if (scopes !== undefined) {408 const owned = new Set(opts.downstream.writes.ownedTables);409 const rogue = scopes410 .filter((s) => s.writable.kind !== "none" && !owned.has(s.table))411 .map((s) => s.table);412 if (rogue.length > 0) {413 throw new Error(414 `writes.scopes marks ${rogue.map((t) => `\`${t}\``).join(", ")} writable, but the ` +415 `host's writes.ownedTables does not include ${rogue.length === 1 ? "it" : "them"} — ` +416 `the boot-wire scopes may only ever narrow the host's owned set, never extend it`,417 );418 }419 }420 // The armed-scope fingerprint: what the gate enforces, hashed once, for the421 // skew check on the token path. `undefined` scopes = v1 ungated → no check.422 if (scopes !== undefined) this.armedScopesHash = scopeSpecsHash(scopes);423 this.journal = opts.downstream.writes.journal ?? memoryJournal();424 }425 this.wss = new WebSocketServer({ port: opts.port ?? 0, host: "127.0.0.1" });426 this.wss.on("connection", (ws) => this.serveDownstream(ws));427 }428429 async start(): Promise<void> {430 await initRoomWasm();431 await new Promise<void>((resolve) => {432 if (this.wss.address() !== null) return resolve();433 this.wss.on("listening", () => resolve());434 });435 if (this.opts.downstream.control) {436 await this.startControl(this.opts.downstream.control);437 }438 const interval = this.opts.downstream.sweepIntervalMs ?? DEFAULT_SWEEP_INTERVAL_MS;439 this.sweepTimer = setInterval(() => this.sweep(), interval);440 this.sweepTimer.unref?.();441 const authority = this.opts.downstream.writes?.authority;442 if (authority && this.journal) {443 // Seed the flush stream past everything this journal ever numbered, then claim444 // the placement epoch (§2.5) — once per process, BEFORE anything else touches445 // the authority: from this moment every stale body a dead predecessor left446 // mid-network is fenced, which is what closes the probe race for good.447 // Unconfirmed batches settle per incarnation in finishBoot — a prior process's448 // records are fenced there (routine: their mutations recover by envelope449 // replay under OUR epoch), our own land or dedup.450 const { maxSeq } = await this.journal.replayFlushes();451 this.flushSeq = maxSeq + 1;452 this.lastCommittedFlushSeq = maxSeq; // journal-seeded (self-corrects upward on each flush_ok)453 this.placementEpoch = await authority.claimEpoch(this.opts.downstream.docId);454 this.log(`placement epoch ${this.placementEpoch}`);455 }456 this.leaseToken = await mintLease(this.opts.upstream);457 this.connectUpstream();458 }459460 get port(): number {461 const addr = this.wss.address();462 return typeof addr === "object" && addr !== null ? addr.port : 0;463 }464465 get controlPort(): number {466 const addr = this.control?.address();467 return typeof addr === "object" && addr !== null ? addr.port : 0;468 }469470 awaitLive(): Promise<void> {471 if (this.live) return Promise.resolve();472 return new Promise((resolve) => this.liveWaiters.push(resolve));473 }474475 cv(): number | undefined {476 return this.room?.cv();477 }478479 upstreamEpoch(): number | undefined {480 return this.room?.epoch();481 }482483 bootId(): string {484 return this.incarnationBootId;485 }486487 async close(): Promise<void> {488 this.closed = true;489 if (this.sweepTimer) clearInterval(this.sweepTimer);490 if (this.ackTimer !== null) {491 clearTimeout(this.ackTimer);492 this.ackTimer = null;493 }494 if (this.flushTimer !== null) {495 clearTimeout(this.flushTimer);496 this.flushTimer = null;497 }498 this.upstream?.close();499 for (const client of this.wss.clients) client.close();500 await new Promise<void>((resolve) => this.wss.close(() => resolve()));501 if (this.control) {502 await new Promise<void>((resolve) => this.control?.close(() => resolve()));503 }504 if (this.room) {505 this.room.free();506 this.room = null;507 }508 }509510 // ------------------------------ upstream leg ------------------------------511512 private connectUpstream(): void {513 if (this.closed) return;514 const ws = new WebSocket(this.opts.upstream.wsUrl);515 this.upstream = ws;516 ws.on("open", () => {517 this.reconnectDelayMs = RESUBSCRIBE_DELAY_MS;518 send(ws, {519 t: "init",520 clientID: this.opts.upstream.clientId ?? `room-shell-${this.incarnationBootId}`,521 });522 this.sendSubscribe();523 });524 ws.on("message", (data) => this.onUpstreamFrame(data));525 ws.on("error", (err) => this.log(`upstream socket error: ${String(err)}`));526 ws.on("close", () => {527 if (this.closed) return;528 // A subscribe may have died with the socket — the fresh connection must be free529 // to send its own.530 this.subscribeInFlight = false;531 this.incarnationDead("upstream socket closed");532 const delay = this.reconnectDelayMs;533 this.reconnectDelayMs = Math.min(delay * 2, RECONNECT_MAX_MS);534 this.log(`upstream reconnect in ${delay}ms`);535 setTimeout(() => this.connectUpstream(), delay).unref?.();536 });537 }538539 /** (Re-)send the upstream subscribe. On an already-open socket rindled treats a540 * re-send for the same queryId as gap recovery: tears down the old attachment, bumps541 * the epoch, re-snapshots. */542 private sendSubscribe(): void {543 if (this.closed || this.subscribeInFlight) return;544 if (this.upstream?.readyState !== WebSocket.OPEN) return; // reconnect path re-enters545 this.subscribeInFlight = true;546 send(this.upstream, { t: "subscribe", queryId: UPSTREAM_QID, leaseToken: this.leaseToken });547 }548549 private onUpstreamFrame(data: RawData): void {550 let frame: Record<string, unknown>;551 try {552 frame = JSON.parse(String(data)) as Record<string, unknown>;553 } catch {554 // Not even JSON: the transport is garbage — violation posture.555 this.violation("upstream frame is not JSON");556 return;557 }558 if (frame.queryId !== undefined && frame.queryId !== UPSTREAM_QID) return;559 switch (frame.t) {560 case "nhello": {561 this.subscribeInFlight = false;562 // A hello supersedes any prior incarnation (e.g. rindled restarted and563 // re-served us): the old store is gone either way.564 if (this.room) this.incarnationDead("superseded by a new upstream hello");565 try {566 this.room = WasmRoom.open(567 JSON.stringify(frame.hello),568 this.opts.downstream.idleTtlMs ?? DEFAULT_IDLE_TTL_MS,569 );570 this.tableShapes = shapesOf(frame.hello);571 this.log(`upstream hello: epoch ${this.room.epoch()}`);572 } catch (e) {573 this.violation(`upstream hello rejected: ${String(e)}`);574 }575 break;576 }577 case "nbatch": {578 if (!this.room) return; // pre-hello or post-death stragglers: drop579 let status: { applied: string; rows?: number; ops?: number };580 try {581 status = JSON.parse(this.room.apply(JSON.stringify(frame.batch))) as typeof status;582 } catch (e) {583 this.violation(`upstream apply failed: ${String(e)}`);584 return;585 }586 if (status.applied === "snapshot") {587 this.log(`upstream snapshot: ${status.rows} rows @ cv ${this.room.cv()}`);588 void this.finishBoot(this.room);589 } else if (status.applied === "live") {590 this.drainDownstream();591 } else if (status.applied === "staleEpoch") {592 this.log("dropped stale-epoch upstream frame");593 }594 break;595 }596 case "progress":597 break; // optimistic release signal — the room's own downstream owns its cvMin598 case "queryError": {599 // The lease died (expiry, dematerialize, daemon restart): re-mint, re-subscribe.600 this.subscribeInFlight = false;601 this.incarnationDead(`upstream queryError: ${String(frame.message)}`);602 void mintLease(this.opts.upstream)603 .then((token) => {604 this.leaseToken = token;605 setTimeout(() => this.sendSubscribe(), RESUBSCRIBE_DELAY_MS).unref?.();606 })607 .catch((e) => {608 this.log(`lease re-mint failed: ${String(e)}`);609 // The socket is still up; retry the whole path after a beat.610 setTimeout(() => {611 if (!this.closed) this.onUpstreamFrame(data);612 }, RECONNECT_MAX_MS).unref?.();613 });614 break;615 }616 default:617 break;618 }619 }620621 /** A §3.4 violation: kill the incarnation and re-subscribe (never patch up). */622 private violation(reason: string): void {623 this.incarnationDead(reason);624 setTimeout(() => this.sendSubscribe(), RESUBSCRIBE_DELAY_MS).unref?.();625 }626627 private incarnationDead(reason: string): void {628 this.log(`incarnation dead: ${reason}`);629 if (this.room) {630 this.room.free();631 this.room = null;632 }633 this.live = false;634 this.incarnationBootId = randomUUID();635 // Downstream subscriptions die with the incarnation (§3.4): close the sockets;636 // reconnecting clients re-subscribe against the next incarnation.637 for (const meta of this.subs.values()) {638 meta.ws.close(1012, "room re-subscribing");639 }640 this.subs.clear();641 this.pendingSubs = [];642 // The write plane's optimism dies with the incarnation too: whatever was not yet643 // journaled was never acked, and never will be (§8.1's `applied` class). The644 // journal itself outlives us — the next incarnation replays it.645 this.pendingMutes = [];646 this.unacked.clear();647 // Recorded outcomes die with the incarnation too — the next incarnation reseeds648 // them from its own journal replay (finishBoot), which is the source of truth for649 // what THAT incarnation's base produced.650 this.outcomes.clear();651 this.ackQueue = [];652 if (this.ackTimer !== null) {653 clearTimeout(this.ackTimer);654 this.ackTimer = null;655 }656 // The flush debounce dies with the incarnation; an in-flight settlement keeps657 // running (its batch is journaled and epoch-bound — the bootId guard keeps its658 // outcome from touching the next incarnation's state).659 if (this.flushTimer !== null) {660 clearTimeout(this.flushTimer);661 this.flushTimer = null;662 }663 }664665 /** After the seq-0 snapshot: enable the write plane, settle any unconfirmed flush666 * batches, probe the authority's durable lmids, and replay the journal onto the667 * fresh base (§3.3 — pending re-invoked; a mid the probe covered replays as dedup,668 * its effects already in the snapshot; everything replayed acked, since journal669 * presence IS the ack), then start serving. The order is load-bearing: the670 * resubmission SETTLES before the probe, so a batch a dead incarnation left671 * mid-network can never land between the probe and our first flush. No subscriber672 * exists yet (they died with the previous incarnation), so replay fans out to673 * nobody — the snapshots the queued subscribes get below already include the674 * replayed rows. */675 private async finishBoot(room: WasmRoom): Promise<void> {676 const writes = this.opts.downstream.writes;677 if (writes && this.journal) {678 let entries: RoomJournalEntry[];679 try {680 // H-iv-b: boot-wire scopes arm the §3.3 commit gate (v2); without them the v1681 // table-granular plane enables byte-identically to before. Either enable682 // throws loudly on a malformed input and leaves writes disabled — the683 // violation below keeps that from becoming a half-enabled room.684 if (writes.scopes !== undefined) {685 room.enableWritesV2(JSON.stringify(writes.scopes));686 } else {687 room.enableWrites(JSON.stringify(writes.ownedTables));688 }689 entries = await this.journal.replay();690 } catch (e) {691 this.violation(`write-plane boot failed: ${String(e)}`);692 return;693 }694 if (this.room !== room) return; // an incarnation death raced the replay695 if (writes.authority) {696 try {697 await this.settleUnconfirmedFlushes(writes.authority);698 } catch (e) {699 if (this.moved || this.closed) return;700 this.violation(`flush resubmission failed: ${String(e)}`);701 return;702 }703 if (this.room !== room) return;704 const clients = [...new Set(entries.map((e) => e.clientID))];705 if (clients.length > 0) {706 let lmids: Record<string, number>;707 try {708 lmids = await withNetRetry(709 () => writes.authority!.lmids(this.opts.downstream.docId, clients),710 () => this.closed || this.room !== room,711 this.log,712 );713 } catch (e) {714 if (this.room !== room || this.closed) return;715 this.violation(`durable-lmid probe failed: ${String(e)}`);716 return;717 }718 if (this.room !== room) return;719 const seeds = Object.entries(lmids)720 .filter(([, lmid]) => lmid > 0)721 .map(([clientID, lmid]) => ({ clientID, lmid }));722 if (seeds.length > 0) {723 try {724 room.seedDurable(JSON.stringify(seeds));725 room.commitAll(); // drop the (subscriber-less) ledger fanout726 } catch (e) {727 this.violation(`durable seed failed: ${String(e)}`);728 return;729 }730 this.log(`durable seed: ${seeds.map((s) => `${s.clientID}→${s.lmid}`).join(", ")}`);731 }732 }733 }734 try {735 for (const entry of entries) {736 const res = this.runMutation(room, entry);737 // Seed the recorded-outcome map with what THIS incarnation's replay738 // produced. That includes the H-iv-a replay gotcha: an entry journaled739 // APPLIED can legitimately DEOPT (or reject) re-invoked against the moved740 // base — the §3.3 rebase class. The journal record is never rewritten741 // (history stays what the acking incarnation observed); the map is this742 // incarnation's answer sheet for re-sent mids.743 if (res.outcome === "rejected" || res.outcome === "deopt") {744 this.recordOutcome(entry.clientID, entry.mid, {745 kind: res.outcome,746 ...(res.reason !== undefined ? { reason: res.reason } : {}),747 ...(res.outcome === "deopt" ? { name: entry.name, args: entry.args } : {}),748 });749 if (journalEntryOutcome(entry) === "applied") {750 this.log(751 `replayed APPLIED mutation ${entry.clientID}:${entry.mid} (\`${entry.name}\`) ` +752 `now ${res.outcome}s against the moved base — mid stays burnt, no effects`,753 );754 }755 }756 }757 if (entries.length > 0) {758 room.ack(759 JSON.stringify(entries.map(({ clientID, mid }) => ({ clientID, mid }))),760 );761 room.commitAll(); // drop the (subscriber-less) fanout of the replay762 }763 } catch (e) {764 this.violation(`journal replay failed: ${String(e)}`);765 return;766 }767 if (entries.length > 0) {768 this.log(`journal replay: ${entries.length} mutation(s) re-invoked and acked`);769 }770 }771 this.becomeLive();772 this.scheduleFlush(); // replay may have left dirty entries / unflushed lmids773 }774775 /** Resubmit every unconfirmed journaled flush batch, byte-identically, in seq order776 * (§3.3/§5.3 step 4). Outcomes: committed or deduped → confirm; fenced under an777 * OLD epoch → routine cleanup (a prior process's batch — its mutations recover by778 * replay under OUR epoch); fenced under OUR epoch → another room claimed the doc:779 * terminal (`room_moved`); conflict → drop (the replay + next flush re-derive the780 * net effect against the authority's current rows); identity mismatch → fatal. */781 private async settleUnconfirmedFlushes(authority: RoomAuthority): Promise<void> {782 if (!this.journal) return;783 const { records } = await this.journal.replayFlushes();784 for (const record of records) {785 if (this.closed || this.moved) return;786 let res: Awaited<ReturnType<RoomAuthority["applyRowChangeTxn"]>>;787 try {788 res = await withNetRetry(789 () => authority.applyRowChangeTxn(record.body),790 () => this.closed || this.moved,791 this.log,792 );793 } catch (e) {794 if ((e as { fatal?: boolean })?.fatal !== true) throw e;795 // The §8.3 identity check tripped on a journaled record — a same-id/796 // different-body bug. LOUD, dropped, never a silent dedup and never a797 // violation loop: the batch is dead; the envelope replay re-derives its798 // effects under a fresh flush id.799 this.log(`flush ${record.seq} DROPPED: batch identity mismatch: ${String(e)}`);800 await this.journal.confirmFlush(record.seq);801 continue;802 }803 if (res.kind === "ok") {804 await this.journal.confirmFlush(record.seq);805 this.log(806 `flush ${record.seq} settled at boot: ${res.applied ? "applied" : "already applied"}`,807 );808 } else if (res.kind === "conflict") {809 await this.journal.confirmFlush(record.seq);810 this.log(`flush ${record.seq} dropped at boot: conflict (replay supersedes)`);811 } else {812 await this.journal.confirmFlush(record.seq);813 if (record.epoch === this.placementEpoch) {814 this.roomMoved(`flush ${record.seq} fenced under our epoch`);815 return;816 }817 this.log(`flush ${record.seq} (old epoch ${record.epoch}) fenced: dropped`);818 }819 }820 }821822 private becomeLive(): void {823 this.live = true;824 const waiters = this.liveWaiters;825 this.liveWaiters = [];826 for (const w of waiters) w();827 const queued = this.pendingSubs;828 this.pendingSubs = [];829 for (const { ws, conn, msg } of queued) {830 if (ws.readyState === WebSocket.OPEN) {831 conn.busy = conn.busy.then(() => this.handleSubscribe(ws, conn, msg));832 }833 }834 const mutes = this.pendingMutes;835 this.pendingMutes = [];836 for (const { ws, conn, envelope } of mutes) {837 if (ws.readyState === WebSocket.OPEN) {838 conn.busy = conn.busy.then(() => this.handlePushMutation(ws, conn, envelope));839 }840 }841 }842843 /** After any commit (an upstream apply, a local mutation, an ack): one commitAll844 * drains every query's net delta, fanned out per subscriber envelope (empty → no845 * frame, no seq — every stream stays gap-free over emitted frames), then each846 * touched connection gets its `progress` release point. */847 private drainDownstream(): void {848 if (!this.room) return;849 const frames = JSON.parse(this.room.commitAll()) as Array<{ sub: string; batch: unknown }>;850 const touched = new Set<ConnState>();851 for (const { sub, batch } of frames) {852 const meta = this.subs.get(sub);853 if (meta) {854 this.sendBudgeted(meta.conn, { t: "nbatch", queryId: meta.clientQid, batch });855 touched.add(meta.conn);856 }857 }858 for (const conn of touched) this.sendProgress(conn);859 }860861 /** The connection-level release point (`{t:"progress", frame:{cvMin}}`): `head_cv`,862 * except an author with un-acked mutations holds below its oldest one's apply-cv —863 * its own data must never release ahead of the lmid advance that confirms it864 * (§1.3.1: the client would re-invoke the mutator on top of its own effect). */865 private sendProgress(conn: ConnState): void {866 if (!this.room) return;867 let cvMin = this.room.headCv();868 if (conn.clientID) {869 const held = this.unacked.get(conn.clientID);870 if (held && held.length > 0) {871 cvMin = Math.min(cvMin, held[0].applyCv - 1);872 }873 }874 this.sendBudgeted(conn, { t: "progress", frame: { cvMin } });875 }876877 /** §9's per-socket send budget (T8): a socket whose queued bytes would exceed it is878 * TERMINATED — the close handler reclaims its subscriptions, so room memory stays879 * bounded and the client repairs by re-subscribing (a gap, §8.5). Never buffer880 * unboundedly on behalf of one slow reader. */881 private sendBudgeted(conn: ConnState, frame: unknown): void {882 const ws = conn.ws;883 if (ws.readyState !== WebSocket.OPEN) return;884 const text = JSON.stringify(frame);885 const budget = this.opts.downstream.sendBudgetBytes ?? DEFAULT_SEND_BUDGET_BYTES;886 if (ws.bufferedAmount + text.length > budget) {887 this.log(`send budget exceeded (conn ${conn.id}): closed with a gap`);888 ws.terminate();889 return;890 }891 ws.send(text);892 }893894 // ------------------------------ the write path ------------------------------895896 /** Apply one envelope to the room: dedup → run the mutator against a staged tx →897 * commit (or reject/deopt, consuming the mid). Shared verbatim by the live path and898 * the boot-time journal replay (§3.3 — recovery is re-invocation; a journaled899 * NON-APPLIED entry replays its recorded outcome without running anything). Verdict900 * classification (H-iv-b): the §3.3 commit gate's structured verdict and an901 * environment shortfall (a capability the room lacks — `tx.query`) are DEOPTS (the902 * client re-routes the mutation to the daemon stream); an unknown mutator or an903 * authz/validation throw is a FINAL rejection. Throws only when the room is torn904 * (a mid gap, a poisoned commit) — the caller decides the blast radius. */905 private runMutation(906 room: WasmRoom,907 entry: RoomJournalEntry,908 ):909 | { outcome: "applied"; applyCv: number }910 | { outcome: "rejected"; reason?: string }911 | { outcome: "deopt"; reason?: string }912 | { outcome: "dedup" } {913 // The envelope rides the wasm tx (Slice I-ii): should this mid end non-applied,914 // its durable outcome row echoes `name`/`args` on a DEOPT — the same915 // self-contained-re-invoke rule as the frame (H-iv-b), but readable through the916 // daemon after downgrade when no room socket exists to deliver one.917 const begin = JSON.parse(918 room.beginMutation(919 entry.clientID,920 entry.mid,921 entry.name,922 entry.args === undefined ? undefined : JSON.stringify(entry.args),923 ),924 ) as {925 begin: "tx" | "dedup";926 };927 if (begin.begin === "dedup") return { outcome: "dedup" };928 // A journaled non-applied entry replays its RECORDED verdict without running: the929 // live run already consumed the mid with no effects, and re-judging (a deopt that930 // would now PASS against the moved base) would invent effects for a mutation the931 // client was told to re-route elsewhere — a double apply. Passing the verdict932 // through the reject IS the replay re-seed (I-ii): a recorded-but-never-flushed933 // outcome re-enters the core buffer here, so the next flush carries its row —934 // while a mid the boot probe covered dedups above and never re-seeds (its row935 // co-committed with the flush that made it durable). The journal keeps no936 // `reason`, so a replayed row carries kind + envelope only, like the replayed937 // frame.938 const journaled = journalEntryOutcome(entry);939 if (journaled !== "applied") {940 room.rejectMutation(journaled, undefined);941 return { outcome: journaled };942 }943 const mutator = this.opts.downstream.writes?.mutators[entry.name];944 if (!mutator) {945 const reason = `unknown mutator \`${entry.name}\``;946 room.rejectMutation("rejected", reason);947 this.log(`mutation ${entry.clientID}:${entry.mid} rejected: ${reason}`);948 return { outcome: "rejected", reason };949 }950 try {951 // The ambient auth context (managed-writes §3.2): the journaled subject, so a952 // replayed invocation sees exactly the identity the live one did (`""` for953 // entries journaled before the identity plane — unauthenticated).954 const returned = (mutator as (...a: unknown[]) => unknown)(955 mutationTx(room, this.tableShapes),956 entry.args as never,957 { user: entry.sub ?? "" },958 );959 assertSyncMutatorReturn(returned, entry.name);960 } catch (e) {961 // The shell owns the H-iv-b classification, so it rides the reject into the962 // core's outcome record (the flush carries the durable row) exactly as it rides963 // the frame below.964 const message = String((e as Error)?.message ?? e);965 if (isEnvironmentShortfall(e)) {966 room.rejectMutation("deopt", "environment");967 this.log(968 `mutation \`${entry.name}\` (${entry.clientID}:${entry.mid}) deopted: ` +969 `environment shortfall: ${message}`,970 );971 return { outcome: "deopt", reason: "environment" };972 }973 room.rejectMutation("rejected", message);974 this.log(`mutation \`${entry.name}\` (${entry.clientID}:${entry.mid}) rejected: ${message}`);975 return { outcome: "rejected", reason: message };976 }977 const out = JSON.parse(room.commitMutation()) as {978 headCv?: number;979 deopt?: { reason: string; table: string; pk: unknown[] };980 };981 if (out.deopt !== undefined) {982 // The gate refused at commit and consumed the tx INTERNALLY (mid burnt,983 // watermark advanced, no head commit) — do NOT call rejectMutation here.984 this.log(985 `mutation \`${entry.name}\` (${entry.clientID}:${entry.mid}) deopted: ` +986 `${out.deopt.reason} on \`${out.deopt.table}\` pk ${JSON.stringify(out.deopt.pk)}`,987 );988 return { outcome: "deopt", reason: out.deopt.reason };989 }990 return { outcome: "applied", applyCv: out.headCv as number };991 }992993 /** The live `pushMutation` path (§5.1): apply → fan the data out NOW → journal under994 * the group commit → ack (the ledger advance) once the append resolves. */995 private handlePushMutation(ws: WebSocket, conn: ConnState, envelope: PushEnvelope): void {996 const fail = (message: string) => send(ws, { t: "error", message });997 if (!this.opts.downstream.writes || !this.journal) {998 fail("this room is read-only (no write plane configured)");999 return;1000 }1001 if (!conn.clientID) {1002 fail("init with a clientID before pushMutation");1003 return;1004 }1005 if (envelope.clientID !== conn.clientID) {1006 // The envelope's clientID is bound to the connection identity — a session may1007 // not write another client's mid stream.1008 fail("envelope clientID does not match the connection identity");1009 return;1010 }1011 if (!this.live || !this.room) {1012 this.pendingMutes.push({ ws, conn, envelope });1013 return;1014 }1015 // Checked at execution time (after the live gate), so a mutation queued behind a1016 // still-pending subscribe is judged AFTER that subscribe bound the subject.1017 if (conn.sub === null) {1018 fail("pushMutation requires an authenticated subject — subscribe with a lease token first");1019 return;1020 }1021 const room = this.room;1022 let result: ReturnType<Shell["runMutation"]>;1023 try {1024 result = this.runMutation(room, { ...envelope, sub: conn.sub });1025 } catch (e) {1026 const message = String((e as Error)?.message ?? e);1027 if (room.isPoisoned()) {1028 this.violation(`mutation commit tore the head: ${message}`);1029 } else {1030 // The mid-gap contract: the exact "mutation gap …" text the client's1031 // recovery keys on rides an error frame.1032 fail(message);1033 }1034 return;1035 }1036 if (result.outcome === "dedup") {1037 // Absorbed; the ledger already covers it (or its in-flight ack will). But a1038 // re-sent NON-APPLIED mid is ANSWERED with its recorded outcome (H-iv-b): the1039 // client may have missed the original frame (reconnect), and silence would1040 // leave its deopted mutation parked forever.1041 const recorded = this.outcomes.get(conn.clientID)?.get(envelope.mid);1042 if (recorded !== undefined) this.sendOutcome(conn, envelope.mid, recorded);1043 return;1044 }1045 if (result.outcome === "applied") {1046 const held = this.unacked.get(conn.clientID) ?? [];1047 held.push({ mid: envelope.mid, applyCv: result.applyCv });1048 this.unacked.set(conn.clientID, held);1049 this.drainDownstream(); // §5.1 step 2: never gated on durability1050 } else {1051 // Non-applied (deopt/rejected): record + answer NOW — synchronously, before the1052 // journal enqueue below, so on this ordered socket the `mutationOutcome` frame1053 // always precedes the lmid ack that burns the mid. Applied mutations send1054 // NOTHING (additive frame: old clients drop unknown `t`).1055 const recorded: RecordedOutcome = {1056 kind: result.outcome,1057 ...(result.reason !== undefined ? { reason: result.reason } : {}),1058 ...(result.outcome === "deopt" ? { name: envelope.name, args: envelope.args } : {}),1059 };1060 this.recordOutcome(conn.clientID, envelope.mid, recorded);1061 this.sendOutcome(conn, envelope.mid, recorded);1062 }1063 this.enqueueJournal({1064 clientID: envelope.clientID,1065 mid: envelope.mid,1066 name: envelope.name,1067 args: envelope.args,1068 sub: conn.sub,1069 outcome: result.outcome,1070 // The legacy flag rides alongside BOTH non-applied kinds: a pre-H-iv-b reader1071 // replays either as a consumed-mid-no-effect — exactly right (journal.ts).1072 ...(result.outcome !== "applied" ? { rejected: true } : {}),1073 });1074 // §5.1 step 5: the write-behind rides its own debounce — applied mutations dirty1075 // rows, rejected/deopted ones still advance an lmid the authority must eventually1076 // hold.1077 this.scheduleFlush();1078 }10791080 /** Record a non-applied verdict for `(clientID, mid)` — the re-send answer sheet.1081 * Insertion-ordered per client; past {@link MAX_RECORDED_OUTCOMES_PER_CLIENT} the1082 * oldest evicts (see the constant's retention rationale). */1083 private recordOutcome(clientID: string, mid: number, outcome: RecordedOutcome): void {1084 let byMid = this.outcomes.get(clientID);1085 if (byMid === undefined) {1086 byMid = new Map();1087 this.outcomes.set(clientID, byMid);1088 }1089 byMid.delete(mid); // re-recording refreshes recency1090 byMid.set(mid, outcome);1091 while (byMid.size > MAX_RECORDED_OUTCOMES_PER_CLIENT) {1092 byMid.delete(byMid.keys().next().value as number);1093 }1094 }10951096 /** The `mutationOutcome` frame (H-iv-b): `{t, mid, kind, reason?, name?, args?}` on1097 * the mutating client's socket. DEOPT frames echo `name`/`args` so they are1098 * self-contained — a client that already retired the entry can re-invoke from the1099 * frame alone. Sent before the mid's journal append is even enqueued, so it always1100 * precedes the lmid ack on this ordered socket. */1101 private sendOutcome(conn: ConnState, mid: number, o: RecordedOutcome): void {1102 this.sendBudgeted(conn, {1103 t: "mutationOutcome",1104 mid,1105 kind: o.kind,1106 ...(o.reason !== undefined ? { reason: o.reason } : {}),1107 ...(o.kind === "deopt" ? { name: o.name, args: o.args } : {}),1108 });1109 }11101111 /** Collect entries for the next group commit (§5.1 step 3: one append per window). */1112 private enqueueJournal(entry: RoomJournalEntry): void {1113 this.ackQueue.push(entry);1114 if (this.ackTimer === null) {1115 const window = this.opts.downstream.writes?.groupCommitMs ?? DEFAULT_GROUP_COMMIT_MS;1116 this.ackTimer = setTimeout(() => this.flushJournal(), window);1117 this.ackTimer.unref?.();1118 }1119 }11201121 private flushJournal(): void {1122 this.ackTimer = null;1123 const batch = this.ackQueue;1124 this.ackQueue = [];1125 if (batch.length === 0 || !this.journal) return;1126 const journal = this.journal;1127 const bootAtFlush = this.incarnationBootId;1128 // One append in flight at a time; acks apply in append order.1129 this.ackChain = this.ackChain.then(async () => {1130 try {1131 await journal.append(batch);1132 } catch (e) {1133 // An ack that might not survive must never be sent (§8.1): the incarnation1134 // dies loudly instead.1135 if (this.incarnationBootId === bootAtFlush && !this.closed) {1136 this.violation(`journal append failed: ${String(e)}`);1137 }1138 return;1139 }1140 if (this.incarnationBootId !== bootAtFlush || !this.room) {1141 return; // the incarnation died mid-append: its optimism died with it1142 }1143 this.applyAck(this.room, batch);1144 });1145 }11461147 /** §5.1 step 4 — the ack: advance the ledger rows, ship the lmid frames, and raise1148 * every author's release point past its confirmed data. */1149 private applyAck(room: WasmRoom, batch: RoomJournalEntry[]): void {1150 let headCv: number | null;1151 try {1152 const entries = batch.map(({ clientID, mid }) => ({ clientID, mid }));1153 headCv = (JSON.parse(room.ack(JSON.stringify(entries))) as { headCv: number | null })1154 .headCv;1155 } catch (e) {1156 this.violation(`ack failed: ${String(e)}`);1157 return;1158 }1159 const authors = new Set<string>();1160 for (const { clientID, mid } of batch) {1161 authors.add(clientID);1162 const held = this.unacked.get(clientID);1163 if (held) {1164 const rest = held.filter((u) => u.mid > mid);1165 if (rest.length > 0) this.unacked.set(clientID, rest);1166 else this.unacked.delete(clientID);1167 }1168 }1169 if (headCv !== null) {1170 this.drainDownstream(); // the lmid frames (+ progress to their receivers)1171 }1172 // Raise the release point for every author connection even if no frame reached1173 // it in this drain (e.g. its lmid slice was already current).1174 for (const meta of this.subs.values()) {1175 if (meta.conn.clientID && authors.has(meta.conn.clientID)) {1176 this.sendProgress(meta.conn);1177 authors.delete(meta.conn.clientID);1178 }1179 }1180 }11811182 // --------------------------- the write-behind flush ---------------------------11831184 /** Arm the flush debounce (or fire now past the dirty-size threshold). Cheap and1185 * idempotent — called after every applied/rejected mutation and every settlement. */1186 private scheduleFlush(): void {1187 const writes = this.opts.downstream.writes;1188 if (!writes?.authority || this.flushBusy || this.flushTimer !== null) return;1189 if (this.closed || this.moved || !this.live || !this.room) return;1190 const dirtyMax = writes.flushDirtyMax ?? DEFAULT_FLUSH_DIRTY_MAX;1191 const delay =1192 this.room.dirtyLen() >= dirtyMax ? 0 : writes.flushDebounceMs ?? DEFAULT_FLUSH_DEBOUNCE_MS;1193 this.flushTimer = setTimeout(() => {1194 this.flushTimer = null;1195 this.fireFlush();1196 }, delay);1197 this.flushTimer.unref?.();1198 }11991200 /** Fire the write-behind flush now and await its settlement (tests, drain paths). */1201 flushNow(): Promise<void> {1202 if (this.flushTimer !== null) {1203 clearTimeout(this.flushTimer);1204 this.flushTimer = null;1205 }1206 this.fireFlush();1207 return this.flushChain;1208 }12091210 /** §4.2/§7.4 drain-before-downgrade: fire the write-behind repeatedly until pending AND dirty1211 * are both empty, awaiting each settlement, then report the last COMMITTED flush seq — the1212 * value a downgraded client's frozen ghost fences against1213 * (`_rindle_room_watermark(doc) ≥ finalFlushSeq`). A never-flushed room reports 0. Idempotent:1214 * re-draining an already-quiescent room re-reports the same seq without flushing. It does NOT1215 * refuse later pushes (§4.2 R2: a straggler push after drain flushes at seq+1; the client1216 * ghost's second conjunct — no sent room-domain pending — carries correctness). A room that1217 * fails to quiesce within {@link DRAIN_MAX_ITERATIONS} throws LOUD (never silently). */1218 async drainForDowngrade(): Promise<number> {1219 const writes = this.opts.downstream.writes;1220 let iterations = 0;1221 while (1222 writes?.authority &&1223 this.live &&1224 !this.moved &&1225 !this.closed &&1226 this.room !== null &&1227 (this.room.dirtyLen() > 0 || this.room.pendingLen() > 0)1228 ) {1229 if (++iterations > DRAIN_MAX_ITERATIONS) {1230 throw new Error(1231 `drain: room ${this.opts.downstream.docId} failed to quiesce after ${DRAIN_MAX_ITERATIONS} flushes ` +1232 `(dirty=${this.room.dirtyLen()}, pending=${this.room.pendingLen()})`,1233 );1234 }1235 await this.flushNow();1236 }1237 return this.lastCommittedFlushSeq;1238 }12391240 /** §5.3: build the batch (synchronously — the build IS the snapshot), journal its1241 * exact body bytes, then settle it against the authority. One in flight; the1242 * settlement outcome drives the state machine (§5.4). */1243 private fireFlush(): void {1244 const writes = this.opts.downstream.writes;1245 const authority = writes?.authority;1246 if (!authority || !this.journal || this.flushBusy) return;1247 if (this.closed || this.moved || !this.live || !this.room) return;1248 const room = this.room;1249 let out: string | undefined;1250 try {1251 out = room.beginFlush();1252 } catch (e) {1253 this.violation(`beginFlush failed: ${String(e)}`);1254 return;1255 }1256 if (!out) return; // nothing dirty, nothing unflushed1257 const { changes, batchHash } = JSON.parse(out) as {1258 changes: unknown[];1259 batchHash: string;1260 };1261 const seq = this.flushSeq++;1262 // Composed ONCE; journaled and sent as this exact string forever (§5.3 step 4).1263 const body = JSON.stringify({1264 source: `room:${this.opts.downstream.docId}:${this.placementEpoch}`,1265 offset: padOffset(seq),1266 doc: this.opts.downstream.docId,1267 epoch: this.placementEpoch,1268 batchHash,1269 cas: true,1270 changes,1271 });1272 const journal = this.journal;1273 const bootAtFlush = this.incarnationBootId;1274 this.flushBusy = true;1275 this.flushChain = (async () => {1276 try {1277 await journal.appendFlush({ seq, epoch: this.placementEpoch, body });1278 } catch (e) {1279 // An unjournaled batch must never reach the wire: a retry could rebuild1280 // different bytes under the same id (§8.3). Loud incarnation death.1281 this.flushBusy = false;1282 if (this.incarnationBootId === bootAtFlush && !this.closed) {1283 this.violation(`flush journal append failed: ${String(e)}`);1284 }1285 return;1286 }1287 let res: Awaited<ReturnType<RoomAuthority["applyRowChangeTxn"]>>;1288 try {1289 res = await withNetRetry(1290 () => authority.applyRowChangeTxn(body),1291 () => this.closed || this.moved,1292 this.log,1293 );1294 } catch (e) {1295 this.flushBusy = false;1296 if (this.closed || this.moved) return;1297 // Fatal-class apply error (identity mismatch): our bug, never retried.1298 if (this.incarnationBootId === bootAtFlush) {1299 this.violation(`flush apply failed fatally: ${String(e)}`);1300 }1301 return;1302 }1303 this.flushBusy = false;1304 const sameIncarnation = this.incarnationBootId === bootAtFlush && this.room !== null;1305 if (res.kind === "ok") {1306 this.flushesConfirmed += 1;1307 this.lastCommittedFlushSeq = Math.max(this.lastCommittedFlushSeq, seq); // §4.2 fence input1308 await journal.confirmFlush(seq);1309 if (!sameIncarnation) {1310 // Settled; the new incarnation replayed its own state — let it flush.1311 this.scheduleFlush();1312 return;1313 }1314 try {1315 this.room!.flushOk();1316 } catch (e) {1317 this.violation(`flushOk failed: ${String(e)}`);1318 return;1319 }1320 this.scheduleFlush(); // in-flight re-dirties / new lmids1321 } else if (res.kind === "conflict") {1322 await journal.confirmFlush(seq); // nothing applied; the retry re-derives1323 if (!sameIncarnation) {1324 this.scheduleFlush();1325 return;1326 }1327 try {1328 const out = JSON.parse(this.room!.flushConflict(JSON.stringify(res.conflicts))) as {1329 headCv: number | null;1330 };1331 if (out.headCv !== null) {1332 this.drainDownstream(); // the corrective frames (§5.4: ordinary edits)1333 }1334 } catch (e) {1335 this.violation(`flushConflict failed: ${String(e)}`);1336 return;1337 }1338 this.log(`flush ${seq} CAS-conflicted: converged to the authority, retrying the rest`);1339 this.scheduleFlush();1340 } else {1341 // Fenced under OUR epoch: this room is superseded (§2.5) — terminal.1342 await journal.confirmFlush(seq);1343 this.roomMoved(`flush ${seq} fenced (authority epoch ${res.currentEpoch ?? "?"})`);1344 }1345 })();1346 }13471348 /** §2.5 stale-room behavior: the authority fenced us — another placement owns the1349 * doc. Final state discarded, downstream closed with `room_moved`, nothing1350 * reconnects. Clients re-open through the API server onto the current epoch. */1351 private roomMoved(reason: string): void {1352 if (this.moved) return;1353 this.moved = true;1354 this.log(`room moved: ${reason}`);1355 for (const meta of this.subs.values()) {1356 meta.ws.close(4009, "room_moved");1357 }1358 void this.close();1359 }13601361 // ----------------------------- downstream leg -----------------------------13621363 private serveDownstream(ws: WebSocket): void {1364 const conn: ConnState = {1365 id: this.nextConnId++,1366 ws,1367 queries: new Map(),1368 busy: Promise.resolve(),1369 clientID: null,1370 sub: null,1371 };1372 this.conns.add(conn);1373 ws.on("message", (data) => {1374 let msg: Record<string, unknown>;1375 try {1376 msg = JSON.parse(String(data)) as Record<string, unknown>;1377 } catch {1378 return;1379 }1380 // Serialize per connection: subscribe verification is async, and a client's1381 // subscribe → unsubscribe order must hold. A per-message throw is isolated to1382 // THIS connection (the reference server's #12).1383 conn.busy = conn.busy.then(async () => {1384 try {1385 await this.handleDownstreamMsg(ws, conn, msg);1386 } catch (err) {1387 send(ws, {1388 t: "error",1389 queryId: msg.queryId,1390 message: String((err as Error)?.message ?? err),1391 });1392 }1393 });1394 });1395 ws.on("close", () => {1396 const now = Date.now();1397 for (const q of conn.queries.values()) {1398 if (this.subs.get(q.subKey)?.ws === ws) {1399 this.subs.delete(q.subKey);1400 this.room?.unsubscribe(q.subKey, now);1401 }1402 }1403 conn.queries.clear();1404 this.conns.delete(conn);1405 });1406 }14071408 private async handleDownstreamMsg(1409 ws: WebSocket,1410 conn: ConnState,1411 msg: Record<string, unknown>,1412 ): Promise<void> {1413 switch (msg.t) {1414 case "init": {1415 // The connection identity: what the lmid subscribe and pushMutation key on.1416 // No reply frame — exactly rindled's wire.1417 if (typeof msg.clientID === "string" && msg.clientID.length > 0) {1418 conn.clientID = msg.clientID;1419 }1420 break;1421 }1422 case "subscribe": {1423 if (typeof msg.queryId !== "number") return;1424 if (!this.live || !this.room) {1425 // Not live yet: hold the subscribe until the seq-0 snapshot lands, so a1426 // booting room doesn't refuse its first clients.1427 this.pendingSubs.push({ ws, conn, msg: msg as unknown as SubscribeMsg });1428 return;1429 }1430 await this.handleSubscribe(ws, conn, msg as unknown as SubscribeMsg);1431 break;1432 }1433 case "unsubscribe": {1434 if (typeof msg.queryId !== "number") return;1435 const prev = conn.queries.get(msg.queryId);1436 if (prev) {1437 conn.queries.delete(msg.queryId);1438 this.subs.delete(prev.subKey);1439 this.room?.unsubscribe(prev.subKey, Date.now());1440 }1441 break;1442 }1443 case "pushMutation": {1444 const e = msg.envelope as Partial<PushEnvelope> | undefined;1445 if (1446 e === null ||1447 typeof e !== "object" ||1448 typeof e.clientID !== "string" ||1449 typeof e.mid !== "number" ||1450 !Number.isFinite(e.mid) ||1451 typeof e.name !== "string"1452 ) {1453 send(ws, { t: "error", message: "malformed pushMutation envelope" });1454 break;1455 }1456 this.handlePushMutation(ws, conn, {1457 clientID: e.clientID,1458 mid: e.mid,1459 name: e.name,1460 args: e.args,1461 });1462 break;1463 }1464 case "mutate":1465 // Raw CRUD never crosses this wire: room writes are named mutators (§4.2),1466 // validated against the owned set at the transaction boundary.1467 send(ws, {1468 t: "error",1469 queryId: msg.queryId,1470 message: "this room accepts named mutators only (pushMutation)",1471 });1472 break;1473 default:1474 break;1475 }1476 }14771478 private async handleSubscribe(1479 ws: WebSocket,1480 conn: ConnState,1481 msg: SubscribeMsg,1482 ): Promise<void> {1483 if (!this.room) return; // raced an incarnation death; the socket is being closed1484 const queryError = (message: string) =>1485 send(ws, { t: "queryError", queryId: msg.queryId, message });14861487 // The reserved lmid system query arrives BY NAME even in lease mode — the write1488 // plane's confirmation stream. The room composes the AST itself from the1489 // connection's `init` identity (client args are ignored, exactly as on rindled).1490 if (typeof msg.name === "string") {1491 if (msg.name !== LMID_QUERY_NAME) {1492 queryError("subscribe requires a lease token");1493 return;1494 }1495 if (!this.opts.downstream.writes) {1496 queryError("this room is read-only (no write plane configured)");1497 return;1498 }1499 if (!conn.clientID) {1500 queryError("subscribe lmid query before init");1501 return;1502 }1503 const prev = conn.queries.get(msg.queryId);1504 const epoch = prev ? prev.epoch + 1 : 1;1505 const subKey = `${conn.id}:${msg.queryId}`;1506 if (prev) this.subs.delete(prev.subKey);1507 let res: { hello: unknown; snapshot: unknown };1508 try {1509 res = JSON.parse(1510 this.room.lmidSubscribe(subKey, epoch, conn.clientID, Date.now()),1511 ) as typeof res;1512 } catch (e) {1513 queryError(`materialize failed: ${String((e as Error)?.message ?? e)}`);1514 return;1515 }1516 conn.queries.set(msg.queryId, { subKey, epoch });1517 this.subs.set(subKey, {1518 ws,1519 conn,1520 clientQid: msg.queryId,1521 // Not lease-gated: it lives exactly as long as its connection — never1522 // swept by exp, never a revocation key (revoking a user closes the socket).1523 user: "",1524 exp: Number.POSITIVE_INFINITY,1525 });1526 send(ws, {1527 t: "nhello",1528 queryId: msg.queryId,1529 hello: res.hello,1530 bootId: this.incarnationBootId,1531 });1532 send(ws, { t: "nbatch", queryId: msg.queryId, batch: res.snapshot });1533 this.sendProgress(conn);1534 return;1535 }1536 if (typeof msg.leaseToken !== "string") {1537 queryError("subscribe requires a lease token");1538 return;1539 }15401541 // The §10.1 gate: signature, doc, expiry — then the §4.1 revocation check.1542 const now = Date.now();1543 let payload;1544 try {1545 payload = await verifyRoomToken(msg.leaseToken, {1546 doc: this.opts.downstream.docId,1547 keys: this.opts.downstream.tokenKeys,1548 now,1549 });1550 } catch (e) {1551 queryError(e instanceof RoomTokenError ? e.message : "lease token refused");1552 return;1553 }1554 const revokedAt = this.revoked.get(payload.sub);1555 if (revokedAt !== undefined && payload.iat <= revokedAt) {1556 queryError("lease token refused: revoked");1557 return;1558 }15591560 // Scope-skew tripwire: a lease proving against scopes this room's gate did NOT arm1561 // with means a profile was edited after this room booted (the gate arms once). The1562 // subscribe is NOT refused — the gate is still sound, it will just deopt routed1563 // writes until the room re-boots. Log once per distinct skew so the otherwise-silent1564 // deopt loop is diagnosable. Both hashes present required (a pre-stamp token or v11565 // gate skips).1566 if (1567 this.armedScopesHash !== undefined &&1568 payload.scopesHash !== undefined &&1569 payload.scopesHash !== this.armedScopesHash1570 ) {1571 const key = `${payload.scopesHash} ${this.armedScopesHash}`;1572 if (!this.loggedScopeSkew.has(key)) {1573 this.loggedScopeSkew.add(key);1574 this.log(1575 `scope skew: this room's gate armed with scopes ${this.armedScopesHash} but a lease ` +1576 `proves against ${payload.scopesHash} — a room profile was edited under this live ` +1577 `room. Routed writes will deopt (safe, but degraded) until the room re-boots.`,1578 );1579 }1580 }15811582 // Subject binding (managed-writes §3.1): the FIRST verified lease binds the1583 // connection's authenticated subject; a later lease whose `sub` differs is refused1584 // — one principal per connection, so "the connection's user" is well-defined and1585 // privilege-mixing is closed.1586 if (conn.sub === null) {1587 conn.sub = payload.sub;1588 } else if (conn.sub !== payload.sub) {1589 queryError("lease token refused: connection is bound to another subject");1590 return;1591 }15921593 // A re-subscribe (gap recovery) replaces the prior envelope and bumps the1594 // DOWNSTREAM epoch — the same rule rindled applies per (conn, queryId).1595 const prev = conn.queries.get(msg.queryId);1596 const epoch = prev ? prev.epoch + 1 : 1;1597 const subKey = `${conn.id}:${msg.queryId}`;1598 if (prev) this.subs.delete(prev.subKey);15991600 let res: { queryKey: string; reused: boolean; hello: unknown; snapshot: unknown };1601 try {1602 res = JSON.parse(1603 this.room.subscribe(subKey, epoch, JSON.stringify(payload.ast), now),1604 ) as typeof res;1605 } catch (e) {1606 queryError(`materialize failed: ${String((e as Error)?.message ?? e)}`);1607 return;1608 }1609 conn.queries.set(msg.queryId, { subKey, epoch });1610 this.subs.set(subKey, {1611 ws,1612 conn,1613 clientQid: msg.queryId,1614 user: payload.sub,1615 exp: payload.exp,1616 });1617 send(ws, {1618 t: "nhello",1619 queryId: msg.queryId,1620 hello: res.hello,1621 bootId: this.incarnationBootId,1622 });1623 send(ws, { t: "nbatch", queryId: msg.queryId, batch: res.snapshot });1624 // The release point for the snapshot: an optimistic client buffers every nbatch1625 // until a progress frame's cvMin covers its cv.1626 this.sendProgress(conn);1627 }16281629 /** The periodic enforcement tick: drop subscriptions whose lease passed `exp`1630 * unrenewed (§4.1's TTL backstop), reclaim idle pipelines, prune old revocations. */1631 private sweep(): void {1632 const now = Date.now();1633 for (const [subKey, meta] of [...this.subs]) {1634 if (meta.exp <= now) {1635 send(meta.ws, {1636 t: "queryError",1637 queryId: meta.clientQid,1638 message: "expired lease — renew through the API server",1639 });1640 this.dropSub(subKey, meta, now);1641 }1642 }1643 if (this.room && this.live) {1644 this.room.sweep(now);1645 }1646 const window = this.opts.downstream.revocationWindowMs ?? DEFAULT_REVOCATION_WINDOW_MS;1647 for (const [user, at] of [...this.revoked]) {1648 if (now - at > window) this.revoked.delete(user);1649 }1650 }16511652 private dropSub(subKey: string, meta: SubMeta, now: number): void {1653 this.subs.delete(subKey);1654 meta.conn.queries.delete(meta.clientQid);1655 this.room?.unsubscribe(subKey, now);1656 }16571658 /** §4.1 layer 2: synchronous revocation. Terminates every subscription and socket of1659 * `user` and refuses their pre-revocation tokens from here on. */1660 private revokeUser(user: string): number {1661 const now = Date.now();1662 this.revoked.set(user, now);1663 const sockets = new Set<WebSocket>();1664 let dropped = 0;1665 for (const [subKey, meta] of [...this.subs]) {1666 if (meta.user !== user) continue;1667 send(meta.ws, {1668 t: "queryError",1669 queryId: meta.clientQid,1670 message: "lease revoked",1671 });1672 this.dropSub(subKey, meta, now);1673 sockets.add(meta.ws);1674 dropped++;1675 }1676 // A conn BOUND to the subject is write-capable even with no live subscription1677 // (managed-writes §3.1/§3.4) — its socket goes too.1678 for (const conn of this.conns) {1679 if (conn.sub === user) sockets.add(conn.ws);1680 }1681 for (const ws of sockets) {1682 ws.close(1008, "revoked");1683 }1684 this.log(`revoked ${user}: ${dropped} subscription(s)`);1685 return dropped;1686 }16871688 // ----------------------------- control plane ------------------------------16891690 private startControl(control: { authToken: string; port?: number }): Promise<void> {1691 const server = createServer((req, res) => {1692 const reply = (status: number, body: unknown) => {1693 res.writeHead(status, { "content-type": "application/json" });1694 res.end(JSON.stringify(body));1695 };1696 if (req.headers.authorization !== `Bearer ${control.authToken}`) {1697 return reply(401, { error: "unauthorized" });1698 }1699 if (req.method === "GET" && req.url === "/stats") {1700 return reply(200, {1701 live: this.live,1702 connections: this.wss.clients.size,1703 subscriptions: this.subs.size,1704 materializations: this.room?.materializationCount() ?? 0,1705 pendingMutations: this.room?.pendingLen() ?? 0,1706 dirtyKeys: this.room?.dirtyLen() ?? 0,1707 placementEpoch: this.placementEpoch,1708 flushesConfirmed: this.flushesConfirmed,1709 flushInFlight: this.flushBusy,1710 });1711 }1712 if (req.method === "POST" && req.url === "/revoke") {1713 let body = "";1714 req.on("data", (c) => (body += c));1715 req.on("end", () => {1716 try {1717 const { userId } = JSON.parse(body) as { userId?: string };1718 if (typeof userId !== "string" || userId.length === 0) {1719 return reply(400, { error: "userId required" });1720 }1721 reply(200, { revoked: this.revokeUser(userId) });1722 } catch {1723 reply(400, { error: "invalid JSON" });1724 }1725 });1726 return;1727 }1728 if (req.method === "POST" && req.url === "/drain") {1729 // §4.2 drain-before-downgrade: the api-server's `drainRoom` hook lands here. No body.1730 this.drainForDowngrade().then(1731 (finalFlushSeq) => reply(200, { finalFlushSeq, drained: true }),1732 (e) => reply(500, { error: String((e as Error)?.message ?? e) }),1733 );1734 return;1735 }1736 reply(404, { error: "unknown endpoint" });1737 });1738 this.control = server;1739 return new Promise((resolve) => {1740 server.listen(control.port ?? 0, "127.0.0.1", () => resolve());1741 });1742 }1743}17441745interface SubscribeMsg {1746 queryId: number;1747 name?: unknown;1748 leaseToken?: unknown;1749}17501751/** The wire mutation envelope (`{t:"pushMutation", envelope}`) — `clientID` capital1752 * ID, exactly as `@rindle/remote` sends it. */1753interface PushEnvelope {1754 clientID: string;1755 mid: number;1756 name: string;1757 args: unknown;1758}17591760/** A flush seq as its wire offset: zero-padded so `rindled`'s lexicographic keyset1761 * compare (`run_already_applied`) orders it like the number it is. */1762function padOffset(seq: number): string {1763 return String(seq).padStart(20, "0");1764}17651766/** Retry `op` on network-class failures (backoff, same inputs — for the flush that1767 * means the same journaled bytes) until it settles, `stop()` says quit, or the error1768 * is fatal (`{fatal: true}` — retrying cannot help). */1769async function withNetRetry<T>(1770 op: () => Promise<T>,1771 stop: () => boolean,1772 log: (line: string) => void,1773): Promise<T> {1774 let delay = FLUSH_RETRY_MIN_MS;1775 for (;;) {1776 try {1777 return await op();1778 } catch (e) {1779 if ((e as { fatal?: boolean })?.fatal === true || stop()) throw e;1780 log(`authority call failed (retrying in ${delay}ms): ${String(e)}`);1781 await new Promise((r) => {1782 const t = setTimeout(r, delay);1783 (t as { unref?: () => void }).unref?.();1784 });1785 if (stop()) throw e;1786 delay = Math.min(delay * 2, FLUSH_RETRY_MAX_MS);1787 }1788 }1789}17901791/** Positional table shapes from the upstream hello — the keyed MutationTx layer's1792 * schema. (The hello was already validated by `WasmRoom.open` before this runs.) */1793function shapesOf(hello: unknown): Map<string, TableShape> {1794 const shapes = new Map<string, TableShape>();1795 const tables = (1796 hello as { tables?: Array<{ name: string; columns: string[]; primaryKey: number[] }> }1797 )?.tables;1798 if (Array.isArray(tables)) {1799 for (const t of tables) {1800 shapes.set(t.name, { columns: t.columns, primaryKey: t.primaryKey });1801 }1802 }1803 return shapes;1804}18051806/** Boot a room shell: init the wasm, mint the upstream lease, connect the upstream leg,1807 * and serve the downstream ws (+ the private control plane, if configured). Returns1808 * once the ports are bound and the upstream connection is underway — await1809 * `shell.awaitLive()` for the seed. */1810export async function createRoomShell(opts: RoomShellOptions): Promise<RoomShell> {1811 const shell = new Shell(opts);1812 try {1813 await shell.start();1814 } catch (e) {1815 await shell.close();1816 throw e;1817 }1818 return shell;1819}1820