[Bug]: `ArgumentException: An item with the same key was already added` with Filter on 9.4.31+

[mycroes]

Describe the bug 🐞

I ran into the following exception in my application:

System.ArgumentException: An item with the same key has already been added. Key: GroupId { Value = 2047 }
   at System.Collections.Generic.Dictionary`2.TryInsert(TKey key, TValue value, InsertionBehavior behavior)
   at System.Collections.Generic.Dictionary`2.Add(TKey key, TValue value)
   at DynamicData.ChangeAwareCache`2.Add(TObject item, TKey key) in /_/src/DynamicData/Cache/ChangeAwareCache.cs:line 75
   at DynamicData.Cache.Internal.Filter.Static`2.<>c__DisplayClass0_1.<Create>b__1(IChangeSet`2 upstreamChanges) in /_/src/DynamicData/Cache/Internal/Filter.Static.cs:line 36
   at System.Reactive.Linq.ObservableImpl.Select`2.Selector._.OnNext(TSource value)

After investigation it appears in DynamicData 9.4.31 and newer and only seems to happen when multiple SourceCaches, multiple subscriptions and multiple schedulers are applied.

Step to reproduce

using System.Reactive.Linq;
using DynamicData;
using Microsoft.Reactive.Testing;

namespace FilterTests;

public class FilterTests
{
    private static TestScheduler scheduler = new TestScheduler();

    [Fact]
    public async Task Filter_Throws_ArgumentException()
    {
        var viewModels = new SourceCache<GroupViewModel, int>(x => x.Group.Id);
        var groups = new SourceCache<Group, int>(x => x.Id);

        viewModels.Connect().Subscribe();

        groups.Connect()
            .Subscribe(changes => {
                viewModels.Edit(inner => {
                    foreach (var change in changes) {
                        switch (change.Reason)
                        {
                            case ChangeReason.Add:
                                inner.AddOrUpdate(new GroupViewModel(change.Current, viewModels));
                                break;
                        }
                    }
                });
            });

        groups.Edit(inner => inner.AddOrUpdate([
                new (1, 0),
                new (0, null)
            ]));

        scheduler.AdvanceBy(10);
    }

    private record Group(int Id, int? Parent);

    private class GroupViewModel
    {
        public Group Group { get; }

        public GroupViewModel(Group group, SourceCache<GroupViewModel, int> masterList)
        {
            Group = group;

            masterList
                .Connect()
                .Filter(n => n.Group.Parent == group.Id)
                .ObserveOn(scheduler)
                .Subscribe();
        }
    }
}

Reproduction repository

See above.

Expected behavior

Should not throw ArgumentException.

DynamicData Version

9.4.31

Additional information ℹ️

Took me over 8 hours to get this reproduced, I hope you guys fix this once and for all.

[mycroes]

I did my best at creating the minimal reproduction, it seems any line I delete from this will make the error go away. Rest assured this doesn't even cover 10% of my production code.

[JakenVeina]

Boy, you found a fun one.

This appears to be a long-standing defect. It's been around since at least 2020, which is the last time the relevant SourceCache<> code was touched. Beyond that, I don't know, cause the solution doesn't even build anymore, when rolling back that far.

The reason this is popping up only now is becaues .Filter() was rebuilt in 9.4.31, for performance, and incidentally, the new version now throws an exception when receiving the particular type of corrupt changeset involved here (where the operator is instructed to add the same item twice). I.E. the old version called .AddOrUpdate() internally for each new item, while the new one calls .Add().

The actual problem is that SourceCache<> is publishing bogus changesets in your highly-specific scenario of constructing multiple subscriptions to a source collection, within an edit operation upon itself. Here's the simplified reproduction...

using var items = new SourceCache<int, int>(static item => item);

using var subscriptions = new CompositeDisposable();

subscriptions.Add(items
    .Connect()
    .Subscribe());
    
items.Edit(inner =>
{
    inner.AddOrUpdate(1);

    subscriptions.Add(items
        .Connect()
        .Do(changes =>
        {
            _output.WriteLine($"{changes.Count} changes:");
            foreach (var change in changes)
                _output.WriteLine($"\t{change}");
        })
        .ValidateChangeSets(static item => item)
        .Subscribe());

    inner.AddOrUpdate(2);
});

Looking at the console output, we can see the issue directly:

1 changes:
	Add, Key: 1, Current: 1, Previous: <None>
2 changes:
	Add, Key: 1, Current: 1, Previous: <None>
	Add, Key: 2, Current: 2, Previous: <None>

When a subscription occurs upon a source collection, the observer immediately receives an initial changeset, containing the current items within the collection. This happens synchronously, so it's actually happening INSIDE an active .Edit() call. Because we're inside an .Edit(), any changes that occur before the new subscription get buffered (but only when there's already another active subscription). So, those items added within the .Edit() before the subscription get published to the subscription both in the initial changeset, and then again when the .Edit() finishes, simply because they're in the buffer.

Simple enough fix in concept: subscriptions should not receive their initial changeset if there's an .Edit() in progress. That process should be deferred until the .Edit() is complete. Might be tricky to implement, but it should be doable.

Thanks for the detailed report. You're right, just about every single line in your snippet was necessary to bring this one out. Really, the only superfluous thing is the scheduler.

[mycroes]

Thanks for checking in. I believe that without the scheduler it would still succeed, but like I said, I had been at it for over 8 hours (in multple sessions), so maybe I was getting delusional. I think same is true for order of the different blocks, if I would change whatever, it would suddenly succeed again. In my application the data is a combination of cached data from Akavache and data fetched from GitLab, but my API key had expired so I knew it was only the cache. When I finally had a working reproduction using my production code and swapped out the Sqlite cache for InMemoryBlobCache it also suddenly wouldn't throw anymore, this bug really got under my skin.

Anyway, I hope this will get battle-tested now that there's a reproduction (or did yours perhaps pre-exist?), due to the incompatibilities between different versions of ReactiveUI and DynamicData I hope I don't have to deal with this for long.

[JakenVeina]

In the meantime, I THINK you can work around the issue using .SuspendNotifications() around any .Edit() calls that are making subscriptions in this fashion. Or you can swap out the .Edit() call entirely, and just use .SuspendNotifications(). Or just wait for a fix. I'm workshopping it now, and we should be able to get it out to a pre-release as soon as I get to play nice with all our existing tests.