server: stream response items as segments so large fields are not copied

[rkk-ant]

Summary

Extends #232's segmented encode to streaming responses. Today every stream item is encoded to one contiguous Bytes before framing, so Encodable::encode_segments is never consulted on the streaming path and a large field the encoder could hand over by reference count is memcpy'd once per item. #219 already removed the second (framing) copy; this removes the first for any body that yields segments.

With this change each item of a server-streaming or bidi response (and a client-streaming or unary response served over gRPC, which rides the same framing) goes through encode_segments. BatchingEnvelopeStream writes the 5-byte envelope header into its batch buffer as before, then walks the item's segments in wire order: a segment at or above MIN_CHAIN_SIZE becomes its own body frame, unmoved; a smaller one (the tag/length fragment between two large fields, or a short tail) is copied into the batch buffer and rides with whatever is batched next. Wire bytes are unchanged; only HTTP frame boundaries move, as in #219.

Measured on a dev machine, one OwnedView item with a single dominant field through the Connect framing stream (encode + frame, release build):

payload	contiguous (main)	segmented
16 KiB	374 ns	425 ns
512 KiB	11.3 µs	0.41 µs
2 MiB	130 µs	0.42 µs

Above the threshold the per-item cost is flat, since only the framing is still being written.

Where this deliberately does nothing

• Owned-message items keep the contiguous default from server: encode view responses without copying their large fields #232 (their encode_segments is not overridden), so a handler streaming plain owned messages sees no change. See the question below.
• Compressed streams flatten the item to feed the compressor and chain the compressed output as one segment, as before.
• A call through an interceptor flattens each item into a Payload, matching the unary path.
• Connect unary still hardcodes Full<Bytes>; not touched here.

Design notes

• EncodedStream (Pin<Box<dyn Stream<Item = Result<EncodedBody, ConnectError>> + Send>>) is the streaming counterpart of EncodedResponse's body and is what StreamingResult now carries.
• EnvelopeEncoder::encode_chained takes an EncodedBody and returns the segments to chain (empty when the body was copied in below the threshold). The header-then-segments policy is one private helper, write_envelope_chained, shared with Envelope::encode_body_parts, so the unary and streaming paths cannot drift.
• DeadlineStream is generic over the item type rather than naming EncodedBody, since it never inspects items.
• The two sites that flattened a unary EncodedBody into the streaming machinery (into_contiguous() for unary-over-gRPC and client-streaming responses) now pass it through, so those responses keep their segments too.

Breaking change, confined to custom dispatch

StreamingResult's body is EncodedStream rather than BoxStream<Result<Bytes, ConnectError>>, and StreamResponse::{from_encoded, into_encoded} follow. Hand-written Dispatcher impls and test doubles map items with Bytes::into() / .map(|r| r.map(Into::into)) and recover a buffer with EncodedBody::into_contiguous(). Generated dispatchers only name the alias and encode_response_stream, so no regeneration is needed; handler traits are unchanged.

Question for review: owned messages with bytes::Bytes fields

buffa's bytes_type(BytesRepr::Bytes) gives an owned message bytes::Bytes fields whose ProtoBytes::as_shared lets a Rope capture them, but the blanket impl Encodable<M> for M cannot tell such a message from one with Vec<u8> fields, so it stays contiguous. A downstream crate can opt in today with a small Encodable wrapper that encodes through buffa::Rope. Would you prefer that as a provided wrapper here (e.g. Segmented<M>, alongside PreEncoded/MaybeBorrowed), or a type-level signal from buffa codegen that the blanket impl can branch on? Happy to send either as a follow-up.

Testing

• New unit tests: encode_chained keeps segments of a large body by pointer and declares the total; copies a small segmented body; compresses a segmented body as one (gzip). BatchingEnvelopeStream: a segmented item interleaved with a small item yields header+lead / large A / fragment / large B / tail+next envelope, large frames by pointer, reassembly decodes to the contiguous envelopes; error after a segmented item preserves order. encode_response_stream forwards encode_segments; StreamResponse::from_encoded flattens.
• New e2e test in tests/streaming: a server stream of OwnedView items with 20 KiB / 7 B / 48 KiB / 16 KiB fields round-trips over HTTP/1.1 (compression disabled so the segmented path is the one exercised).
• cargo test --workspace --all-features, cargo test -p connectrpc --no-default-features, clippy -D warnings, cargo +nightly-2026-02-27 fmt --check, cargo doc with -Dwarnings: clean.
• Server conformance suite: 3600 passed, 0 failed.

[iainmcgin]

[claude code] Reviewed for the 0.9.0 cut. Two passes (correctness and downstream-API) found no correctness issues: every ordering path through BatchingEnvelopeStream::drain_segments was traced and none emits a chained segment while buf is non-empty or reorders a small tail ahead of a large segment (the push_front when buf is non-empty is the load-bearing line, and it's right); the "never poll the source while pending_segments is non-empty" invariant is structural and debug_asserted at both sites; write_envelope_chained computes the length prefix once from body.len() before choosing copy-vs-chain, so the two branches can't disagree, including on the compressed path; and streaming_segmented_item_chains_each_large_segment / …_tail_flushes_before_pending / …_error_after_segmented_item_preserves_order assert order rather than membership. No unsafe, no Bytes clone that's a real copy, DeadlineStream still ticks per message. Server conformance 3600/0 on the rebased tree.

To make tomorrow's cut I've rebased onto current main (only conflict: one min_size → with_min_size in a test, from #261) and pushed one small commit on top:

• #[non_exhaustive] on EncodedBody. It arrived in server: encode view responses without copying their large fields #232 and hasn't shipped; this PR makes it the Dispatcher streaming contract, so 0.9.0 is the last free moment to seal it. segments() / into_contiguous() / From<Bytes> already cover every use.
• The Bytes → EncodedBody migration for hand-written dispatchers is now a compiled doctest on EncodedStream, since that's the page the type error sends people to; the changelog snippet uses Response::stream(items.map(|r| r.map(EncodedBody::from))).
• Response::compress and the guide's "Returning a view body" section now say that a compressed response flattens the segmented encode — with the default CompressionPolicy (1 KiB) and a gzip-advertising client that's every item large enough to segment, so a large-field stream wants .compress(false). The fragment's "wire bytes are unchanged" became "encoded message bytes are unchanged; a segmented message is now split across more HTTP body frames", and min_size → with_min_size.

Two things the review flagged as measurements owed after the cut rather than blockers: (1) on the compressed path each item is now rope-encoded, into_contiguous()'d, then compressed — one more full copy per item than main, in what is the default configuration; worth a benches/rpc streaming-with-gzip number. (2) worth_segmenting gates on whole-message size, so a 20 KiB item of uniformly small fields takes the rope, captures nothing, and pays the tail-doubling — pre-existing from #232 but now per item for the life of a stream.

On your Segmented<M> question: checked buffa 0.9.1's source and the wrapper works today with no buffa change. Message::write_to is generic over impl EncodeSink; the owned-message generator emits put_shared_bytes_field for every bytes field shape; encode_shared_bytes does if S::IS_SEGMENTED { if let Some(shared) = value.as_shared() { buf.put_shared(shared) } }; and impl ProtoBytes for Bytes returns Some(self.clone()) (refcount) while Vec<u8> returns None. So a Segmented<M> alongside PreEncoded/MaybeBorrowed that encodes through Rope gets the handoff for bytes_type(Bytes) fields. Two limits worth stating in its docs: Vec<u8> fields still copy, and owned string fields never get the handoff (they go through put_string_field with no as_shared hook). Reviewer's take, which I agree with: wrapper for 0.9.x, and a buffa codegen type-level signal later only if you want it to stop being opt-in.