Restore OnError routing for NPC-sourced WhenPropertyChanged emissions

src/DynamicData.Tests/Binding/WhenPropertyChangedBehaviorFixture.cs

@@ -154,6 +154,109 @@ public void DeepChain_MidChainSwap_DeeperLevelsRetargetCorrectly()
         emissions.Should().Equal(new[] { 10, 20, 30 }, "leaf event on detached subtree is ignored");
     }
 
+    [Fact]
+    public void Shallow_ObserverThrowsInOnNext_RoutedToOnError_DoesNotEscapeSetter()
+    {
+        // Emissions originate from an INotifyPropertyChanged callback. A subscriber whose OnNext
+        // throws must not surface that exception raw out of the property setter; it is routed
+        // through OnError so a subscriber that supplied an error handler can observe it, and the
+        // setter that raised PropertyChanged stays safe.
+        var model = new TestModel { Value = 10 };
+
+        using var sub = model.WhenPropertyChanged(m => m.Value, notifyOnInitialValue: false)
+            .Subscribe(_ => throw new InvalidOperationException("boom"));
+
+        var setter = () => model.Value = 20;
+        var laterSetter = () => model.Value = 30;
+
+        setter.Should().NotThrow("a PropertyChanged-sourced OnNext throw is routed to OnError, not escaped to the setter");
+        laterSetter.Should().NotThrow("the terminated subscription must not resurface the exception on a later setter");
+    }
+
+    [Fact]
+    public void Shallow_AccessorThrows_ErrorHandlerRethrows_PropagatesToSetter()
+    {
+        // We route accessor/OnNext failures to OnError but do NOT swallow a throw from OnError
+        // itself: an unhandled downstream error (here a rethrowing error handler) propagates to
+        // the setter, matching Subject<T>.OnNext semantics. Locks in the non-swallow decision.
+        var model = new ThrowingGetterModel { Value = 10 };
+
+        using var sub = model.WhenPropertyChanged(m => m.Value, notifyOnInitialValue: false)
+            .Subscribe(_ => { }, _ => throw new InvalidOperationException("onerror"));
+
+        model.ThrowOnGet = true;
+        var setter = () => model.Value = 20;
+
+        setter.Should().Throw<InvalidOperationException>()
+            .WithMessage("onerror", "a rethrowing error handler must not be swallowed");
+    }
+
+    [Fact]
+    public void DeepChain_ObserverThrowsInOnNext_RoutedToOnError_DoesNotEscapeSetter()
+    {
+        var parent = new ParentModel { Child = new ChildModel { Age = 10 } };
+
+        using var sub = parent.WhenPropertyChanged(p => p.Child!.Age, notifyOnInitialValue: false)
+            .Subscribe(_ => throw new InvalidOperationException("boom"));
+
+        var setter = () => parent.Child!.Age = 20;
+        var laterSetter = () => parent.Child!.Age = 30;
+
+        setter.Should().NotThrow("a PropertyChanged-sourced OnNext throw is routed to OnError, not escaped to the setter");
+        laterSetter.Should().NotThrow("the terminated subscription must not resurface the exception on a later setter");
+    }
+
+    [Fact]
+    public void DeepChain_AccessorThrows_ErrorHandlerRethrows_PropagatesToSetter()
+    {
+        var parent = new ThrowingLeafParent { Child = new ThrowingGetterModel { Value = 10 } };
+
+        using var sub = parent.WhenPropertyChanged(p => p.Child!.Value, notifyOnInitialValue: false)
+            .Subscribe(_ => { }, _ => throw new InvalidOperationException("onerror"));
+
+        parent.Child!.ThrowOnGet = true;
+        var setter = () => parent.Child!.Value = 20;
+
+        setter.Should().Throw<InvalidOperationException>()
+            .WithMessage("onerror", "a rethrowing error handler must not be swallowed");
+    }
+
+    private sealed class ThrowingGetterModel : INotifyPropertyChanged
+    {
+        private int _value;
+
+        public event PropertyChangedEventHandler? PropertyChanged;
+
+        public bool ThrowOnGet { get; set; }
+
+        public int Value
+        {
+            get => ThrowOnGet ? throw new InvalidOperationException("getter") : _value;
+            set
+            {
+                _value = value;
+                PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Value)));
+            }
+        }
+    }
+
+    private sealed class ThrowingLeafParent : INotifyPropertyChanged
+    {
+        private ThrowingGetterModel? _child;
+
+        public event PropertyChangedEventHandler? PropertyChanged;
+
+        public ThrowingGetterModel? Child
+        {
+            get => _child;
+            set
+            {
+                _child = value;
+                PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(Child)));
+            }
+        }
+    }
+
     private sealed class TestModel : INotifyPropertyChanged
     {
         private int _value;

src/DynamicData/Binding/ObservablePropertyFactory.cs

@@ -89,25 +89,25 @@ private void OnPropertyChanged(object? sender, PropertyChangedEventArgs args)
             }
         }
 
-        // Reads the current property value and forwards it through the queue. The accessor is
-        // user code and may throw; that exception routes to OnError. The downstream OnNext call
-        // is NOT wrapped: per the Rx contract, if the user observer throws, the exception
-        // propagates back to whoever invoked the PropertyChanged setter, matching what a plain
-        // Subject<T>.OnNext would do.
+        // Wraps both the accessor read AND the queue OnNext (which may invoke the downstream
+        // observer synchronously via the DeliveryQueue drain) so that exceptions from either are
+        // routed through OnError rather than escaping raw into the PropertyChanged setter that
+        // fired the event. These emissions originate from an INotifyPropertyChanged callback, so a
+        // raw escaping exception would bypass the subscriber's error handler and surface out of the
+        // property setter. Routing through OnError gives a subscriber that supplied an onError
+        // handler the chance to observe and absorb the failure. We deliberately do NOT swallow a
+        // throw from OnError itself: if the downstream error is unhandled (or the error handler
+        // rethrows) it propagates to the setter, matching Subject<T>.OnNext semantics.
         private void EmitCurrent()
         {
-            PropertyValue<TObject, TProperty> value;
             try
             {
-                value = new PropertyValue<TObject, TProperty>(_source, _accessor(_source));
+                _queue.OnNext(new PropertyValue<TObject, TProperty>(_source, _accessor(_source)));
             }
             catch (Exception ex)
             {
                 _queue.OnError(ex);
-                return;
             }
-
-            _queue.OnNext(value);
         }
     }
 
@@ -208,35 +208,31 @@ public void Dispose()
 
         private void ProcessSignal(int level)
         {
-            // Drainer thread. The chain walk (Invoker / notifier Factory / ReadCurrent's accessor)
-            // is user code and may throw; those exceptions route to OnError. The downstream
-            // OnNext call is NOT wrapped: per the Rx contract, if the user observer throws, the
-            // exception propagates back through the drainer, matching what a plain Subject<T>
-            // would do.
+            // Drainer thread. Wraps the entire signal processing so that exceptions from any
+            // user-provided invoker, notifier factory, value accessor, or downstream observer are
+            // routed through OnError rather than escaping raw into the SharedDeliveryQueue drainer
+            // (and ultimately the INotifyPropertyChanged setter that raised the event). Routing
+            // through OnError gives a subscriber that supplied an onError handler the chance to
+            // observe and absorb the failure. We deliberately do NOT swallow a throw from OnError
+            // itself: an unhandled downstream error (or a rethrowing error handler) propagates,
+            // matching Subject<T>.OnNext semantics.
             //
             // The two cases (initial setup vs level-fire) collapse to:
             //   startLevel = (initial) ? 0 : level + 1
             //   emit       = (level-fire) || _notifyInitial
-            var isInitial = level == InitialSetupSignal;
-            var shouldEmit = !isInitial || _notifyInitial;
-            PropertyValue<TObject, TProperty> value;
             try
             {
+                var isInitial = level == InitialSetupSignal;
                 ResubscribeFrom(isInitial ? 0 : level + 1);
-                if (!shouldEmit)
+                if (!isInitial || _notifyInitial)
                 {
-                    return;
+                    _userSub.OnNext(ReadCurrent());
                 }
-
-                value = ReadCurrent();
             }
             catch (Exception ex)
             {
                 _userSub.OnError(ex);
-                return;
             }
-
-            _userSub.OnNext(value);
         }
 
         private void ResubscribeFrom(int startLevel)