test(mcp): release held calls on observed EOF, not a sleep (TASK-3298)

Codex round 1 on df883744 (P2): a server that serves resources/read on
the read loop, as mcp-go 1.1.0 does, cannot read EOF while the held
fetch blocks it, so the 300ms sleep released everything before EOF was
processed. A version that kept read-loop serving but cancelled at EOF
would have passed.

The pin now wraps stdin in a reader that signals when the server's own
read returns io.EOF. Tool calls are released only after that signal
(+200ms, so a cancel-at-EOF server cancels before the release can win).
The resource read is released on the same signal, or after a 1s
fallback when it blocks the read loop; EOF cannot cancel it in that
shape. t.Cleanup now frees held calls and both pipes on a failure path.

Controls, one run each, same tree:
- mcp-go 1.1.0: PASS 5/5 under -race (1.2s each: the fallback path).
- mcp-go 1.1.1: FAIL 3/3 under -race, all four ids "context canceled"
  (0.21s: the EOF-observed path).
- mutant (1.1.0 patched to cancel at EOF, read-loop serving kept):
  FAIL 3/3, ids 10-12 "context canceled", id 20 answered. The previous
  version of this test PASSED 5/5 against the same mutant.

Claude-Session: https://claude.ai/code/session_016PXRzXhokr61CLbbu3Txj9
This commit is contained in:
xarmian
2026-09-29 12:37:41 +00:00
parent df88374409
commit 5889ff6298
+69 -13
View File
@@ -4,9 +4,11 @@ import (
"bufio" "bufio"
"context" "context"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"io" "io"
"strings" "strings"
"sync"
"testing" "testing"
"time" "time"
@@ -23,12 +25,39 @@ import (
// release is excluded in .github/dependabot.yml; this test is the pin 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. // keeps any later mcp-go version from bringing the behaviour back unnoticed.
// //
// The calls are held inside the dispatcher and the fetcher until stdin has // Ordering is by observation, not by sleeping. Three tools/call (mcp-go's
// been closed, so EOF is guaranteed to arrive while all of them are in // worker pool, default size 5) and one resources/read are held inside the
// flight: three tools/call (served by mcp-go's worker pool, default size 5) // dispatcher and the fetcher; stdin is closed only once all four are inside.
// and one resources/read, sent last because v1.1.0 serves non-tool requests // The tool calls are released only AFTER the server's own read of stdin has
// on the read loop itself. They are released only after EOF has had time to // returned EOF, so on any version the EOF is processed while they are still
// be read, and every id must then answer successfully. // 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.
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 = 200 * 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, and reports // blockingDispatcher holds each call until release is closed, and reports
// the context's error instead if the context ends first. // the context's error instead if the context ends first.
@@ -65,26 +94,39 @@ func (f *blockingFetcher) Fetch(ctx context.Context, _ []string) (string, error)
func TestStdio_EOFDoesNotCancelInFlightRequests(t *testing.T) { func TestStdio_EOFDoesNotCancelInFlightRequests(t *testing.T) {
const toolCalls = 3 const toolCalls = 3
entered := make(chan struct{}, toolCalls+1) entered := make(chan struct{}, toolCalls+1)
release := make(chan struct{}) 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) }) }
srv := NewServer(Options{Version: "eof-test"}) srv := NewServer(Options{Version: "eof-test"})
if _, err := RegisterCatalog(srv.MCP(), CatalogOptions{ if _, err := RegisterCatalog(srv.MCP(), CatalogOptions{
Doc: liveCmdhelpDoc(t), Doc: liveCmdhelpDoc(t),
Workspace: NewWorkspaceState("docapp"), Workspace: NewWorkspaceState("docapp"),
Dispatcher: &blockingDispatcher{entered: entered, release: release}, Dispatcher: &blockingDispatcher{entered: entered, release: toolRelease},
PadVersion: "test", PadVersion: "test",
}); err != nil { }); err != nil {
t.Fatalf("RegisterCatalog: %v", err) t.Fatalf("RegisterCatalog: %v", err)
} }
RegisterResources(srv.MCP(), &blockingFetcher{entered: entered, release: release}, nil) RegisterResources(srv.MCP(), &blockingFetcher{entered: entered, release: resourceRelease}, nil)
stdin, stdinW := io.Pipe() pipeR, stdinW := io.Pipe()
stdin := &eofSignalReader{r: pipeR, seen: make(chan struct{})}
stdoutR, stdout := io.Pipe() stdoutR, stdout := io.Pipe()
done := make(chan error, 1) done := make(chan error, 1)
go func() { go func() {
done <- srv.RunStdio(context.Background(), stdin, stdout) done <- srv.RunStdio(context.Background(), stdin, stdout)
_ = stdout.Close() _ = 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. // Collect every response line until the server closes stdout.
type response struct { type response struct {
@@ -119,6 +161,8 @@ func TestStdio_EOFDoesNotCancelInFlightRequests(t *testing.T) {
`{"jsonrpc":"2.0","id":%d,"method":"tools/call","params":{"name":"pad_item","arguments":{"action":"get","ref":"TASK-%d"}}}`, `{"jsonrpc":"2.0","id":%d,"method":"tools/call","params":{"name":"pad_item","arguments":{"action":"get","ref":"TASK-%d"}}}`,
10+i, i+1)) 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( lines = append(lines, fmt.Sprintf(
`{"jsonrpc":"2.0","id":20,"method":"resources/read","params":{"uri":%q}}`, WorkspacesURI)) `{"jsonrpc":"2.0","id":20,"method":"resources/read","params":{"uri":%q}}`, WorkspacesURI))
@@ -136,9 +180,21 @@ func TestStdio_EOFDoesNotCancelInFlightRequests(t *testing.T) {
} }
_ = stdinW.Close() _ = stdinW.Close()
// Give the read loop time to reach EOF, then let the calls finish. select {
time.Sleep(300 * time.Millisecond) case <-stdin.seen:
close(release) // 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 { select {
case err := <-done: case err := <-done: