From 1d6b66a998cf0a038e8336161185bc1f76e60be9 Mon Sep 17 00:00:00 2001 From: azik Date: Sun, 8 Nov 2026 13:47:54 +0000 Subject: [PATCH] Fix stdio send hang and bare send failure when the server exits at once When a stdio server exits right after it starts, the read loop sees EOF and CleanupAsync closes the server's stdin while the client's first send may still be writing to it. On Unix, disposing the pipe while that write is creating its socket in SafePipeHandle.CreatePipeSocket hangs the write forever in SafeSocketHandle.CloseAsIs. Take the send lock, for at most ShutdownTimeout, before closing stdin. A send still holding it after that leaves stdin open and the server to the process-tree kill. A send that failed with an IOException checked for the server's exit before a server that had just exited might have been reaped, so it threw a bare "Failed to send message." instead of the exit code and stderr. Wait for the process to exit, for at most ShutdownTimeout, before building the exit exception. --- .../Client/StdioClientSessionTransport.cs | 21 +++- .../Client/StreamClientSessionTransport.cs | 2 +- .../Transport/StdioClientTransportTests.cs | 95 +++++++++++++++++++ 3 files changed, 115 insertions(+), 3 deletions(-) diff --git a/src/ModelContextProtocol.Core/Client/StdioClientSessionTransport.cs b/src/ModelContextProtocol.Core/Client/StdioClientSessionTransport.cs index bb446af55..6f95dd15c 100644 --- a/src/ModelContextProtocol.Core/Client/StdioClientSessionTransport.cs +++ b/src/ModelContextProtocol.Core/Client/StdioClientSessionTransport.cs @@ -34,7 +34,9 @@ public override async Task SendMessageAsync(JsonRpcMessage message, Cancellation catch (IOException) { // We failed to send due to an I/O error. If the server process has exited, which is then very likely the cause - // for the I/O error, we should throw an exception for that instead. + // for the I/O error, we should throw an exception for that instead. A server that has just exited may not + // have been reaped yet, so give it up to ShutdownTimeout to exit before checking. + await WaitForProcessExitAsync().ConfigureAwait(false); if (await GetUnexpectedExitExceptionAsync().ConfigureAwait(false) is Exception processExitException) { throw processExitException; @@ -63,14 +65,29 @@ protected override async ValueTask CleanupAsync(Exception? error = null, Cancell // Closing the server's stdin is the portable graceful-shutdown signal for stdio transports. // Do this before waiting so a well-behaved server can finish its own cleanup and exit. + // Close it only once no send is writing to it: on Unix, disposing the pipe while a first write is + // still creating its socket hangs that write forever. A send still writing after ShutdownTimeout + // leaves the server to the process kill below. + bool sendLockTaken = false; try { - _process.StandardInput.Close(); + sendLockTaken = await _sendLock.WaitAsync(_options.ShutdownTimeout).ConfigureAwait(false); + if (sendLockTaken) + { + _process.StandardInput.Close(); + } } catch (Exception ex) { LogStandardInputCloseFailed(Name, ex); } + finally + { + if (sendLockTaken) + { + _sendLock.Release(); + } + } // Ensure all pending ErrorDataReceived events are drained before detaching // the handler. GetUnexpectedExitExceptionAsync does this when HasExited is diff --git a/src/ModelContextProtocol.Core/Client/StreamClientSessionTransport.cs b/src/ModelContextProtocol.Core/Client/StreamClientSessionTransport.cs index 7a69c1c83..598f4d330 100644 --- a/src/ModelContextProtocol.Core/Client/StreamClientSessionTransport.cs +++ b/src/ModelContextProtocol.Core/Client/StreamClientSessionTransport.cs @@ -14,7 +14,7 @@ internal class StreamClientSessionTransport : TransportBase private readonly TextReader _serverOutput; private readonly Stream _serverInputStream; - private readonly SemaphoreSlim _sendLock = new(1, 1); + private protected readonly SemaphoreSlim _sendLock = new(1, 1); private readonly CancellationTokenSource _shutdownCts = new(); private Task? _readTask; diff --git a/tests/ModelContextProtocol.Tests/Transport/StdioClientTransportTests.cs b/tests/ModelContextProtocol.Tests/Transport/StdioClientTransportTests.cs index 5a3705c4e..8f8f3a3fc 100644 --- a/tests/ModelContextProtocol.Tests/Transport/StdioClientTransportTests.cs +++ b/tests/ModelContextProtocol.Tests/Transport/StdioClientTransportTests.cs @@ -12,6 +12,7 @@ namespace ModelContextProtocol.Tests.Transport; public class StdioClientTransportTests(ITestOutputHelper testOutputHelper) : LoggedTest(testOutputHelper) { public static bool IsStdErrCallbackSupported => !PlatformDetection.IsMonoRuntime; + public static bool IsUnixWithStdErrCallbackSupported => !PlatformDetection.IsWindows && IsStdErrCallbackSupported; [Fact] public async Task DisposeAsync_ClosesServerStandardInputForGracefulExit() @@ -164,6 +165,100 @@ public async Task CreateAsync_StdErrCallbackThrows_DoesNotCrashProcess() await Assert.ThrowsAnyAsync(() => McpClient.CreateAsync(transport, loggerFactory: LoggerFactory, cancellationToken: TestContext.Current.CancellationToken)); } + [Fact(Skip = "Platform not supported by this test.", SkipUnless = nameof(IsUnixWithStdErrCallbackSupported))] + public async Task SendMessageAsync_ServerExited_ThrowsWithExitDetails() + { + string id = Guid.NewGuid().ToString("N"); + var stdErrReceived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + // The server closes its stdin, so the send fails, and exits a second later. That holds open the + // window a server exiting at once leaves until it is reaped: the send has failed, but the + // process has not yet been seen to exit. + StdioClientTransport transport = new(new() + { + Command = "sh", + Arguments = ["-c", $"exec 0<&-; echo {id} >&2; sleep 1; exit 3"], + ShutdownTimeout = TestConstants.DefaultTimeout, + StandardErrorLines = line => + { + if (line.Contains(id)) + { + stdErrReceived.TrySetResult(true); + } + }, + }, LoggerFactory); + + await using ITransport session = await transport.ConnectAsync(TestContext.Current.CancellationToken); + await stdErrReceived.Task.WaitAsync(TestConstants.DefaultTimeout, TestContext.Current.CancellationToken); + + var exception = await Assert.ThrowsAnyAsync(() => session.SendMessageAsync( + new JsonRpcNotification { Method = NotificationMethods.InitializedNotification }, + TestContext.Current.CancellationToken)); + Assert.StartsWith("MCP server process exited unexpectedly", exception.Message); + Assert.Contains(id, exception.Message); + + var closedException = await Assert.ThrowsAsync( + async () => await session.MessageReader.Completion.WaitAsync( + TestConstants.DefaultTimeout, + TestContext.Current.CancellationToken)); + var completionDetails = Assert.IsType(closedException.Details); + Assert.Equal(3, completionDetails.ExitCode); + Assert.Contains(id, completionDetails.StandardErrorTail ?? []); + } + + [Fact(Skip = "Platform not supported by this test.", SkipUnless = nameof(IsUnixWithStdErrCallbackSupported))] + public async Task ConnectSendAndDispose_ServerExitsImmediately_CompletesWithExitDetails() + { + // A server that exits at once makes the transport close the server's stdin while the first send + // may still be writing to it. On Unix, closing the pipe at that moment could hang the send forever, + // and a send that failed before the exit was observed reported a bare "Failed to send message.". + // Both races are narrow, so run many short sessions in parallel. Thousands of sessions would + // flood the test output, so this test does not log. + const int WorkerCount = 16; + const int SessionsPerWorker = 200; + string id = Guid.NewGuid().ToString("N"); + + StdioClientTransport transport = new(new() + { + Command = "sh", + Arguments = ["-c", $"echo {id} >&2; exit 3"], + ShutdownTimeout = TestConstants.DefaultTimeout, + }); + + int sendFailuresWithoutExitDetails = 0; + Task[] workers = Enumerable.Range(0, WorkerCount).Select(_ => Task.Run(async () => + { + for (int i = 0; i < SessionsPerWorker; i++) + { + await using ITransport session = await transport.ConnectAsync(TestContext.Current.CancellationToken); + try + { + await session.SendMessageAsync( + new JsonRpcNotification { Method = NotificationMethods.InitializedNotification }, + TestContext.Current.CancellationToken); + } + catch (IOException ex) + { + if (!ex.Message.Contains(id)) + { + Interlocked.Increment(ref sendFailuresWithoutExitDetails); + } + } + } + }, TestContext.Current.CancellationToken)).ToArray(); + + try + { + await Task.WhenAll(workers).WaitAsync(TestConstants.DefaultTimeout, TestContext.Current.CancellationToken); + } + catch (TimeoutException) + { + Assert.Fail($"{workers.Count(w => !w.IsCompleted)} of {WorkerCount} workers did not complete."); + } + + Assert.Equal(0, sendFailuresWithoutExitDetails); + } + [Theory] [InlineData(null)] [InlineData("argument with spaces")]