Assemble streamed messages per message instead of slicing a per-stream read buffer

[rpb-ant]

The server request reader (BodyReader) and the client response stream (ServerStream) accumulated body frames in a per-stream BytesMut and handed each message out as a split_to().freeze() slice of it. That buffer grows to the largest message, never shrinks, and stays referenced by the reader until the stream ends — so a long-lived stream that once carried a multi-MiB message keeps that allocation alive after the handler has dropped the message (#301 measured 32 MiB/stream; GiBs per pod in the reported service). Every multi-frame message was also copied through a shared doubling buffer.

This replaces the buffer with a small EnvelopeAssembler fed one transport frame at a time:

• carries at most the 5 envelope-header bytes between frames; once the header is complete, rejects len > max_message_size before allocating (same resource_exhausted text as Envelope::decode_with_limit);
• payloads larger than 4 KiB are copied into one Vec per message whose capacity grows as min(declared, 2 × received) — a peer that sends 5 bytes cannot make the receiver allocate max_message_size, and the final allocation is exact — and emitted as Bytes::from(vec): the message is the sole owner of its memory (is_unique()) and the reader retains nothing of it;
• payloads of 4 KiB or less are copied into a lazily-allocated 8 KiB per-stream slab and handed out with split_to().freeze(). That costs one Shared allocation per slab rather than per message, and BytesMut::reserve reclaims the slab in place once every message cut from it has been dropped (otherwise the reader rolls a fresh 8 KiB block). Steady-state small messages therefore do not allocate; a held small message keeps at most its 8 KiB slab alive; an idle stream retains at most one slab. The slab size is a compile-time constant checked against bytes' original-capacity bucketing;
• gRPC-Web trailer frames come out of the same assembler as ordinary 0x80 envelopes, which deletes the client's sentinel peeks and max_buf_size accounting; trailers are capped at the parser's existing 1 MiB MAX_GRPC_WEB_TRAILER_SIZE on the header;
• Envelope::decode* and all public API are untouched; EnvelopeDecoder (crate-private) is now frame-fed and returns Decoded::{Message, EndStream} instead of implementing tokio_util's Decoder.

reader loop, 16 KiB frames (jemalloc)	main	this PR
1 MiB message	135 µs, reader retains 2 MiB	109 µs, retains 0, message is_unique()
20 × 1 MiB, previous still held	191 µs/msg	109 µs/msg
3 MiB then 100 × 300 B	retains 4 MiB until stream end	retains ≤ 8 KiB
300 B single-frame message, previous dropped / held	104 ns, 0 allocs / 172 ns, 2 allocs	98 ns / 105 ns, 0 allocs (steady state)
300 B messages packed in 256 KiB h1 chunks	95 ns	69 ns
300 B messages straddling 1000 B gRPC-Web chunks	81 ns	77 ns
benches/rpc unary / streaming arms	—	within noise; client_stream_many_small −2…−3.5%

Trade-offs, stated plainly: an idle stream that has carried a small message now keeps up to 8 KiB (10k idle streams ≈ 80 MiB) where the first revision of this PR kept 0 and main keeps up to 2× its largest message; and because each large message is a fresh exact allocation, client_stream_large (10 × 1 MiB) is +10–19% under glibc malloc end-to-end (page-fault/munmap churn) while flat-to-slightly-faster under jemalloc — see the perf evaluation comment for the full matrix.

Compatibility: no public API or wire change; error codes and texts preserved except that the client-side "response buffer exceeds limit" error no longer exists (memory is bounded per envelope, whose over-limit header reports "message size N exceeds limit M" as before) and a gRPC-Web trailer frame larger than 1 MiB now fails on its header with resource_exhausted. The workspace bytes floor moves to 1.6.1 only because the new tests rely on Bytes::is_unique being correct for BytesMut-derived values. Conformance (v1.0.5): server 3600, client connect 2580 / grpc 1454 / grpc-web 2838 passed, 0 failed.

Alternative to #301, and built on it — its analysis of the pinning mechanism and its is_unique() tests are what this builds on; the difference is that the invariant is structural rather than restored by a 64 KiB release heuristic, so the completing-frame-carries-the-next-header case, all-large streams, and large h1 chunks need no special handling.

[rpb-ant]

We benchmarked #302 against main (758bd3a) and main + #301, at the reader level and end to end, under both jemalloc and glibc malloc. The retention fix is real and identical in the two PRs: an idle stream that once carried a 3 MiB message keeps 4.2 MB on main and ~5 KB on either PR; #302 additionally keeps a still-held message at its exact size (3.15 vs 4.2 MB) and halves the peak during assembly. Reader CPU is equal or better almost everywhere and clearly better from 64 KiB up. The one real loss is MiB-message streams end to end under glibc malloc (+10–19% per call, absent under jemalloc); beyond that, small single-frame messages cost one extra small malloc.

How it was measured

One harness compiled into all three builds drives the real server BodyReader (directly, and as the spawned reader task + mpsc(1) + handler loop) and the real client ServerStream::message(), fed Bytes slices as hyper delivers them: ≤16 KiB h2 DATA frames, or fixed-size cuts across message boundaries (h1/proxy). A counting allocator gives allocations and retained bytes; timings are medians of 10, pinned, builds interleaved, loopback on a Xeon 8488C. On top of that, the repo's rpc_bench and a 200-concurrent-stream memory test.

Memory

• Idle stream after a 3 MiB message: main 4.2 MB, Release a stream's read buffer after a large message is handed off #301 ≈5 KB, Assemble streamed messages per message instead of slicing a per-stream read buffer #302 ≈5 KB. In the 200-stream test with the handler keeping an owned decode of the first message, RSS per stream is 6.3 MB on main vs 3.1 MB on both PRs.
• Big message still held as a zero-copy view: main and Release a stream's read buffer after a large message is handed off #301 4.2 MB (the message owns the doubled buffer), Assemble streamed messages per message instead of slicing a per-stream read buffer #302 3.15 MB.
• Peak per in-flight 1 MiB message: 2.1 MB on main and Release a stream's read buffer after a large message is handed off #301, 1.05 MB on Assemble streamed messages per message instead of slicing a per-stream read buffer #302.

CPU

Reader step per message, jemalloc; glibc in parentheses where it changes the picture.

workload	main	#301	#302
300 B, one per frame, client, previous message dropped	103 ns	109	138 (+1 malloc)
same, previous message still held	173 ns	177	125
same, real server pipeline (reader task one message ahead)	583 ns	572	519
64 KiB in 16 KiB frames	6.65 µs	8.23	6.35
1 MiB in 16 KiB frames, previous dropped	135 µs (97)	165 (74)	109 (77)
20 × 1 MiB back-to-back, previous held	191 µs	154	109
300 B messages packed in 256 KiB h1 chunks, client	94 ns (81)	104 (88)	106 (102–109)
gRPC-Web, 300 B messages straddling 1000 B chunks	85 ns	89	125 (0.3 reallocs)

("Previous dropped/held" = whether the consumer still holds the previous message's Bytes when the next frame arrives: on main that decides whether the shared BytesMut can be reclaimed in place (0 allocs) or must be reallocated and re-shared (2 allocs); #302 never shares its buffer, so it always pays one exact allocation.) Row 1 is #302's per-message cost, one exact allocation where main could slice: about 1% of the ~4 µs end-to-end per-message cost. In the real server pipeline the reader is always a message ahead, so main takes its two-allocation path and #302 is 11% faster.

End-to-end

unary, server_stream, client_stream ×10 and bidi: within ±1–2% on either allocator. client_stream_many_small: #302 −3 to −5% under glibc, noise under jemalloc; #301 +0 to +3%. client_stream_large (10 × 1 MiB) is the exception.

Where #302 loses

1. MiB messages under glibc, end to end: +18 / +19 / +10% per call (connect / gRPC / gRPC-Web), 60–170 µs per 1 MiB message, both passes. Each message is a fresh exact ~1 MiB allocation that glibc maps, faults in and unmaps (3.35 M vs 2.32 M minor faults, +57% sys time), where main amortises one 2 MiB buffer over the call. Under jemalloc: −1 to −5%. This is real if you serve MiB streams on the system allocator. The fix is allocator-side (jemalloc/mimalloc, mallopt thresholds); doing it in the library means a bounded reusable buffer, i.e. Release a stream's read buffer after a large message is handed off #301's trade — and Release a stream's read buffer after a large message is handed off #301 already pays about a third of it (+6% connect).
2. Small messages main could have sliced: single-frame with the previous one dropped (+25–35 ns, one 300 B malloc), or many per coalesced h1 chunk (client +12% jemalloc, +25–35% glibc; not significant in the server pipeline). Negligible per RPC. A ~6-line whole-frame zero-copy path makes both 0-alloc again, at the price of a held small message pinning its frame.
3. Small messages straddling frames (gRPC-Web): +40 ns, 0.3 reallocs per message. A two-line reserve_exact(min(expected, 16 KiB)) on header completion removes it with bounded amplification.

Where #301 loses

Messages on the 64 KiB threshold regrow from zero every time (+18–24% CPU). Drop-aligned MiB streams are +22–27% under jemalloc only; the earlier "+33–38%" doesn't reproduce as stated, and with the previous message held #301 beats main. A held zero-copy message still owns the doubled buffer, and since client-side release is per decode rather than per frame, a held previous message or coalesced framing leaves the client holding 131–167 KB — bounded, but the silent-when-wrong mode of a threshold.

Caveats

• Loopback, synthesized frames, no TLS or flow control. glibc keeps freed heap, so compare the PRs on live bytes or under jemalloc, not RSS-after-drop.
• Allocator state dominates MiB timings, and the single-threaded reader harness can't see glibc's fault/munmap cost; for MiB messages on glibc trust the end-to-end numbers (rpc_bench has no #[global_allocator], so stock runs are glibc). The main-vs-Assemble streamed messages per message instead of slicing a per-stream read buffer #302 ordering held in every regime; Release a stream's read buffer after a large message is handed off #301-vs-Assemble streamed messages per message instead of slicing a per-stream read buffer #302 on drop-aligned 1 MiB is a tie under warmed glibc and a 34% Assemble streamed messages per message instead of slicing a per-stream read buffer #302 win under jemalloc.
• Small-message cells move ±5–10% between passes; under 10 ns is noise. The counting allocator adds 2–5 ns per allocation, biased against Assemble streamed messages per message instead of slicing a per-stream read buffer #302.

Harness, raw logs and the full write-up are available on request.

[rpb-ant]

Updated per the discussion with @iainmcgin: the 8 KiB small-message slab is folded into this PR (payloads ≤ 4 KiB are cut from a per-stream slab, larger ones keep the exact per-message Vec), so small single-frame messages are back to 0 allocations in steady state (300 B: 98 ns vs main's 104 ns; straddling gRPC-Web chunks 77 ns vs 81 ns; h1-coalesced 69 ns vs 95 ns) while a held small message pins at most 8 KiB and large messages stay exactly owned. Also applied the repo's rust-code-reviewer / rust-api-ergonomics-reviewer findings and a further intent/correctness/elegance review pass (dead post-decompress size check removed, EnvelopeDecoder::decode now returns Decoded::{Message, EndStream}, single site for trailing-after-END_STREAM, trailer cap back at the unary parser, test merges). Net non-test delta vs main is now about +67 lines. Description updated with the slab, the idle-memory trade-off, and the glibc client_stream_large caveat.