Set a total retry limit on streams, or enforce progress

[findleyr]

Right now, the streamable client connection retries broken resumable connections up to StreamableClientTransport.MaxRetries, but once a connection has been re-established, the counter resets.

This can lead to a failure mode where a server keeps terminating the connection with status 200. We should either:

1. Set a global limit on the number of retries across a single logical stream.
2. Only reset the retry counter when the Last-Event-ID actually makes progress.

I'm leaning toward 2: if the stream is active and long-running, a global limit could be problematic. On the other hand, (2) doesn't have that problem, and still avoids the degenerate failure mode.

CC @jba @markus-kusano

[findleyr]

@markus-kusano this one might be an interesting and smallish bug to pick up. LMK if you're interested.

[markus-kusano]

I agree with your analysis, (2) seems to be the way to go.

Here's my thoughts:

Using (1), global limit, has the issue that the limit needs to be reset occasionally for long running connections. This suggests at a high-level we need a form of short-term quota, i.e., i.e., the connection fails if it continuously fails within a short time window, and we reset the quota outside that time window. Using (2) to monitor Last-Event-ID give us that short-term quota property without introducing new variables to track it (the event ID updates if the stream makes progress, IIUC)

[findleyr]

This could be a good starter bug (also, an agent could probably knock this out pretty easily).

[jba]

Agreed, 2 is better.

[rioloc]

I would like to help with this, feel free to assign it to me if it hasn’t been assigned yet.

[rioloc]

Hello, I have some doubts about the desired implementation. Especially, a first approach would be

 func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary string, resp *http.Response, forCall *jsonrpc2.Request) {
+       var currentLastEventID string
+       retryCount := 0
        for {
                // Connection was successful. Continue the loop with the new response.
                //
-               // TODO(#679): we should set a reasonable limit on the number of times
-               // we'll try getting a response for a given request, or enforce that we
-               // actually make progress.
-               //
                // Eventually, if we don't get the response, we should stop trying and
                // fail the request.
                lastEventID, reconnectDelay, clientClosed := c.processStream(ctx, requestSummary, resp, forCall)

+               if currentLastEventID != "" && lastEventID != currentLastEventID {
+                       // stream is not progressing even after retries
+                       if retryCount == c.maxRetries {
+                               // stream didn't progress within the retry limit
+                               c.fail(fmt.Errorf("%s: stream failed to progress within the retry limit (session ID: %v)", requestSummary, c.sessionID))
+                               return
+                       }
+
+                       // increase stream processing counter
+                       retryCount++
+               }
+               currentLastEventID = lastEventID
+
                // If the connection was closed by the client, we're done.
                if clientClosed {
                        return

Should c.maxRetries be used for both connection establishment and this case? At the moment it's only used in

go-sdk/mcp/streamable.go

Line 1952 in 76e6854

for ; attempt <= c.maxRetries; attempt++ {

If c.maxRetries is used as retry counter also for tracking a stream which is not progressing, in the worst case there will be c.maxRetries * c.maxRetries retries.

Idk if this is the kind of approach you were thinking about, or I missed something. Still new to the codebase so I may be easily on the wrong path.

Last question I have is about testing, is there already something in place to mock event ids from StreamableHTTPHandler ?