Resuming an Event Stream with a Cursor, a Bounded Log and a Snapshot

// A cursor is only safe if sequence numbers become visible in order.
// Writer A takes number 5 and is slow to publish; writer B takes 6 and publishes first. Run: node visibility_gap.mjs
const published = [];                                   // what readers can see, in publication order
const publish = (seq, data) => published.push({ seq, data });
const readAfter = (cursor) => published.filter((e) => e.seq > cursor);

publish(4, "x");                                        // everything up to 4 is visible
// writer A allocated seq 5 but has not committed yet
publish(6, "y");                                        // writer B allocated 6 and committed first

let cursor = 4;
const first = readAfter(cursor);                        // the reader polls: sees only 6
cursor = Math.max(cursor, ...first.map((e) => e.seq));  // cursor advances to 6
publish(5, "z");                                        // writer A commits late
const second = readAfter(cursor);                       // the reader polls again with cursor 6

console.log("first read :", first.map((e) => e.seq));
console.log("second read:", second.map((e) => e.seq), "<- event 5 is never delivered");

// What does a real browser EventSource do on reconnect, and on a non-200 response?
// Run: node eventsource_check.mjs   (needs a Chrome/Chromium binary; set CHROME=/path if needed)
import http from "node:http";
import { spawn } from "node:child_process";

const seenHeaders = [];   // Last-Event-ID values the server received on /sse
let attempts410 = 0;

const page = `<script>
const out = { messages: [], sseErrors: 0, gone: { errors: 0, readyState: null } };
const es = new EventSource("/sse");
es.onmessage = (e) => { out.messages.push(e.lastEventId); };
es.onerror = () => { out.sseErrors++; };
const gone = new EventSource("/gone");
gone.onerror = () => { out.gone.errors++; out.gone.readyState = gone.readyState; };
setTimeout(() => { fetch("/report", { method: "POST", body: JSON.stringify(out) }); }, 2500);
</script>`;

const server = http.createServer(async (req, res) => {
  if (req.url === "/") { res.writeHead(200, { "content-type": "text/html" }); return res.end(page); }
  if (req.url === "/sse") {
    const last = req.headers["last-event-id"]; seenHeaders.push(last ?? null);
    res.writeHead(200, { "content-type": "text/event-stream" });
    res.write("retry: 100\n\n");
    const start = last ? Number(last) + 1 : 1;
    for (let id = start; id < start + 3; id++) res.write(`id: ${id}\ndata: hello\n\n`);
    if (!last) setTimeout(() => res.destroy(), 100);          // first connection: drop it after 3 events
    return;
  }
  if (req.url === "/gone") { attempts410++; res.writeHead(410, { "content-type": "application/json" }); return res.end('{"reset":true}'); }
  if (req.url === "/report") {
    let b = ""; for await (const c of req) b += c;
    console.log("page reported:", b);
    console.log("Last-Event-ID headers seen by /sse:", JSON.stringify(seenHeaders));
    console.log("connection attempts to the 410 endpoint:", attempts410);
    res.end("ok"); clearTimeout(guard); chrome.kill(); server.close(); server.closeAllConnections();
  }
});
await new Promise((r) => server.listen(0, "127.0.0.1", r));
const chrome = spawn(process.env.CHROME ?? "google-chrome", [
  "--headless=new", "--no-sandbox", "--disable-gpu", "--user-data-dir=./es-check-profile",
  `http://127.0.0.1:${server.address().port}/`,
], { stdio: "ignore" });
const guard = setTimeout(() => { console.log("timeout"); chrome.kill(); process.exit(1); }, 20000);

// A resumable event stream in ~100 lines: cursor + bounded log + snapshot fallback.
// Requires Node 18+. Run: node resume.mjs
import http from "node:http";

const RETAIN = 50;                                   // the server keeps only the last 50 events
const log = [];                                      // [{ seq, key, value }]
const state = {};                                    // key -> { value, seq }   (the "current" truth)
let seq = 0;

const write = (key, value) => {                      // every state change appends to the log
  const ev = { seq: ++seq, key, value };
  state[key] = { value, seq };
  log.push(ev);
  if (log.length > RETAIN) log.shift();
  return ev;
};
const snapshot = () => ({ seq, state: structuredClone(state) });   // (cursor, state) read together: no await in between

const sse = (id, event, data) => `id: ${id}\nevent: ${event}\ndata: ${JSON.stringify(data)}\n\n`;

export const server = http.createServer((req, res) => {
  res.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-store" });
  const header = req.headers["last-event-id"];
  let cursor = header === undefined ? null : Number(header);

  const oldest = log.length ? log[0].seq : seq + 1;  // first sequence number still available
  if (cursor === null || !Number.isInteger(cursor) || cursor + 1 < oldest || cursor > seq) {
    const snap = snapshot();                          // cannot resume: tell the client in-band, then continue from the snapshot
    res.write(sse(snap.seq, "reset", snap));
    cursor = snap.seq;
  }
  for (const ev of log) if (ev.seq > cursor) { res.write(sse(ev.seq, "put", ev)); cursor = ev.seq; }

  const timer = setInterval(() => {                   // tail the log
    for (const ev of log) if (ev.seq > cursor) { res.write(sse(ev.seq, "put", ev)); cursor = ev.seq; }
  }, 5);
  req.on("close", () => clearInterval(timer));
});

// ---- a minimal client that behaves like EventSource: remember the last id, send it back on reconnect ----
export async function follow(url, onEvent, { signal }) {
  let lastId = null;
  while (!signal.aborted) {
    try {
      const res = await fetch(url, { headers: lastId === null ? {} : { "last-event-id": String(lastId) }, signal });
      let buf = "";
      for await (const chunk of res.body.pipeThrough(new TextDecoderStream())) {
        buf += chunk;
        for (let i; (i = buf.indexOf("\n\n")) >= 0; ) {
          const raw = buf.slice(0, i); buf = buf.slice(i + 2);
          const ev = Object.fromEntries(raw.split("\n").map((l) => [l.slice(0, l.indexOf(":")), l.slice(l.indexOf(":") + 2)]));
          lastId = Number(ev.id);
          onEvent(ev.event, JSON.parse(ev.data));
        }
      }
    } catch (e) { if (signal.aborted) return; }
    await new Promise((r) => setTimeout(r, 20));      // reconnect delay (EventSource's `retry`)
  }
}

if (import.meta.url === `file://${process.argv[1]}`) {
  await new Promise((r) => server.listen(0, "127.0.0.1", r));
  const url = `http://127.0.0.1:${server.address().port}/`;

  const mine = {}; let expected = null, gaps = 0, dups = 0, resets = 0, applied = 0;
  const ac = new AbortController();
  const done = follow(url, (type, d) => {
    if (type === "reset") { resets++; for (const k in mine) delete mine[k]; Object.assign(mine, d.state); expected = d.seq + 1; return; }
    if (expected !== null && d.seq < expected) { dups++; return; }
    if (expected !== null && d.seq > expected) gaps++;
    mine[d.key] = { value: d.value, seq: d.seq }; expected = d.seq + 1; applied++;
  }, { signal: ac.signal });

  // Producer: 400 writes, sometimes in bursts. Chaos: kill every open connection now and then,
  // and once keep the client away for long enough that the log moves past its cursor.
  for (let i = 0; i < 400; i++) {
    write(`k${i % 7}`, i);
    if (i % 60 === 59) server.closeAllConnections();                  // short outage: resume from the cursor
    if (i === 200) { server.closeAllConnections(); for (let j = 0; j < 120; j++) write(`k${j % 7}`, 1000 + j); } // long outage: > RETAIN events missed
    if (i % 5 === 0) await new Promise((r) => setTimeout(r, 3));
  }
  await new Promise((r) => setTimeout(r, 300));
  ac.abort(); await done; server.closeAllConnections(); server.close();

  const same = JSON.stringify(Object.keys(state).sort().map((k) => [k, state[k].value])) ===
               JSON.stringify(Object.keys(mine).sort().map((k) => [k, mine[k].value]));
  console.log(`server seq=${seq}  applied=${applied}  resets=${resets}  gaps=${gaps}  duplicates=${dups}  client state equals server state: ${same}`);
}

// A snapshot is a pair (state, cursor). Read them in the wrong order and an update can vanish.
// Run: node snapshot_order.mjs
const rng = (a) => () => { a = (a + 0x6d2b79f5) | 0; let t = Math.imul(a ^ (a >>> 15), 1 | a); t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; return ((t ^ (t >>> 14)) >>> 0) / 2 ** 32; };

function trial(order, rand) {
  const log = [], state = {}; let seq = 0;
  const write = () => { const key = "k" + Math.floor(rand() * 3); log.push({ seq: ++seq, key, value: seq }); state[key] = { value: seq, seq }; };
  const writes = (n) => { for (let i = 0; i < n; i++) write(); };
  writes(3);

  // Reading the state and reading the cursor are two steps. Writes can land between them (0 to 3 here).
  let snapState, cursor;
  if (order === "state, then cursor") { snapState = structuredClone(state); writes(Math.floor(rand() * 4)); cursor = seq; }
  else                                { cursor = seq; writes(Math.floor(rand() * 4)); snapState = structuredClone(state); }
  writes(Math.floor(rand() * 2));                                   // and a few more after the snapshot

  // The client installs the snapshot, then replays every log entry after the cursor,
  // skipping entries that are not newer than what it already holds for that key.
  const mine = structuredClone(snapState);
  for (const ev of log) if (ev.seq > cursor && ev.seq > (mine[ev.key]?.seq ?? 0)) mine[ev.key] = { value: ev.value, seq: ev.seq };
  return JSON.stringify(Object.entries(mine).sort()) === JSON.stringify(Object.entries(state).sort());
}

for (const order of ["state, then cursor", "cursor, then state"]) {
  const rand = rng(7); let ok = 0;
  for (let i = 0; i < 1000; i++) ok += trial(order, rand) ? 1 : 0;
  console.log(`${order.padEnd(20)} client matches server in ${ok}/1000 trials`);
}