fix: handle graceful cancellation in CloudEventsMiddleware to prevent 499 errors (#1713)

* fix: handle graceful cancellation in CloudEventsMiddleware to prevent 499 errors

Propagate CancellationToken to JsonSerializer and handle OperationCanceledException
gracefully when the response has already started. This prevents erroneous 499
Client Closed Connection errors in PubSub consumers under high load.

Fixes #1237

Signed-off-by: ali-Hamza817 <alihamzaminhas.21@gmail.com>

* test: fix cancellation test ordering and add response-not-started test

- Fix test ordering: move app.Run() before app.Build() to match
  the pattern of all other tests in the file
- Use IHttpResponseFeature to properly simulate HasStarted=true
  since DefaultHttpContext doesn't set it from WriteAsync
- Add new test: PropagatesCancellation_WhenResponseHasNotStarted
  to verify exceptions are NOT swallowed when the response has
  not started (correct ASP.NET Core behavior)

Signed-off-by: ali-Hamza817 <alihamzaminhas.21@gmail.com>

---------

Signed-off-by: ali-Hamza817 <alihamzaminhas.21@gmail.com>

Author: ali-Hamza817

src/Dapr.AspNetCore/CloudEventsMiddleware.cs

@@ -48,36 +48,30 @@ public CloudEventsMiddleware(RequestDelegate next, CloudEventsMiddlewareOptions
         this.options = options;
     }
 
-    public Task InvokeAsync(HttpContext httpContext)
+    public async Task InvokeAsync(HttpContext httpContext)
     {
-        // This middleware unwraps any requests with a cloud events (JSON) content type
-        // and replaces the request body + request content type so that it can be read by a
-        // non-cloud-events-aware piece of code.
-        //
-        // This corresponds to cloud events in the *structured* format:
-        // https://github.com/cloudevents/spec/blob/master/http-transport-binding.md#13-content-modes
-        //
-        // For *binary* format, we don't have to do anything
-        //
-        // We don't support batching.
-        //
-        // The philosophy here is that we don't report an error for things we don't support, because
-        // that would block someone from implementing their own support for it. We only report an error
-        // when something we do support isn't correct.
         if (!MatchesContentType(httpContext, out var charSet))
         {
-            return this.next(httpContext);
+            await this.next(httpContext);
+            return;
         }
 
-        return this.ProcessBodyAsync(httpContext, charSet);
+        try
+        {
+            await this.ProcessBodyAsync(httpContext, charSet);
+        }
+        catch (OperationCanceledException) when (httpContext.RequestAborted.IsCancellationRequested && httpContext.Response.HasStarted)
+        {
+            // Swallow
+        }
     }
 
     private async Task ProcessBodyAsync(HttpContext httpContext, string charSet)
     {
         JsonElement json;
         if (string.Equals(charSet, Encoding.UTF8.WebName, StringComparison.OrdinalIgnoreCase))
         {
-            json = await JsonSerializer.DeserializeAsync<JsonElement>(httpContext.Request.Body);
+            json = await JsonSerializer.DeserializeAsync<JsonElement>(httpContext.Request.Body, cancellationToken: httpContext.RequestAborted);
         }
         else
         {
@@ -142,7 +136,7 @@ private async Task ProcessBodyAsync(HttpContext httpContext, string charSet)
             if (isJson || options.SuppressJsonDecodingOfTextPayloads)
             {
                 // Rehydrate body from JSON payload
-                await JsonSerializer.SerializeAsync<JsonElement>(body, data);
+                await JsonSerializer.SerializeAsync<JsonElement>(body, data, cancellationToken: httpContext.RequestAborted);
             }
             else
             {

test/Dapr.AspNetCore.Test/CloudEventsMiddlewareTest.cs

@@ -20,9 +20,12 @@ namespace Dapr.AspNetCore.Test;
 using System.Text;
 using System.Text.Json;
 using System.Threading.Tasks;
+using System.Threading;
+using System;
 using Shouldly;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Http.Features;
 using Xunit;
 
 public class CloudEventsMiddlewareTest
@@ -478,4 +481,83 @@ private static string ReadBody(Stream stream, Encoding encoding = null)
         var str = encoding.GetString(bytes);
         return str;
     }
+
+    [Fact]
+    public async Task InvokeAsync_SwallowsCancellation_WhenResponseHasStarted()
+    {
+        using var cts = new CancellationTokenSource();
+
+        var serviceCollection = new ServiceCollection();
+        var provider = serviceCollection.BuildServiceProvider();
+
+        var app = new ApplicationBuilder(provider);
+        app.UseCloudEvents();
+
+        // Register terminal middleware BEFORE building the pipeline
+        app.Run(httpContext =>
+        {
+            // Simulate that the response has already started by replacing the response feature
+            httpContext.Features.Set<IHttpResponseFeature>(new TestHttpResponseFeature { HasStarted = true });
+
+            // Simulate the client disconnecting after the response started
+            cts.Cancel();
+            throw new OperationCanceledException(cts.Token);
+        });
+
+        var pipeline = app.Build();
+
+        var context = new DefaultHttpContext();
+        context.Request.ContentType = "application/cloudevents+json";
+        context.Request.Body = MakeBody("{ \"data\": { \"name\":\"jimmy\" } }");
+        context.RequestAborted = cts.Token;
+
+        // The middleware should catch and swallow the OperationCanceledException
+        // because the response has already started and the request was aborted
+        await pipeline.Invoke(context);
+    }
+
+    [Fact]
+    public async Task InvokeAsync_PropagatesCancellation_WhenResponseHasNotStarted()
+    {
+        using var cts = new CancellationTokenSource();
+
+        var serviceCollection = new ServiceCollection();
+        var provider = serviceCollection.BuildServiceProvider();
+
+        var app = new ApplicationBuilder(provider);
+        app.UseCloudEvents();
+
+        // Register terminal middleware that cancels BEFORE the response starts
+        app.Run(httpContext =>
+        {
+            // Response has NOT started (default), client disconnects
+            cts.Cancel();
+            throw new OperationCanceledException(cts.Token);
+        });
+
+        var pipeline = app.Build();
+
+        var context = new DefaultHttpContext();
+        context.Request.ContentType = "application/cloudevents+json";
+        context.Request.Body = MakeBody("{ \"data\": { \"name\":\"jimmy\" } }");
+        context.RequestAborted = cts.Token;
+
+        // The middleware should NOT swallow the exception when the response has not started
+        await Should.ThrowAsync<OperationCanceledException>(() => pipeline.Invoke(context));
+    }
+
+    /// <summary>
+    /// A test implementation of <see cref="IHttpResponseFeature"/> that allows controlling
+    /// the <see cref="HasStarted"/> property for unit testing.
+    /// </summary>
+    private sealed class TestHttpResponseFeature : IHttpResponseFeature
+    {
+        public int StatusCode { get; set; } = 200;
+        public string ReasonPhrase { get; set; }
+        public IHeaderDictionary Headers { get; set; } = new HeaderDictionary();
+        public Stream Body { get; set; } = new MemoryStream();
+        public bool HasStarted { get; set; }
+        public void OnStarting(Func<object, Task> callback, object state) { }
+        public void OnCompleted(Func<object, Task> callback, object state) { }
+    }
 }