Source snapshot

packages/remote/src/normalized.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 NORMALIZED subscription protocol — the wire contract shared by the server2// (`@rindle/server`, which relays the native engine's already-stamped frames) and the client3// (`RemoteNormalizedSource`, which validates via {@link NormalizedSubscriber}). The path-free4// twin of `./protocol.ts`: same epoch/seq/gap envelope, normalized payloads (table-tagged5// `NormalizedOp`s), and a slim per-table-schema hello (NORMALIZED-CHANGES-DESIGN.md §3).6//7// As with the flat path, the Subscriber does ONLY validation/sequencing and hands clean8// `NormalizedEvent`s up — the `NormalizedSync` layer (in `@rindle/normalized`) folds them.910import { COMPARATOR_VERSION } from "@rindle/client";11import type { NormalizedEvent, NormalizedOp, NormalizedTableSchema } from "@rindle/client";1213import { Fnv } from "./protocol.ts";14import { ProtocolError } from "./protocol.ts";1516// ----------------------------- normalized fingerprint -----------------------------1718/** A `tables` set's content fingerprint — FNV-1a 64 over the canonical, length-prefixed byte19 *  stream of `rindle-replica::normalize_protocol::normalized_fp` (PK resolved to column NAMES),20 *  as 16-char lowercase hex (=== the Rust hex). `tables` MUST be sorted by name (the server's21 *  `NormalizedPublisher` guarantees it) so the fingerprint is order-stable. */22export function normalizedFp(tables: NormalizedTableSchema[]): string {23  const f = new Fnv();24  f.u32(tables.length);25  for (const t of tables) {26    f.s(t.name);27    f.u32(t.columns.length);28    for (const c of t.columns) f.s(c);29    f.u32(t.primaryKey.length);30    for (const pk of t.primaryKey) f.s(t.columns[pk]); // PK by NAME31  }32  return f.h.toString(16).padStart(16, "0");33}3435/** Is `sub` a subsequence of `sup` by NAME — every entry of `sub` present in `sup`, in order?36 *  Accepts a projection (server ⊆ client) or an expansion (client ⊆ server); a rename/reorder37 *  satisfies neither direction. */38function isSubsequenceByName(sub: readonly string[], sup: readonly string[]): boolean {39  let i = 0;40  for (const name of sup) if (i < sub.length && sub[i] === name) i++;41  return i === sub.length;42}4344/** Validate the tables a `hello` advertises against the CLIENT's own typed schema, by name45 *  (column order + PK indices). `hello.tables` is the query's table subtree, so each one must46 *  be column-compatible with a client table. Throws {@link ProtocolError} `"schema-mismatch"`47 *  on the first unknown table / column-order skew / PK skew — the §3 "drift ⇒ re-subscribe"48 *  guard (CRIT#4). */49function validateAgainstClientSchema(50  serverTables: NormalizedTableSchema[],51  clientTables: NormalizedTableSchema[],52): void {53  const byName = new Map(clientTables.map((t) => [t.name, t]));54  const eq = (a: readonly (string | number)[], b: readonly (string | number)[]) =>55    a.length === b.length && a.every((x, i) => x === b[i]);56  for (const st of serverTables) {57    const ct = byName.get(st.name);58    if (!ct) {59      throw new ProtocolError("schema-mismatch", `server advertises table "${st.name}" not in the client schema`);60    }61    // The server may advertise FEWER columns than the client (a projected query —62    // PROJECTION-SUPPORT-DESIGN.md §5.2/§7) OR MORE (an EXPANDED server table mid an63    // `expand-then-contract` migration — the client drops the columns it doesn't yet have).64    // Both are safe: the client maps every advertised column to a base position BY NAME, so65    // the relative order is preserved either way. Accept when one column list is a SUBSEQUENCE66    // of the other by name — a narrowing (projection) or a widening (expand) — while still67    // rejecting a genuine skew, where NEITHER is a subsequence of the other: a renamed column68    // or a reordered one (the CRIT#4 guard). A server with more columns than the client is no69    // longer drift; without this, `expand-then-contract` is impossible.70    if (!isSubsequenceByName(st.columns, ct.columns) && !isSubsequenceByName(ct.columns, st.columns)) {71      throw new ProtocolError(72        "schema-mismatch",73        `column drift on "${st.name}": server [${st.columns}] is neither a projection nor an expansion of client [${ct.columns}] (column order skew)`,74      );75    }76    // PK compared by NAME (the server forces the PK into every projection, so it is always77    // present): resolve each side's PK indices to names and require equality.78    const serverPk = st.primaryKey.map((i) => st.columns[i]);79    const clientPk = ct.primaryKey.map((i) => ct.columns[i]);80    if (!eq(serverPk, clientPk)) {81      throw new ProtocolError(82        "schema-mismatch",83        `primary-key drift on "${st.name}": client [${clientPk}] != server [${serverPk}]`,84      );85    }86  }87}8889// ----------------------------- the wire frames -----------------------------9091/** The slim normalized handshake (§3), sent once before any {@link NormalizedBatch}. */92export interface NormalizedHello {93  epoch: number;94  comparatorVersion: number;95  tables: NormalizedTableSchema[];96  normalizedFp: string;97}9899/** One committed transaction's normalized ops (or the seq-0 hydrate snapshot). `cv` (the100 *  global commit version the frame reflects) is stamped by optimistic-protocol servers and101 *  rides through to the `NormalizedEvent` (OPTIMISTIC-WRITES-DESIGN.md §8.3). */102export interface NormalizedBatch {103  epoch: number;104  seq: number;105  normalizedFp: string;106  ops: NormalizedOp[];107  cv?: number;108}109110// ----------------------------- Subscriber (client side) -----------------------------111112/** Receiver side: validates a normalized frame stream (comparator at `hello`; per batch —113 *  epoch match, fingerprint match, strict in-order seq) and emits clean `NormalizedEvent`s.114 *  It does NOT fold (the `NormalizedSync` layer does). Mirrors the flat {@link Subscriber}. */115export class NormalizedSubscriber {116  readonly epoch: number;117  readonly normalizedFp: string;118  private readonly emit: (ev: NormalizedEvent) => void;119  private phase: "snapshot" | "live" = "snapshot";120  private lastSeq = 0;121122  /**123   * @param hello         the server's normalized handshake.124   * @param emit          clean-event sink (the `NormalizedSync` fold).125   * @param clientTables  the CLIENT's own typed per-table schemas (all tables). When given,126   *   each table the hello advertises is validated by NAME against it (column order + PK127   *   indices). The hello's `tables` is a per-query SUBSET (the query's table tree), so this128   *   checks each advertised table rather than one global fingerprint. A mismatch (routine129   *   deployment / schema skew) is rejected here — without it, positional rows aligned to the130   *   SERVER's column order are stored verbatim under the CLIENT's order, silently swapping131   *   cells and mis-keying the refcount/GC (CRIT#4 / §3 "drift ⇒ re-subscribe").132   */133  constructor(134    hello: NormalizedHello,135    emit: (ev: NormalizedEvent) => void,136    clientTables?: NormalizedTableSchema[],137  ) {138    this.emit = emit;139    if (hello.comparatorVersion !== COMPARATOR_VERSION) {140      throw new ProtocolError(141        "comparator-mismatch",142        `comparator version ${hello.comparatorVersion} != ${COMPARATOR_VERSION}`,143      );144    }145    const computed = normalizedFp(hello.tables);146    if (computed !== hello.normalizedFp) {147      throw new ProtocolError("schema-mismatch", `advertised fp ${hello.normalizedFp} != computed ${computed}`);148    }149    if (clientTables) validateAgainstClientSchema(hello.tables, clientTables);150    this.epoch = hello.epoch;151    this.normalizedFp = hello.normalizedFp;152    emit({153      type: "hello",154      tables: hello.tables,155      comparatorVersion: hello.comparatorVersion,156      normalizedFp: hello.normalizedFp,157    });158  }159160  /** Apply one normalized batch (or the seq-0 snapshot). Returns `"duplicate"` for an161   *  already-applied seq; throws {@link ProtocolError} on a gap / epoch / fp mismatch (the162   *  caller re-hydrates under a new epoch). */163  apply(batch: NormalizedBatch): "applied" | "duplicate" {164    if (batch.epoch !== this.epoch) {165      throw new ProtocolError("epoch-mismatch", `expected epoch ${this.epoch}, got ${batch.epoch}`);166    }167    if (batch.normalizedFp !== this.normalizedFp) {168      throw new ProtocolError("schema-mismatch", `expected fp ${this.normalizedFp}, got ${batch.normalizedFp}`);169    }170    if (this.phase === "snapshot") {171      if (batch.seq === 0) {172        this.phase = "live";173        this.lastSeq = 0;174        this.emit({ type: "snapshot", ops: batch.ops, cv: batch.cv });175        return "applied";176      }177      throw new ProtocolError("gap", `expected the seq-0 snapshot, got seq ${batch.seq}`);178    }179    const expected = this.lastSeq + 1;180    if (batch.seq < expected) return "duplicate";181    if (batch.seq > expected) {182      throw new ProtocolError("gap", `expected seq ${expected}, got ${batch.seq}`);183    }184    this.lastSeq = batch.seq;185    this.emit({ type: "batch", ops: batch.ops, cv: batch.cv });186    return "applied";187  }188}189