Per-User Fair Queuing for SSE Broadcasts

Part of Rate Limiting & Backpressure Handling, under Backend Stream Generation & Connection Management.

A broadcast loop that writes to every subscriber in order gives the slowest consumer control over everyone else’s latency. Fair queuing puts each connection in its own lane so one stalled reader costs only that reader.

Symptom & Developer Intent

  • Event latency for all users tracks the worst-connected user rather than the median.
  • A single client on a poor mobile connection visibly slows a dashboard for everyone.
  • One account subscribed to many topics receives a disproportionate share of write capacity.
  • Memory grows because the broadcast loop buffers for whoever cannot keep up.
  • p99 delivery latency is an order of magnitude worse than p50 with no obvious cause.

The intent is delivery whose latency is independent of the worst consumer, with a bounded memory cost per connection.

Root Cause Analysis

The naive broadcast is a loop over subscribers with a synchronous write per subscriber. When one write blocks — because that client’s socket buffer is full — the loop stalls, and every subscriber after it in the iteration waits. Worse, the position in the iteration is arbitrary, so which users suffer changes per event and the effect looks random.

One slow reader stalls the broadcast loop Sequence showing a synchronous broadcast loop blocking on a slow subscriber so every subscriber after it in the iteration waits. One slow reader stalls the broadcast loop Broadcast Fast client Slow client write → accepted write → blocks (buffer full) everyone after waits loop stalled on one socket
Position in the iteration is arbitrary, so which users suffer changes per event — which is why the symptom looks random rather than like one bad client.

Making writes asynchronous removes the stall but replaces it with unbounded buffering: the runtime queues bytes for the slow socket while the loop races ahead, and memory grows until something breaks. This is the failure mode described in Handling Slow Consumers with SSE Backpressure.

Fair queuing separates the two concerns. Publication enqueues into a small per-connection buffer and returns immediately; a separate drain loop moves data from each queue to its socket at whatever rate that socket accepts. A full queue applies a drop policy to that connection alone, so a slow consumer degrades only itself. Round-robin draining then guarantees that a connection with a large backlog cannot monopolise the drain loop.

Step-by-Step Resolution

Publication and draining, separated Pipeline where publication enqueues into per-connection buffers and a round-robin drain loop moves data to each socket at its own pace. Publication and draining, separated a blocked lane waits for drain while every other lane keeps moving Publish enqueue only Per-conn queue bounded + policy Round robin 8 frames per pass Socket at its own pace never blocks drop when full fair budget
Publication never touches a socket, so it cannot block; draining is budgeted per pass, so a large backlog cannot monopolise the loop.

Step 1 — Give every connection a bounded queue 

class ConnectionQueue {
  constructor(res, { capacity = 64, policy = 'drop_oldest' } = {}) {
    this.res = res;
    this.capacity = capacity;
    this.policy = policy;
    this.items = [];
    this.dropped = 0;
    this.blocked = false;         // true when the socket last refused a write
  }

  enqueue(frame) {
    if (this.items.length >= this.capacity) {
      if (this.policy === 'drop_oldest') this.items.shift();
      else { this.dropped++; return false; }        // drop_newest
      this.dropped++;
    }
    this.items.push(frame);
    return true;
  }
}

Step 2 — Make publication a fan-out of enqueues, never of writes 

// O(subscribers) memory writes, zero socket I/O — this never blocks.
function broadcast(topic, event) {
  const frame = `id: ${event.id}\nevent: ${event.type}\ndata: ${event.json}\n\n`;
  for (const q of subscribers.get(topic) ?? []) q.enqueue(frame);
}

Step 3 — Drain round-robin so no connection monopolises the loop 

// One pass writes at most BUDGET frames per connection, then moves on.
const BUDGET = 8;

function drainOnce(queues) {
  for (const q of queues) {
    if (q.blocked) continue;                     // waiting for a drain event
    let written = 0;
    while (q.items.length && written < BUDGET) {
      const accepted = q.res.write(q.items[0]);
      q.items.shift();
      written++;
      if (!accepted) {                           // socket buffer full: stop this lane
        q.blocked = true;
        q.res.once('drain', () => { q.blocked = false; });
        break;
      }
    }
  }
}

setInterval(() => drainOnce(allQueues()), 10);

The per-pass budget is what makes it fair: a connection with two hundred queued frames gets eight per pass, exactly like everyone else.

Step 4 — Add weights where fairness should not be equal 

// Weighted round robin: give some connections a larger per-pass budget.
function budgetFor(q) {
  if (q.tier === 'realtime') return 16;          // trading views, live ops dashboards
  if (q.tier === 'background') return 2;         // hidden tabs, low-priority feeds
  return 8;
}

Weighting by declared client state — a hidden tab asking for less — is more honest than weighting by account tier, and it reduces total work rather than redistributing it.

Step 5 — Enforce per-user limits across connections

Fairness per connection is gameable by opening more connections. Cap and share per user.

const perUser = new Map();                       // userId → Set<ConnectionQueue>

function registerConnection(userId, q) {
  const set = perUser.get(userId) ?? new Set();
  if (set.size >= 5) {
    q.res.write('event: rejected\ndata: {"reason":"too_many_connections"}\n\n');
    q.res.end();
    return false;
  }
  set.add(q);
  perUser.set(userId, set);
  // Share one user's budget across their connections so five tabs are not five times the capacity.
  for (const other of set) other.share = 1 / set.size;
  return true;
}

Step 6 — Record what the policy dropped 

metrics.inc('sse_events_dropped_total', { policy: q.policy });
metrics.observe('sse_queue_depth', q.items.length);

An unmeasured drop is indistinguishable from a bug, and a drop policy nobody can see is one nobody will trust.

Validation & Monitoring

The test that matters is a deliberately slow consumer alongside fast ones.

Delivery latency for fast readers, with one slow consumer present Bar chart of p50 and p99 delivery latency for fast readers under a synchronous broadcast loop versus per-connection fair queuing. Delivery latency for fast readers, with one slow consumer present ten fast readers, one reader at 1 byte/s Loop, p50 4 ms Loop, p99 2 400 ms Fair queue, p50 4 ms Fair queue, p99 9 ms delivery latency for fast readers
The p99-to-p50 ratio is the fairness measure: close to one means the tail is no longer being set by the worst-connected user.
# 1. Ten fast readers and one that reads a byte per second.
for i in $(seq 1 10); do curl -N -s localhost:8080/events > /dev/null & done
curl -N -s --limit-rate 1 localhost:8080/events > /dev/null &

# 2. Publish and measure delivery latency for the FAST readers.
curl -sX POST localhost:8080/admin/publish -d '{"topic":"prices"}'
curl -s localhost:9090/metrics | grep sse_publish_to_write_seconds

# 3. The slow reader should be dropping; the fast ones should not.
curl -s localhost:9090/metrics | grep sse_events_dropped_total

Fast-reader latency must stay flat while the slow reader’s queue fills and drops. If it does not, publication is still doing socket I/O somewhere.

# Fairness check: p99 delivery latency should not be far above p50.
histogram_quantile(0.99, sum(rate(sse_publish_to_write_seconds_bucket[5m])) by (le))
  / histogram_quantile(0.50, sum(rate(sse_publish_to_write_seconds_bucket[5m])) by (le))

# Queue pressure across the fleet.
histogram_quantile(0.95, sum(rate(sse_queue_depth_bucket[5m])) by (le))

A p99-to-p50 ratio near one is the goal — it means the tail is no longer being set by the worst consumer.

Production Checklist

Frequently Asked Questions

Why not just write asynchronously to every subscriber?

Because asynchronous writes remove the stall but not the pressure: the runtime buffers for whoever cannot keep up, and memory grows without bound. A bounded per-connection queue with an explicit drop policy makes the cost visible and finite, which is the whole point of fair queuing.

What queue depth should each connection get?

Enough to absorb a normal burst and no more — typically 16 to 64 events. Multiply by your connection count to see the real cost: 64 events of 500 bytes across 10 000 connections is roughly 320 MB of headroom, which is usually acceptable, while 1 000 events per connection is not.

Is per-connection fairness enough on its own?

No, because a user can open more connections and claim a larger share. Cap connections per user and divide that user's drain budget across their connections, so five tabs receive the same total capacity as one.

How do I choose the drop policy?

From the data. Drop oldest for telemetry and prices, where the newest value is what matters. Coalesce by key for state snapshots, where several updates collapse into one. For ordered data that cannot lose events — ledgers, audit trails — disconnect and force a resync instead of dropping silently.