Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 10 additions & 11 deletions internal/jsonrpc2/conn.go
Original file line number Diff line number Diff line change
Expand Up @@ -297,7 +297,7 @@ func (c *Connection) Notify(ctx context.Context, method string, params any) (err
// cannot increase once shutdown begins, and allowing outgoing notifications
// may permit notifications that will cancel in-flight calls.
if len(s.outgoingCalls) == 0 && len(s.incomingByID) == 0 {
err = s.shuttingDown(ErrClientClosing)
err = s.shuttingDown(ErrConnClosing)
if err != nil {
return
}
Expand Down Expand Up @@ -341,7 +341,7 @@ func (c *Connection) Call(ctx context.Context, method string, params any) *Async
}

c.updateInFlight(func(s *inFlightState) {
err = s.shuttingDown(ErrClientClosing)
err = s.shuttingDown(ErrConnClosing)
if err != nil {
return
}
Expand Down Expand Up @@ -546,7 +546,7 @@ func (c *Connection) wait(fromWait bool) error {
//
// After the start of a Close, notification requests (that lack IDs and do not
// receive responses) will continue to be passed to the Preempter, but calls
// with IDs will receive immediate responses with ErrServerClosing, and no new
// with IDs will receive immediate responses with ErrConnClosing, and no new
// requests (not even notifications!) will be enqueued to the Handler.
func (c *Connection) Close() error {
// Stop handling new requests, and interrupt the reader (by closing the
Expand Down Expand Up @@ -592,16 +592,15 @@ func (c *Connection) readIncoming(ctx context.Context, reader Reader, preempter
// Retire any outgoing requests that were still in flight: with the Reader no
// longer being processed, they necessarily cannot receive a response.
for id, ac := range s.outgoingCalls {
ac.retire(&Response{ID: id, Error: fmt.Errorf("%w: %w", ErrClientClosing, err)})
ac.retire(&Response{ID: id, Error: fmt.Errorf("%w: %w", ErrConnClosing, err)})
}
s.outgoingCalls = nil

// Cancel any incoming requests still in flight: with the reader gone we
// cannot receive cancellation notifications, and likely cannot write a
// response either, so parked handlers have nothing useful left to do.
// The read error (typically io.EOF from the peer disconnecting) is the
// cause, rather than a directional ErrServerClosing/ErrClientClosing
// sentinel: at this layer the connection has no designated end.
// The cause is the read error, typically io.EOF from the peer
// disconnecting.
for _, r := range s.incomingByID {
r.cancel(err)
}
Expand Down Expand Up @@ -642,7 +641,7 @@ func (c *Connection) acceptRequest(ctx context.Context, msg *Request, preempter
// theoretically be handled by the preempter. The preempter could return
// ErrAsyncResponse, which would increase the amount of work in flight
// when we're trying to ensure that it strictly decreases.
err = s.shuttingDown(ErrServerClosing)
err = s.shuttingDown(ErrConnClosing)
}
})
if err != nil {
Expand All @@ -664,7 +663,7 @@ func (c *Connection) acceptRequest(ctx context.Context, msg *Request, preempter
// handler — not even notifications. That ensures that if the handler
// continues to make progress, it will eventually become idle and
// close the connection.
err = s.shuttingDown(ErrServerClosing)
err = s.shuttingDown(ErrConnClosing)
if err != nil {
return
}
Expand Down Expand Up @@ -807,7 +806,7 @@ func (c *Connection) write(ctx context.Context, msg Message) error {
if req, ok := msg.(*Request); ok && !req.IsCall() && s.outgoingNotifications > 0 {
return
}
err = s.shuttingDown(ErrServerClosing)
err = s.shuttingDown(ErrConnClosing)
})
if err == nil {
err = c.writer.Write(ctx, msg)
Expand All @@ -828,7 +827,7 @@ func (c *Connection) write(ctx context.Context, msg Message) error {
if s.writeErr == nil {
s.writeErr = err
for _, r := range s.incomingByID {
r.cancel(fmt.Errorf("%w: %v", ErrServerClosing, err))
r.cancel(fmt.Errorf("%w: %v", ErrConnClosing, err))
}
}
})
Expand Down
14 changes: 7 additions & 7 deletions internal/jsonrpc2/conn_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -88,9 +88,9 @@ func TestUnmarshalableResultSendsError(t *testing.T) {
// failure).
func TestShuttingDownWrapsWriteError(t *testing.T) {
s := &inFlightState{writeErr: io.EOF}
err := s.shuttingDown(ErrServerClosing)
if !errors.Is(err, ErrServerClosing) {
t.Errorf("shuttingDown() error = %v, want it to wrap ErrServerClosing", err)
err := s.shuttingDown(ErrConnClosing)
if !errors.Is(err, ErrConnClosing) {
t.Errorf("shuttingDown() error = %v, want it to wrap ErrConnClosing", err)
}
if !errors.Is(err, io.EOF) {
t.Errorf("shuttingDown() error = %v, want it to wrap io.EOF", err)
Expand All @@ -99,9 +99,9 @@ func TestShuttingDownWrapsWriteError(t *testing.T) {

func TestShuttingDownWrapsReadError(t *testing.T) {
s := &inFlightState{readErr: io.EOF}
err := s.shuttingDown(ErrServerClosing)
if !errors.Is(err, ErrServerClosing) {
t.Errorf("shuttingDown() error = %v, want it to wrap ErrServerClosing", err)
err := s.shuttingDown(ErrConnClosing)
if !errors.Is(err, ErrConnClosing) {
t.Errorf("shuttingDown() error = %v, want it to wrap ErrConnClosing", err)
}
if !errors.Is(err, io.EOF) {
t.Errorf("shuttingDown() error = %v, want it to wrap io.EOF", err)
Expand Down Expand Up @@ -137,7 +137,7 @@ func TestReadErrorRetiresCalls(t *testing.T) {
<-w.out
fail()
err := call.Await(ctx, nil)
for _, want := range []error{ErrClientClosing, readErr} {
for _, want := range []error{ErrConnClosing, readErr} {
if !errors.Is(err, want) {
t.Errorf("Await() = %v, want an error wrapping %v", err, want)
}
Expand Down
9 changes: 5 additions & 4 deletions internal/jsonrpc2/wire.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,11 @@ var (

// ErrUnknown should be used for all non coded errors.
ErrUnknown = NewError(-32001, "unknown error")
// ErrServerClosing is returned for calls that arrive while the server is closing.
ErrServerClosing = NewError(-32004, "server is closing")
// ErrClientClosing is a dummy error returned for calls initiated while the client is closing.
ErrClientClosing = NewError(-32003, "client is closing")
// ErrConnClosing is returned for calls and notifications that are sent or
// received while the connection is closing. Calls received while closing
// are answered with it, so it keeps the code that peers already send for
// this condition.
ErrConnClosing = NewError(-32004, "connection is closing")

// The following errors have special semantics for MCP transports

Expand Down
8 changes: 3 additions & 5 deletions mcp/cmd_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -94,13 +94,11 @@ func TestStdioCallInterruptedByPeerExit(t *testing.T) {
t.Errorf("%s = %v, want an error wrapping %v", name, err, want)
}
}
for _, closing := range []error{jsonrpc2.ErrClientClosing, jsonrpc2.ErrServerClosing} {
if errors.Is(err, closing) {
t.Errorf("%s = %v, must not wrap %v", name, err, closing)
}
if errors.Is(err, jsonrpc2.ErrConnClosing) {
t.Errorf("%s = %v, must not wrap %v", name, err, jsonrpc2.ErrConnClosing)
}
var wireErr *jsonrpc2.WireError
if errors.As(err, &wireErr) && (wireErr.Code == -32003 || wireErr.Code == -32004) {
if errors.As(err, &wireErr) && wireErr.Code == -32004 {
t.Errorf("%s = %v, exposes internal closing code %d", name, err, wireErr.Code)
}
}
Expand Down
2 changes: 1 addition & 1 deletion mcp/shared.go
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,7 @@ func defaultSendingMethodHandler(ctx context.Context, method string, req Request
// Notifications don't have results.
if strings.HasPrefix(method, "notifications/") {
err := req.GetSession().getConn().Notify(ctx, method, params)
if errors.Is(err, jsonrpc2.ErrClientClosing) || errors.Is(err, jsonrpc2.ErrServerClosing) {
if errors.Is(err, jsonrpc2.ErrConnClosing) {
if cause := withoutClosingErrors(err); cause != nil {
err = fmt.Errorf("%w: %w", ErrConnectionClosed, cause)
} else {
Expand Down
10 changes: 5 additions & 5 deletions mcp/transport.go
Original file line number Diff line number Diff line change
Expand Up @@ -313,7 +313,7 @@ func call(ctx context.Context, conn *jsonrpc2.Connection, method string, params
// is nothing to cancel, and the spec permits notifications/cancelled
// only for requests believed to still be in progress.
return nil
case errors.Is(err, jsonrpc2.ErrClientClosing), errors.Is(err, jsonrpc2.ErrServerClosing):
case errors.Is(err, jsonrpc2.ErrConnClosing):
if cause := withoutClosingErrors(err); cause != nil {
return fmt.Errorf("%w: calling %q: %w", ErrConnectionClosed, method, cause)
}
Expand Down Expand Up @@ -347,11 +347,11 @@ func call(ctx context.Context, conn *jsonrpc2.Connection, method string, params
return nil
}

// withoutClosingErrors returns err with jsonrpc2.ErrClientClosing and
// jsonrpc2.ErrServerClosing removed from its tree, or nil if nothing else is
// left. Those errors are internal: wrapping them would expose their codes.
// withoutClosingErrors returns err with jsonrpc2.ErrConnClosing removed from
// its tree, or nil if nothing else is left. That error is internal: wrapping it
// would expose its code.
func withoutClosingErrors(err error) error {
if !errors.Is(err, jsonrpc2.ErrClientClosing) && !errors.Is(err, jsonrpc2.ErrServerClosing) {
if !errors.Is(err, jsonrpc2.ErrConnClosing) {
return err
}
switch err := err.(type) {
Expand Down
19 changes: 9 additions & 10 deletions mcp/transport_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -110,23 +110,24 @@ type connectionClosedTest struct {
func connectionClosedTests() []connectionClosedTest {
readErr := errors.New("read failed")
wrappedReadErr := fmt.Errorf("reading: %w", readErr)
sameMessageErr := errors.New("client is closing")
sameMessageErr := errors.New("connection is closing")
return []connectionClosedTest{
{name: "local close", close: true},
{name: "EOF", readErr: io.EOF, causes: []error{io.EOF}},
{name: "unexpected EOF", readErr: io.ErrUnexpectedEOF, causes: []error{io.ErrUnexpectedEOF}},
{name: "read error", readErr: readErr, causes: []error{readErr}},
{name: "wrapped read error", readErr: wrappedReadErr, causes: []error{wrappedReadErr, readErr}},
{name: "client closing", readErr: jsonrpc2.NewError(-32003, "client is closing")},
{name: "server closing", readErr: jsonrpc2.NewError(-32004, "server is closing")},
{name: "closing", readErr: jsonrpc2.NewError(-32004, "connection is closing")},
// Older SDK versions send the same code with a different message.
{name: "closing, older message", readErr: jsonrpc2.NewError(-32004, "server is closing")},
{
name: "wrapped closing error",
readErr: fmt.Errorf("outer: %w", fmt.Errorf("%w: %w", jsonrpc2.ErrServerClosing, wrappedReadErr)),
readErr: fmt.Errorf("outer: %w", fmt.Errorf("%w: %w", jsonrpc2.ErrConnClosing, wrappedReadErr)),
causes: []error{wrappedReadErr, readErr},
},
{
name: "joined closing errors",
readErr: errors.Join(jsonrpc2.ErrClientClosing, errors.Join(jsonrpc2.ErrServerClosing, readErr), io.ErrUnexpectedEOF),
readErr: errors.Join(jsonrpc2.ErrConnClosing, errors.Join(jsonrpc2.ErrConnClosing, readErr), io.ErrUnexpectedEOF),
causes: []error{readErr, io.ErrUnexpectedEOF},
},
{name: "wire error cause", readErr: jsonrpc2.ErrInvalidParams, causes: []error{jsonrpc2.ErrInvalidParams}},
Expand Down Expand Up @@ -181,13 +182,11 @@ func checkConnectionClosedError(t *testing.T, err error, causes ...error) {
t.Errorf("error = %v, want an error wrapping %v", err, want)
}
}
for _, closing := range []error{jsonrpc2.ErrClientClosing, jsonrpc2.ErrServerClosing} {
if errors.Is(err, closing) {
t.Errorf("error = %v, must not wrap %v", err, closing)
}
if errors.Is(err, jsonrpc2.ErrConnClosing) {
t.Errorf("error = %v, must not wrap %v", err, jsonrpc2.ErrConnClosing)
}
var wireErr *jsonrpc2.WireError
if errors.As(err, &wireErr) && (wireErr.Code == -32003 || wireErr.Code == -32004) {
if errors.As(err, &wireErr) && wireErr.Code == -32004 {
t.Errorf("error = %v, exposes internal closing code %d", err, wireErr.Code)
}
}
Expand Down
Loading