Source snapshot

packages/remote/src/remote-source.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// RemoteNormalizedSource: a `NormalizedSource` over a network transport — the ws sibling of2// the in-process native source (NORMALIZED-CHANGES-DESIGN.md §7). It subscribes in normalized3// mode, owns the epoch/seq/gap protocol (via {@link NormalizedSubscriber}), and emits clean4// `NormalizedEvent`s upward — so `@rindle/normalized`'s `NormalizedBackend` drives the local5// engine identically whether the footprint stream comes from in-process or over the wire.6//7// On a gap (or epoch/fp drift) it re-subscribes; the server re-registers the query under a NEW8// epoch and replies with a fresh hello + snapshot, and `NormalizedSync` diffs the new footprint9// against the old (the set-analogue re-hydrate, §5.3).1011import type {12  Mutation,13  NormalizedEvent,14  NormalizedSource,15  NormalizedTableSchema,16  QueryId,17  RemoteQuery,18} from "@rindle/client";1920import { NormalizedSubscriber } from "./normalized.ts";21import { retryDelayMs } from "./query-error.ts";22import type { NormalizedBatch, NormalizedHello } from "./normalized.ts";23import { ProtocolError } from "./protocol.ts";24import type { ServerMsg } from "./protocol.ts";25import {26  defaultSubscribeTarget,27  isThenable,28  subscribeMessage,29  type RawMutationSender,30  type SubscribeResolver,31  type SubscribeTarget,32} from "./subscribe.ts";33import { WsTransport } from "./transport.ts";34import type { Transport } from "./transport.ts";3536interface QState {37  remote: RemoteQuery;38  subscriber: NormalizedSubscriber | null;39  /** The epoch of the current subscription (0 before the first hello). */40  epoch: number;41  /** True between sending a re-subscribe and receiving its hello (so a second gap is ignored). */42  resubscribing: boolean;43  /** Monotonic token that cancels stale async lease resolutions. */44  subscribeTicket: number;45  /** Pending retryable-error re-subscribe timer (FOLLOWER-LAG-SHED §6.3), if any. */46  retryTimer: ReturnType<typeof setTimeout> | undefined;47  /** Consecutive retryable errors without a successful hello — the backoff exponent. */48  retryAttempt: number;49}5051export interface RemoteNormalizedSourceOptions {52  /** Resolve the upstream subscribe target. Defaults to embedded-server `{name,args}`. */53  resolveSubscribe?: SubscribeResolver;54  /** Override raw authoritative writes, e.g. POST them to an app API server. */55  sendMutation?: RawMutationSender;56}5758export class RemoteNormalizedSource implements NormalizedSource {59  private readonly transport: Transport;60  private readonly resolveSubscribe: SubscribeResolver;61  private readonly sendMutation?: RawMutationSender;62  private handler: (qid: QueryId, ev: NormalizedEvent) => void = () => {};63  private readonly subs = new Map<QueryId, QState>();64  /** The client's own typed per-table schemas, for hello validation (CRIT#4); set by the backend. */65  private clientTables: NormalizedTableSchema[] | undefined;6667  constructor(transport: Transport, opts: RemoteNormalizedSourceOptions = {}) {68    this.transport = transport;69    this.resolveSubscribe = opts.resolveSubscribe ?? defaultSubscribeTarget;70    this.sendMutation = opts.sendMutation;71    this.transport.onMessage((msg) => this.onServerMsg(msg));72  }7374  expectClientSchema(tables: NormalizedTableSchema[]): void {75    this.clientTables = tables;76  }7778  registerQuery(qid: QueryId, remote: RemoteQuery): void {79    this.subs.set(qid, { remote, subscriber: null, epoch: 0, resubscribing: false, subscribeTicket: 0, retryTimer: undefined, retryAttempt: 0 });80    this.subscribe(qid, remote);81  }8283  unregisterQuery(qid: QueryId): void {84    const s = this.subs.get(qid);85    if (s?.retryTimer !== undefined) clearTimeout(s.retryTimer);86    this.subs.delete(qid);87    this.transport.send({ t: "unsubscribe", queryId: qid });88  }8990  mutate(mutations: Mutation[]): Promise<void> {91    if (this.sendMutation) return Promise.resolve(this.sendMutation(mutations));92    this.transport.send({ t: "mutate", mutations });93    return Promise.resolve();94  }9596  onNormalized(handler: (qid: QueryId, ev: NormalizedEvent) => void): void {97    this.handler = handler;98  }99100  // --- internals ---------------------------------------------------------------101102  private subscribe(qid: QueryId, remote: RemoteQuery): void {103    const s = this.subs.get(qid);104    if (!s) return;105    // A fresh subscribe (gap recovery, the retry timer itself) supersedes any scheduled106    // retryable-error retry — never leave two subscribe paths racing for one query.107    if (s.retryTimer !== undefined) {108      clearTimeout(s.retryTimer);109      s.retryTimer = undefined;110    }111    const request = { queryId: qid, remote, mode: "normalized" as const };112    const ticket = ++s.subscribeTicket;113    const send = (target: SubscribeTarget) => {114      const cur = this.subs.get(qid);115      if (cur !== s || cur.subscribeTicket !== ticket) return;116      this.transport.send(subscribeMessage(request, target));117    };118    const fail = (err: unknown) => {119      const cur = this.subs.get(qid);120      if (cur !== s || cur.subscribeTicket !== ticket) return;121      s.resubscribing = false;122      console.error(123        `[rindle-remote] normalized query ${qid} subscribe resolution failed: ${String((err as Error)?.message ?? err)}`,124      );125    };126    try {127      const target = this.resolveSubscribe(request);128      if (isThenable(target)) void target.then(send, fail);129      else send(target);130    } catch (err) {131      fail(err);132    }133  }134135  private onServerMsg(msg: ServerMsg): void {136    if (msg.t === "queryError") {137      this.onQueryError(msg.queryId, msg);138      return;139    }140    // This source is normalized-only; it sees `nhello`/`nbatch` (flat frames are ignored).141    if (msg.t !== "nhello" && msg.t !== "nbatch") return;142    const s = this.subs.get(msg.queryId);143    if (!s) return; // unsubscribed / unknown query144    if (msg.t === "nhello") this.openSubscriber(msg.queryId, s, msg.hello);145    else this.applyBatch(msg.queryId, s, msg.batch);146  }147148  /** Route a `queryError` by its 101 §5 classification: retryable ⇒ keep the QState (and the149   *  rows already folded downstream — 101 §6) and re-subscribe after a jittered backoff150   *  honoring `retryAfterMs`; terminal (or pre-classification servers) ⇒ drop the151   *  subscription, exactly as before (FOLLOWER-LAG-SHED §6.3). */152  private onQueryError(qid: QueryId, err: { message: string; code?: string; retryable?: boolean; retryAfterMs?: number }): void {153    const s = this.subs.get(qid);154    if (!s) return;155    if (err.retryable !== true) {156      if (s.retryTimer !== undefined) clearTimeout(s.retryTimer);157      this.subs.delete(qid);158      console.error(`[rindle-remote] normalized query ${qid} subscription rejected: ${err.message}`);159      return;160    }161    if (s.retryTimer !== undefined) return; // a retry is already scheduled — don't stack them162    s.subscriber = null; // stop validating the dead epoch; recovery is a fresh seq-0 hydrate163    s.resubscribing = true;164    const delay = retryDelayMs(s.retryAttempt++, err.retryAfterMs);165    console.warn(166      `[rindle-remote] normalized query ${qid} ${err.code ?? "error"} (retryable): re-subscribing in ${delay}ms: ${err.message}`,167    );168    const timer = setTimeout(() => {169      const cur = this.subs.get(qid);170      if (cur !== s) return;171      s.retryTimer = undefined;172      this.subscribe(qid, s.remote);173    }, delay);174    (timer as { unref?: () => void }).unref?.();175    s.retryTimer = timer;176  }177178  private openSubscriber(qid: QueryId, s: QState, hello: NormalizedHello): void {179    try {180      s.subscriber = new NormalizedSubscriber(hello, (ev) => this.handler(qid, ev), this.clientTables);181      s.epoch = hello.epoch;182      s.resubscribing = false;183      s.retryAttempt = 0; // a successful hello resets the retryable-error backoff184    } catch (e) {185      // A comparator/fp mismatch at hello is unrecoverable (a code-contract divergence).186      s.subscriber = null;187      console.error(`[rindle-remote] normalized query ${qid} subscription rejected: ${(e as Error).message}`);188    }189  }190191  private applyBatch(qid: QueryId, s: QState, batch: NormalizedBatch): void {192    if (!s.subscriber) return; // no hello yet (or mid re-hydrate)193    if (batch.epoch < s.epoch) return; // a stale batch from a superseded epoch — drop194    try {195      s.subscriber.apply(batch);196    } catch (e) {197      if (!(e instanceof ProtocolError)) throw e;198      if (s.resubscribing) return; // already recovering199      // Gap / drift → re-hydrate under a new epoch (the server bumps it on re-subscribe).200      s.resubscribing = true;201      s.subscriber = null;202      this.subscribe(qid, s.remote);203    }204  }205}206207/** Convenience: a `RemoteNormalizedSource` over a ws URL or a custom transport. */208export function createRemoteNormalizedSource(209  urlOrTransport: string | Transport,210  opts: RemoteNormalizedSourceOptions = {},211): RemoteNormalizedSource {212  const transport = typeof urlOrTransport === "string" ? new WsTransport(urlOrTransport) : urlOrTransport;213  return new RemoteNormalizedSource(transport, opts);214}215