Fix that a source failure is reported as successful completion to deferred subscriptions

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<IChangeSet<int, int>>();
+            using var cache = new IntermediateCache<int, int>(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<IChangeSet<int, int>>();
+            using var cache = new IntermediateCache<int, int>(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<IChangeSet<int, int>>();
+            using var cache = new IntermediateCache<int, int>(source);
+
+            var suspend = cache.SuspendNotifications();
+            using var results = cache.Connect().AsAggregator();
+            source.OnNext(new ChangeSet<int, int> { 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();

src/DynamicData/Cache/ObservableCache.cs

@@ -120,9 +120,14 @@ public IObservable<IChangeSet<TObject, TKey>> Connect(Func<TObject, bool>? 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<Change<TObject, TKey>> 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();