diff --git a/src/DynamicData.Tests/List/OnItemRemovedFixture.cs b/src/DynamicData.Tests/List/OnItemRemovedFixture.cs
index 56bf05d05..6b91fc11d 100644
--- a/src/DynamicData.Tests/List/OnItemRemovedFixture.cs
+++ b/src/DynamicData.Tests/List/OnItemRemovedFixture.cs
@@ -1,7 +1,8 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
-
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
 using FluentAssertions;
 using Xunit;
 
@@ -11,6 +12,29 @@ namespace DynamicData.Tests.List;
 
 public class OnItemRemovedFixture
 {
+    // https://github.com/reactivemarbles/DynamicData/issues/1061
+    [Theory]
+    [InlineData(true)]
+    [InlineData(false)]
+    public void SubscriberDoesNotHandleErrors_ErrorBubblesUpstream(bool invokeOnUnsubscribe)
+    {
+        using var source = new TestSourceList<int>();
+
+        using var subscription = source.Connect()
+            .OnItemRemoved(
+                removeAction:           static _ => { },
+                invokeOnUnsubscribe:    invokeOnUnsubscribe)
+            .Subscribe();
+        
+        var error = new Exception("Test");
+
+        FluentActions.Invoking(() => source.SetError(error))
+            .Should()
+            .Throw<Exception>("errors not handled by the subscriber should propagate upstream to the caller")
+            .Which
+            .Should().BeSameAs(error);
+    }
+
     [Theory]
     [InlineData(0,  0,  0)]
     [InlineData(1,  0,  0)]
diff --git a/src/DynamicData/List/Internal/OnBeingRemoved.cs b/src/DynamicData/List/Internal/OnBeingRemoved.cs
deleted file mode 100644
index 48b8e01e9..000000000
--- a/src/DynamicData/List/Internal/OnBeingRemoved.cs
+++ /dev/null
@@ -1,61 +0,0 @@
-// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved.
-// Roland Pheasant licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for full license information.
-
-using System.Reactive.Disposables;
-using System.Reactive.Linq;
-
-namespace DynamicData.List.Internal;
-
-internal sealed class OnBeingRemoved<T>(IObservable<IChangeSet<T>> source, Action<T> callback, bool invokeOnUnsubscribe)
-    where T : notnull
-{
-    private readonly Action<T> _callback = callback ?? throw new ArgumentNullException(nameof(callback));
-    private readonly IObservable<IChangeSet<T>> _source = source ?? throw new ArgumentNullException(nameof(source));
-
-    public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
-            observer =>
-            {
-                var locker = InternalEx.NewLock();
-                var items = new List<T>();
-                var subscriber = _source.Synchronize(locker).Do(changes => RegisterForRemoval(items, changes), observer.OnError).SubscribeSafe(observer);
-
-                return Disposable.Create(
-                    () =>
-                    {
-                        subscriber.Dispose();
-
-                        if (invokeOnUnsubscribe)
-                        {
-                            items.ForEach(t => _callback(t));
-                        }
-                    });
-            });
-
-    private void RegisterForRemoval(IList<T> items, IChangeSet<T> changes)
-    {
-        foreach (var change in changes)
-        {
-            switch (change.Reason)
-            {
-                case ListChangeReason.Replace:
-                    change.Item.Previous.IfHasValue(t => _callback(t));
-                    break;
-
-                case ListChangeReason.Remove:
-                    _callback(change.Item.Current);
-                    break;
-
-                case ListChangeReason.RemoveRange:
-                    change.Range.ForEach(_callback);
-                    break;
-
-                case ListChangeReason.Clear:
-                    items.ForEach(_callback);
-                    break;
-            }
-        }
-
-        items.Clone(changes);
-    }
-}
diff --git a/src/DynamicData/List/Internal/OnItemRemoved.cs b/src/DynamicData/List/Internal/OnItemRemoved.cs
new file mode 100644
index 000000000..178066bf2
--- /dev/null
+++ b/src/DynamicData/List/Internal/OnItemRemoved.cs
@@ -0,0 +1,59 @@
+// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved.
+// Roland Pheasant licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for full license information.
+
+using System.Reactive.Linq;
+
+namespace DynamicData.List.Internal;
+
+internal static class OnItemRemoved<T>
+    where T : notnull
+{
+    public static IObservable<IChangeSet<T>> Create(
+        IObservable<IChangeSet<T>> source,
+        Action<T> removeAction,
+        bool invokeOnUnsubscribe)
+    {
+        source.ThrowArgumentNullExceptionIfNull(nameof(source));
+        removeAction.ThrowArgumentNullExceptionIfNull(nameof(removeAction));
+
+        var removalProcessor = source.Do(changeSet =>
+        {
+            foreach (var change in changeSet)
+            {
+                switch (change.Reason)
+                {
+                    case ListChangeReason.Clear:
+                    case ListChangeReason.RemoveRange:
+                        foreach (var item in change.Range)
+                            removeAction.Invoke(item);
+                        break;
+
+                    case ListChangeReason.Remove:
+                        removeAction.Invoke(change.Item.Current);
+                        break;
+
+                    case ListChangeReason.Replace:
+                        removeAction.Invoke(change.Item.Previous.Value);
+                        break;
+                }
+            }
+        });
+
+        return invokeOnUnsubscribe
+            ? Observable.Create<IChangeSet<T>>(observer =>
+            {
+                var items = new List<T>();
+
+                return removalProcessor
+                    .Do(changeSet => items.Clone(changeSet))
+                    .Finally(() =>
+                    {
+                        foreach (var item in items)
+                            removeAction.Invoke(item);
+                    })
+                    .SubscribeSafe(observer);
+            })
+            : removalProcessor;
+    }
+}
diff --git a/src/DynamicData/List/ObservableListEx.cs b/src/DynamicData/List/ObservableListEx.cs
index 92cf54e23..2f871bf4c 100644
--- a/src/DynamicData/List/ObservableListEx.cs
+++ b/src/DynamicData/List/ObservableListEx.cs
@@ -1276,26 +1276,24 @@ public static IObservable<IChangeSet<T>> OnItemRefreshed<T>(
             refreshAction: refreshAction);
 
     /// <summary>
-    /// Callback for each item as and when it is being removed from the stream.
+    /// Invokes a given action for every item removed from the source list stream.
     /// </summary>
-    /// <typeparam name="T">The type of the object.</typeparam>
-    /// <param name="source">The source.</param>
-    /// <param name="removeAction">The remove action.</param>
-    /// <param name="invokeOnUnsubscribe"> Should the remove action be invoked when the subscription is disposed.</param>
-    /// <returns>An observable which emits the change set.</returns>
-    /// <exception cref="ArgumentNullException">
-    /// source
-    /// or
-    /// removeAction.
-    /// </exception>
-    public static IObservable<IChangeSet<T>> OnItemRemoved<T>(this IObservable<IChangeSet<T>> source, Action<T> removeAction, bool invokeOnUnsubscribe = true)
-        where T : notnull
-    {
-        source.ThrowArgumentNullExceptionIfNull(nameof(source));
-        removeAction.ThrowArgumentNullExceptionIfNull(nameof(removeAction));
-
-        return new OnBeingRemoved<T>(source, removeAction, invokeOnUnsubscribe).Run();
-    }
+    /// <typeparam name="T">The type of items in the list.</typeparam>
+    /// <param name="source">The list stream whose items are to be passed to <paramref name="removeAction"/>.</param>
+    /// <param name="removeAction">The action to invoke upon each removed item.</param>
+    /// <param name="invokeOnUnsubscribe">A flag indicating whether <paramref name="removeAction"/> should be invoked upon teardown of the stream. This includes disposal of subscriptions, completion notifications, and error notifications.</param>
+    /// <returns>A list stream, containing all items in <paramref name="source"/>, with changes published after <paramref name="removeAction"/> has been invoked.</returns>
+    /// <exception cref="ArgumentNullException">Throws for <paramref name="source"/> and <paramref name="removeAction"/>.</exception>
+    /// <remarks>Note that "removed" items includes items from <see cref="ListChangeReason.Remove"/>, <see cref="ListChangeReason.RemoveRange"/>, <see cref="ListChangeReason.Replace"/>, and <see cref="ListChangeReason.Clear"/> changes.</remarks>
+    public static IObservable<IChangeSet<T>> OnItemRemoved<T>(
+                this IObservable<IChangeSet<T>> source,
+                Action<T> removeAction,
+                bool invokeOnUnsubscribe = true)
+            where T : notnull
+        => List.Internal.OnItemRemoved<T>.Create(
+            source: source,
+            removeAction: removeAction,
+            invokeOnUnsubscribe: invokeOnUnsubscribe);
 
     /// <summary>
     /// Apply a logical Or operator between the collections.