fix: SubscribeAsync throws DaprException when sidecar is unavailable

src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs

@@ -11,8 +11,17 @@
 // limitations under the License.
 // ------------------------------------------------------------------------
 
+using Dapr;
+
 namespace Dapr.Messaging.PublishSubscribe;
 
+/// <summary>
+/// A delegate that handles errors occurring during the subscription lifecycle after a successful initial connection.
+/// </summary>
+/// <param name="exception">The exception that occurred.</param>
+/// <returns>A task representing the asynchronous error handling operation.</returns>
+public delegate Task SubscriptionErrorHandler(DaprException exception);
+
 /// <summary>
 /// Options used to configure the dynamic Dapr subscription.
 /// </summary>
@@ -40,5 +49,14 @@ public sealed record DaprSubscriptionOptions(MessageHandlingPolicy MessageHandli
     /// been signaled.
     /// </summary>
     public TimeSpan MaximumCleanupTimeout { get; init; } = TimeSpan.FromSeconds(30);
+
+    /// <summary>
+    /// 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, which may terminate the application depending on
+    /// runtime configuration. If the error handler itself throws, those exceptions are suppressed to prevent cascading
+    /// failures.
+    /// </summary>
+    public SubscriptionErrorHandler? ErrorHandler { get; init; }
 }
 

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,38 @@ internal async Task SubscribeAsync(CancellationToken cancellationToken = default
             return;
         }
 
-        var stream = await GetStreamAsync(cancellationToken);
+        AsyncDuplexStreamingCall<P.SubscribeTopicEventsRequestAlpha1, P.SubscribeTopicEventsResponseAlpha1> 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
+        // CancellationToken.None is used for continuations to ensure error handling runs even during disposal
         _ = 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 +170,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;
         }
     }
 

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<AggregateException>(() =>
-            PublishSubscribeReceiver.HandleTaskCompletion(task, null));
+        var options = new DaprSubscriptionOptions(new MessageHandlingPolicy(TimeSpan.FromSeconds(5), TopicResponseAction.Success));
+        var mockDaprClient = new Mock<P.Dapr.DaprClient>();
+        var receiver = new PublishSubscribeReceiver("testPubSub", "testTopic", options,
+            (_, _) => Task.FromResult(TopicResponseAction.Success), mockDaprClient.Object);
+
+        var exception = Assert.Throws<DaprException>(() =>
+            receiver.HandleTaskCompletion(task));
 
         Assert.IsType<InvalidOperationException>(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<P.Dapr.DaprClient>();
+        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<P.Dapr.DaprClient>();
+        mockDaprClient.Setup(c => c.SubscribeTopicEventsAlpha1(null, null, It.IsAny<CancellationToken>()))
+            .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<DaprException>(() => receiver.SubscribeAsync(CancellationToken.None));
+        Assert.IsType<RpcException>(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<P.Dapr.DaprClient>();
+        var callCount = 0;
+
+        var mockRequestStream = new Mock<IClientStreamWriter<P.SubscribeTopicEventsRequestAlpha1>>();
+        var mockResponseStream = new Mock<IAsyncStreamReader<P.SubscribeTopicEventsResponseAlpha1>>();
+        var mockCall = new AsyncDuplexStreamingCall<P.SubscribeTopicEventsRequestAlpha1, P.SubscribeTopicEventsResponseAlpha1>(
+            mockRequestStream.Object, mockResponseStream.Object,
+            Task.FromResult(new Metadata()), () => new Status(), () => new Metadata(), () => { });
+
+        mockDaprClient.Setup(c => c.SubscribeTopicEventsAlpha1(null, null, It.IsAny<CancellationToken>()))
+            .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<DaprException>(() => 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<P.Dapr.DaprClient>();
+        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<InvalidOperationException>(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<AggregateException>(() =>
-            PublishSubscribeReceiver.HandleTaskCompletion(faultedStub, null));
+        var ex = Assert.Throws<DaprException>(() =>
+            receiver.HandleTaskCompletion(faultedStub));
         Assert.IsType<InvalidOperationException>(ex.InnerException);
         Assert.Contains("99", ex.InnerException!.Message);