Skip to content

Commit 48ad26e

Browse files
authored
fix(upstream): detect a dead stdio transport and respawn the process (RC4-STDIO-001) (#1489)
# Pull Request ## Description Fixes **RC4-STDIO-001**, from the codex QA of v0.70.0-rc.4. The bug also exists in rc.2. When a stdio upstream's child process dies, every call fails with `transport error: transport closed`, but REST keeps reporting the server connected, healthy and usable, and it is never respawned. Only a manual restart recovered it. **Cause.** After the child dies, mcp-go's stdio reader hits EOF and every request returns `transport.ErrTransportClosed`. `isConnectionError` in `internal/upstream/managed/client.go` had no marker for that error, so: - the health loop logged it as "timeout (high activity), ignoring"; - a failed `tools/call` took the not-a-connection-error branch. `SetError` was therefore never called, and the backoff reconnect, which only runs from `StateError`, never fired. **Fix** (managed layer only): - **`isDeadTransportError`:** matches `ErrTransportClosed`, `io.ErrClosedPipe` and `os.ErrClosed` through the error chain. Its text fallback counts only when the text sits inside a wrapped `transport.Error`, so a healthy server's JSON-RPC error text can't trigger it. `io.EOF`/`io.ErrUnexpectedEOF` are deliberately excluded, because HTTP/SSE transports wrap request EOFs that way on live connections. - **Effect:** a dead transport is now a hard connection error and is not treated as a transient health-check error. The first failed ping or call moves the server to Error, and the existing `ShouldRetry` backoff and `tryReconnect` respawn the process. - **Generation guard:** both the call path (`recordDeadConnection`) and the health path record the error only while the client is still Ready on the same connection generation, serialized under `epochMu`. A ping or call that races a deliberate `Disconnect` can't flip a reset client to Error, and a burst of calls on one dead transport costs only one retry. - **No new risks:** the reconnect path is unchanged, the managed `Disconnect` is never called from the health goroutine (no self-join), and backoff and the supervisor behave as before. ## Testing - [x] I have tested these changes locally - [x] I have added/updated tests that prove my fix is effective or my feature works - [x] All existing tests pass - **Tests:** - New: `dead_transport_test.go` covers the classification table (including negatives for JSON-RPC text and HTTP EOF), health check flips to Error, a call marks Error, a burst counts one retry, and a call after Disconnect stays Disconnected. - New: `dead_transport_fixture_test.go` uses a real `mcpfixture` child: SIGKILL, assert it leaves Ready, assert it respawns to a new `instance_id`. - All of these failed before the fix. - **Race suites:** `go test -race` passes for `./internal/upstream/...` and `./internal/runtime/supervisor/...`. The fixture test passed with `-count=3` and `-count=5`. - **Live probe on a real core:** within 5s of SIGKILLing the stdio child, REST shows `connected=false, unhealthy`. By 30s a new child was spawned and the server was healthy again. - **Review:** zcode round 1 raised three mediums (HTTP EOF, JSON-RPC text, health-path epoch). All are fixed in `603614463`. - **Known low follow-ups, not in this PR:** - `runListToolsAsLeader`, `GetCachedToolCount` and `GetPrompt` still call `SetError` on connection errors with no epoch guard. - The epoch branch of the `recordDeadConnection` guard has no dedicated test. 🤖 Generated with [Claude Code](https://claude.com/claude-code)
1 parent a010eca commit 48ad26e

4 files changed

Lines changed: 485 additions & 11 deletions

File tree

‎internal/upstream/managed/client.go‎

Lines changed: 139 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,10 @@ import (
44
"context"
55
"errors"
66
"fmt"
7+
"io"
78
"net"
89
"net/url"
10+
"os"
911
"strings"
1012
"sync"
1113
"sync/atomic"
@@ -22,6 +24,7 @@ import (
2224
"github.com/smart-mcp-proxy/mcpproxy-go/internal/upstream/core"
2325
"github.com/smart-mcp-proxy/mcpproxy-go/internal/upstream/types"
2426

27+
mcptransport "github.com/mark3labs/mcp-go/client/transport"
2528
"github.com/mark3labs/mcp-go/mcp"
2629
"go.uber.org/zap"
2730
)
@@ -1058,6 +1061,7 @@ func (mc *Client) runListToolsAsLeader(listCtx context.Context, release func() b
10581061
}
10591062
}()
10601063

1064+
listEpoch := mc.connectionEpoch.Load()
10611065
tools, err := mc.coreClient.ListTools(listCtx)
10621066
mc.publishListToolsResult(tools, err)
10631067

@@ -1074,7 +1078,9 @@ func (mc *Client) runListToolsAsLeader(listCtx context.Context, release func() b
10741078
mc.logger.Warn("Connection error detected during ListTools, updating server state",
10751079
zap.String("server", mc.GetConfig().Name),
10761080
zap.Error(err))
1077-
mc.StateManager.SetError(err)
1081+
// Guarded like the call and health paths: only the connection
1082+
// that produced the failure may be marked, once.
1083+
mc.setErrorIfCurrentConnection(err, listEpoch)
10781084
}
10791085
return nil, fmt.Errorf("ListTools failed: %w", err)
10801086
}
@@ -1138,7 +1144,6 @@ func (mc *Client) callTool(ctx context.Context, toolName string, args map[string
11381144
if !mc.IsConnected() {
11391145
return nil, fmt.Errorf("client not connected (state: %s)", mc.StateManager.GetState().String())
11401146
}
1141-
11421147
// #1317 round 2: counted from HERE, before admission control, not just
11431148
// around the transport call below — a call already queued in
11441149
// acquireAdmission's wait is just as "in flight" from tryReconnect()'s
@@ -1192,6 +1197,10 @@ func (mc *Client) callTool(ctx context.Context, toolName string, args map[string
11921197
invoker = mc.coreClient
11931198
}
11941199

1200+
// The connection generation this call runs on, read AFTER the admission
1201+
// wait (a queued call may outlive a reconnect), so a transport failure is
1202+
// only ever charged to the session that produced it (RC4-STDIO-001).
1203+
callEpoch := mc.connectionEpoch.Load()
11951204
result, err := invoker.CallTool(ctx, toolName, args)
11961205
if err != nil {
11971206
mc.recordCallToolOAuthSignal(toolName, err)
@@ -1247,7 +1256,7 @@ func (mc *Client) callTool(ctx context.Context, toolName string, args map[string
12471256
zap.String("tool", toolName),
12481257
zap.Error(err))
12491258
}
1250-
mc.StateManager.SetError(err)
1259+
mc.recordDeadConnection(toolName, err, callEpoch)
12511260
}
12521261
return nil, err
12531262
}
@@ -1262,6 +1271,47 @@ func (mc *Client) callTool(ctx context.Context, toolName string, args map[string
12621271
return result, nil
12631272
}
12641273

1274+
// recordDeadConnection flips the server to Error after a tools/call produced
1275+
// hard evidence that its connection is broken (refused/reset, broken pipe, or
1276+
// a dead stdio transport). The health loop's next tick then reconnects through
1277+
// the normal ShouldRetry backoff.
1278+
//
1279+
// The verdict is applied only while the client is still Ready on the SAME
1280+
// connection generation the call ran on, serialized with Connect/Disconnect
1281+
// through epochMu (the same pairing probeAfterAmbiguousCallError uses):
1282+
// - a call that was on the wire when the user disconnected the server sees
1283+
// "transport closed" too; that belongs to the closed generation and must
1284+
// not flip the Disconnected client to Error;
1285+
// - a burst of concurrent calls hitting the same dead transport records ONE
1286+
// failure, not one per call — every extra SetError would bump retryCount
1287+
// and stretch the reconnect backoff for a single process death.
1288+
func (mc *Client) recordDeadConnection(toolName string, err error, callEpoch int64) {
1289+
if !mc.setErrorIfCurrentConnection(err, callEpoch) {
1290+
mc.logger.Debug("Tool call connection failure belongs to a connection that is already gone; not re-marking",
1291+
zap.String("server", mc.GetConfig().Name),
1292+
zap.String("tool", toolName),
1293+
zap.Error(err))
1294+
}
1295+
}
1296+
1297+
// setErrorIfCurrentConnection applies SetError(err) only while the client is
1298+
// still Ready on connection generation epoch, serialized with Connect and
1299+
// Disconnect through epochMu. It reports whether the error was recorded.
1300+
//
1301+
// Both the tools/call path and the health loop use it: Disconnect closes the
1302+
// transport (every in-flight request then fails with "transport closed")
1303+
// BEFORE it bumps the epoch and Resets the state, so a verdict computed from
1304+
// that failure must not land on the freshly Disconnected client.
1305+
func (mc *Client) setErrorIfCurrentConnection(err error, epoch int64) bool {
1306+
mc.epochMu.Lock()
1307+
defer mc.epochMu.Unlock()
1308+
if !mc.IsConnected() || mc.connectionEpoch.Load() != epoch {
1309+
return false
1310+
}
1311+
mc.StateManager.SetError(err)
1312+
return true
1313+
}
1314+
12651315
// recordCallToolOAuthSignal inspects a failed tools/call error and, when it is an
12661316
// "authorization required" / 401 from an otherwise-connected server (NOT a
12671317
// connection error), flags the server as needing OAuth sign-in. This covers
@@ -1698,6 +1748,10 @@ func (mc *Client) performHealthCheck() {
16981748
if prober == nil {
16991749
prober = mc.coreClient
17001750
}
1751+
// The generation this ping runs on: a Disconnect racing the ping closes
1752+
// the transport (the ping fails with "transport closed") and only then
1753+
// bumps the epoch, so the verdict below must land on this generation only.
1754+
pingEpoch := mc.connectionEpoch.Load()
17011755
err := prober.Ping(ctx)
17021756

17031757
if err != nil {
@@ -1747,11 +1801,16 @@ func (mc *Client) performHealthCheck() {
17471801
mc.resetInFlightSuppression()
17481802
}
17491803
if mc.recordHealthCheckFailure(err) {
1750-
mc.logger.Warn("Health check failed repeatedly, marking as error",
1751-
zap.String("server", mc.GetConfig().Name),
1752-
zap.Int("consecutive_failures", mc.consecutiveHealthFailures),
1753-
zap.Error(err))
1754-
mc.StateManager.SetError(err)
1804+
if mc.setErrorIfCurrentConnection(err, pingEpoch) {
1805+
mc.logger.Warn("Health check failed repeatedly, marking as error",
1806+
zap.String("server", mc.GetConfig().Name),
1807+
zap.Int("consecutive_failures", mc.consecutiveHealthFailures),
1808+
zap.Error(err))
1809+
} else {
1810+
mc.logger.Debug("Health check failure belongs to a connection that is already gone; not marking",
1811+
zap.String("server", mc.GetConfig().Name),
1812+
zap.Error(err))
1813+
}
17551814
} else {
17561815
mc.logger.Info("Health check failed transiently, tolerating below threshold",
17571816
zap.String("server", mc.GetConfig().Name),
@@ -1836,6 +1895,12 @@ func isTransientHealthCheckError(err error) bool {
18361895
if err == nil {
18371896
return false
18381897
}
1898+
// A dead transport (the stdio child exited, its pipes closed) never
1899+
// recovers by waiting: every later request fails the same way until the
1900+
// process is respawned (RC4-STDIO-001).
1901+
if isDeadTransportError(err) {
1902+
return false
1903+
}
18391904
msg := strings.ToLower(err.Error())
18401905
// Hard failures: short-circuit to "not transient" so the caller flips
18411906
// Error on the first miss. Order matters — check these BEFORE the
@@ -2133,11 +2198,73 @@ func (mc *Client) waitForReconnectCompletion(ctx context.Context) error {
21332198
}
21342199
}
21352200

2201+
// isDeadTransportError reports whether err proves the transport itself is
2202+
// closed, as opposed to one request failing on a live connection. The stdio
2203+
// case is RC4-STDIO-001: when the upstream's child process dies, mcp-go's
2204+
// stdio reader hits EOF and closes the transport, and every later request
2205+
// returns transport.ErrTransportClosed ("transport closed") — which matched
2206+
// none of isConnectionError's markers, so the health loop ignored it as a
2207+
// "high activity timeout" and the server stayed Ready with every call failing.
2208+
//
2209+
// The classifier is shared by every protocol, so it only accepts evidence that
2210+
// the LOCAL end of the transport is closed:
2211+
// - io.EOF / io.ErrUnexpectedEOF are deliberately NOT matched. net/http wraps
2212+
// them for a single truncated or reset response on a live HTTP/SSE
2213+
// upstream (LB idle timeout, keep-alive race), which must stay subject to
2214+
// the normal flap tolerance. A stdio child's death never surfaces as them:
2215+
// mcp-go's stdio reader turns its EOF into ErrTransportClosed.
2216+
// - The text fallbacks (for chains flattened before they reach us) only
2217+
// count after mcp-go's "transport error: " prefix, i.e. text produced by
2218+
// transport.Error — never a live server's JSON-RPC error message, which
2219+
// mcp-go returns unwrapped and which may echo errno text such as
2220+
// "file already closed".
2221+
func isDeadTransportError(err error) bool {
2222+
if err == nil {
2223+
return false
2224+
}
2225+
if errors.Is(err, mcptransport.ErrTransportClosed) ||
2226+
errors.Is(err, io.ErrClosedPipe) ||
2227+
errors.Is(err, os.ErrClosed) {
2228+
return true
2229+
}
2230+
// Text fallback, for chains that were flattened into a string: only the
2231+
// payload of mcp-go's own transport wrapper counts — either the typed
2232+
// *transport.Error, or a message that STARTS with its prefix. A server's
2233+
// JSON-RPC error is surfaced unwrapped (errors.New(message)), so text it
2234+
// supplies can only match if it is the entire message, not by containing
2235+
// the phrase somewhere.
2236+
var transportMsg string
2237+
var tErr *mcptransport.Error
2238+
if errors.As(err, &tErr) && tErr.Err != nil {
2239+
transportMsg = tErr.Err.Error()
2240+
} else {
2241+
const transportPrefix = "transport error: "
2242+
msg := err.Error()
2243+
if !strings.HasPrefix(msg, transportPrefix) {
2244+
return false
2245+
}
2246+
transportMsg = strings.TrimPrefix(msg, transportPrefix)
2247+
}
2248+
for _, marker := range []string{
2249+
mcptransport.ErrTransportClosed.Error(), // "transport closed"
2250+
os.ErrClosed.Error(), // "file already closed"
2251+
io.ErrClosedPipe.Error(), // "io: read/write on closed pipe"
2252+
} {
2253+
if strings.Contains(transportMsg, marker) {
2254+
return true
2255+
}
2256+
}
2257+
return false
2258+
}
2259+
21362260
// isConnectionError checks if an error indicates a connection problem
21372261
func (mc *Client) isConnectionError(err error) bool {
21382262
if err == nil {
21392263
return false
21402264
}
2265+
if isDeadTransportError(err) {
2266+
return true
2267+
}
21412268

21422269
errStr := err.Error()
21432270
connectionErrors := []string{
@@ -2352,6 +2479,7 @@ func (mc *Client) GetCachedToolCount(ctx context.Context) (int, error) {
23522479

23532480
// Fetch fresh tool count with timeout. Publish the result so any concurrent
23542481
// ListTools waiter coalesced behind us receives the real tools list.
2482+
countEpoch := mc.connectionEpoch.Load()
23552483
tools, err := mc.coreClient.ListTools(listCtx)
23562484
mc.publishListToolsResult(tools, err)
23572485
if err != nil {
@@ -2360,9 +2488,10 @@ func (mc *Client) GetCachedToolCount(ctx context.Context) (int, error) {
23602488
zap.Error(err),
23612489
zap.Int("cached_count", cachedCount))
23622490

2363-
// Check if it's a connection error and update state
2491+
// Check if it's a connection error and update state (guarded: only
2492+
// the connection that produced the failure may be marked, once).
23642493
if mc.isConnectionError(err) {
2365-
mc.StateManager.SetError(err)
2494+
mc.setErrorIfCurrentConnection(err, countEpoch)
23662495
}
23672496

23682497
// Return cached count if available, even if stale
Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
package managed
2+
3+
import (
4+
"context"
5+
"encoding/json"
6+
"os/exec"
7+
"path/filepath"
8+
"runtime"
9+
"testing"
10+
"time"
11+
12+
"github.com/mark3labs/mcp-go/mcp"
13+
"github.com/stretchr/testify/require"
14+
"go.uber.org/zap"
15+
16+
"github.com/smart-mcp-proxy/mcpproxy-go/internal/config"
17+
"github.com/smart-mcp-proxy/mcpproxy-go/internal/secret"
18+
)
19+
20+
// pingInstanceID calls the fixture's deterministic `ping` tool and returns the
21+
// per-process instance_id, which changes on every process start.
22+
func pingInstanceID(ctx context.Context, mc *Client) (string, error) {
23+
res, err := mc.CallTool(ctx, "ping", nil)
24+
if err != nil {
25+
return "", err
26+
}
27+
for _, c := range res.Content {
28+
if tc, ok := c.(mcp.TextContent); ok {
29+
var body struct {
30+
InstanceID string `json:"instance_id"`
31+
}
32+
if jerr := json.Unmarshal([]byte(tc.Text), &body); jerr == nil && body.InstanceID != "" {
33+
return body.InstanceID, nil
34+
}
35+
}
36+
}
37+
return "", nil
38+
}
39+
40+
// TestStdioChildKilled_ClientLeavesReadyAndRespawns is the end-to-end
41+
// RC4-STDIO-001 regression: an admitted stdio upstream (cmd/mcpfixture) whose
42+
// child process is SIGKILLed must stop reporting Ready on the first failed
43+
// call ("transport closed") and be respawned by the managed client's own
44+
// health loop + backoff, without any explicit restart.
45+
func TestStdioChildKilled_ClientLeavesReadyAndRespawns(t *testing.T) {
46+
if testing.Short() {
47+
t.Skip("builds and spawns the mcpfixture stdio binary")
48+
}
49+
if runtime.GOOS == "windows" {
50+
t.Skip("uses pkill to SIGKILL the child")
51+
}
52+
if _, err := exec.LookPath("pkill"); err != nil {
53+
t.Skip("pkill not available")
54+
}
55+
56+
bin := filepath.Join(t.TempDir(), "mcpfixture-deadtransport")
57+
build := exec.Command("go", "build", "-o", bin, "github.com/smart-mcp-proxy/mcpproxy-go/cmd/mcpfixture")
58+
out, err := build.CombinedOutput()
59+
require.NoError(t, err, "build mcpfixture: %s", out)
60+
61+
interval := config.Duration(200 * time.Millisecond)
62+
cfg := &config.ServerConfig{
63+
Name: "stdio-dead-transport",
64+
Command: bin,
65+
Args: []string{"--transport", "stdio"},
66+
Protocol: "stdio",
67+
Enabled: true,
68+
HealthCheckInterval: &interval,
69+
}
70+
mc, err := NewClient(cfg.Name, cfg, zap.NewNop(), nil, &config.Config{}, nil, secret.NewResolver())
71+
require.NoError(t, err)
72+
73+
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
74+
defer cancel()
75+
require.NoError(t, mc.Connect(ctx))
76+
t.Cleanup(func() { _ = mc.Disconnect() })
77+
78+
firstID, err := pingInstanceID(ctx, mc)
79+
require.NoError(t, err)
80+
require.NotEmpty(t, firstID)
81+
epochBefore := mc.ConnectionEpoch()
82+
83+
// SIGKILL only the child: the proxy-side client is left untouched.
84+
require.NoError(t, exec.Command("pkill", "-KILL", "-f", bin).Run())
85+
86+
// The dead transport must surface as a failed call, and that call must
87+
// stop the client from reporting Ready on the dead generation.
88+
require.Eventually(t, func() bool {
89+
_, callErr := mc.CallTool(ctx, "ping", nil)
90+
return callErr != nil
91+
}, 5*time.Second, 20*time.Millisecond, "calls must start failing once the child is dead")
92+
require.True(t, !mc.IsConnected() || mc.ConnectionEpoch() != epochBefore,
93+
"after a call failed on the dead transport the client must not still report Ready on that connection (state=%s)",
94+
mc.GetState())
95+
96+
// ...and the managed client must respawn the process on its own.
97+
var newID string
98+
require.Eventually(t, func() bool {
99+
if !mc.IsConnected() {
100+
return false
101+
}
102+
id, callErr := pingInstanceID(ctx, mc)
103+
if callErr != nil || id == "" {
104+
return false
105+
}
106+
newID = id
107+
return true
108+
}, 20*time.Second, 100*time.Millisecond, "the client must reconnect a fresh stdio process (state=%s)", mc.GetState())
109+
require.NotEqual(t, firstID, newID, "the reconnected upstream must be a new process")
110+
}

0 commit comments

Comments
 (0)