fix: make STDIO transport sendMessage thread-safe (#304)#935
Open
nquinquenel wants to merge 1 commit intomodelcontextprotocol:mainfrom
Open
fix: make STDIO transport sendMessage thread-safe (#304)#935nquinquenel wants to merge 1 commit intomodelcontextprotocol:mainfrom
nquinquenel wants to merge 1 commit intomodelcontextprotocol:mainfrom
Conversation
…ol#304) Concurrent calls to `sendMessage` on the STDIO transports raced on the unicast sink and sporadically failed with `FAIL_NON_SERIALIZED`, dropping the message with `RuntimeException("Failed to enqueue message")`. Switch `tryEmitNext` to `emitNext` with a `busyLooping` EmitFailureHandler so concurrent emissions are serialized as recommended by the Reactor maintainers. Applied on both `StdioServerTransportProvider` and `StdioClientTransport`. Also adds a regression test that fans `sendMessage` out across 16 parallel rails and verifies all 500 emissions complete successfully and every message is written to the output stream.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Hello, we've had this issue #304 for a while in our MCP server, and I implemented this fix following your recommendation.
As a summary, concurrent calls to
sendMessageon the STDIO transports raced on the unicastSinks.Manyand sporadically failed withFAIL_NON_SERIALIZED, dropping the message withRuntimeException("Failed to enqueue message"). This happens in practice whenever tools andloggingNotificationare emitted from different threads on the server, and can happen on the client under any concurrent sender.I switched
tryEmitNexttoemitNext(message, Sinks.EmitFailureHandler.busyLooping(Duration.ofMillis(100)))as suggested by the maintainers in the issue, so concurrent emissions are serialized. Applied on both sides:mcp-core/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.javamcp-core/src/main/java/io/modelcontextprotocol/client/transport/StdioClientTransport.javaThe remaining
catch (Sinks.EmissionException)path still returns a failedMonofor the non-retryable modes (FAIL_CANCELLED,FAIL_TERMINATED, busyLooping timeout), preserving the existing error semantics.I also removed the TODO, as I believe that the only realistic failure (
FAIL_NON_SERIALIZED) is now handled in-place by busyLooping. The other scenario such asFAIL_OVERFLOWis unreachable with the current sink -onBackpressureBuffer()is unbounded so there's no overflow to reschedule. Happy to revert if you'd rather keep the hint for a future bounded-buffer refactor.Test plan
shouldHandleConcurrentSendMessageregression test that fans 500sendMessagecalls across 16 parallel rails and verifies all complete and every message lands on the output stream. The test fails onmain(RuntimeException: Failed to enqueue messageafter ~44 emissions) and passes with the fix../mvnw -pl mcp-test -am test -Dtest='StdioServerTransportProviderTests,StdioMcpSyncClientTests'- 52/52 pass.StdioMcpAsyncClientTestspass in isolation with the fix.