mcp: bound and abandon transport reads

The stdio reader ran inline under the transport lock, and bufio never observes a context. A server that accepted a request and then wrote nothing held that lock forever. alive() takes the same lock and Refresh calls alive() while holding the manager lock, so one mute python server wedged Tools, Status and every Call, including turns that touch no MCP tool at all. The read now runs on its own goroutine feeding a channel, the call selects on the context, and a call that gives up drops the connection so the manager re-dials. The frame bound was measured after the line had been assembled, which is not a bound. A server emitting 500 MB with no newline had all 500 MB in mavend before the check could reject it, which on the deploy target is an OOM kill of the core daemon. The scanner's own buffer limit enforces it now. The HTTP transport never checked the response id. A server request sent mid-stream, sampling/createMessage or roots/list, unmarshalled into a response with neither result nor error, so the call reported success with an empty string. The act was logged as done and the tool never ran. The id must match and the frame must carry a result or an error. Found in review of #70.
2026-08-01 14:11:24 +04:00
parent 7f42cc73be
commit 87d03cf8c6
3 changed files with 223 additions and 58 deletions
@@ -35,13 +35,14 @@ type PostResponse struct {
 type httpTransport struct {
 	poster Poster
 	url    string
+	extra  map[string]string // static headers, e.g. an Authorization bearer

 	mu      sync.Mutex
 	session string // Mcp-Session-Id, echoed back when the server issues one
 }

-func newHTTPTransport(post Poster, endpoint string) *httpTransport {
-	return &httpTransport{poster: post, url: endpoint}
+func newHTTPTransport(post Poster, endpoint string, extra map[string]string) *httpTransport {
+	return &httpTransport{poster: post, url: endpoint, extra: extra}
 }

 func (t *httpTransport) Call(ctx context.Context, req *rpcRequest) (*rpcResponse, error) {
@@ -49,7 +50,7 @@ func (t *httpTransport) Call(ctx context.Context, req *rpcRequest) (*rpcResponse
 	if err != nil {
 		return nil, err
 	}
-	frame, err := decodeFrame(body)
+	frame, err := decodeFrame(body, req.ID)
 	if err != nil {
 		return nil, err
 	}
@@ -57,6 +58,18 @@ func (t *httpTransport) Call(ctx context.Context, req *rpcRequest) (*rpcResponse
 	if err := json.Unmarshal(frame, &resp); err != nil {
 		return nil, fmt.Errorf("mcp: decode response: %w", err)
 	}
+	// The id check the stdio transport already did. Without it a server that
+	// sends a request of its own (sampling/createMessage, roots/list) mid-stream
+	// has that request accepted as the answer: it unmarshals into an rpcResponse
+	// with neither result nor error, and the call reports success with nothing
+	// in it. An empty string and no error is the one answer that lies — the act
+	// is logged as done and the tool never ran.
+	if resp.ID == nil || *resp.ID != req.ID {
+		return nil, fmt.Errorf("mcp: response id mismatch (wanted %d)", req.ID)
+	}
+	if resp.Error == nil && len(resp.Result) == 0 {
+		return nil, errors.New("mcp: response carries neither result nor error")
+	}
 	return &resp, nil
 }

@@ -71,7 +84,14 @@ func (t *httpTransport) send(ctx context.Context, req *rpcRequest) ([]byte, erro
 	if err != nil {
 		return nil, err
 	}
-	hdr := map[string]string{"Accept": "application/json, text/event-stream"}
+	hdr := map[string]string{}
+	// Configured headers first, so nothing here can be overwritten by them:
+	// a real remote server needs a bearer token, and the Vikunja one on
+	// loopback is only reachable without one because it is unauthenticated.
+	for k, v := range t.extra {
+		hdr[k] = v
+	}
+	hdr["Accept"] = "application/json, text/event-stream"
 	t.mu.Lock()
 	if t.session != "" {
 		hdr["Mcp-Session-Id"] = t.session
@@ -114,9 +134,10 @@ func headerGet(h map[string]string, key string) string {
 }

 // decodeFrame pulls the JSON object out of a body that is either raw JSON or
-// SSE. For SSE we take the LAST data: payload that parses, which is the
-// response — earlier frames on the stream are progress notifications.
-func decodeFrame(body []byte) ([]byte, error) {
+// SSE. For SSE we take the last data: payload that parses AND carries our own
+// id with a result or an error in it. Matching on the presence of an "id" key
+// alone is not enough: a JSON-RPC request from the server has one too.
+func decodeFrame(body []byte, id int64) ([]byte, error) {
 	trimmed := bytes.TrimSpace(body)
 	if len(trimmed) == 0 {
 		return nil, errors.New("mcp: empty response body")
@@ -136,19 +157,28 @@ func decodeFrame(body []byte) ([]byte, error) {
 		if payload == "" {
 			continue
 		}
-		var probe map[string]json.RawMessage
+		var probe struct {
+			ID     *int64          `json:"id"`
+			Result json.RawMessage `json:"result"`
+			Error  json.RawMessage `json:"error"`
+			Method string          `json:"method"`
+		}
 		if json.Unmarshal([]byte(payload), &probe) != nil {
 			continue
 		}
-		if _, isResp := probe["id"]; isResp {
-			last = []byte(payload)
+		if probe.Method != "" || probe.ID == nil || *probe.ID != id {
+			continue
 		}
+		if len(probe.Result) == 0 && len(probe.Error) == 0 {
+			continue
+		}
+		last = []byte(payload)
 	}
 	if err := sc.Err(); err != nil {
 		return nil, fmt.Errorf("mcp: read event stream: %w", err)
 	}
 	if last == nil {
-		return nil, errors.New("mcp: no JSON-RPC response in event stream")
+		return nil, fmt.Errorf("mcp: no JSON-RPC response for id %d in event stream", id)
 	}
 	return last, nil
 }
@@ -2,6 +2,7 @@ package mcp

 import (
 	"bufio"
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -9,12 +10,17 @@ import (
 	"io"
 	"os"
 	"os/exec"
-	"strings"
 	"sync"
 )

 // maxLine bounds one JSON-RPC frame from a subprocess. A tool result bigger
 // than this is a misbehaving server, not something to buffer.
+//
+// The bound is enforced by bufio.Scanner's own buffer limit, not by measuring
+// the line after it was assembled. Measuring afterwards is not a bound: a
+// server that emits 500 MB with no newline would have all 500 MB in mavend's
+// heap before the check could reject it, which on the deploy target is an OOM
+// kill of the core daemon.
 const maxLine = 1 << 20 // 1 MiB

 // stdioTransport speaks newline-delimited JSON-RPC to a child process. This is
@@ -23,12 +29,27 @@ const maxLine = 1 << 20 // 1 MiB
 //
 // Args are argv, never a shell string — the same discipline internal/tool
 // keeps, for the same reason.
+//
+// Reading happens on its own goroutine, feeding frames down a channel. That is
+// what makes a call abandonable: bufio never observes a context, so a server
+// that accepts a request and then writes nothing at all would otherwise block
+// the reader forever with the transport lock held, and every other server in
+// the manager behind it.
 type stdioTransport struct {
-	mu   sync.Mutex
-	cmd  *exec.Cmd
-	in   io.WriteCloser
-	out  *bufio.Reader
-	dead bool
+	cmd   *exec.Cmd
+	in    io.WriteCloser
+	lines chan []byte
+	stop  chan struct{} // closed by Close, so the reader can give up
+
+	// callMu serialises whole calls, so two callers cannot consume each
+	// other's frames off the shared channel. It is deliberately NOT the lock
+	// alive() takes: a hung call must not make the manager's health check
+	// block on it.
+	callMu sync.Mutex
+
+	mu      sync.Mutex
+	dead    bool
+	readErr error
 }

 func newStdioTransport(ctx context.Context, argv []string, env []string, dir string) (*stdioTransport, error) {
@@ -52,38 +73,92 @@ func newStdioTransport(ctx context.Context, argv []string, env []string, dir str
 	if err := cmd.Start(); err != nil {
 		return nil, fmt.Errorf("mcp: start %q: %w", argv[0], err)
 	}
-	return &stdioTransport{cmd: cmd, in: in, out: bufio.NewReaderSize(out, 64<<10)}, nil
+	t := &stdioTransport{
+		cmd:   cmd,
+		in:    in,
+		lines: make(chan []byte),
+		stop:  make(chan struct{}),
+	}
+	go t.readLoop(out)
+	return t, nil
+}
+
+// readLoop pushes one frame per line onto t.lines until the pipe ends. The
+// scanner's own buffer limit is the frame bound: a line longer than maxLine
+// ends the scan with bufio.ErrTooLong having buffered at most maxLine, rather
+// than assembling the whole thing first and rejecting it afterwards.
+func (t *stdioTransport) readLoop(out io.Reader) {
+	defer close(t.lines)
+	sc := bufio.NewScanner(out)
+	sc.Buffer(make([]byte, 0, 64<<10), maxLine)
+	for sc.Scan() {
+		line := bytes.TrimSpace(sc.Bytes())
+		if len(line) == 0 {
+			continue
+		}
+		frame := append([]byte(nil), line...)
+		select {
+		case t.lines <- frame:
+		case <-t.stop:
+			return
+		}
+	}
+	err := sc.Err()
+	switch {
+	case errors.Is(err, bufio.ErrTooLong):
+		err = fmt.Errorf("mcp: frame exceeds %d bytes", maxLine)
+	case err == nil:
+		err = io.EOF
+	}
+	t.mu.Lock()
+	t.readErr = err
+	t.mu.Unlock()
 }

 func (t *stdioTransport) Call(ctx context.Context, req *rpcRequest) (*rpcResponse, error) {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	if t.dead {
+	t.callMu.Lock()
+	defer t.callMu.Unlock()
+	if !t.alive() {
 		return nil, ErrClosed
 	}
-	if err := t.write(req); err != nil {
-		t.dead = true
+	t.mu.Lock()
+	err := t.write(req)
+	t.mu.Unlock()
+	if err != nil {
+		_ = t.Close()
 		return nil, err
 	}
 	// Read until the frame with our id turns up; anything else on the pipe is
 	// a notification or a server-initiated request we do not answer.
 	for {
-		if err := ctx.Err(); err != nil {
-			return nil, err
+		select {
+		case <-ctx.Done():
+			// A server that took the request and answered nothing is not a
+			// server this connection can be reused with: the next call would
+			// read into a pipe whose state we no longer know. Drop it and let
+			// the manager re-dial.
+			_ = t.Close()
+			return nil, ctx.Err()
+		case line, ok := <-t.lines:
+			if !ok {
+				t.mu.Lock()
+				rerr := t.readErr
+				t.mu.Unlock()
+				_ = t.Close()
+				if rerr == nil {
+					rerr = ErrClosed
+				}
+				return nil, fmt.Errorf("mcp: read: %w", rerr)
+			}
+			var resp rpcResponse
+			if err := json.Unmarshal(line, &resp); err != nil {
+				continue // not a response frame; ignore rather than break the turn
+			}
+			if resp.ID == nil || *resp.ID != req.ID {
+				continue
+			}
+			return &resp, nil
 		}
-		line, err := t.readLine()
-		if err != nil {
-			t.dead = true
-			return nil, err
-		}
-		var resp rpcResponse
-		if err := json.Unmarshal(line, &resp); err != nil {
-			continue // not a response frame; ignore rather than break the turn
-		}
-		if resp.ID == nil || *resp.ID != req.ID {
-			continue
-		}
-		return &resp, nil
 	}
 }

@@ -96,6 +171,7 @@ func (t *stdioTransport) Notify(ctx context.Context, method string, params any)
 	return t.write(&rpcRequest{JSONRPC: "2.0", Method: method, Params: params})
 }

+// write must be called with t.mu held.
 func (t *stdioTransport) write(req *rpcRequest) error {
 	req.JSONRPC = "2.0"
 	raw, err := json.Marshal(req)
@@ -108,31 +184,20 @@ func (t *stdioTransport) write(req *rpcRequest) error {
 	return nil
 }

-func (t *stdioTransport) readLine() ([]byte, error) {
-	for {
-		line, err := t.out.ReadString('\n')
-		if err != nil {
-			if len(strings.TrimSpace(line)) == 0 {
-				return nil, fmt.Errorf("mcp: read: %w", err)
-			}
-			return []byte(line), nil
-		}
-		if len(line) > maxLine {
-			return nil, fmt.Errorf("mcp: frame exceeds %d bytes", maxLine)
-		}
-		if s := strings.TrimSpace(line); s != "" {
-			return []byte(s), nil
-		}
-	}
-}
-
+// Close is idempotent: a call that abandoned a silent pipe calls it, and so
+// does the manager.
 func (t *stdioTransport) Close() error {
 	t.mu.Lock()
-	defer t.mu.Unlock()
+	if t.dead {
+		t.mu.Unlock()
+		return nil
+	}
 	t.dead = true
+	close(t.stop)
 	if t.in != nil {
 		_ = t.in.Close()
 	}
+	t.mu.Unlock()
 	if t.cmd.Process != nil {
 		_ = t.cmd.Process.Kill()
 		_ = t.cmd.Wait()
@@ -145,5 +210,8 @@ func (t *stdioTransport) Close() error {
 func (t *stdioTransport) alive() bool {
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	return !t.dead
+	if t.dead {
+		return false
+	}
+	return t.readErr == nil
 }
@@ -9,6 +9,7 @@ import (
 	"os/exec"
 	"strings"
 	"testing"
+	"time"
 )

 // The stdio transport is tested against a real subprocess — this test binary,
@@ -47,6 +48,20 @@ func fakeStdioServer() {
 			_ = out.Flush()
 			continue
 		}
+		// "mute" answers the handshake and then goes silent: a python server
+		// that hit an unhandled exception in its own read loop but did not
+		// exit is the ordinary way to get here.
+		if os.Getenv("MAVEN_MCP_FAKE") == "mute" && req.Method == "tools/call" {
+			select {} // never answer, never exit
+		}
+		// "flood" writes one enormous line with no newline in it.
+		if os.Getenv("MAVEN_MCP_FAKE") == "flood" && req.Method == "tools/call" {
+			for i := 0; i < 64; i++ {
+				_, _ = out.Write(make([]byte, 1<<20))
+			}
+			_ = out.Flush()
+			continue
+		}
 		result, rerr := h(req.Method, req.Params)
 		resp := map[string]any{"jsonrpc": "2.0", "id": *req.ID}
 		if rerr != nil {
@@ -147,3 +162,55 @@ func TestStdioMissingCommand(t *testing.T) {
 		t.Logf("err = %q", st[0].Err)
 	}
 }
+
+// A stdio server that accepts a call and then answers nothing must not wedge
+// the manager. Before the read moved onto its own goroutine, the read held the
+// transport lock, Refresh took that lock through alive() while holding the
+// manager lock, and from then on Tools, Status and Call blocked for EVERY
+// server — including turns that touch no MCP tool at all.
+func TestStdioSilentServerDoesNotWedgeTheManager(t *testing.T) {
+	m := stdioManager(t, "mute")
+	defer m.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+	done := make(chan error, 1)
+	go func() {
+		_, err := m.Call(ctx, "fake", "read_thing", nil)
+		done <- err
+	}()
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("a call into a silent server must fail, not succeed")
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("the call never returned: the context is not observed during the read")
+	}
+
+	// The manager must still answer while (and after) that call was stuck.
+	ready := make(chan struct{})
+	go func() {
+		m.Refresh(context.Background())
+		m.Tools()
+		m.Status()
+		close(ready)
+	}()
+	select {
+	case <-ready:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Refresh/Tools/Status deadlocked behind the hung call")
+	}
+}
+
+// One frame is bounded by the reader's buffer, not measured after the whole
+// thing has already been assembled in mavend's heap.
+func TestStdioOversizedFrameIsRefused(t *testing.T) {
+	m := stdioManager(t, "flood")
+	defer m.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if _, err := m.Call(ctx, "fake", "read_thing", nil); err == nil {
+		t.Fatal("a 64 MiB frame must be refused")
+	}
+}