Streamable HTTP resumability + redelivery + SSE polling via server-side disconnect #1077
Conversation
| /// For other transports, this property will be <see langword="null"/>. | ||
| /// </para> | ||
| /// </remarks> | ||
| public Action? CloseStandaloneSseStream { get; set; } |
There was a problem hiding this comment.
Are the changes to the JsonRpcMessageContext necessary? I think I see the purpose. You cannot always rely on the request ID from the POST body to identify the final message in the SSE stream, but I wonder if there isn't a cleaner abstraction. Maybe disposing the RelatedTransport? That wouldn't cover the GET stream, but I don't know why you need to be able to close the GET stream via a random JsonRpcMessage at all.
| public void CloseStandaloneSseStream() | ||
| { | ||
| JsonRpcRequest.Context?.CloseStandaloneSseStream?.Invoke(); | ||
| } |
There was a problem hiding this comment.
I think this can probably be reverted for the same reasons as the JsonRpcMessageContext changes
| } | ||
|
|
||
| // If this is a POST stream, we're done - the replay was the complete response | ||
| if (streamId != GetStreamId) |
There was a problem hiding this comment.
Shouldn't there be similar changes to HandlePostRequestAsync? Also, we need to do more to avoid stream ID collisions. Even though there's one "get" per session, IEventStore is supplied via HttpServerTransportOptions.EventStore which makes it quite difficult to configure a per-session store.
I don't think people want to be forced to configure the store per-session anyway. I think we need to include the session ID or maybe a random GUID for "Stateless" requests if we support stateless resumption at all.
I think it could make sense to support stateless, but we need to have tests for stateless and non-stateless tests. I recommend looking at the MapMcpTests and MapMcpStreamableHttpTests. By adding test cases to MapMcpStreamableHttpTests, you should get both stateless and non-stateless test coverage.
| /// Priming events are only supported in protocol version >= 2025-11-25. | ||
| /// Older clients may crash when receiving SSE events with empty data. | ||
| /// </remarks> | ||
| internal static bool SupportsResumability(string? protocolVersion) |
There was a problem hiding this comment.
Nit: This should be renamed. It's not about whether the protocol supports resumability, but it's specifically about the "priming" event.
| /// In-memory event store for testing resumability. | ||
| /// This is a simple implementation intended for testing, not for production use. | ||
| /// </summary> | ||
| public class InMemoryEventStore : IEventStore |
There was a problem hiding this comment.
What do you think about the idea of including a default implementation of IEventStore based on IDistributedCache? It'd be opt-in ofc. I could definitely see it being a follow up. Figuring out how to get the usability of that right might influence the design of the abstraction.
| /// Gets or sets the retry interval to suggest to clients in SSE retry field. | ||
| /// </summary> | ||
| /// <value> | ||
| /// The retry interval. The default is <see langword="null"/>, meaning no retry field is sent. |
There was a problem hiding this comment.
The spec states "it SHOULD send an SSE event with a standard retry field". Does that mean our default is going against the spec's recommendation?
| response = await _httpClient.SendAsync(request, message: null, _connectionCts.Token).ConfigureAwait(false); | ||
| } | ||
| catch (HttpRequestException) | ||
| // Continuously receive unsolicited messages until cancelled |
| else | ||
| { | ||
| // We have an event ID, so reconnection should work - reset attempts | ||
| attempt = 0; |
There was a problem hiding this comment.
What prevents us from ending up in an infinite loop?
| /// Implementations should be thread-safe, as events may be stored and replayed concurrently. | ||
| /// </para> | ||
| /// </remarks> | ||
| public interface IEventStore |
There was a problem hiding this comment.
This is a fairly general name. Could we make it ISseEventStore or something similarly more specific?
| /// Stores an event for later retrieval. | ||
| /// </summary> | ||
| /// <param name="streamId"> | ||
| /// The ID of the stream the event belongs to. This is typically the JSON-RPC request ID | ||
| /// for POST SSE responses, or a special identifier for the standalone GET SSE stream. | ||
| /// </param> | ||
| /// <param name="message"> | ||
| /// The JSON-RPC message to store, or <see langword="null"/> for priming events. | ||
| /// Priming events establish the event ID without carrying a message payload. | ||
| /// </param> |
There was a problem hiding this comment.
What is the definition of "event" here? It says this is for storing an event, but then accepts the JSON-RPC message to store. Is the intimation that the event contains that message?
| ValueTask<string?> ReplayEventsAfterAsync( | ||
| string lastEventId, | ||
| Func<JsonRpcMessage, string, CancellationToken, ValueTask> sendCallback, | ||
| CancellationToken cancellationToken = default); |
There was a problem hiding this comment.
I'm curious why you chose to expose it like this rather than, say, exposing a GetEventsAfter method that returned an enumerable.
| await foreach (var item in messages.WithCancellation(cancellationToken).ConfigureAwait(false)) | ||
| { | ||
| // Skip endpoint events, priming events, and replayed events (which already have IDs) | ||
| if (item.EventType == "endpoint" || item.EventType == "priming" || item.EventId is not null) |
There was a problem hiding this comment.
nit:
| if (item.EventType == "endpoint" || item.EventType == "priming" || item.EventId is not null) | |
| if (item.EventType is "endpoint" or "priming" || item.EventId is not null) |
Opening initially as a draft while adding/debugging remaining E2E tests
Fixes #510
Fixes #1020