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

[shvbsle]

Fixes #311

Process each frame inline in ClientReader::handle_msg instead of tokio::spawn-ing per frame. The read loop in connection.rs already awaits handle_msg per frame, so inline preserves per-stream wire order and restores natural back-pressure on the per-stream mpsc. handle_err adjusted the same way for consistency.

Server dispatch also spawns per frame, but there the spawn is load-bearing (user handlers can take arbitrary time) and a oneshot already preserves ordering. Left alone.

Regression tests

Added two examples that exercise the race on a multi_thread runtime:

• async-stream-close-order - server-streaming handler sends one payload and returns (DATA immediately followed by REMOTE_CLOSED). Repeats 2000 times in batches of 200 and asserts every iteration receives the payload before EOF. Targets data loss from the close-frame removing the stream before the data-frame task looks it up.

• async-data-order - server-streaming handler sends 50 sequential DATA frames per stream. Opens 100 concurrent streams and asserts every stream receives frames in sequence order. Targets reordering from spawned tasks executing out of submission order.

Without the fix:

repeating the experiment 2000 times, we find data loss 148/2000 times
repeating the experiment 100 times, we find out of order frames 8/100 times

With the fix both pass cleanly across repeated runs.

Compatibility

Public API, wire format, and observable behavior for correctly-behaving consumers are unchanged.

[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.