Fix background-task silent failures in PublishSubscribeReceiver

The initial fix for #1663 addressed the synchronous subscribe path but left a
near-identical silent-failure hole in the background continuation path:
HandleTaskCompletion was an async Task wired via _ = ContinueWith(...), so when
no ErrorHandler was configured its "throw daprException" faulted an unobserved
inner Task that never surfaced — exactly the bug the PR set out to fix.

Changes:

- HandleTaskCompletion no longer throws. It caches the first fault per subscribe
  cycle in a new pendingBackgroundFault field, which the next SubscribeAsync
  call drains and rethrows. Callers that skip the ErrorHandler still observe
  the error reliably rather than losing it to unobserved-task-exception void.

- A new hasFaulted dedupe flag ensures that when multiple sibling continuations
  (FetchDataFromSidecar / ProcessAcknowledgementChannel / ProcessTopicChannel)
  all fault on a single sidecar outage, only the first is reported.

- ResetStreamStateAsync now acquires the same semaphore used by GetStreamAsync,
  nulls clientStream before resetting hasInitialized, and disposes the prior
  stream — closing the race where a concurrent SubscribeAsync could observe a
  stale clientStream and leaking the gRPC streaming call on reset.

- OperationCanceledException handling now inspects every inner exception (via
  AggregateException.InnerExceptions.All(...)) instead of just the first, so a
  fault wrapping both an OCE and a real RpcException still surfaces the real
  error.

- When the user-supplied ErrorHandler itself throws, both the original fault
  and the handler failure are combined into an AggregateException and cached,
  replacing the previous empty catch that silently swallowed handler bugs.

- DaprSubscriptionOptions.ErrorHandler and SubscriptionErrorHandler XML docs
  updated to describe the new cache-and-rethrow contract accurately.

Tests:

Five new tests drive the real SubscribeAsync -> ContinueWith -> HandleTaskCompletion
chain (instead of awaiting HandleTaskCompletion directly) to reproduce each
scenario end-to-end:

  - NextSubscribeRethrows — verifies the no-handler cache path
  - InvokesErrorHandlerExactlyOnce — verifies hasFaulted dedupe
  - NextSubscribeRecreatesStream — verifies reset + retry works end-to-end
  - BackgroundFetchCancelled_DoesNotCachePendingFault — verifies OCE short-circuit
  - ErrorHandlerThrows_CachesCombinedFaultForNextSubscribe — verifies handler-fault capture

I verified these five reproduce the bug: after temporarily reverting the source
fix and re-running them, they all fail (with the exact "background fault was
not cached" / synchronous-throw symptoms the fix addresses).

Two pre-existing unit tests were adjusted to match the new contract — they now
exercise HandleTaskCompletion followed by SubscribeAsync to confirm the fault
is surfaced via the cache-and-rethrow path instead of via a synchronous throw.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Signed-off-by: George Tsiokos <george@tsiokos.com>

Author: GeorgeTsiokos

src/Dapr.Messaging/PublishSubscribe/DaprSubscriptionOptions.cs

@@ -44,7 +44,10 @@ public sealed record DaprSubscriptionOptions(MessageHandlingPolicy MessageHandli
     /// <summary>
     /// An optional callback invoked when errors occur in background tasks during an active subscription.
     /// Errors during the initial subscription attempt are thrown directly to the caller.
-    /// If not set, background task errors surface as unobserved task exceptions (default .NET behavior).
+    /// If not set, the first background fault per subscribe cycle is cached on the receiver and rethrown
+    /// on the next call to <c>SubscribeAsync</c>, so the caller can observe it and decide whether to retry.
+    /// If the handler itself throws, both the original fault and the handler failure are cached and
+    /// surfaced together as an <see cref="AggregateException"/> on the next subscribe attempt.
     /// </summary>
     public SubscriptionErrorHandler? ErrorHandler { get; init; }
 }

src/Dapr.Messaging/PublishSubscribe/PublishSubscribeReceiver.cs

@@ -74,6 +74,17 @@ internal sealed class PublishSubscribeReceiver : IAsyncDisposable
     /// </summary>
     private int hasInitialized;
     /// <summary>
+    /// Dedupes <see cref="HandleTaskCompletion"/> across the three background continuations so a single
+    /// sidecar failure surfaces exactly once per subscribe cycle. Reset at the start of each new cycle.
+    /// </summary>
+    private int hasFaulted;
+    /// <summary>
+    /// Stores the first unhandled background fault per subscribe cycle. Rethrown on the next call to
+    /// <see cref="SubscribeAsync"/> so a caller that did not configure an <see cref="DaprSubscriptionOptions.ErrorHandler"/>
+    /// still observes the error rather than losing it as an unobserved task exception.
+    /// </summary>
+    private Exception? pendingBackgroundFault;
+    /// <summary>
     /// Flag that ensures the instance is only disposed a single time.
     /// </summary>
     private bool isDisposed;
@@ -82,6 +93,9 @@ internal sealed class PublishSubscribeReceiver : IAsyncDisposable
     internal Task TopicMessagesChannelCompletion => topicMessagesChannel.Reader.Completion;
     // Internal property for testing purposes
     internal Task AcknowledgementsChannelCompletion => acknowledgementsChannel.Reader.Completion;
+    // Internal property for testing purposes — exposes whether a background fault has been cached
+    // and is waiting to be surfaced on the next SubscribeAsync call.
+    internal bool HasPendingBackgroundFault => Volatile.Read(ref pendingBackgroundFault) is not null;
 
     /// <summary>
     /// Constructs a new instance of a <see cref="PublishSubscribeReceiver"/> instance.
@@ -113,12 +127,23 @@ internal PublishSubscribeReceiver(string pubSubName, string topicName, DaprSubsc
     /// <returns>A task representing the asynchronous subscribe operation.</returns>
     internal async Task SubscribeAsync(CancellationToken cancellationToken = default)
     {
+        // Surface any unhandled background fault from a prior subscribe cycle so the caller
+        // can observe it explicitly before re-subscribing.
+        var carried = Interlocked.Exchange(ref pendingBackgroundFault, null);
+        if (carried is not null)
+        {
+            throw carried;
+        }
+
         //Prevents the receiver from performing the subscribe operation more than once (as the multiple initialization messages would cancel the stream).
         if (Interlocked.Exchange(ref hasInitialized, 1) == 1)
         {
             return;
         }
 
+        // Reset the per-cycle fault dedupe flag.
+        Interlocked.Exchange(ref hasFaulted, 0);
+
         AsyncDuplexStreamingCall<P.SubscribeTopicEventsRequestAlpha1, P.SubscribeTopicEventsResponseAlpha1> stream;
         try
         {
@@ -168,42 +193,93 @@ internal async Task WriteAcknowledgementToChannelAsync(TopicAcknowledgement ackn
     }
 
     /// <summary>
-    /// Handles faulted background tasks by resetting the initialization flag and invoking the configured error handler, if any.
+    /// Handles faulted background tasks by resetting the subscription state and either invoking the
+    /// configured error handler or caching the fault for the next <see cref="SubscribeAsync"/> call.
     /// </summary>
+    /// <remarks>
+    /// This method is the terminal continuation for <see cref="FetchDataFromSidecarAsync"/>,
+    /// <see cref="ProcessAcknowledgementChannelMessagesAsync"/>, and <see cref="ProcessTopicChannelMessagesAsync"/>.
+    /// It never re-throws: doing so would fault an unobserved inner <see cref="Task"/> (the original bug
+    /// this class was meant to fix). Instead, unhandled faults are stored in
+    /// <see cref="pendingBackgroundFault"/> and surfaced on the caller's next subscribe attempt.
+    /// </remarks>
     internal async Task HandleTaskCompletion(Task task, object? state)
     {
         if (task.Exception is null)
         {
             return;
         }
 
-        // Reset initialization flag so a future SubscribeAsync call can re-establish the subscription
-        Interlocked.Exchange(ref hasInitialized, 0);
-        clientStream = null;
+        // Dedupe: the three sibling continuations typically all fault when the sidecar dies.
+        // Only the first one is reported; the rest observe the dedupe flag and exit.
+        if (Interlocked.CompareExchange(ref hasFaulted, 1, 0) != 0)
+        {
+            return;
+        }
 
-        var innerException = task.Exception.InnerException ?? task.Exception;
+        var innerExceptions = task.Exception.InnerExceptions;
 
-        if (innerException is OperationCanceledException)
+        // If every inner exception is a cancellation, treat the fault as a clean cancel:
+        // reset the subscription state but do not cache or surface an error.
+        if (innerExceptions.All(e => e is OperationCanceledException))
         {
+            await ResetStreamStateAsync().ConfigureAwait(false);
             return;
         }
 
+        // Prefer a non-cancellation inner exception for the user-facing message.
+        var innerException = innerExceptions.FirstOrDefault(e => e is not OperationCanceledException)
+                             ?? task.Exception;
+
+        await ResetStreamStateAsync().ConfigureAwait(false);
+
         var daprException = new DaprException(
             $"An error occurred during an active subscription to topic '{topicName}' on pubsub '{pubSubName}'.",
             innerException);
 
         if (options.ErrorHandler is null)
         {
-            throw daprException;
+            // No handler configured: cache so the next SubscribeAsync surfaces it.
+            Interlocked.CompareExchange(ref pendingBackgroundFault, daprException, null);
+            return;
+        }
+
+        try
+        {
+            await options.ErrorHandler.Invoke(daprException).ConfigureAwait(false);
         }
+        catch (Exception handlerEx)
+        {
+            // User-supplied handler threw. Cache a combined fault so the next SubscribeAsync
+            // surfaces both the original error and the handler failure — never silently drop either.
+            var combined = new AggregateException(
+                $"The SubscriptionErrorHandler for topic '{topicName}' on pubsub '{pubSubName}' threw while handling a prior fault.",
+                daprException, handlerEx);
+            Interlocked.CompareExchange(ref pendingBackgroundFault, combined, null);
+        }
+    }
 
+    /// <summary>
+    /// Atomically tears down the active stream and resets the initialization flag.
+    /// </summary>
+    /// <remarks>
+    /// Serialized under <see cref="semaphore"/> (the same lock used by <see cref="GetStreamAsync"/>)
+    /// so a concurrent <see cref="SubscribeAsync"/> cannot observe <see cref="hasInitialized"/> reset
+    /// while <see cref="clientStream"/> still holds the stale reference.
+    /// </remarks>
+    private async Task ResetStreamStateAsync()
+    {
+        await semaphore.WaitAsync().ConfigureAwait(false);
         try
         {
-            await options.ErrorHandler.Invoke(daprException);
+            var old = clientStream;
+            clientStream = null;
+            Interlocked.Exchange(ref hasInitialized, 0);
+            old?.Dispose();
         }
-        catch (Exception)
+        finally
         {
-            // No logger available; prevent a faulty error handler from becoming an unobserved task exception
+            semaphore.Release();
         }
     }
 

src/Dapr.Messaging/PublishSubscribe/SubscriptionErrorHandler.cs

@@ -19,6 +19,7 @@ namespace Dapr.Messaging.PublishSubscribe;
 /// <param name="exception">The <see cref="DaprException"/> wrapping the original error.</param>
 /// <remarks>
 /// This handler is invoked on a thread pool thread. Implementations should be thread-safe.
-/// If the returned task faults, the exception will be suppressed.
+/// If the returned task faults, the handler's exception is combined with the original fault and
+/// surfaced as an <see cref="AggregateException"/> on the next call to <c>SubscribeAsync</c>.
 /// </remarks>
 public delegate Task SubscriptionErrorHandler(DaprException exception);