diff --git a/src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs b/src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs
index 73838b605..51fba6a6c 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,14 @@ 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, 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 4b0d608ff..8d22c2041 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,38 @@ 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
+ // 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;
}
}
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);