fix(aspnetcore): make SEP-2575 status mapping deterministic (#1837)
This commit is contained in:
parent
df11d7da3b
commit
cb3e6e87fe
4 changed files with 60 additions and 85 deletions
|
|
@ -76,9 +76,6 @@ internal sealed partial class StreamableHttpPostTransport(
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
CancellationTokenSource? deferredFlushCts = null;
|
|
||||||
Task? deferredFlushTask = null;
|
|
||||||
bool deferHeaderFlush = false;
|
|
||||||
using (await _messageLock.LockAsync(cancellationToken).ConfigureAwait(false))
|
using (await _messageLock.LockAsync(cancellationToken).ConfigureAwait(false))
|
||||||
{
|
{
|
||||||
var primingItem = await TryStartSseEventStreamAsync(_pendingRequest).ConfigureAwait(false);
|
var primingItem = await TryStartSseEventStreamAsync(_pendingRequest).ConfigureAwait(false);
|
||||||
|
|
@ -87,88 +84,33 @@ internal sealed partial class StreamableHttpPostTransport(
|
||||||
await NotifyResponseStartingAsync(firstMessage: null).ConfigureAwait(false);
|
await NotifyResponseStartingAsync(firstMessage: null).ConfigureAwait(false);
|
||||||
await _httpSseWriter.WriteAsync(primingItem.Value, cancellationToken).ConfigureAwait(false);
|
await _httpSseWriter.WriteAsync(primingItem.Value, cancellationToken).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
else if (onResponseStarting is null)
|
else if (onResponseStarting is not null &&
|
||||||
|
McpProtocolVersions.RequiresPerRequestMetadata(message.Context.ProtocolVersion))
|
||||||
{
|
{
|
||||||
// If there's no priming write, flush the stream to ensure HTTP response headers are
|
// Per-request-metadata protocol revisions map the first JSON-RPC error onto the HTTP
|
||||||
// sent to the client now that the server is ready to process the request.
|
// status line. Keep the headers uncommitted until that message arrives so the mapping
|
||||||
// This prevents HttpClient timeout for long-running requests.
|
// cannot depend on how long dispatch takes.
|
||||||
await responseStream.FlushAsync(cancellationToken).ConfigureAwait(false);
|
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
deferHeaderFlush = true;
|
// Earlier protocol revisions keep the eager header flush for long-running handlers.
|
||||||
|
// Mark the response as started before flushing because any later JSON-RPC error must
|
||||||
|
// not attempt to change an already committed status line.
|
||||||
|
await NotifyResponseStartingAsync(firstMessage: null).ConfigureAwait(false);
|
||||||
|
await responseStream.FlushAsync(cancellationToken).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Ensure that we've sent the priming event before processing the incoming request.
|
// Ensure that we've sent the priming event before processing the incoming request.
|
||||||
await parentTransport.MessageWriter.WriteAsync(message, cancellationToken).ConfigureAwait(false);
|
await parentTransport.MessageWriter.WriteAsync(message, cancellationToken).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (deferHeaderFlush)
|
// Wait for the response to be written before returning from the handler.
|
||||||
{
|
// This keeps the HTTP response open until the final response message is sent.
|
||||||
// Defer the flush (and the header commit it implies) so the callback can still choose
|
await _httpResponseTcs.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||||
// the HTTP status line for an immediate JSON-RPC error. Start the bounded grace period
|
|
||||||
// only after the request has been queued for dispatch.
|
|
||||||
deferredFlushCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
|
||||||
deferredFlushTask = DeferredHeaderFlushAsync(deferredFlushCts.Token);
|
|
||||||
}
|
|
||||||
|
|
||||||
try
|
|
||||||
{
|
|
||||||
// Wait for the response to be written before returning from the handler.
|
|
||||||
// This keeps the HTTP response open until the final response message is sent.
|
|
||||||
await _httpResponseTcs.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
finally
|
|
||||||
{
|
|
||||||
if (deferredFlushCts is not null)
|
|
||||||
{
|
|
||||||
deferredFlushCts.Cancel();
|
|
||||||
await deferredFlushTask!.ConfigureAwait(false);
|
|
||||||
deferredFlushCts.Dispose();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Bounds the deferred header flush: after a short grace window, flushes the response headers
|
|
||||||
/// if no response message has been written yet. Immediate rejections land well inside the
|
|
||||||
/// window, so the response-starting callback can still map their JSON-RPC error codes onto the
|
|
||||||
/// HTTP status line; a handler that runs longer commits the headers here so clients see them
|
|
||||||
/// promptly (long-running tool calls must not trip HttpClient's response timeout).
|
|
||||||
/// </summary>
|
|
||||||
private async Task DeferredHeaderFlushAsync(CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
|
||||||
await Task.Delay(DeferredHeaderFlushGrace, cancellationToken).ConfigureAwait(false);
|
|
||||||
using var _ = await _messageLock.LockAsync(cancellationToken).ConfigureAwait(false);
|
|
||||||
if (!_httpResponseStarted && !_httpResponseCompleted)
|
|
||||||
{
|
|
||||||
await NotifyResponseStartingAsync(firstMessage: null).ConfigureAwait(false);
|
|
||||||
await responseStream.FlushAsync(cancellationToken).ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
|
||||||
{
|
|
||||||
// The response was written or the request ended before the grace window elapsed.
|
|
||||||
}
|
|
||||||
catch (Exception ex)
|
|
||||||
{
|
|
||||||
// Surface the failure to the awaiting HandlePostAsync when possible. If the response
|
|
||||||
// future has already been resolved (the response started or completed on another path),
|
|
||||||
// TrySetException is a no-op, so log here to keep the deferred-flush failure diagnosable.
|
|
||||||
if (!_httpResponseTcs.TrySetException(ex))
|
|
||||||
{
|
|
||||||
LogDeferredHeaderFlushFailed(ex);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>How long the response-header flush may be deferred waiting for the first response message.</summary>
|
|
||||||
internal static readonly TimeSpan DeferredHeaderFlushGrace = TimeSpan.FromMilliseconds(250);
|
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Invokes the response-starting callback exactly once, immediately before the first write to
|
/// Invokes the response-starting callback exactly once, immediately before the first write to
|
||||||
/// the HTTP response stream, so the HTTP application can still set the response status line.
|
/// the HTTP response stream, so the HTTP application can still set the response status line.
|
||||||
|
|
@ -343,6 +285,4 @@ internal sealed partial class StreamableHttpPostTransport(
|
||||||
[LoggerMessage(Level = LogLevel.Warning, Message = "Failed to dispose SSE event stream writer.")]
|
[LoggerMessage(Level = LogLevel.Warning, Message = "Failed to dispose SSE event stream writer.")]
|
||||||
private partial void LogStoreStreamDisposalFailed(Exception exception);
|
private partial void LogStoreStreamDisposalFailed(Exception exception);
|
||||||
|
|
||||||
[LoggerMessage(Level = LogLevel.Warning, Message = "Failed to flush deferred Streamable HTTP response headers.")]
|
|
||||||
private partial void LogDeferredHeaderFlushFailed(Exception exception);
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -216,14 +216,14 @@ public sealed partial class StreamableHttpServerTransport : ITransport
|
||||||
/// This overload additionally reports the first JSON-RPC message written to the response via
|
/// This overload additionally reports the first JSON-RPC message written to the response via
|
||||||
/// <paramref name="onResponseStarting"/>, before any response bytes are written, so the HTTP
|
/// <paramref name="onResponseStarting"/>, before any response bytes are written, so the HTTP
|
||||||
/// application can still choose the response status line (SEP-2575 maps some JSON-RPC error
|
/// application can still choose the response status line (SEP-2575 maps some JSON-RPC error
|
||||||
/// codes to HTTP statuses). When <paramref name="onResponseStarting"/> is provided, the eager
|
/// codes to HTTP statuses). When <paramref name="onResponseStarting"/> is provided for a
|
||||||
/// response-header flush that normally precedes request processing is deferred until that first
|
/// per-request-metadata protocol revision, the eager response-header flush that normally precedes
|
||||||
/// message; the callback receives <see langword="null"/> when the first write is not a JSON-RPC
|
/// request processing is deferred until that first message; the callback receives
|
||||||
/// message (e.g. a resumability priming event).
|
/// <see langword="null"/> when the first write is not a JSON-RPC message (e.g. a resumability
|
||||||
|
/// priming event).
|
||||||
/// The status line can only be influenced by the FIRST write: when a handler streams a
|
/// The status line can only be influenced by the FIRST write: when a handler streams a
|
||||||
/// notification (e.g. progress) before failing, or runs past the transport's bounded
|
/// notification (e.g. progress) before failing, the status is already committed and a later
|
||||||
/// header-flush grace window, the status is already committed and a later JSON-RPC error
|
/// JSON-RPC error rides the committed status.
|
||||||
/// rides the committed status.
|
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="message">The JSON-RPC message to process.</param>
|
/// <param name="message">The JSON-RPC message to process.</param>
|
||||||
/// <param name="responseStream">The response stream to write any JSON-RPC responses to.</param>
|
/// <param name="responseStream">The response stream to write any JSON-RPC responses to.</param>
|
||||||
|
|
|
||||||
|
|
@ -239,10 +239,9 @@ public abstract partial class MapMcpTests(ITestOutputHelper testOutputHelper) :
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task LongRunningToolCall_DoesNotTimeout_WhenNoEventStreamStore()
|
public async Task LongRunningToolCall_DoesNotTimeout_WhenNoEventStreamStore()
|
||||||
{
|
{
|
||||||
// Regression test for: Tool calls that last over HttpClient timeout without producing
|
// Legacy Streamable HTTP tool calls that run past the HttpClient timeout without producing
|
||||||
// intermediate notifications will timeout because HttpClient doesn't see the 200 response
|
// intermediate notifications need an eager response flush. Per-request-metadata protocol
|
||||||
// until the first message is written. When primingItem is null (no ISseEventStreamStore),
|
// revisions instead wait for the first JSON-RPC message so SEP-2575 can map its status.
|
||||||
// we should flush the response stream so HttpClient sees the 200 immediately.
|
|
||||||
|
|
||||||
Builder.Services.AddMcpServer().WithHttpTransport(ConfigureStateless).WithTools<LongRunningTools>();
|
Builder.Services.AddMcpServer().WithHttpTransport(ConfigureStateless).WithTools<LongRunningTools>();
|
||||||
|
|
||||||
|
|
@ -274,7 +273,11 @@ public abstract partial class MapMcpTests(ITestOutputHelper testOutputHelper) :
|
||||||
TransportMode = transportMode,
|
TransportMode = transportMode,
|
||||||
}, shortTimeoutClient, LoggerFactory);
|
}, shortTimeoutClient, LoggerFactory);
|
||||||
|
|
||||||
await using var mcpClient = await McpClient.CreateAsync(transport, loggerFactory: LoggerFactory, cancellationToken: TestContext.Current.CancellationToken);
|
await using var mcpClient = await McpClient.CreateAsync(
|
||||||
|
transport,
|
||||||
|
new McpClientOptions { ProtocolVersion = "2025-11-25" },
|
||||||
|
LoggerFactory,
|
||||||
|
TestContext.Current.CancellationToken);
|
||||||
|
|
||||||
// Call a tool that takes 2 seconds - this should succeed despite the 1 second HttpClient timeout
|
// Call a tool that takes 2 seconds - this should succeed despite the 1 second HttpClient timeout
|
||||||
// because the response stream is flushed immediately after receiving the request
|
// because the response stream is flushed immediately after receiving the request
|
||||||
|
|
|
||||||
|
|
@ -207,6 +207,27 @@ public class RawHttpConformanceTests(ITestOutputHelper outputHelper) : KestrelIn
|
||||||
Assert.Equal((int)McpErrorCode.MissingRequiredClientCapability, json["error"]!["code"]!.GetValue<int>());
|
Assert.Equal((int)McpErrorCode.MissingRequiredClientCapability, json["error"]!["code"]!.GetValue<int>());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task July2026Post_SlowHandler_MissingRequiredCapability_Returns400()
|
||||||
|
{
|
||||||
|
await StartAsync();
|
||||||
|
|
||||||
|
var body =
|
||||||
|
@"{""jsonrpc"":""2.0"",""id"":20,""method"":""tools/call"",""params"":{""name"":""slow_requires_sampling"",""arguments"":{}," +
|
||||||
|
July2026ProtocolMetaFragment() + "}}";
|
||||||
|
|
||||||
|
using var request = new HttpRequestMessage(HttpMethod.Post, "") { Content = JsonContent(body) };
|
||||||
|
request.Headers.Add(ProtocolVersionHeader, McpProtocolVersions.July2026ProtocolVersion);
|
||||||
|
request.Headers.Add("Mcp-Method", "tools/call");
|
||||||
|
request.Headers.Add("Mcp-Name", "slow_requires_sampling");
|
||||||
|
using var response = await HttpClient.SendAsync(request, TestContext.Current.CancellationToken);
|
||||||
|
|
||||||
|
Assert.Equal(HttpStatusCode.BadRequest, response.StatusCode);
|
||||||
|
var json = await ReadJsonResponseAsync(response, TestContext.Current.CancellationToken);
|
||||||
|
Assert.Equal(20, json["id"]!.GetValue<long>());
|
||||||
|
Assert.Equal((int)McpErrorCode.MissingRequiredClientCapability, json["error"]!["code"]!.GetValue<int>());
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task ServerDiscover_WithConfiguredPerRequestMetadataProtocol_ReturnsOnlyConfiguredVersion()
|
public async Task ServerDiscover_WithConfiguredPerRequestMetadataProtocol_ReturnsOnlyConfiguredVersion()
|
||||||
{
|
{
|
||||||
|
|
@ -500,10 +521,21 @@ public class RawHttpConformanceTests(ITestOutputHelper outputHelper) : KestrelIn
|
||||||
[McpServerToolType]
|
[McpServerToolType]
|
||||||
private sealed class CapabilityTools
|
private sealed class CapabilityTools
|
||||||
{
|
{
|
||||||
|
private static readonly TimeSpan SlowHandlerDelay = TimeSpan.FromSeconds(1);
|
||||||
|
|
||||||
[McpServerTool(Name = "requires_sampling")]
|
[McpServerTool(Name = "requires_sampling")]
|
||||||
public static string RequiresSampling() =>
|
public static string RequiresSampling() =>
|
||||||
throw new MissingRequiredClientCapabilityException(
|
throw new MissingRequiredClientCapabilityException(
|
||||||
new ClientCapabilities { Sampling = new() },
|
new ClientCapabilities { Sampling = new() },
|
||||||
"sampling capability required but not declared by client");
|
"sampling capability required but not declared by client");
|
||||||
|
|
||||||
|
[McpServerTool(Name = "slow_requires_sampling")]
|
||||||
|
public static async Task<string> SlowRequiresSampling(CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
await Task.Delay(SlowHandlerDelay, cancellationToken);
|
||||||
|
throw new MissingRequiredClientCapabilityException(
|
||||||
|
new ClientCapabilities { Sampling = new() },
|
||||||
|
"sampling capability required but not declared by client");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue