mcp: add race and 1.25 tests, fix a bug in streamable transport

.github/workflows/test.yml

@@ -1,22 +1,22 @@
 name: Test
 on:
   # Manual trigger
-  workflow_dispatch:  
+  workflow_dispatch:
   push:
     branches: main
   pull_request:
-  
+
 permissions:
   contents: read
 
 jobs:
   lint:
     runs-on: ubuntu-latest
     steps:
-    - name: Set up Go
-      uses: actions/setup-go@v5
     - name: Check out code
       uses: actions/checkout@v4
+    - name: Set up Go
+      uses: actions/setup-go@v5
     - name: Check formatting
       run: |
         unformatted=$(gofmt -l .)
@@ -26,18 +26,30 @@ jobs:
           exit 1
         fi
         echo "All Go files are properly formatted"
-  
+
   test:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        go: [ '1.23', '1.24' ]
+        go: [ '1.23', '1.24', '1.25.0-rc.2' ]
     steps:
+    - name: Check out code
+      uses: actions/checkout@v4
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
         go-version: ${{ matrix.go }}
-    - name: Check out code
-      uses: actions/checkout@v4
     - name: Test
       run: go test -v ./...
+
+  race-test:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Check out code
+      uses: actions/checkout@v4
+    - name: Set up Go
+      uses: actions/setup-go@v5
+      with:
+        go-version: '1.24'
+    - name: Test with -race
+      run: go test -v -race ./...

mcp/streamable.go

@@ -267,9 +267,7 @@ type stream struct {
 
 	// streamRequests is the set of unanswered incoming RPCs for the stream.
 	//
-	// Lifecycle: requests values persist until the requests have been
-	// replied to by the server. Notably, NOT until they are sent to an HTTP
-	// response, as delivery is not guaranteed.
+	// Requests persist until their response data has been added to outgoing.
 	requests map[jsonrpc.ID]struct{}
 }
 
@@ -482,6 +480,7 @@ stream:
 		t.mu.Lock()
 		outgoing := stream.outgoing
 		stream.outgoing = nil
+		nOutstanding := len(stream.requests)
 		t.mu.Unlock()
 
 		for _, data := range outgoing {
@@ -493,9 +492,6 @@ stream:
 			}
 		}
 
-		t.mu.Lock()
-		nOutstanding := len(stream.requests)
-		t.mu.Unlock()
 		// If all requests have been handled and replied to, we should terminate this connection.
 		// "After the JSON-RPC response has been sent, the server SHOULD close the SSE stream."
 		// §6.4, https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#sending-messages-to-the-server