client: restore Send on stream message() futures with concrete view types (#215)

Fixes #214.

**Before** (0.8.0 — fails to compile with any concrete generated view
type):
```rust
tokio::spawn(async move {
    while let Some(msg) = stream.message().await? { /* … */ }
});
// error: implementation of `MessageView` is not general enough
```

**After**: the same code compiles and runs; verified live with a spawned
consumer on a multi-threaded runtime across server-stream,
mid-stream-error, and bidi shapes.

## Root cause

The 0.8.0 client-stream change bounded both `message()` methods with
`RespView::Owned: HasMessageView<View<'static> = RespView>`. Projecting
through the GAT inside an `async fn`'s where-clause trips rustc's
coroutine-witness auto-trait check on monomorphization — the future is
`Send` when the message type is a generic parameter but loses `Send`
with any concrete generated type, exactly as isolated in the issue
(whose report was excellent — the generic-vs-concrete observation
pointed straight at the fix).

## The fix

Reshape the bound as an output type parameter:

```rust
pub async fn message<M>(&mut self) -> Result<Option<StreamMessage<M>>, ConnectError>
where
    RespView: MessageView<'static, Owned = M>,
    M: HasMessageView<View<'static> = RespView>,
```

`M` is uniquely determined by `RespView` (the impl blocks already carry
`RespView: MessageView<'static>`), so the bound is logically equivalent
and `M` is inferred at every call site — the entire workspace compiles
with zero call-site changes.

The regression pin in `tests/streaming` asserts `Send` through
async-block wrappers deliberately: the bare `message()` future is `Send`
even with the buggy bound — the failure only manifests in the
*enclosing* coroutine's witness, i.e. exactly the `tokio::spawn(async
move { .. })` shape users write. Verified fail-before (4 errors) /
pass-after, and covered by concrete-shape probes including the issue's
message-typed-`oneof`.

## Semver note

`cargo semver-checks` flags one lint: `message` gains a generic type
parameter (classified major). The formal break surface is explicit
turbofish on a method that previously had no generic parameters — code
that cannot meaningfully exist against 0.8.0. Weighed against the real
regression (the streaming client being unusable in `Send` tasks), this
ships as **0.8.1**.

Signed-off-by: Iain McGinniss <309153+iainmcgin@users.noreply.github.com>

Author: iainmcgin

.changes/unreleased/Fixed-20260702-090031.yaml

@@ -0,0 +1,6 @@
+kind: Fixed
+body: |-
+    Client stream `message()` futures are `Send` again with concrete generated view types ([#214]). 0.8.0's bound shape on `ServerStream::message`/`BidiStream::message` (projecting `RespView::Owned` in the where-clause) tripped rustc's coroutine-witness auto-trait check on monomorphization, so a server-stream could not be consumed inside `tokio::spawn`; the bound is reshaped as an output type parameter (`RespView: MessageView<'static, Owned = M>`), which is equivalent for every caller and restores `Send`. Call sites are unaffected — `M` is inferred.
+
+    [#214]: https://github.com/connectrpc/connect-rust/issues/214
+time: 2026-07-02T09:00:31.44759722-07:00

connectrpc/src/client/mod.rs

@@ -2401,11 +2401,15 @@ where
     /// matching grpc-go's treatment of each case. A Trailers-Only response
     /// carrying `grpc-status: 0` in the headers (empty body) is a clean
     /// end.
-    pub async fn message(
-        &mut self,
-    ) -> Result<Option<crate::StreamMessage<RespView::Owned>>, ConnectError>
+    pub async fn message<M>(&mut self) -> Result<Option<crate::StreamMessage<M>>, ConnectError>
     where
-        RespView::Owned: HasMessageView<View<'static> = RespView>,
+        // `M` is an output parameter pinned to `RespView`'s owned message —
+        // spelled this way round (rather than bounding `RespView::Owned`
+        // directly) so the future stays `Send` for concrete generated view
+        // types: projecting through the GAT in the bound trips rustc's
+        // coroutine-witness auto-trait check (#214).
+        RespView: MessageView<'static, Owned = M>,
+        M: HasMessageView<View<'static> = RespView>,
     {
         // The outcome is immutable once reported: replay the terminal
         // record without re-entering the body.
@@ -3139,11 +3143,12 @@ where
     /// server error carried in the termination metadata is returned as
     /// `Err`, sticky across calls — see [`ServerStream::message()`] for the
     /// full contract.
-    pub async fn message(
-        &mut self,
-    ) -> Result<Option<crate::StreamMessage<RespView::Owned>>, ConnectError>
+    pub async fn message<M>(&mut self) -> Result<Option<crate::StreamMessage<M>>, ConnectError>
     where
-        RespView::Owned: HasMessageView<View<'static> = RespView>,
+        // Same output-parameter shape as `ServerStream::message` — see the
+        // bound comment there (#214).
+        RespView: MessageView<'static, Owned = M>,
+        M: HasMessageView<View<'static> = RespView>,
     {
         // If we already failed during construction or first await, return that.
         if let Some(ref err) = self.construct_err {

tests/streaming/src/lib.rs

@@ -21,6 +21,33 @@ mod tests {
 
     const MSG_DELAY: Duration = Duration::from_millis(100);
 
+    /// Regression pin for issue #214: the `message()` futures must be `Send`
+    /// with CONCRETE generated view types, not just generic parameters — the
+    /// pre-fix bound shape (projecting `RespView::Owned` in the where-clause)
+    /// compiled generically but lost `Send` on monomorphization via rustc's
+    /// coroutine-witness auto-trait check. Compile-time-only assertion; never
+    /// called.
+    #[allow(dead_code)]
+    fn stream_message_futures_are_send<B>(
+        mut server_stream: connectrpc::client::ServerStream<B, EchoResponseView<'static>>,
+        mut bidi: connectrpc::client::BidiStream<B, EchoRequest, EchoResponseView<'static>>,
+    ) where
+        B: connectrpc::http_body::Body<Data = bytes::Bytes> + Send + Unpin + 'static,
+        B::Error: std::fmt::Display,
+    {
+        fn assert_send<T: Send>(_: T) {}
+        // The async-block wrappers are load-bearing: asserting the bare
+        // `message()` future is Send passes even with the buggy bound —
+        // the failure only manifests in the ENCLOSING coroutine's witness,
+        // i.e. exactly the `tokio::spawn(async move { .. })` shape users hit.
+        assert_send(async move {
+            let _ = server_stream.message().await;
+        });
+        assert_send(async move {
+            let _ = bidi.message().await;
+        });
+    }
+
     /// Test echo service that echoes messages back with configurable delays.
     struct TestEchoService;