A subscriber that is draining, but slower than the producer, is disconnected mid-stream — silently, and indistinguishably from a crash

[arohou]

Crate: agent-client-protocol-http 1.0.1 (also present in 2.0.0, current latest)
Affects: both the WebSocket and the SSE/HTTP transports — they share OutboundStream::push

Summary

OutboundStream::push gives each subscriber a bounded queue and, on a full try_send, removes
the subscriber:

// src/connection.rs:85
pub(crate) const OUTBOUND_STREAM_CAPACITY: usize = 1024;

// src/connection.rs:61-70
state.subscribers.retain(|subscriber| match subscriber.try_send(msg.clone()) {
    Ok(()) => true,
    Err(mpsc::error::TrySendError::Full(_)) => {
        debug!("outbound subscriber queue full; closing subscriber stream");
        false
    }
    Err(mpsc::error::TrySendError::Closed(_)) => false,
});

I can see from sse_closes_slow_subscriber_before_skipping_messages in src/http_server.rs that
this is deliberate, and I agree with the intent: silently skipping messages in the middle of a
JSON-RPC stream would be worse. This report is not asking for messages to be dropped instead.

The problem is that the policy cannot distinguish a peer that is stuck from a peer that is
merely slower than a burst, and it reacts to the second case in a way the client cannot
diagnose.

What it looks like from the client

An agent implementation built on this crate replays a stored conversation to a reconnecting
client as a burst of notifications — one per content block, read from a local database and
pushed as fast as it can be deserialised. For a long session that is thousands of messages and
tens of megabytes, produced far faster than any socket can drain.

The subscriber is draining. It is just slower than a producer with no flow control. Once it
falls 1024 messages behind:

• the subscriber is removed;
• the writer task's recv() returns None and breaks its loop;
• the socket is dropped with no close frame — the client sees WebSocket 1006;
• the only record is one debug! line, invisible under any normal filter;
• the server process is otherwise healthy and keeps running.

From the client there is no way to tell this from the agent process crashing. In our case it cost
several hours and five rounds of log collection with an affected user before we found the
debug! line, because every artifact we had said the backend was fine — and it was.

TLS is what makes it reproducible: it slows the writer just enough to lose the race. The exact
same store over plain ws:// completes cleanly, which is a good illustration of how
timing-dependent the current behaviour is.

Measured on one machine, replaying a synthetic session:

messages	payload	ws://	wss://
3000	0.7 MiB	clean	clean
900	92 MiB	clean	clean
1674	16 MiB	clean	clean
1674	28 MiB	clean	disconnected
1674	171 MiB	clean	disconnected

Both conditions are needed: more than OUTBOUND_STREAM_CAPACITY messages in flight, and
enough bytes for the writer to be the bottleneck.

Reproducer

rust_sdk_repro.rs in this report adds one #[tokio::test] to the module in
src/http_server.rs. It differs from the existing capacity test in exactly one way that matters:
the subscriber drains continuously, just more slowly than the producer. It asserts no message
loss, and fails on main.

rust_sdk_repro.rs.zip

Three concrete problems

1. No backpressure option. push is already async; awaiting capacity is available and is
   the correct response to a producer burst. Disconnecting is the right answer only when the peer
   has genuinely stopped consuming.
2. The disconnect is undiagnosable. No close frame, no reason code, debug! level. A client
   cannot distinguish it from a crashed agent, and a server operator sees nothing at all.
3. The threshold is a fixed message count. 1024 regardless of message size, not configurable
   through ServerOptions. 1024 one-kilobyte notifications and 1024 two-megabyte notifications
   are very different amounts of buffered memory and very different amounts of writer time.

Suggested fix

Preserve the no-skipping invariant, but pace instead of disconnecting, and keep disconnection for
peers that are actually stuck:

// in OutboundStream::push, subscriber branch
let subscribers = {
    let mut state = self.state.lock().await;
    // ... replay branch unchanged, returns early ...
    state.subscribers.clone()
};                                    // release the guard BEFORE awaiting

let mut stalled = Vec::new();
for subscriber in &subscribers {
    match tokio::time::timeout(SUBSCRIBER_SEND_TIMEOUT, subscriber.send(msg.clone())).await {
        Ok(Ok(())) => {}                            // paced, nothing lost
        Ok(Err(_)) => stalled.push(subscriber.clone()),   // receiver gone
        Err(_) => {                                 // genuinely stuck
            warn!("outbound subscriber stalled for {SUBSCRIBER_SEND_TIMEOUT:?}; closing stream");
            stalled.push(subscriber.clone());
        }
    }
}

if !stalled.is_empty() {
    let mut state = self.state.lock().await;
    state.subscribers.retain(|s| !stalled.iter().any(|dead| dead.same_channel(s)));
}

Three things I got wrong on a first attempt, in case they save you the same detour:

Do not hold the state lock across the await. It looks like the natural way to preserve
ordering, but ordering does not depend on it: every runtime call reaches push through
Connection::route_outbound, which runs on the single task spawned by start_router, so there
is exactly one producer per stream and nothing to interleave. Meanwhile subscribe() takes the
same mutex, so holding it would block handle_get inside its axum handler for the whole
timeout — precisely when an SSE client is reconnecting to restore the drain. (run_ws only polls
outbound_rx.recv(), so the WebSocket path would not have shown this.)

Keep SUBSCRIBER_SEND_TIMEOUT short. In HTTP/SSE mode one router task serialises the
connection stream and every session stream, so the timeout doubles as the worst-case stall a
single wedged subscriber imposes on the other sessions sharing that connection. try_send could
never stall anything, so this is a genuine new cost of pacing. I settled on 5 s: a 255 MiB /
4003-notification replay drains end-to-end in ~5–7 s, so no single message legitimately waits
anywhere near that long, and there is still ~2 orders of magnitude of headroom.

This is not end-to-end backpressure. The agent feeds the router over an unbounded channel, so
a slow subscriber makes the pending burst accumulate there rather than costing a dropped
connection. That is the trade I wanted, but it does move an unbounded amount of the replay into
memory. Bounding the agent→router channel would close that properly, at the cost of a deadlock
question I have not worked through.

Two smaller changes would help independently of the above, and would have saved us most of the
investigation on their own:

• raise the removal log from debug! to warn!, with the connection id;
• send a close frame with a reason before dropping the socket, so the client can report something
  better than 1006.

Making OUTBOUND_STREAM_CAPACITY configurable via ServerOptions would also let embedders tune
it, though it would not fix the underlying race on its own.

I have this running as a local patch and it does resolve the symptom: a session that reliably
died at ~1500 notifications now completes, and I have taken it to 4003 notifications / 255 MiB
over TLS without a disconnect. Happy to open a PR if the approach looks right.