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")]