client: take an async Stream for client-streaming requests - #227
Conversation
Client-streaming calls previously took a synchronous iterator, which cannot express uploads whose messages become available over time (paced producers, socket readers, channel consumers) without buffering the whole request or blocking a thread. call_client_stream and the generated client methods now take impl Stream<Item = Req> + Send + 'static. The stream backs the HTTP request body directly: the new EncodingBody envelope-encodes each message in poll_frame, replacing the previous bounded-channel-plus-detached-send-task pump. This hands upload liveness to the HTTP layer: backpressure is flow control, a server that ends the RPC mid-upload surfaces immediately even while the caller's stream is idle, and a server that sends response headers early while continuing to read the upload keeps receiving messages. An encode failure aborts the request and is surfaced as the call's error even when a success response races the abort. The Stream trait and a stream_iter adapter (futures::stream::iter) are re-exported from connectrpc::client, so migrating a ready collection is client.sum(connectrpc::client::stream_iter(requests)) with no new caller dependency. Adds sync_wrapper so the boxed body is Sync without requiring Sync of the caller's stream. Signed-off-by: Iain McGinniss <309153+iainmcgin@users.noreply.github.com>
Replace the raw impl Stream<Item = Req> + Send + 'static bound on client-streaming calls with impl ClientRequestStream<Req> - a sealed trait implemented automatically for every qualifying stream. Its #[diagnostic::on_unimplemented] attribute turns the common misuse (passing a collection directly) into actionable compiler output that names the stream_iter adapter and explains the Send + 'static requirement, and a borrowing stream's lifetime error now points at a documented, named bound. Named ClientRequestStream to avoid colliding with the server-side dispatcher::RequestStream byte-stream alias. Signed-off-by: Iain McGinniss <309153+iainmcgin@users.noreply.github.com>
client_stream_many_small (1000 small messages) isolates the per-message path from call setup; client_stream_large (10 x 1 MiB) makes the per-message encode/copy cost significant relative to the wire. Both were used to compare the stream-backed request body against the previous channel design on dedicated hardware. iter_batched keeps the input clone out of the timed section. No-Verification-Needed: bench-only change, no runtime surface Signed-off-by: Iain McGinniss <309153+iainmcgin@users.noreply.github.com>
|
[claude code] Benchmarked the stream-as-body design against the previous channel-based client-streaming (base
Reading:
The two new bench groups ( |
#226 fixed the channel-era client-stream leak by racing the detached send task against receiver closure. The stream-as-body design has no detached task — dropping the call future drops the transport send and the request body directly — so the mechanism is superseded while the guarantees are kept: - the # Cancellation rustdoc sections are ported (client-stream text adapted to the direct-drop mechanism; server-stream text unchanged) - all five #226 tests are ported to the stream API; the mid-drain backpressure case becomes an unread-request-body case, since there is no drain loop or channel to park in - tokio/macros stays on the base dependency per #226's consolidation, with the comment narrowed to the server accept loop now that the client-stream path no longer uses select! - the #226 changelog fragment is reworded to describe the behavior (abandonment stops the in-flight send) rather than the superseded background-task mechanism Signed-off-by: Iain McGinniss <309153+iainmcgin@users.noreply.github.com>
|
[claude code] Rebased onto main via merge: #226's abandonment fix targeted the channel-era detached send task, which this PR deletes — dropping the call future now drops the transport send and request body directly, so the guarantee holds structurally. All five #226 tests are ported to the stream API (the mid-drain backpressure case becomes an unread-request-body case, since there is no drain loop to park in), the |
| // the precise encode error instead of the generic transport failure — | ||
| // unconditionally, because a server's early response can race the abort | ||
| // and produce an `Ok` result for a truncated, encode-aborted upload. | ||
| if let Some(err) = encode_error |
There was a problem hiding this comment.
This covers the early-Ok race, but there's a second direction: with a background-driven body, an encode error can land after this check and is then never read — the call returns Ok for a truncated tail. I think that's fine (the server already finished its response), but a sentence saying it's intentional would help a future reader.
There was a problem hiding this comment.
[claude code] Documented in f948aba. The comment now states that direction is deliberate and why: with the body driven in the background, an encode failure can land after the check and is then never read, so the call reports the server Ok. That is the intended outcome, because the server had already produced a complete response and the truncated tail did not affect it.
| #[doc = " are sent as the stream yields them. It backs the request"] | ||
| #[doc = " body, so yield owned messages or feed the call from a"] | ||
| #[doc = " channel-backed stream. For a collection that is already in"] | ||
| #[doc = " hand, wrap it with `::connectrpc::client::stream_iter(...)`."] |
There was a problem hiding this comment.
Could we add a cancellation note here? This is the doc users actually read, and dropping the returned future (e.g. a timeout firing) now silently drops unsent messages. Something like: "Dropping the returned future cancels the call; unsent messages are never delivered." A matching sentence in the guide would be nice too.
There was a problem hiding this comment.
[claude code] Added in f948aba, on both the plain and _with_options generated methods, plus a matching paragraph in the guide (moved out of the code fence into the prose that follows it). The generated text calls out the timeout case specifically, since that is the shape most likely to hit it.
| /// so the compiler can point at the two usual fixes when the bound is not | ||
| /// met: wrap a ready collection with [`stream_iter`], and make a borrowing | ||
| /// stream yield owned messages (the stream backs the request body, which can | ||
| /// outlive the call frame and move across threads — hence `Send + 'static`). |
There was a problem hiding this comment.
Worth a sentence here on panics: the stream is polled on the transport's task now, so a panic in poll_next surfaces as a generic transport error instead of propagating (and can take down a shared connection). Would save someone a confusing debug session.
There was a problem hiding this comment.
[claude code] Added as a # Panics in poll_next section in f948aba. Worth recording that the blast radius is wider than it first looks: hyper eagerly polls the request body inline on the shared dispatch task before handing the pipe to a spawned task (proto/h2/client.rs in 1.9.0), so a panic on that first poll can fault a SharedHttp2Connection for every concurrent RPC on it, not just the one call. I deliberately did not put that mechanism in the doc, though, since it is a hyper internal that a refactor there would silently falsify. The doc states the durable part: the stream is polled on the task driving the request, a panic surfaces as a generic transport error, and where that task is shared it faults every RPC on the connection. It also notes the stream yields Req rather than a Result, so ending the stream early is the only way to report failure.
| Req: buffa::Message + crate::codec::JsonSerialize, | ||
| { | ||
| type Data = Bytes; | ||
| type Error = ConnectError; |
There was a problem hiding this comment.
Might as well add:
fn is_end_stream(&self) -> bool {
self.done
}The default is always false, so downstream layers need an extra poll_frame to notice exhaustion — done already means exactly this.
There was a problem hiding this comment.
[claude code] Checked this one against hyper rather than reasoning about it, and I think it would be a no-op, so I have left it out.
done is set in exactly two places, both inside poll_frame: when poll_next returns None, and in record_error. So it only becomes true as a result of the very poll that already returned None or an error, never before it.
Hyper checks is_end_stream() immediately after receiving a data frame, to decide whether to set END_STREAM on it (proto/h2/mod.rs:158 in 1.9.0, let is_eos = me.stream.is_end_stream(); right after poll_frame yields data). At that moment done is still false, so the optimization would not fire and hyper would poll once more anyway, which is the extra poll the suggestion aims to avoid. The other call site (proto/h2/client.rs:712) runs before the first poll, where done is false by construction.
Getting the actual saving needs one-item lookahead so the body knows the stream is exhausted while returning the last frame, which is a design change rather than a three-line addition. Happy to file that as a follow-up if you think the saved frame is worth it.
| // can surface the precise error instead of a generic transport failure. | ||
| let encode_error: std::sync::Arc<std::sync::Mutex<Option<ConnectError>>> = | ||
| std::sync::Arc::default(); | ||
| let body: ClientBody = EncodingBody { |
There was a problem hiding this comment.
Nothing forces a construction site to also drain the mirror — the pairing is convention only. Maybe EncodingBody::new(...) -> (Self, EncodeErrorSlot) with a #[must_use] slot? Small change, makes the contract compiler-visible.
There was a problem hiding this comment.
[claude code] Agreed the pairing is convention-only. I have deferred it rather than doing it here: there is exactly one construction site today, and unlike the rest of this round it is a refactor rather than a doc change, so it would mean re-running the verification battery on an already-verified PR. Filing it as a follow-up issue so the constraint gets made compiler-visible before a second construction site appears, which is when it would actually start costing something.
| service: &str, | ||
| method: &str, | ||
| requests: impl IntoIterator<Item = Req>, | ||
| requests: impl ClientRequestStream<Req>, |
There was a problem hiding this comment.
Style nit, take it or leave it: the other call_* fns name all their generics; S: ClientRequestStream<Req> would match them and show the bound in rustdoc. (Not about turbofish — that already works; naming it would just add a _ to the conformance call.)
There was a problem hiding this comment.
[claude code] Left as is. Naming the generic would show the bound in a where-clause, but it also adds a type parameter to the turbofish arity, so the existing call sites need an extra _ as you note. That felt like a poor trade for a cosmetic gain, even in a release where we are already breaking things. Happy to take it if you feel strongly.
| /// Re-export of [`futures::stream::iter`]: adapts a collection that is | ||
| /// already in hand into a request stream for a client-streaming call, | ||
| /// without a direct `futures` dependency. | ||
| pub use futures::stream::iter as stream_iter; |
There was a problem hiding this comment.
Small one: we re-export UnaryResponse/ServerStream/BidiStream at the crate root, but not stream_iter, which most callers will want. pub use client::stream_iter; in lib.rs would shorten the common call site.
There was a problem hiding this comment.
[claude code] Done in f948aba, and it turned out to be the most valuable comment of the batch. Adding the root re-export exposed that we were teaching two spellings: the guide and the unmet-bound diagnostic said one thing, while the generated method docs, the call_client_stream example, the streaming-tour example and its README, and the streaming e2e test all still said connectrpc::client::stream_iter. So a caller who hit the compile error, followed the compiler hint, then hovered the generated method would read a different path than the one they had just typed. Everything is now unified on connectrpc::stream_iter; the only remaining mention of the long path is the re-export declaration itself. Verified end-to-end with a live client-streaming call over a socket through the root path.
…tion Applies the non-blocking review feedback on the client-streaming redesign. `stream_iter` is the adapter every migrating caller has to reach for, but it was the one item not re-exported alongside the other client-facing names, so the common call site carried a module path the neighbouring types did not. Re-export it at the crate root and settle on `connectrpc::stream_iter` as the single spelling: the guide, the unmet-bound diagnostic, the `call_client_stream` example, the streaming-tour example and its README, the streaming e2e test, and the generated method docs all used to disagree about the path, so a caller who fixed a compile error by following the compiler's hint then read a different spelling in the generated docs it sent them to. Three behaviours are correct but surprising, and none of them was written down: - Dropping the call cancels it, so messages the stream had not yet yielded never reach the server. Recorded on the generated methods, where a caller reaching for `timeout` will actually see it, and in the guide. - A panic in the caller's `poll_next` cannot reach the caller, because the stream is polled on the task driving the request. It surfaces as a generic transport error, and where that task is shared it faults every RPC on the connection. The stream yields `Req` rather than a `Result` and so has no way to report its own failure, which makes ending the stream early the only option. - An encode failure that lands after the call has already read the error slot is deliberately not surfaced: the server had produced a complete response by then, so the truncated tail did not affect it. The generated directories are regenerated for the doc change. Signed-off-by: Iain McGinniss <309153+iainmcgin@users.noreply.github.com>
…nnectrpc#228) ## Summary Adds `BidiStream::into_split`, splitting a bidirectional stream into independently owned halves so the two sides can be driven from separate tasks — true full duplex: ```rust let (mut send, mut recv) = client.running_sum().await?.into_split(); let reader = tokio::spawn(async move { while let Some(reply) = recv.message().await? { /* ... */ } Ok::<_, connectrpc::ConnectError>(()) }); for req in requests { send.send(req).await?; } send.close_send(); reader.await.expect("reader task")?; ``` Builds on connectrpc#227, now merged, and rebased onto `main`; additive — no existing API changes, no codegen changes (generated bidi methods still return `BidiStream`). ## Design `BidiStream` was already two disjoint sides internally; the split is a plain destructure with no locking added: - **`BidiSendHalf<Req>`** owns `send`/`close_send` (and only needs `Req` bounds — looser than the combined impl). It carries its own copy of the whole-call deadline. Dropping it ends the request body cleanly; the RPC continues until the receive half finishes. - **`BidiRecvHalf<B, RespView>`** owns `message`/`headers`/`trailers`/`error` and the lazy response-initialization state machine. The abort-on-drop of in-flight initialization tasks (connectrpc#221) moves here from `BidiStream`'s `Drop` — which also frees `BidiStream` of its `Drop` impl so `into_split` can destructure. Dropping this half cancels the RPC, exactly as dropping a whole `BidiStream` did; a code comment records that the `send`-before-`recv` field order is load-bearing for the drop sequence. - `BidiStream` keeps its full API via thin delegation, and `into_split` sits in an unbounded impl (a pure move needs no bounds). No `reunite` — documented on both halves. - Docs state the HTTP/2 requirement for interleaved use prominently on `into_split` itself (an HTTP/1.1 caller doing response-dependent sends would deadlock), and steer users toward spawned tasks over naming the halves' body type parameter. ## Testing - Three new e2e tests over a real gRPC/h2 connection: full-duplex ping-pong with the halves owned by different tasks (each echo received before the next send), dropping the send half ends the stream cleanly, dropping the receive half fails subsequent sends instead of hanging. - All 562 all-features unit tests pass unchanged through the delegating methods, including the connectrpc#221 cancellation-safety tests. - Live socket drive: split ping-pong over h2 completes in ~50 ms alongside the existing client-stream probes. - Conformance client suites re-run: Connect 2580/2580; gRPC shows only the known environment-flaky Timeouts cases (including unary cases this diff does not touch). - `cargo semver-checks` vs 0.8.1: 196/196 (additive). Net non-test size: +177 lines. Signed-off-by: Iain McGinniss <309153+iainmcgin@users.noreply.github.com>
2 participants
Summary
Client-streaming calls now take an async
Streamof requests instead of a synchronous iterator.call_client_streamand the generated client-streaming methods acceptimpl ClientRequestStream<Req>— a sealed trait implemented automatically for everyStream<Item = Req> + Send + 'static— so request messages can be produced as they become available — paced by timers, read from sockets, forwarded from channels — without buffering the whole upload or blocking a thread. The synchronous-iterator shape could not express this: there was no way to await between messages, which ruled out pump-style uploads (e.g. forwarding chunks under a per-chunk idle deadline).Breaking change, targeted at 0.9.0. Migration for a ready collection is one line, with no new dependency:
Stream(thefutures0.3 trait) andstream_iter(futures::stream::iter) are re-exported fromconnectrpc::client, and every generated client-streaming method carries the migration hint in its rustdoc. The bound trait carries#[diagnostic::on_unimplemented], so passing a collection directly produces compiler output that namesstream_iterand explains theSend + 'staticrequirement, rather than a bare unmet-trait error.Design: the stream is the request body
The stream is not pumped by a library-side drain loop — it backs the HTTP request body directly (
EncodingBody, which envelope-encodes each message inpoll_frame). The transport polls for the next frame only while it can send, so:An encode failure is emitted through the body (aborting the request) and mirrored into the call's return value so the caller sees the precise error. The stream is held in a
sync_wrapper::SyncWrapper(new tiny zero-dep dependency, the axum pattern) so the boxed body isSyncwithout requiringSyncof the caller's stream. The previous channel + detached-send-task machinery for client-streaming is deleted; bidi is unchanged (its push-stylesend()API keeps the channel).Changes
connectrpc/src/client/mod.rs: signature,EncodingBody,Stream/stream_iterre-exports, rewritten docs with# Errors.connectrpc-codegen/src/codegen.rs: generated arg type + doc hint; checked-in generated dirs regenerated.docs/guide.mdshows bothstream_iterand a channel-backed producer.Testing
task lint/task testgreen; clippy all-targets-D warningsclean; docs with broken-links denied; MSRV (1.88), wasm32, and minimal-features checks pass.cargo semver-checksvs 0.8.1 passes (the break is in animpl Traitargument bound, outside its lint coverage); declared in the changelog fragment.