mirror of
https://github.com/PerpetualSoftware/pad.git
synced 2026-10-25 05:49:50 +00:00
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:
@@ -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:
|
||||||
|
|||||||
Reference in New Issue
Block a user