API index and search · Build metadata

Source snapshot

packages/remote/src/affinity.ts

1// Browser-side follower-affinity ticket transport (FOLLOWER-AFFINITY-DESIGN.md §3, §5). The ticket2// is OPAQUE here — the browser never mints or verifies it (placement, not authorization; §3, §9),3// so this module carries no crypto. It is the thin seam three pieces share:4//5//   - {@link WsTransport} offers the held ticket as a ws subprotocol at each connect so the fleet6//     edge selects the pinned follower;7//   - {@link RemoteOptimisticSource} writes the follower's minted ticket from the `{t:"affinity"}`8//     frame, and CLEARS it on a sustained outage (the pinned follower is gone → the next reconnect9//     anycasts to a live one and re-pins, §8);10//   - the lease POST forwards the held ticket as `affinity`, sequenced AFTER a fresh connect's mint11//     frame ({@link AffinityTicketStore.waitForTicket}) so both legs land on the same follower (§4.1).1213/** The base ws subprotocol the browser offers alongside its `aff.*` ticket; the fleet follower14 *  echoes it on the 101 (a strict browser closes a socket whose selected subprotocol it never15 *  offered, so the base is always offered in affinity mode). Inlined — NOT imported from16 *  `@rindle/affinity`'s `WS_SUBPROTOCOL` — to keep that crate's `node:crypto` out of the browser17 *  bundle (the same duplicate-to-stay-bundle-clean discipline the lease wire types use). Keep the18 *  two spellings in lock-step. */19export const WS_SUBPROTOCOL = "rindle.v1";2021/** A mutable holder for the current opaque affinity ticket, shared by the transport, the source,22 *  and the lease POST (see the module header). */23export interface AffinityTicketStore {24  /** The ticket to offer on the next ws handshake, or undefined — ticketless, so the edge25   *  anycasts + mints a fresh one. A persisted/previous-connection ticket may be returned here26   *  even while it is NOT yet safe for a lease; see {@link leaseTicket}. */27  get(): string | undefined;28  /** Record a freshly minted/refreshed ticket (from a connection's `{t:"affinity"}` frame). */29  set(ticket: string): void;30  /** Drop the ticket so the next connect goes ticketless (the pinned follower is gone, §8). */31  clear(): void;32  /** Mark the held ticket handshake-only until this connection emits a fresh mint frame. Called33   *  before the first connection and every reconnect, after the held ticket was already selected34   *  for the ws subprotocol offer. This prevents a restored/rotated/expired persisted ticket from35   *  racing an HTTP lease onto an independently-anycast follower. */36  connectionPending(): void;37  /** Resolve with a ticket minted/refreshed on the CURRENT connection, or the next one38   *  {@link set}. A persisted ticket deliberately does not resolve this wait. */39  waitForTicket(): Promise<string>;40  /** Obtain the current connection's ticket for an HTTP lease. The first missing-ticket wait is41   *  bounded; its timeout removes the waiter and latches ticketless fallback, so later leases42   *  return immediately until a mint frame arrives. Concurrent callers share the one timer.43   *  `timedOut` is true only for that transition, allowing one warning per fallback episode. */44  leaseTicket(timeoutMs: number): Promise<{ ticket?: string; timedOut: boolean }>;45}4647/** Where a store persists the ticket across reloads. The browser backs this with sessionStorage48 *  (per-TAB, so two tabs can pin two regions — design §13); tests pass none (pure in-memory). */49export interface TicketPersistence {50  load(): string | undefined;51  save(ticket: string): void;52  clear(): void;53}5455/** Build an {@link AffinityTicketStore}, optionally persisted. Pure/in-memory when `persist` is56 *  omitted. */57export function createAffinityTicketStore(persist?: TicketPersistence): AffinityTicketStore {58  let current = persist?.load();59  // A loaded ticket is useful for the ws handshake, but is not lease-safe until the follower on60  // THIS connection confirms it by minting a fresh frame. The same distinction is re-armed on61  // every reconnect via `connectionPending`.62  let currentConnectionFresh = false;63  let ticketlessFallback = false;64  let waiters: Array<(ticket: string) => void> = [];65  let leaseWait:66    | {67        promise: Promise<{ ticket?: string; timedOut: boolean }>;68        resolve: (result: { ticket?: string; timedOut: boolean }) => void;69        timer: ReturnType<typeof setTimeout>;70      }71    | undefined;7273  const finishLeaseWait = (result: { ticket?: string; timedOut: boolean }): void => {74    const pending = leaseWait;75    if (!pending) return;76    clearTimeout(pending.timer);77    leaseWait = undefined;78    pending.resolve(result);79  };8081  return {82    get: () => current,83    set: (ticket) => {84      current = ticket;85      currentConnectionFresh = true;86      ticketlessFallback = false;87      persist?.save(ticket);88      // Resolve everyone waiting for the (now-arrived) ticket, in FIFO order.89      const pending = waiters;90      waiters = [];91      for (const resolve of pending) resolve(ticket);92      finishLeaseWait({ ticket, timedOut: false });93    },94    clear: () => {95      // Drop the ticket but leave any waiters pending — the next connect's mint frame resolves them96      // (a dead-follower reassignment must still be able to lease once the new ticket arrives).97      current = undefined;98      currentConnectionFresh = false;99      persist?.clear();100    },101    connectionPending: () => {102      currentConnectionFresh = false;103    },104    waitForTicket: () =>105      currentConnectionFresh && current !== undefined106        ? Promise.resolve(current)107        : new Promise<string>((resolve) => waiters.push(resolve)),108    leaseTicket: (timeoutMs) => {109      if (currentConnectionFresh && current !== undefined) {110        return Promise.resolve({ ticket: current, timedOut: false });111      }112      if (ticketlessFallback) return Promise.resolve({ timedOut: false });113      if (leaseWait) return leaseWait.promise;114115      let resolve!: (result: { ticket?: string; timedOut: boolean }) => void;116      const promise = new Promise<{ ticket?: string; timedOut: boolean }>((r) => {117        resolve = r;118      });119      const timer = setTimeout(() => {120        // Clear the shared waiter before resolving callers. A later mint starts a new, fresh121        // episode; no abandoned resolver accumulates per lease while affinity is off.122        leaseWait = undefined;123        ticketlessFallback = true;124        resolve({ timedOut: true });125      }, Math.max(0, timeoutMs));126      leaseWait = { promise, resolve, timer };127      return promise;128    },129  };130}131132/** The subprotocol list to offer for one connection: the base protocol always, plus the held ticket133 *  (already an `aff.<…>` token) when one exists. */134export function offerSubprotocols(store: AffinityTicketStore): string[] {135  const ticket = store.get();136  return ticket ? [WS_SUBPROTOCOL, ticket] : [WS_SUBPROTOCOL];137}138