diff --git a/client/transport/stdio.go b/client/transport/stdio.go index 11268f83c..6cd0177e5 100644 --- a/client/transport/stdio.go +++ b/client/transport/stdio.go @@ -43,6 +43,7 @@ type Stdio struct { stdinMu sync.Mutex stdout *bufio.Reader stderr io.ReadCloser + stderrCh chan []byte responses map[string]chan *JSONRPCResponse mu sync.RWMutex done chan struct{} @@ -62,6 +63,13 @@ type Stdio struct { const ( gracefulShutdownTimeout = 2 * time.Second forceKillTimeout = 3 * time.Second + + // stderrChunkSize is the read buffer size used by the stderr drain goroutine. + stderrChunkSize = 32 * 1024 + // stderrBufferChunks bounds how much undelivered stderr output is retained + // for Stderr() consumers. When full, the newest chunk is dropped so the + // subprocess can never block writing to a full, unread pipe. + stderrBufferChunks = 16 ) func waitForProcessExit(waitErrCh <-chan error, timeout time.Duration) (error, bool) { @@ -113,6 +121,7 @@ func NewIO(input io.Reader, output io.WriteCloser, logging io.ReadCloser) *Stdio stderr: logging, responses: make(map[string]chan *JSONRPCResponse), + stderrCh: make(chan []byte, stderrBufferChunks), done: make(chan struct{}), ctx: context.Background(), logger: slog.Default(), @@ -147,6 +156,7 @@ func NewStdioWithOptions( env: env, responses: make(map[string]chan *JSONRPCResponse), + stderrCh: make(chan []byte, stderrBufferChunks), done: make(chan struct{}), ctx: context.Background(), logger: slog.Default(), @@ -186,6 +196,13 @@ func (c *Stdio) Start(ctx context.Context) error { }() <-ready + // Drain the subprocess's stderr so a server that logs more than the OS pipe + // buffer can never block in write(2) and stop answering on stdout. Stderr() + // consumers read the same stream through the internal buffer (see readStderr). + if c.stderr != nil { + go c.readStderr() + } + return nil } @@ -585,8 +602,67 @@ func (c *Stdio) sendResponse(response JSONRPCResponse) { } } +// readStderr continuously drains the subprocess's stderr into an internal +// buffer so that the OS pipe never fills up. A server that writes more than the +// pipe buffer to stderr would otherwise block in write(2) and stop answering on +// stdout, taking the whole stdio channel down. The buffered stream is exposed +// to callers via Stderr(). When the buffer is full the newest chunk is dropped +// rather than letting the subprocess block. +func (c *Stdio) readStderr() { + defer close(c.stderrCh) + buf := make([]byte, stderrChunkSize) + for { + n, err := c.stderr.Read(buf) + if n > 0 { + chunk := make([]byte, n) + copy(chunk, buf[:n]) + select { + case c.stderrCh <- chunk: + default: + // Buffer full: drop the newest chunk so the subprocess keeps + // making progress instead of blocking on the unread pipe. + } + } + if err != nil { + return + } + } +} + +// stderrReader reads buffered stderr output produced by the readStderr drain +// goroutine. It returns io.EOF once the subprocess's stderr stream is closed. +type stderrReader struct { + ch chan []byte + buf []byte +} + +func (r *stderrReader) Read(p []byte) (int, error) { + if len(p) == 0 { + return 0, nil + } + if len(r.buf) == 0 { + chunk, ok := <-r.ch + if !ok { + return 0, io.EOF + } + r.buf = chunk + } + n := copy(p, r.buf) + r.buf = r.buf[n:] + return n, nil +} + // Stderr returns a reader for the stderr output of the subprocess. // This can be used to capture error messages or logs from the subprocess. +// +// The underlying stderr pipe is drained continuously by the transport, so a +// caller that never reads it cannot deadlock the subprocess. The returned +// reader replays that stream through a bounded buffer (up to ~512KB of the most +// recent output); if a consumer reads slower than the subprocess writes, the +// newest output is dropped rather than blocking the subprocess. func (c *Stdio) Stderr() io.Reader { - return c.stderr + if c.stderr == nil { + return nil + } + return &stderrReader{ch: c.stderrCh} } diff --git a/client/transport/stdio_test.go b/client/transport/stdio_test.go index fd96f84d7..e2e771081 100644 --- a/client/transport/stdio_test.go +++ b/client/transport/stdio_test.go @@ -1050,3 +1050,46 @@ func generateRandomString(size int) string { } return string(b) } + +func TestStdio_StderrDrainPreventsDeadlock(t *testing.T) { + // Regression test for issue #956: the stdio transport must keep reading the + // subprocess's stderr pipe. A server that writes more than the OS pipe buffer + // (~64KB) to stderr would otherwise block in write(2), stop answering on + // stdout, and take the whole stdio channel down with an unexplained hang. + tempFile, err := os.CreateTemp(t.TempDir(), "mockstdio_server") + require.NoError(t, err) + tempFile.Close() + mockServerPath := tempFile.Name() + ".exe" + + require.NoError(t, compileTestServer(mockServerPath)) + + stdio := NewStdio(mockServerPath, nil) + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + require.NoError(t, stdio.Start(ctx)) + defer stdio.Close() + + // Make the server write 256KB to stderr (well past the ~64KB pipe buffer) + // before replying. + logReq := JSONRPCRequest{ + JSONRPC: "2.0", + ID: mcp.NewRequestId(int64(1)), + Method: "debug/log_stderr", + Params: map[string]any{"kilobytes": 256}, + } + logResp, err := stdio.SendRequest(ctx, logReq) + require.NoError(t, err) + require.NotNil(t, logResp) + + // The channel must still be alive after the heavy stderr write: a follow-up + // echo has to complete within the remaining timeout. + echoReq := JSONRPCRequest{ + JSONRPC: "2.0", + ID: mcp.NewRequestId(int64(2)), + Method: "debug/echo", + Params: map[string]any{"alive": true}, + } + echoResp, err := stdio.SendRequest(ctx, echoReq) + require.NoError(t, err) + require.NotNil(t, echoResp) +} diff --git a/testdata/mockstdio_server.go b/testdata/mockstdio_server.go index 6a83f1841..353892ca6 100644 --- a/testdata/mockstdio_server.go +++ b/testdata/mockstdio_server.go @@ -160,6 +160,19 @@ func handleRequest(request JSONRPCRequest) JSONRPCResponse { }) fmt.Fprintf(os.Stdout, "%s\n", responseBytes) + case "debug/log_stderr": + // Write N kilobytes to stderr, then reply. Used to reproduce the + // stdio stderr-pipe deadlock (issue #956): without a draining reader + // the subprocess blocks once the OS pipe buffer (~64KB) fills. + var params struct { + Kilobytes int `json:"kilobytes"` + } + _ = json.Unmarshal(request.Params, ¶ms) + line := strings.Repeat("x", 1023) + "\n" + for i := 0; i < params.Kilobytes; i++ { + _, _ = os.Stderr.WriteString(line) + } + response.Result = map[string]any{"logged": fmt.Sprintf("%dKB", params.Kilobytes)} case "debug/echo_error_string": all, _ := json.Marshal(request) details := mcp.NewJSONRPCErrorDetails(mcp.METHOD_NOT_FOUND, string(all), nil)