From 5a8dd758f8e754f9e9aa7b0a4974166a98e9f00a Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Sun, 26 Jul 2026 11:15:16 -0700 Subject: [PATCH] Report a source failure as a failure to deferred subscriptions A Connect() or Watch() made while notifications are suspended is deferred until the suspension lifts. When the cache's source failed, that deferral reported the failure as a successful completion, so the subscriber's error handling never ran and its data looked complete. Two things caused it. The suspension tracker was disposed on failure, which completes the subject that the deferral is waiting on, and the deferral mapped that completion straight onto observer.OnCompleted. Faulting the subject instead carries the exception, and dropping the Do lets the terminal event travel through the chain rather than bypassing it. SelectMany replaces Select followed by Switch. Take(1) means there is only ever one inner sequence, so the two are equivalent for delivery, but SelectMany propagates the gate's terminal event by itself. That leaves the deferral self-contained rather than dependent on the behaviour of another operator. --- .../SuspendNotificationsFixture.UnitTests.cs | 61 ++++++++++++++++++- src/DynamicData/Cache/ObservableCache.cs | 28 ++++++--- 2 files changed, 81 insertions(+), 8 deletions(-) diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs index ddc9d6c07..521fbebc3 100644 --- a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs @@ -1,7 +1,8 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; using System.Reactive.Linq; +using System.Reactive.Subjects; using System.Threading.Tasks; using FluentAssertions; @@ -453,6 +454,64 @@ public void ReSuspendThenResumeDeliversAllInSingleBatch() results.IsCompleted.Should().BeFalse(); } + [Fact] + public void OnErrorFiresIfCacheFailsWhileSuspended() + { + // A connection made while suspended is deferred, and must still be told when the source + // fails. Reporting the failure as a successful completion leaves the subscriber's error + // handling unrun and its data looking complete. + using var source = new Subject>(); + using var cache = new IntermediateCache(source); + + using var suspend = cache.SuspendNotifications(); + using var results = cache.Connect().AsAggregator(); + + var expectedError = new Exception("Test Exception"); + source.OnError(expectedError); + + results.Error.Should().Be(expectedError, "a connection deferred by a suspension should see the source fail"); + results.IsCompleted.Should().BeFalse("the source failed, it did not complete"); + } + + [Fact] + public void OnErrorFiresIfCacheFailsWhileWatchIsSuspended() + { + // Watch defers the same way Connect does, and has the same obligation. + using var source = new Subject>(); + using var cache = new IntermediateCache(source); + + using var suspend = cache.SuspendNotifications(); + Exception? actualError = null; + var isCompleted = false; + using var subscription = cache.Watch(1).Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true); + + var expectedError = new Exception("Test Exception"); + source.OnError(expectedError); + + actualError.Should().Be(expectedError, "a watch deferred by a suspension should see the source fail"); + isCompleted.Should().BeFalse("the source failed, it did not complete"); + } + + [Fact] + public void OnErrorFiresIfCacheFailsAfterResumingWhileConnectionWasSuspended() + { + // The deferred connection has activated by the time the failure arrives, so this covers + // the path through the connection itself rather than through the suspension gate. + using var source = new Subject>(); + using var cache = new IntermediateCache(source); + + var suspend = cache.SuspendNotifications(); + using var results = cache.Connect().AsAggregator(); + source.OnNext(new ChangeSet { new(ChangeReason.Add, 1, 1) }); + + suspend.Dispose(); + var expectedError = new Exception("Test Exception"); + source.OnError(expectedError); + + results.Error.Should().Be(expectedError, "an activated connection should still see the source fail"); + results.Data.Count.Should().Be(1, "the data written before the failure should have arrived"); + } + public void Dispose() { _source.Dispose(); diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index afce06cc6..60642ee29 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -120,9 +120,14 @@ public IObservable> Connect(Func? predi // Create the Connection Observable ? CreateConnectObservable(predicate, suppressEmptyChangeSets) - // Defer until notifications are no longer suspended - : _suspensionTracker.Value.NotificationsSuspendedObservable.Do(static _ => { }, observer.OnCompleted) - .Where(static b => !b).Take(1).Select(_ => CreateConnectObservable(predicate, suppressEmptyChangeSets)).Switch(); + // Defer until notifications are no longer suspended. Take(1) means there is only + // ever one inner sequence, so SelectMany carries the terminal event of the gate + // through on its own: the connection ends when the cache does, and fails when it + // fails, rather than reporting a failure as a successful completion. + : _suspensionTracker.Value.NotificationsSuspendedObservable + .Where(static areNotificationsSuspended => !areNotificationsSuspended) + .Take(1) + .SelectMany(_ => CreateConnectObservable(predicate, suppressEmptyChangeSets)); return observable.SubscribeSafe(observer); } @@ -144,9 +149,12 @@ public IObservable> Watch(TKey key) => // Create the Watch Observable ? CreateWatchObservable(key) - // Defer until notifications are no longer suspended - : _suspensionTracker.Value.NotificationsSuspendedObservable.Do(static _ => { }, observer.OnCompleted) - .Where(static b => !b).Take(1).Select(_ => CreateWatchObservable(key)).Switch(); + // Defer until notifications are no longer suspended. See Connect() for why + // SelectMany is used here. + : _suspensionTracker.Value.NotificationsSuspendedObservable + .Where(static areNotificationsSuspended => !areNotificationsSuspended) + .Take(1) + .SelectMany(_ => CreateWatchObservable(key)); return observable.SubscribeSafe(observer); } @@ -394,7 +402,7 @@ public void OnError(Exception error) if (cache._suspensionTracker.IsValueCreated) { - cache._suspensionTracker.Value.Dispose(); + cache._suspensionTracker.Value.Fault(error); } } @@ -517,6 +525,12 @@ public void EmitResumeNotification() } } + public void Fault(Exception error) + { + _areNotificationsSuspended.OnError(error); + _areNotificationsSuspended.Dispose(); + } + public void Dispose() { _areNotificationsSuspended.OnCompleted();