Update Grpc.Core.Api to 2.45.0 and support write cancellation (#41525)

eng/Versions.props

@@ -243,11 +243,11 @@
     <FSharpCoreVersion>6.0.0</FSharpCoreVersion>
     <GoogleApiCommonProtosVersion>2.5.0</GoogleApiCommonProtosVersion>
     <GoogleProtobufVersion>3.18.1</GoogleProtobufVersion>
-    <GrpcAspNetCoreVersion>2.43.0</GrpcAspNetCoreVersion>
-    <GrpcAspNetCoreServerVersion>2.43.0</GrpcAspNetCoreServerVersion>
-    <GrpcAuthVersion>2.43.0</GrpcAuthVersion>
-    <GrpcNetClientVersion>2.43.0</GrpcNetClientVersion>
-    <GrpcToolsVersion>2.43.0</GrpcToolsVersion>
+    <GrpcAspNetCoreVersion>2.45.0</GrpcAspNetCoreVersion>
+    <GrpcAspNetCoreServerVersion>2.45.0</GrpcAspNetCoreServerVersion>
+    <GrpcAuthVersion>2.45.0</GrpcAuthVersion>
+    <GrpcNetClientVersion>2.45.0</GrpcNetClientVersion>
+    <GrpcToolsVersion>2.45.0</GrpcToolsVersion>
     <DuendeIdentityServerAspNetIdentityVersion>5.2.0</DuendeIdentityServerAspNetIdentityVersion>
     <DuendeIdentityServerEntityFrameworkVersion>5.2.0</DuendeIdentityServerEntityFrameworkVersion>
     <DuendeIdentityServerVersion>5.2.0</DuendeIdentityServerVersion>

src/Grpc/Grpc.slnf

@@ -3,6 +3,9 @@
     "path": "..\\..\\AspNetCore.sln",
     "projects": [
       "src\\Extensions\\Features\\src\\Microsoft.Extensions.Features.csproj",
+      "src\\Grpc\\Interop\\test\\InteropTests\\InteropTests.csproj",
+      "src\\Grpc\\Interop\\test\\testassets\\InteropClient\\InteropClient.csproj",
+      "src\\Grpc\\Interop\\test\\testassets\\InteropWebsite\\InteropWebsite.csproj",
       "src\\Grpc\\JsonTranscoding\\perf\\Microsoft.AspNetCore.Grpc.Microbenchmarks\\Microsoft.AspNetCore.Grpc.Microbenchmarks.csproj",
       "src\\Grpc\\JsonTranscoding\\src\\Microsoft.AspNetCore.Grpc.JsonTranscoding\\Microsoft.AspNetCore.Grpc.JsonTranscoding.csproj",
       "src\\Grpc\\JsonTranscoding\\src\\Microsoft.AspNetCore.Grpc.Swagger\\Microsoft.AspNetCore.Grpc.Swagger.csproj",
@@ -11,9 +14,6 @@
       "src\\Grpc\\JsonTranscoding\\test\\Microsoft.AspNetCore.Grpc.Swagger.Tests\\Microsoft.AspNetCore.Grpc.Swagger.Tests.csproj",
       "src\\Grpc\\JsonTranscoding\\test\\testassets\\IntegrationTestsWebsite\\IntegrationTestsWebsite.csproj",
       "src\\Grpc\\JsonTranscoding\\test\\testassets\\Sandbox\\Sandbox.csproj",
-      "src\\Grpc\\Interop\\test\\InteropTests\\InteropTests.csproj",
-      "src\\Grpc\\Interop\\test\\testassets\\InteropClient\\InteropClient.csproj",
-      "src\\Grpc\\Interop\\test\\testassets\\InteropWebsite\\InteropWebsite.csproj",
       "src\\Hosting\\Abstractions\\src\\Microsoft.AspNetCore.Hosting.Abstractions.csproj",
       "src\\Hosting\\Hosting\\src\\Microsoft.AspNetCore.Hosting.csproj",
       "src\\Hosting\\Server.Abstractions\\src\\Microsoft.AspNetCore.Hosting.Server.Abstractions.csproj",

src/Grpc/Interop/test/testassets/InteropWebsite/TestServiceImpl.cs

@@ -47,10 +47,9 @@ public class TestServiceImpl : TestService.TestServiceBase
 
         foreach (var responseParam in request.ResponseParameters)
         {
-            // TODO(JamesNK): Remove nullable override after Grpc.Core.Api update
             responseStream.WriteOptions = !(responseParam.Compressed?.Value ?? false)
                 ? new WriteOptions(WriteFlags.NoCompress)
-                : null!;
+                : null;
 
             var response = new StreamingOutputCallResponse { Payload = CreateZerosPayload(responseParam.Size) };
             await responseStream.WriteAsync(response);

src/Grpc/JsonTranscoding/src/Microsoft.AspNetCore.Grpc.JsonTranscoding/Internal/CallHandlers/UnaryServerCallHandler.cs

@@ -44,6 +44,6 @@ internal sealed class UnaryServerCallHandler<TService, TRequest, TResponse> : Se
 
         serverCallContext.EnsureResponseHeaders();
 
-        await JsonRequestHelpers.SendMessage(serverCallContext, SerializerOptions, response);
+        await JsonRequestHelpers.SendMessage(serverCallContext, SerializerOptions, response, CancellationToken.None);
     }
 }

src/Grpc/JsonTranscoding/src/Microsoft.AspNetCore.Grpc.JsonTranscoding/Internal/HttpContextStreamWriter.cs

@@ -22,43 +22,71 @@ internal sealed class HttpContextStreamWriter<TResponse> : IServerStreamWriter<T
         _writeLock = new object();
     }
 
-    public WriteOptions WriteOptions
+    public WriteOptions? WriteOptions
     {
         get => _context.WriteOptions;
         set => _context.WriteOptions = value;
     }
 
+    Task IAsyncStreamWriter<TResponse>.WriteAsync(TResponse message, CancellationToken cancellationToken)
+    {
+        return WriteAsyncCore(message, cancellationToken);
+    }
+
     public Task WriteAsync(TResponse message)
+    {
+        return WriteAsyncCore(message, CancellationToken.None);
+    }
+
+    private async Task WriteAsyncCore(TResponse message, CancellationToken cancellationToken)
     {
         if (message == null)
         {
-            return Task.FromException(new ArgumentNullException(nameof(message)));
+            throw new ArgumentNullException(nameof(message));
         }
 
-        if (_completed || _context.CancellationToken.IsCancellationRequested)
+        // Register cancellation token early to ensure request is canceled if cancellation is requested.
+        CancellationTokenRegistration? registration = null;
+        if (cancellationToken.CanBeCanceled)
         {
-            return Task.FromException(new InvalidOperationException("Can't write the message because the request is complete."));
+            registration = cancellationToken.Register(
+                static (state) => ((JsonTranscodingServerCallContext)state!).HttpContext.Abort(),
+                _context);
         }
 
-        lock (_writeLock)
+        try
         {
-            // Pending writes need to be awaited first
-            if (IsWriteInProgressUnsynchronized)
+            cancellationToken.ThrowIfCancellationRequested();
+
+            if (_completed || _context.CancellationToken.IsCancellationRequested)
             {
-                return Task.FromException(new InvalidOperationException("Can't write the message because the previous write is in progress."));
+                throw new InvalidOperationException("Can't write the message because the request is complete.");
             }
 
-            // Save write task to track whether it is complete. Must be set inside lock.
-            _writeTask = WriteMessageAndDelimiter(message);
-        }
+            lock (_writeLock)
+            {
+                // Pending writes need to be awaited first
+                if (IsWriteInProgressUnsynchronized)
+                {
+                    throw new InvalidOperationException("Can't write the message because the previous write is in progress.");
+                }
+
+                // Save write task to track whether it is complete. Must be set inside lock.
+                _writeTask = WriteMessageAndDelimiter(message, cancellationToken);
+            }
 
-        return _writeTask;
+            await _writeTask;
+        }
+        finally
+        {
+            registration?.Dispose();
+        }
     }
 
-    private async Task WriteMessageAndDelimiter(TResponse message)
+    private async Task WriteMessageAndDelimiter(TResponse message, CancellationToken cancellationToken)
     {
-        await JsonRequestHelpers.SendMessage(_context, _serializerOptions, message);
-        await _context.HttpContext.Response.Body.WriteAsync(GrpcProtocolConstants.StreamingDelimiter);
+        await JsonRequestHelpers.SendMessage(_context, _serializerOptions, message, cancellationToken);
+        await _context.HttpContext.Response.Body.WriteAsync(GrpcProtocolConstants.StreamingDelimiter, cancellationToken);
     }
 
     public void Complete()

src/Grpc/JsonTranscoding/src/Microsoft.AspNetCore.Grpc.JsonTranscoding/Internal/JsonRequestHelpers.cs

@@ -100,7 +100,7 @@ internal static class JsonRequestHelpers
             Code = (int)status.StatusCode
         };
 
-        await WriteResponseMessage(response, encoding, e, options);
+        await WriteResponseMessage(response, encoding, e, options, CancellationToken.None);
     }
 
     public static int MapStatusCodeToHttpStatus(StatusCode statusCode)
@@ -147,13 +147,13 @@ internal static class JsonRequestHelpers
         return StatusCodes.Status500InternalServerError;
     }
 
-    public static async Task WriteResponseMessage(HttpResponse response, Encoding encoding, object responseBody, JsonSerializerOptions options)
+    public static async Task WriteResponseMessage(HttpResponse response, Encoding encoding, object responseBody, JsonSerializerOptions options, CancellationToken cancellationToken)
     {
         var (stream, usesTranscodingStream) = GetStream(response.Body, encoding);
 
         try
         {
-            await JsonSerializer.SerializeAsync(stream, responseBody, options);
+            await JsonSerializer.SerializeAsync(stream, responseBody, options, cancellationToken);
         }
         finally
         {
@@ -281,7 +281,7 @@ internal static class JsonRequestHelpers
         });
     }
 
-    public static async Task SendMessage<TResponse>(JsonTranscodingServerCallContext serverCallContext, JsonSerializerOptions serializerOptions, TResponse message) where TResponse : class
+    public static async Task SendMessage<TResponse>(JsonTranscodingServerCallContext serverCallContext, JsonSerializerOptions serializerOptions, TResponse message, CancellationToken cancellationToken) where TResponse : class
     {
         var response = serverCallContext.HttpContext.Response;
 
@@ -304,7 +304,7 @@ internal static class JsonRequestHelpers
                 responseType = message.GetType();
             }
 
-            await JsonRequestHelpers.WriteResponseMessage(response, serverCallContext.RequestEncoding, responseBody, serializerOptions);
+            await JsonRequestHelpers.WriteResponseMessage(response, serverCallContext.RequestEncoding, responseBody, serializerOptions, cancellationToken);
 
             GrpcServerLog.SerializedMessage(serverCallContext.Logger, responseType);
             GrpcServerLog.MessageSent(serverCallContext.Logger);

src/Grpc/JsonTranscoding/src/Microsoft.AspNetCore.Grpc.JsonTranscoding/Internal/HttpApiServerCallContext.cs → src/Grpc/JsonTranscoding/src/Microsoft.AspNetCore.Grpc.JsonTranscoding/Internal/JsonTranscodingServerCallContext.cs

@@ -18,8 +18,7 @@ namespace Microsoft.AspNetCore.Grpc.JsonTranscoding.Internal;
 
 internal sealed class JsonTranscodingServerCallContext : ServerCallContext, IServerCallContextFeature
 {
-    // TODO(JamesNK): Remove nullable override after Grpc.Core.Api update
-    private static readonly AuthContext UnauthenticatedContext = new AuthContext(null!, new Dictionary<string, List<AuthProperty>>());
+    private static readonly AuthContext UnauthenticatedContext = new AuthContext(null, new Dictionary<string, List<AuthProperty>>());
 
     private readonly IMethod _method;
 
@@ -168,7 +167,7 @@ internal sealed class JsonTranscodingServerCallContext : ServerCallContext, ISer
 
     protected override Status StatusCore { get; set; }
 
-    protected override WriteOptions WriteOptionsCore
+    protected override WriteOptions? WriteOptionsCore
     {
         get => throw new NotImplementedException();
         set => throw new NotImplementedException();
@@ -191,7 +190,6 @@ internal sealed class JsonTranscodingServerCallContext : ServerCallContext, ISer
         }
     }
 
-    // TODO(JamesNK): Remove nullable override after Grpc.Core.Api update
     protected override IDictionary<object, object> UserStateCore => HttpContext.Items!;
 
     protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options)

src/Grpc/JsonTranscoding/src/Shared/AuthContextHelpers.cs

@@ -54,8 +54,7 @@ internal static class AuthContextHelpers
             }
         }
 
-        // TODO(JamesNK): Remove nullable override after Grpc.Core.Api update
-        return new AuthContext(peerIdentityPropertyName!, properties);
+        return new AuthContext(peerIdentityPropertyName, properties);
 
         static void AddProperty(Dictionary<string, List<AuthProperty>> properties, string name, string value)
         {

src/Grpc/JsonTranscoding/test/Microsoft.AspNetCore.Grpc.JsonTranscoding.IntegrationTests/ServerStreamingTests.cs

@@ -77,4 +77,40 @@ public class ServerStreamingTests : IntegrationTestBase
         // Assert 2
         Assert.Equal("Hello test 2!", result2.RootElement.GetProperty("message").GetString());
     }
+
+    [Fact]
+    public async Task GetWithRouteParameter_WriteMultiple_CancellationBefore_CallCanceled()
+    {
+        // Arrange
+        var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        async Task ServerStreamingMethod(HelloRequest request, IServerStreamWriter<HelloReply> writer, ServerCallContext context)
+        {
+            await writer.WriteAsync(new HelloReply { Message = $"Hello {request.Name} 1!" });
+            await tcs.Task;
+            await writer.WriteAsync(new HelloReply { Message = $"Hello {request.Name} 2!" }, new CancellationToken(canceled: true));
+        }
+        var method = Fixture.DynamicGrpc.AddServerStreamingMethod<HelloRequest, HelloReply>(
+            ServerStreamingMethod,
+            Greeter.Descriptor.FindMethodByName("SayHello"));
+
+        var client = new HttpClient(Fixture.Handler) { BaseAddress = new Uri("http://localhost") };
+
+        // Act 1
+        var response = await client.GetAsync("/v1/greeter/test", HttpCompletionOption.ResponseHeadersRead).DefaultTimeout();
+        var responseStream = await response.Content.ReadAsStreamAsync();
+        var streamReader = new StreamReader(responseStream);
+
+        var line1 = await streamReader.ReadLineAsync();
+        using var result1 = JsonDocument.Parse(line1!);
+
+        // Assert 1
+        Assert.Equal("Hello test 1!", result1.RootElement.GetProperty("message").GetString());
+
+        // Act 2
+        tcs.SetResult();
+        var ex = await Assert.ThrowsAsync<IOException>(() => streamReader.ReadLineAsync());
+
+        // Assert 2
+        Assert.Equal("The application aborted the request.", ex.InnerException!.Message);
+    }
 }