Streaming SSE by Converting Backend Cursor Events into Frontend Flow Control

I stumbled into a weird performance cliff while building a “live search” feature: the backend was streaming results incrementally, but the frontend sometimes got ahead of the browser’s ability to paint. The symptom was dramatic—CPU spikes and delayed UI updates—even though network throughput looked fine.

What fixed it for me was a very specific architecture pattern: use Server-Sent Events (SSE) on the backend, but add explicit flow control in the frontend by pausing the fetch stream consumption based on UI backpressure. I did this by wiring SSE “cursor events” into a bounded async queue and only resuming the reader when the UI drains the queue.

Below is the exact setup I ended up with.

---

The architecture I built

Why SSE

SSE (Server-Sent Events) is a HTTP-based streaming mechanism where the server pushes text events to the browser over a single long-lived connection. It’s simpler than WebSockets when you only need server-to-client streaming.

The hidden failure mode

I initially had:

• Backend: stream event: result lines as they’re found
• Frontend: append each result to the DOM as soon as it arrives

When results arrive faster than the DOM can update, the browser’s event loop backs up. In practice, this means your “stream reader” keeps consuming data, but the UI rendering can’t keep up—so memory grows and latency increases.

The fix: bounded queue + controlled consumption

I used:

• Backend: stream results using SSE (one event per result)
• Frontend: parse SSE lines, push results into a bounded queue
• UI renderer: drains the queue at a controlled pace
• Stream reader: pauses/resumes based on queue size

This turns “best effort streaming” into a predictable pipeline.

---

Backend: SSE endpoint that emits cursor-based events

I used Node.js with Express. The core idea is:

• For each item in an in-memory “database,” emit:
  • event: result with JSON payload
  • event: cursor with progress metadata
• Support cancellation via client disconnect.

Backend code (Node.js + Express)

// server.js
import express from "express";
import cors from "cors";

const app = express();
app.use(cors());

function sleep(ms) {
  return new Promise((r) => setTimeout(r, ms));
}

// Fake dataset
const DATASET = Array.from({ length: 5000 }, (_, i) => ({
  id: i + 1,
  title: `Item ${i + 1}`,
}));

app.get("/api/stream-search", async (req, res) => {
  const query = String(req.query.q ?? "").toLowerCase();

  // SSE headers
  res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
  res.setHeader("Cache-Control", "no-cache, no-transform");
  res.setHeader("Connection", "keep-alive");

  // Helps some proxies flush immediately
  res.flushHeaders?.();

  let disconnected = false;
  req.on("close", () => {
    disconnected = true;
  });

  // Emit a comment line to establish the stream promptly (optional but common)
  res.write(`: stream-start\n\n`);

  // Cursor-based streaming (cursor = last emitted index)
  let cursor = 0;

  // Filter lazily-ish (still in-memory here, but models real DB cursor behavior)
  for (let i = 0; i < DATASET.length; i++) {
    if (disconnected) return;

    const row = DATASET[i];
    if (query && !row.title.toLowerCase().includes(query)) continue;

    // Simulate backend work
    await sleep(1);

    cursor = i;

    // SSE event: result
    res.write(
      [
        `event: result`,
        `data: ${JSON.stringify(row)}`,
        ``,
      ].join("\n")
    );

    // SSE event: cursor (progress metadata)
    res.write(
      [
        `event: cursor`,
        `data: ${JSON.stringify({ cursor })}`,
        ``,
      ].join("\n")
    );
  }

  // Finish
  res.write(`event: done\ndata: ${JSON.stringify({ cursor })}\n\n`);
  res.end();
});

app.listen(3000, () => {
  console.log("SSE server running on http://localhost:3000");
});

Run it:

node server.js

---

Frontend: turn SSE text into a bounded async queue

The key part is that I did not rely on EventSource directly. The browser’s EventSource API doesn’t give me precise control over reading and pausing when my UI can’t keep up.

Instead, I:

1. Fetch the SSE endpoint as a stream (fetch(...); response.body.getReader()).
2. Parse the SSE protocol lines manually.
3. Push parsed results into a bounded queue.
4. Pause reading when the queue is full, resume when it’s drained.

A minimal SSE parser for my specific event set

I only handle:

• event: result
• event: cursor
• event: done

The parsing logic reads chunks, splits into lines, and builds events.

Frontend code (plain JS + DOM)

Create index.html:

<!doctype html>
<html>
  <head>
    <meta charset="utf-8" />
    <title>SSE Cursor Streaming with Flow Control</title>
    <style>
      body { font-family: system-ui, Arial; margin: 24px; }
      #results { margin-top: 12px; display: grid; gap: 6px; max-width: 520px; }
      .row { padding: 8px; border: 1px solid #ddd; border-radius: 8px; }
      #meta { color: #444; font-size: 14px; }
      button { margin-top: 12px; }
    </style>
  </head>
  <body>
    <h1>SSE Cursor Streaming with Frontend Flow Control</h1>

    <div>
      <input id="q" placeholder="Type to filter titles" value="1" />
      <button id="start">Start stream</button>
      <button id="stop">Stop</button>
    </div>

    <div id="meta">Queue: <span id="queueSize">0</span></div>
    <div id="results"></div>

    <script type="module" src="./main.js"></script>
  </body>
</html>

And main.js:

// main.js

class BoundedQueue {
  constructor(maxSize) {
    this.maxSize = maxSize;
    this.items = [];
    this.waitingDequeue = [];
    this.waitingEnqueue = [];
    this.closed = false;
  }

  get size() {
    return this.items.length;
  }

  close() {
    this.closed = true;
    // Wake up any waiters so the system can exit cleanly
    while (this.waitingEnqueue.length) this.waitingEnqueue.shift()();
    while (this.waitingDequeue.length) this.waitingDequeue.shift()();
  }

  async enqueue(item) {
    if (this.closed) return;

    // Wait until space is available
    while (!this.closed && this.items.length >= this.maxSize) {
      await new Promise((resolve) => this.waitingEnqueue.push(resolve));
    }
    if (this.closed) return;

    this.items.push(item);

    // Wake up one dequeuer
    const resolveDequeue = this.waitingDequeue.shift();
    if (resolveDequeue) resolveDequeue();
  }

  async dequeue() {
    while (!this.closed && this.items.length === 0) {
      await new Promise((resolve) => this.waitingDequeue.push(resolve));
    }
    if (this.items.length === 0) return null;

    const item = this.items.shift();

    // Wake up one enqueuer due to new space
    const resolveEnqueue = this.waitingEnqueue.shift();
    if (resolveEnqueue) resolveEnqueue();

    return item;
  }
}

function parseSSEBuffer(lines, state) {
  // state: { currentEvent, currentDataLines }
  for (const line of lines) {
    // SSE uses empty line to terminate an event
    if (line === "") {
      const { currentEvent, currentDataLines } = state;
      const data = currentDataLines.join("\n");
      if (currentEvent && data !== undefined) {
        state.events.push({ event: currentEvent, data });
      }
      state.currentEvent = null;
      state.currentDataLines = [];
      continue;
    }

    if (line.startsWith("event:")) {
      state.currentEvent = line.slice("event:".length).trim();
      continue;
    }

    if (line.startsWith("data:")) {
      state.currentDataLines.push(line.slice("data:".length).trim());
      continue;
    }

    // Ignore other fields (like id:, retry:, comments)
  }
}

async function streamSearchWithFlowControl({ url, query, maxQueueSize }) {
  const controller = new AbortController();

  const queue = new BoundedQueue(maxQueueSize);

  const metaQueueSize = document.getElementById("queueSize");
  const resultsEl = document.getElementById("results");

  let lastCursor = null;
  let done = false;

  // UI drain loop: render at a controlled pace
  const uiLoop = (async () => {
    // Render in small batches to avoid DOM thrash
    const BATCH_SIZE = 15;

    while (!done) {
      const batch = [];
      for (let i = 0; i < BATCH_SIZE; i++) {
        const item = await queue.dequeue();
        if (item === null) break;
        batch.push(item);
      }

      if (batch.length) {
        for (const row of batch) {
          const div = document.createElement("div");
          div.className = "row";
          div.textContent = `${row.id}: ${row.title}`;
          resultsEl.appendChild(div);
        }
      }

      // Let the browser breathe (especially important when results are dense)
      await new Promise((r) => setTimeout(r, 0));

      metaQueueSize.textContent = String(queue.size);
    }
  })();

  // Stream reader + parser
  const readerLoop = (async () => {
    const resp = await fetch(`${url}?q=${encodeURIComponent(query)}`, {
      headers: { Accept: "text/event-stream" },
      signal: controller.signal,
    });

    if (!resp.ok || !resp.body) {
      throw new Error(`Bad response: ${resp.status}`);
    }

    const streamReader = resp.body.getReader();
    const decoder = new TextDecoder("utf-8");

    let buffer = "";
    const sseState = {
      currentEvent: null,
      currentDataLines: [],
      events: [],
    };

    try {
      while (true) {
        // Flow control: stop reading when queue is full
        if (queue.size >= maxQueueSize) {
          await new Promise((resolve) => setTimeout(resolve, 10));
          continue;
        }

        const { value, done: streamDone } = await streamReader.read();
        if (streamDone) break;

        buffer += decoder.decode(value, { stream: true });

        // Process complete lines; keep remainder
        const lines = buffer.split(/\r?\n/);
        buffer = lines.pop() ?? "";

        parseSSEBuffer(lines, sseState);

        // Handle parsed events
        while (sseState.events.length) {
          const { event, data } = sseState.events.shift();

          if (event === "result") {
            const row = JSON.parse(data);
            // Enqueue can wait if the queue fills (bounded backpressure)
            await queue.enqueue(row);
          } else if (event === "cursor") {
            lastCursor = JSON.parse(data).cursor;
          } else if (event === "done") {
            done = true;
            queue.close();
            // ensure UI loop exits promptly
            break;
          }
        }
      }
    } finally {
      done = true;
      queue.close();
    }
  })();

  return {
    stop() {
      controller.abort();
      done = true;
      queue.close();
    },
    uiLoop,
    readerLoop,
    get lastCursor() {
      return lastCursor;
    },
  };
}

// Wire up page controls
let current = null;

document.getElementById("start").addEventListener("click", async () => {
  const q = document.getElementById("q").value;

  // Reset UI
  document.getElementById("results").innerHTML = "";

  // Start a new stream
  current?.stop();
  current = await streamSearchWithFlowControl({
    url: "http://localhost:3000/api/stream-search",
    query: q,
    maxQueueSize: 120, // bounded queue size is the whole point
  });
});

document.getElementById("stop").addEventListener("click", () => {
  current?.stop();
});

Run it by serving the static files (e.g., http-server or any local static server). Your browser will call the backend at localhost:3000.

---

What happens when you run this

Without flow control (what I originally did)

• Results arrive fast from SSE
• The renderer appends to the DOM for every event
• DOM work piles up
• Memory grows because updates can’t keep up
• UI becomes sluggish even though network is “fine”

With the bounded queue + paused reading

• When queue.size hits maxQueueSize, the reader loop temporarily stops reading more chunks
• The UI drain loop keeps rendering in batches
• Once the queue drops below the limit, streaming resumes
• The app stays responsive because work is bounded and coordinated

The important architectural point is that I used SSE for incremental transport, but I added a second layer of backpressure on the frontend.

---

Why cursor events matter in this design

The cursor event is useful even if you don’t display it. In real systems, cursor metadata lets you:

• reconnect/resume after interruption
• correlate UI state with backend progress
• debug “why am I not seeing results?” scenarios

In my implementation, the backend updates cursor for each emitted match, and the frontend tracks lastCursor for potential diagnostics.

---

Conclusion

I learned that SSE streaming alone doesn’t guarantee smooth UX: the missing piece is flow control across the transport/UI boundary. By parsing SSE manually, feeding results into a bounded async queue, and only reading from the stream when the UI can drain it, I turned an unstable live-search pipeline into a predictable architecture. The combination—cursor-based SSE + bounded queue backpressure + batched DOM rendering—is what finally eliminated the performance cliff.