handleSSE manages the lifecycle of an SSE connection. It can be either persistent (for the main GET listener) or temporary (for a POST response). If forCall is set, it is the call that initiated the stream, and the stream is complete when we receive its response. Otherwise, this is the standalone s
(ctx context.Context, requestSummary string, resp *http.Response, forCall *jsonrpc2.Request)
| 1983 | // stream is complete when we receive its response. Otherwise, this is the |
| 1984 | // standalone stream. |
| 1985 | func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary string, resp *http.Response, forCall *jsonrpc2.Request) { |
| 1986 | // Track the last event ID to detect progress. |
| 1987 | // The retry counter is only reset when progress is made (lastEventID advances). |
| 1988 | // This prevents infinite retry loops when a server repeatedly terminates |
| 1989 | // connections without making progress (#679). |
| 1990 | var prevLastEventID string |
| 1991 | retriesWithoutProgress := 0 |
| 1992 | |
| 1993 | for { |
| 1994 | lastEventID, reconnectDelay, clientClosed := c.processStream(ctx, requestSummary, resp, forCall) |
| 1995 | |
| 1996 | // If the connection was closed by the client, we're done. |
| 1997 | if clientClosed { |
| 1998 | return |
| 1999 | } |
| 2000 | // If we don't have a last event ID, we can never get the call response, so |
| 2001 | // there's nothing to resume. For the standalone stream, we can reconnect, |
| 2002 | // but we may just miss messages. |
| 2003 | if lastEventID == "" && forCall != nil { |
| 2004 | return |
| 2005 | } |
| 2006 | |
| 2007 | // Check if we made progress (lastEventID advanced). |
| 2008 | // Only reset the retry counter when actual progress is made. |
| 2009 | if lastEventID != "" && lastEventID != prevLastEventID { |
| 2010 | // Progress was made: reset the retry counter. |
| 2011 | retriesWithoutProgress = 0 |
| 2012 | prevLastEventID = lastEventID |
| 2013 | } else { |
| 2014 | // No progress: increment the retry counter. |
| 2015 | retriesWithoutProgress++ |
| 2016 | if retriesWithoutProgress > c.maxRetries { |
| 2017 | if ctx.Err() == nil { |
| 2018 | c.fail(fmt.Errorf("%s: exceeded %d retries without progress (session ID: %v)", requestSummary, c.maxRetries, c.sessionID)) |
| 2019 | } |
| 2020 | return |
| 2021 | } |
| 2022 | } |
| 2023 | |
| 2024 | // The stream was interrupted or ended by the server. Attempt to reconnect. |
| 2025 | newResp, err := c.connectSSE(ctx, lastEventID, reconnectDelay, false) |
| 2026 | if err != nil { |
| 2027 | // If the client didn't cancel this request, any failure to execute it |
| 2028 | // breaks the logical MCP session. |
| 2029 | if ctx.Err() == nil { |
| 2030 | // All reconnection attempts failed: fail the connection. |
| 2031 | c.fail(fmt.Errorf("%s: failed to reconnect (session ID: %v): %v", requestSummary, c.sessionID, err)) |
| 2032 | } |
| 2033 | return |
| 2034 | } |
| 2035 | |
| 2036 | resp = newResp |
| 2037 | if err := c.checkResponse(requestSummary, resp); err != nil { |
| 2038 | c.fail(err) |
| 2039 | return |
| 2040 | } |
| 2041 | } |
| 2042 | } |
no test coverage detected