fix(transport): make AsyncRwTransport::receive cancel-safe (#941)

[ameyypawar]

Fixes#941.

Problem

An rmcp stdio server handling many concurrent requests that produce large responses can intermittently drop an incoming request: it never reaches a handler, so no response is produced and the client waits forever for that id.

Root cause

AsyncRwTransport::receive() reads incoming lines with read_until, which is not cancellation-safe, and receive() is polled inside the service loop's select! (service.rs). When another branch (e.g. an outgoing response on the sink-proxy channel) becomes ready while a request line is only partially read, the receive() future is dropped, and the next call's self.line_buf.clear() at the top of the loop discards the partially-read bytes — so that request is lost.

Under many concurrent large responses the read gets cancelled often, which is why the loss is intermittent and load-dependent. The loss is on the read side: instrumenting the loop shows the missing id never produces a "received request" event, i.e. it never arrives at all.

Fix

Keep the partially-read bytes in line_buf across calls and clear the buffer only after a whole line has been consumed. read_until returns only at a delimiter or EOF (a cancelled read never returns), so on return the buffer holds a complete line; a cancelled read leaves its partial bytes in place and the next call resumes the same line instead of discarding it. The buffer is cleared (retaining its capacity) rather than reallocated per message.

The fix is in AsyncRwTransport, so it also covers the child-process client transport (TokioChildProcess), which reuses the same transport.

Test

Adds crates/rmcp/tests/test_stdio_response_concurrency.rs — the reproducer from the issue. It sends 200 concurrent tools/call requests that each return 64 KiB over real stdio pipes and asserts every response id eventually arrives. It fails reliably on main and passes with this change (40/40 runs locally); the full rmcp test suite remains green, and cargo fmt/clippy are clean.

[DaleSeo]

Thanks for the fix, @ameyypawar!

[ameyypawar]

Fixes#941.

Problem

An rmcp stdio server handling many concurrent requests that produce large responses can intermittently drop an incoming request: it never reaches a handler, so no response is produced and the client waits forever for that id.

Root cause

AsyncRwTransport::receive() reads incoming lines with read_until, which is not cancellation-safe, and receive() is polled inside the service loop's select! (service.rs). When another branch (e.g. an outgoing response on the sink-proxy channel) becomes ready while a request line is only partially read, the receive() future is dropped, and the next call's self.line_buf.clear() at the top of the loop discards the partially-read bytes — so that request is lost.

Under many concurrent large responses the read gets cancelled often, which is why the loss is intermittent and load-dependent. The loss is on the read side: instrumenting the loop shows the missing id never produces a "received request" event, i.e. it never arrives at all.

Fix

Keep the partially-read bytes in line_buf across calls and clear the buffer only after a whole line has been consumed. read_until returns only at a delimiter or EOF (a cancelled read never returns), so on return the buffer holds a complete line; a cancelled read leaves its partial bytes in place and the next call resumes the same line instead of discarding it. The buffer is cleared (retaining its capacity) rather than reallocated per message.

The fix is in AsyncRwTransport, so it also covers the child-process client transport (TokioChildProcess), which reuses the same transport.

Test

Adds crates/rmcp/tests/test_stdio_response_concurrency.rs — the reproducer from the issue. It sends 200 concurrent tools/call requests that each return 64 KiB over real stdio pipes and asserts every response id eventually arrives. It fails reliably on main and passes with this change (40/40 runs locally); the full rmcp test suite remains green, and cargo fmt/clippy are clean.

[DaleSeo]

Thanks for the fix, @ameyypawar!

[ameyypawar]

Fixes#941.

Problem

An rmcp stdio server handling many concurrent requests that produce large responses can intermittently drop an incoming request: it never reaches a handler, so no response is produced and the client waits forever for that id.

Root cause

AsyncRwTransport::receive() reads incoming lines with read_until, which is not cancellation-safe, and receive() is polled inside the service loop's select! (service.rs). When another branch (e.g. an outgoing response on the sink-proxy channel) becomes ready while a request line is only partially read, the receive() future is dropped, and the next call's self.line_buf.clear() at the top of the loop discards the partially-read bytes — so that request is lost.

Under many concurrent large responses the read gets cancelled often, which is why the loss is intermittent and load-dependent. The loss is on the read side: instrumenting the loop shows the missing id never produces a "received request" event, i.e. it never arrives at all.

Fix

Keep the partially-read bytes in line_buf across calls and clear the buffer only after a whole line has been consumed. read_until returns only at a delimiter or EOF (a cancelled read never returns), so on return the buffer holds a complete line; a cancelled read leaves its partial bytes in place and the next call resumes the same line instead of discarding it. The buffer is cleared (retaining its capacity) rather than reallocated per message.

The fix is in AsyncRwTransport, so it also covers the child-process client transport (TokioChildProcess), which reuses the same transport.

Test

Adds crates/rmcp/tests/test_stdio_response_concurrency.rs — the reproducer from the issue. It sends 200 concurrent tools/call requests that each return 64 KiB over real stdio pipes and asserts every response id eventually arrives. It fails reliably on main and passes with this change (40/40 runs locally); the full rmcp test suite remains green, and cargo fmt/clippy are clean.

[DaleSeo]

Thanks for the fix, @ameyypawar!

[ameyypawar]

Fixes#941.

Problem

An rmcp stdio server handling many concurrent requests that produce large responses can intermittently drop an incoming request: it never reaches a handler, so no response is produced and the client waits forever for that id.

Root cause

AsyncRwTransport::receive() reads incoming lines with read_until, which is not cancellation-safe, and receive() is polled inside the service loop's select! (service.rs). When another branch (e.g. an outgoing response on the sink-proxy channel) becomes ready while a request line is only partially read, the receive() future is dropped, and the next call's self.line_buf.clear() at the top of the loop discards the partially-read bytes — so that request is lost.

Under many concurrent large responses the read gets cancelled often, which is why the loss is intermittent and load-dependent. The loss is on the read side: instrumenting the loop shows the missing id never produces a "received request" event, i.e. it never arrives at all.

Fix

Keep the partially-read bytes in line_buf across calls and clear the buffer only after a whole line has been consumed. read_until returns only at a delimiter or EOF (a cancelled read never returns), so on return the buffer holds a complete line; a cancelled read leaves its partial bytes in place and the next call resumes the same line instead of discarding it. The buffer is cleared (retaining its capacity) rather than reallocated per message.

The fix is in AsyncRwTransport, so it also covers the child-process client transport (TokioChildProcess), which reuses the same transport.

Test

Adds crates/rmcp/tests/test_stdio_response_concurrency.rs — the reproducer from the issue. It sends 200 concurrent tools/call requests that each return 64 KiB over real stdio pipes and asserts every response id eventually arrives. It fails reliably on main and passes with this change (40/40 runs locally); the full rmcp test suite remains green, and cargo fmt/clippy are clean.

[DaleSeo]

Thanks for the fix, @ameyypawar!

[ameyypawar]

Fixes#941.

Problem

An rmcp stdio server handling many concurrent requests that produce large responses can intermittently drop an incoming request: it never reaches a handler, so no response is produced and the client waits forever for that id.

Root cause

AsyncRwTransport::receive() reads incoming lines with read_until, which is not cancellation-safe, and receive() is polled inside the service loop's select! (service.rs). When another branch (e.g. an outgoing response on the sink-proxy channel) becomes ready while a request line is only partially read, the receive() future is dropped, and the next call's self.line_buf.clear() at the top of the loop discards the partially-read bytes — so that request is lost.

Under many concurrent large responses the read gets cancelled often, which is why the loss is intermittent and load-dependent. The loss is on the read side: instrumenting the loop shows the missing id never produces a "received request" event, i.e. it never arrives at all.

Fix

Keep the partially-read bytes in line_buf across calls and clear the buffer only after a whole line has been consumed. read_until returns only at a delimiter or EOF (a cancelled read never returns), so on return the buffer holds a complete line; a cancelled read leaves its partial bytes in place and the next call resumes the same line instead of discarding it. The buffer is cleared (retaining its capacity) rather than reallocated per message.

The fix is in AsyncRwTransport, so it also covers the child-process client transport (TokioChildProcess), which reuses the same transport.

Test

Adds crates/rmcp/tests/test_stdio_response_concurrency.rs — the reproducer from the issue. It sends 200 concurrent tools/call requests that each return 64 KiB over real stdio pipes and asserts every response id eventually arrives. It fails reliably on main and passes with this change (40/40 runs locally); the full rmcp test suite remains green, and cargo fmt/clippy are clean.

[DaleSeo]

Thanks for the fix, @ameyypawar!

[ameyypawar]

Fixes#941.

Problem

An rmcp stdio server handling many concurrent requests that produce large responses can intermittently drop an incoming request: it never reaches a handler, so no response is produced and the client waits forever for that id.

Root cause

AsyncRwTransport::receive() reads incoming lines with read_until, which is not cancellation-safe, and receive() is polled inside the service loop's select! (service.rs). When another branch (e.g. an outgoing response on the sink-proxy channel) becomes ready while a request line is only partially read, the receive() future is dropped, and the next call's self.line_buf.clear() at the top of the loop discards the partially-read bytes — so that request is lost.

Under many concurrent large responses the read gets cancelled often, which is why the loss is intermittent and load-dependent. The loss is on the read side: instrumenting the loop shows the missing id never produces a "received request" event, i.e. it never arrives at all.

Fix

Keep the partially-read bytes in line_buf across calls and clear the buffer only after a whole line has been consumed. read_until returns only at a delimiter or EOF (a cancelled read never returns), so on return the buffer holds a complete line; a cancelled read leaves its partial bytes in place and the next call resumes the same line instead of discarding it. The buffer is cleared (retaining its capacity) rather than reallocated per message.

The fix is in AsyncRwTransport, so it also covers the child-process client transport (TokioChildProcess), which reuses the same transport.

Test

Adds crates/rmcp/tests/test_stdio_response_concurrency.rs — the reproducer from the issue. It sends 200 concurrent tools/call requests that each return 64 KiB over real stdio pipes and asserts every response id eventually arrives. It fails reliably on main and passes with this change (40/40 runs locally); the full rmcp test suite remains green, and cargo fmt/clippy are clean.

[DaleSeo]

Thanks for the fix, @ameyypawar!

[ameyypawar]

Fixes#941.

Problem

An rmcp stdio server handling many concurrent requests that produce large responses can intermittently drop an incoming request: it never reaches a handler, so no response is produced and the client waits forever for that id.

Root cause

AsyncRwTransport::receive() reads incoming lines with read_until, which is not cancellation-safe, and receive() is polled inside the service loop's select! (service.rs). When another branch (e.g. an outgoing response on the sink-proxy channel) becomes ready while a request line is only partially read, the receive() future is dropped, and the next call's self.line_buf.clear() at the top of the loop discards the partially-read bytes — so that request is lost.

Under many concurrent large responses the read gets cancelled often, which is why the loss is intermittent and load-dependent. The loss is on the read side: instrumenting the loop shows the missing id never produces a "received request" event, i.e. it never arrives at all.

Fix

Keep the partially-read bytes in line_buf across calls and clear the buffer only after a whole line has been consumed. read_until returns only at a delimiter or EOF (a cancelled read never returns), so on return the buffer holds a complete line; a cancelled read leaves its partial bytes in place and the next call resumes the same line instead of discarding it. The buffer is cleared (retaining its capacity) rather than reallocated per message.

The fix is in AsyncRwTransport, so it also covers the child-process client transport (TokioChildProcess), which reuses the same transport.

Test

Adds crates/rmcp/tests/test_stdio_response_concurrency.rs — the reproducer from the issue. It sends 200 concurrent tools/call requests that each return 64 KiB over real stdio pipes and asserts every response id eventually arrives. It fails reliably on main and passes with this change (40/40 runs locally); the full rmcp test suite remains green, and cargo fmt/clippy are clean.

[DaleSeo]

Thanks for the fix, @ameyypawar!