This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new b88aedc34 fix(rmqctl): reconnect on session termination even when the
id was cleared (#4876)
b88aedc34 is described below
commit b88aedc34949e2da81f9fb862572b8f879627a13
Author: Apulupie <[email protected]>
AuthorDate: Mon Oct 19 18:14:16 2026 +0800
fix(rmqctl): reconnect on session termination even when the id was cleared
(#4876)
Verified against `rocketmq-studio` at 86aa49c34 before merging: 缺陷仍在。主干
`rmqctl/internal/studio/mcp_message.go:196` 仍是 `return generation,
session.transport.GetSessionId() != ""`，`:177` 用它决定是否重连；`MB..trunk` 对 `rmqc
merge-tree clean against the current trunk and no outstanding must-fix
items; merged as-is (squash) per the maintainer review of 2026-10-19.
---
rmqctl/internal/studio/mcp_message.go | 17 +--
rmqctl/internal/studio/mcp_message_test.go | 168 +++++++++++++++++++++++++++++
2 files changed, 179 insertions(+), 6 deletions(-)
diff --git a/rmqctl/internal/studio/mcp_message.go
b/rmqctl/internal/studio/mcp_message.go
index 7c4d7a6b7..41f3d2360 100644
--- a/rmqctl/internal/studio/mcp_message.go
+++ b/rmqctl/internal/studio/mcp_message.go
@@ -161,7 +161,7 @@ func (session *MCPClientSession) sendWithReconnect(ctx
context.Context, skipReco
if err := session.sendGate.acquire(ctx, session.closed, false);
err != nil {
return err
}
- generation, hadSession := session.sendSnapshot()
+ generation, canReconnect := session.sendSnapshot()
session.state.RLock()
pending := session.state.reconnectPending
session.state.RUnlock()
@@ -174,7 +174,7 @@ func (session *MCPClientSession) sendWithReconnect(ctx
context.Context, skipReco
if err == nil || skipReconnect || attempt == 1 {
return err
}
- if !pending && !(hadSession && errors.Is(err,
mcptransport.ErrSessionTerminated)) {
+ if !pending && !(canReconnect && errors.Is(err,
mcptransport.ErrSessionTerminated)) {
return err
}
if reconnectErr := session.reinitialize(ctx, generation);
reconnectErr != nil {
@@ -186,14 +186,19 @@ func (session *MCPClientSession) sendWithReconnect(ctx
context.Context, skipReco
return err
}
-// sendSnapshot returns the current session generation and whether a session
-// id is already established. Callers must hold a sendGate permit so the
-// snapshot is consistent with the transport state used for the send.
+// sendSnapshot returns the current session generation and whether a terminated
+// session can be re-initialized, which is possible whenever the initialize
+// request has been recorded. The transport's session id cannot be used for
+// this decision: mcp-go clears it as soon as any request fails with a 404, so
+// a sender racing with that failure would wrongly see "no session" and skip
+// the reconnect. Callers must hold a sendGate permit so the snapshot is
+// consistent with the transport state used for the send.
func (session *MCPClientSession) sendSnapshot() (uint64, bool) {
session.state.RLock()
generation := session.state.generation
+ initialize := session.state.initialize
session.state.RUnlock()
- return generation, session.transport.GetSessionId() != ""
+ return generation, initialize != nil
}
func (session *MCPClientSession) recordInitialization(
diff --git a/rmqctl/internal/studio/mcp_message_test.go
b/rmqctl/internal/studio/mcp_message_test.go
index aee3b0542..06fddf724 100644
--- a/rmqctl/internal/studio/mcp_message_test.go
+++ b/rmqctl/internal/studio/mcp_message_test.go
@@ -17,11 +17,179 @@
package studio
import (
+ "context"
"encoding/json"
"strings"
+ "sync"
"testing"
+
+ mcptransport "github.com/mark3labs/mcp-go/client/transport"
+ "github.com/mark3labs/mcp-go/mcp"
)
+// reconnectStubTransport answers requests the way a Studio server that lost a
+// session does: every non-initialize request fails with a 404 until a fresh
+// initialize exchange establishes "session-2". Like mcp-go's StreamableHTTP
+// client, it clears the stored session id when a request fails with a 404.
+type reconnectStubTransport struct {
+ mu sync.Mutex
+ sessionID string
+ sends int
+ initializes int
+}
+
+func newReconnectStubTransport(sessionID string) *reconnectStubTransport {
+ return &reconnectStubTransport{sessionID: sessionID}
+}
+
+func (transport *reconnectStubTransport) Start(context.Context) error { return
nil }
+func (transport *reconnectStubTransport) Close() error { return
nil }
+
+func (transport *reconnectStubTransport) SetNotificationHandler(
+ func(mcp.JSONRPCNotification),
+) {
+}
+
+func (transport *reconnectStubTransport) SetProtocolVersion(string) {}
+
+func (transport *reconnectStubTransport) GetSessionId() string {
+ transport.mu.Lock()
+ defer transport.mu.Unlock()
+ return transport.sessionID
+}
+
+func (transport *reconnectStubTransport) SendRequest(
+ _ context.Context,
+ request mcptransport.JSONRPCRequest,
+) (*mcptransport.JSONRPCResponse, error) {
+ transport.mu.Lock()
+ if request.Method == string(mcp.MethodInitialize) {
+ transport.initializes++
+ transport.sessionID = "session-2"
+ transport.mu.Unlock()
+ return &mcptransport.JSONRPCResponse{
+ JSONRPC: mcp.JSONRPC_VERSION,
+ ID: request.ID,
+ Result: json.RawMessage(`{"protocolVersion":"` +
mcp.LATEST_PROTOCOL_VERSION + `"}`),
+ }, nil
+ }
+ transport.sends++
+ established := transport.sessionID == "session-2"
+ used := transport.sessionID
+ transport.mu.Unlock()
+ if !established {
+ // mcp-go clears the session id on a 404, but only when it
still holds
+ // the value the failed request was sent with (CompareAndSwap).
+ transport.mu.Lock()
+ if transport.sessionID == used {
+ transport.sessionID = ""
+ }
+ transport.mu.Unlock()
+ return nil, mcptransport.ErrSessionTerminated
+ }
+ return &mcptransport.JSONRPCResponse{
+ JSONRPC: mcp.JSONRPC_VERSION,
+ ID: request.ID,
+ Result: json.RawMessage(`{}`),
+ }, nil
+}
+
+func (transport *reconnectStubTransport) SendNotification(
+ context.Context,
+ mcp.JSONRPCNotification,
+) error {
+ return nil
+}
+
+func (transport *reconnectStubTransport) initializeRequest()
*mcptransport.JSONRPCRequest {
+ return &mcptransport.JSONRPCRequest{
+ JSONRPC: mcp.JSONRPC_VERSION,
+ ID: mcp.NewRequestId("initialize-1"),
+ Method: string(mcp.MethodInitialize),
+ }
+}
+
+func newReconnectTestSession(transport *reconnectStubTransport)
*MCPClientSession {
+ session := &MCPClientSession{
+ transport: transport,
+ closed: make(chan struct{}),
+ notifications: make(chan json.RawMessage,
notificationBufferSize),
+ }
+ session.state.initialize = transport.initializeRequest()
+ return session
+}
+
+func sendToolsList(session *MCPClientSession) error {
+ return session.sendWithReconnect(context.Background(), false, func()
error {
+ _, err := session.transport.SendRequest(context.Background(),
mcptransport.JSONRPCRequest{
+ JSONRPC: mcp.JSONRPC_VERSION,
+ ID: mcp.NewRequestId("send-1"),
+ Method: string(mcp.MethodToolsList),
+ })
+ return err
+ })
+}
+
+// TestSendWithReconnectRetriesAfterConcurrentSessionClear reproduces the
+// concurrent-404 race: mcp-go clears the transport session id as soon as any
+// request fails with a 404, so another in-flight sender can observe an empty
+// session id in its snapshot even though a session had been established and
+// recorded. Reconnect-worthiness must therefore come from the recorded
+// initialize request, not from the transport's session id.
+func TestSendWithReconnectRetriesAfterConcurrentSessionClear(t *testing.T) {
+ transport := newReconnectStubTransport("")
+ session := newReconnectTestSession(transport)
+
+ // A concurrent sender just failed with a 404, which cleared the
transport
+ // session id before that sender managed to reinitialize.
+ if err := sendToolsList(session); err != nil {
+ t.Fatalf("sendWithReconnect after a concurrent 404 cleared the
session: %v", err)
+ }
+ transport.mu.Lock()
+ initializes := transport.initializes
+ transport.mu.Unlock()
+ if initializes != 1 {
+ t.Fatalf("initialize exchanges = %d, want 1", initializes)
+ }
+ if transport.GetSessionId() != "session-2" {
+ t.Fatalf("session id = %q, want session-2",
transport.GetSessionId())
+ }
+}
+
+// TestSendWithReconnectConcurrent404sAllRecover verifies that concurrent
+// senders hitting a terminated session all recover once one of them
+// reinitializes, without duplicating the initialize exchange per sender.
+func TestSendWithReconnectConcurrent404sAllRecover(t *testing.T) {
+ transport := newReconnectStubTransport("session-1")
+ session := newReconnectTestSession(transport)
+
+ const senders = 8
+ var wg sync.WaitGroup
+ errs := make([]error, senders)
+ for i := range errs {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ errs[i] = sendToolsList(session)
+ }()
+ }
+ wg.Wait()
+ for i, err := range errs {
+ if err != nil {
+ t.Fatalf("sender %d: %v", i, err)
+ }
+ }
+ transport.mu.Lock()
+ initializes := transport.initializes
+ transport.mu.Unlock()
+ // At least one initialize exchange must have re-established the
session;
+ // a sender whose request raced the establishment may legitimately
repeat
+ // it, but each sender must still recover.
+ if initializes < 1 {
+ t.Fatalf("initialize exchanges = %d, want at least 1",
initializes)
+ }
+}
+
func toolCallPayload(t *testing.T, tool string, arguments map[string]any)
[]byte {
t.Helper()
params := map[string]any{"name": tool}
