From 8fff63016bf3c23f3b7cce3c82ff1038cfb334ce Mon Sep 17 00:00:00 2001 From: Whit Waldo Date: Tue, 2 Jun 2026 14:38:57 -0500 Subject: [PATCH] No longer passing exception when workflow is shut down per cancellation request Signed-off-by: Whit Waldo --- src/Dapr.Workflow/Logging.cs | 2 +- .../Worker/Grpc/GrpcProtocolHandler.cs | 7 +- .../Worker/Grpc/GrpcProtocolHandlerTests.cs | 130 ++++++++++++++++++ 3 files changed, 134 insertions(+), 5 deletions(-) diff --git a/src/Dapr.Workflow/Logging.cs b/src/Dapr.Workflow/Logging.cs index 14b117fbc..cc73c35ff 100644 --- a/src/Dapr.Workflow/Logging.cs +++ b/src/Dapr.Workflow/Logging.cs @@ -83,7 +83,7 @@ internal static partial class Logging public static partial void LogGrpcProtocolHandlerReceiveLoopError(this ILogger logger, Exception ex); [LoggerMessage(LogLevel.Information, "Workflow worker gRPC stream canceled during shutdown (expected)")] - public static partial void LogGrpcProtocolHandlerReceiveLoopCanceled(this ILogger logger, Exception ex); + public static partial void LogGrpcProtocolHandlerReceiveLoopCanceled(this ILogger logger); [LoggerMessage(LogLevel.Information, "Disposing gRPC protocol handler")] public static partial void LogGrpcProtocolHandlerDisposing(this ILogger logger); diff --git a/src/Dapr.Workflow/Worker/Grpc/GrpcProtocolHandler.cs b/src/Dapr.Workflow/Worker/Grpc/GrpcProtocolHandler.cs index f5b178333..722f44d60 100644 --- a/src/Dapr.Workflow/Worker/Grpc/GrpcProtocolHandler.cs +++ b/src/Dapr.Workflow/Worker/Grpc/GrpcProtocolHandler.cs @@ -17,7 +17,6 @@ using System.Threading.Tasks; using Dapr.Common; using Dapr.DurableTask.Protobuf; -using Dapr.Workflow.Worker; using Google.Protobuf.WellKnownTypes; using Grpc.Core; using Microsoft.Extensions.Logging; @@ -193,15 +192,15 @@ private async Task ReceiveLoopAsync( await Task.WhenAll(activeWorkItems); } } - catch (OperationCanceledException ex) when (cancellationToken.IsCancellationRequested) + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { // Normal shutdown path (host stopping / handler disposing / token canceled) - _logger.LogGrpcProtocolHandlerReceiveLoopCanceled(ex); + _logger.LogGrpcProtocolHandlerReceiveLoopCanceled(); } catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled && cancellationToken.IsCancellationRequested) { // gRPC surfaces token/dispose cancellation as StatusCode.Cancelled - _logger.LogGrpcProtocolHandlerReceiveLoopCanceled(ex); + _logger.LogGrpcProtocolHandlerReceiveLoopCanceled(); } catch (Exception ex) { diff --git a/test/Dapr.Workflow.Test/Worker/Grpc/GrpcProtocolHandlerTests.cs b/test/Dapr.Workflow.Test/Worker/Grpc/GrpcProtocolHandlerTests.cs index 843195d19..17524472d 100644 --- a/test/Dapr.Workflow.Test/Worker/Grpc/GrpcProtocolHandlerTests.cs +++ b/test/Dapr.Workflow.Test/Worker/Grpc/GrpcProtocolHandlerTests.cs @@ -13,9 +13,11 @@ using System.Diagnostics; using System.Reflection; +using System.Collections.Concurrent; using Dapr.DurableTask.Protobuf; using Dapr.Workflow.Worker.Grpc; using Grpc.Core; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Moq; @@ -1063,6 +1065,79 @@ await RunHandlerUntilAsync( Assert.True(Volatile.Read(ref getWorkItemsCalls) >= 1); } + [Fact] + public async Task StartAsync_ShouldNotLogException_WhenReceiveLoopCancellationIsRequested() + { + var grpcClientMock = CreateGrpcClientMock(); + using var loggerFactory = new CapturingLoggerFactory(); + + var loggedCancellation = CreateTcs(); + + loggerFactory.Logged += entry => + { + if (entry.Message == "Workflow worker gRPC stream canceled during shutdown (expected)") + { + loggedCancellation.TrySetResult(entry); + } + }; + + grpcClientMock + .Setup(x => x.GetWorkItems(It.IsAny(), It.IsAny())) + .Returns(() => CreateServerStreamingCallFromReader(new CancelledWhenTokenCanceledStreamReader())); + + var handler = new GrpcProtocolHandler(grpcClientMock.Object, loggerFactory); + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(2)); + + var runTask = handler.StartAsync( + workflowHandler: (_, _) => Task.FromResult(new WorkflowResponse()), + activityHandler: (_, _) => Task.FromResult(new ActivityResponse()), + cancellationToken: cts.Token); + + await Task.Delay(50, cts.Token); + cts.Cancel(); + + await runTask; + + var cancellationLog = await loggedCancellation.Task.WaitAsync(TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + Assert.Equal(LogLevel.Information, cancellationLog.Level); + Assert.Null(cancellationLog.Exception); + } + + [Fact] + public async Task StartAsync_ShouldLogException_WhenReceiveLoopCancellationIsUnexpected() + { + var grpcClientMock = CreateGrpcClientMock(); + using var loggerFactory = new CapturingLoggerFactory(); + + var loggedUnexpectedCancellation = CreateTcs(); + + loggerFactory.Logged += entry => + { + if (entry.Message == "Error in receive loop" && entry.Exception is RpcException { StatusCode: StatusCode.Cancelled }) + { + loggedUnexpectedCancellation.TrySetResult(entry); + } + }; + + grpcClientMock + .Setup(x => x.GetWorkItems(It.IsAny(), It.IsAny())) + .Returns(CreateServerStreamingCallFromReader( + new ThrowingAsyncStreamReader(new RpcException(new Status(StatusCode.Cancelled, "unexpected"))))); + + var handler = new GrpcProtocolHandler(grpcClientMock.Object, loggerFactory); + + await RunHandlerUntilAsync( + handler, + workflowHandler: (_, _) => Task.FromResult(new WorkflowResponse()), + activityHandler: (_, _) => Task.FromResult(new ActivityResponse()), + until: loggedUnexpectedCancellation.Task, + timeout: TimeSpan.FromSeconds(2)); + + var cancellationLog = await loggedUnexpectedCancellation.Task; + Assert.Equal(LogLevel.Error, cancellationLog.Level); + Assert.NotNull(cancellationLog.Exception); + } + /// /// Regression test for: "Task failed: Status(StatusCode="Unknown", Detail="no such instance exists")" /// @@ -1219,6 +1294,17 @@ public Task MoveNext(CancellationToken cancellationToken) } } + private sealed class CancelledWhenTokenCanceledStreamReader : IAsyncStreamReader + { + public WorkItem Current => new(); + + public async Task MoveNext(CancellationToken cancellationToken) + { + await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + return false; + } + } + private sealed class ThrowingAfterOneAsyncStreamReader(WorkItem first, Exception thenThrow) : IAsyncStreamReader { private int _state; // 0 = before first, 1 = after first (throw), 2 = done (never reached) @@ -1275,6 +1361,50 @@ private sealed class NullStackTraceException(string message) : Exception(message public override string? StackTrace => null; } + private sealed record LogEntry(LogLevel Level, string Category, string Message, Exception? Exception); + + private sealed class CapturingLoggerFactory : ILoggerFactory + { + private readonly ConcurrentQueue _entries = new(); + + public event Action? Logged; + + public IReadOnlyCollection Entries => _entries.ToArray(); + + public ILogger CreateLogger(string categoryName) => new CapturingLogger(categoryName, Capture); + + public void AddProvider(ILoggerProvider provider) + { + } + + public void Dispose() + { + } + + private void Capture(LogEntry entry) + { + _entries.Enqueue(entry); + Logged?.Invoke(entry); + } + + private sealed class CapturingLogger(string categoryName, Action capture) : ILogger + { + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + capture(new LogEntry(logLevel, categoryName, formatter(state, exception), exception)); + } + } + } + private static Mock CreateGrpcClientMock() { var callInvoker = new Mock(MockBehavior.Loose);