一次版本升级导致的血案

- github.com/mark3labs/mcp-go v0.39.1
+ github.com/mark3labs/mcp-go v0.48.0

github.com/mark3labs/mcp-go

mcpSrv := server.NewMCPServer(
	"internal-mcp",
	"1.0.0",
	server.WithToolCapabilities(false),
)

handler := server.NewStreamableHTTPServer(mcpSrv)

http.Handle("/mcp", handler)

- github.com/mark3labs/mcp-go v0.39.1
+ github.com/mark3labs/mcp-go v0.48.0

- github.com/mark3labs/mcp-go v0.39.1
+ github.com/mark3labs/mcp-go v0.48.0

go tool pprof heap.pprof

(pprof) top
(pprof) top -cum
(pprof) list FunctionName

github.com/mark3labs/mcp-go/server.newStreamableHttpSession
github.com/mark3labs/mcp-go/server.(*StreamableHTTPServer).handlePost
github.com/mark3labs/mcp-go/server.(*MCPServer).RegisterSession
sync.(*Map).Store

server/streamable_http.go

StreamableHTTPServer
NewStreamableHTTPServer
handlePost
handleGet
handleDelete
newStreamableHttpSession
cleanupSessionState
WithSessionIdleTTL

// Prepare the session for the mcp server
// The session is ephemeral. Its life is the same as the request. It's only created
// for interaction with the mcp server.

session := newStreamableHttpSession(sessionID, s.sessionTools, s.sessionLogLevels)

// streamableHttpSession is a session for streamable-http transport
// When in POST handlers(request/notification), it's ephemeral, and only exists in the life of the request handler.
// When in GET handlers(listening), it's a real session, and will be registered in the MCP server.

type StreamableHTTPServer struct {
	server *MCPServer

	sessionTools             *sessionToolsStore
	sessionResources         *sessionResourcesStore
	sessionResourceTemplates *sessionResourceTemplatesStore

	sessionRequestIDs sync.Map // sessionId --> last requestID(*atomic.Int64)
	activeSessions    sync.Map // sessionId --> *streamableHttpSession (for sampling responses)

	httpServer *http.Server

	// ...

	sessionIdleTTL    time.Duration
	sessionLastActive sync.Map // sessionID → *atomic.Int64 (unix nanos)
	sweeperCancel    context.CancelFunc
}

sessionRequestIDs sync.Map
activeSessions    sync.Map

activeSessions sync.Map // sessionId --> *streamableHttpSession

原来：请求结束 -> 引用链断开 -> 等待 GC
现在：请求结束 -> activeSessions 仍然引用 -> GC 不能回收

isInitializeRequest := jsonMessage.Method == mcp.MethodInitialize

var sessionID string
sessionIdManager := s.sessionIdManagerResolver.ResolveSessionIdManager(r)

if isInitializeRequest {
	// generate a new one for initialize request
	sessionID = sessionIdManager.Generate()
} else {
	// Get session ID from header.
	// Stateful servers need the client to carry the session ID.
	sessionID = r.Header.Get(HeaderKeySessionID)

	isTerminated, err := sessionIdManager.Validate(sessionID)
	if err != nil {
		http.Error(w, "Invalid session ID", http.StatusNotFound)
		return
	}
	if isTerminated {
		http.Error(w, "Session terminated", http.StatusNotFound)
		return
	}
}

// For non-initialize requests, try to reuse existing registered session
var session *streamableHttpSession
if !isInitializeRequest {
	if sessionValue, ok := s.server.sessions.Load(sessionID); ok {
		if existingSession, ok := sessionValue.(*streamableHttpSession); ok {
			session = existingSession
		}
	}
}

// Check if a persistent session exists (for sampling support), otherwise create ephemeral session
// Persistent sessions are created by GET (continuous listening) connections
if session == nil {
	if sessionInterface, exists := s.activeSessions.Load(sessionID); exists {
		if persistentSession, ok := sessionInterface.(*streamableHttpSession); ok {
			session = persistentSession
		}
	}
}

// Create ephemeral session if no persistent session exists
if session == nil {
	session = newStreamableHttpSession(
		sessionID,
		s.sessionTools,
		s.sessionResources,
		s.sessionResourceTemplates,
		s.sessionLogLevels,
	)
}

// Register session after successful initialization
// Only register if not already registered (e.g., by a GET connection)
if isInitializeRequest && sessionID != "" {
	if _, exists := s.server.sessions.Load(sessionID); !exists {
		// Store in activeSessions to prevent duplicate registration from GET
		s.activeSessions.Store(sessionID, session)

		// Register the session with the MCPServer for notification support
		if err := s.server.RegisterSession(ctx, session); err != nil {
			s.logger.Errorf("Failed to register POST session: %v", err)
			s.activeSessions.Delete(sessionID)
			// Don't fail the request, just log the error
		}
	}
}

StreamableHTTPServer
    └── activeSessions sync.Map
            └── sessionID -> *streamableHttpSession
                                ├── notificationChannel
                                ├── samplingRequestChan
                                ├── elicitationRequestChan
                                ├── rootsRequestChan
                                ├── samplingRequests sync.Map
                                ├── tools store
                                ├── resources store
                                ├── resource templates store
                                └── log levels store

root
  -> http server
  -> StreamableHTTPServer
  -> activeSessions
  -> streamableHttpSession

func (s *StreamableHTTPServer) handleDelete(w http.ResponseWriter, r *http.Request) {
	// delete request terminate the session
	sessionID := r.Header.Get(HeaderKeySessionID)

	sessionIdManager := s.sessionIdManagerResolver.ResolveSessionIdManager(r)
	notAllowed, err := sessionIdManager.Terminate(sessionID)
	if err != nil {
		http.Error(w, fmt.Sprintf("Session termination failed: %v", err), http.StatusInternalServerError)
		return
	}
	if notAllowed {
		http.Error(w, "Session termination not allowed", http.StatusMethodNotAllowed)
		return
	}

	s.cleanupSessionState(r.Context(), sessionID)

	w.WriteHeader(http.StatusOK)
}

s.cleanupSessionState(r.Context(), sessionID)

func (s *StreamableHTTPServer) cleanupSessionState(ctx context.Context, sessionID string) {
	// Unregister first to stop notification routing before deleting data.
	s.server.UnregisterSession(ctx, sessionID)

	s.activeSessions.Delete(sessionID)
	s.sessionTools.delete(sessionID)
	s.sessionResources.delete(sessionID)
	s.sessionResourceTemplates.delete(sessionID)
	s.sessionLogLevels.delete(sessionID)
	s.sessionRequestIDs.Delete(sessionID)
	s.sessionLastActive.Delete(sessionID)
}

server.sessions
activeSessions
sessionTools
sessionResources
sessionResourceTemplates
sessionLogLevels
sessionRequestIDs
sessionLastActive

session 数量增长
    -> activeSessions 增长
    -> server.sessions 增长
    -> sessionRequestIDs 增长
    -> sessionTools/sessionResources 等 store 增长
    -> heap 持续上涨

func WithSessionIdleTTL(ttl time.Duration) StreamableHTTPOption {
	return func(s *StreamableHTTPServer) {
		s.sessionIdleTTL = ttl
	}
}

// WithSessionIdleTTL sets the idle TTL for per-session transport state.
//
// When enabled, a background sweeper periodically removes entries from
// per-session stores (tools, resources, resource templates, log levels,
// request IDs) for sessions that have been idle longer than the given
// duration. This prevents memory leaks when clients disconnect without
// sending a DELETE request. A zero or negative value disables the sweeper
// (the default).

This prevents memory leaks when clients disconnect without sending a DELETE request.
A zero or negative value disables the sweeper (the default).

handler := server.NewStreamableHTTPServer(mcpSrv)

session 会被持久化保存
+
客户端不一定发送 DELETE
+
服务端没有开启 idle TTL sweeper
=
session 无界增长

if s.sessionIdleTTL > 0 {
	ctx, cancel := context.WithCancel(context.Background())
	s.sweeperCancel = cancel
	s.startSessionSweeper(ctx)
}

func (s *StreamableHTTPServer) startSessionSweeper(ctx context.Context) {
	interval := max(s.sessionIdleTTL/2, time.Second)

	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()

		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				s.sweepExpiredSessions()
			}
		}
	}()
}

s.sweepExpiredSessions()

func (s *StreamableHTTPServer) sweepExpiredSessions() {
	now := time.Now().UnixNano()
	ttlNanos := s.sessionIdleTTL.Nanoseconds()

	s.sessionLastActive.Range(func(key, value any) bool {
		sessionID, ok := key.(string)
		if !ok {
			s.sessionLastActive.Delete(key)
			return true
		}

		lastActive, ok := value.(*atomic.Int64)
		if !ok {
			s.sessionLastActive.Delete(key)
			return true
		}

		capturedLastActive := lastActive.Load()
		if now-capturedLastActive < ttlNanos {
			return true
		}

		// Re-check: if lastActive changed since we read it, the session
		// was touched concurrently — skip it.
		if lastActive.Load() != capturedLastActive {
			return true
		}

		s.logger.Infof("Sweeping expired session: %s", sessionID)

		mgr := s.sessionIdManager
		if mgr == nil {
			mgr = s.sessionIdManagerResolver.ResolveSessionIdManager(nil)
		}
		_, _ = mgr.Terminate(sessionID)

		s.cleanupSessionState(context.Background(), sessionID)

		return true
	})
}

func (s *StreamableHTTPServer) touchSession(sessionID string) {
	if sessionID == "" || s.sessionIdleTTL <= 0 {
		return
	}

	now := time.Now().UnixNano()
	actual, _ := s.sessionLastActive.LoadOrStore(sessionID, new(atomic.Int64))
	actual.(*atomic.Int64).Store(now)
}

s.cleanupSessionState(context.Background(), sessionID)

1. 升级 mcp-go：v0.39.1 -> v0.48.0
        ↓
2. StreamableHTTPServer 新增 activeSessions / sessionRequestIDs / sessionLastActive 等服务级状态
        ↓
3. initialize POST 成功后，session 被 Store 到 activeSessions，并 Register 到 MCPServer
        ↓
4. session 生命周期从“请求级”变成“服务级”
        ↓
5. 客户端没有稳定发送 HTTP DELETE
        ↓
6. 服务端没有配置 WithSessionIdleTTL
        ↓
7. session 相关状态无法释放
        ↓
8. Go GC 认为对象仍然可达
        ↓
9. heap 持续上涨
        ↓
10. OOM

handler := server.NewStreamableHTTPServer(
	mcpSrv,
	server.WithSessionIdleTTL(30*time.Minute),
)

handler := server.NewStreamableHTTPServer(
	mcpSrv,
	server.WithSessionIdleTTL(5*time.Minute),
)

func closeMCPSession(ctx context.Context, endpoint string, sessionID string) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodDelete, endpoint, nil)
	if err != nil {
		return err
	}

	req.Header.Set("Mcp-Session-Id", sessionID)

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 300 {
		return fmt.Errorf("close mcp session failed: status=%d", resp.StatusCode)
	}

	return nil
}

handler := server.NewStreamableHTTPServer(
	mcpSrv,
	server.WithStateLess(true),
)

mu.Lock()
sessions[id] = session
mu.Unlock()

s.activeSessions.Store(sessionID, session)

s.activeSessions.Delete(sessionID)

sync.(*Map).Store
xxx.newSession
xxx.RegisterSession

1. 是否新增全局状态？
2. 是否新增服务级 map/cache？
3. 是否新增后台 goroutine？
4. 是否新增连接池？
5. 是否新增 session 管理？
6. 是否新增 Close/Delete/Shutdown/TTL 之类的选项？
7. 默认值是否发生变化？
8. 请求级对象是否变成服务级对象？
9. 是否有异常路径清理问题？
10. 是否需要补充压测和 soak test？

handler := server.NewStreamableHTTPServer(
	mcpSrv,
	server.WithSessionIdleTTL(10*time.Minute),
)