diff --git a/.github/dependabot.yml b/.github/dependabot.yml index fdce94799..b2488547a 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -25,6 +25,14 @@ updates: commit-message: prefix: "chore(deps)" include: scope + ignore: + # mcp-go v1.1.1 cancels every in-flight stdio request at stdin EOF + # (upstream #977), so `pad mcp serve` fed a piped batch answers + # `context canceled` for each dispatcher-backed call (TASK-3298). + # Only this release is excluded; internal/mcp's + # TestStdio_EOFDoesNotCancelInFlightRequests guards the next one. + - dependency-name: "github.com/mark3labs/mcp-go" + versions: ["1.1.1"] groups: go-minor-and-patch: applies-to: version-updates diff --git a/go.mod b/go.mod index 5aaec3709..8b5aa498e 100644 --- a/go.mod +++ b/go.mod @@ -30,7 +30,7 @@ require ( github.com/jackc/pgx/v5 v5.11.0 github.com/mark3labs/mcp-go v1.1.0 github.com/ory/fosite v0.49.0 - github.com/pb33f/libopenapi v0.38.7 + github.com/pb33f/libopenapi v0.40.1 github.com/pquerna/otp v1.5.0 github.com/prometheus/client_golang v1.24.1 github.com/prometheus/client_model v0.6.3 @@ -98,7 +98,7 @@ require ( github.com/ory/go-acc v0.2.9-0.20230103102148-6b1c9a70dbbe // indirect github.com/ory/go-convenience v0.1.0 // indirect github.com/ory/x v0.0.665 // indirect - github.com/pb33f/jsonpath v0.8.2 // indirect + github.com/pb33f/jsonpath v0.8.3 // indirect github.com/pb33f/ordered-map/v2 v2.3.1 // indirect github.com/pelletier/go-toml/v2 v2.0.9 // indirect github.com/pkg/errors v0.9.1 // indirect diff --git a/go.sum b/go.sum index 187c2cfaf..a8dc7a3e8 100644 --- a/go.sum +++ b/go.sum @@ -391,10 +391,10 @@ github.com/ory/jsonschema/v3 v3.0.8 h1:Ssdb3eJ4lDZ/+XnGkvQS/te0p+EkolqwTsDOCxr/F github.com/ory/jsonschema/v3 v3.0.8/go.mod h1:ZPzqjDkwd3QTnb2Z6PAS+OTvBE2x5i6m25wCGx54W/0= github.com/ory/x v0.0.665 h1:61vv0ObCDSX1vOQYbxBeqDiv4YiPmMT91lYxDaaKX08= github.com/ory/x v0.0.665/go.mod h1:7SCTki3N0De3ZpqlxhxU/94ZrOCfNEnXwVtd0xVt+L8= -github.com/pb33f/jsonpath v0.8.2 h1:Ou4C7zjYClBm97dfZjDCjdZGusJoynv/vrtiEKNfj2Y= -github.com/pb33f/jsonpath v0.8.2/go.mod h1:zBV5LJW4OQOPatmQE2QdKpGQJvhDTlE5IEj6ASaRNTo= -github.com/pb33f/libopenapi v0.38.7 h1:Q2jfgRPdnU38WW8wQvrX2HEPGiqsxj01PX1BHmAEihc= -github.com/pb33f/libopenapi v0.38.7/go.mod h1:naZ03Auhn7i+RJtMv8ck8l7Ag8E2/x2w66j9vsDFL38= +github.com/pb33f/jsonpath v0.8.3 h1:gdOUYn31vic8iRSSR4I2SEv7UQf2WxLeDcUnDPpGZNE= +github.com/pb33f/jsonpath v0.8.3/go.mod h1:zBV5LJW4OQOPatmQE2QdKpGQJvhDTlE5IEj6ASaRNTo= +github.com/pb33f/libopenapi v0.40.1 h1:Ma+2kxag5ox70sSP4dcgGY/4QGZ0VS5Ov8S/z4POyIc= +github.com/pb33f/libopenapi v0.40.1/go.mod h1:nISN8KLZBliHRwZY0V53JZJtuzI8P52WGfJ6J56z9Tg= github.com/pb33f/ordered-map/v2 v2.3.1 h1:5319HDO0aw4DA4gzi+zv4FXU9UlSs3xGZ40wcP1nBjY= github.com/pb33f/ordered-map/v2 v2.3.1/go.mod h1:qxFQgd0PkVUtOMCkTapqotNgzRhMPL7VvaHKbd1HnmQ= github.com/pb33f/testify v0.1.0 h1:g48/HDU/jn2COspS4nM0scptxiKTJ4DnbX/4ehK6IZ8= diff --git a/internal/mcp/stdio_eof_inflight_test.go b/internal/mcp/stdio_eof_inflight_test.go new file mode 100644 index 000000000..2b317f164 --- /dev/null +++ b/internal/mcp/stdio_eof_inflight_test.go @@ -0,0 +1,258 @@ +package mcp + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/mark3labs/mcp-go/mcp" +) + +// TASK-3298: stdin EOF must not cancel requests that are already in flight. +// +// `pad mcp serve` is also driven non-interactively: a script pipes a batch of +// JSON-RPC requests in and closes stdin, expecting one answer per id. mcp-go +// v1.1.1 changed StdioServer.Listen to cancel every in-flight request's +// context the moment it reads EOF (upstream #977), so each dispatcher-backed +// call answered `server_error: context canceled` instead of its result. That +// release is excluded in .github/dependabot.yml; this test is the pin that +// keeps any later mcp-go version from bringing the behaviour back unnoticed. +// +// Ordering is by observation, not by sleeping. Three tools/call (mcp-go's +// worker pool, default size 5) and one resources/read are held inside the +// dispatcher and the fetcher; stdin is closed only once all four are inside. +// The tool calls are released only AFTER the server's own read of stdin has +// returned EOF, so on any version the EOF is processed while they are still +// in flight. The resource read is released on the same signal, or after +// resourceFallback when the server serves it on the read loop itself (as +// v1.1.0 does) and therefore cannot read EOF until it returns; in that shape +// EOF cannot cancel it, and the tool calls are still held across the EOF. +// +// Each held call reads its own context when it returns, and the test asserts +// none was cancelled, independently of the JSON-RPC responses: a select does +// not prefer ctx.Done over a ready release, and a server could cancel yet +// still answer successfully, so the responses alone cannot prove it. +// +// Two bounds, stated rather than implied. (1) The signal fires inside Read, +// before the server has handled the EOF, and a server that never cancels +// emits nothing to wait for, so the release after it is a timed settle: a +// cancel-at-EOF server whose EOF handling is delayed past eofSettle would +// pass. That error runs one way only; a server that keeps in-flight calls +// alive can never be failed by it. (2) The entry gate needs all four calls +// in flight at once, so a server that cannot hold them (tool calls served on +// the read loop, fewer than three concurrent tool workers, or a change that +// stops these requests reaching the injected handlers) fails it with "only N +// of 4 calls reached"; that means this pin's premise no longer holds, not +// that EOF cancelled anything. + +const ( + // eofSettle is how long a release waits after EOF was observed, so a + // cancel-at-EOF server has cancelled before the release can win the race. + eofSettle = 500 * time.Millisecond + // resourceFallback releases a resource read that blocks the read loop. + resourceFallback = time.Second +) + +// eofSignalReader closes seen the first time the wrapped reader returns +// io.EOF, i.e. when the server's read loop has actually reached EOF. +type eofSignalReader struct { + r io.Reader + once sync.Once + seen chan struct{} +} + +func (e *eofSignalReader) Read(p []byte) (int, error) { + n, err := e.r.Read(p) + if errors.Is(err, io.EOF) { + e.once.Do(func() { close(e.seen) }) + } + return n, err +} + +// blockingDispatcher holds each call until release is closed or its context +// ends, then counts the call in cancelled if its context had ended, whichever +// select branch ran. +type blockingDispatcher struct { + entered chan struct{} + release chan struct{} + cancelled *atomic.Int32 +} + +func (d *blockingDispatcher) Dispatch(ctx context.Context, _ []string, _ []string) (*mcp.CallToolResult, error) { + d.entered <- struct{}{} + select { + case <-d.release: + case <-ctx.Done(): + } + if err := ctx.Err(); err != nil { + d.cancelled.Add(1) + return nil, err + } + return mcp.NewToolResultText("dispatched"), nil +} + +type blockingFetcher struct { + entered chan struct{} + release chan struct{} + cancelled *atomic.Int32 +} + +func (f *blockingFetcher) Fetch(ctx context.Context, _ []string) (string, error) { + f.entered <- struct{}{} + select { + case <-f.release: + case <-ctx.Done(): + } + if err := ctx.Err(); err != nil { + f.cancelled.Add(1) + return "", err + } + return `[{"slug":"docapp","name":"Pad"}]`, nil +} + +func TestStdio_EOFDoesNotCancelInFlightRequests(t *testing.T) { + const toolCalls = 3 + entered := make(chan struct{}, toolCalls+1) + toolRelease := make(chan struct{}) + resourceRelease := make(chan struct{}) + var releaseTools, releaseResource sync.Once + freeTools := func() { releaseTools.Do(func() { close(toolRelease) }) } + freeResource := func() { releaseResource.Do(func() { close(resourceRelease) }) } + var cancelled atomic.Int32 + + srv := NewServer(Options{Version: "eof-test"}) + if _, err := RegisterCatalog(srv.MCP(), CatalogOptions{ + Doc: liveCmdhelpDoc(t), + Workspace: NewWorkspaceState("docapp"), + Dispatcher: &blockingDispatcher{entered: entered, release: toolRelease, cancelled: &cancelled}, + PadVersion: "test", + }); err != nil { + t.Fatalf("RegisterCatalog: %v", err) + } + RegisterResources(srv.MCP(), &blockingFetcher{entered: entered, release: resourceRelease, cancelled: &cancelled}, nil) + + pipeR, stdinW := io.Pipe() + stdin := &eofSignalReader{r: pipeR, seen: make(chan struct{})} + stdoutR, stdout := io.Pipe() + done := make(chan error, 1) + go func() { + done <- srv.RunStdio(context.Background(), stdin, stdout) + _ = stdout.Close() + }() + // On any failure path, unblock every held call and both pipes so no + // goroutine outlives the test. + t.Cleanup(func() { + freeTools() + freeResource() + _ = stdinW.Close() + _ = stdoutR.Close() + }) + + // Collect every response line until the server closes stdout. + type response struct { + ID int `json:"id"` + Error *json.RawMessage `json:"error"` + Result struct { + IsError bool `json:"isError"` + } `json:"result"` + raw string + } + responses := make(chan []response, 1) + go func() { + var got []response + sc := bufio.NewScanner(stdoutR) + sc.Buffer(make([]byte, 1<<20), 1<<24) + for sc.Scan() { + var r response + if json.Unmarshal(sc.Bytes(), &r) == nil { + r.raw = sc.Text() + got = append(got, r) + } + } + responses <- got + }() + + lines := []string{ + `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"eof-test","version":"0"}}}`, + `{"jsonrpc":"2.0","method":"notifications/initialized"}`, + } + for i := 0; i < toolCalls; i++ { + lines = append(lines, fmt.Sprintf( + `{"jsonrpc":"2.0","id":%d,"method":"tools/call","params":{"name":"pad_item","arguments":{"action":"get","ref":"TASK-%d"}}}`, + 10+i, i+1)) + } + // Last, because a server that serves it on the read loop reads nothing + // after it until it returns. + lines = append(lines, fmt.Sprintf( + `{"jsonrpc":"2.0","id":20,"method":"resources/read","params":{"uri":%q}}`, WorkspacesURI)) + + go func() { + _, _ = io.WriteString(stdinW, strings.Join(lines, "\n")+"\n") + }() + + // Every call is inside the dispatcher or fetcher before stdin closes. + for i := 0; i < toolCalls+1; i++ { + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatalf("only %d of %d calls reached the dispatcher/fetcher", i, toolCalls+1) + } + } + _ = stdinW.Close() + + select { + case <-stdin.seen: + // Off-loop resource read: EOF arrived with everything still held. + case <-time.After(resourceFallback): + // On-loop resource read: it must return before EOF can be read. + freeResource() + select { + case <-stdin.seen: + case <-time.After(5 * time.Second): + t.Fatal("server never read stdin EOF") + } + } + time.Sleep(eofSettle) + freeResource() + freeTools() + + select { + case err := <-done: + if err != nil { + t.Fatalf("RunStdio: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("RunStdio did not return after stdin EOF") + } + got := <-responses + + if n := cancelled.Load(); n != 0 { + t.Errorf("%d held call(s) returned with a cancelled context after stdin EOF", n) + } + byID := map[int]response{} + for _, r := range got { + byID[r.ID] = r + } + want := []int{1, 20} + for i := 0; i < toolCalls; i++ { + want = append(want, 10+i) + } + for _, id := range want { + r, ok := byID[id] + if !ok { + t.Errorf("id %d: no response after stdin EOF", id) + continue + } + if r.Error != nil || r.Result.IsError { + t.Errorf("id %d: answered with an error after stdin EOF: %s", id, r.raw) + } + } +}