From f076eb2a8c35b1835a9f4527cda10e4e29839adb Mon Sep 17 00:00:00 2001 From: Debasish Padhy Date: Sun, 20 Sep 2026 12:16:00 +0530 Subject: [PATCH 1/2] fix stdio transport readiness barrier using wrong Mono operator sendMessage waited on inboundReady and outboundReady (two Sinks.One) via Mono.zip, but Mono.zip completes as soon as the first of two value-less sources completes -- it never actually waits for the second. Mono.when is the correct operator for waiting on multiple completion-only signals, and does wait for both. Fixes #303 --- .../server/transport/StdioServerTransportProvider.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java b/mcp-core/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java index 8370e41ed..c702ebd23 100644 --- a/mcp-core/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java +++ b/mcp-core/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java @@ -174,7 +174,7 @@ public StdioMcpSessionTransport() { @Override public Mono sendMessage(McpSchema.JSONRPCMessage message) { - return Mono.zip(inboundReady.asMono(), outboundReady.asMono()).then(Mono.defer(() -> { + return Mono.when(inboundReady.asMono(), outboundReady.asMono()).then(Mono.defer(() -> { try { outboundSink.emitNext(message, Sinks.EmitFailureHandler.busyLooping(Duration.ofMillis(100))); return Mono.empty(); From b9375e27edf896cf8ee3e9c5a21a1d34017c8cf9 Mon Sep 17 00:00:00 2001 From: Debasish Padhy Date: Sun, 20 Sep 2026 12:16:04 +0530 Subject: [PATCH 2/2] add regression test for Mono.zip vs Mono.when readiness semantics Pins down why Mono.when is the correct operator for the sendMessage readiness barrier: verifies Mono.zip incorrectly proceeds after only one of two Mono signals fires, while Mono.when correctly waits for both. --- .../StdioServerTransportProviderTests.java | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerTransportProviderTests.java b/mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerTransportProviderTests.java index 72d0dfcf8..f8bd452b3 100644 --- a/mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerTransportProviderTests.java +++ b/mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerTransportProviderTests.java @@ -18,6 +18,7 @@ import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import io.modelcontextprotocol.json.McpJsonDefaults; @@ -30,6 +31,7 @@ import org.junit.jupiter.api.Test; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; import reactor.core.scheduler.Schedulers; import reactor.test.StepVerifier; @@ -103,6 +105,48 @@ void shouldCreateSessionWhenSessionFactoryIsSet() { assertThat(testErr.toString()).doesNotContain("Error"); } + /** + * Regression test for https://github.com/modelcontextprotocol/java-sdk/issues/303. + * + * {@code StdioMcpSessionTransport#sendMessage} gates the actual write on + * {@code inboundReady} and {@code outboundReady} (two {@code Sinks.One}) both + * completing before it proceeds. This mirrors that exact combinator: {@code + * Mono.zip} completes as soon as the FIRST of two value-less {@code Mono} + * sources completes -- it never actually waits for the second one -- silently + * breaking the readiness barrier. {@code Mono.when} is the correct operator for + * waiting on multiple completion-only signals and does wait for both. + */ + @Test + void monoZipDoesNotWaitForBothVoidSignals_monoWhenDoes() { + Sinks.One inboundReady = Sinks.one(); + Sinks.One outboundReady = Sinks.one(); + AtomicBoolean zipProceeded = new AtomicBoolean(false); + + Mono.zip(inboundReady.asMono(), outboundReady.asMono()) + .then(Mono.fromRunnable(() -> zipProceeded.set(true))) + .subscribe(); + + // Only outboundReady has fired; inboundReady is still pending. + outboundReady.tryEmitValue(null); + + assertThat(zipProceeded).as("Mono.zip incorrectly proceeds after only ONE of the two signals fires").isTrue(); + + Sinks.One inboundReady2 = Sinks.one(); + Sinks.One outboundReady2 = Sinks.one(); + AtomicBoolean whenProceeded = new AtomicBoolean(false); + + Mono.when(inboundReady2.asMono(), outboundReady2.asMono()) + .then(Mono.fromRunnable(() -> whenProceeded.set(true))) + .subscribe(); + + // Only outboundReady2 has fired; inboundReady2 is still pending. + outboundReady2.tryEmitValue(null); + assertThat(whenProceeded).as("Mono.when must not proceed until BOTH signals are ready").isFalse(); + + inboundReady2.tryEmitValue(null); + assertThat(whenProceeded).as("Mono.when must proceed once BOTH signals are ready").isTrue(); + } + @Test void shouldHandleIncomingMessages() {