fix(streamable-client): make Close() synchronously terminate GET SSE - #103
Open
bytethm wants to merge 1 commit into
Open
fix(streamable-client): make Close() synchronously terminate GET SSE#103bytethm wants to merge 1 commit into
bytethm wants to merge 1 commit into
Conversation
Client.Close() previously returned as soon as it had canceled the GET
SSE context, without waiting for the background SSE goroutine to exit or
closing the in-flight HTTP response body. The only way for the blocking
bufio.Scanner.Scan() -> resp.Body.Read() to return was for the ctx
cancellation to propagate through http.Transport and tear down the TCP
connection. On CI this propagation could stall long enough that
httptest.Server.Close() (and real-world server shutdown paths) hung
waiting for the handler it held open, surfacing upstream as the 10-minute
timeout worked around in trpc-agent-go#1721.
This change turns close() into a synchronous contract:
* Hold resp.Body in getSSEConn.body and forcefully Close() it, so the
blocking read wakes up immediately instead of waiting on
http.Transport cancellation traversal.
* Track the SSE goroutine with a sync.WaitGroup and Wait() for it at
the end of close(). When close() returns, the goroutine is gone.
* Guard establishGetSSE with a 'closed' flag so a late spawn coming
from Initialize's 'go establishGetSSEConnection(ctx)' cannot race
past sseWg.Wait() and leak a goroutine.
The same pattern was applied to sse_client.go in trpc-group#93; this commit
extends it to the Streamable HTTP transport.
Protocol compliance: spec 2025-11-25 §Transports explicitly allows
'The client MAY close the SSE stream at any time', and §Lifecycle
states 'For HTTP transports, shutdown is indicated by closing the
associated HTTP connection(s)'. This fix does exactly that, without
altering TerminateSession() (HTTP DELETE) or Last-Event-ID semantics.
Regression tests cover:
* Client.Close() returning leaves no handleGetSSEEvents goroutine.
* Close() returns in sub-ms even when the server produces no SSE
traffic for tens of seconds (proving resp.Body.Close drives the
wake-up, not ctx propagation).
* 50 tight Initialize/Close cycles do not leak SSE goroutines
(covers the late-spawn race fixed by the 'closed' guard).
End-to-end verified against trpc-agent-go tool/mcpbroker: with this
fix in place, removing the httptest.Server.CloseClientConnections()
workaround from broker_test.go and running -count=20 passes cleanly
(201s, no hangs).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Client.Close()returned before the GET SSE goroutine actually exited: itonly cancelled the ctx, leaving
scanner.Scan()blocked until the cancelpropagated through
http.Transport. Under CI load this window stalleddownstream server shutdown — the symptom behind trpc-agent-go#1721.
Fix: close
resp.Bodydirectly,sync.WaitGroup-join the SSE goroutine,and guard
establishGetSSEagainst late spawns fromInitialize.Same pattern as #93, extended to the streamable HTTP transport.
Regression tests in
client_test.go(TestClient_Close_*).