API index and search · Build metadata

Source snapshot

packages/remote/src/transport.ts

1// The transport seam: the RemoteBackend talks JSON messages through a `Transport`, so the2// wire (ws / sse / http) is swappable + mockable. The default {@link WsTransport} uses the3// global `WebSocket` (Node 22+ and browsers both provide it — zero dependency).45import type { ClientMsg, ServerMsg } from "./protocol.ts";67export interface Transport {8  /** Send a message up to the server. */9  send(msg: ClientMsg): void;10  /** Register the single handler for incoming server messages. */11  onMessage(handler: (msg: ServerMsg) => void): void;12  /** Register a handler fired after the connection is RE-established (not the first open) — the13   *  source uses it to re-`init` + re-subscribe so a dropped/restarted daemon heals. Optional:14   *  transports without reconnect (mocks, in-process) may omit it. */15  onReconnect?(handler: () => void): void;16  /** Register a handler fired when the connection is SUSTAINEDLY down — repeated reconnects to the17   *  same endpoint have failed (a dead/removed follower, READ-ROUTER-DESIGN.md §3). The source uses18   *  it to re-lease: the router returns a (possibly new) `wsEndpoint`, and a changed one migrates the19   *  whole session off the dead node. Optional: transports without failover (mocks, in-process,20   *  fixed endpoints) may omit it. */21  onDown?(handler: () => void): void;22  /** Tear down the connection. */23  close(): void;24}2526/** A `Transport` over a `WebSocket` (text JSON frames). Messages sent before the socket27 *  opens are buffered and flushed on open (so `registerQuery`/`mutate` can be called eagerly).28 *  Reconnects with capped exponential backoff: if the socket drops (e.g. the daemon restarted)29 *  it reopens and fires `onReconnect` so the source rebuilds its subscriptions. */30export class WsTransport implements Transport {31  private readonly url: string;32  /** Reads the subprotocols to offer at each (re)connect — in affinity mode, `["rindle.v1", "aff.…"]`33   *  with the CURRENT ticket (FOLLOWER-AFFINITY-DESIGN.md §5). Undefined ⇒ offer none (today's34   *  single-daemon behavior, byte-identical). Evaluated per connect so a reconnect presents the35   *  freshest (or freshly cleared) ticket. */36  private readonly subprotocols?: () => string[];37  private ws: WebSocket;38  private handler: (msg: ServerMsg) => void = () => {};39  private reconnectHandler: () => void = () => {};40  private downHandler: () => void = () => {};41  /** Buffered as PRE-SERIALIZED frames: serialization happens at `send` time so an42   *  unserializable message (a `bigint` query arg — `JSON.stringify` throws on bigint)43   *  throws typed INTO ITS CALLER instead of detonating later inside the socket's44   *  `open` listener, where it would strand every frame queued behind it. */45  private readonly pending: string[] = [];46  private open = false;47  private everOpened = false;48  private closedByUser = false;49  private attempt = 0;50  private reconnectTimer: ReturnType<typeof setTimeout> | undefined;51  /** Failed reconnect attempts after which the connection is declared "down" (fires `onDown`). */52  private readonly downThreshold: number;53  /** True once `onDown` has fired for the CURRENT down episode; reset on the next successful open54   *  so a later outage fires again (but a single episode fires `onDown` exactly once — no re-lease55   *  storm while a follower is gone). */56  private downFired = false;5758  constructor(url: string, opts: { downThreshold?: number; subprotocols?: () => string[] } = {}) {59    this.url = url;60    this.downThreshold = opts.downThreshold ?? 4;61    this.subprotocols = opts.subprotocols;62    this.ws = this.connect();63  }6465  private connect(): WebSocket {66    // Offer the affinity subprotocols (base + current ticket) when configured; otherwise open bare,67    // exactly as before. An empty list is treated as bare (never send `Sec-WebSocket-Protocol: `).68    const protocols = this.subprotocols?.();69    const ws = protocols && protocols.length > 0 ? new WebSocket(this.url, protocols) : new WebSocket(this.url);70    ws.addEventListener("open", () => {71      this.open = true;72      this.attempt = 0;73      this.downFired = false; // a fresh connection clears the down episode74      if (!this.everOpened) {75        // First connection: flush whatever was buffered eagerly (init + subscribes).76        // Frames were serialized at `send` time, so this loop cannot throw.77        this.everOpened = true;78        for (const m of this.pending) ws.send(m);79        this.pending.length = 0;80      } else {81        // A reconnect: drop any stale buffered frames — the source rebuilds the full desired82        // state (re-init + re-subscribe, re-leasing as it goes) in onReconnect.83        this.pending.length = 0;84        this.reconnectHandler();85      }86    });87    ws.addEventListener("message", (ev: MessageEvent) => {88      this.handler(JSON.parse(typeof ev.data === "string" ? ev.data : String(ev.data)) as ServerMsg);89    });90    ws.addEventListener("close", () => {91      this.open = false;92      if (!this.closedByUser) this.scheduleReconnect();93    });94    // `error` is followed by `close`; let the close handler own reconnection.95    ws.addEventListener("error", () => {});96    return ws;97  }9899  private scheduleReconnect(): void {100    if (this.reconnectTimer !== undefined) return;101    const delay = Math.min(250 * 2 ** this.attempt, 5000);102    this.attempt++;103    // Sustained failure (we had a connection, and several reconnects to this endpoint have failed):104    // declare the connection down ONCE so the source can re-lease and migrate (§3). We keep105    // reconnecting underneath in case the same follower returns (a reboot, same endpoint).106    if (this.everOpened && !this.downFired && this.attempt >= this.downThreshold) {107      this.downFired = true;108      this.downHandler();109    }110    this.reconnectTimer = setTimeout(() => {111      this.reconnectTimer = undefined;112      if (!this.closedByUser) this.ws = this.connect();113    }, delay);114  }115116  send(msg: ClientMsg): void {117    // Serialize HERE, open or not: a frame the wire cannot carry (a `bigint` query118    // arg on the live-query plane — the browser bigint lane ships with design 226119    // Stage E) must throw typed into its caller, never poison the pending queue or120    // the socket's `open` flush.121    let text: string;122    try {123      text = JSON.stringify(msg);124    } catch (e) {125      throw new Error(126        "query args must be JSON-serializable: bigint values are not supported on the " +127          "live-query wire until the browser bigint lane ships (design 226) — " +128          `${e instanceof Error ? e.message : String(e)}`,129      );130    }131    if (this.open) this.ws.send(text);132    else this.pending.push(text);133  }134135  onMessage(handler: (msg: ServerMsg) => void): void {136    this.handler = handler;137  }138139  onReconnect(handler: () => void): void {140    this.reconnectHandler = handler;141  }142143  onDown(handler: () => void): void {144    this.downHandler = handler;145  }146147  close(): void {148    this.closedByUser = true;149    if (this.reconnectTimer !== undefined) {150      clearTimeout(this.reconnectTimer);151      this.reconnectTimer = undefined;152    }153    this.ws.close();154  }155}156