[Bug]: Cross-cache deadlock when concurrent updates notify subscribers that modify other SourceCache instances

[dwcullop]

Describe the bug 🐞

ObservableCache holds its internal _locker during InvokeNext, which calls _changes.OnNext synchronously while the lock is held. When a subscriber modifies a different SourceCache from inside the OnNext callback, that second cache acquires its own _locker and also calls InvokeNext under its lock. If two threads concurrently update two caches that notify into each other, the lock acquisition order inverts between threads, causing a deadlock.

This affects any scenario where SourceCache instances are wired together via Connect subscribers — including PopulateInto, MergeManyChangeSets, or any custom subscriber that writes to another cache in response to changes.

The deadlock is difficult to trigger because it requires concurrent updates to two interdependent caches with subscriber chains that cross cache boundaries.

Reproduction Test

const int iterations = 100;

for (var iter = 0; iter < iterations; iter++)
{
    using var cacheA = new SourceCache<TestItem, string>(static x => x.Key);
    using var cacheB = new SourceCache<TestItem, string>(static x => x.Key);

    using var subA = cacheA.Connect().Subscribe(changes =>
    {
        foreach (var c in changes)
        {
            if (c.Reason == ChangeReason.Add && !c.Current.Key.StartsWith("x"))
            {
                cacheB.AddOrUpdate(new TestItem("x" + c.Current.Key, c.Current.Value));
            }
        }
    });

    using var subB = cacheB.Connect().Subscribe(changes =>
    {
        foreach (var c in changes)
        {
            if (c.Reason == ChangeReason.Add && !c.Current.Key.StartsWith("x"))
            {
                cacheA.AddOrUpdate(new TestItem("x" + c.Current.Key, c.Current.Value));
            }
        }
    });

    var barrier = new Barrier(2);

    var taskA = Task.Run(() =>
    {
        barrier.SignalAndWait();
        for (var i = 0; i < 1000; i++)
        {
            cacheA.AddOrUpdate(new TestItem("a" + i, "V" + i));
        }
    });

    var taskB = Task.Run(() =>
    {
        barrier.SignalAndWait();
        for (var i = 0; i < 1000; i++)
        {
            cacheB.AddOrUpdate(new TestItem("b" + i, "V" + i));
        }
    });

    var completed = Task.WhenAll(taskA, taskB);
    var finished = await Task.WhenAny(completed, Task.Delay(TimeSpan.FromSeconds(5)));
}

private sealed record TestItem(string Key, string Value);

This test deadlocks on the current main branch.

Reproduction repository

https://gist.github.com/dwcullop/fe877f72ac7d8df4265c8affd771453d

Expected behavior

The deadlock shouldn't happen.

DynamicData Version

DynamicData 9.4.3

[dwcullop]

I believe the fix is to use two locks: one to protect mutating the container and one for Rx notifications. Operations like "Connect" will need to hold both locks, but other operations can have one or the other.

[JakenVeina]

There's some plumbing already in the project for doing coordinated upstream/downstream locking. I think added it for one of the .Filter() variants. Or maybe it was .ToObservableChangeSet().

I will say I don't think this is really valid to call a bug. There's a reason side-effects are heavily frowned upon in RX-land, and it's pretty much this. Really, I think the onus is on the consumer to not allow for side-effects to feed back upstream, in a synchronous manner. This can be accomplished pretty easily by tossing your side-effect over to an appropriate IScheduler instead of calling it directly. That's effectively how Redux does things, in JS land, it just looks a little different in a single-threaded environment.

That being said, I'm not against an upstream/downstream locking approach within SourceCache, if it solves the issue. But there's a gotcha involved in making sure that lock handover is synchronized. I.E. the upstream lock can't be released until the downstream lock is taken, otherwise it's possible for notification order to get lost during the small window when the collection has been mutated, but the downstream notification hasn't been published. That's what that SwappableLock class (or whatever I called it) I implemented does.