Connection Lifecycle, Heartbeats and Reconnection

sleep = random_between(0, min(cap, base * 2 ** attempt))

{
  "name": "chatterbox-reconnect-demo",
  "private": true,
  "type": "module",
  "scripts": {
    "build": "tsc",
    "demo": "tsc && node dist/demo.js",
    "simulate": "tsc && node dist/simulate-backoff.js"
  },
  "dependencies": { "ws": "8.22.0" },
  "devDependencies": { "@types/node": "24.19.2", "@types/ws": "8.18.1", "typescript": "7.0.2" }
}

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "lib": ["ES2023", "DOM"],
    "types": ["node"],
    "strict": true,
    "outDir": "dist",
    "rootDir": "src",
    "skipLibCheck": true
  },
  "include": ["src"]
}

export type State =
  | "idle"
  | "connecting"
  | "handshaking"
  | "online"
  | "backoff"
  | "offline"
  | "suspended"
  | "closed";

export interface Envelope {
  v: 1;
  type: string;
  id: string;
  room: string | null;
  ts: number;
  payload: Record<string, any>;
}

export interface Options {
  url: string;
  getTicket?: () => Promise<string | undefined>;
  baseMs?: number;
  capMs?: number;
  stableMs?: number;
  ackTimeoutMs?: number;
  /** Where visibilitychange is dispatched and visibilityState is read: document in a page. */
  documentTarget?: EventTarget & { visibilityState?: string };
  /** Where pagehide, pageshow, online and offline are dispatched: window in a page. */
  windowTarget?: EventTarget & { navigator?: { onLine: boolean } };
  random?: () => number;
}

const NO_ACK = new Set(["typing", "receipt"]);
const FATAL = new Set([4400, 4401, 4403, 4413, 4426]);
let counter = 0;
const newId = () => `c${Date.now().toString(36)}${(counter++).toString(36)}${Math.random().toString(36).slice(2, 8)}`;

export class ReconnectingChatterboxSocket {
  state: State = "idle";
  onState: (s: State, why: string) => void = () => {};
  onEnvelope: (e: Envelope) => void = () => {};

  private ws?: WebSocket;
  private attempt = 0;
  private heartbeatMs = 20_000;
  private lastRx = 0;
  private sessionId?: string;
  private hadSession = false;
  private ackTimer?: ReturnType<typeof setTimeout>;
  private readonly lastSeq = new Map<string, number>();
  private readonly rooms = new Set<string>();
  private readonly outbox: { env: Envelope; sentAt: number }[] = [];
  private timers = new Set<ReturnType<typeof setTimeout>>();
  private hbTimer?: ReturnType<typeof setInterval>;
  private stableTimer?: ReturnType<typeof setTimeout>;
  private backoffTimer?: ReturnType<typeof setTimeout>;
  private wantOpen = false;
  private readonly o: Required<Pick<Options, "baseMs" | "capMs" | "stableMs" | "ackTimeoutMs" | "random">> & Options;

  constructor(options: Options) {
    this.o = { baseMs: 1000, capMs: 30_000, stableMs: 30_000, ackTimeoutMs: 10_000, random: Math.random, ...options };
    this.o.documentTarget?.addEventListener("visibilitychange", () => this.onVisibility());
    const w = this.o.windowTarget;
    w?.addEventListener("pagehide", () => this.suspend("pagehide"));
    w?.addEventListener("pageshow", (e) => {
      if ((e as PageTransitionEvent).persisted && this.wantOpen) this.connectNow("pageshow");
    });
    w?.addEventListener("online", () => this.onOnline());
    w?.addEventListener("offline", () => this.onOffline());
  }

  connect(): void {
    this.wantOpen = true;
    this.connectNow("connect()");
  }

  close(): void {
    this.wantOpen = false;
    this.teardown(1000, "client close");
    this.set("closed", "close()");
  }

  join(room: string): void {
    this.rooms.add(room);
    this.send("join", room, { since: this.lastSeq.get(room) ?? null });
  }

  /** Queues the envelope. Anything that needs an ack stays in the outbox until the ack arrives. */
  send(type: string, room: string | null, payload: Record<string, any>): string {
    const env: Envelope = { v: 1, type, id: newId(), room, ts: 0, payload };
    if (NO_ACK.has(type)) {
      if (this.state === "online") this.ws!.send(JSON.stringify(env));
      return env.id;
    }
    this.outbox.push({ env, sentAt: 0 });
    this.flush();
    return env.id;
  }

  get pending(): number {
    return this.outbox.length;
  }

  // ---- connection management ----

  private async connectNow(why: string): Promise<void> {
    if (!this.wantOpen || this.state === "connecting" || this.state === "handshaking" || this.state === "online") return;
    clearTimeout(this.backoffTimer);
    this.set("connecting", why);
    let ticket: string | undefined;
    try {
      ticket = await this.o.getTicket?.();
    } catch {
      return this.scheduleRetry("ticket failed");
    }
    const url = new URL(this.o.url);
    if (ticket) url.searchParams.set("ticket", ticket);
    const ws = new WebSocket(url, "chatterbox.v1");
    this.ws = ws;
    ws.addEventListener("open", () => {
      if (this.ws !== ws) return;
      this.set("handshaking", "socket open");
      const hello: Record<string, any> = { client: "web/1.4" };
      if (this.sessionId) hello.resume = { sessionId: this.sessionId, rooms: Object.fromEntries(this.lastSeq) };
      ws.send(JSON.stringify({ v: 1, type: "hello", id: newId(), room: null, ts: 0, payload: hello }));
      this.lastRx = Date.now();
      // The server closes with 4408 when hello does not arrive in 5 s, we give up on a missing reply in 5 s too.
      this.after(5000, () => this.ws === ws && this.state === "handshaking" && this.abandon(4408, "hello reply timeout"));
    });
    ws.addEventListener("message", (ev) => this.onMessage(ws, String(ev.data)));
    ws.addEventListener("close", (ev) => {
      if (this.ws === ws) this.onClose(ev.code, ev.reason);
    });
    ws.addEventListener("error", () => {
      // The browser gives no detail here; the close event that follows carries the code.
    });
  }

  private onMessage(ws: WebSocket, data: string): void {
    if (this.ws !== ws) return;
    this.lastRx = Date.now();
    const env = JSON.parse(data) as Envelope;
    if (env.type === "hello") return this.onHello(env);
    if (env.type === "ack" || env.type === "error") {
      const ref = env.payload.ref as string | null;
      const i = this.outbox.findIndex((x) => x.env.id === ref);
      if (i >= 0 && (env.type === "ack" || env.payload.retryable === false)) this.outbox.splice(i, 1);
      if (env.type === "ack" && env.room && typeof env.payload.seq === "number") this.bump(env.room, env.payload.seq);
    } else if (env.room && typeof env.payload.seq === "number") {
      const known = this.lastSeq.get(env.room) ?? 0;
      if (env.payload.seq <= known) return; // replay overlap, already delivered
      this.bump(env.room, env.payload.seq);
    }
    this.onEnvelope(env);
  }

  private onHello(env: Envelope): void {
    const p = env.payload;
    this.sessionId = p.sessionId;
    const hadSession = this.hadSession;
    this.hadSession = true;
    this.heartbeatMs = p.heartbeatMs;
    this.set("online", p.resumed ? "hello (resumed)" : "hello (fresh session)");
    if (!p.resumed && hadSession) {
      // The server forgot us: join again from our cursors and let it replay what we missed.
      for (const room of this.rooms) {
        if (this.outbox.some((x) => x.env.type === "join" && x.env.room === room)) continue;
        this.outbox.unshift({ env: { v: 1, type: "join", id: newId(), room, ts: 0, payload: { since: this.lastSeq.get(room) ?? null } }, sentAt: 0 });
      }
    }
    this.flush(true);
    clearInterval(this.hbTimer);
    this.hbTimer = setInterval(() => this.heartbeat(), this.heartbeatMs);
    // Reset the attempt counter only after the connection proved stable.
    clearTimeout(this.stableTimer);
    this.stableTimer = setTimeout(() => (this.attempt = 0), this.o.stableMs);
  }

  private heartbeat(): void {
    if (this.state !== "online") return;
    if (Date.now() - this.lastRx > 2 * this.heartbeatMs) return this.abandon(4408, "no frame for 2 heartbeats");
    this.ws!.send(JSON.stringify({ v: 1, type: "presence", id: newId(), room: null, ts: 0, payload: { status: "online" } }));
  }

  private flush(all = false): void {
    if (this.state !== "online") return;
    const now = Date.now();
    for (const item of this.outbox) {
      const due = item.sentAt === 0 || all || now - item.sentAt >= this.o.ackTimeoutMs;
      if (!due) continue;
      item.sentAt = now;
      this.ws!.send(JSON.stringify(item.env)); // the same id every time
    }
    clearTimeout(this.ackTimer);
    this.ackTimer = setTimeout(() => this.flush(), this.o.ackTimeoutMs);
  }

  private onClose(code: number, reason: string): void {
    clearInterval(this.hbTimer);
    clearTimeout(this.stableTimer);
    this.ws = undefined;
    if (!this.wantOpen) return;
    if (FATAL.has(code)) {
      this.wantOpen = false;
      return this.set("closed", `fatal close ${code} ${reason}`);
    }
    if (code === 4503 || code === 1001) return this.scheduleRetry(`close ${code}`, { fast: true });
    if (code === 4429) return this.scheduleRetry(`close ${code}`, { minMs: 5000 });
    this.scheduleRetry(`close ${code}`);
  }

  /** Half-open or silent socket: do not wait for the close handshake, it may never complete. */
  private abandon(code: number, why: string): void {
    this.teardown(code, why);
    this.scheduleRetry(why);
  }

  private teardown(code: number, reason: string): void {
    const ws = this.ws;
    this.ws = undefined;
    clearInterval(this.hbTimer);
    clearTimeout(this.stableTimer);
    clearTimeout(this.ackTimer);
    for (const t of this.timers) clearTimeout(t);
    this.timers.clear();
    try {
      ws?.close(code, reason);
    } catch {
      /* already closing */
    }
  }

  // ---- retry policy ----

  /** Full jitter: a random delay between 0 and min(cap, base * 2^attempt). */
  nextDelay(opts: { fast?: boolean; minMs?: number } = {}): number {
    const ceiling = opts.fast ? 1000 : Math.min(this.o.capMs, this.o.baseMs * 2 ** this.attempt);
    return Math.max(opts.minMs ?? 0, Math.floor(this.o.random() * ceiling));
  }

  private scheduleRetry(why: string, opts: { fast?: boolean; minMs?: number } = {}): void {
    if (!this.wantOpen) return;
    if (this.o.windowTarget?.navigator?.onLine === false) return this.set("offline", `${why}; waiting for online`);
    const delay = this.nextDelay(opts);
    if (!opts.fast) this.attempt++;
    this.set("backoff", `${why}; retry in ${delay} ms`);
    clearTimeout(this.backoffTimer);
    this.backoffTimer = setTimeout(() => this.connectNow("backoff elapsed"), delay);
  }

  // ---- page and network signals ----

  private onVisibility(): void {
    if (this.o.documentTarget?.visibilityState !== "visible" || !this.wantOpen) return;
    // Timers were throttled while hidden: judge the socket by timestamps, not by missed ticks.
    if (this.state === "online" && Date.now() - this.lastRx > 2 * this.heartbeatMs) this.abandon(4408, "stale after hidden");
    else if (this.state === "backoff" || this.state === "suspended") this.connectNow("visible again");
  }

  private onOffline(): void {
    // The event is a hint, not proof. An open socket keeps running and the watchdog decides;
    // a pending retry is dropped so that no attempt is wasted while there is no network.
    if (this.state !== "backoff") return;
    clearTimeout(this.backoffTimer);
    this.set("offline", "offline event");
  }

  private onOnline(): void {
    if (!this.wantOpen) return;
    if (this.state === "offline" || this.state === "backoff") {
      clearTimeout(this.backoffTimer);
      // A small random delay stops every tab on the network from reconnecting in the same instant.
      this.backoffTimer = setTimeout(() => this.connectNow("online event"), Math.floor(this.o.random() * 1000));
    }
  }

  private suspend(why: string): void {
    if (!this.wantOpen) return;
    this.teardown(1000, why); // 1000 is allowed in a browser; the page may enter the bfcache
    this.set("suspended", why);
  }

  // ---- helpers ----

  private bump(room: string, seq: number): void {
    if (seq > (this.lastSeq.get(room) ?? 0)) this.lastSeq.set(room, seq);
  }

  private after(ms: number, fn: () => void): void {
    const t = setTimeout(() => {
      this.timers.delete(t);
      fn();
    }, ms);
    this.timers.add(t);
  }

  private set(s: State, why: string): void {
    this.state = s;
    this.onState(s, why);
  }
}

import { WebSocketServer, WebSocket } from "ws";
import { randomBytes } from "node:crypto";

interface Session { userId: string; rooms: Set<string>; socket?: WebSocket; expires: number }
interface Stored { seq: number | null; room: string | null }

export interface TestServerOptions {
  heartbeatMs?: number;
  sessionTtlMs?: number;
}

/** A tiny Chatterbox v1 server that can misbehave on demand. */
export async function startServer(opts: TestServerOptions = {}) {
  const heartbeatMs = opts.heartbeatMs ?? 200;
  const sessionTtlMs = opts.sessionTtlMs ?? 120_000;
  const wss = new WebSocketServer({ port: 0, handleProtocols: (p) => (p.has("chatterbox.v1") ? "chatterbox.v1" : false) });
  await new Promise((r) => wss.once("listening", r));
  const port = (wss.address() as { port: number }).port;

  const sessions = new Map<string, Session>();
  const log = new Map<string, any[]>(); // room -> message envelopes in seq order
  const seen = new Map<string, Stored>(); // idempotency: client id -> original ack data
  const stats = { connections: 0, messageEnvelopes: 0, duplicates: 0, helloResumed: 0, helloFresh: 0 };
  let muted = false; // true simulates a half-open connection: nothing in, nothing out
  let swallowAcks = 0; // process the next N messages but never send their acks

  const sid = () => "s_" + randomBytes(8).toString("hex");
  const out = (ws: WebSocket, type: string, room: string | null, payload: object) => {
    if (!muted && ws.readyState === ws.OPEN) ws.send(JSON.stringify({ v: 1, type, id: sid(), room, ts: Date.now(), payload }));
  };
  const nextSeq = (room: string) => (log.get(room)?.length ?? 0) + 1;
  const append = (room: string, from: string, id: string, text: string) => {
    const env = { v: 1, type: "message", id, room, ts: Date.now(), payload: { from: { userId: from }, text, seq: nextSeq(room) } };
    log.set(room, [...(log.get(room) ?? []), env]);
    for (const s of sessions.values()) if (s.rooms.has(room) && s.socket) out(s.socket, "message", room, env.payload);
    return env.payload.seq;
  };

  wss.on("connection", (ws) => {
    stats.connections++;
    let session: Session | undefined;
    let lastRx = Date.now();
    const helloTimer = setTimeout(() => ws.close(4408, "hello timeout"), 5000);
    const beat = setInterval(() => {
      if (muted) return;
      if (Date.now() - lastRx > 2 * heartbeatMs) return ws.close(4408, "heartbeat timeout");
      ws.ping();
    }, heartbeatMs);
    ws.on("close", () => {
      clearTimeout(helloTimer);
      clearInterval(beat);
      if (session && session.socket === ws) { session.socket = undefined; session.expires = Date.now() + sessionTtlMs; }
    });
    ws.on("message", (raw) => {
      if (muted) return;
      lastRx = Date.now();
      const env = JSON.parse(String(raw));
      const ack = (extra: object = {}, seq: number | null = null, dup = false) =>
        out(ws, "ack", env.room, { ref: env.id, ok: true, seq, dup, result: {}, ...extra });

      if (env.type === "hello") {
        clearTimeout(helloTimer);
        const r = env.payload.resume;
        const old = r && sessions.get(r.sessionId);
        const resumed = !!old && old.expires > Date.now();
        const me: Session = resumed ? old : { userId: "u_" + randomBytes(3).toString("hex"), rooms: new Set(), expires: 0 };
        session = me;
        const id = resumed ? r.sessionId : sid();
        sessions.set(id, me);
        me.socket = ws;
        resumed ? stats.helloResumed++ : stats.helloFresh++;
        out(ws, "hello", null, { ref: env.id, userId: me.userId, name: "demo", sessionId: id, heartbeatMs, maxFrameBytes: 16384, resumed, rooms: [...me.rooms] });
        if (resumed) for (const [room, last] of Object.entries<number>(r.rooms ?? {})) {
          for (const m of log.get(room) ?? []) if (m.payload.seq > last) out(ws, "message", room, m.payload);
        }
        return;
      }
      if (!session) return ws.close(4400, "hello first");
      if (env.type === "join") {
        session.rooms.add(env.room);
        const since: number | null = env.payload.since;
        ack({ result: { members: [], lastSeq: nextSeq(env.room) - 1 } });
        for (const m of log.get(env.room) ?? []) if (since !== null && m.payload.seq > since) out(ws, "message", env.room, m.payload);
      } else if (env.type === "message") {
        stats.messageEnvelopes++;
        const prior = seen.get(env.id);
        if (prior) { stats.duplicates++; return ack({}, prior.seq, true); }
        const seq = append(env.room, session.userId, env.id, env.payload.text);
        seen.set(env.id, { seq, room: env.room });
        if (swallowAcks > 0) { swallowAcks--; return; }
        ack({}, seq);
      } else if (env.type === "presence") {
        ack();
      }
    });
  });

  return {
    port, stats, url: `ws://127.0.0.1:${port}/ws`,
    /** Kill every TCP connection without a close frame (the client sees 1006). */
    dropAll: () => wss.clients.forEach((c) => c.terminate()),
    /** Close every socket with an application code. */
    closeAll: (code: number, reason = "") => wss.clients.forEach((c) => c.close(code, reason)),
    setMuted: (m: boolean) => { muted = m; },
    swallowNextAcks: (n: number) => { swallowAcks = n; },
    /** A message from "someone else", delivered live or replayed later. */
    inject: (room: string, text: string) => append(room, "u_other", sid(), text),
    stop: () => new Promise<void>((r) => { wss.clients.forEach((c) => c.terminate()); wss.close(() => r()); }),
  };
}

import assert from "node:assert/strict";
import { startServer } from "./chatterbox-test-server.js";
import { ReconnectingChatterboxSocket, type Envelope } from "./reconnecting-chatterbox-socket.js";

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
const t0 = Date.now();
const stamp = () => String(Date.now() - t0).padStart(5);

async function until(cond: () => boolean, what: string, ms = 5000) {
  const end = Date.now() + ms;
  while (!cond()) {
    if (Date.now() > end) throw new Error("timeout waiting for " + what);
    await sleep(20);
  }
}

const server = await startServer({ heartbeatMs: 200, sessionTtlMs: 1500 });

// Fake page and network signal sources, so the browser-only code paths run under Node.
const doc = Object.assign(new EventTarget(), { visibilityState: "visible" });
const win = Object.assign(new EventTarget(), { navigator: { onLine: true } });

const received: Envelope[] = [];
const client = new ReconnectingChatterboxSocket({ url: server.url, baseMs: 100, capMs: 800, stableMs: 600, ackTimeoutMs: 1000, documentTarget: doc, windowTarget: win });
client.onState = (s, why) => console.log(`${stamp()} ms  state=${s.padEnd(10)} ${why}`);
client.onEnvelope = (e) => { if (e.type === "message") received.push(e); };

const texts = () => received.map((e) => e.payload.text);

console.log("-- 1. connect, join, send");
client.connect();
client.join("general");
client.send("message", "general", { text: "one", replyTo: null, attachments: [] });
await until(() => client.state === "online" && client.pending === 0, "first ack");

console.log("-- 2. TCP dropped, a message arrives meanwhile, session resumed");
server.dropAll();
server.inject("general", "missed while away");
await until(() => client.state === "online" && texts().includes("missed while away"), "resume replay");
assert.equal(server.stats.helloResumed, 1);

console.log("-- 3. ack lost with the connection: retry with the same id, no duplicate");
server.swallowNextAcks(1);
client.send("message", "general", { text: "two", replyTo: null, attachments: [] });
await sleep(100);
server.dropAll();
await until(() => client.state === "online" && client.pending === 0, "dup ack");
assert.equal(server.stats.duplicates, 1);
assert.equal(texts().filter((t) => t === "two").length, 1);

console.log("-- 4. half-open connection: server goes silent, watchdog notices");
server.setMuted(true);
await until(() => client.state === "backoff", "watchdog", 3000);
server.setMuted(false);
await until(() => client.state === "online", "recovery after watchdog");

console.log("-- 5. close-code policy: 4503 retries fast, 4429 waits at least 5 s (a visible tab skips the wait)");
server.closeAll(4503, "restarting");
await until(() => client.state === "backoff", "4503 backoff");
await until(() => client.state === "online", "4503 reconnect fast");
server.closeAll(4429, "slow down");
await until(() => client.state === "backoff", "4429 backoff");
doc.dispatchEvent(new Event("visibilitychange"));
await until(() => client.state === "online", "wake on visibility");

console.log("-- 5b. offline event: no attempts while offline, immediate after online");
win.navigator.onLine = false;
const conns = server.stats.connections;
server.dropAll();
await until(() => client.state === "offline", "offline state");
await sleep(500);
assert.equal(server.stats.connections, conns, "no connection attempt while offline");
win.navigator.onLine = true;
win.dispatchEvent(new Event("online"));
await until(() => client.state === "online", "online event reconnect");

console.log("-- 6. session expired on the server: fresh hello, join with since fills the gap");
server.inject("general", "before expiry");
await until(() => texts().includes("before expiry"), "live message");
server.setMuted(false);
doc.visibilityState = "hidden";
win.dispatchEvent(new Event("pagehide"));
await until(() => client.state === "suspended", "pagehide suspend");
server.inject("general", "while suspended");
await sleep(1700); // longer than sessionTtlMs
doc.visibilityState = "visible";
win.dispatchEvent(Object.assign(new Event("pageshow"), { persisted: true }));
await until(() => client.state === "online" && texts().includes("while suspended"), "since replay");
assert.ok(server.stats.helloFresh >= 2, "second fresh hello");
assert.deepEqual(texts().filter((t) => t === "while suspended").length, 1);

console.log("-- 7. unauthorized close is fatal");
server.closeAll(4401, "token expired");
await until(() => client.state === "closed", "fatal close");
await sleep(400);
const finalConns = server.stats.connections;
await sleep(400);
assert.equal(server.stats.connections, finalConns, "no retry after 4401");

console.log("\nmessages seen by the UI:", texts());
console.log("server stats:", server.stats);
await server.stop();
console.log("all scenarios passed");

-- 1. connect, join, send
    1 ms  state=connecting connect()
   14 ms  state=handshaking socket open
   15 ms  state=online     hello (fresh session)
-- 2. TCP dropped, a message arrives meanwhile, session resumed
   23 ms  state=backoff    close 1006; retry in 72 ms
   96 ms  state=connecting backoff elapsed
   97 ms  state=handshaking socket open
   97 ms  state=online     hello (resumed)
-- 3. ack lost with the connection: retry with the same id, no duplicate
  208 ms  state=backoff    close 1006; retry in 38 ms
  247 ms  state=connecting backoff elapsed
  248 ms  state=handshaking socket open
  249 ms  state=online     hello (resumed)
-- 4. half-open connection: server goes silent, watchdog notices
  850 ms  state=backoff    no frame for 2 heartbeats; retry in 58 ms
  908 ms  state=connecting backoff elapsed
  909 ms  state=handshaking socket open
  910 ms  state=online     hello (resumed)
-- 5. close-code policy: 4503 retries fast, 4429 waits at least 5 s (a visible tab skips the wait)
  920 ms  state=backoff    close 4503; retry in 735 ms
 1655 ms  state=connecting backoff elapsed
 1657 ms  state=handshaking socket open
 1657 ms  state=online     hello (resumed)
 1674 ms  state=backoff    close 4429; retry in 5000 ms
 1694 ms  state=connecting visible again
 1695 ms  state=handshaking socket open
 1696 ms  state=online     hello (resumed)
-- 5b. offline event: no attempts while offline, immediate after online
 1715 ms  state=offline    close 1006; waiting for online
 2749 ms  state=connecting online event
 2751 ms  state=handshaking socket open
 2751 ms  state=online     hello (resumed)
-- 6. session expired on the server: fresh hello, join with since fills the gap
 2782 ms  state=suspended  pagehide
 4483 ms  state=connecting pageshow
 4484 ms  state=handshaking socket open
 4485 ms  state=online     hello (fresh session)
-- 7. unauthorized close is fatal
 4505 ms  state=closed     fatal close 4401 token expired

messages seen by the UI: [
  'one',
  'missed while away',
  'two',
  'before expiry',
  'while suspended'
]
server stats: {
  connections: 8,
  messageEnvelopes: 3,
  duplicates: 1,
  helloResumed: 6,
  helloFresh: 2
}
all scenarios passed

// Prints the reconnect delays that 1000 clients would pick after the same outage.
const CLIENTS = 1000;
const BASE_MS = 1000;
const CAP_MS = 30_000;

type Policy = (attempt: number) => number;
const policies: Record<string, Policy> = {
  "no backoff (fixed 1 s)": () => BASE_MS,
  "exponential, no jitter": (n) => Math.min(CAP_MS, BASE_MS * 2 ** n),
  "full jitter": (n) => Math.random() * Math.min(CAP_MS, BASE_MS * 2 ** n),
};

function histogram(values: number[], bucketMs: number, buckets: number): number[] {
  const h = new Array<number>(buckets).fill(0);
  for (const v of values) h[Math.min(buckets - 1, Math.floor(v / bucketMs))]++;
  return h;
}

for (const [name, policy] of Object.entries(policies)) {
  console.log(`\n${name}`);
  for (const attempt of [0, 1, 2, 3, 4]) {
    const delays = Array.from({ length: CLIENTS }, () => policy(attempt)).sort((a, b) => a - b);
    const ceiling = Math.min(CAP_MS, BASE_MS * 2 ** attempt);
    const h = histogram(delays, ceiling / 10, 10);
    const peak = Math.max(...h);
    console.log(
      `attempt ${attempt}: window ${ceiling} ms, min ${delays[0].toFixed(0)}, median ${delays[CLIENTS / 2].toFixed(0)}, max ${delays[CLIENTS - 1].toFixed(0)}, busiest tenth of the window ${peak} of ${CLIENTS} clients`,
    );
  }
}

// The server stays down for 10 s. Every client keeps retrying; count the attempts that reach the server per 100 ms slot.
const OUTAGE_MS = 10_000;
for (const [name, policy] of Object.entries(policies)) {
  const slots = new Array<number>(OUTAGE_MS / 100).fill(0);
  let total = 0;
  for (let c = 0; c < CLIENTS; c++) {
    let t = 0;
    for (let attempt = 0; ; attempt++) {
      t += policy(attempt);
      if (t >= OUTAGE_MS) break;
      slots[Math.floor(t / 100)]++;
      total++;
    }
  }
  console.log(`\n${name}: ${total} attempts in 10 s, peak ${Math.max(...slots)} in one 100 ms slot`);
}

no backoff (fixed 1 s): 9000 attempts in 10 s, peak 1000 in one 100 ms slot
exponential, no jitter: 3000 attempts in 10 s, peak 1000 in one 100 ms slot
full jitter: 3957 attempts in 10 s, peak 156 in one 100 ms slot

import * as signalR from "@microsoft/signalr";

const connection = new signalR.HubConnectionBuilder()
  .withUrl("/chathub")
  .withStatefulReconnect({ bufferSize: 100000 })
  .withAutomaticReconnect({
    nextRetryDelayInMilliseconds: (ctx) => {
      if (ctx.elapsedMilliseconds > 5 * 60 * 1000) return null; // give up after five minutes
      const window = Math.min(30000, 1000 * 2 ** ctx.previousRetryCount);
      return Math.random() * window; // full jitter
    },
  })
  .build();

connection.onreconnecting(() => console.log("state: Reconnecting"));
connection.onreconnected((id) => console.log("state: Connected again", id));
connection.onclose(() => console.log("state: Disconnected, show the sign-in dialog"));

await connection.start();

// server.js
import { createServer } from "node:http";
import { Server } from "socket.io";

const io = new Server(createServer().listen(3000), {
  pingInterval: 20000,
  pingTimeout: 20000,
  connectionStateRecovery: { maxDisconnectionDuration: 2 * 60 * 1000, skipMiddlewares: false },
});

io.on("connection", (socket) => {
  console.log(socket.recovered ? "recovered session" : "new session", socket.id);
  socket.on("message", (m) => io.to("general").emit("message", m));
  socket.join("general");
});

// client.js
import { io } from "socket.io-client";

const socket = io("http://localhost:3000", { reconnectionDelay: 1000, reconnectionDelayMax: 30000, randomizationFactor: 0.5 });
socket.on("connect", () => console.log(socket.recovered ? "missed events follow" : "fresh state, reload history"));