async client: request timeout does not cover outbound queue and socket write

[PiotrProkop]

Description

The async ttrpc client can remain blocked beyond Request.timeout_nano when the connection writer stops making progress.

Client::request registers the stream and waits to enqueue the message before starting the timeout:

self.streams.lock()?.insert(stream_id, tx);

self.req_tx
    .send(SendingMessage::new(msg))
    .await?;

let result = tokio::time::timeout(
    Duration::from_nanos(timeout_nano as u64),
    rx.recv(),
)
.await?;

Relevant source:

• Client::request
• Connection writer task

All messages for a connection are serialized through one writer task and a bounded mpsc channel with a capacity of 100. The writer calls GenMessage::write_to, whose write_all/flush operations have no deadline.

If the peer remains connected but stops reading:

1. The writer task blocks in write_to.
2. The writer stops draining the outbound channel.
3. The 100 outbound queue slots fill.
4. Later Client::request calls block in req_tx.send(...).await.
5. Their timeout_nano is never started because execution has not reached tokio::time::timeout.
6. These calls can therefore remain pending indefinitely.

This is connection-wide head-of-line blocking: one blocked socket write prevents unrelated RPCs from being submitted.

Timed-out requests also leak stream registrations

The client inserts the stream into self.streams before queue admission. There is no cleanup guard covering all early-return and cancellation paths.

If:

• queue admission is canceled externally;
• response waiting times out; or
• the request future is dropped;

the corresponding entry can remain in self.streams until a response arrives or the complete connection disconnects.

A successfully enqueued request that times out also remains in the outbound queue. If the writer later recovers, the expired request may still be transmitted after its caller has returned.

Expected behavior

A non-zero Request.timeout_nano should bound the complete client-side request lifecycle:

stream registration
  -> outbound queue admission
  -> socket write
  -> response wait

Specifically:

• Waiting for outbound queue capacity must honor the request deadline.
• A queued request that expires before being written must be discarded.
• A request must not be transmitted after its deadline if no bytes were written.
• Cancellation during a possibly partial frame write must close or poison the connection.
• Every timeout, cancellation, send failure, and dropped future must remove its stream registration.
• Other callers must not remain indefinitely blocked behind an expired request.

Actual behavior

The timeout only wraps rx.recv() after queue admission. It does not cover:

• waiting for space in the outbound channel;
• the actual socket write;
• streaming sends;
• stream initialization; or
• cleanup of self.streams.

Streaming paths

The same writer is used by streaming operations.

new_stream waits for req_tx.send(...).await without applying Request.timeout_nano. StreamSender::send and close_send wait for both queue admission and the writer result without a request deadline.

These paths should be included in the fix and regression coverage.

Async server behavior

The async server uses the same Connection writer task and another bounded 100-message channel.

The per-request timeout wraps the service handler, but sending the response happens afterward. If the writer is blocked and its response queue fills, completed handler tasks can remain blocked while enqueueing responses even though the request timeout has expired.

The server response and shutdown paths should therefore also be audited for the same transport invariant.

Sync client behavior

The sync client has a related manifestation:

• It uses one sender thread for all requests.
• write_message performs blocking writes without a deadline.
• Requests are submitted through an unbounded std::sync::mpsc channel.
• recv_timeout lets individual callers return, but it does not cancel their queued or in-progress messages.
• Continued calls can grow the outbound queue without bound while the sender thread remains blocked.

Relevant source:

• sync::Client::request
• write_message

The async implementation is the primary subject of this issue, but the sync implementation should either be fixed together or tracked separately.

Impact

A peer that remains connected but stops reading can cause:

• all unrelated RPCs on that connection to be starved;
• async request futures to outlive their configured timeout;
• unbounded growth of the sync outbound queue;
• accumulation of stale stream-map entries;
• expired requests to be transmitted later;
• streaming operations to hang indefinitely; and
• connection shutdown to wait on blocked response tasks.

This is particularly harmful when lifecycle/control RPCs and diagnostic traffic share one long-lived ttrpc connection.

[PiotrProkop]

This is simillar issue as in containerd/ttrpc#258