Streamable HTTP resumability + redelivery + SSE polling via server-side disconnect (#1077)
Co-authored-by: Stephen Halter <halter73@gmail.com>
This commit is contained in:
Mackinnon Buck
GitHub
parent
4cb2fd4655
commit
b8311756b4
3 changed files with 2675 additions and 246 deletions
|
|
@ -0,0 +1,175 @@
|
|||
using System.Diagnostics;
|
||||
using System.Net;
|
||||
|
||||
namespace ModelContextProtocol.AspNetCore.Tests.Utils;
|
||||
|
||||
/// <summary>
|
||||
/// A message handler that wraps SSE response streams and can trigger faults mid-stream
|
||||
/// to simulate network disconnections during SSE streaming.
|
||||
/// </summary>
|
||||
internal sealed class FaultingStreamHandler : DelegatingHandler
|
||||
{
|
||||
private FaultingStream? _lastStream;
|
||||
private TaskCompletionSource? _reconnectTcs;
|
||||
|
||||
public async Task<ReconnectAttempt> TriggerFaultAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (_lastStream is null or { IsDisposed: true })
|
||||
{
|
||||
throw new InvalidOperationException("There is no active response stream to fault.");
|
||||
}
|
||||
|
||||
if (_reconnectTcs is not null)
|
||||
{
|
||||
throw new InvalidOperationException("Cannot trigger a fault while already waiting for reconnection.");
|
||||
}
|
||||
|
||||
_reconnectTcs = new();
|
||||
await _lastStream.TriggerFaultAsync(cancellationToken);
|
||||
|
||||
return new(_reconnectTcs);
|
||||
}
|
||||
|
||||
public sealed class ReconnectAttempt(TaskCompletionSource reconnectTcs)
|
||||
{
|
||||
public void Continue()
|
||||
=> reconnectTcs.SetResult();
|
||||
}
|
||||
|
||||
protected override async Task<HttpResponseMessage> SendAsync(
|
||||
HttpRequestMessage request, CancellationToken cancellationToken)
|
||||
{
|
||||
if (_reconnectTcs is not null && request.Headers.Accept.Contains(new("text/event-stream")))
|
||||
{
|
||||
// If we're blocking reconnection, wait until we're allowed to continue.
|
||||
await _reconnectTcs.Task.WaitAsync(cancellationToken);
|
||||
_reconnectTcs = null;
|
||||
}
|
||||
|
||||
var response = await base.SendAsync(request, cancellationToken);
|
||||
|
||||
// Only wrap SSE streams (text/event-stream)
|
||||
if (response.Content.Headers.ContentType?.MediaType == "text/event-stream")
|
||||
{
|
||||
var originalStream = await response.Content.ReadAsStreamAsync(cancellationToken);
|
||||
_lastStream = new FaultingStream(originalStream);
|
||||
var faultingContent = new FaultingStreamContent(_lastStream);
|
||||
|
||||
// Copy headers from original content
|
||||
var newContent = faultingContent;
|
||||
foreach (var header in response.Content.Headers)
|
||||
{
|
||||
newContent.Headers.TryAddWithoutValidation(header.Key, header.Value);
|
||||
}
|
||||
|
||||
response.Content = newContent;
|
||||
}
|
||||
|
||||
return response;
|
||||
}
|
||||
|
||||
private sealed class FaultingStreamContent(FaultingStream stream) : HttpContent
|
||||
{
|
||||
private readonly FaultingStream _manualStream = new(stream);
|
||||
|
||||
protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context)
|
||||
=> throw new NotSupportedException();
|
||||
|
||||
protected override Task<Stream> CreateContentReadStreamAsync()
|
||||
=> Task.FromResult<Stream>(stream);
|
||||
|
||||
protected override bool TryComputeLength(out long length)
|
||||
{
|
||||
length = -1;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class FaultingStream(Stream innerStream) : Stream
|
||||
{
|
||||
private readonly CancellationTokenSource _cts = new();
|
||||
private TaskCompletionSource? _faultTcs;
|
||||
private bool _disposed;
|
||||
|
||||
public bool IsDisposed => _disposed;
|
||||
|
||||
public async Task TriggerFaultAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (_faultTcs is not null)
|
||||
{
|
||||
throw new InvalidOperationException("Only one fault can be triggered per stream.");
|
||||
}
|
||||
|
||||
_faultTcs = new TaskCompletionSource();
|
||||
|
||||
await _cts.CancelAsync();
|
||||
|
||||
// Use a timeout to detect if the fault is not observed by a read operation.
|
||||
using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
timeoutCts.CancelAfter(TimeSpan.FromSeconds(30));
|
||||
|
||||
try
|
||||
{
|
||||
await _faultTcs.Task.WaitAsync(timeoutCts.Token);
|
||||
}
|
||||
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw new TimeoutException(
|
||||
$"TriggerFaultAsync timed out after 30 seconds waiting for a read to observe the cancellation. " +
|
||||
$"Stream disposed: {_disposed}, CTS cancelled: {_cts.IsCancellationRequested}");
|
||||
}
|
||||
}
|
||||
|
||||
public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
|
||||
{
|
||||
try
|
||||
{
|
||||
_cts.Token.ThrowIfCancellationRequested();
|
||||
|
||||
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cts.Token);
|
||||
var bytesRead = await innerStream.ReadAsync(buffer, linkedCts.Token);
|
||||
|
||||
_cts.Token.ThrowIfCancellationRequested();
|
||||
|
||||
return bytesRead;
|
||||
}
|
||||
catch (OperationCanceledException) when (_cts.IsCancellationRequested)
|
||||
{
|
||||
Debug.Assert(_faultTcs is not null);
|
||||
|
||||
if (!_faultTcs.TrySetResult())
|
||||
{
|
||||
throw new InvalidOperationException("Attempted to read an already-faulted stream.");
|
||||
}
|
||||
|
||||
throw new IOException("Simulated network disconnection.");
|
||||
}
|
||||
}
|
||||
|
||||
public override int Read(byte[] buffer, int offset, int count)
|
||||
=> throw new NotSupportedException("Synchronous reads are not supported.");
|
||||
|
||||
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
|
||||
=> ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();
|
||||
|
||||
public override bool CanRead => innerStream.CanRead;
|
||||
public override bool CanSeek => innerStream.CanSeek;
|
||||
public override bool CanWrite => innerStream.CanWrite;
|
||||
public override long Length => innerStream.Length;
|
||||
public override long Position { get => innerStream.Position; set => innerStream.Position = value; }
|
||||
public override void Flush() => innerStream.Flush();
|
||||
public override long Seek(long offset, SeekOrigin origin) => innerStream.Seek(offset, origin);
|
||||
public override void SetLength(long value) => innerStream.SetLength(value);
|
||||
public override void Write(byte[] buffer, int offset, int count) => innerStream.Write(buffer, offset, count);
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
if (!disposing || _disposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_disposed = true;
|
||||
innerStream.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -27,21 +27,24 @@ public class KestrelInMemoryTest : LoggedTest
|
|||
return new(connection.ClientStream);
|
||||
};
|
||||
|
||||
HttpClient = new HttpClient(SocketsHttpHandler)
|
||||
{
|
||||
BaseAddress = new Uri("http://localhost:5000/"),
|
||||
Timeout = TimeSpan.FromSeconds(10),
|
||||
};
|
||||
HttpClient = new HttpClient(SocketsHttpHandler);
|
||||
ConfigureHttpClient(HttpClient);
|
||||
}
|
||||
|
||||
public WebApplicationBuilder Builder { get; }
|
||||
|
||||
public HttpClient HttpClient { get; }
|
||||
public HttpClient HttpClient { get; set; }
|
||||
|
||||
public SocketsHttpHandler SocketsHttpHandler { get; } = new();
|
||||
|
||||
public KestrelInMemoryTransport KestrelInMemoryTransport { get; } = new();
|
||||
|
||||
protected static void ConfigureHttpClient(HttpClient httpClient)
|
||||
{
|
||||
httpClient.BaseAddress = new Uri("http://localhost:5000/");
|
||||
httpClient.Timeout = TimeSpan.FromSeconds(10);
|
||||
}
|
||||
|
||||
public override void Dispose()
|
||||
{
|
||||
HttpClient.Dispose();
|
||||
|
|
|
|||
|
|
@ -0,0 +1,268 @@
|
|||
using ModelContextProtocol.Protocol;
|
||||
using ModelContextProtocol.Server;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Net.ServerSentEvents;
|
||||
using System.Runtime.CompilerServices;
|
||||
|
||||
namespace ModelContextProtocol.AspNetCore.Tests.Utils;
|
||||
|
||||
/// <summary>
|
||||
/// In-memory event store for testing resumability.
|
||||
/// This is a simple implementation intended for testing, not for production use.
|
||||
/// </summary>
|
||||
public sealed class TestSseEventStreamStore : ISseEventStreamStore
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, StreamState> _streams = new();
|
||||
private readonly ConcurrentDictionary<string, (StreamState Stream, long Sequence)> _eventLookup = new();
|
||||
private readonly List<string> _storedEventIds = [];
|
||||
private readonly List<TimeSpan> _storedReconnectionIntervals = [];
|
||||
private readonly object _storedEventIdsLock = new();
|
||||
private int _storeEventCallCount;
|
||||
private long _globalSequence;
|
||||
|
||||
/// <summary>
|
||||
/// Gets the number of times events have been stored.
|
||||
/// </summary>
|
||||
public int StoreEventCallCount => _storeEventCallCount;
|
||||
|
||||
/// <summary>
|
||||
/// Gets the list of stored event IDs in order.
|
||||
/// </summary>
|
||||
public IReadOnlyList<string> StoredEventIds
|
||||
{
|
||||
get
|
||||
{
|
||||
lock (_storedEventIdsLock)
|
||||
{
|
||||
return [.. _storedEventIds];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the list of stored reconnection intervals in order.
|
||||
/// </summary>
|
||||
public IReadOnlyList<TimeSpan> StoredReconnectionIntervals
|
||||
{
|
||||
get
|
||||
{
|
||||
lock (_storedEventIdsLock)
|
||||
{
|
||||
return [.. _storedReconnectionIntervals];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask<ISseEventStreamWriter> CreateStreamAsync(SseEventStreamOptions options, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var streamKey = GetStreamKey(options.SessionId, options.StreamId);
|
||||
var state = new StreamState(options.SessionId, options.StreamId, options.Mode);
|
||||
if (!_streams.TryAdd(streamKey, state))
|
||||
{
|
||||
throw new InvalidOperationException($"A stream with key '{streamKey}' has already been created.");
|
||||
}
|
||||
var writer = new InMemoryEventStreamWriter(this, state);
|
||||
return new ValueTask<ISseEventStreamWriter>(writer);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask<ISseEventStreamReader?> GetStreamReaderAsync(string lastEventId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
// Look up the event by its ID to find which stream it belongs to
|
||||
if (!_eventLookup.TryGetValue(lastEventId, out var lookup))
|
||||
{
|
||||
return new ValueTask<ISseEventStreamReader?>((ISseEventStreamReader?)null);
|
||||
}
|
||||
|
||||
var reader = new InMemoryEventStreamReader(lookup.Stream, lookup.Sequence);
|
||||
return new ValueTask<ISseEventStreamReader?>(reader);
|
||||
}
|
||||
|
||||
private string GenerateEventId() => Interlocked.Increment(ref _globalSequence).ToString();
|
||||
|
||||
private void TrackEvent(string eventId, StreamState stream, long sequence, TimeSpan? reconnectionInterval = null)
|
||||
{
|
||||
_eventLookup[eventId] = (stream, sequence);
|
||||
lock (_storedEventIdsLock)
|
||||
{
|
||||
_storedEventIds.Add(eventId);
|
||||
if (reconnectionInterval.HasValue)
|
||||
{
|
||||
_storedReconnectionIntervals.Add(reconnectionInterval.Value);
|
||||
}
|
||||
}
|
||||
Interlocked.Increment(ref _storeEventCallCount);
|
||||
}
|
||||
|
||||
private static string GetStreamKey(string sessionId, string streamId) => $"{sessionId}:{streamId}";
|
||||
|
||||
/// <summary>
|
||||
/// Holds the state for a single stream.
|
||||
/// </summary>
|
||||
private sealed class StreamState
|
||||
{
|
||||
private readonly List<(SseItem<JsonRpcMessage?> Item, long Sequence)> _events = [];
|
||||
private readonly object _lock = new();
|
||||
private TaskCompletionSource _newEventSignal = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private long _sequence;
|
||||
|
||||
public StreamState(string sessionId, string streamId, SseEventStreamMode mode)
|
||||
{
|
||||
SessionId = sessionId;
|
||||
StreamId = streamId;
|
||||
Mode = mode;
|
||||
}
|
||||
|
||||
public string SessionId { get; }
|
||||
public string StreamId { get; }
|
||||
public SseEventStreamMode Mode { get; set; }
|
||||
public bool IsCompleted { get; private set; }
|
||||
|
||||
public long NextSequence() => Interlocked.Increment(ref _sequence);
|
||||
|
||||
public void AddEvent(SseItem<JsonRpcMessage?> item, long sequence)
|
||||
{
|
||||
lock (_lock)
|
||||
{
|
||||
if (IsCompleted)
|
||||
{
|
||||
throw new InvalidOperationException("Cannot add events to a completed stream.");
|
||||
}
|
||||
|
||||
_events.Add((item, sequence));
|
||||
|
||||
var oldSignal = _newEventSignal;
|
||||
_newEventSignal = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
oldSignal.TrySetResult();
|
||||
}
|
||||
}
|
||||
|
||||
public (List<SseItem<JsonRpcMessage?>> Events, long LastSequence, Task NewEventSignal) GetEventsAfter(long sequence)
|
||||
{
|
||||
lock (_lock)
|
||||
{
|
||||
var result = new List<SseItem<JsonRpcMessage?>>();
|
||||
long lastSequence = sequence;
|
||||
|
||||
foreach (var (item, seq) in _events)
|
||||
{
|
||||
if (seq > sequence)
|
||||
{
|
||||
result.Add(item);
|
||||
lastSequence = seq;
|
||||
}
|
||||
}
|
||||
|
||||
return (result, lastSequence, _newEventSignal.Task);
|
||||
}
|
||||
}
|
||||
|
||||
public void Complete()
|
||||
{
|
||||
lock (_lock)
|
||||
{
|
||||
IsCompleted = true;
|
||||
_newEventSignal.TrySetResult();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class InMemoryEventStreamWriter : ISseEventStreamWriter
|
||||
{
|
||||
private readonly TestSseEventStreamStore _store;
|
||||
private readonly StreamState _state;
|
||||
private bool _disposed;
|
||||
|
||||
public InMemoryEventStreamWriter(TestSseEventStreamStore store, StreamState state)
|
||||
{
|
||||
_store = store;
|
||||
_state = state;
|
||||
}
|
||||
|
||||
public ValueTask SetModeAsync(SseEventStreamMode mode, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_state.Mode = mode;
|
||||
return default;
|
||||
}
|
||||
|
||||
public ValueTask<SseItem<JsonRpcMessage?>> WriteEventAsync(SseItem<JsonRpcMessage?> sseItem, CancellationToken cancellationToken = default)
|
||||
{
|
||||
// Skip if already has an event ID
|
||||
if (sseItem.EventId is not null)
|
||||
{
|
||||
return new ValueTask<SseItem<JsonRpcMessage?>>(sseItem);
|
||||
}
|
||||
|
||||
var sequence = _state.NextSequence();
|
||||
var eventId = _store.GenerateEventId();
|
||||
var newItem = sseItem with { EventId = eventId };
|
||||
|
||||
_state.AddEvent(newItem, sequence);
|
||||
_store.TrackEvent(eventId, _state, sequence, sseItem.ReconnectionInterval);
|
||||
|
||||
return new ValueTask<SseItem<JsonRpcMessage?>>(newItem);
|
||||
}
|
||||
|
||||
public ValueTask DisposeAsync()
|
||||
{
|
||||
if (_disposed)
|
||||
{
|
||||
return default;
|
||||
}
|
||||
|
||||
_disposed = true;
|
||||
_state.Complete();
|
||||
return default;
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class InMemoryEventStreamReader : ISseEventStreamReader
|
||||
{
|
||||
private readonly StreamState _state;
|
||||
private readonly long _startSequence;
|
||||
|
||||
public InMemoryEventStreamReader(StreamState state, long startSequence)
|
||||
{
|
||||
_state = state;
|
||||
_startSequence = startSequence;
|
||||
}
|
||||
|
||||
public string SessionId => _state.SessionId;
|
||||
public string StreamId => _state.StreamId;
|
||||
|
||||
public async IAsyncEnumerable<SseItem<JsonRpcMessage?>> ReadEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
|
||||
{
|
||||
long lastSeenSequence = _startSequence;
|
||||
|
||||
while (true)
|
||||
{
|
||||
// Get events after the last seen sequence
|
||||
var (events, lastSequence, newEventSignal) = _state.GetEventsAfter(lastSeenSequence);
|
||||
|
||||
foreach (var evt in events)
|
||||
{
|
||||
yield return evt;
|
||||
}
|
||||
|
||||
// Update to the sequence we actually retrieved
|
||||
lastSeenSequence = lastSequence;
|
||||
|
||||
// If in polling mode, stop after returning currently available events
|
||||
if (_state.Mode == SseEventStreamMode.Polling)
|
||||
{
|
||||
yield break;
|
||||
}
|
||||
|
||||
// If the stream is completed, stop
|
||||
if (_state.IsCompleted)
|
||||
{
|
||||
yield break;
|
||||
}
|
||||
|
||||
// Wait for new events or cancellation
|
||||
await newEventSignal.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue