fix: command-backed MCP services get one subprocess per session, not a shared multiplexed bridge - #387
Conversation
Subscribe()'s only caller is bridgeTransport, i.e. an MCP request/response session: every line matters, since the one dropped could be the exact reply a Read() call is blocked on, with nothing left to ever wake it. It shared the same bounded (cap 10), drop-on-full channel as the SSE feed in ServeHTTP, where dropping is fine - a browser tab just misses a live update. A periodic health-check Probe() and a live tools/call session subscribing to the same backend concurrently was enough to lose a message that way, surfacing as a client-side EOF even though the backend had answered correctly. Every Subscribe() caller now gets its own unbounded, order-preserving queue instead, pushed to non-blockingly from the bridge's single stdout-scanning goroutine and drained by the session's own reader. The SSE feed is untouched - it keeps its own bounded, drop-tolerant channel, appropriate for a best-effort live stream.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request replaces the channel-based subscription mechanism in StdioBridge with an unbounded, order-preserving subscriberQueue to prevent dropping stdout lines on slow consumers. Feedback on the changes points out a potential memory retention issue in subscriberQueue.pop where slicing the buffer does not release the memory of the popped strings, and suggests explicitly zeroing out the popped element and resetting the slice when empty.
Re-slicing alone (q.buf = q.buf[1:]) leaves the popped element's string header live in the backing array: the array is one GC-tracked object, so as long as any slice into it is reachable, every element within it is too, including ones logically before the current slice's start. For a long-lived session relaying many messages, that's every line ever queued, retained for the queue's whole lifetime regardless of how many have actually been popped. pop now clears the popped slot before advancing, releasing that string, and drops the backing array entirely once the queue drains to empty rather than let it sit at its peak size indefinitely. New test asserts both: a second reference into the backing array reads back "" after its line is popped, and buf is nil after a full drain - confirmed to fail without the fix (backing array slot still holds "a" after being popped).
…ribe Both prior tests for this fix drove StdioBridge/bridgeTransport through an in-memory io.Pipe and never exercised two concurrent subscribers, so neither would have caught a regression in the exact scenario Subscribe's own doc comment describes (a health probe racing a live session). These two tests spawn a real subprocess and drive it through the actual bridgeTransport path: - TestIntegration_ProbeAndLiveSessionConcurrently: a live session bursting calls without reading between them, racing a continuously-reconnecting probe, asserting neither ever loses a reply. Fails deterministically on the pre-fix code (verified against 17da7f6~1) and passes on this branch. - TestIntegration_StuckSubscriberDoesNotStallOthers: one subscriber that connects and never reads, asserting a second, healthy subscriber on the same backend is unaffected. This is the test that rules out the obvious simpler alternative to the unbounded queue (a private bounded channel per subscriber with a blocking send) - verified separately that it deadlocks this exact test, since the scan loop dispatches to every subscriber under one lock that Send also takes. Signed-off-by: Fernando Marino <fernando.marino85@gmail.com>
|
Thanks for the review, @aojea — and for the earlier patience on #375 while I chased that one down to its actual cause. Pushed one more commit (d951715) adding two integration tests that drive a real subprocess through the actual StdioBridge → bridgeTransport path (not the in-memory pipe the other tests use), since that's the level the original bug and the memory-retention fix both actually live at:
This is ready for another look whenever you have a chance. |
|
@fer-marino I was looking into this and I think we are working in a direction that will cost us later. The bug here is in the sharing model, not in the buffer. Subscribe() fans out every stdout line to every session, so a session's channel fills with replies that belong to other sessions. Worse, the go-sdk numbers requests from 1 per connection, so two concurrent sessions on the same bridge both send id: 1 and both receive both replies — session A can consume session B's tools/call result today. An unbounded per-subscriber queue keeps all of that, and adds a new failure mode: a peer that stops reading its stream (HandleStreamPassThrough has no timeout while the backend leg is alive) retains every other session's traffic indefinitely. "Silent drop → hang" becomes "unbounded growth → node OOM", triggered by an authorized peer. I don't feel sam-node should own a JSON-RPC multiplexer. Stdio MCP is single-session by spec; no SDK multiplexes it, because there's no correct way to do so (server-initiated requests like sampling or elicitation have no attributable destination on a shared process). Any mux here — the current bridge, this queue, or a smarter id-rewriting version I prototyped — is protocol-level code we'd maintain and get subtly wrong over years. We've been there in Kubernetes. What I'd like to do instead: use case *api.RegisterServiceRequest_Command:
cmd := exec.Command(x.Command.Command[0], x.Command.Command[1:]...)
// env setup as today
return &mcp.CommandTransport{Command: cmd}, nil
Two things to keep in mind while doing it:
Would you be up for reworking this PR along those lines, keeping your subprocess-based regression tests adapted to the new transport? If you'd rather not, no problem — say so and I'll open it myself. |
…a shared multiplexed bridge StdioBridge/bridgeTransport multiplexed every session of a command backend over one child process and one stdout stream, broadcasting every line to every subscriber and relying on each session's own JSON-RPC id to sort out which reply was whose. That doesn't hold: the go-sdk client numbers requests from 1 per connection, so two concurrent sessions on the same bridge could both send id:1 and each read the other's reply, not just their own - a silent cross-session data leak, independent of and worse than the drop bug fixed earlier on this branch. The earlier fix (an unbounded per-subscriber queue) kept all of that and added its own failure mode: a peer that stops reading its stream retains every other session's traffic indefinitely, turning "silent drop" into "unbounded growth" under the same trigger. Stdio MCP is single-session by spec, and no SDK multiplexes it, because a server-initiated request (sampling, elicitation) has no attributable destination on a shared process. Rather than teach the bridge a per-session id space (protocol-level code this package shouldn't own, and would still be new code to get subtly wrong), MCPService.backendTransport now builds a fresh mcp.CommandTransport - one subprocess per call - for command backends, the same "fresh transport per session" shape it already used for URL backends via StreamableClientTransport. Probe, Tools and HandleStreamPassThrough are unchanged: they only ever see an mcp.Transport. StdioBridge and bridgeTransport are deleted. Command backends lose the local SSE/POST HTTP ingress route StdioBridge.ServeHTTP used to serve (baseService.Init no longer builds a handler for them); the mesh stream path is what actually matters for cross-partner discovery/invocation and is unaffected. Handler()'s existing callers already treat a nil handler as "no local ingress for this service" (a 404), not a crash. Restoring a local-ingress equivalent is left for a follow-up. New tests (mcp_service_command_test.go) pin the property this exists for: two concurrent command-backend sessions, both using id:1 the way independent go-sdk clients naturally would, each only ever see their own subprocess's replies. Signed-off-by: Fernando Marino <fernando.marino85@gmail.com>
fer-marino
changed the title
|
@aojea You're right, and the id-collision you found is worse than what I was chasing - a shared process silently handing session A session B's tools/call result is a real correctness bug, not just a liveness one. Went with reworking it rather than leaving it to you, since it's a fairly contained change once you see it your way. Pushed the rework (commit 82a6a86): Two notes on the parts you flagged:
New tests in PR title/description updated to match. Full |
|
failures look related
|
| // A command backend gets its own subprocess, via mcp.CommandTransport, for | ||
| // every call - the same "one fresh transport per session" shape as the URL | ||
| // case, not a shared one. An earlier version of this multiplexed every | ||
| // session for a command backend over one child process and one stdout | ||
| // stream (StdioBridge/bridgeTransport, since deleted): every line | ||
| // broadcast to every subscriber, relying on each session's own JSON-RPC id | ||
| // to sort out which reply was whose. That doesn't hold - the go-sdk client | ||
| // numbers requests from 1 per connection, so two concurrent sessions on the | ||
| // same bridge could both send id:1 and each read the other's reply, not | ||
| // just their own. Fixing the multiplexer to not do that (id rewriting, a | ||
| // bridge-wide id space) is exactly the kind of protocol-level code this | ||
| // package shouldn't own: stdio MCP is single-session by spec, and no SDK | ||
| // multiplexes it, because a server-initiated request (sampling, | ||
| // elicitation) has no attributable destination on a shared process. One | ||
| // subprocess per session sidesteps all of it by construction, matching | ||
| // what backendTransport already does for URL backends. | ||
| // | ||
| // The cost is a fresh process per Probe/Tools call and per mesh stream | ||
| // instead of one long-lived one; command backends with slow startup pay | ||
| // that repeatedly. Command backends also lose the local SSE/POST HTTP | ||
| // ingress route StdioBridge.ServeHTTP used to provide (baseService.Init | ||
| // no longer builds a handler for them) - the mesh stream path | ||
| // (HandleStreamPassThrough) is what actually matters and is unaffected; | ||
| // restoring a local-ingress equivalent is left for later, not folded into | ||
| // this change. |
There was a problem hiding this comment.
try to tell the AI agent to not create this very large and verbose comments, I like the comments but this is a book :)
There was a problem hiding this comment.
Fair, noted - trimmed that one and a couple others down in 0f40c80.
The bridgeTransport removal also deleted StdioBridge.ServeHTTP, the local ingress route command backends use for direct HTTP+SSE access (distinct from the mesh path via HandleStreamPassThrough). Deleting it broke a real, CI-tested feature - tests/e2e/datapath.bats' "Stdio services are reachable across nodes" - not just a hypothetical. StdioBridge is back, trimmed to only what that route needs: its own bounded, drop-tolerant broadcast for ServeHTTP's GET/SSE and POST/call handling. Subscribe() and Send() are gone for good - those existed only for bridgeTransport, which stays deleted. This route was never the part of the id-collision bug aojea flagged: it already used a fresh callCh per POST, not a shared per-connection id space multiplexed across sessions the way bridgeTransport did. baseService goes back to building a StdioBridge handler and owning its process for command backends; MCPService.backendTransport is unchanged from the previous commit (still a fresh mcp.CommandTransport per session for Probe/Tools/HandleStreamPassThrough). Also trimmed several comments down that were too long for what they needed to say. Signed-off-by: Fernando Marino <fernando.marino85@gmail.com>
|
Yep - deleting StdioBridge.ServeHTTP along with bridgeTransport broke that route for real, not just hypothetically (the bats test proved it). Fixed in 0f40c80: StdioBridge is back, but trimmed to only what ServeHTTP needs. That route was never part of the id-collision bug - it already used a fresh channel per POST, no shared connection-wide id space. bridgeTransport/Subscribe() are still gone; MCPService.backendTransport is unchanged from 82a6a86. |
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request refactors the MCP service's command backend to use a fresh subprocess per call via mcp.CommandTransport instead of a shared StdioBridge, preventing session crosstalk. Unused bridge transport code and tests have been removed. Feedback highlights a potential nil pointer dereference or out-of-bounds panic when starting the command, suggesting a defensive check to avoid a denial-of-service vulnerability.
| case *api.RegisterServiceRequest_Command: | ||
| bridge, ok := m.handler.(*StdioBridge) | ||
| if !ok { | ||
| return nil, fmt.Errorf("expected *StdioBridge handler for command-backed MCP service %q, got %T", m.info.GetName(), m.handler) | ||
| cmd := exec.Command(x.Command.Command[0], x.Command.Command[1:]...) |
There was a problem hiding this comment.
Accessing x.Command.Command[0] directly without checking if x.Command is nil or if x.Command.Command is empty can lead to a nil pointer dereference or an out-of-bounds panic. Since this registration request can be triggered by external API clients or configurations, this represents a potential denial-of-service (DoS) vulnerability.
Adding a defensive check to ensure x.Command is non-nil and contains at least one command element prevents these panics and improves the robustness of the service.
case *api.RegisterServiceRequest_Command:
if x.Command == nil || len(x.Command.Command) == 0 {
return nil, fmt.Errorf("missing command for command-backed MCP service %q", m.info.GetName())
}
cmd := exec.Command(x.Command.Command[0], x.Command.Command[1:]...)gemini-code-assist flagged this on review: x.Command.Command[0] was read without checking x.Command for nil or Command for empty, so a RegisterService call with a command-backed MCP service but no actual command panicked backendTransport() instead of returning an error. Reachable from an external RegisterService call, not just local misconfiguration.
|
Fixed in ac74055, per your suggested check: |
ac74055 guarded backendTransport, but RegisterService reaches baseService.Init first, and createStdioBridgeHandler indexes Command[0] there: the same request still panicked before the new check ran. Reject it in NewServiceFromRequest, which every registration path goes through and which covers the Inference and A2A command backends too. The backendTransport check stays as defense in depth.
One subprocess per session turns an authorized peer holding streams open into a fork bomb; the only limits were the per-peer stream rate and the libp2p resource manager. Bound it per service with a slot pool: Connect takes a slot or fails fast, Close returns it after the go-sdk has reaped the child. Probe, Tools and HandleStreamPassThrough inherit the bound through backendTransport unchanged.
|
/gemini review added 3 commits for completeness and preserved the previous |
There was a problem hiding this comment.
Code Review
This pull request refactors command-backed MCP services to run each session in its own subprocess rather than sharing a single StdioBridge, preventing cross-talk between concurrent sessions. To prevent resource exhaustion, a session limit of 16 concurrent subprocesses is introduced using a bounded transport. Additionally, validation is added to reject command backends missing a command during service construction. The review feedback highlights potential flakiness in the new tests due to arbitrary time.Sleep calls and suggests replacing them with deterministic polling.
| // Give the handler a moment to register as a client before writing. | ||
| time.Sleep(50 * time.Millisecond) | ||
| _, _ = stdoutWriter.Write([]byte("hello\n")) | ||
| time.Sleep(50 * time.Millisecond) | ||
| cancel() |
There was a problem hiding this comment.
Rule 3: Test coverage must follow the pyramid
Using arbitrary time.Sleep calls to wait for client registration and response processing can lead to flaky tests in slow or resource-constrained CI environments. We can make this test 100% deterministic and faster by polling b.clients and rec.Body instead of sleeping.
| // Give the handler a moment to register as a client before writing. | |
| time.Sleep(50 * time.Millisecond) | |
| _, _ = stdoutWriter.Write([]byte("hello\n")) | |
| time.Sleep(50 * time.Millisecond) | |
| cancel() | |
| // Wait for the client to be registered to avoid flakiness. | |
| for { | |
| b.mu.Lock() | |
| registered := len(b.clients) > 0 | |
| b.mu.Unlock() | |
| if registered { | |
| break | |
| } | |
| time.Sleep(1 * time.Millisecond) | |
| } | |
| _, _ = stdoutWriter.Write([]byte("hello\n")) | |
| // Wait for the data to be written to the response. | |
| for { | |
| if strings.Contains(rec.Body.String(), "data: hello\n\n") { | |
| break | |
| } | |
| time.Sleep(1 * time.Millisecond) | |
| } | |
| cancel() |
References
- Integration tests must be time bounded. Look for unbounded waits, time.Sleep longer than a few hundred milliseconds, or polling without a deadline. (link)
| time.Sleep(50 * time.Millisecond) | ||
| reply := `{"jsonrpc":"2.0","id":1,"result":{}}` | ||
| _, _ = stdoutWriter.Write([]byte(reply + "\n")) |
There was a problem hiding this comment.
Rule 3: Test coverage must follow the pyramid
Using an arbitrary time.Sleep to wait for the call to be registered in b.calls can cause flakiness in CI. We can poll b.calls to ensure the call is registered before writing the reply, making the test deterministic and faster.
// Wait for the call to be registered to avoid flakiness.
for {
b.mu.Lock()
registered := len(b.calls) > 0
b.mu.Unlock()
if registered {
break
}
time.Sleep(1 * time.Millisecond)
}
reply := "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}"
_, _ = stdoutWriter.Write([]byte(reply + "\n"))References
- Integration tests must be time bounded. Look for unbounded waits, time.Sleep longer than a few hundred milliseconds, or polling without a deadline. (link)
The GET and POST tests synchronized with the handler goroutine via time.Sleep. Poll the bridge's client/call registration under its lock instead, and for the SSE test observe the handler's Write through a recorder wrapper rather than reading rec.Body while it is being written. Both waits fail the test after 2s instead of hanging.
|
once CI passes I merge, thanks @fer-marino |
investigating flake, but seems unrelated to the PR |
2 participants
Reworked following @aojea's review below - the original approach (an unbounded per-subscriber queue on a shared
StdioBridge) has been replaced entirely. History of how this PR got here is preserved in the commits and the comment thread; this description now describes the current state.The bug is in the sharing model, not the buffer.
StdioBridgemultiplexed every session of a command backend over one child process and one stdout stream, broadcasting every line to every subscriber and relying on each session's own JSON-RPC id to sort out which reply was whose. That doesn't hold: the go-sdk client numbers requests from 1 per connection, so two concurrent sessions on the same bridge could both sendid: 1and each read the other's reply - a silent cross-session data leak, not just the drop bug this PR originally set out to fix. The queue fix kept all of that and added its own failure mode: a peer that stops reading its stream retains every other session's traffic indefinitely (unbounded growth instead of a drop).Stdio MCP is single-session by spec, and no SDK multiplexes it, because a server-initiated request (sampling, elicitation) has no attributable destination on a shared process. So instead of teaching the bridge a per-session id space,
MCPService.backendTransportnow builds a freshmcp.CommandTransport- one subprocess per call - for command backends, the same "fresh transport per session" shape it already used for URL backends viaStreamableClientTransport.Probe,ToolsandHandleStreamPassThroughare unchanged: they only ever see anmcp.Transport.StdioBridgeandbridgeTransportare deleted. Command backends lose the local SSE/POST HTTP ingress routeStdioBridge.ServeHTTPused to serve (baseService.Initno longer builds a handler for them) - the mesh stream path is what actually matters for cross-partner discovery/invocation and is unaffected.Handler()'s existing callers already treat a nil handler as "no local ingress for this service" (a 404), not a crash. Restoring a local-ingress equivalent is left for a follow-up, per @aojea's own scoping below.New tests (
mcp_service_command_test.go) pin the property this exists for: two concurrent command-backend sessions, both usingid: 1the way independent go-sdk clients naturally would, each only ever see their own subprocess's replies.Tests:
go test ./internal/node/...- full suite unaffected beyond the five pre-existing, unrelated Windows-environment failures noted in the original description (reproduce identically on unmodified main). The new tests pass reliably under repeated runs (-count=10).