From e49fe1dae3005c4e25f04fab7a55f6cfbe0faf3d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 3 May 2026 00:35:59 +0000 Subject: [PATCH 1/2] fix: SubscribeAsync now throws DaprException when sidecar is unavailable and supports error handler callback Fixes #1663 - SubscribeAsync catches RpcException during initial connection and wraps in DaprException - Resets hasInitialized flag on failure to allow retry - Added SubscriptionErrorHandler delegate to DaprSubscriptionOptions for handling background task failures - HandleTaskCompletion invokes error handler when configured, preserving existing behavior when not - Updated and added unit tests for all scenarios Agent-Logs-Url: https://github.com/dapr/dotnet-sdk/sessions/43bcaf39-8a3f-49c1-91df-f57b43a700bb Co-authored-by: WhitWaldo <2238529+WhitWaldo@users.noreply.github.com> --- .../DaprSubscriptionOptions.cs | 16 +++ .../PublishSubscribeReceiver.cs | 60 ++++++++-- .../PublishSubscribeReceiverTests.cs | 113 +++++++++++++++++- 3 files changed, 175 insertions(+), 14 deletions(-) diff --git a/src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs b/src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs index 73838b605..403fe0e1b 100644 --- a/src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs +++ b/src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs @@ -11,8 +11,17 @@ // limitations under the License. // ------------------------------------------------------------------------ +using Dapr; + namespace Dapr.Messaging.PublishSubscribe; +/// +/// A delegate that handles errors occurring during the subscription lifecycle after a successful initial connection. +/// +/// The exception that occurred. +/// A task representing the asynchronous error handling operation. +public delegate Task SubscriptionErrorHandler(DaprException exception); + /// /// Options used to configure the dynamic Dapr subscription. /// @@ -40,5 +49,12 @@ public sealed record DaprSubscriptionOptions(MessageHandlingPolicy MessageHandli /// been signaled. /// public TimeSpan MaximumCleanupTimeout { get; init; } = TimeSpan.FromSeconds(30); + + /// + /// An optional error handler that is invoked when background tasks (sidecar streaming, acknowledgement processing, + /// message processing) fault after a successful initial subscription. If not configured, exceptions will become + /// unobserved task exceptions following the default .NET behavior. + /// + public SubscriptionErrorHandler? ErrorHandler { get; init; } } diff --git a/src/Dapr.Messaging/PublishSubscribe/PublishSubscribeReceiver.cs b/src/Dapr.Messaging/PublishSubscribe/PublishSubscribeReceiver.cs index 4b0d608ff..4c9c316fd 100644 --- a/src/Dapr.Messaging/PublishSubscribe/PublishSubscribeReceiver.cs +++ b/src/Dapr.Messaging/PublishSubscribe/PublishSubscribeReceiver.cs @@ -12,6 +12,7 @@ // ------------------------------------------------------------------------ using System.Threading.Channels; +using Dapr; using Dapr.AppCallback.Autogen.Grpc.v1; using Grpc.Core; using P = Dapr.Client.Autogen.Grpc.v1; @@ -118,18 +119,37 @@ internal async Task SubscribeAsync(CancellationToken cancellationToken = default return; } - var stream = await GetStreamAsync(cancellationToken); + AsyncDuplexStreamingCall stream; + + try + { + stream = await GetStreamAsync(cancellationToken); + } + catch (RpcException ex) + { + // Reset the flag so callers can retry after the sidecar becomes available + Interlocked.Exchange(ref hasInitialized, 0); + throw new DaprException( + $"Failed to establish a subscription stream for topic '{topicName}' on pubsub '{pubSubName}'. Ensure the Dapr sidecar is available.", + ex); + } + catch (Exception) + { + // Reset the flag so callers can retry + Interlocked.Exchange(ref hasInitialized, 0); + throw; + } //Retrieve the messages from the sidecar and write to the messages channel - start without awaiting so this isn't blocking _ = FetchDataFromSidecarAsync(stream, topicMessagesChannel.Writer, cancellationToken) - .ContinueWith(HandleTaskCompletion, null, cancellationToken, TaskContinuationOptions.OnlyOnFaulted, + .ContinueWith(task => HandleTaskCompletion(task), CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default); //Process the messages as they're written to either channel - _ = ProcessAcknowledgementChannelMessagesAsync(stream, cancellationToken).ContinueWith(HandleTaskCompletion, - null, cancellationToken, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default); - _ = ProcessTopicChannelMessagesAsync(cancellationToken).ContinueWith(HandleTaskCompletion, null, - cancellationToken, + _ = ProcessAcknowledgementChannelMessagesAsync(stream, cancellationToken).ContinueWith( + task => HandleTaskCompletion(task), CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default); + _ = ProcessTopicChannelMessagesAsync(cancellationToken).ContinueWith( + task => HandleTaskCompletion(task), CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default); } @@ -149,11 +169,33 @@ internal async Task WriteAcknowledgementToChannelAsync(TopicAcknowledgement ackn } //Exposed for testing purposes only - internal static void HandleTaskCompletion(Task task, object? state) + internal void HandleTaskCompletion(Task task) { - if (task.Exception != null) + if (task.Exception is null) + { + return; + } + + var daprException = new DaprException( + $"A background task faulted for the subscription on topic '{topicName}' on pubsub '{pubSubName}'.", + task.Exception.GetBaseException()); + + if (options.ErrorHandler is not null) + { + try + { + // Fire-and-forget the error handler; suppress its exceptions to avoid cascading unobserved task exceptions + _ = options.ErrorHandler(daprException).ContinueWith(static _ => { }, CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default); + } + catch + { + // Suppress synchronous exceptions from the error handler invocation + } + } + else { - throw task.Exception; + throw daprException; } } diff --git a/test/Dapr.Messaging.Test/PublishSubscribe/PublishSubscribeReceiverTests.cs b/test/Dapr.Messaging.Test/PublishSubscribe/PublishSubscribeReceiverTests.cs index d266ff0cb..2e4228116 100644 --- a/test/Dapr.Messaging.Test/PublishSubscribe/PublishSubscribeReceiverTests.cs +++ b/test/Dapr.Messaging.Test/PublishSubscribe/PublishSubscribeReceiverTests.cs @@ -12,6 +12,7 @@ // ------------------------------------------------------------------------ using System.Threading.Channels; +using Dapr; using Dapr.AppCallback.Autogen.Grpc.v1; using Dapr.Messaging.PublishSubscribe; using Grpc.Core; @@ -197,8 +198,13 @@ public void HandleTaskCompletion_ShouldThrowException_WhenTaskHasException() { var task = Task.FromException(new InvalidOperationException("Test exception")); - var exception = Assert.Throws(() => - PublishSubscribeReceiver.HandleTaskCompletion(task, null)); + var options = new DaprSubscriptionOptions(new MessageHandlingPolicy(TimeSpan.FromSeconds(5), TopicResponseAction.Success)); + var mockDaprClient = new Mock(); + var receiver = new PublishSubscribeReceiver("testPubSub", "testTopic", options, + (_, _) => Task.FromResult(TopicResponseAction.Success), mockDaprClient.Object); + + var exception = Assert.Throws(() => + receiver.HandleTaskCompletion(task)); Assert.IsType(exception.InnerException); Assert.Equal("Test exception", exception.InnerException.Message); @@ -208,8 +214,105 @@ public void HandleTaskCompletion_ShouldThrowException_WhenTaskHasException() public void HandleTaskCompletion_SuccessfulTask_DoesNotThrow() { var task = Task.CompletedTask; + var options = new DaprSubscriptionOptions(new MessageHandlingPolicy(TimeSpan.FromSeconds(5), TopicResponseAction.Success)); + var mockDaprClient = new Mock(); + var receiver = new PublishSubscribeReceiver("testPubSub", "testTopic", options, + (_, _) => Task.FromResult(TopicResponseAction.Success), mockDaprClient.Object); + // Should not throw for a completed (non-faulted) task - PublishSubscribeReceiver.HandleTaskCompletion(task, null); + receiver.HandleTaskCompletion(task); + } + + [Fact] + public async Task SubscribeAsync_ShouldThrowDaprException_WhenSidecarUnavailable() + { + const string pubSubName = "testPubSub"; + const string topicName = "testTopic"; + var options = new DaprSubscriptionOptions( + new MessageHandlingPolicy(TimeSpan.FromSeconds(5), TopicResponseAction.Success)); + + var mockDaprClient = new Mock(); + mockDaprClient.Setup(c => c.SubscribeTopicEventsAlpha1(null, null, It.IsAny())) + .Throws(new RpcException(new Status(StatusCode.Unavailable, "Connection refused"))); + + var receiver = new PublishSubscribeReceiver(pubSubName, topicName, options, + (_, _) => Task.FromResult(TopicResponseAction.Success), mockDaprClient.Object); + + var ex = await Assert.ThrowsAsync(() => receiver.SubscribeAsync(CancellationToken.None)); + Assert.IsType(ex.InnerException); + Assert.Contains("testTopic", ex.Message); + Assert.Contains("testPubSub", ex.Message); + } + + [Fact] + public async Task SubscribeAsync_ShouldAllowRetry_AfterSidecarFailure() + { + const string pubSubName = "testPubSub"; + const string topicName = "testTopic"; + var options = new DaprSubscriptionOptions( + new MessageHandlingPolicy(TimeSpan.FromSeconds(5), TopicResponseAction.Success)); + + var mockDaprClient = new Mock(); + var callCount = 0; + + var mockRequestStream = new Mock>(); + var mockResponseStream = new Mock>(); + var mockCall = new AsyncDuplexStreamingCall( + mockRequestStream.Object, mockResponseStream.Object, + Task.FromResult(new Metadata()), () => new Status(), () => new Metadata(), () => { }); + + mockDaprClient.Setup(c => c.SubscribeTopicEventsAlpha1(null, null, It.IsAny())) + .Returns(() => + { + callCount++; + if (callCount == 1) + throw new RpcException(new Status(StatusCode.Unavailable, "Connection refused")); + return mockCall; + }); + + var receiver = new PublishSubscribeReceiver(pubSubName, topicName, options, + (_, _) => Task.FromResult(TopicResponseAction.Success), mockDaprClient.Object); + + // First call fails + await Assert.ThrowsAsync(() => receiver.SubscribeAsync(CancellationToken.None)); + + // Second call should succeed (hasInitialized was reset) + await receiver.SubscribeAsync(CancellationToken.None); + Assert.Equal(2, callCount); + } + + [Fact] + public async Task HandleTaskCompletion_ShouldInvokeErrorHandler_WhenConfigured() + { + DaprException? capturedEx = null; + var tcs = new TaskCompletionSource(); + var options = new DaprSubscriptionOptions( + new MessageHandlingPolicy(TimeSpan.FromSeconds(5), TopicResponseAction.Success)) + { + ErrorHandler = ex => + { + capturedEx = ex; + tcs.SetResult(); + return Task.CompletedTask; + } + }; + + var mockDaprClient = new Mock(); + var receiver = new PublishSubscribeReceiver("testPubSub", "testTopic", options, + (_, _) => Task.FromResult(TopicResponseAction.Success), mockDaprClient.Object); + + var faultedTask = Task.FromException(new InvalidOperationException("background error")); + + // Should not throw when error handler is provided + receiver.HandleTaskCompletion(faultedTask); + + // Wait for the error handler to be invoked + await tcs.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + Assert.NotNull(capturedEx); + Assert.IsType(capturedEx.InnerException); + Assert.Contains("background error", capturedEx.InnerException!.Message); + Assert.Contains("testTopic", capturedEx.Message); } [Fact] @@ -656,8 +759,8 @@ public async Task AcknowledgeMessageAsync_UnrecognisedAction_FaultsProcessingTas // Verify HandleTaskCompletion correctly re-throws when given the faulted task. var faultedStub = Task.FromException(new InvalidOperationException("Unrecognized topic acknowledgement action: 99")); - var ex = Assert.Throws(() => - PublishSubscribeReceiver.HandleTaskCompletion(faultedStub, null)); + var ex = Assert.Throws(() => + receiver.HandleTaskCompletion(faultedStub)); Assert.IsType(ex.InnerException); Assert.Contains("99", ex.InnerException!.Message); From a3927bb77a1feaa95c19a7c89237a5367711a645 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 3 May 2026 00:39:30 +0000 Subject: [PATCH 2/2] docs: improve error handler documentation based on code review feedback Agent-Logs-Url: https://github.com/dapr/dotnet-sdk/sessions/43bcaf39-8a3f-49c1-91df-f57b43a700bb Co-authored-by: WhitWaldo <2238529+WhitWaldo@users.noreply.github.com> --- .../PublishSubscribe/DaprSubscriptionOptions.cs | 4 +++- .../PublishSubscribe/PublishSubscribeReceiver.cs | 1 + 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs b/src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs index 403fe0e1b..51fba6a6c 100644 --- a/src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs +++ b/src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs @@ -53,7 +53,9 @@ public sealed record DaprSubscriptionOptions(MessageHandlingPolicy MessageHandli /// /// An optional error handler that is invoked when background tasks (sidecar streaming, acknowledgement processing, /// message processing) fault after a successful initial subscription. If not configured, exceptions will become - /// unobserved task exceptions following the default .NET behavior. + /// unobserved task exceptions following the default .NET behavior, which may terminate the application depending on + /// runtime configuration. If the error handler itself throws, those exceptions are suppressed to prevent cascading + /// failures. /// public SubscriptionErrorHandler? ErrorHandler { get; init; } } diff --git a/src/Dapr.Messaging/PublishSubscribe/PublishSubscribeReceiver.cs b/src/Dapr.Messaging/PublishSubscribe/PublishSubscribeReceiver.cs index 4c9c316fd..8d22c2041 100644 --- a/src/Dapr.Messaging/PublishSubscribe/PublishSubscribeReceiver.cs +++ b/src/Dapr.Messaging/PublishSubscribe/PublishSubscribeReceiver.cs @@ -141,6 +141,7 @@ internal async Task SubscribeAsync(CancellationToken cancellationToken = default } //Retrieve the messages from the sidecar and write to the messages channel - start without awaiting so this isn't blocking + // CancellationToken.None is used for continuations to ensure error handling runs even during disposal _ = FetchDataFromSidecarAsync(stream, topicMessagesChannel.Writer, cancellationToken) .ContinueWith(task => HandleTaskCompletion(task), CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default);