API index and search · Build metadata
Source snapshot
packages/optimistic/src/backend.ts
1// OptimisticBackend — the composition that adds OPTIMISTIC WRITES to the normalized2// local-first path (OPTIMISTIC-WRITES-DESIGN.md §8.5, §9). The richer cousin of3// `NormalizedBackend`: the same substrate (a local `@rindle/wasm` engine + a server4// stream + `NormalizedSync`), plus5//6// - a **named client-mutator registry**: `invoke("createIssue", args)` runs the7// mutator against the live engine NOW (the prediction shows instantly), pushes8// `(mid, name, args)` — never effects — onto the pending stack, and ships the9// envelope upstream (§4);10// - **lmid-as-data**: at construction the backend subscribes its own one-row11// SYSTEM QUERY (`LMID_QUERY_NAME` → `_rindle_client_mutations WHERE client_id =12// me`). The server writes `lmid` in the SAME transaction as a mutation's13// effects, so the confirmation arrives as an ordinary cv-tagged data frame and14// is released by the same `cvMin` — confirmation can never skew from the data15// of its own commit (the drain race the old frame-carried lmid suffered);16// - **cv-buffering**: every `cv`-tagged data frame buffers; a progress frame17// releases all `cv ≤ cvMin` as ONE coherent step — data ops into the engine's18// `sync` side, lmid-table ops into `confirmedLmid` (§1.3.1/§8.5 — no torn reads19// across queries, no apply ahead of the release point);20// - the **§1.3 reconcile cycle** per release: the wasm engine rewinds21// (`serverBatchBegin`), every still-pending mutator is **re-invoked** against the22// rebased base (read-dependent mutators read current values — §4.1), and the23// whole cycle coalesces to one minimal delivery (`serverBatchEnd`) — a24// confirmed-correct prediction produces zero churn;25// - per-query **ResultType** is the SERVER CHANNEL's state only (FOLDED-MUTATIONS-DESIGN26// §7): `unknown` while not hydrated, else `complete`. A pending local mutation no longer27// moves it — "is a prediction pending here?" is its own reactive axis (`pending(qid)` /28// `onPending`), so a fold held through its debounce window doesn't pin a query loading.29// There is NO server rejection signal: a failed mutation is processed-as-no-op (lmid30// advances, effects rolled back) and the prediction snaps back on the ordinary release.31// - **folded mutations** (FOLDED-MUTATIONS-DESIGN): `invokeFolded` collapses a run of32// same-key absorbing writes to ONE pending entry whose `args` are overwritten in place,33// debounces the server write (a virtual `clock` seam keeps it test-deterministic), and34// ships only the last value. The `mid` is assigned at FLUSH (not invoke), so the gapless35// wire sequence is preserved under debounce (§4.1); an overlapping non-fold write drains36// the interacting fold first (`drainOverlapping`, §4.2 flush-on-enqueue).37//38// `Store`/`ArrayView` are untouched: this implements `Backend`, so39// `new Store(schema, new OptimisticBackend(...))` reuses the whole machinery.4041import {42 CLIENT_MUTATIONS_SCHEMA,43 driveMutationSync,44 insertCell,45 insertPlan,46 isGeneratorMutator,47 isoTx,48 LMID_QUERY_NAME,49 localTableNames,50 normalizedTableSchemas,51 rowsEqual,52 tableSpec,53 toCell,54} from "@rindle/client";55import type {56 Ast,57 Backend,58 BackendDevObserver,59 ChangeEvent,60 ColsMap,61 ColType,62 Condition,63 CorrelatedSubquery,64 IsoTx,65 KeyedRow,66 Mutation,67 MutationEnvelope,68 MutationGen,69 MutationOp,70 MutationOutcomeFrame,71 MutatorCtx,72 NormalizedEvent,73 NormalizedOp,74 NormalizedTableSchema,75 OptimisticSource,76 ProgressFrame,77 QueryArg,78 QueryId,79 QueryResultRow,80 RemoteQuery,81 ResultType,82 Schema,83 WireValue,84} from "@rindle/client";85import { aggTableSchemas, NormalizedSync, rewriteAggregates, type ColCounts, type PkCols } from "@rindle/normalized";86import { WasmBackend, type ServerDeltaOp, type WasmWriteTxn } from "@rindle/wasm";8788import { AggOverlay, type ChildOp, collectAggDefs } from "./agg-overlay.ts";89import {90 decodeOutcomeRow,91 LIFECYCLE_TABLE_SCHEMAS,92 ROOM_CLIENT_MUTATIONS_TABLE,93 ROOM_MUTATION_OUTCOMES_TABLE,94 ROOM_WATERMARK_TABLE,95 roomDomainKey,96 SCOPE_SESSIONS_TABLE,97 type SystemStreamSpec,98} from "./system-streams.ts";99100export type { SystemStreamSpec, SystemStreamTable } from "./system-streams.ts";101102/** A keyed row: column name → cell. The ergonomic shape — column names are validated against the103 * schema at runtime, so a typo throws immediately with the valid names. Re-exported from104 * `@rindle/client` (the leaf both tiers share). */105export type { KeyedRow };106107/** {@link MutationTx.query}'s query handle ({@link QueryArg}, `{ ast(): Ast }`) and result row108 * ({@link QueryResultRow}) — the SAME types the isomorphic seam uses (203-MUTATOR-READS-DESIGN.md109 * §5.2/§9.1), re-exported from `@rindle/client` so the client and shared-generator surfaces share110 * one definition. */111export type { QueryArg, QueryResultRow };112113/** The write handle a client mutator runs against (the client `MutationTx`, §4.2):114 * reads see the current base + this transaction's own staged writes (§4.1).115 *116 * Prefer the KEYED methods (`insert`/`update`/`upsert`/`delete`/`row`) — named columns,117 * schema-checked. The positional methods (`get`/`add`/`remove`/`edit`) are the raw wire118 * shape: bare cells in schema column order, `pk` cells in `primaryKey` order. */119export interface MutationTx {120 // --- keyed (schema-aware) ---121 /** Read one row by primary key (e.g. `tx.row("issue", { id: 1 })`). */122 row(table: string, pk: KeyedRow): KeyedRow | undefined;123 /** Insert a FULL row (every column named; missing or unknown columns throw). */124 insert(table: string, row: KeyedRow): void;125 /** Update the row identified by the pk columns; only the named non-pk columns change.126 * A missing row is a NO-OP (rebase-friendly: the row may have vanished upstream). */127 update(table: string, row: KeyedRow): void;128 /** Insert, or fully replace when the pk already exists (a FULL row, like `insert`). */129 upsert(table: string, row: KeyedRow): void;130 /** Insert a FULL row, or do nothing if the pk already exists (the isomorphic form of the classic131 * `if (!tx.row(pk)) tx.insert(row)` upsert-if-absent; renders `ON CONFLICT DO NOTHING` server-side). */132 insertIgnore(table: string, row: KeyedRow): void;133 /** Delete the row identified by the pk columns. A missing row is a NO-OP. */134 delete(table: string, pk: KeyedRow): void;135 /** Run a one-shot read query (`where`/`orderBy`/`limit`/join) over the state this136 * mutator is mutating — it sees this transaction's own writes-so-far, the same137 * read-your-writes contract as `get`/`row` (§4.1; 203-MUTATOR-READS-DESIGN.md §5.2).138 * Synchronous; returns the matching rows in the query's order, each with its materialized139 * relationship children nested by name (presented identically to a `view.data` row). Pass140 * a query from the typed builder, e.g. `tx.query(q.issue.where("owner", "=", me))`.141 * Refused inside a FOLDED mutator (a reading mutator is non-absorbing, §9.1). */142 query(query: QueryArg): QueryResultRow[];143 // --- positional (the wire shape) ---144 get(table: string, pk: WireValue[]): WireValue[] | undefined;145 add(table: string, row: WireValue[]): void;146 remove(table: string, row: WireValue[]): void;147 edit(table: string, oldRow: WireValue[], newRow: WireValue[]): void;148}149150/** A client mutator: optimistic, deterministic, replayable — a pure function of `(base, args)` (§5:151 * no clock, no randomness; it is RE-INVOKED on every rebase). Either shape is accepted:152 * - a plain synchronous function `(tx, args) => void` (client-only), OR153 * - a shared GENERATOR `(tx, args, ctx) => MutationGen` (the isomorphic form: the SAME body the API154 * server runs against a live async transaction — MUTATORS-ISOMORPHIC). The driver detects which. */155export type ClientMutator =156 | ((tx: MutationTx, args: never) => void)157 | ((tx: IsoTx, args: never, ctx: MutatorCtx) => MutationGen);158159/** The client registry (§4.2) — one of the two registries; the server's authoritative160 * twin shares names (and possibly code), never the wire. */161export type ClientRegistry = Record<string, ClientMutator>;162163export type { ResultType };164165/** The reserved source-qid of the backend's own lmid system query. User/local query ids166 * are assigned by the `Store` starting at 1, so 0 never collides. */167const LMID_QID: QueryId = 0;168169/** One captured write, pk-granular (RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md §3.2 #1: "the170 * writers already receive `(table, row)`... this is pure capture, no semantic change"). `row` is171 * the post-write image (positional wire cells, schema column order); `undefined` for a `remove` —172 * no row survives it, and `row === undefined` stays the remove marker.173 *174 * `oldRow` is the PRE-IMAGE, captured for a `remove` AND an `edit` (H-ii): the full-width row175 * read via `tx.get` immediately BEFORE the write staged — read-your-writes, so a write to the176 * same pk earlier in the SAME invocation shows through — falling back to the caller's asserted177 * old row when the pk is not txn-visible (a raw remove/edit of an absent row; incl. the178 * pk-MOVING raw edit, whose record is keyed by the NEW pk yet whose pre-image is the caller's179 * OLD row). A captured remove/edit thus always carries a full-width pre-image — with ONE180 * exception: a record that collapses to a (re-)insert has none, see the matrix.181 *182 * Coalescing within one invocation is last-write-wins per pk on `row` (the final image matches183 * the engine head's own semantics for that pk) with `oldRow` pinned to the TXN-ENTRY BASE — the184 * H-ii matrix:185 * - edit-after-add / edit-after-remove: the record collapses to a (re-)insert — post-image186 * only, NO `oldRow` (the pk did not pre-exist this invocation's base; presence hold-back187 * uses `row`).188 * - edit-after-edit: keeps the FIRST pre-image (the txn-entry base — the chain nets to ONE189 * edit from the base to the final image).190 * - remove-after-edit / remove-after-remove: keeps the ORIGINAL pre-image (the first write's191 * captured base), NOT the edited transient — the net effect is a remove of the row the192 * external world last knew.193 * - remove-after-add: keeps the txn-visible pre-image (the transient added row — the pk had194 * no base, and this is the only truthful full-width row there is; G-iii pinned it).195 * - add-after-remove (a re-insert): drops `oldRow` (presence hold-back uses `row`).196 * Across rebase re-invocations the write-set is union-never-shrink ({@link mergeWriteSet}): a197 * re-run that no-ops keeps the prior record — and its pre-image — intact. */198export interface WriteRecord {199 table: string;200 pk: WireValue[];201 row: WireValue[] | undefined;202 oldRow?: WireValue[];203}204205/** The pk-granular write-set captured over ONE mutator invocation: table → pk-key (a stable-JSON206 * encoding of the pk cells, {@link stableJson}) → that pk's write, LAST-WRITE-WINS within the207 * invocation — an add-then-edit or edit-then-edit of the SAME pk collapses to its final image,208 * matching the engine head's own semantics for that pk. Chosen (over a flat array) because the209 * later routing proof needs "is pk P in the writable scope" / "did we already see this pk in this210 * invocation" as cheap lookups, and rebase re-invocation needs to MERGE a fresh write-set into an211 * accumulated one ({@link mergeWriteSet}) — both are Map operations, not scans.212 *213 * `touched` (the pre-existing table-granular `Set<string>` the pending axis reads, §7.2) is214 * exactly `new Set(writeSet.keys())` — derived from this, never separately populated, so the two215 * can never drift. */216export type WriteSet = Map<string, Map<string, WriteRecord>>;217218/** Whether a recorded point read (`tx.get`/`tx.row`) found a row. */219export type ReadOutcome = "present" | "absent";220221/** One recorded point read, pk-granular (§3.2 #2). Recording-mode only — see {@link ReadLog}.222 * Since H-ii this covers BOTH the public reads (`tx.get`/`tx.row`) and the keyed writers'223 * internal pre-existence probes (§3.2 #3 — see the `rawGet` note in {@link trackingTx}). */224export interface ReadRecord {225 table: string;226 pk: WireValue[];227 outcome: ReadOutcome;228}229230/** The read-log captured over ONE mutator invocation when recording is armed (RINDLE-REALTIME-231 * QUERY-ENABLEMENT-DESIGN.md §3.2 #2): every point read (`reads` — the public `tx.get`/`tx.row`232 * and, since H-ii, the keyed writers' internal pre-existence probes) plus every resolved query AST233 * (`queries`, from `tx.query`). A SIBLING of the folded read TRAP (`FoldReadError` below) — the234 * trap arms on the folded path and throws before any read completes (recording never runs there);235 * recording arms on the ordinary (non-folded) prediction run and never throws. Pure capture for236 * devtools/inspection (the §3 routing derivation it once fed was removed by237 * 302-ROOM-STORE-SEPARATION-DESIGN.md §5 — mutators DECLARE their domain now). */238export interface ReadLog {239 reads: ReadRecord[];240 queries: Ast[];241}242243interface PendingMutation {244 /** The wire mutation id. A FOLDED entry carries `null` until its window flushes — the `mid`245 * is dealt from `nextMid` in SEND order, never reserved at invoke, so the wire sequence stays246 * gapless under debounce (FOLDED-MUTATIONS-DESIGN §4.1). A `null` entry is never confirmable247 * (the confirm-drop retains it) and re-invokes AFTER every assigned mid. */248 mid: number | null;249 /** The client-global deal sequence, stamped in the same breath as {@link mid} (`null` while the250 * mid is). Mids are PER-DOMAIN (each authority numbers its own confirms, §7.1), so mids from251 * different domains are incomparable — a daemon mid 5 and a room mid 1 say nothing about which252 * was sent first. `seq` is the ONE total order across domains: **confirmation order is253 * per-domain; replay order is client-global** — the reconcile's re-invocation sort keys on254 * `seq`, never on `mid`. Within one domain `seq` order equals `mid` order (both dealt at the255 * same send-time choke point) EXCEPT across an H-v deopt re-enqueue: a flipped entry keeps its256 * ORIGINAL seq while its fresh daemon mid is dealt later, so its seq may undercut daemon257 * entries with smaller mids. That is the point — seq is the REPLAY order and the flip must not258 * move the entry's overlay position (a read-dependent sibling invoked after it replays on its259 * value); the only consumer of the ordering is the seq-keyed reconcile sort, which wants260 * exactly this. Single-domain behavior without deopts is unchanged. */261 seq: number | null;262 name: string;263 args: unknown;264 /** The confirming stream (RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md §7.1): which domain's ledger265 * dealt this entry's `mid` and whose confirm watermark can retire it. Resolved from the injectable266 * `domainPolicy` when the mid is dealt (for a folded entry, re-resolved at flush). `"daemon"` in267 * the single-domain configuration. An un-flushed fold (`mid == null`) carries its provisional268 * domain but is never confirmable until the flush stamps a real mid. */269 domain: string;270 /** Tables this mutator touched at its LAST invocation (drives the pending axis, §7.2). Derived271 * from `writes.keys()` — see {@link WriteSet}. */272 touched: Set<string>;273 /** The pk-granular write-set captured at this entry's LAST invocation (§3.2 #1). A rebase274 * re-invocation MERGES its fresh write-set into this one ({@link mergeWriteSet}), mirroring the275 * `touched` union: the key set only grows across re-invocations (a re-run that no-ops must not276 * shrink it, §7.2), each key's value is always the newest. Tables are ENGINE names: a277 * room-domain entry's writes on the room's own tables record the namespaced name (302 §2). */278 writes: WriteSet;279 /** The read-log captured at this entry's LAST *recorded* invocation (§3.2 #2). Empty for a280 * FOLDED entry (the trap, not recording, arms on that path — nothing is ever recorded there)281 * and left as the ORIGINAL invoke's log across a rebase re-invocation: recording is armed only282 * on the initial `invoke`, not the reconcile replay (a re-invocation runs against a rebased283 * base, so its read outcomes would need re-proving against the NEW base anyway — re-arming284 * recording there is future work, not required for pure capture). */285 reads: ReadLog;286}287288/** A virtual-clock seam for fold debounce timers and elapsed-time checks (FOLDED-MUTATIONS-DESIGN §9): the289 * oracle injects a deterministic scheduler; production defaults to real timers + `Date.now`. */290export interface FoldClock {291 setTimeout(cb: () => void, ms: number): unknown;292 clearTimeout(handle: unknown): void;293 now(): number;294}295296/** Options for a folded call site (FOLDED-MUTATIONS-DESIGN §3). `key` is the identity half of the297 * fold key (combined with the mutator name); the rest tune the debounce policy. */298export interface FoldOptions {299 /** The identity half of the fold key (typically the targeted primary key). Required. */300 key: unknown;301 /** Trailing debounce: the flush fires this long after the LAST invoke for `key`. Default 120ms. */302 debounceMs?: number;303 /** Elapsed-time threshold checked on each invocation. Once reached, that invocation flushes304 * immediately. This does not arm a separate timer: without another invocation, the trailing305 * debounce still controls the flush. Omit to use only the trailing debounce. */306 maxWaitMs?: number;307 /** For a declared room route, use this value for both `debounceMs` and `maxWaitMs`.308 * The domain policy selects the route on the window's first invocation; it stays fixed309 * for that window. `0` flushes each invocation. Omit to use the ordinary fold options.310 * This does not detect collaborators or infer a route from the writes. */311 roomDebounceMs?: number;312 /** Keep deferring across overlapping non-fold writes for maximum economy, accepting the §4.2313 * read-dependent reorder snap. Default `false` (flush-on-enqueue — correct-and-boring). */314 deferAcrossWrites?: boolean;315}316317/** A folded mutation window. `flush()` assigns its mutation ID and queues the latest arguments318 * for delivery. `mid` resolves with that ID at flush; it does not wait for server acceptance319 * or confirming-stream progress. Each call in the same window shares this promise. */320export interface FoldHandle {321 flush(): void;322 readonly mid: Promise<number>;323}324325/** The single live fold entry for one fold key, plus its debounce bookkeeping (§8). */326interface FoldRecord {327 /** The single pending entry; `entry.mid` stays `null` until `flushFold`. */328 entry: PendingMutation;329 /** Latest args observed for this key — the value the flush envelope ships (mirror of entry.args). */330 args: unknown;331 /** The live debounce timer handle (cleared on re-arm / flush). */332 timer: unknown;333 /** `clock.now()` at the first invoke of this window — for the `maxWaitMs` invocation threshold. */334 firstAt: number;335 debounceMs: number;336 maxWaitMs?: number;337 deferAcrossWrites: boolean;338 /** Resolves with the assigned mid at flush (the handle's `mid` promise). */339 midPromise: Promise<number>;340 resolveMid: (mid: number) => void;341}342343/** Thrown by the read-trap tx when a FOLDED mutator reads state (`tx.get`/`tx.row`) — the classic344 * non-absorbing shape, refused at the folded path (FOLDED-MUTATIONS-DESIGN §5). */345class FoldReadError extends Error {}346347interface BufferedFrame {348 cv: number;349 qid: QueryId;350 kind: "snapshot" | "batch";351 ops: NormalizedOp[];352 /** Arrival order — the tiebreak for equal-`cv` frames (release applies in order). */353 seq: number;354}355356/** ONE authority channel's coherence gate (RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md §5.1): the357 * per-source generalization of what used to be the backend's single `(buffer, appliedCv)` pair.358 * Each connected source — the daemon always; a `room:doc:X` once Slice G wires the live feed —359 * buffers its own cv-tagged frames and releases on its OWN `cvMin`, an independent cycle feeding360 * the one `applyRelease`: **coherent within a source; eventual across sources** (no joint361 * barrier, by design). `key` doubles as the lmid fold domain (§7.1) and the physical source the362 * release rebases (§5.3). `sync` is this source's OWN refcount/baseline space — folding two363 * sources through one `NormalizedSync` would refcount an overlapping row 1→2 and emit NOTHING364 * for the second source, silently starving its per-source baseline (§5.2 "a real store: its own365 * baseline"). */366interface SourceGate {367 key: string;368 source: OptimisticSource;369 sync: NormalizedSync;370 buffer: BufferedFrame[];371 /** Arrival counter behind {@link BufferedFrame.seq}, scoped to THIS gate's buffer. */372 nextSeq: number;373 appliedCv: number;374 /** A ROOM gate's wire-table → engine-table map (302 §2 — one source per table): the channel's375 * released deltas rename into the room's own namespaced tables, and a wire table NOT in the376 * map (context the room still publishes, or an unknown table) is DROPPED — the daemon is the377 * sole authority for context, so its copy must never enter the store from a room channel378 * (302 §6). Absent on the daemon gate (its deltas apply verbatim). */379 tableMap?: ReadonlyMap<string, string>;380}381382/** One I-iv doorbell event ({@link OptimisticBackend.onScopeSessions}, §4.1): a release folded383 * scope-session rows for `scope`, and `others` is the count of OTHER clients' unexpired sessions384 * there — {@link OptimisticBackend.otherScopeSessions} evaluated at fold time (the same one rule,385 * on the injectable {@link FoldClock}, so a virtual-clock harness gets deterministic verdicts).386 * The consumer (client.ts) triggers its one debounced re-lease on the 0→≥1 transition; expired387 * and own-clientID rows never count, so a solo client's own row can never ring its own bell. */388export interface ScopeSessionsEvent {389 scope: string;390 others: number;391}392393/** One demoted room source's §4.2 SWAP-BACK gate record (Slice I-v, re-expressed by 302 §4.2):394 * after a downgrade the room's namespaced tables keep backing their views — frozen at the room's395 * last state (the channel is disconnected) — until the daemon plane has provably absorbed the396 * room's final flush. Swapping the views back earlier would show the falling-back follower's397 * PRE-flush images (a visibly rolled-back document). The drop condition is evaluated after every398 * release ({@link OptimisticBackend.evaluateGhosts}):399 *400 * `roomWatermarks[doc] ≥ finalFlushSeq` (0 ⇒ trivially true — a never-flushed room)401 * AND no pending mutation with `domain === sourceKey` remains (sent-pins-domain, §7.5 —402 * room-domain entries retire ONLY through the outcome-resolved daemon-carried folds, I-iii).403 *404 * Both satisfied ⇒ {@link OptimisticBackend.dropGhost}: every room-swapped view re-registers on405 * its ORIGINAL (daemon-table) AST — value-equal under the fence, so visually a no-op — and the406 * room's namespaced tables unregister. */407interface RoomGhost {408 doc: string;409 finalFlushSeq: number;410 /** Whether the ONE stuck-downgrade event already fired for this ghost. */411 stuckReported: boolean;412}413414/** The I-v stuck-downgrade event ({@link OptimisticBackend.onDowngradeStuck}): the ghost's fence415 * is satisfied but these SENT room-domain mids never resolved (an entry that never reached the416 * room — sent-but-undelivered when the socket died — is undecidable in general, §7.5). The ghost417 * HOLDS (fail LOUD, never silent; no timeout-retire is invented) and the mids are surfaced once,418 * actionably. */419export interface DowngradeStuckEvent {420 sourceKey: string;421 doc: string;422 mids: number[];423}424425export interface OptimisticBackendOptions {426 /** Stable per-client identity for the upstream envelopes (§8.1). */427 clientID: string;428 /** Supplies a shared mutator's local `ctx.user` on every run, including reconciliation.429 * Keep the principal stable for this client's lifetime; recreate the client on account changes.430 * The server obtains its own authenticated principal. Defaults to the empty string. Plain431 * client-only mutators ignore this option. */432 user?: () => string;433 /** Buffered-frame ceiling before the §8.5 escape (drop + re-hydrate). */434 bufferCap?: number;435 /** Virtual-clock seam for the fold debounce timers (FOLDED-MUTATIONS-DESIGN §9). Defaults to436 * real `setTimeout`/`clearTimeout`/`Date.now`; the fold oracle injects a deterministic clock. */437 clock?: FoldClock;438 /** The DECLARED confirming stream per mutation (302 §5: declared, not derived — there is no439 * routing proof). A policy returning a string pins that domain verbatim: the mutation stages440 * onto that room's namespaced tables and ships on its channel. Returning `undefined` (or441 * configuring no policy) means `"daemon"`. The client layer builds this from the app's declared442 * realtime mutators + the currently attached rooms; a misdeclaration fails SOFT (302 §5.1) —443 * the write lands on the other authority's tables and the view simply stops feeling instant444 * until the echo relays it. */445 domainPolicy?: (name: string, args: unknown) => string | undefined;446 /** A FINAL (authz/validation) mutation rejection's reason surface — the room plane's twin of the447 * HTTP mutate route's `onRejected` (H-v; the H-iv-b `mutationOutcome {kind:"rejected"}` frame).448 * The prediction's snap-back is NOT this callback's job: the room burns the mid and its lmid449 * release drops the entry exactly as a daemon-path rejection does (processed-as-no-op) — this450 * is where the REASON reaches the app, same contract as the queue's callback. Also invoked when451 * a DEOPT's fresh re-invocation (the already-retired arm) throws — that mutation is dead on the452 * current base with no stream left to confirm it, the closest thing to a rejection there is. */453 onRejected?: (envelope: MutationEnvelope, reason: string) => void;454}455456// --- dev-only introspection (DEBUG-TOOLS-BROWSER-DESIGN §2/§4.1) -----------------457// A single read-only snapshot of the optimistic loop's state for a devtools pane — the source the458// "mutation timeline" reconstructs the fork/rebase lifecycle from. Every field below is already459// held by the backend; `__inspect()` just copies it out (no new hot-path state). The shapes are460// mirrored by `@rindle/devtools`' own `OptimisticInspect` (kept structurally identical there so the461// core need not import this package and drag in the wasm artifact at typecheck time).462463/** One folded entry's debounce window, for the timeline's fold drill-down (§4.1). */464export interface FoldInspect {465 /** The fold key (`${name}\0${identityJSON}`) collapsing same-key invokes into one entry. */466 foldKey: string;467 debounceMs: number;468 maxWaitMs?: number;469 deferAcrossWrites: boolean;470 /** Whether the window has flushed (a real `mid` was dealt); an un-flushed fold has `mid == null`. */471 flushed: boolean;472}473474/** One pending mutation, as the timeline sees it (DEBUG-TOOLS-BROWSER-DESIGN §4.1). */475export interface PendingInspect {476 /** Stable identity across snapshots: `m:<mid>` once a mid is assigned, else `f:<foldKey>` for an477 * un-flushed folded entry (the mid is dealt at flush, FOLDED-MUTATIONS-DESIGN §4.1). */478 key: string;479 /** The wire mutation id, or `null` for an un-flushed fold. */480 mid: number | null;481 name: string;482 args: unknown;483 /** Tables this mutator touched at its last (re)invocation — the pending-axis basis (§7.2). */484 tables: string[];485 /** The pk-granular write-set captured at this entry's LAST invocation (RINDLE-REALTIME-QUERY-486 * ENABLEMENT-DESIGN.md §3.2 #1), flattened from the {@link WriteSet} map for inspection — one487 * entry per `(table, pk)` currently held. Pure capture; no routing consumer yet. */488 writes: WriteRecord[];489 /** The read-log captured at this entry's LAST *recorded* invocation (§3.2 #2). Empty for a490 * folded entry — the read TRAP arms there, not recording (see {@link PendingMutation.reads}). */491 reads: ReadLog;492 /** Present iff this entry is a folded (debounced) write. */493 fold?: FoldInspect;494}495496/** A read-only snapshot of the optimistic loop ({@link OptimisticBackend.__inspect}). */497export interface OptimisticInspect {498 /** The pending stack in array order (assigned mids ascending, un-flushed folds interleaved by499 * creation — the re-invoke order is derived from this in `runReconcileCycle`). */500 pending: PendingInspect[];501 /** High-water confirmed mid: an entry with `mid <= confirmedLmid` has been confirmed/dropped. */502 confirmedLmid: number;503 /** The next mid to be dealt (so `nextMid - 1` is the highest issued). */504 nextMid: number;505 /** The applied coherent-release watermark (`cvMin`, §8.6). */506 appliedCv: number;507 /** Frames still buffered awaiting their release point (§8.5) — a backpressure gauge. */508 bufferedFrames: number;509 /** Every table some pending mutation currently touches (the coarse pending indicator set, §7.2). */510 pendingTables: string[];511}512513/** Default trailing-debounce window for a folded call site (FOLDED-MUTATIONS-DESIGN §3 example). */514const DEFAULT_FOLD_DEBOUNCE_MS = 120;515516/** The real-timer clock used when none is injected. */517const REAL_CLOCK: FoldClock = {518 setTimeout: (cb, ms) => setTimeout(cb, ms),519 clearTimeout: (h) => clearTimeout(h as ReturnType<typeof setTimeout>),520 now: () => Date.now(),521};522523/** The (shared, frozen-by-convention) empty map {@link OptimisticBackend.roomTablesFor} answers524 * for a room with no registered tables. */525const EMPTY_ROOM_TABLES: ReadonlyMap<string, string> = new Map();526527/** Per-domain retention cap for the processed-outcome set (H-v) — mirrors the shell's528 * `MAX_RECORDED_OUTCOMES_PER_CLIENT`: the sender caps what it can re-answer at 512 per client,529 * so remembering more than 512 processed mids per domain buys nothing. */530const MAX_PROCESSED_OUTCOMES_PER_DOMAIN = 512;531532export class OptimisticBackend<S extends ColsMap> implements Backend {533 private readonly local: WasmBackend<S>;534 private readonly sync: NormalizedSync;535 private readonly source: OptimisticSource;536 private readonly registry: ClientRegistry;537 private readonly clientID: string;538 /** The acting principal provider for a shared mutator's `ctx.user` (§ shared mutators). */539 private readonly user: () => string;540 private readonly bufferCap: number;541 /** Column order + pk indices per table, for the keyed `MutationTx` methods. */542 private readonly specs: TableSpecs;543 /** Local-only table names (`201-LOCAL-ONLY-TABLES-DESIGN.md` §4). Drives: the agg-rewrite gate544 * (L1 — a local-child count stays a native reduce), the mutator guard (M1 — a replayable545 * mutator may not read/write one), and `writeLocal` (M2 — it accepts ONLY these). */546 private readonly localTables: Set<string>;547 /** Each table's full column count (union-row width) + column-name → base ColId — to learn a548 * projected query's per-table projection off its `hello` and register it with the sync layer,549 * so it scatters that query's narrower rows into the shared union (PROJECTION-SUPPORT-DESIGN550 * §5.2). Without this a projected query's short rows reach the wasm `Db` un-scattered and fail551 * its width check. */552 private readonly colCounts: ColCounts;553 private readonly colIndex: Record<string, Map<string, number>>;554 /** Per-table pk column indices — held so `connectSource` can build a fresh per-source555 * `NormalizedSync` with the same layout the daemon's uses. */556 private readonly pkCols: PkCols;557 /** The client's OWN typed per-table schemas + the reserved lmid table — the fixed base558 * of the expected-schema set (CRIT#4 validation). Synthetic agg tables are appended as559 * queries arrive (`ensureSyntheticTables`). */560 private readonly clientTablesBase: NormalizedTableSchema[];561 /** Synthetic aggregate tables (`__agg_*`) registered so far, by name (AGGREGATE-SYNC-DESIGN562 * §3.3). Per aggregate DEFINITION (not per query), so two queries over the same count563 * share one table. */564 private readonly synthetic = new Map<string, NormalizedTableSchema>();565 /** Synthetic table name → how many registered LOCAL queries reference it. A table is566 * materialized on the `0→1` transition and reclaimed (engine source + baseline + refcount567 * layer + overlay def) on `1→0` — so aggregate state is not permanent (§4). */568 private readonly syntheticRefs = new Map<string, number>();569 /** Local qid → the synthetic tables it referenced at registration, to decrement on teardown. */570 private readonly queryAggTables = new Map<QueryId, string[]>();571 /** The optimistic aggregate overlay (§4–§6): the per-aggregate definitions + the per-group572 * pending delta `displayed = server_base ⊕ local_pending_delta` is applied from. */573 private readonly overlay = new AggOverlay();574 private handler: (qid: QueryId, ev: ChangeEvent) => void = () => {};575 // The local qids whose reconcile-cycle batch delivery this release is their FIRST hydration (empty576 // → full): their batch is the initial result set arriving as a delta, so it is stamped `catchUp`577 // and the Store maps it to the `snapshot` change-phase (a narrator ignores it by default). Non-null578 // only for the duration of the reconcile cycle in `onProgress`; a re-hydrate after a drop is a real579 // footprint diff (genuine change) and is NOT remapped. See {@link ChangeEvent} `catchUp`.580 private catchUpQids: Set<QueryId> | null = null;581 /** Newly-hydrated qids whose reconcile ACTUALLY emitted a (catch-up-stamped) batch — recorded by the582 * local-event forwarder alongside {@link catchUpQids}. After the reconcile, any newly-hydrated qid583 * NOT in here folded nothing (0 rows, or its result already present via a sibling → 0 net muts, or584 * the reconcile was skipped), so `onProgress` sends it an explicit empty catch-up — else its SSR585 * seed would never retire (the view freezes). Non-null only for the reconcile's duration. */586 private catchUpEmitted: Set<QueryId> | null = null;587 /** The Store's commit-boundary handler ({@link Backend.onCommitBoundary}), forwarded from the588 * local engine's `dispatch` brackets so the Store folds every affected view before notifying any589 * subscriber (cross-view-atomic notification). All this backend's data deltas originate from the590 * local engine, so its commit brackets are this backend's commit brackets. */591 private boundaryHandler: (phase: "begin" | "end") => void = () => {};592 private readonly devObservers = new Set<BackendDevObserver>();593594 private pendingMutations: PendingMutation[] = [];595 /** The next mid to deal, PER DOMAIN (RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md §7.1): a client596 * writing through room + daemon concurrently must not alias one lmid counter. Seeded with the597 * `"daemon"` stream at 1; a domain absent from the map starts at 1. In the single-domain598 * configuration only `"daemon"` is ever touched, so the sequence is byte-for-byte as before. */599 private nextMid = new Map<string, number>([["daemon", 1]]);600 /** The client-global deal counter behind {@link PendingMutation.seq}: one sequence across ALL601 * domains, bumped whenever any domain's mid is dealt. The replay order (mids are per-domain and602 * incomparable across domains — see the `seq` field doc). */603 private dealSeq = 0;604 /** The explicit confirming-stream override (§7.1/§3) — see605 * {@link OptimisticBackendOptions.domainPolicy}. `undefined` from it ⇒ H-iii derivation. */606 private readonly domainPolicy: (name: string, args: unknown) => string | undefined;607 /** The final-rejection reason surface ({@link OptimisticBackendOptions.onRejected}). */608 private readonly rejectedHandler: (envelope: MutationEnvelope, reason: string) => void;609 /** Processed `(domain, mid)` outcome frames (H-v) — the deopt handshake's idempotence guard: a610 * duplicate frame (the original plus a reconnect re-send's re-answer, or two re-answers across611 * two reconnects) must not double-invoke. Needed precisely because a deopt frame can arrive for612 * an ALREADY-RETIRED mid (the replay gotcha) — "no matching entry" alone cannot distinguish613 * "handle it fresh" from "already handled". Per-domain FIFO, capped like the shell's614 * recorded-outcome map ({@link MAX_PROCESSED_OUTCOMES_PER_DOMAIN}); past the cap a duplicate of615 * an evicted mid would be re-processed — the same bounded-window trade the shell makes, and it616 * takes 512 interleaving non-applied outcomes on one domain to open it. */617 private readonly outcomesProcessed = new Map<string, Set<number>>();618 /** THE room-table registry (302 §2 — one source per table): per connected room `sourceKey`, the619 * wire-table → engine-table map for the tables that room OWNS (its writable scope). Written by620 * {@link registerRoomTables} (same breath as the engine registration); read by the gate's621 * release rename/filter, the mutator staging map, the view swap ({@link processSwapIns}), and622 * the client's `__realtimeInspect` bookkeeping. The record outlives a downgrade's disconnect —623 * the ghost's views still read the engine tables — and drops at {@link dropGhost} (or the last624 * clean release via {@link unregisterRoomTables}). */625 private readonly roomTables = new Map<string, Map<string, string>>();626 /** Local view qids currently REGISTERED on a room's namespaced tables (302 §4 swap-in), →627 * their sourceKey. Set by {@link processSwapIns}; cleared by the swap-back ({@link dropGhost})628 * and view teardown. The original AST stays in {@link asts} throughout — the swap re-registers629 * only the ENGINE query. */630 private readonly roomSwappedViews = new Map<QueryId, string>();631 /** Room subs whose FIRST snapshot released in the current release — their views swap onto the632 * room tables at the release tail ({@link processSwapIns}), strictly AFTER the reconcile folded633 * the snapshot into those tables (swapping earlier would hydrate the view EMPTY, a flash). */634 private readonly pendingSwapIns = new Set<RemoteSub>();635 /** The live fold entries, by fold key `${name}\0${identityJSON}` — at most one per key636 * (FOLDED-MUTATIONS-DESIGN §8). Insertion order is creation order (the drain/flush tiebreak). */637 private readonly folds = new Map<string, FoldRecord>();638 /** The fold debounce clock (real timers by default; the oracle injects a virtual one). */639 private readonly clock: FoldClock;640 /** The high-water confirmed mutation id, PER DOMAIN (RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md641 * §7.2 per-domain confirm-drop): an entry with `mid <= watermark[entry.domain]` has been642 * confirmed. The `"daemon"` domain is folded from the lmid system query's RELEASED ops643 * (lmid-as-data) — never from a frame; a room domain will fold from its own lmid stream (later644 * slice). Seeded with `"daemon"` at 0; the daemon scalar `confirmedLmid` (devtools) is645 * `watermark.get("daemon")`. */646 private watermark = new Map<string, number>([["daemon", 0]]);647 /** The per-source coherence gates (§5.1), by source key. Seeded with the daemon gate at648 * construction; a room channel attaches later (`connectSource`). Single-domain: one entry,649 * and every gate-generalized path degenerates to the old single-buffer code. NOT the same650 * space as {@link watermark}/{@link nextMid}: a DOMAIN can confirm with no gate connected651 * (the `__testRelease` seam); a gate's `key` names the domain its lmid stream folds into. */652 private readonly gates = new Map<string, SourceGate>();653 /** The daemon's gate — the always-present channel (constructor-attached). The devtools654 * scalars (`__inspect`) read it directly; its `sync` IS {@link sync} (the agg overlay and655 * synthetic tables are daemon-tracked by design). */656 private readonly daemonGate: SourceGate;657 // --- the §4 lifecycle SYSTEM-STREAM plane (Slice I-iii) --------------------------------658 /** System retains by source qid ({@link retainSystemQuery}): a subscription with NO store view659 * and NO user-visible table — its frames buffer on its gate exactly like {@link LMID_QID}'s and660 * fold at RELEASE time ({@link foldSystemFrames}), never entering the sync layer or the local661 * engine. The spec names which system table the qid serves and the scope/doc it was minted for662 * (the fold's row filter). Empty on every non-lifecycle client — every partition below is then663 * a structural no-op and the release path is byte-identical to before. */664 private readonly systemQids = new Map<QueryId, SystemStreamSpec>();665 /** The §4.2 fence state: room doc → highest `flush_seq` delivered through the daemon plane666 * (monotone max-fold; a remove never regresses it). Slice I-v's ghost-drop consumer — I-iii667 * only maintains + exposes it (`__inspectDomains().lifecycle`). */668 private readonly roomWatermarks = new Map<string, number>();669 /** The §4.1 occupancy state: scope → (client_id → expires_at) from the doorbell stream. Slice670 * I-iv's doorbell consumer (the 1→2 re-lease reaction) — I-iii only maintains + exposes it.671 * A snapshot REPLACES the scope's map (authoritative re-hydrate); a batch folds add/edit/remove672 * incrementally (the age-out sweep's deletes arrive as removes). */673 private readonly scopeSessions = new Map<string, Map<string, number>>();674 /** The I-iv doorbell event sink ({@link onScopeSessions}) — fired once per scope a release's675 * scope-session fold touched, AFTER the whole release applied. Default no-op: a client that676 * never registers (no lifecycle plane) pays nothing. */677 private scopeSessionsHandler: (event: ScopeSessionsEvent) => void = () => {};678 /** Deferred old-channel row GC for in-flight upgrade retargets ({@link retargetRemoteQuery}):679 * sub sourceQid → the channel it left. The rows the OLD gate's sync holds for the qid stay680 * visible (merge: daemon tier) until the sub's first snapshot RELEASES on its new room channel681 * ({@link flushRetargetGc}) — dropping them at retarget time would emit net removes ahead of682 * the room's re-adds, the flicker the two-phase cutover exists to avoid. Doubles as the683 * wrong-channel GRACE window in {@link onFrame}: a frame already in flight from the old684 * channel when the sub moved is stale, not a wiring bug. Empty on every non-upgrade client —685 * every consultation below is then a structural no-op. */686 private readonly pendingRetargetGc = new Map<QueryId, string>();687 /** The §4.2 GHOSTS (Slice I-v): demoted room sources awaiting their watermark fence, by688 * sourceKey. Written only by {@link demoteRoomSource}; evaluated after every release689 * ({@link evaluateGhosts}) and dropped by {@link dropGhost} once the fence clears with no690 * sent room-domain pending left. Empty on every non-downgrade client — the per-release691 * evaluation is then a structural no-op. */692 private readonly ghosts = new Map<string, RoomGhost>();693 /** The I-v stuck-downgrade surface ({@link onDowngradeStuck}) — fired AT MOST ONCE per ghost694 * when its fence is satisfied but sent room-domain mids remain unresolved (§7.5: they retire695 * only through outcome resolution; the ghost holds rather than inventing a timeout-retire).696 * Default no-op. */697 private downgradeStuckHandler: (event: DowngradeStuckEvent) => void = () => {};698699 private readonly asts = new Map<QueryId, Ast>();700 /** Per query: the base tables its result can draw from (from the AST tree). */701 private readonly queryTables = new Map<QueryId, Set<string>>();702 private readonly remoteSubs = new Map<string, RemoteSub>();703 private readonly sourceToRemote = new Map<QueryId, string>();704 private readonly localToRemote = new Map<QueryId, string>();705 private readonly remoteRetainToLocal = new Map<QueryId, QueryId | undefined>();706 private readonly resultTypes = new Map<QueryId, ResultType>();707 /** Local view qids that are server-authoritative: a query with no remote sub (purely local) is708 * hydrated on registration; a remote query is hydrated when its sub's first snapshot releases.709 * An un-hydrated query reports `unknown` (still loading) — the basis of `resultType`. */710 private readonly hydrated = new Set<QueryId>();711 private resultTypeHandler: (qid: QueryId, rt: ResultType) => void = () => {};712 /** The pending AXIS (§7.2), split off `ResultType`: per query, whether any pending mutation713 * touches its tables. Cached so `onPending` fires only on transitions (invoke ↔ confirm). */714 private readonly pendingState = new Map<QueryId, boolean>();715 private pendingHandler: (qid: QueryId, pending: boolean) => void = () => {};716 /** The local-persistence write-through tap (`207-LOCAL-TABLE-PERSISTENCE-DESIGN.md` §5.1):717 * {@link writeLocal} invokes it post-commit; {@link applyLocalReplica} deliberately does not. */718 private localWriteObserver: ((mutations: Mutation[]) => void) | null = null;719720 constructor(721 schema: Schema<S>,722 source: OptimisticSource,723 registry: ClientRegistry,724 opts: OptimisticBackendOptions,725 ) {726 this.local = new WasmBackend(schema);727 this.local.onEvent((qid, ev) => {728 // Stamp a newly-hydrating query's reconcile batch as a catch-up (initial-hydration) delivery,729 // so the Store phases it as a `snapshot` rather than narrating the whole first result set. Record730 // that we emitted a hydration batch for this qid, so `onProgress` knows which newly-hydrated qids731 // still need an explicit empty catch-up (they folded nothing — see {@link catchUpEmitted}).732 const stamp = ev.type === "batch" && this.catchUpQids?.has(qid) === true;733 if (stamp) this.catchUpEmitted?.add(qid);734 this.handler(qid, stamp ? { ...ev, catchUp: true } : ev);735 });736 // Forward the local engine's commit brackets up to the Store (cross-view-atomic notification):737 // every data delta this backend emits comes from `this.local`, so its commit boundaries are ours.738 this.local.onCommitBoundary((phase) => this.boundaryHandler(phase));739 this.colCounts = colCountsFromSchema(schema);740 this.colIndex = colIndexFromSchema(schema);741 this.pkCols = pkColsFromSchema(schema);742 this.sync = new NormalizedSync(this.pkCols, this.colCounts);743 this.specs = tableSpecsFromSchema(schema);744 this.localTables = localTableNames(schema);745 this.source = source;746 this.registry = registry;747 this.clientID = opts.clientID;748 this.user = opts.user ?? (() => "");749 this.bufferCap = opts.bufferCap ?? 1024;750 this.clock = opts.clock ?? REAL_CLOCK;751 // No policy configured ⇒ every route DERIVES (H-iii §3). With no room gate connected the752 // derivation short-circuits to "daemon", so a single-domain app is byte-for-byte as before.753 this.domainPolicy = opts.domainPolicy ?? (() => undefined);754 this.rejectedHandler = opts.onRejected ?? (() => {});755 // The reserved lmid table + (I-iii) the four lifecycle system tables join the expected set so756 // a system subscription's hello passes CRIT#4 validation. Extra CLIENT-side entries are inert757 // for every other server hello (validation only checks tables a server advertises), so a758 // client that never receives a lifecycle block is byte-identical.759 this.clientTablesBase = [...normalizedTableSchemas(schema), CLIENT_MUTATIONS_SCHEMA, ...LIFECYCLE_TABLE_SCHEMAS];760 // The daemon is the always-present channel: its gate is attached at construction, and its761 // per-source refcount space IS `this.sync` (the agg overlay reads it directly). A room762 // channel attaches through the same seam later (§5.1; Slice G).763 this.daemonGate = this.attachGate("daemon", source, this.sync);764 }765766 /** Wire one authority channel into its own coherence gate (§5.1): every frame the channel767 * delivers buffers on THIS gate's cv timeline, its progress frames release THIS buffer, its768 * restart resets THIS gate alone, and its reserved lmid stream folds into `watermark[key]`.769 * Validates each server hello against our OWN typed schema → reject a schema skew (CRIT#4);770 * the reserved lmid table is part of the expected set so the system query's hello passes, and771 * synthetic agg tables join the set as queries register them. */772 private attachGate(key: string, source: OptimisticSource, sync: NormalizedSync): SourceGate {773 const gate: SourceGate = { key, source, sync, buffer: [], nextSeq: 0, appliedCv: 0 };774 this.gates.set(key, gate);775 source.expectClientSchema?.([...this.clientTablesBase, ...this.synthetic.values()]);776 source.onNormalized((qid, ev) => this.onFrame(gate, qid, ev));777 source.onProgress((frame) => this.onGateProgress(gate, frame));778 source.onRestart?.(() => this.resetGate(gate));779 // The deopt handshake's client half (H-v §3.3): the channel's `mutationOutcome` frames arrive780 // as `(domain = gate.key, frame)`. OUT-OF-BAND — the source dispatches on arrival and this781 // handler runs immediately, NEVER behind the gate's cv buffer: a deopt must migrate its entry782 // BEFORE the buffered lmid release that would otherwise retire it as a success (and the §7.3783 // hold-back trigger, keyed on `p.domain`, would park its staged writes the wrong way).784 source.onMutationOutcome?.((frame) => this.handleMutationOutcome(gate.key, frame));785 // §7.5 rule 3 (H-v): a re-established session re-sends this DOMAIN's unconfirmed pending786 // envelopes with their ORIGINAL mids — the authority's own ledger dedups (an applied mid is787 // silent; a non-applied one is re-answered from the recorded-outcome map into the handler788 // above). This is the deopt crash-window closer: a frame lost with its socket is re-earned.789 source.onResync?.(() => this.resendPending(gate.key));790 // The lmid system query (lmid-as-data): confirmations arrive on this channel's stream,791 // cv-tagged, released by the same cvMin as the data they belong to. The server derives792 // the identity from the connection; args are advisory. Qid 0 is reserved PER CHANNEL —793 // it never collides with Store-dealt qids and never enters the sync layer.794 source.registerQuery(LMID_QID, { name: LMID_QUERY_NAME, args: {} });795 return gate;796 }797798 /** Attach a SECOND authority channel (§5.1) — the seam Slice G's room upgrade calls with the799 * ws-backed room feed. Rooms speak the daemon protocol verbatim (§2.4: the client cannot tell800 * a room from the daemon), so the argument is a full {@link OptimisticSource} — exactly what801 * `@rindle/remote` builds from `{roomUrl, leaseToken}`. The channel buffers/releases on its802 * own cv timeline (an independent §5.1 gate: coherent within, eventual across) and its803 * reserved lmid stream folds into `watermark[sourceKey]` — so `sourceKey` must equal the804 * `domainPolicy` name for the mutations this authority confirms. The converse is NOT required:805 * a domain may exist with no connected gate (`__testRelease` drives confirms gate-less); the806 * live production path stays daemon-only until G calls this. */807 connectSource(sourceKey: string, source: OptimisticSource): void {808 if (this.gates.has(sourceKey)) {809 throw new Error(`optimistic backend: source ${sourceKey} is already connected`);810 }811 const gate = this.attachGate(sourceKey, source, new NormalizedSync(this.pkCols, this.colCounts));812 // A re-upgrade of a doc whose tables are still registered (a ghost that never dropped, or a813 // quick down/up bounce) adopts the surviving record as this incarnation's rename map.814 const tables = this.roomTables.get(sourceKey);815 if (tables !== undefined) gate.tableMap = tables;816 // …and CANCELS the pending swap-back: the room is the authority again, its views stay swapped,817 // and a ghost left armed would fire against this LIVE gate when the old fence clears —818 // un-swapping the views and unregistering the namespaced tables the gate's tableMap still819 // renames deltas into (the next release would then throw from serverBatchBegin and poison the820 // rebase state). A future downgrade arms a fresh ghost with its own fence.821 this.ghosts.delete(sourceKey);822 }823824 /** Register the tables room `sourceKey` OWNS (its writable scope — 302 §2): each wire table825 * gets its own namespaced ENGINE table (`{@link roomEngineTable}`), an ordinary tracked table826 * whose sole authority is the room channel. From here on the channel's released deltas rename827 * into these tables (wire tables outside the map are DROPPED — context stays daemon-owned,828 * 302 §6), room-domain mutators stage onto them, and a room-homed view swaps onto them once829 * the room sub hydrates ({@link processSwapIns}). Idempotent per (sourceKey, table); a wire830 * table unknown to the schema is skipped (nothing to hold rows for). */831 registerRoomTables(sourceKey: string, tables: readonly string[]): void {832 if (sourceKey === "daemon") {833 throw new Error("optimistic backend: the daemon is not a room — no namespaced tables");834 }835 let map = this.roomTables.get(sourceKey);836 if (!map) this.roomTables.set(sourceKey, (map = new Map()));837 for (const table of tables) {838 if (map.has(table)) continue;839 const spec = this.specs[table];840 if (spec === undefined || this.localTables.has(table)) continue;841 const engineTable = roomEngineTable(table, sourceKey);842 this.local.registerTable(engineTable, { columns: spec.columns, primaryKey: spec.primaryKey });843 map.set(table, engineTable);844 }845 const gate = this.gates.get(sourceKey);846 if (gate !== undefined) gate.tableMap = map;847 }848849 /** The wire-table → engine-table map for room `sourceKey`'s owned tables (empty when none) —850 * the client's idempotence check and `__realtimeInspect` read THIS record (one source of851 * truth; the client keeps no shadow copy). */852 roomTablesFor(sourceKey: string): ReadonlyMap<string, string> {853 return this.roomTables.get(sourceKey) ?? EMPTY_ROOM_TABLES;854 }855856 // --- the Backend seam ---------------------------------------------------------857858 /** `channel` (G-iii registration-time routing) names the authority channel the remote sub859 * registers on — a `connectSource`d gate key; default `"daemon"` (every existing caller is860 * byte-identical). Slice G-v threads the lease's `realtime.sourceKey` here. Validated FIRST861 * (like the E3 check below): a bad channel must throw before any per-query state is recorded. */862 registerQuery(qid: QueryId, ast: Ast, remote?: RemoteQuery, channel?: string): void {863 if (remote) this.requireGate(channel ?? "daemon");864 // queryTables is derived from the ORIGINAL ast — its `count(comments)` subquery names865 // `comment`, so an optimistic comment mutation flips this query to `unknown` (§6). The866 // local engine, by contrast, runs the REWRITTEN ast (reads the synthetic `__agg_*`).867 const tables = collectTables(ast);868 // E3/Q1 (`201-LOCAL-ONLY-TABLES-DESIGN.md`, backend chokepoint): a REMOTE (named) query may869 // never reference a local-only table — the server has no such table, and a smuggled ref would870 // either leak its existence or hit an unknown-table error upstream. A local query is nameless871 // (no `remote`) and runs entirely on the local engine, so it is exempt. This runs BEFORE we872 // record any per-query state (asts/queryTables): the throw path never reaches unregisterQuery873 // (the only cleanup), and refreshPending() iterates queryTables.keys(), so a rejected qid left874 // in those maps would be an orphaned leak processed by the pending axis.875 if (remote) {876 for (const t of tables) {877 if (this.localTables.has(t)) {878 throw new Error(879 `remote query "${remote.name}" references local-only table "${t}" — local tables never cross the wire (201-LOCAL-ONLY-TABLES-DESIGN.md E3).`,880 );881 }882 }883 }884 this.asts.set(qid, ast);885 this.queryTables.set(qid, tables);886 // A relationship `count` is DISPLAYED from a server-authoritative synthetic base table,887 // not recomputed locally (AGGREGATE-SYNC-DESIGN.md §3.3): register that table (engine +888 // refcount layer + hello validation), then drive the local engine off a rewritten AST889 // whose `count` relationships read it with a plain projected join. The remote query stays890 // un-rewritten — the server always emits the synthetic `__agg_*` rows. A count over a LOCAL891 // child is left a native reduce (L1) — `rewriteAggregates`/`ensureSyntheticTables` skip it.892 this.ensureSyntheticTables(qid, ast);893 // Local first (synchronous empty view), then the server stream hydrates it.894 this.local.registerQuery(qid, this.plainEngineAst(ast));895 if (remote) {896 // A remote query is `unknown` until its first server snapshot lands (hydration); retainRemote897 // attaches it to the sub and sets the lifecycle against the sub's hydration state.898 this.retainRemote(qid, remote, qid, channel);899 } else {900 // No server stream (a purely local AST view — or the local half of a split retain whose901 // remote attaches separately via `retainRemoteQuery`): local data is synchronous, so it is902 // `complete` with nothing to await. A later remote retain flips it back to `unknown` if it903 // attaches an un-hydrated sub.904 this.hydrated.add(qid);905 this.setResultType(qid, "complete");906 }907 }908909 /** Register every synthetic aggregate table `ast` needs that we haven't seen yet: on the910 * local engine (which auto-tracks it for the optimistic rebase loop), on `NormalizedSync`911 * (so its rows refcount/GC by group key), and into the source's expected-schema set (so912 * the server's `hello` — which advertises the same table — passes CRIT#4 validation).913 * Idempotent across queries that share an aggregate definition. */914 private ensureSyntheticTables(qid: QueryId, ast: Ast): void {915 // Idempotent per qid: a re-register of the same query keeps the refcounts balanced.916 if (this.queryAggTables.has(qid)) return;917 let added = false;918 const names: string[] = [];919 // L1: a count over a LOCAL child is a native reduce with no synthetic `__agg_*` base — skip it.920 for (const t of aggTableSchemas(ast, (table) => this.localTables.has(table))) {921 names.push(t.name);922 const prev = this.syntheticRefs.get(t.name) ?? 0;923 this.syntheticRefs.set(t.name, prev + 1);924 if (prev > 0) continue; // another query already materialized this table — just refcount925 this.synthetic.set(t.name, t);926 this.local.registerTable(t.name, { columns: t.columns, primaryKey: t.primaryKey });927 this.sync.registerTable(t.name, t.primaryKey);928 added = true;929 }930 if (names.length) this.queryAggTables.set(qid, names);931 // The optimistic delta (§4) needs each aggregate's child table + group key + filter, which932 // the synthetic schema alone doesn't carry — derive the definitions from the original AST.933 this.overlay.register(collectAggDefs(ast, (t) => this.specs[t]?.columns, (t) => this.localTables.has(t)));934 if (added) this.source.expectClientSchema?.([...this.clientTablesBase, ...this.synthetic.values()]);935 }936937 /** Decrement the refcount of every synthetic table query `qid` referenced; for each one that938 * reaches 0 (no live reader left), remove it from the engine, the refcount layer, and the939 * overlay — so aggregate state is reclaimed, not permanent (§4). Must run AFTER940 * `local.unregisterQuery(qid)` so the engine source has no live connection when941 * `unregisterTable` frees it (the engine refuses otherwise). */942 private releaseSyntheticTables(qid: QueryId): void {943 const names = this.queryAggTables.get(qid);944 if (!names) return;945 this.queryAggTables.delete(qid);946 let removed = false;947 for (const name of names) {948 const next = (this.syntheticRefs.get(name) ?? 1) - 1;949 if (next > 0) {950 this.syntheticRefs.set(name, next);951 continue;952 }953 this.syntheticRefs.delete(name);954 this.local.unregisterTable(name);955 this.sync.unregisterTable(name);956 this.overlay.unregister(name);957 this.synthetic.delete(name);958 removed = true;959 }960 if (removed) this.source.expectClientSchema?.([...this.clientTablesBase, ...this.synthetic.values()]);961 }962963 unregisterQuery(qid: QueryId): void {964 this.roomSwappedViews.delete(qid); // a swapped view's teardown forgets its room backing965 const remoteQid = this.releaseRemote(qid);966 // GC: rows this remote footprint SOLELY referenced fall to refcount 0 → net removes. A qid967 // lives on ONE channel, so at most one gate's dropQuery is non-empty (dropQuery of an968 // unknown qid returns []) — but sweep every gate so this needs no ownership lookup.969 const gcs: [string, Mutation[]][] = [];970 if (remoteQid !== undefined) {971 this.pendingRetargetGc.delete(remoteQid); // the sweep below covers a mid-retarget teardown972 for (const gate of this.gates.values()) {973 gate.buffer = gate.buffer.filter((f) => f.qid !== remoteQid);974 const gc = mapGateDeltas(gate, gate.sync.dropQuery(remoteQid));975 if (gc.length) gcs.push([gate.key, gc]);976 }977 }978 // Tear down the local pipeline+view first so the reconcile cycle below skips it.979 this.local.unregisterQuery(qid);980 // The GC removals must leave BOTH head AND the engine's `sync` baseline. A plain981 // `local.mutate` (a HEAD-only write) leaves them in `sync`, so they look like a pending982 // optimistic REMOVE: the next release's rewind diffs head against sync+D and RESURRECTS983 // them, GC never frees anything, and a later query is served the stale/deleted row984 // forever (CRIT#2). Deliver them as a coherent SERVER delta instead — the same985 // sync-moving boundary the release gate uses — so head and sync both drop the rows,986 // against the SOURCE whose baseline held them.987 for (const [key, gc] of gcs) this.runReconcileCycle(key, gc);988 // The local pipeline is gone (no live conn) and the remote footprint's `__agg` rows were989 // GC'd above, so any synthetic table this was the last reader of can now be freed (§4).990 this.releaseSyntheticTables(qid);991 this.asts.delete(qid);992 this.queryTables.delete(qid);993 this.resultTypes.delete(qid);994 this.hydrated.delete(qid);995 this.pendingState.delete(qid); // §7.2 cache, keyed by the local materialized qid (a monotonic996 // Store counter — re-materialize gets a fresh id, never this one again), so drop it on teardown.997 }998999 /** `channel` as in {@link registerQuery} (G-iii): the gate the remote sub registers on; default1000 * `"daemon"`. This is the split-retain seam G-v's resolve-then-register drives — resolve the1001 * lease, learn `realtime.sourceKey`, `connectSource` it, then retain the query on that channel.1002 * Validated FIRST so a bad channel throws before any synthetic-table refcount moves. */1003 retainRemoteQuery(qid: QueryId, remote: RemoteQuery, localQueryId?: QueryId, ast?: Ast, channel?: string): void {1004 this.requireGate(channel ?? "daemon");1005 if (ast) this.ensureSyntheticTables(qid, ast);1006 this.retainRemote(qid, remote, localQueryId, channel);1007 }10081009 releaseRemoteQuery(qid: QueryId): void {1010 const remoteQid = this.releaseRemote(qid);1011 this.releaseSyntheticTables(qid);1012 if (!this.queryTables.has(qid)) {1013 this.resultTypes.delete(qid);1014 this.hydrated.delete(qid);1015 }1016 if (remoteQid === undefined) return;1017 // A mid-retarget release: the every-gate sweep below IS the deferred old-channel GC1018 // (dropQuery hits the old gate's sync too), so retire the pending record — and its1019 // wrong-channel grace — with it.1020 this.pendingRetargetGc.delete(remoteQid);1021 // Per-gate sweep, like `unregisterQuery`: at most one gate owned this qid's frames/rows.1022 for (const gate of this.gates.values()) {1023 gate.buffer = gate.buffer.filter((f) => f.qid !== remoteQid);1024 const gc = mapGateDeltas(gate, gate.sync.dropQuery(remoteQid));1025 if (gc.length) this.runReconcileCycle(gate.key, gc);1026 }1027 }10281029 /** The Slice I-iv upgrade retarget (§4.1 "Retarget" / the doorbell reaction): move a LIVE1030 * (name, args) sub — every retain of it and every local view it feeds, wholesale — from the1031 * channel it lives on onto `sourceKey`'s (already-`connectSource`d, already-promoted) room1032 * channel, WITHOUT the view ever dropping its rows. Returns the sub's wire `sourceQid` (the1033 * identity the client's renewal loop re-subscribes with).1034 *1035 * Why a dedicated primitive: the one-channel-per-(name,args) invariant ({@link retainRemote}'s1036 * loud throw) is correct — a sub's frames must never split across two cv timelines — so the1037 * upgrade cannot simply retain a second sub on the room and release the daemon one; and the1038 * naive release-then-retain order GCs the daemon sync's rows synchronously (net removes emit,1039 * the view flashes empty) a full ws round trip before the room's seq-0 snapshot refills it.1040 * The cutover is therefore TWO-PHASE around the room's first release:1041 *1042 * 1. NOW (here): unsubscribe the old channel's wire sub, sweep its still-buffered frames for1043 * this qid (their cv timeline continues without the sub — the hello-supersession1044 * precedent), flip `sub.channel`, re-arm `sub.hydrated` (the room's own snapshot is the1045 * cutover point), and register on the room source (its resolver presents the handed1046 * roomToken). The old gate's SYNC rows are deliberately NOT dropped: they keep the view's1047 * plain tables populated through the window — the view still reads them until the swap.1048 * 2. AT THE ROOM'S FIRST RELEASED SNAPSHOT: the reconcile folds the snapshot into the room's1049 * namespaced tables, the release tail SWAPS every local view onto them (302 §4.1,1050 * {@link processSwapIns} — the accepted-flash boundary), and {@link flushRetargetGc}'s1051 * deferred `dropQuery`+reconcile on the OLD gate then GCs the plain-table rows the sub1052 * alone referenced — invisible to the swapped views.1053 *1054 * Idempotent per target channel: a sub already on `sourceKey` returns immediately (the1055 * double-doorbell / re-entrancy guard — one retarget per (query, sourceKey)). Validates before1056 * mutating: a throw here leaves the sub fully daemon-attached (the client's fail-open). */1057 retargetRemoteQuery(remote: RemoteQuery, sourceKey: string): QueryId {1058 const newGate = this.requireGate(sourceKey); // throw loudly BEFORE any sub state moves1059 const key = remoteKey(remote);1060 const sub = this.remoteSubs.get(key);1061 if (!sub) {1062 throw new Error(1063 `optimistic backend: no live sub for query "${remote.name}" — nothing to retarget`,1064 );1065 }1066 if (sub.channel === sourceKey) return sub.sourceQid; // already there — idempotent1067 const oldGate = this.gates.get(sub.channel) ?? this.daemonGate;1068 oldGate.source.unregisterQuery(sub.sourceQid);1069 oldGate.buffer = oldGate.buffer.filter((f) => f.qid !== sub.sourceQid);1070 this.pendingRetargetGc.set(sub.sourceQid, sub.channel);1071 sub.channel = sourceKey;1072 sub.hydrated = false;1073 newGate.source.registerQuery(sub.sourceQid, remote);1074 return sub.sourceQid;1075 }10761077 /** Phase 2 of {@link retargetRemoteQuery}, run at the end of every gate release: once a1078 * retargeted sub's first snapshot has RELEASED on its new channel (`sub.hydrated` re-armed at1079 * retarget, re-set by {@link markSubHydrated} inside this very release), drop the qid's rows1080 * from the OLD gate's sync and reconcile them out — after the room's rows are already applied,1081 * so the winner flip is value-equal (net-zero; see the phase table above). A sub torn down1082 * mid-window was already swept by `releaseRemoteQuery`/`unregisterQuery` (which delete the1083 * record); a vanished record here is pruned defensively. */1084 private flushRetargetGc(gate: SourceGate): void {1085 if (this.pendingRetargetGc.size === 0) return; // every non-upgrade release: structural no-op1086 for (const [sourceQid, oldGateKey] of this.pendingRetargetGc) {1087 const key = this.sourceToRemote.get(sourceQid);1088 const sub = key !== undefined ? this.remoteSubs.get(key) : undefined;1089 if (!sub) {1090 this.pendingRetargetGc.delete(sourceQid);1091 continue;1092 }1093 if (sub.channel !== gate.key || !sub.hydrated) continue; // not this gate / not yet cut over1094 this.pendingRetargetGc.delete(sourceQid);1095 const oldGate = this.gates.get(oldGateKey);1096 if (!oldGate) continue;1097 const gc = mapGateDeltas(oldGate, oldGate.sync.dropQuery(sourceQid));1098 if (gc.length) this.runReconcileCycle(oldGateKey, gc);1099 }1100 }11011102 // --- the §4.2 downgrade: demote → ghost → fence → drop (Slice I-v) ----------------------11031104 /** The I-v downgrade orchestration primitive (§4.2/§7.4, re-expressed by 302 §4.2 as the1105 * SWAP-BACK GATE): retire room `sourceKey` behind the watermark fence. The caller has ALREADY1106 * retargeted every live sub off the channel ({@link retargetRemoteQuery} room→daemon —1107 * validated loudly below) and holds the fence from the api-server's downgrade response1108 * (`finalFlushSeq` = the room's last COMMITTED flush seq; `doc` keys the §4.2 watermark fold,1109 * {@link roomWatermarks}). Steps, in order:1110 *1111 * 1. **Disconnect** the channel ({@link disconnectSource}): handlers detached, gate + buffer1112 * dropped. `nextMid`/`watermark`/processed-outcomes for the domain are KEPT FOREVER (§7.1:1113 * an assigned mid pins its domain; a later re-upgrade of the same doc continues the1114 * sequence — {@link connectSource} attaches a fresh gate and the lmid snapshot max-folds1115 * into the surviving watermark). Disconnecting BEFORE the daemon sub's first release is1116 * load-bearing: it makes {@link flushRetargetGc}'s deferred old-channel GC a no-op (gate1117 * gone ⇒ record deleted, nothing dropped). The room's namespaced tables — and the views1118 * swapped onto them — deliberately stay: frozen at the room's last state, they keep the1119 * document visible while the falling-back follower may still lack the final flush.1120 * Swapping back earlier would show its pre-flush images — the regression §4.2 prevents.1121 * 2. **Ghost + first evaluation**: the record joins {@link ghosts} and is evaluated once1122 * immediately — `finalFlushSeq === 0` (a never-flushed room) with no room-domain pending1123 * drops on the spot, the single-daemon first-frame case.1124 *1125 * In-flight discipline (§7.5): entries with `mid !== null` on `sourceKey` stay PINNED (rule1126 * 2 — never re-route a sent mutation); their resolution arrives via the daemon-carried1127 * ledger+outcome folds (I-iii) and blocks the drop until then. Idempotent per sourceKey (a1128 * second labeled query sharing the room demotes into the existing ghost). */1129 demoteRoomSource(sourceKey: string, doc: string, finalFlushSeq: number): void {1130 if (sourceKey === "daemon") {1131 throw new Error("optimistic backend: the daemon source cannot be demoted");1132 }1133 // Validate FIRST (nothing mutated yet): a live sub still on the channel would silently1134 // starve once the gate detaches — the caller must retarget every sub off the room first.1135 for (const sub of this.remoteSubs.values()) {1136 if (sub.channel === sourceKey) {1137 throw new Error(1138 `optimistic backend: cannot demote ${JSON.stringify(sourceKey)} — query "${sub.remote.name}" is still retained on it (retarget it to the daemon first)`,1139 );1140 }1141 }1142 // Idempotent per sourceKey (co-tenant queries sharing the room demote into the existing1143 // ghost) — but NEVER a bare early-return: each demote carries its own fence, so keep the1144 // NEWEST flush (monotone max — swapping back on an older fence would show pre-flush images),1145 // and disconnect defensively in case a gate re-attached since the ghost was armed (a1146 // down→up→down bounce; {@link connectSource} cancels the ghost on re-upgrade, so this arm1147 // normally finds no gate — but a stale gate left connected would let the next daemon release1148 // GC the room slice out from under the still-swapped views, the §4.2 regression).1149 const existing = this.ghosts.get(sourceKey);1150 if (existing) {1151 this.disconnectSource(sourceKey);1152 existing.finalFlushSeq = Math.max(existing.finalFlushSeq, finalFlushSeq);1153 this.evaluateGhosts();1154 return;1155 }1156 this.disconnectSource(sourceKey); // (1) the channel1157 this.ghosts.set(sourceKey, { doc, finalFlushSeq, stuckReported: false }); // (2)1158 this.evaluateGhosts();1159 }11601161 /** Detach one connected room channel (Slice I-v step 3): the source's handlers are replaced1162 * with no-ops (the {@link OptimisticSource} handler seam is single-registration, so this IS1163 * the detach — a late frame from a dying socket can no longer touch any bookkeeping), its1164 * reserved lmid sub is unregistered, and the gate — buffer, per-source sync, cv watermark —1165 * is dropped from {@link gates}. The DOMAIN state deliberately survives forever:1166 * `nextMid[sourceKey]`, `watermark[sourceKey]`, and the processed-outcome set are untouched1167 * (§7.1 — an assigned mid pins its domain; a re-upgrade must continue, never restart, the mid1168 * sequence; {@link connectSource} then attaches a fresh gate whose lmid snapshot max-folds1169 * into the surviving watermark via {@link foldConfirm}). Closing the underlying transport is1170 * the caller's job. Idempotent (a missing gate is a no-op). */1171 disconnectSource(sourceKey: string): void {1172 if (sourceKey === "daemon") {1173 throw new Error("optimistic backend: the daemon source cannot be disconnected");1174 }1175 const gate = this.gates.get(sourceKey);1176 if (!gate) return;1177 this.gates.delete(sourceKey);1178 gate.source.onNormalized(() => {});1179 gate.source.onProgress(() => {});1180 gate.source.onRestart?.(() => {});1181 gate.source.onMutationOutcome?.(() => {});1182 gate.source.onResync?.(() => {});1183 gate.source.unregisterQuery(LMID_QID);1184 }11851186 /** Register the I-v stuck-downgrade sink — see {@link DowngradeStuckEvent}. One handler (a1187 * later registration replaces it, the {@link onScopeSessions} convention); client.ts maps it1188 * onto the loud anomaly surface. */1189 onDowngradeStuck(handler: (event: DowngradeStuckEvent) => void): void {1190 this.downgradeStuckHandler = handler;1191 }11921193 /** The I-v ghost-drop watcher (§4.2), run after every applied release ({@link applyRelease} —1194 * the seam where {@link roomWatermarks} has just folded and the confirm-drop has just run) and1195 * once at demote time. For each ghost: the fence must be satisfied1196 * (`roomWatermarks[doc] ≥ finalFlushSeq`; 0 is trivially satisfied) AND no SENT room-domain1197 * pending may remain (§7.5 — such entries resolve only through the daemon-carried1198 * outcome/ledger folds; an entry that never reached the room is undecidable, so the ghost1199 * HOLDS and the stuck event fires exactly once, naming the mids). Both satisfied ⇒1200 * {@link dropGhost}. */1201 private evaluateGhosts(): void {1202 if (this.ghosts.size === 0) return; // every non-downgrade release: structural no-op1203 for (const [sourceKey, ghost] of [...this.ghosts]) {1204 // A LIVE gate means the doc re-upgraded — dropping now would dismantle the live room1205 // (un-swap its views, unregister the tables its tableMap renames into). connectSource1206 // cancels the ghost on re-upgrade, so this guard is purely defensive; hold, never drop.1207 if (this.gates.has(sourceKey)) continue;1208 if ((this.roomWatermarks.get(ghost.doc) ?? 0) < ghost.finalFlushSeq) continue; // fence holds1209 const stuck = this.pendingMutations.filter((p) => p.domain === sourceKey && p.mid !== null);1210 if (stuck.length > 0) {1211 if (!ghost.stuckReported) {1212 ghost.stuckReported = true;1213 this.downgradeStuckHandler({ sourceKey, doc: ghost.doc, mids: stuck.map((p) => p.mid as number) });1214 }1215 continue; // hold — never a timeout-retire (§7.5 rule 2)1216 }1217 this.dropGhost(sourceKey);1218 }1219 }12201221 /** Drop one cleared ghost — the 302 §4.2 SWAP-BACK: under the fence the daemon tables are1222 * value-equal-or-ahead of the room's final state, so (1) every view swapped onto the room's1223 * namespaced tables re-registers on its ORIGINAL (daemon-table) AST — visually a no-op, the1224 * Store folds the re-hello as an in-place reset; (2) the namespaced tables unregister (no1225 * reader is left after the swap); (3) ONE daemon reconcile re-invokes the pending set so any1226 * entry whose writes had staged onto the now-gone room tables re-stages onto the daemon tables1227 * (its domain policy stopped naming the dead room when the client dropped it). The whole drop1228 * runs under one commit boundary so the swap and the re-staged predictions notify as ONE step.1229 * After this, a FUTURE upgrade of the same doc registers again from scratch. */1230 private dropGhost(sourceKey: string): void {1231 this.ghosts.delete(sourceKey);1232 this.inOneCommit(() => {1233 for (const [qid, key] of [...this.roomSwappedViews]) {1234 if (key !== sourceKey) continue;1235 this.roomSwappedViews.delete(qid);1236 const ast = this.asts.get(qid);1237 if (ast === undefined) continue;1238 this.local.unregisterQuery(qid);1239 this.local.registerQuery(qid, this.plainEngineAst(ast));1240 }1241 this.unregisterRoomTables(sourceKey);1242 // One daemon reconcile re-stages the pending set onto the surviving tables. Run whenever1243 // any pending exists: unregistering the room tables took their staged copies with the tree.1244 if (this.pendingMutations.length > 0) this.runReconcileCycle("daemon", []);1245 });1246 this.refreshPending(); // the reconcile may have dropped a throwing re-invocation1247 }12481249 /** Unregister room `sourceKey`'s namespaced engine tables and drop the {@link roomTables}1250 * record. Callers must have no view registered on them (the engine refuses otherwise —1251 * loud by design). No-op for an unknown sourceKey. */1252 unregisterRoomTables(sourceKey: string): void {1253 const map = this.roomTables.get(sourceKey);1254 if (!map) return;1255 this.roomTables.delete(sourceKey);1256 for (const engineTable of map.values()) this.local.unregisterTable(engineTable);1257 }12581259 // --- the §4 lifecycle SYSTEM-STREAM retains (Slice I-iii) ------------------------------12601261 /** Retain one minted SYSTEM subscription (RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md §4, Slice1262 * I-iii): a wire sub with NO store view and NO user-visible table. Registered through the same1263 * {@link RemoteSub} bookkeeping as any remote retain — so qid→channel ownership, the overflow1264 * re-subscribe, and refcounted release all work unchanged — but with an EMPTY `localQids` set1265 * (no hydration/resultType coupling) and a {@link systemQids} record telling the release path1266 * which system table this qid's frames carry (`spec.table`) and which scope/doc it was minted1267 * for (the fold's row filter). Its frames then buffer on the channel's gate exactly like1268 * {@link LMID_QID}'s and fold at RELEASE time in {@link foldSystemFrames} — riding the SAME1269 * buffered cv path as the data they co-committed with (fence coherence: an out-of-band1270 * shortcut would break I-ii's co-commit ordering guarantee).1271 *1272 * `channel` defaults to `"daemon"` — the system tables live in the DAEMON store (that is the1273 * point: outcome/ledger/watermark rows must be readable with no room socket alive, §7.11274 * "load-bearing for §7.5"). Idempotence per (table, scope/doc) is the CALLER's job (client.ts1275 * keys its retains on exactly that); a duplicate retain of the SAME remote identity refcounts1276 * like any sub. */1277 retainSystemQuery(retainQid: QueryId, remote: RemoteQuery, spec: SystemStreamSpec, channel = "daemon"): void {1278 const gate = this.requireGate(channel); // throw loudly BEFORE any sub state moves1279 const key = remoteKey(remote);1280 let sub = this.remoteSubs.get(key);1281 if (sub) {1282 if (sub.channel !== channel) {1283 throw new Error(1284 `optimistic backend: system query "${remote.name}" is already retained on channel ${JSON.stringify(sub.channel)} — cannot retain it on ${JSON.stringify(channel)}`,1285 );1286 }1287 sub.refCount++;1288 this.localToRemote.set(retainQid, key);1289 this.remoteRetainToLocal.set(retainQid, undefined);1290 return;1291 }1292 // A fresh sub: deliberately NOT `retainRemote` — its `localQueryId` default would couple this1293 // retain's qid to the view-hydration machinery (`hydrated`/`resultType`), and a system stream1294 // has no view to hydrate.1295 sub = { sourceQid: retainQid, remote, refCount: 1, localQids: new Map(), hydrated: false, channel };1296 this.remoteSubs.set(key, sub);1297 this.sourceToRemote.set(retainQid, key);1298 this.localToRemote.set(retainQid, key);1299 this.remoteRetainToLocal.set(retainQid, undefined);1300 this.systemQids.set(retainQid, { ...spec });1301 gate.source.registerQuery(retainQid, remote);1302 }13031304 /** Release a {@link retainSystemQuery} retain. Refcounted like any sub; the LAST release1305 * unregisters from the owning channel, sweeps its buffered frames, and drops the1306 * {@link systemQids} record. The folded lifecycle STATE (`roomWatermarks`/`scopeSessions`/1307 * processed outcomes) deliberately survives — the fence is monotone truth about the store, not1308 * about the subscription (a re-retained fence must not forget a cleared watermark). */1309 releaseSystemQuery(retainQid: QueryId): void {1310 const remoteQid = this.releaseRemote(retainQid);1311 if (remoteQid === undefined) return; // still refcounted (or unknown)1312 for (const gate of this.gates.values()) {1313 gate.buffer = gate.buffer.filter((f) => f.qid !== remoteQid);1314 }1315 this.systemQids.delete(remoteQid);1316 }13171318 /** Raw CRUD has no optimistic story (§9 replaces it with named mutators). Register a1319 * mutator — even a trivial one — and `invoke` it. */1320 mutate(_mutations: Mutation[]): Promise<void> {1321 return Promise.reject(1322 new Error("optimistic backend: writes go through named mutators — use invoke(name, args)"),1323 );1324 }13251326 /** Direct-commit a batch of LOCAL-only writes (`201-LOCAL-ONLY-TABLES-DESIGN.md` §6 / M2):1327 * straight to the local engine, OUTSIDE the optimistic pending stack — a local table is1328 * untracked, so it never rebases, reverts, or waits on a confirmation. Rejects a synced/tracked1329 * table (the local engine's `writeLocal` is the chokepoint). Reconcile is synchronous and1330 * non-reentrant (A5), so a local write can never interleave with an open server cycle.1331 *1332 * Fires the {@link onLocalWrite} observer AFTER the engine commit but BEFORE subscriber1333 * delivery — i.e. for exactly the batches that passed the M2 guard and committed (the1334 * persistence layer's write-through tap, `207-LOCAL-TABLE-PERSISTENCE-DESIGN.md` §5.1). A1335 * subscriber throwing during delivery re-raises out of this call, but only after the tap has1336 * seen the batch: a committed write can never be invisible to the persistence plane. */1337 writeLocal(mutations: Mutation[]): void {1338 this.local.writeLocal(mutations, () => this.localWriteObserver?.(mutations));1339 }13401341 /** The write-through tap for the local-persistence layer (207 §5.1): `observer` sees every1342 * {@link writeLocal} batch post-commit. One observer (the layer); a later registration1343 * replaces it. The observer must not throw — a persistence failure degrades durability, never1344 * the write path (P9); the layer catches internally. */1345 onLocalWrite(observer: (mutations: Mutation[]) => void): void {1346 this.localWriteObserver = observer;1347 }13481349 /** Apply a REPLICATED local batch (a restore snapshot / a leader `commit` — 207 §5.1): delegates1350 * to the engine's `writeLocal`, so the M2 locality guard still fires (P8 — a corrupt record1351 * naming a synced table dies loudly here), but does NOT invoke the {@link onLocalWrite}1352 * observer — the echo guard is structural, so the persistence layer can never re-enter itself.1353 * `onCommitted` fires post-commit pre-delivery (same anchor as {@link writeLocal}'s tap): the1354 * layer updates its mirror there, so a subscriber throw can never desync mirror from engine. */1355 applyLocalReplica(mutations: Mutation[], onCommitted?: () => void): void {1356 this.local.writeLocal(mutations, onCommitted);1357 }13581359 onEvent(handler: (qid: QueryId, ev: ChangeEvent) => void): void {1360 this.handler = handler;1361 }13621363 onCommitBoundary(handler: (phase: "begin" | "end") => void): void {1364 this.boundaryHandler = handler;1365 }13661367 // --- the named-mutator entry (§9) ----------------------------------------------13681369 /** Bracket a multi-step optimistic apply as ONE notification commit (cross-view-atomic1370 * notification — {@link Backend.onCommitBoundary}). `invoke`/`invokeFolded` apply the prediction1371 * and then reconcile the `__agg` head in TWO `this.local` commits, but they are two halves of one1372 * logical mutation: a relationship-`count` view and the data view it counts must update together.1373 * The Store's `commitDepth` is a counter, so each inner commit's own `begin`/`end` nests under this1374 * outer pair and the Store flushes every affected view (data AND count) once, together, at the1375 * outer `end` — a subscriber re-reading a sibling view then sees post-commit data, never a torn1376 * half. Balanced on throw (the prediction mutator may reject) via the `finally`, so a thrown1377 * prediction never wedges the Store in deferred mode. */1378 private inOneCommit<T>(apply: () => T): T {1379 this.boundaryHandler("begin");1380 try {1381 return apply();1382 } finally {1383 this.boundaryHandler("end");1384 }1385 }13861387 /** Run one client mutator against the staged `tx`, accepting BOTH forms (§ shared mutators):1388 * a plain sync function runs as-is; a shared GENERATOR is driven synchronously — every yielded1389 * write applies to the wasm txn now, every `tx.row` read is resolved against the same staged1390 * state (read-your-writes), the SAME body the API server drives asynchronously. `ctx.user` is1391 * the acting principal (re-read per invoke, stable across a rebase re-invoke). */1392 private runMutator(mutator: ClientMutator, tx: MutationTx, args: unknown): void {1393 if (isGeneratorMutator(mutator)) {1394 const gen = (mutator as (t: IsoTx, a: never, c: MutatorCtx) => MutationGen)(1395 isoTx,1396 args as never,1397 { user: this.user() },1398 );1399 driveMutationSync(gen, {1400 apply: (op) => applyOpToTx(tx, op),1401 read: (table, pk) => tx.row(table, pk),1402 query: (q) => tx.query(q),1403 });1404 } else {1405 (mutator as (t: MutationTx, a: never) => void)(tx, args as never);1406 }1407 }14081409 /** Deal the next wire mid from `domain`'s ledger (RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md1410 * §7.1) and advance that counter. A domain absent from the map starts at 1. Per-domain, so a1411 * client writing through room + daemon concurrently keeps two gapless, non-aliasing sequences.1412 * The client-global `seq` is stamped in the same breath — the ONE cross-domain total order1413 * (confirmation order is per-domain; replay order is client-global). Bundled here so no call1414 * site can deal a mid without its seq. ONE caller discards the seq deliberately: the H-v deopt1415 * flip ({@link handleMutationOutcome}) keeps the entry's ORIGINAL seq — its replay position —1416 * and takes only the fresh mid (the dealSeq bump is harmless: seq consumers order, never1417 * count). */1418 private dealMid(domain: string): { mid: number; seq: number } {1419 const mid = this.nextMid.get(domain) ?? 1;1420 this.nextMid.set(domain, mid + 1);1421 return { mid, seq: ++this.dealSeq };1422 }14231424 // --- the DECLARED router (302 §5: declared, not derived) --------------------------------1425 //1426 // The user declares which mutators are room mutators; the client neither proves, derives,1427 // widens, nor falls back. The declaration reaches this backend as `domainPolicy` — the client1428 // layer resolves (mutator name, args) against its declared realtime mutators and the currently1429 // attached rooms. A misdeclaration fails SOFT (302 §5.1): a daemon-declared mutator touching1430 // room-visible data stages onto the daemon tables while the room-homed view reads the room1431 // tables — no optimistic feedback until the echo relays it a hop later, never a divergence.1432 // The room GATE stays the authoritative backstop: a room-routed mutation the room refuses comes1433 // back as a `mutationOutcome` deopt/reject frame and the H-v machinery below re-enqueues or1434 // surfaces it.14351436 /** The declared confirming stream for one invocation: the `domainPolicy`'s verdict, `"daemon"`1437 * when it abstains. Resolved BEFORE the prediction runs — the domain picks the staging map1438 * (a room domain stages its owned tables onto the room's namespaced twins). */1439 private resolveDomain(name: string, args: unknown): string {1440 return this.domainPolicy(name, args) ?? "daemon";1441 }14421443 /** The staging table map for a `domain`-routed prediction ({@link trackingTx}'s `stage`):1444 * wire table → the room's namespaced engine table for the tables the room owns; identity for1445 * everything else (including the whole map for the daemon domain). */1446 private stagingMap(domain: string): ReadonlyMap<string, string> | undefined {1447 return domain === "daemon" ? undefined : this.roomTables.get(domain);1448 }14491450 /** The PLAIN (daemon-homed) engine AST for `ast` — aggregate relationships rewritten to their1451 * synthetic `__agg_*` reads, no room renames. The ONE form every non-swapped engine1452 * registration uses ({@link registerQuery}, {@link dropGhost}'s swap-back) and the base the1453 * swap-in renames ({@link processSwapIns}). */1454 private plainEngineAst(ast: Ast): Ast {1455 return rewriteAggregates(ast, (t) => this.localTables.has(t));1456 }14571458 /** Mutator names the cross-authority warn below already fired for (once per name). */1459 private readonly warnedCrossAuthority = new Set<string>();14601461 /** 302 §5.1 dev-time guard: a room-DECLARED mutator wrote tables the room does not own. Those1462 * writes staged onto the PLAIN daemon tables (the staging map covers only owned tables), but1463 * the entry confirms on the ROOM stream — and only the room's OWNED tables flush back to the1464 * daemon, so nothing upstream ever echoes them: once the room confirm retires the entry, the1465 * next release's whole-store rewind reverts them for good. The first-party room shell refuses1466 * such a mutation (the §3.3 deopt/reject backstop re-routes it to the daemon), so this warns1467 * for the shapes where that backstop may be absent (a BYO relay) — loud, once, soft (§5.1:1468 * misdeclarations never throw). */1469 private warnCrossAuthorityWrites(name: string, domain: string, touched: ReadonlySet<string>): void {1470 if (domain === "daemon" || this.warnedCrossAuthority.has(name)) return;1471 const map = this.roomTables.get(domain);1472 const staged = new Set(map?.values() ?? []);1473 const outside = [...touched].filter((t) => !staged.has(t));1474 if (outside.length === 0) return;1475 this.warnedCrossAuthority.add(name);1476 console.warn(1477 `[rindle] room mutator "${name}" wrote table(s) ${outside.join(", ")} that room ${JSON.stringify(domain)} does not own` +1478 ` (owned: ${map !== undefined && map.size > 0 ? [...map.keys()].join(", ") : "none"}) — these writes rely on the room` +1479 ` shell's deopt backstop and revert after the room confirm if the shell applies the mutation anyway (302 §5.1).`,1480 );1481 }14821483 /** Run the named client mutator optimistically: the prediction applies to the live1484 * engine now (affected views update synchronously), `(mid, name, args)` joins the1485 * pending stack, and the envelope ships upstream. Returns the assigned `mid`. */1486 invoke(name: string, args: unknown): number {1487 return this.invokeWith(name, args);1488 }14891490 /** {@link invoke} with an optional PINNED confirming domain (H-v): the deopt handshake's1491 * already-retired arm re-invokes the frame's echoed `(name, args)` as a FRESH invocation pinned1492 * to `"daemon"` — an honest re-prediction on the current base, never derived (`pin` bypasses1493 * {@link resolveDomain} entirely, so the router never runs and no Q6 counter moves). Every1494 * other step is `invoke` verbatim: prediction now, capture, drainOverlapping, mid dealt from1495 * the pinned domain's ledger, envelope on its channel. */1496 private invokeWith(name: string, args: unknown, pin?: string): number {1497 const mutator = this.registry[name];1498 if (!mutator) throw new Error(`unknown client mutator: ${name}`);1499 // One commit boundary spans the prediction AND the `__agg`-head reconcile below, so their views1500 // (data + count) flush together rather than tearing across two engine commits.1501 return this.inOneCommit(() => {1502 // The confirming stream is DECLARED (302 §5), so it resolves BEFORE the prediction: the1503 // domain picks the staging map — a room-domain mutator's writes to the room's owned tables1504 // land on the namespaced engine twins the room-homed views read. An H-v deopt re-invocation1505 // pins via `pin` and the policy never runs.1506 const domain = pin ?? this.resolveDomain(name, args);1507 // Apply the prediction. If the mutator throws (client-side validation, a bad read),1508 // the staged write is discarded (the wasm txn is a clean no-op until commit) and the throw1509 // propagates with NO mid consumed — a burnt mid is a permanent server-side gap that1510 // silently refuses every later mutation from this client (#10).1511 const writes: WriteSet = new Map();1512 const reads: ReadLog = { reads: [], queries: [] };1513 const ops: ChildOp[] = [];1514 this.local.writeWith((tx) => {1515 this.runMutator(1516 mutator,1517 trackingTx(tx, writes, this.specs, this.localTables, this.opCollector(ops), false, reads, this.stagingMap(domain)),1518 args,1519 );1520 });1521 // `touched` is DERIVED, never separately populated (§3.2 #1) — see {@link WriteSet}.1522 const touched = new Set(writes.keys());1523 this.warnCrossAuthorityWrites(name, domain, touched);1524 // Flush-on-enqueue (§4.2): a fold whose tables overlap this write must take its mid NOW, BEFORE1525 // this write does, so wire order == local-apply order for any pair that can observe each other1526 // (a read-dependent write reading a folded cell sees the same value optimistically and on the1527 // wire — no snap). Drained folds ship with smaller mids; this write's mid is dealt after.1528 this.drainOverlapping(touched);1529 // The confirming stream's ledger deals the mid and its watermark alone retires the entry1530 // (§7.1). An assigned mid pins its domain forever — a re-invocation never re-routes.1531 const { mid, seq } = this.dealMid(domain);1532 this.pendingMutations.push({ mid, seq, name, args, domain, touched, writes, reads });1533 // The prediction stuck — fold its child ops into the optimistic agg delta and push it onto1534 // the `__agg` head rows (§4). No reset here (this is the §1.3 trivial case, no rewind): the1535 // delta accumulates on top of the prior pending set, and `reconcileAggHead` recomputes each1536 // touched group's head as the absolute `server_base ⊕ delta`.1537 for (const op of ops) this.overlay.observe(op);1538 this.reconcileAggHead();1539 this.refreshPending(); // §7.2: this write now touches its queries' pending axis (NOT ResultType).1540 void this.channelFor(domain).pushMutation({ clientID: this.clientID, mid, name, args });1541 return mid;1542 });1543 }15441545 /** Run a FOLDED invoke (FOLDED-MUTATIONS-DESIGN §8): apply the prediction to the live engine now1546 * (like `invoke`), but collapse a run of same-key invokes into ONE pending entry whose `args`1547 * are overwritten in place, debounce the server write, and ship only the last value. The `mid`1548 * is assigned at flush, not here (§4.1) — so the return is a {@link FoldHandle}, not a mid. */1549 invokeFolded(name: string, opts: FoldOptions, args: unknown): FoldHandle {1550 const mutator = this.registry[name];1551 if (!mutator) throw new Error(`unknown client mutator: ${name}`);1552 const foldKey = `${name}\0${stableJson(opts.key)}`;1553 // One commit boundary spans the prediction AND the `__agg`-head reconcile (see {@link inOneCommit}),1554 // so a folded mutation's list view and count view flush together, never torn across two commits.1555 // The declared domain (302 §5) — resolved up front, like `invoke`'s: it picks the staging1556 // map, the §9.3 cadence, and the provisional confirming stream (the flush re-resolves).1557 const domain = this.resolveDomain(name, args);1558 return this.inOneCommit(() => {1559 // Apply the prediction with the read trap armed (§5): a folded mutator that reads state to1560 // compute its write is non-absorbing and refused. A throw discards the staged write (clean1561 // no-op) and consumes no mid — exactly `invoke`'s guarantee. NO `readLog` here — the trap1562 // path stays byte-for-byte as it was; recording (§3.2 #2) never arms alongside the trap.1563 const writes: WriteSet = new Map();1564 const ops: ChildOp[] = [];1565 try {1566 this.local.writeWith((tx) => {1567 this.runMutator(mutator, trackingTx(tx, writes, this.specs, this.localTables, this.opCollector(ops), true, undefined, this.stagingMap(domain)), args);1568 });1569 } catch (e) {1570 if (e instanceof FoldReadError) {1571 throw new Error(1572 `cannot fold "${name}": it reads state via tx.get/tx.row, so it is not absorbing — folded mutators must be last-writer-wins (FOLDED-MUTATIONS-DESIGN §5)`,1573 );1574 }1575 throw e;1576 }1577 for (const op of ops) this.overlay.observe(op);1578 this.reconcileAggHead();15791580 // `touched` is DERIVED, never separately populated (§3.2 #1) — see {@link WriteSet}.1581 const touched = new Set(writes.keys());1582 this.warnCrossAuthorityWrites(name, domain, touched);1583 const now = this.clock.now();1584 let f = this.folds.get(foldKey);1585 if (f) {1586 // Overwrite the single entry in place — the pending stack does NOT grow (§1 #2). The head1587 // already carries this new prediction (absorbing, last-wins on the cell); the entry holds1588 // only the LATEST args, which is what a rebase re-derives from and what the flush ships.1589 // `domain` too: THIS invocation staged through the freshly-resolved domain's map above, so1590 // a mid-window rebase must re-stage through the same one (the flush re-resolves anyway;1591 // no mid is pinned yet — `entry.mid` is null until flush).1592 f.entry.args = args;1593 f.entry.touched = touched;1594 f.entry.writes = writes;1595 f.entry.domain = domain;1596 f.args = args;1597 this.clock.clearTimeout(f.timer);1598 } else {1599 // §9.3: pick the window's cadence. Routing into a room ⇒ flush at roomDebounceMs so1600 // intermediates stream to the shared head; off the room, the caller's collapse debounce1601 // governs.1602 const inRoom = opts.roomDebounceMs !== undefined && domain !== "daemon";1603 const debounceMs = inRoom ? opts.roomDebounceMs! : opts.debounceMs ?? DEFAULT_FOLD_DEBOUNCE_MS;1604 const maxWaitMs = inRoom ? opts.roomDebounceMs! : opts.maxWaitMs;1605 const entry: PendingMutation = { mid: null, seq: null, name, args, domain, touched, writes, reads: { reads: [], queries: [] } };1606 this.pendingMutations.push(entry);1607 let resolveMid!: (mid: number) => void;1608 const midPromise = new Promise<number>((res) => (resolveMid = res));1609 f = {1610 entry,1611 args,1612 timer: undefined,1613 firstAt: now,1614 debounceMs,1615 maxWaitMs,1616 deferAcrossWrites: opts.deferAcrossWrites ?? false,1617 midPromise,1618 resolveMid,1619 };1620 this.folds.set(foldKey, f);1621 }1622 // Check the elapsed-time threshold on this invocation. There is no independent maxWait1623 // timer; if the threshold is not reached, re-arm the full trailing debounce.1624 if (f.maxWaitMs !== undefined && now - f.firstAt >= f.maxWaitMs) {1625 this.flushFold(foldKey);1626 } else {1627 f.timer = this.clock.setTimeout(() => this.flushFold(foldKey), f.debounceMs);1628 }1629 this.refreshPending();1630 const handle: FoldHandle = { flush: () => this.flushFold(foldKey), mid: f.midPromise };1631 return handle;1632 });1633 }16341635 /** Flush-on-enqueue (§4.2): for each outstanding fold whose touched tables overlap `tables`,1636 * assign its mid NOW and ship it — in creation (insertion) order, so the wire stays gapless. A1637 * `deferAcrossWrites` fold opts out (it keeps deferring, accepting the read-dependent snap). The1638 * incoming write's own fold key (if any) is skipped — it is being folded into, not flushed. */1639 private drainOverlapping(tables: Set<string>, exceptKey?: string): void {1640 // Snapshot the entries first: `flushFold` mutates `this.folds` mid-iteration.1641 for (const [key, f] of [...this.folds]) {1642 if (key === exceptKey || f.deferAcrossWrites) continue;1643 if (intersects(f.entry.touched, tables)) this.flushFold(key);1644 }1645 }16461647 /** Flush one fold (§8): deal its `mid` from `nextMid` (SEND order — never reserved, so gapless1648 * by construction), stamp the entry, ship the envelope with the LATEST args, resolve the handle.1649 * The entry stays on `pendingMutations` (now with a real mid) until the lmid release confirms it. */1650 private flushFold(foldKey: string): void {1651 const f = this.folds.get(foldKey);1652 if (!f) return;1653 this.clock.clearTimeout(f.timer);1654 this.folds.delete(foldKey);1655 // Re-resolve the DECLARED confirming stream from the FINAL args (§7.1) and deal the mid from1656 // that domain's ledger — SEND order, never reserved, so gapless within the domain. The mid1657 // dealt below then pins this domain. (A domain that changed since the window opened — a room1658 // attached or dropped mid-window — re-stages on the next reconcile's re-invocation.)1659 const domain = this.resolveDomain(f.entry.name, f.args);1660 f.entry.domain = domain;1661 const { mid, seq } = this.dealMid(domain);1662 f.entry.mid = mid;1663 f.entry.seq = seq;1664 void this.channelFor(domain).pushMutation({ clientID: this.clientID, mid, name: f.entry.name, args: f.args });1665 f.resolveMid(mid);1666 }16671668 /** The transport a `domain`-confirmed mutation ships on (§7.5 sent-pins-domain: only the1669 * domain's own authority can confirm it, so its channel is the only correct transport). A1670 * domain with NO connected gate ships on the daemon channel — the gate-less configurations1671 * (`__testRelease`-driven tests) and today's entire live path resolve `"daemon"` anyway. */1672 private channelFor(domain: string): OptimisticSource {1673 return (this.gates.get(domain) ?? this.daemonGate).source;1674 }16751676 /** The gate a channel-keyed retain registers through (G-iii registration-time routing). The1677 * channel MUST already be connected (`connectSource`; the daemon is constructor-attached) —1678 * loud by design: a typo'd or not-yet-connected sourceKey must throw at retain time, never1679 * silently register on the daemon and split the query's frames across channels. */1680 private requireGate(channel: string): SourceGate {1681 const gate = this.gates.get(channel);1682 if (!gate) {1683 throw new Error(1684 `optimistic backend: no source connected for channel ${JSON.stringify(channel)} — call connectSource(${JSON.stringify(channel)}, source) before retaining a query on it`,1685 );1686 }1687 return gate;1688 }16891690 /** The channel that owns `sourceQid` — {@link RemoteSub.channel}, the ONE source of truth for1691 * qid routing (G-iii). `undefined` when no sub owns the qid (a harness-delivered raw feed, or1692 * a just-released sub): such frames buffer on whatever gate they arrive at. */1693 private channelOf(sourceQid: QueryId): string | undefined {1694 const key = this.sourceToRemote.get(sourceQid);1695 return key ? this.remoteSubs.get(key)?.channel : undefined;1696 }16971698 // --- the §3.3 deopt handshake, client half (H-v) ---------------------------------1699 //1700 // THE NAMED INVARIANT (Slice I inherits it): **never retire a room-domain entry off a1701 // daemon-carried lmid without outcome resolution.** On the room socket it holds by1702 // construction: every room lmid folds through the room's OWN gate, whose socket also carries1703 // the outcome frames — same-socket ordering puts the frame before the ack, and the reconnect1704 // re-send re-earns a lost frame, so a room-domain entry is only ever retired as a success when1705 // the room really applied it. Slice I's downgrade path breaks that coupling: the doc-scoped1706 // ledger row becomes readable THROUGH THE DAEMON with no room socket alive (§7.1 "load-bearing1707 // for §7.5"), and an lmid adopted that way covers burnt non-applied mids with no frame to say1708 // so — retiring a deopted entry there as a silent success is exactly the lost-write this1709 // handshake exists to prevent. ENFORCED since I-iii by {@link foldSystemFrames}: the I-ii1710 // outcome ROWS (co-committed, in ONE daemon transaction, with the ledger row that covers them)1711 // are synthesized into frames and routed through THIS machine BEFORE the ledger fold advances1712 // the domain watermark — one verdict path for frames and rows, with the processed set as the1713 // cross-release resolved-verdict memory, and absence-under-a-covering-lmid = applied (I-ii's1714 // atomicity makes that the sound default).17151716 /** Record `(domain, mid)` as processed; `false` if it already was (a duplicate frame —1717 * ignore it). FIFO-capped per domain ({@link MAX_PROCESSED_OUTCOMES_PER_DOMAIN}). */1718 private markOutcomeProcessed(domain: string, mid: number): boolean {1719 let mids = this.outcomesProcessed.get(domain);1720 if (!mids) this.outcomesProcessed.set(domain, (mids = new Set()));1721 if (mids.has(mid)) return false;1722 mids.add(mid);1723 while (mids.size > MAX_PROCESSED_OUTCOMES_PER_DOMAIN) {1724 mids.delete(mids.values().next().value as number);1725 }1726 return true;1727 }17281729 /** One `mutationOutcome` frame from `domain`'s channel (H-v — the §3.3 handshake's client1730 * half). The frame arrives OUT-OF-BAND (see {@link attachGate}); the state machine:1731 *1732 * 1. `mid` never issued on `domain` ⇒ ignore (a confused/foreign frame must not invent work).1733 * 2. `(domain, mid)` already processed ⇒ ignore — idempotence under duplicate frames (the1734 * original + a re-send's re-answer; a deopt for a mid whose entry ALREADY FLIPPED also1735 * lands here harmlessly on its second frame).1736 * 3. `kind:"rejected"` ⇒ FINAL. Surface the reason through {@link rejectedHandler} (room-plane1737 * parity with the HTTP queue's callback) and STOP — the drop + snap-back is the EXISTING1738 * failed-mutation machinery: the room burnt the mid, its lmid release retires the entry1739 * per-domain and the reconcile rewinds the prediction, exactly the daemon path's1740 * processed-as-no-op rejection. No new drop path.1741 * 4. `kind:"deopt"`, entry found (pending `(domain, mid)`) ⇒ FLIP IN PLACE: `domain` becomes1742 * `"daemon"`, a fresh daemon mid is dealt and the envelope ships NOW on the daemon channel1743 * ("deal-and-send-now" — the conforming §3.3 re-enqueue: there is no flush machinery for1744 * non-fold entries, so the design's "mid: null until the daemon flush" is satisfied1745 * momentarily inside this call). THE ENTRY'S `seq` IS KEPT — settled (§5.3, commit1746 * 68141096): `seq` is the client-global REPLAY order; re-sequencing would move the entry's1747 * overlay position and change read-dependent SIBLINGS' replay base. Everything else stays1748 * (writes/reads/touched/touchedSources/writeSources — union-never-shrink), the prediction1749 * stays applied (the entry never leaves `pendingMutations`, so no rewind fires), and the1750 * router does NOT re-run nor does `drainOverlapping` (§3.3 re-enqueues, never re-derives;1751 * any open overlapping fold was invoked later and flushes later with a larger mid).1752 * 5. `kind:"deopt"`, entry NOT found ⇒ the burnt-mid confirm won the race, or the frame is a1753 * replay re-answer for an entry a previous session retired (the replay gotcha): re-invoke1754 * the frame's echoed `name`/`args` as a FRESH invocation PINNED to `"daemon"` — an honest1755 * re-prediction on the current base, never a derived route ({@link invokeWith}). A frame1756 * without `name` (not self-contained) has nothing to re-invoke and is dropped; a re-invoke1757 * that THROWS (the base moved from under it) is surfaced through {@link rejectedHandler} —1758 * the mutation is dead with no stream left to confirm it.1759 *1760 * A `"deopt"` bump joins the Q6 routing counters either way (`routing.reasons.deopt`) —1761 * derived-and-deopted routes are visible beside derived successes. */1762 private handleMutationOutcome(domain: string, frame: MutationOutcomeFrame): void {1763 if (frame.mid >= (this.nextMid.get(domain) ?? 1)) return; // never issued here — not ours1764 if (!this.markOutcomeProcessed(domain, frame.mid)) return; // duplicate frame1765 if (frame.kind === "rejected") {1766 const entry = this.pendingMutations.find((p) => p.domain === domain && p.mid === frame.mid);1767 this.rejectedHandler(1768 {1769 clientID: this.clientID,1770 mid: frame.mid,1771 name: entry?.name ?? frame.name ?? "",1772 args: entry !== undefined ? entry.args : frame.args,1773 },1774 frame.reason ?? "mutation rejected",1775 );1776 return;1777 }1778 // kind === "deopt": the room gate refused a declared-room mutation — re-enqueue onto the daemon.1779 const entry = this.pendingMutations.find((p) => p.domain === domain && p.mid === frame.mid);1780 if (entry) {1781 entry.domain = "daemon";1782 // Deal the fresh daemon mid but DISCARD its seq — the entry keeps its own (state-machine1783 // step 4 above; the harmless dealSeq bump is accepted). This is the ONE place a dealt seq1784 // is dropped, so "within one domain seq order == mid order" weakens to "except deopt1785 // re-enqueues" — see the {@link PendingMutation.seq} doc.1786 const { mid } = this.dealMid("daemon");1787 entry.mid = mid;1788 void this.channelFor("daemon").pushMutation({ clientID: this.clientID, mid, name: entry.name, args: entry.args });1789 return;1790 }1791 if (frame.name === undefined) return; // not self-contained — nothing to re-invoke1792 try {1793 this.invokeWith(frame.name, frame.args, "daemon");1794 } catch (err) {1795 this.rejectedHandler(1796 { clientID: this.clientID, mid: frame.mid, name: frame.name, args: frame.args },1797 `deopt re-invocation failed: ${String((err as Error)?.message ?? err)}`,1798 );1799 }1800 }18011802 /** §7.5 rule 3 (H-v): re-send `domain`'s unconfirmed pending envelopes with their ORIGINAL1803 * mids, in mid order, on the domain's own channel. Folds with `mid === null` are excluded —1804 * nothing was ever sent for them (the flush deals their mid). Envelopes are reconstructed from1805 * the pending entries exactly as `invoke` shipped them (`clientID`/`mid`/`name`/`args` —1806 * entries carry everything the wire needs). Idempotent under the domain's ledger: an APPLIED1807 * mid dedups silently and its lmid coverage retires the entry; a NON-APPLIED mid is re-answered1808 * from the shell's recorded-outcome map into {@link handleMutationOutcome}. Confirmed entries1809 * are already gone from `pendingMutations`, so no filter against the watermark is needed. */1810 private resendPending(domain: string): void {1811 const unconfirmed = this.pendingMutations1812 .filter((p) => p.domain === domain && p.mid !== null)1813 .sort((a, b) => a.mid! - b.mid!);1814 if (unconfirmed.length === 0) return;1815 const channel = this.channelFor(domain);1816 for (const p of unconfirmed) {1817 void channel.pushMutation({ clientID: this.clientID, mid: p.mid!, name: p.name, args: p.args });1818 }1819 }18201821 /** Drain every outstanding fold immediately (FOLDED-MUTATIONS-DESIGN §3): the explicit1822 * `app.flushFolds()` and the `beforeunload`/`close` hook. Creation (insertion) order. */1823 flushFolds(): void {1824 for (const key of [...this.folds.keys()]) this.flushFold(key);1825 }18261827 /** A `trackingTx` op collector that records only the ops over a tracked aggregate's child1828 * table (the others can't move any count). Applied to the overlay by the caller AFTER the1829 * mutator succeeds, so a throwing mutator (whose staged write is discarded) leaves no delta. */1830 private opCollector(ops: ChildOp[]): (op: ChildOp) => void {1831 return (op) => {1832 if (this.overlay.hasChild(op.table)) ops.push(op);1833 };1834 }18351836 /** Push the optimistic per-group delta onto the `__agg` head rows (§4):1837 * `target = server_base ⊕ delta`. A head-only write to the (tracked) synthetic table, so it1838 * joins the optimistic layer and is rewound/rebuilt by the reconcile cycle like any1839 * prediction. `server_base` is read from `NormalizedSync` (the authoritative base) — NOT1840 * from head, which already carries the optimistic layer (a torn read). Works standalone (an1841 * ordinary delivery) and inside an open cycle (the write buffers into it). */1842 private reconcileAggHead(): void {1843 const entries = this.overlay.entries();1844 if (entries.length === 0) return;1845 this.local.writeWith((tx) => {1846 for (const e of entries) {1847 const countCol = e.def.groupKeyCols.length; // row is [group…, count]; count is at index k1848 const serverRow = this.sync.baseRow(e.aggTable, e.cells);1849 const serverBase = serverRow ? Number(serverRow[countCol]) : 0;1850 const target = Math.max(0, serverBase + e.n); // a displayed count never goes below 01851 const headRow = tx.get(e.aggTable, e.cells) as WireValue[] | undefined;1852 const wantRow = serverRow !== undefined || target > 0;1853 if (wantRow) {1854 const desired = [...e.cells, target];1855 if (!headRow) tx.add(e.aggTable, desired); // §6.1 optimistic birth (no server group yet)1856 else if (!rowsEqual(headRow, desired)) tx.edit(e.aggTable, headRow, desired);1857 } else if (headRow) {1858 tx.remove(e.aggTable, headRow); // delta took a not-yet-on-server group back to identity1859 }1860 }1861 });1862 this.overlay.pruneZeros();1863 }18641865 /** The SERVER CHANNEL's state for a query (§7): `unknown` while not hydrated, else `complete`.1866 * A pending local mutation no longer moves it — see {@link pending}. */1867 resultType(qid: QueryId): ResultType {1868 return this.resultTypes.get(qid) ?? "complete";1869 }18701871 onResultType(handler: (qid: QueryId, rt: ResultType) => void): void {1872 this.resultTypeHandler = handler;1873 }18741875 __attachDevtoolsServerDeltas(observer: BackendDevObserver): () => void {1876 this.devObservers.add(observer);1877 return () => {1878 this.devObservers.delete(observer);1879 };1880 }18811882 // --- the pending AXIS (§7.2): orthogonal to ResultType -------------------------------18831884 /** Whether any pending mutation (folded or not) touches this query's tables — "is a prediction1885 * pending here?" (FOLDED-MUTATIONS-DESIGN §7.2). Orthogonal to {@link resultType}; this is the1886 * same `queryTables ∩ pending.touched` computation that used to be smuggled into `unknown`. */1887 pending(qid: QueryId): boolean {1888 const tables = this.queryTables.get(qid);1889 if (!tables) return false;1890 return this.pendingMutations.some((p) => intersects(tables, p.touched));1891 }18921893 /** Reactive pending axis (§7.2): fires when a query's pending-ness flips (invoke ↔ confirm), so a1894 * "saving…" affordance clears on its own when `lmid` catches up. */1895 onPending(handler: (qid: QueryId, pending: boolean) => void): void {1896 this.pendingHandler = handler;1897 }18981899 /** The coarse, table-level pending indicator set (§7.2): every table some pending mutation touched. */1900 pendingTables(): Set<string> {1901 const out = new Set<string>();1902 for (const p of this.pendingMutations) for (const t of p.touched) out.add(t);1903 return out;1904 }19051906 /** A read-only snapshot of the optimistic loop for a devtools pane (DEBUG-TOOLS-BROWSER-DESIGN1907 * §4.1). Built fresh per call from state the backend already holds — no new instrumentation, no1908 * mutation. Only ever called by `@rindle/devtools` (imported in dev). */1909 __inspect(): OptimisticInspect {1910 // Reverse-map each FoldRecord's entry → its fold key, so an un-flushed folded pending entry can1911 // be tagged with its debounce window. A flushed fold has already been removed from `folds`1912 // (`flushFold`), so it reads here as an ordinary `mid`-bearing entry — the pane links the1913 // `f:<foldKey>` → `m:<mid>` transition itself.1914 const foldByEntry = new Map<PendingMutation, [string, FoldRecord]>();1915 for (const [foldKey, f] of this.folds) foldByEntry.set(f.entry, [foldKey, f]);1916 const pending: PendingInspect[] = this.pendingMutations.map((p) => {1917 const folded = foldByEntry.get(p);1918 const key = p.mid != null ? `m:${p.mid}` : folded ? `f:${folded[0]}` : `?:${p.name}`;1919 const writes: WriteRecord[] = [];1920 for (const byPk of p.writes.values()) for (const rec of byPk.values()) writes.push(rec);1921 const out: PendingInspect = {1922 key,1923 mid: p.mid,1924 name: p.name,1925 args: p.args,1926 tables: [...p.touched],1927 writes,1928 reads: { reads: [...p.reads.reads], queries: [...p.reads.queries] },1929 };1930 if (folded) {1931 const f = folded[1];1932 out.fold = {1933 foldKey: folded[0],1934 debounceMs: f.debounceMs,1935 maxWaitMs: f.maxWaitMs,1936 deferAcrossWrites: f.deferAcrossWrites,1937 flushed: p.mid != null,1938 };1939 }1940 return out;1941 });1942 return {1943 pending,1944 // Back-compat scalars for the devtools `OptimisticInspect` mirror (unchanged shape): the DAEMON1945 // domain's ledger — the only one in single-domain. Per-domain state is `__inspectDomains()`.1946 confirmedLmid: this.watermark.get("daemon") ?? 0,1947 nextMid: this.nextMid.get("daemon") ?? 1,1948 appliedCv: this.daemonGate.appliedCv,1949 bufferedFrames: this.daemonGate.buffer.length,1950 pendingTables: [...this.pendingTables()],1951 };1952 }19531954 /** Test-only per-domain ledger snapshot (RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md §7.1/§8.5).1955 * Kept separate from {@link __inspect} so the devtools `OptimisticInspect` mirror stays byte-for-1956 * byte identical (daemon-scalar-only). Exposes the per-domain `nextMid`/`watermark` maps plus each1957 * pending entry's confirming domain — the axes the §8.5 ledger-isolation assertion checks. */1958 __inspectDomains(): {1959 nextMid: Record<string, number>;1960 watermark: Record<string, number>;1961 /** Per connected CHANNEL (§5.1): its release watermark + buffered-frame depth — the axis the1962 * gate-isolation assertions read (one source's laggy cvMin must never move the other's). */1963 gates: Record<string, { appliedCv: number; bufferedFrames: number }>;1964 /** Per connected/registered room: its wire-table → engine-table map (302 §2) and which local1965 * view qids are currently swapped onto it (302 §4). */1966 roomTables: Record<string, Record<string, string>>;1967 swappedViews: Record<number, string>;1968 /** The §4 lifecycle plane's folded state (Slice I-iii introspection): the per-doc §4.2 fence1969 * value (`roomWatermarks`, I-v's ghost-drop input), the per-scope §4.1 occupancy map1970 * (`scopeSessions`: scope → client_id → expires_at, I-iv's doorbell input), and the live1971 * I-v ghosts (demoted room sources still awaiting their swap-back fence). */1972 lifecycle: {1973 roomWatermarks: Record<string, number>;1974 scopeSessions: Record<string, Record<string, number>>;1975 ghosts: Record<string, { doc: string; finalFlushSeq: number }>;1976 };1977 pending: { mid: number | null; seq: number | null; name: string; domain: string }[];1978 } {1979 return {1980 nextMid: Object.fromEntries(this.nextMid),1981 watermark: Object.fromEntries(this.watermark),1982 gates: Object.fromEntries(1983 [...this.gates].map(([k, g]) => [k, { appliedCv: g.appliedCv, bufferedFrames: g.buffer.length }]),1984 ),1985 roomTables: Object.fromEntries(1986 [...this.roomTables].map(([k, m]) => [k, Object.fromEntries(m)]),1987 ),1988 swappedViews: Object.fromEntries(this.roomSwappedViews),1989 lifecycle: {1990 roomWatermarks: Object.fromEntries(this.roomWatermarks),1991 scopeSessions: Object.fromEntries(1992 [...this.scopeSessions].map(([scope, sessions]) => [scope, Object.fromEntries(sessions)]),1993 ),1994 ghosts: Object.fromEntries(1995 [...this.ghosts].map(([k, g]) => [k, { doc: g.doc, finalFlushSeq: g.finalFlushSeq }]),1996 ),1997 },1998 pending: this.pendingMutations.map((p) => ({1999 mid: p.mid,2000 // The client-global deal sequence — the REPLAY order (mids are per-domain, incomparable2001 // across domains; see PendingMutation.seq). The harness asserts send order with this.2002 seq: p.seq,2003 name: p.name,2004 domain: p.domain,2005 })),2006 };2007 }20082009 /** Recompute the pending axis for every query and fire `onPending` on transitions only. Called2010 * from the two points that move the pending set: invoke/invokeFolded (add) and the confirm-drop2011 * (remove) — exactly where `:359`/`:468` used to flip ResultType (§7.3). */2012 private refreshPending(): void {2013 for (const qid of this.queryTables.keys()) {2014 const now = this.pending(qid);2015 if (this.pendingState.get(qid) !== now) {2016 this.pendingState.set(qid, now);2017 this.pendingHandler(qid, now);2018 }2019 }2020 }20212022 // --- the downstream stream (§8.5: buffer, then release coherently — PER GATE, §5.1) ------20232024 private onFrame(gate: SourceGate, qid: QueryId, ev: NormalizedEvent): void {2025 // Ownership is fixed at RETAIN time (G-iii registration-time routing: {@link RemoteSub.channel},2026 // the one source of truth) — a qid lives on the ONE channel its sub registered on. So a frame2027 // arriving on any OTHER gate means the server routed a qid to the wrong channel: a wiring bug —2028 // fail loudly rather than silently splitting one query's frames across two cv timelines. (This2029 // used to be a lazy first-arrival CLAIM; it is now a pure assertion.) A qid with NO sub (a2030 // harness-delivered raw feed) has no owner and buffers on the arriving gate; the per-channel2031 // reserved LMID_QID is exempt — each gate owns its own.2032 if (qid !== LMID_QID) {2033 const owner = this.channelOf(qid);2034 if (owner !== undefined && owner !== gate.key) {2035 // I-iv retarget grace: a frame already in flight from the sub's PREVIOUS channel when2036 // {@link retargetRemoteQuery} moved it (the unsubscribe races the server's last frames)2037 // is stale, not a wiring bug — drop it. The grace window is exactly the deferred-GC2038 // window: {@link flushRetargetGc} deletes the record, and the loud throw is restored.2039 if (this.pendingRetargetGc.get(qid) === gate.key) return;2040 throw new Error(`optimistic backend: qid ${qid} arrived on ${gate.key} but is owned by ${owner}`);2041 }2042 }2043 // System-plane frames (I-iii) are bookkeeping, not view data: like the lmid stream they skip2044 // the devtools server-delta tap (there is no local view to attribute them to).2045 if (qid !== LMID_QID && !this.systemQids.has(qid)) this.emitServerDelta(qid, ev);2046 if (ev.type === "hello") {2047 // A hello is a (re)subscribe = a NEW epoch. The column map below is mutated eagerly, but2048 // data frames are cv-buffered and drained later (the gate's progress). So any frame still2049 // buffered for this qid is from a SUPERSEDED epoch and must NOT be scattered through this2050 // epoch's (possibly changed) map — drop it. This epoch's snapshot, which always follows the2051 // hello, re-hydrates the qid from scratch, so the dropped frames are redundant. Scoped to2052 // this qid: other queries' frames (and the lmid system query's) keep their coherent release.2053 gate.buffer = gate.buffer.filter((f) => f.qid !== qid);2054 this.addServerDependencyTables(qid, ev.tables.map((t) => t.name));2055 // Learn this query's per-table column map (PROJECTION-SUPPORT-DESIGN.md §5.2): map each2056 // advertised column to its base ColId BY NAME. The hello may carry FEWER columns than the2057 // client's schema (a projection) or MORE (an EXPANDED server table mid an2058 // `expand-then-contract` migration) — a column the client lacks maps to `-1`, a DROP2059 // sentinel the sync layer discards while keeping the rest Absent. Register the map so the2060 // sync layer scatters the rows into the shared union; a table whose map is an in-order,2061 // drop-free full width IS '*' (a verbatim full row) and stays unregistered (a synthetic agg2062 // table is unknown here, also '*'). Idempotent across re-hydrate epochs.2063 for (const t of ev.tables) {2064 const full = this.colCounts[t.name];2065 if (full === undefined) continue; // unknown/synthetic table → '*' (full presence)2066 const index = this.colIndex[t.name];2067 const cols = t.columns.map((name) => index?.get(name) ?? -1);2068 // Register a non-trivial map; otherwise revert to '*' — and CLEAR any stale map a prior2069 // epoch left (a server that expanded then contracted back), so the now-exact rows don't2070 // scatter through a `-1`-bearing layout (silent cell corruption).2071 if (cols.length !== full || cols.some((c, i) => c !== i)) gate.sync.registerProjection(qid, t.name, cols);2072 else gate.sync.unregisterProjection(qid, t.name);2073 }2074 return; // envelope validation is the source's job2075 }2076 const cv = ev.cv ?? 0;2077 if (cv <= gate.appliedCv && ev.type === "batch") return; // stale redelivery ON THIS TIMELINE2078 gate.buffer.push({ cv, qid, kind: ev.type, ops: ev.ops, seq: gate.nextSeq++ });2079 if (gate.buffer.length > this.bufferCap) this.overflow(gate);2080 }20812082 private emitServerDelta(sourceQid: QueryId, ev: NormalizedEvent): void {2083 if (!this.devObservers.size) return;2084 for (const qid of this.localQidsForSource(sourceQid)) {2085 for (const o of this.devObservers) o.onServerDelta?.(qid, { format: "normalized", event: ev });2086 }2087 }20882089 private localQidsForSource(sourceQid: QueryId): QueryId[] {2090 const key = this.sourceToRemote.get(sourceQid);2091 const sub = key ? this.remoteSubs.get(key) : undefined;2092 if (!sub) return [sourceQid];2093 const localQids = [...sub.localQids.keys()];2094 return localQids.length ? localQids : [sourceQid];2095 }20962097 /** One gate's release (§5.1 release gate): compute the coherent delta from THIS gate's cv-buffer,2098 * then apply it against the gate's source/domain. Split into {@link computeRelease} (buffer →2099 * delta, lmid → watermark) and {@link applyRelease} (per-source confirm-drop + reconcile) —2100 * N independent gates all feed the ONE apply half; {@link __testRelease} drives it directly. */2101 private onGateProgress(gate: SourceGate, frame: ProgressFrame): void {2102 const { deltas, newlyHydrated, touchedScopes } = this.computeRelease(gate, frame);2103 this.applyRelease(gate.key, deltas, undefined, newlyHydrated);2104 // I-iv phase 2: a retargeted sub whose first ROOM snapshot released just now gets its old2105 // channel's rows GC'd — AFTER the release fully applied, so the winner flip is value-equal2106 // against the freshly-folded room rows (never a remove-before-the-refill).2107 this.flushRetargetGc(gate);2108 // I-iv doorbell events LAST — everything this release carried (data, confirms, the occupancy2109 // fold itself, the retarget cutover) is already applied when the consumer's reaction (an2110 // async re-lease) is kicked off. One event per touched scope, count evaluated at the fold2111 // clock's now (deterministic under an injected clock).2112 if (touchedScopes !== null) {2113 for (const scope of touchedScopes) {2114 this.scopeSessionsHandler({ scope, others: this.otherScopeSessions(scope) });2115 }2116 }2117 }21182119 /** Compute one coherent release from ONE gate's cv-buffer (§5.1) — gate-scoped: its buffer, its2120 * cvMin timeline. Take every buffered frame at `cv ≤ cvMin`, in (cv, arrival) order, and fold2121 * it: the lmid system-query frame advances `watermark[gate.key]` (via {@link foldLmidOps} — the2122 * daemon stream folds "daemon", a room stream folds its own domain); data frames fold through2123 * this SOURCE's cross-query refcount into ONE net base delta — the §1.3 `D`. Returns that delta2124 * plus the set of local views this release JUST hydrated (so their reconcile batch phases as a2125 * `snapshot`). Mutates the gate's buffer/`appliedCv`, hydration, and the gate's domain2126 * watermark; the pending set and the reconcile are {@link applyRelease}'s job. */2127 private computeRelease(2128 gate: SourceGate,2129 frame: ProgressFrame,2130 ): { deltas: Mutation[]; newlyHydrated: Set<QueryId> | null; touchedScopes: Set<string> | null } {2131 // Snapshot which local views are already hydrated BEFORE this release folds: any that cross into2132 // hydrated below get their first result set as this cycle's batch, which must phase as a snapshot.2133 const wasHydrated = new Set(this.hydrated);2134 const ready = gate.buffer2135 .filter((f) => f.cv <= frame.cvMin)2136 .sort((a, b) => a.cv - b.cv || a.seq - b.seq);2137 gate.buffer = gate.buffer.filter((f) => f.cv > frame.cvMin);2138 // The §4 lifecycle SYSTEM frames fold FIRST, in a FIXED structural category order (Slice2139 // I-iii; see {@link foldSystemFrames} for why the order is load-bearing), then the ordinary2140 // lmid + data frames fold exactly as before. With no system retain the partition is empty and2141 // this release is byte-identical to pre-I-iii. The returned scope set feeds the I-iv doorbell2142 // events `onGateProgress` fires once the WHOLE release has applied.2143 const touchedScopes = this.foldSystemFrames(ready.filter((f) => this.systemQids.has(f.qid)));2144 const muts: Mutation[] = [];2145 for (const f of ready) {2146 if (this.systemQids.has(f.qid)) {2147 // Folded above; a system stream has no store view and MUST NOT enter the sync layer (its2148 // tables are not in the schema) — but its first snapshot still marks the sub hydrated so2149 // the overflow/introspection bookkeeping stays uniform.2150 if (f.kind === "snapshot") this.markSubHydrated(f.qid);2151 continue;2152 }2153 if (f.qid === LMID_QID) {2154 // Confirmation and data of the same commit share a cv, so they release together — each2155 // channel's lmid stream folds into ITS OWN domain's watermark (§7.1).2156 this.foldLmidOps(f.ops, gate.key);2157 continue;2158 }2159 // A ROOM gate's deltas rename into the room's namespaced tables — and a wire table outside2160 // the registered map is DROPPED (302 §6: context comes from the daemon, one authority per2161 // table; a room's relayed context copy must never enter the store).2162 muts.push(2163 ...mapGateDeltas(gate, f.kind === "snapshot" ? gate.sync.rehydrate(f.qid, f.ops) : gate.sync.applyBatch(f.qid, f.ops)),2164 );2165 // A query's first released snapshot is its hydration point — even an empty one (0 rows is an2166 // authoritative answer): lift every local view this sub feeds out of `unknown` (loading).2167 if (f.kind === "snapshot") this.markSubHydrated(f.qid);2168 }2169 gate.appliedCv = Math.max(gate.appliedCv, frame.cvMin);2170 let newlyHydrated: Set<QueryId> | null = null;2171 for (const qid of this.hydrated) {2172 if (!wasHydrated.has(qid)) (newlyHydrated ??= new Set()).add(qid);2173 }2174 return { deltas: muts, newlyHydrated, touchedScopes };2175 }21762177 /** Apply one released delta against `sourceKey`'s domain (§7.2 per-domain confirm-drop + the §1.32178 * reconcile cycle). `watermarkUpdate`, when given, advances `watermark[sourceKey]` first — the2179 * hook a per-source lmid confirm rides on (the daemon path folds its watermark in2180 * {@link computeRelease} and passes `undefined`). Then: drop every pending entry its OWN domain's2181 * watermark now covers (a room confirm can never retire a daemon entry, and vice-versa — the §7.12182 * ledger-collision fix), and run the reconcile cycle against `sourceKey` when the base delta or the2183 * pending set changed. `newlyHydrated` stamps the initial-hydration batch as a catch-up. */2184 private applyRelease(2185 sourceKey: string,2186 deltas: Mutation[],2187 watermarkUpdate?: number,2188 newlyHydrated: Set<QueryId> | null = null,2189 ): void {2190 if (watermarkUpdate !== undefined) {2191 this.watermark.set(sourceKey, Math.max(this.watermark.get(sourceKey) ?? 0, watermarkUpdate));2192 }2193 // Drop confirmed pending (§1.3 step 5's bookkeeping half), PER DOMAIN: an entry is retired only2194 // when ITS domain's watermark reaches its mid — so two concurrent streams never alias one counter2195 // (§7.1). A failed mutation drops the same way (the release carries no effects, so the rewind snaps2196 // the prediction back). An UNFLUSHED fold (`mid == null`) is never confirmable — retained until its2197 // flush stamps a real mid (FOLDED-MUTATIONS-DESIGN §4.1), regardless of any domain's watermark.2198 // H-v NOTE — retiring here treats coverage as SUCCESS, which for a room domain is sound only2199 // because outcome resolution ALWAYS precedes the coverage that retires: on the room socket2200 // the outcome frames outrun the lmid acks (same-socket ordering + the resync re-send), and on2201 // the daemon-carried path (I-iii) `foldSystemFrames` routes the co-committed outcome ROWS2202 // through handleMutationOutcome BEFORE the ledger fold advances the watermark this filter2203 // reads — a deopted entry has already flipped off the domain by the time its burnt mid is2204 // covered, either way (the named invariant above handleMutationOutcome).2205 const before = this.pendingMutations.length;2206 this.pendingMutations = this.pendingMutations.filter(2207 (p) => p.mid === null || p.mid > (this.watermark.get(p.domain) ?? 0),2208 );2209 const pendingChanged = this.pendingMutations.length !== before;22102211 // The reconcile cycle — only when something can have changed: a base delta to fold in, or a2212 // pending set that shrank (its optimistic layer must rewind out). The batch it emits for any view2213 // that JUST became hydrated is that view's initial result set, so mark those qids so the2214 // local-event forwarder stamps their batch `catchUp` (→ Store phases it `snapshot`).2215 if (deltas.length || pendingChanged) {2216 const emitted = (this.catchUpEmitted = new Set<QueryId>());2217 this.catchUpQids = newlyHydrated;2218 try {2219 this.runReconcileCycle(sourceKey, deltas);2220 } finally {2221 this.catchUpQids = null;2222 this.catchUpEmitted = null;2223 }2224 // Drop the qids the reconcile actually delivered a batch for; the rest folded nothing.2225 if (newlyHydrated) for (const qid of emitted) newlyHydrated.delete(qid);2226 }22272228 // ResultType is the SERVER CHANNEL's state only now (§7): `unknown` while not hydrated, else2229 // `complete` — a pending mutation no longer moves it. The pending axis moves separately.2230 for (const qid of this.queryTables.keys()) {2231 this.setResultType(qid, this.hydrated.has(qid) ? "complete" : "unknown");2232 }2233 // A newly-hydrated query whose reconcile emitted NO batch (0 rows, its whole result already present2234 // via a sibling → 0 net muts, or the reconcile was skipped) still needs a hydration signal, or its2235 // SSR seed never retires and the view freezes. Send an explicit empty catch-up (now that it reads2236 // `complete`, the Store retires the seed and reveals whatever is already in its tree).2237 if (newlyHydrated) {2238 for (const qid of newlyHydrated) this.handler(qid, { type: "batch", events: [], catchUp: true });2239 }2240 this.refreshPending();2241 // The 302 §4.1 swap-in — strictly AFTER the reconcile above folded this release's data, so a2242 // room sub whose first snapshot just released swaps its views onto room tables that already2243 // hold the snapshot (swapping earlier would hydrate them empty). Structural no-op with no2244 // pending swap (every single-domain client).2245 this.processSwapIns();2246 // The I-v ghost-drop watcher (§4.2), LAST: this release's watermark rows have folded2247 // (computeRelease) and its confirm-drop has retired what it covers — exactly the two inputs2248 // the drop condition reads. Structural no-op with no ghost.2249 this.evaluateGhosts();2250 }22512252 /** Test-only per-source release seam (RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md §7.2/§8.5): drive2253 * {@link applyRelease} for `sourceKey` directly — an explicit `watermarkUpdate` (a simulated lmid2254 * confirm for that domain) and `deltas` (a coherent base delta), with no real gate. Lets a harness2255 * exercise a room-domain confirm before the real second lmid stream / per-source gate is wired2256 * (E-iii-b/c). The `__`-prefix marks it a test hook, alongside {@link __inspect}. */2257 __testRelease(sourceKey: string, deltas: Mutation[], watermarkUpdate?: number): void {2258 this.applyRelease(sourceKey, deltas, watermarkUpdate);2259 }22602261 // --- the 302 §4 swap-in ------------------------------------------------------------------22622263 /** Swap every view of each just-hydrated ROOM sub onto the room's namespaced tables (302 §4.1):2264 * re-register the local engine query with the AST's room-owned table references renamed2265 * ({@link remapAstTables}); the Store folds the re-hello as an in-place reset, so the caller's2266 * view reference survives and subscribers see ONE transition. Runs at the applyRelease tail —2267 * the reconcile has already folded the sub's snapshot into the room tables, so the swapped2268 * view hydrates straight to the room state (swapping earlier would flash it empty). The2269 * ORIGINAL ast stays in {@link asts}; the swap-back ({@link dropGhost}) re-registers it.2270 *2271 * This is the accepted-flash boundary (302 §4.1/§7.1): the room's copy may be behind the2272 * daemon rows the view showed a moment ago — accepted by decision, revisit on a real2273 * two-region deploy. */2274 private processSwapIns(): void {2275 if (this.pendingSwapIns.size === 0) return; // every single-domain release: structural no-op2276 const subs = [...this.pendingSwapIns];2277 this.pendingSwapIns.clear();2278 this.inOneCommit(() => {2279 for (const sub of subs) {2280 const map = this.roomTables.get(sub.channel);2281 for (const qid of sub.localQids.keys()) {2282 const ast = this.asts.get(qid);2283 if (ast === undefined) continue;2284 if (map === undefined || map.size === 0) continue; // no owned tables — nothing to swap2285 if (this.roomSwappedViews.get(qid) === sub.channel) continue; // already swapped2286 const rewritten = remapAstTables(this.plainEngineAst(ast), map);2287 this.local.unregisterQuery(qid);2288 this.local.registerQuery(qid, rewritten);2289 this.roomSwappedViews.set(qid, sub.channel);2290 // The pending axis follows the engine tables the view now reads (union — the wire2291 // names stay too, conservatively: a daemon-declared write to a room-visible table is2292 // still an honest "pending elsewhere" signal).2293 const tables = this.queryTables.get(qid);2294 if (tables) for (const t of map.values()) tables.add(t);2295 }2296 }2297 });2298 }22992300 /** Fold `domain`'s lmid system query's released ops (lmid-as-data): the one row's2301 * `last_mutation_id` cell is this client's confirmed high-water mid in that domain — it advances2302 * `watermark[domain]` and, on a fresh session ahead of our issued mids, `nextMid[domain]`. The2303 * daemon stream folds `"daemon"`; a room stream folds its own `"room:doc:X"`; the daemon-carried2304 * §7.1 ledger rows fold through the same {@link foldConfirm} core (Slice I-iii). */2305 private foldLmidOps(ops: NormalizedOp[], domain: string): void {2306 for (const op of ops) {2307 const row = op.op === "add" ? op.row : op.op === "edit" ? op.new : undefined;2308 if (!row) continue; // a remove (client GC) confirms nothing2309 this.foldConfirm(domain, Number(row[1]));2310 }2311 }23122313 /** THE one confirm fold (§7.1/§7.2): advance `watermark[domain]` to `lmid` (monotone max) and,2314 * on a fresh session ahead of our issued mids, adopt `nextMid[domain]`. Shared verbatim by the2315 * per-channel lmid system query ({@link foldLmidOps}) and the daemon-carried room-ledger rows2316 * ({@link foldSystemFrames} — one core so the two paths cannot drift). */2317 private foldConfirm(domain: string, lmid: number): void {2318 if (!Number.isFinite(lmid)) return;2319 const highestIssued = (this.nextMid.get(domain) ?? 1) - 1;2320 if (lmid > highestIssued) {2321 // Only in-flight mutations of THIS domain can contradict its watermark (§7.1: the counters2322 // are independent — a room-domain mutation pending while the daemon's historical lmid2323 // snapshot arrives is a normal fresh-session interleaving, not a second writer). An2324 // unflushed fold (`mid == null`) has issued nothing yet either way: it cannot explain a2325 // confirmed-ahead lmid, and its eventual flush deals from the adopted counter below.2326 if (this.pendingMutations.some((p) => p.domain === domain && p.mid !== null)) {2327 // The server confirmed a mid we never issued while we have mutations in2328 // flight on this domain — two writers on one clientID or corrupted state. Unrecoverable.2329 throw new Error(2330 `optimistic backend: confirmed lmid ${lmid} is ahead of issued mids (${highestIssued})`,2331 );2332 }2333 // A fresh session over a clientID with history: adopt the server's high-water2334 // mark so our next mid continues the sequence instead of colliding below it.2335 this.nextMid.set(domain, lmid + 1);2336 }2337 this.watermark.set(domain, Math.max(this.watermark.get(domain) ?? 0, lmid));2338 }23392340 // --- the §4 lifecycle system-stream folds (Slice I-iii) --------------------------------23412342 /** Fold one release's SYSTEM frames in a FIXED category order — the order is STRUCTURAL (one2343 * function, categories in sequence), because it is the client half of THE NAMED INVARIANT2344 * (§3.3's shipped note; documented above {@link handleMutationOutcome}): **never retire a2345 * room-domain entry off a daemon-carried lmid without outcome resolution.**2346 *2347 * 1. **outcome rows** (`_rindle_room_mutation_outcomes`) — each row for OUR clientID is2348 * synthesized into a {@link MutationOutcomeFrame} and routed through2349 * {@link handleMutationOutcome}, the SAME H-v state machine the room socket's frames use2350 * (one verdict path: frames and rows cannot drift). A deopt flips its pending entry to2351 * the daemon IN PLACE (keep-seq, deal-and-send-now); a rejection surfaces + stays for the2352 * ordinary burnt-mid retire; a duplicate (frame already seen, or the row re-delivered) is2353 * absorbed by the processed set — which doubles as the resolved-verdict memory across2354 * releases (per-domain FIFO, {@link MAX_PROCESSED_OUTCOMES_PER_DOMAIN}, mirroring the2355 * shell's recorded-outcome cap).2356 * 2. **room-ledger rows** (`_rindle_room_client_mutations`) — the FIRST daemon-carried2357 * room-lmid path: OUR row's `last_mutation_id` folds into `watermark[room:<doc>]` via2358 * {@link foldConfirm}. Because step 1 ALREADY resolved every non-applied verdict this2359 * release carries (and earlier releases' verdicts were resolved at their own release),2360 * the confirm-drop that follows in {@link applyRelease} retires only entries whose2361 * outcome is resolution-by-absence — which I-ii's co-commit atomicity defines as APPLIED2362 * (a room flush co-commits the ledger row and every non-applied mid's outcome row in ONE2363 * daemon transaction, so a covering lmid without a row IS the applied verdict).2364 * Processing this category before step 1 is the violation, in two proven directions2365 * (each run break→fail→revert against `test/system_streams.test.ts`): (a) the ledger's2366 * fresh-session `nextMid` ADOPTION must not run before historical outcome rows are2367 * judged — adopted-first, a previous session's retained deopt row passes the2368 * "never-issued" guard and spuriously re-invokes a mutation that session already handled2369 * (a double-apply); (b) the RETIRE must not precede resolution — it does not BECAUSE the2370 * confirm-drop runs in {@link applyRelease}, strictly after this whole function. That2371 * deferral is load-bearing: an "optimization" retiring inline with the watermark fold2372 * retires a deopted entry as a silent success (the exact lost-write H-v exists to2373 * prevent) and mis-attributes a rejected row's reason.2374 * 3. **watermark rows** (`_rindle_room_watermark`) — the §4.2 fence value, max-folded per2375 * doc ({@link roomWatermarks}); I-v's ghost-drop consumer, no reaction here.2376 * 4. **scope-session rows** (`_rindle_scope_sessions`) — the §4.1 occupancy map2377 * ({@link scopeSessions}); I-iv's doorbell consumer, no reaction here.2378 *2379 * Ordinary data ops fold AFTER all of these (the caller's main loop) — outcome/ledger state2380 * must be in place before {@link applyRelease}'s confirm-drop + reconcile consume the release.2381 * Every row is filtered against the retain's {@link SystemStreamSpec} scope/doc AND (for the2382 * client-keyed tables) our own `clientID` — defense in depth: the server predicate may have2383 * been minted doc-only (no `clientId` at lease time), so other clients' rows are expected and2384 * must be ignored, and a row for a doc this retain was not minted for is never folded.2385 *2386 * Returns the scopes category 4 touched (snapshot or ops) — the I-iv doorbell events' input;2387 * `null` when none (every non-lifecycle release). The events themselves fire from2388 * `onGateProgress` AFTER the release applies, never from inside the fold. */2389 private foldSystemFrames(frames: BufferedFrame[]): Set<string> | null {2390 if (frames.length === 0) return null;2391 const byTable = (table: string): { spec: SystemStreamSpec; frame: BufferedFrame }[] =>2392 frames.flatMap((frame) => {2393 const spec = this.systemQids.get(frame.qid);2394 return spec !== undefined && spec.table === table ? [{ spec, frame }] : [];2395 });2396 // (1) outcome rows → the H-v machine, BEFORE any ledger fold (the named invariant).2397 for (const { spec, frame } of byTable(ROOM_MUTATION_OUTCOMES_TABLE)) {2398 for (const op of frame.ops) {2399 const cells = op.op === "add" ? op.row : op.op === "edit" ? op.new : undefined;2400 if (!cells) continue; // a remove is retention pruning (mid ≤ lmid − 512), never a verdict2401 const row = decodeOutcomeRow(cells);2402 if (!row || row.clientId !== this.clientID) continue;2403 if (spec.doc !== undefined && row.doc !== spec.doc) continue;2404 const frameShape: MutationOutcomeFrame = {2405 mid: row.mid,2406 kind: row.kind,2407 ...(row.reason !== undefined ? { reason: row.reason } : {}),2408 ...(row.name !== undefined ? { name: row.name } : {}),2409 ...(row.args !== undefined ? { args: row.args } : {}),2410 };2411 // Release-time invocation is sound here where out-of-band was REQUIRED for the socket2412 // frames (`attachGate`): the socket frame races a buffered lmid ack it must beat, so it2413 // may not wait behind the gate — a ROW cannot race its own release (it and the covering2414 // ledger row co-committed at one cv and fold in THIS function's fixed order). The2415 // machine's steps need nothing from an open release: the flip/reject only move pending2416 // bookkeeping + ship an envelope, and the not-found re-invoke arm runs a fresh prediction2417 // — legal before `applyRelease` opens the reconcile cycle, identical to an app invoke2418 // racing the release.2419 this.handleMutationOutcome(roomDomainKey(row.doc), frameShape);2420 }2421 }2422 // (2) room-ledger rows → the daemon-carried per-domain confirm (outcomes above resolved first).2423 for (const { spec, frame } of byTable(ROOM_CLIENT_MUTATIONS_TABLE)) {2424 for (const op of frame.ops) {2425 const cells = op.op === "add" ? op.row : op.op === "edit" ? op.new : undefined;2426 if (!cells) continue; // a ledger remove confirms nothing (mirrors foldLmidOps)2427 const [doc, clientId, lmid] = cells;2428 if (typeof doc !== "string" || clientId !== this.clientID) continue;2429 if (spec.doc !== undefined && doc !== spec.doc) continue;2430 this.foldConfirm(roomDomainKey(doc), Number(lmid));2431 }2432 }2433 // (3) watermark rows → the monotone §4.2 fence value per doc.2434 for (const { spec, frame } of byTable(ROOM_WATERMARK_TABLE)) {2435 for (const op of frame.ops) {2436 const cells = op.op === "add" ? op.row : op.op === "edit" ? op.new : undefined;2437 if (!cells) continue; // the fence is monotone — a remove never regresses it2438 const [doc, flushSeq] = cells;2439 const seq = Number(flushSeq);2440 if (typeof doc !== "string" || !Number.isFinite(seq)) continue;2441 if (spec.doc !== undefined && doc !== spec.doc) continue;2442 this.roomWatermarks.set(doc, Math.max(this.roomWatermarks.get(doc) ?? 0, seq));2443 }2444 }2445 // (4) scope-session rows → the §4.1 occupancy map (a snapshot REPLACES the scope's map — an2446 // authoritative re-hydrate must drop sessions that aged out while the stream was down; a2447 // batch folds add/edit/remove incrementally). Touched scopes are collected for the I-iv2448 // doorbell events (a snapshot touches its minted scope even with zero ops — an emptied-out2449 // scope is a legitimate 1→0 observation for the transition tracker).2450 let touchedScopes: Set<string> | null = null;2451 const touch = (scope: string): void => {2452 (touchedScopes ??= new Set()).add(scope);2453 };2454 for (const { spec, frame } of byTable(SCOPE_SESSIONS_TABLE)) {2455 if (frame.kind === "snapshot" && spec.scope !== undefined) {2456 this.scopeSessions.set(spec.scope, new Map());2457 touch(spec.scope);2458 }2459 for (const op of frame.ops) {2460 // A remove's identity rides its (full) removed row; add/edit carry the post-image.2461 const cells = op.op === "edit" ? op.new : op.row;2462 const [scope, clientId, expiresAt] = cells;2463 if (typeof scope !== "string" || typeof clientId !== "string") continue;2464 if (spec.scope !== undefined && scope !== spec.scope) continue;2465 let sessions = this.scopeSessions.get(scope);2466 if (!sessions) this.scopeSessions.set(scope, (sessions = new Map()));2467 if (op.op === "remove") {2468 sessions.delete(clientId);2469 } else {2470 const exp = Number(expiresAt);2471 if (Number.isFinite(exp)) sessions.set(clientId, exp);2472 }2473 touch(scope);2474 }2475 }2476 return touchedScopes;2477 }24782479 /** The I-iv occupancy count — THE one rule (§4.1/D7): unexpired (`expires_at >` the fold2480 * clock's now) sessions under `scope` from OTHER clientIDs. Shared by the doorbell events2481 * ({@link onGateProgress}) and the client's registration-time check (a doorbell that folded2482 * BEFORE a candidate registered must still be able to trigger it) so the two can never2483 * disagree. Own-clientID rows never count — a solo client cannot ring its own bell — and2484 * expiry is judged on the injectable {@link FoldClock} (deterministic in a virtual-clock2485 * harness, the folded-oracle discipline). */2486 otherScopeSessions(scope: string): number {2487 const sessions = this.scopeSessions.get(scope);2488 if (!sessions) return 0;2489 const now = this.clock.now();2490 let n = 0;2491 for (const [clientId, expiresAt] of sessions) {2492 if (clientId !== this.clientID && expiresAt > now) n++;2493 }2494 return n;2495 }24962497 /** Register the I-iv doorbell event sink — see {@link ScopeSessionsEvent}. One handler (a later2498 * registration replaces it, the {@link onLocalWrite} convention); client.ts is the consumer. */2499 onScopeSessions(handler: (event: ScopeSessionsEvent) => void): void {2500 this.scopeSessionsHandler = handler;2501 }25022503 /** One §1.3 reconcile cycle: rewind the optimistic layer and fold the coherent SERVER2504 * delta into BOTH head AND the `sync` baseline (`serverBatchBegin`), re-invoke every2505 * still-pending mutator to re-stage the optimistic layer (the rewind un-applied it), then2506 * deliver the coalesced result (`serverBatchEnd`). This is the engine's only sync-moving2507 * boundary — `onProgress` releases and `unregisterQuery`'s GC both go through here so head2508 * and sync never diverge (the §1.2 invariant; CRIT#2). */2509 private runReconcileCycle(_sourceKey: string, serverDeltas: Mutation[]): void {2510 // `_sourceKey` names the authority these `deltas` confirm — `"daemon"` on the live daemon2511 // path (and the GC path), a `room:doc:X` string on a room release, whose deltas already carry2512 // the room's ENGINE table names (the gate's rename/filter). Kept for call-site readability2513 // and tracing only: the engine itself is source-agnostic (302: one authority per table) — its2514 // rewind covers EVERY tracked table and every pending mutation re-invokes below regardless of2515 // which channel released, so NOTHING in this cycle may branch on it.2516 this.local.serverBatchBegin(serverDeltas.map(toServerOp));2517 // The rewind covers EVERY tracked table (302: the engine is source-agnostic — there is no2518 // per-source rewind) — including the `__agg_*` head rows — whichever channel released. So the2519 // optimistic agg delta rebuilds on EVERY cycle, room or daemon: reset here, re-observe from2520 // the re-invoked pending set below, re-apply onto the rewound heads at the end. Gating any of2521 // the three on a daemon-only cycle (the pre-302 per-source-rewind contract) would let a room2522 // release wipe the optimistic `__agg` edits and skip the rebuild — every count() view snaps2523 // back to the server base until the next daemon release. The delta stays sound across2524 // domains: `reconcileAggHead` recomputes each head as the absolute `server_base ⊕ delta`,2525 // and the server base (`this.sync`) only moves on daemon releases.2526 this.overlay.reset();2527 // Sort ALL pending into SEND order (the client-global `seq` ascending, then unflushed folds2528 // last by creation order — the deterministic §4.1 slot; the comparator is explicit, NOT2529 // `(seq ?? ∞) - (seq ?? ∞)` which is `∞ - ∞ = NaN` and corrupts V8's sort). The key MUST be2530 // `seq`, never `mid`: mids are per-domain (§7.1) so mids from different domains are2531 // incomparable — a mid-sort would replay a room mid 1 before a daemon mid 5 that was sent2532 // FIRST, letting a read-dependent mutator re-predict from a base it never saw (confirmation2533 // order is per-domain; replay order is client-global). EVERY entry re-invokes — the engine's2534 // rewind covers every tracked table (302: there is no per-source rewind), so every entry's2535 // staged writes were just un-applied, whichever channel released. Single-domain: seq order ==2536 // mid order (except H-v deopt re-enqueues, which keep their ORIGINAL seq under a later daemon2537 // mid — deliberately, so this very sort replays them at their original overlay position).2538 const order = [...this.pendingMutations].sort((a, b) => {2539 if (a.seq === null && b.seq === null) return 0; // both unflushed → stable creation order2540 if (a.seq === null) return 1; // an unflushed fold sorts after every dealt seq2541 if (b.seq === null) return -1;2542 return a.seq - b.seq;2543 });2544 const dropped = new Set<PendingMutation>();2545 try {2546 for (const p of order) {2547 // NO `readLog` here — recording is armed only on the initial `invoke` (§3.2 #2 note on2548 // `PendingMutation.reads`); a re-invocation's write-set still needs fresh capture (below).2549 // The staging map follows the entry's CURRENT domain — a deopt-flipped or re-routed entry2550 // re-stages onto its new domain's tables here.2551 const writes: WriteSet = new Map();2552 const ops: ChildOp[] = [];2553 try {2554 this.local.writeWith((tx) => {2555 this.runMutator(this.registry[p.name], trackingTx(tx, writes, this.specs, this.localTables, this.opCollector(ops), false, undefined, this.stagingMap(p.domain)), p.args);2556 });2557 } catch {2558 // A re-invocation threw — e.g. a read-dependent mutator whose base row the server2559 // deleted, or one reading a row a now-rejected sibling created. Drop it from pending2560 // (its staged write was discarded); its prediction snaps back on this release. This2561 // local replay failure does not call `onRejected`; authority outcomes arrive separately.2562 // The `finally` below closes the cycle even when a mutator throws.2563 dropped.add(p);2564 continue;2565 }2566 // The re-invocation stuck — fold its child ops into the rebuilt optimistic agg delta.2567 for (const op of ops) this.overlay.observe(op);2568 // The pending footprint is the UNION across invocations: a re-run that no-ops (touched =2569 // {}) must NOT shrink it, else a still-pending mutation reports not-pending and its2570 // pending-axis clear fires early (§7.2). `writes` mirrors this: merge, never replace.2571 for (const t of writes.keys()) p.touched.add(t);2572 mergeWriteSet(p.writes, writes);2573 }2574 // Preserve creation order in the live array (the unflushed-fold sort tiebreak depends on it).2575 if (dropped.size) this.pendingMutations = this.pendingMutations.filter((p) => !dropped.has(p));2576 // Re-apply the optimistic agg delta onto the (rewound) `__agg` head rows — INSIDE the open2577 // cycle, so the writes buffer and coalesce into the one per-query delivery `serverBatchEnd`2578 // makes (and never escape as a separate batch). Every cycle (see the reset above).2579 this.reconcileAggHead();2580 } finally {2581 this.local.serverBatchEnd(); // ALWAYS close the cycle — ONE delivery per affected query.2582 }2583 }25842585 /** ONE channel's authority restarted (a new boot id): it lost all materialization + `cv` state2586 * and its `cv` sequence reset, so previously-released `cv`s no longer bound the new stream. The2587 * source has already re-subscribed every query (reconnect → resync); drop THIS gate's buffer2588 * and `cv` watermark so the fresh, low-`cv` snapshots are RELEASED instead of dropped as stale2589 * (`onFrame`/`computeRelease` gate on `appliedCv`). The OTHER gates are untouched — an2590 * authority restart is per-channel (§5.1). Pending optimistic mutations stay put — they2591 * re-apply on the next reconcile, and the channel's lmid system query's fresh snapshot restores2592 * its domain's confirmation watermark. */2593 private resetGate(gate: SourceGate): void {2594 gate.buffer = [];2595 gate.appliedCv = 0;2596 }25972598 /** The §8.5 escape: ONE gate's buffer outgrew its cap (a pinned `cvMin` under churn on that2599 * channel). Drop everything it buffered and re-register every query on that source — the fresh2600 * snapshots arrive as ordinary frames and the next release re-hydrates via the footprint diff2601 * (the §5.3 path); still-pending optimism re-applies in that cycle. The other gates' buffers2602 * and subscriptions are untouched. */2603 private overflow(gate: SourceGate): void {2604 gate.buffer = [];2605 for (const sub of this.remoteSubs.values()) {2606 // Re-register only the subs THIS channel owns ({@link RemoteSub.channel} — the one source2607 // of truth, G-iii): resubscribing another channel's sub here would fork its stream.2608 if (sub.channel !== gate.key) continue;2609 gate.source.unregisterQuery(sub.sourceQid);2610 gate.source.registerQuery(sub.sourceQid, sub.remote);2611 }2612 // The lmid system query's buffered frames were dropped too — re-subscribe it so a2613 // fresh snapshot restores the confirmation watermark.2614 gate.source.unregisterQuery(LMID_QID);2615 gate.source.registerQuery(LMID_QID, { name: LMID_QUERY_NAME, args: {} });2616 }26172618 private setResultType(qid: QueryId, rt: ResultType): void {2619 if (this.resultTypes.get(qid) === rt) return;2620 this.resultTypes.set(qid, rt);2621 this.resultTypeHandler(qid, rt);2622 }26232624 /** Recompute a query's server-channel state from hydration alone (§7): a pending mutation no2625 * longer affects it. Used when a remote sub attaches to or hydrates a local view. */2626 private recomputeResultType(qid: QueryId): void {2627 this.setResultType(qid, this.hydrated.has(qid) ? "complete" : "unknown");2628 }26292630 /** A remote sub's first snapshot landed: mark it (and every local view it feeds) hydrated, then2631 * lift those views out of `unknown` (loading). A ROOM sub's hydration additionally queues the2632 * 302 §4.1 swap-in — performed at the applyRelease TAIL ({@link processSwapIns}), once the2633 * reconcile has folded this snapshot into the room tables. Idempotent — a re-hydrate snapshot2634 * re-marks harmlessly; a source qid with no sub (the lmid system query) is a no-op. */2635 private markSubHydrated(sourceQid: QueryId): void {2636 const key = this.sourceToRemote.get(sourceQid);2637 if (!key) return;2638 const sub = this.remoteSubs.get(key);2639 if (!sub || sub.hydrated) return;2640 sub.hydrated = true;2641 if (sub.channel !== "daemon" && !this.systemQids.has(sub.sourceQid)) this.pendingSwapIns.add(sub);2642 for (const localQid of sub.localQids.keys()) {2643 this.hydrated.add(localQid);2644 this.recomputeResultType(localQid);2645 }2646 }26472648 /** `channel` (G-iii registration-time routing): the gate the sub registers on — the qid's2649 * ownership is fixed HERE, at retain time (no lazy claim; `onFrame` only asserts it). Default2650 * `"daemon"`, so every channel-less caller is byte-identical to before. */2651 private retainRemote(2652 retainQid: QueryId,2653 remote: RemoteQuery,2654 localQueryId: QueryId | undefined = retainQid,2655 channel = "daemon",2656 ): void {2657 const gate = this.requireGate(channel); // throw loudly BEFORE any sub state moves2658 const key = remoteKey(remote);2659 let sub = this.remoteSubs.get(key);2660 let isNew = false;2661 if (!sub) {2662 sub = { sourceQid: retainQid, remote, refCount: 0, localQids: new Map(), hydrated: false, channel };2663 this.remoteSubs.set(key, sub);2664 this.sourceToRemote.set(sub.sourceQid, key);2665 isNew = true;2666 } else if (sub.channel !== channel) {2667 // A (name,args) sub lives on ONE channel — a second retain naming another is a wiring bug2668 // (it would split the query's frames across two cv timelines). Fail loudly.2669 throw new Error(2670 `optimistic backend: query "${remote.name}" is already retained on channel ${JSON.stringify(sub.channel)} — cannot retain it on ${JSON.stringify(channel)}`,2671 );2672 }2673 sub.refCount++;2674 if (localQueryId !== undefined) {2675 sub.localQids.set(localQueryId, (sub.localQids.get(localQueryId) ?? 0) + 1);2676 // A late-joiner to an already-hydrated sub is immediately hydrated; otherwise this view now2677 // awaits the sub's first snapshot (so a split-path local view registered `complete` flips to2678 // `unknown` here). Then recompute its lifecycle.2679 if (sub.hydrated) {2680 this.hydrated.add(localQueryId);2681 // 302 §4.1 LATE JOIN: a ROOM sub's one-shot swap queue ({@link markSubHydrated}) fired at2682 // its first released snapshot — long gone by now — so a view attaching afterwards must2683 // swap onto the room's namespaced tables HERE, or its engine query stays registered on2684 // the plain daemon tables the room channel never feeds (empty/stale, reported complete,2685 // diverging from its already-swapped siblings forever). The room tables already hold the2686 // released state (hydrated ⇒ folded), so swapping immediately is the ordinary2687 // after-the-data order; processSwapIns skips already-swapped siblings, and2688 // pendingSwapIns is empty outside a release, so exactly this sub's un-swapped views move.2689 if (sub.channel !== "daemon" && !this.systemQids.has(sub.sourceQid)) {2690 this.pendingSwapIns.add(sub);2691 this.processSwapIns();2692 }2693 // FORCE the notify past setResultType's dedup: the labeled split registers the local2694 // half `complete`, then flips the STORE view to `unknown` for the lease window WITHOUT2695 // touching our record — so a complete→complete recompute here would swallow the event2696 // and strand the late-joining view `unknown` forever. Redundant notifies are idempotent2697 // Store-side; a swallowed transition is not recoverable.2698 this.resultTypes.set(localQueryId, "complete");2699 this.resultTypeHandler(localQueryId, "complete");2700 } else {2701 this.hydrated.delete(localQueryId);2702 this.recomputeResultType(localQueryId);2703 }2704 }2705 this.localToRemote.set(retainQid, key);2706 this.remoteRetainToLocal.set(retainQid, localQueryId);2707 // Register on the CHANNEL's source (G-iii): the qid's frames will arrive — and buffer, release,2708 // and overflow — on that channel's own §5.1 gate.2709 if (isNew) gate.source.registerQuery(sub.sourceQid, remote);2710 }27112712 private releaseRemote(retainQid: QueryId): QueryId | undefined {2713 const key = this.localToRemote.get(retainQid);2714 if (!key) return undefined;2715 this.localToRemote.delete(retainQid);2716 const localQueryId = this.remoteRetainToLocal.get(retainQid);2717 this.remoteRetainToLocal.delete(retainQid);2718 const sub = this.remoteSubs.get(key);2719 if (!sub) return undefined;2720 sub.refCount--;2721 if (localQueryId !== undefined) {2722 const refs = (sub.localQids.get(localQueryId) ?? 0) - 1;2723 if (refs > 0) sub.localQids.set(localQueryId, refs);2724 else sub.localQids.delete(localQueryId);2725 }2726 if (sub.refCount > 0) return undefined;2727 // Unregister from the SAME gate's source the retain registered on. Since I-v a gate CAN be2728 // removed ({@link disconnectSource}) — but never with a live sub on it ({@link2729 // demoteRoomSource} validates loudly), so the daemon fallback is purely defensive.2730 (this.gates.get(sub.channel) ?? this.daemonGate).source.unregisterQuery(sub.sourceQid);2731 this.sourceToRemote.delete(sub.sourceQid);2732 this.remoteSubs.delete(key);2733 return sub.sourceQid;2734 }27352736 private addServerDependencyTables(sourceQid: QueryId, names: string[]): void {2737 const key = this.sourceToRemote.get(sourceQid);2738 if (!key) return;2739 const sub = this.remoteSubs.get(key);2740 if (!sub) return;2741 let grew = false;2742 for (const localQid of sub.localQids.keys()) {2743 const tables = this.queryTables.get(localQid);2744 if (!tables) continue;2745 for (const name of names) {2746 if (!tables.has(name)) grew = true;2747 tables.add(name);2748 }2749 }2750 // A newly-learned server dependency may bring a query into a pending mutation's footprint (§7.2);2751 // ResultType is unaffected (server-channel-only, §7).2752 if (grew) this.refreshPending();2753 }2754}27552756// --- helpers -----------------------------------------------------------------------27572758interface RemoteSub {2759 sourceQid: QueryId;2760 remote: RemoteQuery;2761 refCount: number;2762 localQids: Map<QueryId, number>;2763 /** Whether this sub's first server snapshot has been released (drives hydration of its views). */2764 hydrated: boolean;2765 /** The authority channel (gate/source key) this sub registered on — fixed at retain time2766 * (G-iii registration-time routing), `"daemon"` unless a channel-keyed retain named another.2767 * The ONE source of truth for qid→channel ownership: `onFrame`'s wrong-channel assertion and2768 * the per-gate `overflow` both read it (via {@link OptimisticBackend.channelOf} / directly). */2769 channel: string;2770}27712772function remoteKey(remote: RemoteQuery): string {2773 return stableJson([remote.name, remote.args]);2774}27752776function stableJson(value: unknown): string {2777 if (value === null || typeof value !== "object") return JSON.stringify(value);2778 if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`;2779 const obj = value as Record<string, unknown>;2780 return `{${Object.keys(obj)2781 .sort()2782 .map((k) => `${JSON.stringify(k)}:${stableJson(obj[k])}`)2783 .join(",")}}`;2784}27852786function pkColsFromSchema<S extends ColsMap>(schema: Schema<S>): PkCols {2787 const out: PkCols = {};2788 for (const name of Object.keys(schema.tables)) out[name] = tableSpec(schema.tables[name]).primaryKey;2789 return out;2790}27912792/** Per-table column order + pk indices + the insert plan (which columns a full insert requires vs2793 * may omit-to-null) + column kinds (for the `json` stringify), for the keyed `MutationTx` methods. */2794type TableSpecs = Record<2795 string,2796 { columns: string[]; primaryKey: number[]; required: string[]; nullable: ReadonlySet<string>; types: Record<string, ColType> }2797>;27982799function tableSpecsFromSchema<S extends ColsMap>(schema: Schema<S>): TableSpecs {2800 const out: TableSpecs = {};2801 for (const name of Object.keys(schema.tables)) {2802 const { columns, primaryKey } = tableSpec(schema.tables[name]);2803 const { required, nullable } = insertPlan(schema.tables[name]);2804 const types: Record<string, ColType> = {};2805 for (const c of columns) types[c] = schema.tables[name].columns[c].type;2806 out[name] = { columns, primaryKey, required, nullable, types };2807 }2808 return out;2809}28102811/** Each table's FULL column count (the union-row width), from the typed schema. */2812function colCountsFromSchema<S extends ColsMap>(schema: Schema<S>): ColCounts {2813 const out: ColCounts = {};2814 for (const name of Object.keys(schema.tables)) out[name] = tableSpec(schema.tables[name]).columns.length;2815 return out;2816}28172818/** Each table's column name → base ColId, from the typed schema (for mapping a projected hello's2819 * columns back to base positions, PROJECTION-SUPPORT-DESIGN.md §5.2). */2820function colIndexFromSchema<S extends ColsMap>(schema: Schema<S>): Record<string, Map<string, number>> {2821 const out: Record<string, Map<string, number>> = {};2822 for (const name of Object.keys(schema.tables)) {2823 const cols = tableSpec(schema.tables[name]).columns;2824 out[name] = new Map(cols.map((c, i) => [c, i]));2825 }2826 return out;2827}28282829/** Merge a fresh invocation's write-set into a `PendingMutation`'s accumulated one (§3.2 #1, rebase2830 * re-invocation): each pk's value is OVERWRITTEN with the newest image (a later invocation ran2831 * against the base the rebase just replaced, so its view supersedes the earlier one), but a key2832 * present only in `dest` is left alone — the same union-never-shrink rule `touched` already2833 * follows (§7.2: "a re-run that no-ops must NOT shrink it"). */2834function mergeWriteSet(dest: WriteSet, src: WriteSet): void {2835 for (const [table, byPk] of src) {2836 let d = dest.get(table);2837 if (!d) dest.set(table, (d = new Map()));2838 for (const [pkKey, rec] of byPk) d.set(pkKey, rec);2839 }2840}28412842// --- the 302 room-table helpers -------------------------------------------------------28432844/** The namespaced ENGINE table backing wire `table` for room `sourceKey` (302 §2: `room_deck` ≠2845 * `deck` — one authority per table). `@` appears in no schema table name — ENFORCED by2846 * `createSchema`/`extendSchema`'s addTableMeta ban (packages/client/src/schema.ts), so the name2847 * cannot collide with a real table. */2848export function roomEngineTable(table: string, sourceKey: string): string {2849 return `${table}@${sourceKey}`;2850}28512852/** Rename a room gate's released deltas into the room's namespaced tables, DROPPING deltas for2853 * wire tables outside the map (context / unknown — the daemon is their sole authority, 302 §6).2854 * Identity (no copy) for a map-less gate — the daemon path is untouched. */2855function mapGateDeltas(gate: SourceGate, muts: Mutation[]): Mutation[] {2856 const map = gate.tableMap;2857 if (map === undefined) return muts;2858 const out: Mutation[] = [];2859 for (const m of muts) {2860 const engineTable = map.get(m.table);2861 if (engineTable === undefined) continue;2862 out.push({ ...m, table: engineTable });2863 }2864 return out;2865}28662867/** Rename every TABLE reference in a query AST through `map` (302 §2 point 3 — the room-homed2868 * view's rewrite): the root `table`, every `related` subquery, every `correlatedSubquery`2869 * (EXISTS) condition — walking the KNOWN wire-AST shape, never a blind key scan: `start.row` is2870 * keyed by COLUMN name (a schema column literally named `table` must keep its bound value), and2871 * the same goes for any future column-keyed record. Tables absent from the map keep their name —2872 * that is the client-side join across kinds (a room table joined to daemon-owned context,2873 * 201-style). Structural clone; the input AST is never mutated. */2874export function remapAstTables(ast: Ast, map: ReadonlyMap<string, string>): Ast {2875 const walkCond = (c: Condition): Condition => {2876 if (c.type === "and" || c.type === "or") return { ...c, conditions: c.conditions.map(walkCond) };2877 if (c.type === "correlatedSubquery") return { ...c, related: walkSub(c.related) };2878 return c; // "simple" — column refs and literals carry no table reference2879 };2880 const walkSub = (s: CorrelatedSubquery): CorrelatedSubquery => ({ ...s, subquery: walk(s.subquery) });2881 const walk = (a: Ast): Ast => ({2882 ...a,2883 table: map.get(a.table) ?? a.table,2884 ...(a.where !== undefined ? { where: walkCond(a.where) } : {}),2885 ...(a.having !== undefined ? { having: walkCond(a.having) } : {}),2886 ...(a.related !== undefined ? { related: a.related.map(walkSub) } : {}),2887 });2888 return walk(ast);2889}28902891/** Wrap the raw wasm txn as the client `MutationTx`, capturing a pk-granular write-set as it2892 * applies (`writes`, a {@link WriteSet} — table → pk-key → last-write-wins image, §3.2 #1);2893 * `touched` (the pending axis's table-granular Set, §7.2) is derived by the CALLER as2894 * `new Set(writes.keys())`, never populated here. The keyed methods validate column names eagerly:2895 * a typo'd table or column throws with the valid names listed, at the moment the mutator runs.2896 *2897 * With `trapReads` (the FOLDED path, §5), the PUBLIC reads `tx.get`/`tx.row`/`tx.query` throw2898 * `FoldReadError` — a folded mutator that reads state to compute its write is non-absorbing and2899 * refused. The keyed writers (`update`/`upsert`/`insertIgnore`/`delete`) still read internally to2900 * preserve unspecified columns / check pre-existence; that is fold-legal (the trap wraps only the2901 * returned object's `get`/`row`/`query` surface, never the writers' internal probe) — unchanged2902 * by H-ii, which records those probes but arms recording only where the trap never is.2903 *2904 * With `readLog` (recording mode, RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md §3.2 #2) — a SIBLING2905 * of `trapReads`, the two never armed together by any call site — the PUBLIC `tx.get`/`tx.row`2906 * push a `(table, pk, outcome, source?)` {@link ReadRecord} (per-read provenance via the2907 * `provenance` probe, H-ii §3.2 #3), `tx.query` pushes its resolved AST, and (H-ii) the keyed2908 * writers' pre-existence probes record through the same path. Pure capture: it changes no return2909 * value, throws nothing, and is a no-op when `readLog` is omitted. */2910/** Apply one logical {@link MutationOp} (yielded by a shared generator mutator) onto the client's2911 * keyed {@link MutationTx} — the same methods a plain client mutator calls directly. Column2912 * validation, write-set capture, and op collection all happen inside those methods. */2913function applyOpToTx(tx: MutationTx, op: MutationOp): void {2914 switch (op.kind) {2915 case "insert":2916 return tx.insert(op.table, op.row);2917 case "upsert":2918 return tx.upsert(op.table, op.row);2919 case "insertIgnore":2920 return tx.insertIgnore(op.table, op.row);2921 case "update":2922 return tx.update(op.table, op.row);2923 case "delete":2924 return tx.delete(op.table, op.pk);2925 }2926}29272928function trackingTx(2929 tx: WasmWriteTxn,2930 writes: WriteSet,2931 specs: TableSpecs,2932 localTables: Set<string>,2933 onOp?: (op: ChildOp) => void,2934 trapReads = false,2935 readLog?: ReadLog,2936 /** The 302 staging map for a room-DECLARED mutation: wire table → the room's namespaced engine2937 * table for the tables the room owns; identity for everything else. Every raw engine access —2938 * reads and writes — goes through it, so a room mutator reads/writes the room's own state2939 * (its optimistic effects land where the room-homed views look) while its envelope still2940 * ships the wire names. Absent (or a non-owned table) ⇒ the plain table, verbatim. */2941 stage?: ReadonlyMap<string, string>,2942): MutationTx {2943 const spec = (table: string) => {2944 const s = specs[table];2945 if (!s) throw new Error(`unknown table ${JSON.stringify(table)} — tables: ${Object.keys(specs).join(", ")}`);2946 return s;2947 };2948 /** The ENGINE table a wire-named access lands on (302 §2). Schema/column validation always2949 * runs on the WIRE name (the namespaced twin shares the spec). */2950 const staged = (table: string): string => stage?.get(table) ?? table;29512952 // M1 (`201-LOCAL-ONLY-TABLES-DESIGN.md` §6): a replayable mutator is a pure function of2953 // (synced base + args) — it neither READS nor WRITES a local-only table. The server runs the2954 // same mutator from `args` alone and cannot see local tables, so any dependence diverges the2955 // prediction from authority by construction, with no confirmation that can reconcile it. Both2956 // directions are refused at stage time, the same loud shape as the unknown-column throw; local2957 // writes go through `store.writeLocal` instead.2958 const assertNotLocal = (table: string, verb: string): void => {2959 if (localTables.has(table)) {2960 throw new Error(2961 `cannot ${verb} local-only table ${JSON.stringify(table)} inside a mutator — a replayable mutator may not touch local tables (use store.writeLocal; 201-LOCAL-ONLY-TABLES-DESIGN.md §6 / M1).`,2962 );2963 }2964 };29652966 /** Validate `obj`'s keys against the table's columns; require the pk columns; with2967 * `full`, require every NON-nullable column (a nullable column may be omitted and is filled with2968 * `null` in {@link toCells}, design 206 §6.2). */2969 const checkColumns = (table: string, obj: KeyedRow, full: boolean): void => {2970 const s = spec(table);2971 const unknown = Object.keys(obj).filter((k) => !s.columns.includes(k));2972 if (unknown.length) {2973 throw new Error(2974 `unknown column${unknown.length > 1 ? "s" : ""} ${unknown.join(", ")} on ${table} — columns: ${s.columns.join(", ")}`,2975 );2976 }2977 const required = full ? s.required : s.primaryKey.map((i) => s.columns[i]);2978 const missing = required.filter((c) => !(c in obj));2979 if (missing.length) {2980 throw new Error(`missing ${full ? "column" : "primary-key column"}${missing.length > 1 ? "s" : ""} ${missing.join(", ")} on ${table}`);2981 }2982 };29832984 const pkCells = (table: string, obj: KeyedRow): WireValue[] =>2985 spec(table).primaryKey.map((i) => obj[spec(table).columns[i]]);29862987 // pk from the POSITIONAL wire shape (raw cells in schema column order) — the counterpart of2988 // `pkCells` (which reads a KeyedRow) for the raw `add`/`remove`/`edit` writers below (§3.2 #1).2989 const pkFromCells = (table: string, cells: WireValue[]): WireValue[] =>2990 spec(table).primaryKey.map((i) => cells[i]);29912992 // Record (or overwrite) this pk's write for the invocation (§3.2 #1): last-write-wins WITHIN2993 // this invocation — an add-then-edit (or edit-then-edit) of the same pk collapses to its final2994 // image, matching the engine head's own semantics for that pk. The record is replaced with2995 // exactly the arguments given: the CALLERS (`edit`/`remove` below, consulting `prior`) decide2996 // the pre-image per the H-ii coalescing matrix on {@link WriteRecord}. Keyed by the STAGED2997 // (engine) table name, so the pending axis and the write-set match what the engine holds.2998 const recordWrite = (engineTable: string, pk: WireValue[], row: WireValue[] | undefined, oldRow?: WireValue[]): void => {2999 let byPk = writes.get(engineTable);3000 if (!byPk) writes.set(engineTable, (byPk = new Map()));3001 const pkKey = stableJson(pk);3002 // Defensive copies: the wasm binding's returned arrays are not contractually immutable/unique,3003 // so a captured record must not alias a cell array the engine could later reuse or mutate.3004 byPk.set(pkKey, { table: engineTable, pk: [...pk], row: row ? [...row] : undefined, ...(oldRow ? { oldRow: [...oldRow] } : {}) });3005 };30063007 // A full insert row: each cell is `obj[c]`, or `null` for an omitted nullable column (design 2063008 // §6.2); a `json` object is stringified for the engine (`toCell`). Non-nullable columns are3009 // guaranteed present by `checkColumns(full)`.3010 const toCells = (table: string, obj: KeyedRow): WireValue[] => {3011 const s = spec(table);3012 return s.columns.map((c) => toCell(insertCell(obj, c), s.types[c]));3013 };30143015 const toKeyed = (table: string, cells: WireValue[]): KeyedRow => {3016 const out: KeyedRow = {};3017 spec(table).columns.forEach((c, i) => (out[c] = cells[i]));3018 return out;3019 };30203021 // The raw, UN-recorded read primitive behind `getImpl`/`rowImpl` — the M1 local guard + the txn3022 // read, nothing else. Slice B deliberately kept the keyed writers (`update`/`upsert`/3023 // `insertIgnore`/`delete`) on this, un-recorded ("recording is about the PUBLIC read entry3024 // points"). H-ii deliberately REVERSES that: the pre-existence probe each keyed writer BRANCHES3025 // on is genuine value-dependence the §3 routing proof must see — the concrete silent-drop shape3026 // is `update("cards", {id:5,…})` where the client's daemon slice has the row but the room's3027 // footprint lacks it: the room-side update no-ops, the commit "succeeds" with zero effects, the3028 // confirm retires the entry, and the user's edit silently vanishes. The proof can only refuse3029 // that route if the probe is on the record. So the keyed writers now probe through `getImpl`3030 // (recorded like any public read, §3.2 #3); the FOLD trap is unaffected — it wraps only the3031 // returned object's `get`/`row`/`query` surface, so keyed writers stay fold-legal and the3032 // trapped path (where `readLog` is never armed) records nothing, exactly as before.3033 const rawGet = (table: string, pk: WireValue[]) => {3034 assertNotLocal(table, "read");3035 return tx.get(staged(table), pk) as WireValue[] | undefined;3036 };30373038 // Push one {@link ReadRecord} when recording is armed (§3.2 #2/#3): outcome from `row`'s3039 // presence. Pure capture for inspection.3040 const recordRead = (table: string, pk: WireValue[], row: WireValue[] | undefined): void => {3041 if (!readLog) return;3042 readLog.reads.push({3043 table,3044 pk: [...pk],3045 outcome: row === undefined ? "absent" : "present",3046 });3047 };30483049 // The PUBLIC positional read (§3.2 #2) — and, since H-ii, the keyed writers' pre-existence3050 // probe (§3.2 #3, see the `rawGet` note): `rawGet` plus a `readLog` record when recording is3051 // armed. A no-op record when `readLog` is omitted — exactly `rawGet`'s behavior then.3052 const getImpl = (table: string, pk: WireValue[]): WireValue[] | undefined => {3053 const result = rawGet(table, pk);3054 recordRead(table, pk, result);3055 return result;3056 };30573058 // The PUBLIC keyed read (§3.2 #2), the `row` counterpart of `getImpl`.3059 const rowImpl = (table: string, pk: KeyedRow): KeyedRow | undefined => {3060 checkColumns(table, pk, false);3061 const pkc = pkCells(table, pk);3062 const cells = rawGet(table, pkc);3063 recordRead(table, pkc, cells);3064 return cells ? toKeyed(table, cells) : undefined;3065 };30663067 // The pk's existing record from THIS invocation, if any — the coalescing-matrix input for3068 // `edit`/`remove` below (see {@link WriteRecord}). Keyed by the STAGED name like the records.3069 const prior = (table: string, pk: WireValue[]): WriteRecord | undefined =>3070 writes.get(staged(table))?.get(stableJson(pk));3071 const add = (table: string, row: WireValue[]) => {3072 assertNotLocal(table, "write");3073 const t = staged(table);3074 recordWrite(t, pkFromCells(table, row), row);3075 // ChildOps carry the WIRE name (unlike the write-set): the agg overlay's defs are keyed by3076 // the ORIGINAL AST's child tables (`collectAggDefs`), and the `__agg_*` heads it feeds are3077 // shared by plain and swapped views alike — a staged name would silently miss the dispatch3078 // and the optimistic count would lag every room-declared write until its echo.3079 onOp?.({ table, kind: "add", row });3080 tx.add(t, row);3081 };3082 const remove = (table: string, row: WireValue[]) => {3083 assertNotLocal(table, "write");3084 const t = staged(table);3085 const pk = pkFromCells(table, row);3086 // The remove PRE-IMAGE (the H-ii matrix on {@link WriteRecord}): remove-after-edit/-remove3087 // keeps the ORIGINAL captured pre-image (the txn-entry base — the net effect is a remove of3088 // the row the external world last knew, never the edited transient). Otherwise (first touch,3089 // or remove-after-add) the truthful full-width row is the txn-visible one — `tx.get` read3090 // BEFORE the remove stages (read-your-writes: an add of this pk earlier in the SAME3091 // invocation shows through). Falls back to the caller's asserted `row` when the pk is not3092 // resident (a raw remove of an absent row) — a captured remove thus always carries a3093 // full-width pre-image.3094 const oldRow = prior(table, pk)?.oldRow ?? (tx.get(t, pk) as WireValue[] | undefined) ?? row;3095 recordWrite(t, pk, undefined, oldRow);3096 onOp?.({ table, kind: "remove", row }); // wire name — see `add`3097 tx.remove(t, row);3098 };3099 const edit = (table: string, oldRow: WireValue[], newRow: WireValue[]) => {3100 assertNotLocal(table, "write");3101 const t = staged(table);3102 const pk = pkFromCells(table, newRow);3103 // The edit PRE-IMAGE (the H-ii matrix on {@link WriteRecord}). First touch: the txn-visible3104 // row read BEFORE staging, falling back to the caller's asserted `oldRow` when the pk is not3105 // resident (covers the pk-MOVING raw edit — the record is keyed by the NEW pk; the pre-image3106 // carries the OLD row). Edit-after-edit: keep the FIRST pre-image (the txn-entry base).3107 // Edit-after-add / edit-after-remove: the record collapses to a (re-)insert — NO pre-image3108 // (the pk did not pre-exist this invocation's base).3109 const p = prior(table, pk);3110 const pre =3111 p === undefined3112 ? ((tx.get(t, pk) as WireValue[] | undefined) ?? oldRow)3113 : p.row !== undefined && p.oldRow !== undefined3114 ? p.oldRow3115 : undefined;3116 recordWrite(t, pk, newRow, pre);3117 onOp?.({ table, kind: "edit", row: newRow, old: oldRow }); // wire name — see `add`3118 tx.edit(t, oldRow, newRow);3119 };31203121 // The folded read trap (§5): a mutator that reads to compute its write is refused. `() => never`3122 // is assignable to the wider read signatures (extra args ignored, `never` widens to the result).3123 const trapped = (): never => {3124 throw new FoldReadError();3125 };31263127 // One-shot query (203 §5.2): lower the builder to an AST, refuse any local-only table it3128 // reads (M1 — same guard as `rawGet`), and run it over the engine's read-cache fork. The3129 // wasm engine returns the rows already keyed by column name with their materialized3130 // relationship children nested by name (`marshal::caught_node_to_js`), in the query's order —3131 // identical in shape to `view.data` — so this is a pass-through.3132 const runQuery = (q: QueryArg): QueryResultRow[] => {3133 const ast = q.ast();3134 for (const t of collectTables(ast)) assertNotLocal(t, "read");3135 readLog?.queries.push(ast);3136 // A room-declared mutator's one-shot query reads the room's own staged state for the tables3137 // the room owns (the same staging rule as the point reads above).3138 return tx.query(stage !== undefined && stage.size > 0 ? remapAstTables(ast, stage) : ast) as QueryResultRow[];3139 };31403141 return {3142 get: trapReads ? trapped : getImpl,3143 query: trapReads ? trapped : runQuery,3144 add,3145 remove,3146 edit,3147 row: trapReads ? trapped : rowImpl,3148 insert: (table, row) => {3149 checkColumns(table, row, true);3150 add(table, toCells(table, row));3151 },3152 // The keyed writers' pre-existence probes go through `getImpl` — RECORDED reads since H-ii3153 // (§3.2 #3): each writer BRANCHES on the probe, a value-dependence the routing proof must see3154 // (the silent-drop rationale on `rawGet` above). Fold-legal exactly as before (the trap wraps3155 // the public surface above, never these), and byte-identical when recording is off.3156 update: (table, row) => {3157 checkColumns(table, row, false);3158 const current = getImpl(table, pkCells(table, row));3159 if (!current) return; // rebase-friendly: the row may have vanished upstream3160 const s = spec(table);3161 // Named columns overwrite (a `json` object stringified via `toCell`); unnamed keep `current`,3162 // which the engine already holds as a wire value (a json string), so no re-encode.3163 const next = s.columns.map((c, i) => (c in row ? toCell(row[c], s.types[c]) : current[i]));3164 edit(table, current, next);3165 },3166 upsert: (table, row) => {3167 checkColumns(table, row, true);3168 const current = getImpl(table, pkCells(table, row));3169 if (current) edit(table, current, toCells(table, row));3170 else add(table, toCells(table, row));3171 },3172 insertIgnore: (table, row) => {3173 checkColumns(table, row, true);3174 if (!getImpl(table, pkCells(table, row))) add(table, toCells(table, row));3175 },3176 delete: (table, pk) => {3177 checkColumns(table, pk, false);3178 const current = getImpl(table, pkCells(table, pk));3179 if (!current) return; // rebase-friendly no-op3180 remove(table, current);3181 },3182 };3183}31843185function toServerOp(m: Mutation): ServerDeltaOp {3186 if (m.op === "add") return { table: m.table, type: "add", row: m.row };3187 if (m.op === "remove") return { table: m.table, type: "remove", row: m.row };3188 return { table: m.table, type: "edit", row: m.new, old: m.old };3189}31903191/** Every base table a query's AST can draw rows from: the root + every related3192 * subquery + EXISTS subqueries (a conservative deep scan for `table` fields). */3193function collectTables(ast: Ast): Set<string> {3194 const out = new Set<string>();3195 const walk = (v: unknown): void => {3196 if (Array.isArray(v)) {3197 for (const x of v) walk(x);3198 } else if (v && typeof v === "object") {3199 const o = v as Record<string, unknown>;3200 if (typeof o.table === "string") out.add(o.table);3201 for (const k of Object.keys(o)) walk(o[k]);3202 }3203 };3204 walk(ast);3205 return out;3206}32073208function intersects(a: Set<string>, b: Set<string>): boolean {3209 for (const x of b) if (a.has(x)) return true;3210 return false;3211}3212