Fix FilterImmutable Update->Remove to carry the previous value as Current (#1113)

dwcullop · web-flow · commit 6d2144c32d03 · 2026-06-16T22:27:20.000-05:00
When an Update changeset entry has Previous matching the predicate and
Current not matching, FilterImmutable emits a Remove. Previously this
Remove carried the new (non-matching) value as Current, violating the
Change&lt;T,K&gt; contract that Remove.Current is the item being removed (the
item that just left downstream). Since the new value never reached
downstream, only the previous value can satisfy this contract.

Consumers that read Current on Remove (e.g. composition with
TransformImmutable, side-effect handlers like DisposeMany or
OnItemRemoved equivalents) received the wrong reference, silently
producing incorrect results or InvalidCastException.
diff --git a/src/DynamicData.Tests/Cache/FilterImmutableFixture.cs b/src/DynamicData.Tests/Cache/FilterImmutableFixture.cs
@@ -4,6 +4,8 @@
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
 
+using DynamicData.Tests.Utilities;
+
 using FluentAssertions;
 using Xunit;
 
@@ -16,9 +18,10 @@ public void ItemsAreManipulated_UnmatchedItemsAreExcludedAndIndexesAreDiscarded(
     {
         using var source = new Subject<IChangeSet<Item, int>>();
 
-        using var results = source
+        using var subscription = source
             .FilterImmutable(predicate: Item.Predicate)
-            .AsAggregator();
+            .ValidateChangeSets(Item.KeySelector)
+            .RecordCacheItems(out var results);
 
 
         // Add items
@@ -31,8 +34,8 @@ public void ItemsAreManipulated_UnmatchedItemsAreExcludedAndIndexesAreDiscarded(
         });
 
         results.Error.Should().BeNull();
-        results.Messages.Count.Should().Be(1, "1 source operation was performed");
-        results.Data.Items.Should().BeEquivalentTo(new[] { item1 }, "2 items were added, with 1 excluded");
+        results.RecordedChangeSets.Count.Should().Be(1, "1 source operation was performed");
+        results.RecordedItemsByKey.Values.Should().BeEquivalentTo(new[] { item1 }, "2 items were added, with 1 excluded");
 
 
         // Replace items, changing inclusion
@@ -45,8 +48,8 @@ public void ItemsAreManipulated_UnmatchedItemsAreExcludedAndIndexesAreDiscarded(
         });
 
         results.Error.Should().BeNull();
-        results.Messages.Skip(1).Count().Should().Be(1, "1 source operation was performed");
-        results.Data.Items.Should().BeEquivalentTo(new[] { item4 }, "2 items were replaced, with 1 excluded");
+        results.RecordedChangeSets.Skip(1).Count().Should().Be(1, "1 source operation was performed");
+        results.RecordedItemsByKey.Values.Should().BeEquivalentTo(new[] { item4 }, "2 items were replaced, with 1 excluded");
 
 
         // Replace items, not changing inclusion
@@ -59,8 +62,8 @@ public void ItemsAreManipulated_UnmatchedItemsAreExcludedAndIndexesAreDiscarded(
         });
 
         results.Error.Should().BeNull();
-        results.Messages.Skip(2).Count().Should().Be(1, "1 source operation was performed");
-        results.Data.Items.Should().BeEquivalentTo(new[] { item6 }, "2 items were replaced, with 1 excluded");
+        results.RecordedChangeSets.Skip(2).Count().Should().Be(1, "1 source operation was performed");
+        results.RecordedItemsByKey.Values.Should().BeEquivalentTo(new[] { item6 }, "2 items were replaced, with 1 excluded");
 
 
         // Refresh items
@@ -71,8 +74,8 @@ public void ItemsAreManipulated_UnmatchedItemsAreExcludedAndIndexesAreDiscarded(
         });
 
         results.Error.Should().BeNull();
-        results.Messages.Skip(3).Count().Should().Be(1, "1 source operation was performed");
-        results.Data.Items.Should().BeEquivalentTo(new[] { item6 }, "2 items were refreshed, with 1 excluded");
+        results.RecordedChangeSets.Skip(3).Count().Should().Be(1, "1 source operation was performed");
+        results.RecordedItemsByKey.Values.Should().BeEquivalentTo(new[] { item6 }, "2 items were refreshed, with 1 excluded");
 
 
         // Remove items
@@ -83,28 +86,29 @@ public void ItemsAreManipulated_UnmatchedItemsAreExcludedAndIndexesAreDiscarded(
         });
 
         results.Error.Should().BeNull();
-        results.Messages.Skip(4).Count().Should().Be(1, "1 source operation was performed");
-        results.Data.Items.Should().BeEmpty("2 items were removed, with one excluded");
+        results.RecordedChangeSets.Skip(4).Count().Should().Be(1, "1 source operation was performed");
+        results.RecordedItemsByKey.Should().BeEmpty("2 items were removed, with one excluded");
 
 
-        results.Messages.SelectMany(static changes => changes).Should().AllSatisfy(
+        results.RecordedChangeSets.SelectMany(static changes => changes).Should().AllSatisfy(
             change =>
             {
                 change.CurrentIndex.Should().Be(-1);
                 change.PreviousIndex.Should().Be(-1);
             },
             because: "indexes should not be preserved");
-        results.IsCompleted.Should().BeFalse();
+        results.HasCompleted.Should().BeFalse();
     }
 
     [Fact]
     public void ItemsAreMoved_ChangesAreNotPropagated()
     {
         using var source = new Subject<IChangeSet<Item, int>>();
 
-        using var results = source
+        using var subscription = source
             .FilterImmutable(predicate: Item.Predicate)
-            .AsAggregator();
+            .ValidateChangeSets(Item.KeySelector)
+            .RecordCacheItems(out var results);
 
         // Initial setup
         var item1 = new Item() { Id = 1, IsIncluded = true };
@@ -116,18 +120,18 @@ public void ItemsAreMoved_ChangesAreNotPropagated()
             new(reason: ChangeReason.Add, key: item2.Id, current: item2, index: 1),
             new(reason: ChangeReason.Add, key: item3.Id, current: item3, index: 2)
         });
-        results.Messages.Clear();
+        var changeSetsBeforeMove = results.RecordedChangeSets.Count;
 
 
         // Move items
         source.OnNext(new ChangeSet<Item, int>()
         {
             new(reason: ChangeReason.Moved, key: item1.Id, current: item1, previous: default, currentIndex: 2, previousIndex: 0),
-            new(reason: ChangeReason.Moved, key: item2.Id, current: item1, previous: default, currentIndex: 0, previousIndex: 1)
+            new(reason: ChangeReason.Moved, key: item2.Id, current: item2, previous: default, currentIndex: 0, previousIndex: 1)
         });
 
         results.Error.Should().BeNull();
-        results.Messages.Should().BeEmpty("move operations should not be propagated");
+        results.RecordedChangeSets.Skip(changeSetsBeforeMove).Should().BeEmpty("move operations should not be propagated");
     }
 
     [Fact]
@@ -144,9 +148,10 @@ public void PredicateThrows_ExceptionIsCaptured()
 
         var error = new Exception();
 
-        using var results = source
+        using var subscription = source
             .FilterImmutable(predicate: _ => throw error)
-            .AsAggregator();
+            .ValidateChangeSets(Item.KeySelector)
+            .RecordCacheItems(out var results);
 
 
         var item1 = new Item() { Id = 1, IsIncluded = true };
@@ -156,18 +161,19 @@ public void PredicateThrows_ExceptionIsCaptured()
         });
 
         results.Error.Should().Be(error);
-        results.Messages.Should().BeEmpty("no source operations should have been processed");
-        results.IsCompleted.Should().BeFalse();
+        results.RecordedChangeSets.Should().BeEmpty("no source operations should have been processed");
+        results.HasCompleted.Should().BeFalse();
     }
 
     [Fact]
     public void SourceCompletes_CompletionIsPropagated()
     {
         using var source = new Subject<IChangeSet<Item, int>>();
 
-        using var results = source
+        using var subscription = source
             .FilterImmutable(predicate: Item.Predicate)
-            .AsAggregator();
+            .ValidateChangeSets(Item.KeySelector)
+            .RecordCacheItems(out var results);
 
 
         var item1 = new Item() { Id = 1, IsIncluded = true };
@@ -178,19 +184,19 @@ public void SourceCompletes_CompletionIsPropagated()
         source.OnCompleted();
 
         results.Error.Should().BeNull();
-        results.IsCompleted.Should().BeTrue();
-        results.Messages.Count.Should().Be(1, "1 source operation was performed");
-        results.Data.Items.Should().BeEquivalentTo(new[] { item1 }, "1 item was added");
+        results.HasCompleted.Should().BeTrue();
+        results.RecordedChangeSets.Count.Should().Be(1, "1 source operation was performed");
+        results.RecordedItemsByKey.Values.Should().BeEquivalentTo(new[] { item1 }, "1 item was added");
+
 
-        
         // Make sure no extraneous notifications are published.
         var item2 = new Item() { Id = 2, IsIncluded = true };
         source.OnNext(new ChangeSet<Item, int>()
         {
             new(reason: ChangeReason.Add, key: item2.Id, current: item2)
         });
 
-        results.Messages.Skip(1).Should().BeEmpty("no source operations should have been processed");
+        results.RecordedChangeSets.Skip(1).Should().BeEmpty("no source operations should have been processed");
     }
 
     [Fact]
@@ -210,17 +216,16 @@ public void SourceCompletesImmediately_CompletionIsPropagated()
             return Disposable.Empty;
         });
 
-        var error = new Exception();
-
-        using var results = source
+        using var subscription = source
             .FilterImmutable(predicate: Item.Predicate)
-            .AsAggregator();
+            .ValidateChangeSets(Item.KeySelector)
+            .RecordCacheItems(out var results);
 
 
         results.Error.Should().BeNull();
-        results.IsCompleted.Should().BeTrue();
-        results.Messages.Count.Should().Be(1, "1 source operation was performed");
-        results.Data.Items.Should().BeEquivalentTo(new[] { item1 }, "1 item was added");
+        results.HasCompleted.Should().BeTrue();
+        results.RecordedChangeSets.Count.Should().Be(1, "1 source operation was performed");
+        results.RecordedItemsByKey.Values.Should().BeEquivalentTo(new[] { item1 }, "1 item was added");
     }
 
     [Fact]
@@ -230,9 +235,10 @@ public void SourceErrors_ErrorIsPropagated()
 
         var error = new Exception();
 
-        using var results = source
+        using var subscription = source
             .FilterImmutable(predicate: Item.Predicate)
-            .AsAggregator();
+            .ValidateChangeSets(Item.KeySelector)
+            .RecordCacheItems(out var results);
 
 
         var item1 = new Item() { Id = 1, IsIncluded = true };
@@ -243,19 +249,19 @@ public void SourceErrors_ErrorIsPropagated()
         source.OnError(error);
 
         results.Error.Should().Be(error);
-        results.IsCompleted.Should().BeFalse();
-        results.Messages.Count.Should().Be(1, "1 source operation was performed");
-        results.Data.Items.Should().BeEquivalentTo(new[] { item1 }, "1 item was added");
+        results.HasCompleted.Should().BeFalse();
+        results.RecordedChangeSets.Count.Should().Be(1, "1 source operation was performed");
+        results.RecordedItemsByKey.Values.Should().BeEquivalentTo(new[] { item1 }, "1 item was added");
+
 
-        
         // Make sure no extraneous notifications are published.
         var item2 = new Item() { Id = 2, IsIncluded = true };
         source.OnNext(new ChangeSet<Item, int>()
         {
             new(reason: ChangeReason.Add, key: item2.Id, current: item2)
         });
 
-        results.Messages.Skip(1).Should().BeEmpty("no source operations should have been processed");
+        results.RecordedChangeSets.Skip(1).Should().BeEmpty("no source operations should have been processed");
     }
 
     [Fact]
@@ -276,15 +282,16 @@ public void SourceErrorsImmediately_ErrorIsPropagated()
             return Disposable.Empty;
         });
 
-        using var results = source
+        using var subscription = source
             .FilterImmutable(predicate: Item.Predicate)
-            .AsAggregator();
+            .ValidateChangeSets(Item.KeySelector)
+            .RecordCacheItems(out var results);
 
 
         results.Error.Should().Be(error);
-        results.IsCompleted.Should().BeFalse();
-        results.Messages.Count.Should().Be(1, "1 source operation was performed");
-        results.Data.Items.Should().BeEquivalentTo(new[] { item1 }, "1 item was added");
+        results.HasCompleted.Should().BeFalse();
+        results.RecordedChangeSets.Count.Should().Be(1, "1 source operation was performed");
+        results.RecordedItemsByKey.Values.Should().BeEquivalentTo(new[] { item1 }, "1 item was added");
     }
 
     [Fact]
@@ -299,36 +306,38 @@ public void SuppressEmptyChangesetsIsFalse_EmptyChangesetsArePublished()
     {
         using var source = new Subject<IChangeSet<Item, int>>();
 
-        using var results = source
+        using var subscription = source
             .FilterImmutable(
                 predicate: Item.Predicate,
                 suppressEmptyChangeSets: false)
-            .AsAggregator();
+            .ValidateChangeSets(Item.KeySelector)
+            .RecordCacheItems(out var results);
 
 
         ManipulateExcludedItems(source);
 
         results.Error.Should().BeNull();
-        results.IsCompleted.Should().BeFalse();
-        results.Messages.Count.Should().Be(5, "5 source operations were performed");
-        results.Messages.Should().AllSatisfy(changes => changes.Should().BeEmpty(), "no included items were manipulated");
+        results.HasCompleted.Should().BeFalse();
+        results.RecordedChangeSets.Count.Should().Be(5, "5 source operations were performed");
+        results.RecordedChangeSets.Should().AllSatisfy(changes => changes.Should().BeEmpty(), "no included items were manipulated");
     }
 
     [Fact]
     public void SuppressEmptyChangesetsIsTrue_EmptyChangesetsAreNotPublished()
     {
         using var source = new Subject<IChangeSet<Item, int>>();
 
-        using var results = source
+        using var subscription = source
             .FilterImmutable(predicate: Item.Predicate)
-            .AsAggregator();
+            .ValidateChangeSets(Item.KeySelector)
+            .RecordCacheItems(out var results);
 
 
         ManipulateExcludedItems(source);
 
         results.Error.Should().BeNull();
-        results.IsCompleted.Should().BeFalse();
-        results.Messages.Should().BeEmpty("no source operations should have generated changes");
+        results.HasCompleted.Should().BeFalse();
+        results.RecordedChangeSets.Should().BeEmpty("no source operations should have generated changes");
     }
 
     private static void ManipulateExcludedItems(ISubject<IChangeSet<Item, int>> source)
@@ -364,6 +373,40 @@ private static void ManipulateExcludedItems(ISubject<IChangeSet<Item, int>> sour
         });
     }
 
+    [Fact]
+    public void Update_PreviousMatchedCurrentDoesNot_EmitsRemoveCarryingPreviousAsCurrent()
+    {
+        // Per Change<T,K> contract, a Remove change carries the item that was removed in Current.
+        // For an Update where Previous matched the predicate but Current does not, the item that
+        // leaves the filtered view is Previous (it was downstream; Current never reached downstream).
+        using var source = new Subject<IChangeSet<Item, int>>();
+
+        using var subscription = source
+            .FilterImmutable(predicate: Item.Predicate)
+            .ValidateChangeSets(Item.KeySelector)
+            .RecordCacheItems(out var results);
+
+        var included = new Item() { Id = 1, IsIncluded = true };
+        var excluded = new Item() { Id = 1, IsIncluded = false };
+
+        source.OnNext(new ChangeSet<Item, int>()
+        {
+            new(reason: ChangeReason.Add, key: included.Id, current: included, index: 0)
+        });
+
+        source.OnNext(new ChangeSet<Item, int>()
+        {
+            new(reason: ChangeReason.Update, key: excluded.Id, current: excluded, previous: included, currentIndex: 0, previousIndex: 0)
+        });
+
+        var lastChangeSet = results.RecordedChangeSets[results.RecordedChangeSets.Count - 1];
+        lastChangeSet.Count.Should().Be(1);
+
+        var removeChange = lastChangeSet.Single();
+        removeChange.Reason.Should().Be(ChangeReason.Remove);
+        removeChange.Current.Should().BeSameAs(included, "Remove.Current must carry the item that left downstream (the previously-matching value), not the new value that never reached downstream");
+    }
+
     private class Item
     {
         public static readonly Func<Item, int> KeySelector
diff --git a/src/DynamicData/Cache/Internal/FilterImmutable.cs b/src/DynamicData/Cache/Internal/FilterImmutable.cs
@@ -74,10 +74,16 @@ public IObservable<IChangeSet<TObject, TKey>> Run()
                             if (downstreamReason is { } reason)
                             {
                                 // Do not propagate indexes, we can't guarantee them to be correct, because we aren't caching items.
+                                //
+                                // For Update->Remove (Previous matched, Current does not), the item that leaves
+                                // downstream is Previous; Current never reached downstream. Per the Change<T,K>
+                                // contract, Remove.Current is the value being removed, so carry Previous here.
                                 downstreamChanges.Add(new(
                                     reason: reason,
                                     key: change.Key,
-                                    current: change.Current,
+                                    current: (reason is ChangeReason.Remove && change.Reason is ChangeReason.Update)
+                                        ? change.Previous.Value
+                                        : change.Current,
                                     previous: (reason is ChangeReason.Update)
                                         ? change.Previous
                                         : default));
