Source snapshot

packages/remote/src/protocol.ts
Source revision 05d0bf2c2e56 · build details
Source revision: 05d0bf2c2e56.
TypeScript input SHA-256: aabe6cfcc4172b870d5e272142958e9ea8d8784c2aa23133156e5a7ee633318e
Generated 2026-09-04T23:58:25.590Z with TypeScript 6.0.3. Public TypeScript checks and declaration emit passed. Package runtime tests are separate.
1// The flat-change subscription protocol, ported from `src/flat_protocol.rs` + the2// `schema_fp` fingerprint from `src/wire_schema.rs`. This is the wire contract shared by the3// server (`@rindle/server`, which frames via {@link Publisher}) and the client RemoteBackend4// (which validates via {@link Subscriber}).5//6// The §2.2 split (WASM-CLIENT-DESIGN.md): the Rust `Subscriber` owns a `Receiver` and does7// validation *and* folding. Here we keep ONLY validation/sequencing in {@link Subscriber} and8// hand the clean `hello`/`snapshot`/`batch` `ChangeEvent`s up to the core `ArrayView` to fold —9// so one `ArrayView` serves both the local and remote backends.1011import { COMPARATOR_VERSION } from "@rindle/client";12import type { ChangeEvent, FlatChange, Mutation, MutationEnvelope, ProgressFrame, WireSchema } from "@rindle/client";13import type { NormalizedBatch, NormalizedHello } from "./normalized.ts";1415export { COMPARATOR_VERSION };1617// ----------------------------- schema fingerprint (FNV-1a 64) -----------------------------1819const enc = new TextEncoder();20const FNV_OFFSET = 0xcbf29ce484222325n;21const FNV_PRIME = 0x00000100000001b3n;22const MASK = 0xffffffffffffffffn;2324export class Fnv {25  h = FNV_OFFSET;26  private byte(b: number): void {27    this.h = ((this.h ^ BigInt(b & 0xff)) * FNV_PRIME) & MASK;28  }29  u8(v: number): void {30    this.byte(v);31  }32  /** A `u32` little-endian (matching Rust `v.to_le_bytes()`). */33  u32(v: number): void {34    this.byte(v & 0xff);35    this.byte((v >>> 8) & 0xff);36    this.byte((v >>> 16) & 0xff);37    this.byte((v >>> 24) & 0xff);38  }39  /** A length-prefixed string: `u32(byteLen)` then the UTF-8 bytes (disambiguates joins). */40  s(str: string): void {41    const bytes = enc.encode(str);42    this.u32(bytes.length);43    for (const b of bytes) this.byte(b);44  }45}4647function hashLevel(f: Fnv, ws: WireSchema): void {48  f.u8(0x53); // 'S'49  f.u32(ws.columns.length);50  for (const c of ws.columns) f.s(c);51  // PK + sort resolved to column NAMES (semantic identity, independent of indices).52  f.u32(ws.primaryKey.length);53  for (const pk of ws.primaryKey) f.s(ws.columns[pk]);54  f.u32(ws.sort.length);55  for (const [c, asc] of ws.sort) {56    f.s(ws.columns[c]);57    f.u8(asc ? 1 : 0);58  }59  f.u8(ws.singular ? 1 : 0);60  f.u32(ws.relationships.length);61  for (const r of ws.relationships) {62    f.s(r.name);63    if (r.child) {64      f.u8(1);65      hashLevel(f, r.child);66      // Scalar-projection marker (REDUCE-DESIGN.md §9): a presence byte, then — when present67      // — the projected column's NAME in the child schema (index-independent, matching68      // PK/sort). The empty-identity value is NOT hashed (a bare cell can't reproduce its69      // type variant). Must match `hash_level` in `src/wire_schema.rs`.70      if (r.project) {71        f.u8(1);72        f.s(r.child.columns[r.project.col]);73      } else {74        f.u8(0);75      }76    } else {77      f.u8(0);78    }79  }80}8182/** A `WireSchema`'s content fingerprint — FNV-1a 64 over the canonical, length-prefixed byte83 *  stream of `src/wire_schema.rs`, rendered as 16-char lowercase hex (=== Rust `SchemaFp`84 *  `Display`). A string (not a JS number) so the full 64 bits survive JSON without precision loss. */85export function schemaFp(ws: WireSchema): string {86  const f = new Fnv();87  hashLevel(f, ws);88  return f.h.toString(16).padStart(16, "0");89}9091// ----------------------------- the wire frames -----------------------------9293/** The subscription handshake, sent once before any {@link Batch}. */94export interface Hello {95  epoch: number;96  comparatorVersion: number;97  schema: WireSchema;98  schemaFp: string;99}100101/** One transaction's flat changes (or the seq-0 hydrate snapshot). `events` apply in order. */102export interface Batch {103  epoch: number;104  seq: number;105  schemaFp: string;106  events: FlatChange[];107}108109/** The multiplexed wire messages (many queries over one connection, tagged by `queryId`).110 *  `subscribe.mode` selects the serializer (default flat); a normalized subscription gets111 *  `nhello`/`nbatch` back instead of `hello`/`batch` (NORMALIZED-CHANGES-DESIGN.md §6).112 *  Embedded servers receive `{name,args}`. Daemon/serverless deployments can instead receive an113 *  opaque `leaseToken` that the app API server minted after auth + named-query resolution.114 *115 *  The OPTIMISTIC path (OPTIMISTIC-WRITES-DESIGN.md §8) adds `init` (the connection116 *  identifies its stable clientID, so progress frames can carry that client's `lmid`) and117 *  `pushMutation` (one named-mutator envelope up); the server answers with the normalized118 *  frames (`cv`-stamped) plus connection-level `progress` frames. */119export type SubscribeClientMsg =120  | { t: "subscribe"; queryId: number; name: string; args: unknown; mode?: "flat" | "normalized" }121  | { t: "subscribe"; queryId: number; leaseToken: string; mode?: "flat" | "normalized" };122123export type ClientMsg =124  | { t: "init"; clientID: string }125  | SubscribeClientMsg126  | { t: "unsubscribe"; queryId: number }127  | { t: "mutate"; mutations: Mutation[] }128  | { t: "pushMutation"; envelope: MutationEnvelope };129130export type ServerMsg =131  | { t: "hello"; queryId: number; hello: Hello }132  | { t: "batch"; queryId: number; batch: Batch }133  // `bootId` is the daemon's per-process id (same on every frame of one connection). A change134  // across (re)connections means the daemon restarted — it keeps no durable lease/materialization135  // or `cv` state — so the client must force a clean re-hydrate (reset its `cv` watermark).136  | { t: "nhello"; queryId: number; hello: NormalizedHello; bootId?: string }137  | { t: "nbatch"; queryId: number; batch: NormalizedBatch }138  // `code`/`retryable`/`retryAfterMs` classify the error per 101-QUERY-ERRORS §5 (absent from139  // older servers ⇒ terminal): a retryable error (`code:"shed"` from a load-shedding follower,140  // `code:"faulted"` from a worker fault) schedules a backed-off re-subscribe instead of141  // stranding the subscription (FOLLOWER-LAG-SHED §6.3).142  | { t: "queryError"; queryId: number; message: string; code?: string; retryable?: boolean; retryAfterMs?: number }143  | { t: "progress"; frame: ProgressFrame }144  // The follower-affinity ticket (FOLLOWER-AFFINITY-DESIGN.md §3), minted by the fleet follower as145  // the FIRST frame after the ws opens (`rust/rindle-server/src/net.rs` `serve_ws_conn`). Opaque +146  // connection-level (no `queryId`, no `cv`): the client persists it and offers it as a subprotocol147  // on the next connect + forwards it on the lease POST so both legs pin the same follower. Absent148  // on a single/affinity-off daemon (never sent) — old clients drop the unknown `t`.149  | { t: "affinity"; ticket: string }150  // The room deopt handshake's verdict frame (RINDLE-REALTIME-QUERY-ENABLEMENT §3.3, H-iv-b):151  // sent on the author's socket for every NON-applied mutation, BEFORE the lmid ack that burns152  // the mid; `name`/`args` ride `kind:"deopt"` only (self-contained re-invoke). Connection-level153  // and cv-less — the client dispatches it OUT-OF-BAND, never behind the cv buffer. Additive:154  // old clients drop unknown `t`.155  | { t: "mutationOutcome"; mid: number; kind: "deopt" | "rejected"; reason?: string; name?: string; args?: unknown }156  // A per-connection error reply: the server could not process this client's message (bad157  // table/AST, a commit/derive failure). Sent INSTEAD of crashing the process — the connection158  // stays open and every other connection is unaffected. Existing clients ignore unknown `t`.159  | { t: "error"; queryId?: number; message: string };160161// ----------------------------- Publisher (server side) -----------------------------162163/** Sender side: stamps batches with the subscription `epoch` + `schemaFp` and drives the164 *  gap-free seq. The hydrate snapshot is seq 0; increments are seq 1, 2, …. An empty165 *  transaction emits no batch and consumes no seq (so a gap always means a lost batch). */166export class Publisher {167  readonly epoch: number;168  readonly schema: WireSchema;169  readonly schemaFp: string;170  private nextSeq = 0;171172  constructor(epoch: number, schema: WireSchema) {173    this.epoch = epoch;174    this.schema = schema;175    this.schemaFp = schemaFp(schema);176  }177178  hello(): Hello {179    return { epoch: this.epoch, comparatorVersion: COMPARATOR_VERSION, schema: this.schema, schemaFp: this.schemaFp };180  }181182  /** The hydrate snapshot (seq 0). Always emitted, even when empty, so the receiver learns it. */183  snapshot(events: FlatChange[]): Batch {184    return this.emit(events);185  }186187  /** One transaction's events — `null` for an empty transaction (no batch, no seq consumed). */188  commit(events: FlatChange[]): Batch | null {189    return events.length ? this.emit(events) : null;190  }191192  private emit(events: FlatChange[]): Batch {193    const seq = this.nextSeq++;194    return { epoch: this.epoch, seq, schemaFp: this.schemaFp, events };195  }196}197198// ----------------------------- Subscriber (client side) -----------------------------199200export type ProtocolErrorKind = "comparator-mismatch" | "epoch-mismatch" | "schema-mismatch" | "gap";201202/** A protocol violation. All but a duplicate (handled silently) are unrecoverable for the203 *  current subscription — the RemoteBackend re-hydrates under a new epoch. */204export class ProtocolError extends Error {205  readonly kind: ProtocolErrorKind;206  constructor(kind: ProtocolErrorKind, message: string) {207    super(message);208    this.kind = kind;209    this.name = "ProtocolError";210  }211}212213/** Receiver side: validates a frame stream (comparator at `hello`; per batch — epoch match,214 *  schema-fp match, strict in-order seq) and emits the clean `hello`/`snapshot`/`batch`215 *  `ChangeEvent`s. It does NOT fold (the core `ArrayView` does) — the §2.2 split. */216export class Subscriber {217  readonly epoch: number;218  readonly schemaFp: string;219  private readonly emit: (ev: ChangeEvent) => void;220  private phase: "snapshot" | "live" = "snapshot";221  private lastSeq = 0;222223  constructor(hello: Hello, emit: (ev: ChangeEvent) => void) {224    this.emit = emit;225    if (hello.comparatorVersion !== COMPARATOR_VERSION) {226      throw new ProtocolError(227        "comparator-mismatch",228        `comparator version ${hello.comparatorVersion} != ${COMPARATOR_VERSION}`,229      );230    }231    const computed = schemaFp(hello.schema);232    if (computed !== hello.schemaFp) {233      throw new ProtocolError("schema-mismatch", `advertised fp ${hello.schemaFp} != computed ${computed}`);234    }235    this.epoch = hello.epoch;236    this.schemaFp = hello.schemaFp;237    emit({ type: "hello", schema: hello.schema, comparatorVersion: hello.comparatorVersion });238  }239240  /** Apply one incremental batch (or the seq-0 snapshot). Returns `"duplicate"` for an241   *  already-applied seq (discarded — rc ops are not idempotent); throws {@link ProtocolError}242   *  on a gap / epoch / schema mismatch (the caller re-hydrates). */243  apply(batch: Batch): "applied" | "duplicate" {244    if (batch.epoch !== this.epoch) {245      throw new ProtocolError("epoch-mismatch", `expected epoch ${this.epoch}, got ${batch.epoch}`);246    }247    if (batch.schemaFp !== this.schemaFp) {248      throw new ProtocolError("schema-mismatch", `expected fp ${this.schemaFp}, got ${batch.schemaFp}`);249    }250    if (this.phase === "snapshot") {251      if (batch.seq === 0) {252        this.phase = "live";253        this.lastSeq = 0;254        this.emit({ type: "snapshot", adds: batch.events, last: true });255        return "applied";256      }257      throw new ProtocolError("gap", `expected the seq-0 snapshot, got seq ${batch.seq}`);258    }259    const expected = this.lastSeq + 1;260    if (batch.seq < expected) return "duplicate";261    if (batch.seq > expected) {262      throw new ProtocolError("gap", `expected seq ${expected}, got ${batch.seq}`);263    }264    this.lastSeq = batch.seq;265    this.emit({ type: "batch", events: batch.events });266    return "applied";267  }268}269