Long polling

Deep dive

How it works

In: poll(cursor, wait) · Out: events with seq > cursor in order, or an empty list on timeout, or "reset cursor" · State: server: append-only log with sequence numbers plus a waiter list per topic; client: the last seq it has processed · Cost: O(1) per held request; one wake per publish per waiter

1. Client sends GET /events?after=C&wait=W (C = last seq it processed; 0 at start).
2. Server locks the topic's log. If it has events with seq > C, it answers at once with up to max of them.
3. Otherwise it parks the request for at most W, below the lowest proxy or load-balancer idle timeout on the path.
4. A publish appends under the same lock and wakes the topic's waiters; each re-checks the log against its own cursor.
5. On timeout the server answers 200 with an empty list (or 204), never an error.
6. Client processes the batch, sets C to the last seq, re-asks at once. Events published while no request was open are found in step 2: batching falls out of the cursor.
7. A cursor older than the retained log, or newer than it (server restarted), gets "reset": refetch a snapshot, restart from its seq.
8. After errors the client backs off with jitter.

The log, not a per-client queue, is the state, so the gap between response and re-ask is harmless. Advance C only after processing: Telegram confirms updates when offset passes their id (at least once).

Code

use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::{Duration, Instant};

#[derive(Clone, Debug, PartialEq)]
pub struct Event { pub seq: u64, pub data: String }

/// The client's cursor is no longer usable: it fell off the retained log, or comes
/// from a different server epoch. It must resync from a snapshot, then poll again.
#[derive(Debug, PartialEq)]
pub struct ResetCursor { pub oldest: u64 }

struct Log { events: VecDeque<Event>, next_seq: u64, cap: usize }

pub struct Topic { log: Mutex<Log>, news: Condvar }

impl Topic {
    pub fn new(cap: usize) -> Self {
        Topic { log: Mutex::new(Log { events: VecDeque::new(), next_seq: 1, cap }), news: Condvar::new() }
    }

    /// Append under the lock, then wake every waiter; each re-checks its own cursor.
    pub fn publish(&self, data: &str) -> u64 {
        let mut l = self.log.lock().unwrap();
        let seq = l.next_seq;
        l.next_seq += 1;
        l.events.push_back(Event { seq, data: data.to_string() });
        if l.events.len() > l.cap { l.events.pop_front(); }
        drop(l);
        self.news.notify_all();
        seq
    }

    /// Events with seq > `after`, oldest first, at most `max`. Blocks up to `wait`
    /// if there are none; an empty Vec means "timed out, ask again".
    pub fn poll(&self, after: u64, wait: Duration, max: usize) -> Result<Vec<Event>, ResetCursor> {
        let l = self.log.lock().unwrap();
        let oldest = l.events.front().map_or(l.next_seq, |e| e.seq);
        if after >= l.next_seq { return Err(ResetCursor { oldest }); }   // cursor from the future
        // Loops over spurious wakeups and recomputes the remaining wait itself.
        let (l, _) = self.news.wait_timeout_while(l, wait, |l| l.next_seq - 1 <= after).unwrap();
        let oldest = l.events.front().map_or(l.next_seq, |e| e.seq);
        if after + 1 < oldest { return Err(ResetCursor { oldest }); }   // trimmed past the cursor
        Ok(l.events.iter().skip((after + 1 - oldest) as usize).take(max).cloned().collect())
    }
}

fn main() {
    let t = Arc::new(Topic::new(1000));
    let s = Duration::from_secs(5);

    // 1. a waiter wakes on publish and gets exactly that event
    let w = { let t = t.clone(); thread::spawn(move || t.poll(0, s, 100)) };
    thread::sleep(Duration::from_millis(50));
    t.publish("a");
    assert_eq!(w.join().unwrap().unwrap(), vec![Event { seq: 1, data: "a".into() }]);

    // 2. timeout returns empty, after about the wait
    let t0 = Instant::now();
    assert!(t.poll(1, Duration::from_millis(100), 100).unwrap().is_empty());
    assert!(t0.elapsed() >= Duration::from_millis(100));

    // 3. a slow client resumes from its cursor and gets everything missed, in order, in batches
    for i in 2..=7 { t.publish(&format!("e{i}")); }
    let (mut cur, mut got) = (1, vec![]);
    loop {
        let b = t.poll(cur, Duration::ZERO, 4).unwrap();
        if b.is_empty() { break; }
        cur = b.last().unwrap().seq;
        got.extend(b);
    }
    assert_eq!(got.iter().map(|e| e.seq).collect::<Vec<_>>(), (2..=7).collect::<Vec<_>>());

    // 4. one publish wakes many waiters; measure publish -> wake on this machine
    for n in [1usize, 100] {
        let mut worst = Duration::ZERO;
        for _ in 0..20 {
            let cur = t.poll(0, Duration::ZERO, 1000).unwrap().last().unwrap().seq;
            let hs: Vec<_> = (0..n).map(|_| { let t = t.clone();
                thread::spawn(move || { let r = t.poll(cur, s, 10).unwrap(); (r, Instant::now()) }) }).collect();
            thread::sleep(Duration::from_millis(30)); // let them park
            let t0 = Instant::now();
            let seq = t.publish("x");
            for h in hs { let (r, woke) = h.join().unwrap(); assert_eq!(r[0].seq, seq); worst = worst.max(woke - t0); }
        }
        println!("{n} waiters: worst publish->wake over 20 rounds = {worst:?}");
    }

    // 5. cursors that cannot be served are reported, not silently skipped
    let small = Topic::new(2);
    for i in 0..5 { small.publish(&i.to_string()); }
    assert_eq!(small.poll(1, Duration::ZERO, 10), Err(ResetCursor { oldest: 4 }));
    assert_eq!(small.poll(99, Duration::ZERO, 10), Err(ResetCursor { oldest: 4 }));
    println!("ok");
}

Measured by me on Apple M4 Pro (release build, thread per waiter, 3 runs): worst publish-to-wake over 20 rounds was 29-78 µs for 1 waiter and 0.96-1.17 ms for 100 waiters woken by one notify_all; mostly scheduler cost.

Complexity and constants

• Publish: O(1) append under the lock, then one wake per waiter. Poll: O(1) offset arithmetic, O(batch) clone.
• Memory: the bounded log plus, per held request, an OS thread (2 MiB virtual stack by default in Rust std) or one small state struct on an event loop. Per-request resources limit gateways and proxies (RFC 6202).
• Idle traffic: 10,000 clients at T = 30 s is about 333 req/s, against 10,000 req/s for 1 s short polling.
• Latency: average about one network transit, worst about three (response, re-ask, response) (RFC 6202).

Efficiency

• Cost: a parked request is a socket plus a waiter entry; a wake is a lock and a response write: tens of µs for one waiter, about 1 ms for 100 thread waiters (measured above, M4 Pro). Idle cost is N/T requests/s, each with full HTTP headers. Bound: per-request memory and scheduler on the server; headers and RTT on the wire.
• Levers:
  • longer hold T — requests/s falls as 1/T (arithmetic above) — until the lowest idle timeout on the path
  • batch with max and a cursor — one response carries everything queued (RFC 6202) — bursty topics, slow clients
  • one waiter list per topic — a publish wakes only that topic's waiters — many topics per process
  • event loop instead of thread per waiter — asynchronous servers need only slightly more than normal traffic (RFC 6202); see epoll — thousands of clients
  • random extra wait — spreads re-asks; Consul adds up to wait/16 — many clients waking on one change
• Naive → standard → state of the art:
  • short poll every second · N req/s, up to 1 s late, mostly empty
  • long poll with cursor and hold under the idle timeout · N/T req/s, about one transit late
  • SSE or WebSocket on a persistent stream · one connection, no per-event request headers

Implementation notes

• Use wait_timeout_while (or while !cond { wait }): spurious wakeups and wakeups for other cursors are normal.
• Notify after releasing the lock, so woken threads do not immediately block on the mutex.
• Send Cache-Control: no-cache (RFC 6202); a cursor in the URL makes retries idempotent.
• Server wait < proxy idle timeout < client read timeout. Defaults: nginx proxy_read_timeout 60 s; ALB idle timeout 60 s (range 1-4000 s).
• Consul: index plus wait (default 5 min, max 10 min); a return does not guarantee a change.
• Kafka's consumer fetch has this shape: the broker holds a fetch until fetch.min.bytes is available or fetch.max.wait.ms (default 500) passes.

Pitfalls

• Polling by time ("events since 12:00:03") or without a cursor: events published between two requests are lost.
• A hold longer than a proxy's idle timeout: 504/502 and a retry storm. Test through the real path.
• Cursor moves backwards (Consul: snapshot restore, deleted highest-index key): detect it and reset, or the client waits for updates that never come. Consul also warns that an index below 1 makes clients busy-loop.
• Re-asking at once after an error hammers a struggling server: back off with jitter.
• Everyone re-asks together after a restart or after one publish wakes 10,000 waiters: add jitter and cap held requests with backpressure.
• Advancing the cursor before processing loses events; an unbounded log eats memory.

Variants and alternatives

In production

Telegram getUpdates (updates kept at most 24 h), Socket.IO/Engine.IO (default transport), CometD (Bayeux requires long polling), Consul blocking queries, Kafka consumer fetches.

Implementations

Sources

• Loreto, Saint-Andre, Salsano, Wilkins, Known Issues and Best Practices for the Use of Long Polling and Streaming in Bidirectional HTTP, RFC 6202, 2011. RFC
• Telegram, Bot API: getUpdates. docs
• HashiCorp, Consul API: Blocking Queries. docs
• Apache Kafka, Consumer configs: fetch.max.wait.ms. docs
• Socket.IO, How it works. docs · CometD, Reference. docs
• nginx, proxy_read_timeout. docs · AWS, ALB idle timeout. docs