WebSocket reconnect: heartbeats, backoff and close codes in production

// npm install ws   (Node.js 18 or newer)
import WebSocket from "ws";

const URL =
  "wss://stream.tickerlayer.com/?apiKey=" +
  encodeURIComponent(process.env.TICKERLAYER_API_KEY);

// Everything this process wants. Re-sent, in full, after every reconnect.
const SUBSCRIPTIONS = [
  { action: "subscribe", channels: ["crypto.quotes", "crypto.trades"], symbols: ["BTCUSD"] },
  { action: "subscribe", channels: ["forex.quotes"], symbols: ["EURUSD"] },
];

const BASE_MS = 1_000;     // first retry after about a second
const MAX_MS = 30_000;     // never wait longer than this
const STABLE_MS = 30_000;  // a connection that lived this long resets the ladder
const PING_MS = 10_000;    // our JSON ping; the server answers with a pong frame
const SILENT_MS = 30_000;  // no frame at all for this long: the socket is dead
const LAG_WARN_MS = 5_000; // a trade older than this on arrival: we are behind

let attempt = 0;
let stopped = false;

// Equal jitter: half the exponential step is fixed, half is random, so a
// thousand clients dropped at the same moment do not return at the same moment.
function nextDelay() {
  const step = Math.min(MAX_MS, BASE_MS * 2 ** attempt);
  attempt += 1;
  return Math.round(step / 2 + Math.random() * (step / 2));
}

function onMarketFrame(msg) {
  if (msg.type === "trade" && !msg.snapshot && Date.now() - msg.ts > LAG_WARN_MS) {
    console.warn(`falling behind: this trade is ${Date.now() - msg.ts} ms old`);
  }
  if (msg.type === "quote") {
    console.log(`${msg.symbol} bid ${Number(msg.bid)} ask ${Number(msg.ask)}${msg.snapshot ? " (snapshot)" : ""}`);
  }
  // Keep this function fast. Hand heavy work (database writes, indicators) to a queue.
}

function connect() {
  const ws = new WebSocket(URL, { perMessageDeflate: false, handshakeTimeout: 10_000 });
  let openedAt = 0;
  let lastFrameAt = Date.now();
  let retryAfterMs = 0;
  let timer;

  // The upgrade was refused with a plain HTTP status; no frames were exchanged.
  ws.on("unexpected-response", (req, res) => {
    const status = res.statusCode;
    retryAfterMs = Number(res.headers["retry-after"] ?? 0) * 1_000;
    if (status === 401 || status === 403) {
      console.error(`upgrade refused with ${status}: fix the key or the plan, not the network`);
      stopped = true;
    } else {
      console.warn(`upgrade refused with ${status}; will retry`);
    }
    ws.terminate(); // emits "error", then "close" with 1006
  });

  ws.on("open", () => {
    openedAt = Date.now();
    timer = setInterval(() => {
      if (Date.now() - lastFrameAt > SILENT_MS) {
        console.warn("no frames and no pong: treating the socket as dead");
        ws.terminate();
        return;
      }
      ws.send(JSON.stringify({ action: "ping" }));
    }, PING_MS);
  });

  ws.on("ping", () => { lastFrameAt = Date.now(); }); // ws answers with a pong for us

  ws.on("message", (data) => {
    lastFrameAt = Date.now();
    const msg = JSON.parse(data.toString());
    if (msg.type === "system" && msg.event === "ready") {
      for (const sub of SUBSCRIPTIONS) ws.send(JSON.stringify(sub));
    } else if (msg.type === "system" && msg.event === "disconnect") {
      console.warn(`server is closing the stream: ${msg.code}`);
    } else if (msg.type === "error") {
      // A rejected subscribe leaves the socket open. Reconnecting will not fix it.
      console.error(`error frame ${msg.code}: ${msg.message}`, msg.rejected ?? "");
    } else if (msg.type === "system") {
      if (msg.event !== "pong") console.log(`${msg.event}: ${msg.channels} ${msg.symbols}`);
    } else {
      onMarketFrame(msg);
    }
  });

  ws.on("error", (err) => console.error(`socket error: ${err.message}`)); // "close" follows

  ws.on("close", (code) => {
    clearInterval(timer);
    if (stopped) return;
    if (openedAt && Date.now() - openedAt > STABLE_MS) attempt = 0;
    let delay;
    if (code === 1012) {
      delay = Math.round(Math.random() * 1_000); // planned release: come back now, spread out
    } else {
      delay = Math.max(nextDelay(), retryAfterMs);
    }
    if (code === 4008) console.warn("4008: the process did not answer heartbeats (blocked event loop?)");
    console.log(`closed with ${code}; reconnecting in ${delay} ms`);
    setTimeout(connect, delay);
  });

  return ws;
}

connect();

subscribed: crypto.quotes,crypto.trades BTCUSD
subscribed: forex.quotes EURUSD
EURUSD bid 1.13723 ask 1.13724
BTCUSD bid 82974.87 ask 82974.88
EURUSD bid 1.137228 ask 1.137237
...
closed with 1006; reconnecting in 800 ms
subscribed: crypto.quotes,crypto.trades BTCUSD
subscribed: forex.quotes EURUSD

// Browser version. Only for a private, signed-in tool: the key sits in the URL,
// so anyone who can open the page can read it. Public sites need a relay.
function openStream({ apiKey, subscriptions, onFrame }) {
  const url = "wss://stream.tickerlayer.com/?apiKey=" + encodeURIComponent(apiKey);
  const lifetime = new AbortController();
  let ws;
  let attempt = 0;
  let refusals = 0;      // closes before any ready frame, in a row
  let healthySince = 0;  // when the current socket sent its ready frame
  let lastFrameAt = 0;
  let pingTimer;
  let retryTimer;

  function connect() {
    clearTimeout(retryTimer);
    healthySince = 0;
    const socket = new WebSocket(url);
    ws = socket;

    socket.onopen = () => {
      lastFrameAt = Date.now();
      pingTimer = setInterval(() => {
        if (Date.now() - lastFrameAt > 30_000) return drop(socket); // half-open
        socket.send(JSON.stringify({ action: "ping" }));
      }, 10_000);
    };

    socket.onmessage = (event) => {
      lastFrameAt = Date.now();
      const msg = JSON.parse(event.data);
      if (msg.type === "system" && msg.event === "ready") {
        healthySince = Date.now();
        refusals = 0;
        for (const sub of subscriptions) socket.send(JSON.stringify(sub));
      } else if (msg.type !== "system") {
        onFrame(msg); // quote, trade and error frames
      }
    };

    // Browsers hide the HTTP status of a refused upgrade: 401 and 403 arrive as 1006.
    socket.onclose = (event) => {
      if (!healthySince) refusals += 1;
      scheduleReconnect(event.code);
    };
  }

  function drop(socket) {
    socket.onclose = null; // do not wait for a close handshake that may never come
    socket.close();
    scheduleReconnect(1006);
  }

  function scheduleReconnect(code) {
    clearInterval(pingTimer);
    if (refusals >= 5) {
      console.error("refused five times in a row: check the key and the plan");
      return;
    }
    if (healthySince && Date.now() - healthySince > 30_000) attempt = 0;
    const step = Math.min(30_000, 1_000 * 2 ** attempt++);
    const delay = code === 1012 ? Math.random() * 1_000 : step / 2 + Math.random() * (step / 2);
    retryTimer = setTimeout(connect, delay);
  }

  // Back from sleep or a network switch: skip the rest of the backoff.
  window.addEventListener("online", () => {
    if (ws.readyState === WebSocket.CLOSED) connect();
  }, { signal: lifetime.signal });

  connect();

  return function stop() {
    lifetime.abort();
    clearTimeout(retryTimer);
    clearInterval(pingTimer);
    ws.onclose = null;
    ws.close(1000);
  };
}

// Usage, in a private tool where each signed-in user supplies their own key.
const stop = openStream({
  apiKey: localStorage.getItem("tickerlayer-key"),
  subscriptions: [{ action: "subscribe", channels: ["forex.quotes"], symbols: ["EURUSD"] }],
  onFrame: (msg) => {
    if (msg.type === "quote") {
      document.querySelector("#eurusd").textContent = `${Number(msg.bid).toFixed(5)} / ${Number(msg.ask).toFixed(5)}`;
    } else if (msg.type === "error") {
      console.warn(`error frame ${msg.code}: ${msg.message}`);
    }
  },
});
window.addEventListener("pagehide", stop);