Async client: fix silent data loss on the final frame of a server stream

[shvbsle]

Fixes #311

Problem

The async client previously spawned one Tokio task per incoming frame. Tasks for the same stream could execute out of order, allowing a terminal REMOTE_CLOSED frame to remove the stream before an earlier DATA frame was delivered. This caused silent data loss and frame reordering. Processing frames directly through the existing bounded response channel preserved ordering but introduced connection-wide head-of-line blocking.

Solution

Give every client RPC an ordered per-stream mailbox:

• ClientReader applies inbound transforms and enqueues frames into the corresponding unbounded mailbox in wire order.
• One dispatcher task per active RPC forwards mailbox entries, in FIFO order, to the existing bounded consumer channel.
• A slow consumer can block only its own dispatcher; the shared connection reader remains free to process other streams.
• DATA frames, responses, close frames, and errors all follow the same ordered path.

The unbounded ingress queue is intentional. TTRPC has no per-stream receive-window mechanism, so a hard bound would require either blocking the whole connection or failing an individual stream at an arbitrary limit. This replaces the previous unbounded task-per-frame behavior with one queue and one dispatcher task per active RPC.

Regression tests

Added three self-contained regression examples:

• async-stream-close-order sends DATA immediately followed by REMOTE_CLOSED across 2,000 calls and verifies the payload always arrives before EOF.
• async-data-order sends 50 sequential frames across 400 concurrent streams and verifies every stream receives them in order.
• async-slow-consumer leaves a 500-frame stream unread, issues an unrelated unary RPC on the same connection, and verifies that the unary RPC completes and all stream frames remain available.

All three examples are executed by tests/run-examples.rs, and spawned-task panics are propagated. Unit tests also verify mailbox ordering beyond the bounded channel capacity and delivery of terminal errors after buffered frames.

Compatibility

The public API and wire format are unchanged. Applications continue using the existing unary and streaming receive APIs without any additional handling.

[shvbsle]

On CI Failures: The CI build failures are unrelated to the code changes. Cargo 1.81 cannot resolve the current crate ecosystem because new crate versions now require edition2024. Seems like infrastructure rot.

[fletcherw]

On CI Failures: The CI build failures are unrelated to the code changes. Cargo 1.81 cannot resolve the current crate ecosystem because new crate versions now require edition2024. Seems like infrastructure rot.

You can update rust-toolchain.toml to 1.95 to fix this.

[shvbsle]

You can update rust-toolchain.toml to 1.95 to fix this.

I had a PR for this but its not as simple as updating the toolchain. Need to fix the new clippy issues that are introduced in new versions:
#313

I can cherry pick or rebase this PR after the above one is merged

[wllenyj]

LGTM.

At the time of implementation, it was simply designed to behave exactly like the Go version, even though the Go version might be incorrect...

[tzneal]

lgtm

[dims]

@tzneal @shvbsle can the reproducer be turned into a test case perhaps? worth it?

[shvbsle]

@tzneal @dims Added a regression test in the second commit. example/async-stream-close-order.rs sends one payload on a server-stream and returns, 1000 times on a multi_thread runtime. Verified it fails without the fix (8/1000 dropped) and passes with it. Wired into tests/run-examples.rs so it runs under cargo test.

[Tim-Zhang]

The ordering bug is real, but this implementation introduces connection-wide head-of-line blocking.

Each stream uses a bounded channel with capacity 100. If one stream is not consumed, resp_tx.send(...).await blocks Connection::run, preventing it from reading responses for every other stream on the same connection.

I reproduced this by leaving a 200-frame stream unread and issuing an unrelated unary RPC: it times out with this patch but succeeds with the previous implementation.

Increasing the channel capacity would only delay the problem. Please preserve ordering without blocking the connection reader, for example with a per-stream ordered mailbox/dispatcher. The reader should enqueue frames without awaiting the consumer-facing bounded channel; if strict memory bounds are required, apply a per-stream quota and fail only the overflowing stream.

There are also two test issues:

1. let _ = h.await ignores task panics and can produce false positives.
2. async-data-order is not executed by tests/run-examples.rs.

[shvbsle]

@Tim-Zhang @wllenyj I've incorporated your feedback and updated my implementation (PR description should have the details). I will appreciate if you can do another round of review!