Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -164,6 +165,100 @@ public async Task CreateAsync_StdErrCallbackThrows_DoesNotCrashProcess()
await Assert.ThrowsAnyAsync<IOException>(() => 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<bool>(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<IOException>(() => 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<ClientTransportClosedException>(
async () => await session.MessageReader.Completion.WaitAsync(
TestConstants.DefaultTimeout,
TestContext.Current.CancellationToken));
var completionDetails = Assert.IsType<StdioClientCompletionDetails>(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")]
Expand Down