Realtime fanout has unbounded head-of-line blocking across all WebSocket connections

[christabel888]

Summary

fanout_message (backend/src/realtime/fanout.rs) delivers a message to every connected WebSocket client by iterating the connection list and sending sequentially, .await-ing overflow handling inline. At any nontrivial connection count, or with even one slow/stalled consumer, this turns real-time fanout latency into a function of every other consumer's behavior — the opposite of what a real-time dashboard needs.

Location

backend/src/realtime/fanout.rs (24 lines, full function below), backend/src/realtime/policy.rs, backend/src/realtime/connection.rs

pub async fn fanout_message(
    registry: &ConnectionRegistry,
    message: Message,
) -> Result<(), String> {
    let connections = registry.get_all().await;
    for (id, sender) in connections {
        match sender.send(message.clone()) {
            Ok(_) => {}
            Err(_) => {
                let _ = handle_overflow(&id).await;
                registry.remove(&id).await;
            }
        }
    }
    Ok(())
}

Current gap / Motivation

• The loop is sequential. handle_overflow(&id).await runs inside the per-connection loop body, so if connection chore(deps): bump criterion from 0.5.1 to 0.8.2 in /contracts #3 (of, say, 5,000) is a slow consumer whose channel is full, connection chore(deps): bump the rust-patch-updates group in /backend with 3 updates #4 through chore(deps): bump rust_xlsxwriter from 0.83.0 to 0.95.0 in /backend #5,000 wait behind whatever handle_overflow does before they get their message.
• There is no per-connection latency budget or isolation: one pathological client degrades the experience of every other client on the same fanout call.
• Failure detection is purely reactive (only fires once sender.send returns an error because the bounded channel is already full) — there's no proactive slow-consumer detection (e.g. tracking per-connection queue depth trend) that could quarantine a degrading connection before it starts dropping messages for everyone else's timing, even though delivery to other connections is logically independent.
• message.clone() on every iteration for what is presumably a Message wrapping a Bytes/String payload — worth confirming this is genuinely cheap (Bytes-backed, reference-counted clone) rather than a full payload copy per connection, since at high fanout counts this is on the hot path too.

The hard part

This is a real concurrent-systems design problem, not a "add join_all" fix:

1. Bounded worst-case latency independent of N. Fanning out to N connections concurrently (e.g. via futures::future::join_all or a bounded task pool) solves the head-of-line blocking, but naively spawning N tasks per message under high message rates can itself exhaust the runtime's task queue or blow past acceptable memory/CPU overhead. The right design bounds concurrency (e.g. a semaphore-limited worker pool) while still guaranteeing no single connection's send can hold up another's beyond a bounded amount.
2. Per-connection isolation must be real, not just "not literally sequential." A design that fans out concurrently but still has a single shared mutex/registry lock taken per-send can reintroduce the same contention under a different name. ConnectionRegistry::get_all() already takes a snapshot — sends themselves must not re-contend on any single shared resource.
3. Ordering guarantees, if any, must be preserved per-connection. If downstream consumers rely on receiving updates in the order they were fanned out (plausible for a live analytics dashboard), converting to concurrent delivery must not reorder messages within a single connection's stream, even though cross-connection ordering is irrelevant.
4. Slow-consumer quarantine without false positives. A connection on a congested but otherwise healthy network path should not be permanently evicted for one transient blip, but a genuinely stalled consumer should be isolated fast enough that it stops affecting fanout latency for everyone else — this needs a real policy (e.g. a leaky-bucket or trailing-window drop-rate threshold in policy.rs), not just "remove on first full channel."

Implementation

• Convert fanout to bounded-concurrency delivery with per-connection isolation (no shared lock/state touched during the actual send).
• Move overflow/removal handling off the hot per-message path where possible (e.g. into a background task or batched cleanup), so it cannot add latency to any other connection's delivery.
• Add proactive queue-depth-trend tracking in policy.rs to quarantine degrading connections before they start dropping.
• Add a benchmark (or load test under tests/) demonstrating p99 fanout latency to N-1 healthy connections stays flat as N grows and as one connection is deliberately stalled.

Acceptance criteria

• A stalled/slow single connection cannot measurably delay delivery to any other connection (demonstrated by a test/benchmark, not just code review).
• Per-connection message ordering is preserved.
• Concurrency is bounded (no unbounded task/future spawning per fanout call).
• policy.rs gains a documented, tested slow-consumer detection policy beyond "channel send failed."
• Existing cargo test suite stays green.

[chiemeliesteve4-art]

Good day maintainer, i will redesign fanout_message with bounded concurrent delivery and per-connection isolation while preserving message ordering. I will move overflow handling off the hot path, add proactive slow-consumer detection in policy.rs, and verify the design with latency/load tests demonstrating that stalled connections do not block healthy clients. Thank you.

[GazzyLee]

good day, I’d like to take this on. I’ll redesign fanout around bounded concurrency and per-connection isolation while preserving message ordering for each client and keeping shared registry state out of the send path.

I’ll also add a documented slow-consumer policy with proactive queue tracking, move cleanup off the hot path, and add a benchmark/test proving a stalled connection doesn’t materially affect healthy clients as fanout scales.
please assign