Streamable HTTP: close replayed request streams once no in-flight requests remain

[felixweinberger]

When a client resumes a request stream with Last-Event-ID, the replayed stream stays open indefinitely (keep-alive now keeps it healthy) even when every correlated response has already been delivered or stored.

Three narrow races follow from this:

• A response that becomes store-visible while a resume is mid-flight is missed by that replay's snapshot and never delivered to the now-permanent successor.
• A superseded successor's dedup state does not transfer, so a double reconnect during an in-flight store write can double-deliver a response.
• A stream resumed after closeSSEStream polling delivers the replayed response but never ends.

Closing a replayed request stream after replay whenever no in-flight requests map to it turns the polling loop into one that converges: the client re-polls until a replay snapshot includes the response.

Applies to both main and v1.x (the send()/replay code has the same shape in both). Split out from the review discussion on #2547.

[felixweinberger]

Widening this issue based on five review rounds on #2547: the individual races reported there (and fixed there) plus the ones below all share one generator, and this issue is the right place to retire it.

The generator: request/response delivery state lives in four separately-mutated maps (_streamMapping, _requestToStreamMapping, _requestResponseMap, _resumableStreams) plus a per-mapping dedup set, updated from five call sites (send(), handlePostRequest, replayEvents, stream cancel callbacks, close()) across six suspension points. Every invariant ("correlations exist iff a response is pending", "an entry belongs to the live stream", "closed means nothing mutates") is enforced by convention at each site, so each close()/cancel/resume/store-failure interleaving needs its own conditional, and each added conditional creates new interleavings.

Additional scenarios in the class (from #2547 review round 5, deliberately not patched there):

• A JSON-RPC batch where one response was recorded before close() and another send was parked on storeEvent: the swept _requestResponseMap makes allResponsesReady permanently false, so no completion arm ever runs.
• close() racing writePrimingEvent: the swept stream's closed controller makes the continuation's enqueue throw, producing a 400 "Parse error" plus a spurious post-close onerror instead of a 404.
• The response-path storeEvent rejecting: send() rejects leaving the stream mapping, armed keep-alive timer, correlations, and resumability entry in place.

Proposed shape: one Exchange object per request id owning {streamId, response?, resumable, state: pending | delivered | released | failed}, a single map of them, and all transitions through one method that checks closed/cancelled once. close() settles every exchange; a stream closes when its exchanges settle (which is also this issue's original ask); event-store writes go through one wrapper with a failure policy. The keep-alive work on #2547 killed its timer class exactly this way (ownership instead of a shared registry), and no timer finding has recurred in five review rounds since.

Applies to both lines; main first, then backport.