Apply locking to fix join race conditions. Fixes #787

src/DynamicData.Tests/Cache/InnerJoinFixtureRaceCondition.cs

@@ -32,6 +32,25 @@ public void LetsSeeWhetherWeCanRandomlyHitARaceCondition()
         ids.InnerJoin(itemsCache.Connect(), x => x.Id, (_, thing) => thing).Subscribe((z) => { }, ex => { }, () => { });
     }
 
+    // See https://github.com/reactivemarbles/DynamicData/issues/787 
+    [Fact(Skip = "Heavyweight test so probably best run manually if this issue comes up again")]
+    public void LetsSeeWhetherWeCanRandomlyHitADifferentRaceCondition()
+    {
+        var leftCache = new SourceCache<Thing, long>(x => x.Id);
+        var rightCache = new SourceCache<Thing, long>(x => x.Id);
+        var joined = leftCache.Connect().InnerJoin(rightCache.Connect(), x => x.Id,
+            (keys, leftThing, rightThing) =>
+                new Thing { Id = keys.leftKey, Name = $"{leftThing.Name} x {rightThing.Name}" });
+
+        IDisposable StartUpdating(ISourceCache<Thing, long> sourceCache) =>
+            Observable.Range(1, 1_000_000, Scheduler.Default)
+                .Subscribe(x => sourceCache.AddOrUpdate(new Thing { Id = x, Name = $"{x}" }));
+
+        using var leftUpdater = StartUpdating(leftCache);
+        using var rightUpdater = StartUpdating(rightCache);
+        using var subscription = joined.Subscribe();
+    }
+
     public class Thing
     {
         public long Id { get; set; }

src/DynamicData/Cache/Internal/FullJoin.cs

@@ -118,6 +118,9 @@ public IObservable<IChangeSet<TDestination, TLeftKey>> Run() => Observable.Creat
                     return joinedCache.CaptureChanges();
                 });
 
-                return new CompositeDisposable(leftLoader.Merge(rightLoader).SubscribeSafe(observer), leftCache, rightCache);
+                lock (locker)
+                {
+                    return new CompositeDisposable(leftLoader.Merge(rightLoader).SubscribeSafe(observer), leftCache, rightCache);
+                }
             });
 }

src/DynamicData/Cache/Internal/InnerJoin.cs

@@ -41,7 +41,7 @@ internal class InnerJoin<TLeft, TLeftKey, TRight, TRightKey, TDestination>(IObse
                 {
                     foreach (var change in changes.ToConcreteType())
                     {
-                        var leftCurent = change.Current;
+                        var leftCurrent = change.Current;
                         var rightLookup = rightGrouped.Lookup(change.Key);
 
                         if (rightLookup.HasValue)
@@ -52,7 +52,7 @@ internal class InnerJoin<TLeft, TLeftKey, TRight, TRightKey, TDestination>(IObse
                                 case ChangeReason.Update:
                                     foreach (var keyvalue in rightLookup.Value.KeyValues)
                                     {
-                                        joinedCache.AddOrUpdate(_resultSelector((change.Key, keyvalue.Key), leftCurent, keyvalue.Value), (change.Key, keyvalue.Key));
+                                        joinedCache.AddOrUpdate(_resultSelector((change.Key, keyvalue.Key), leftCurrent, keyvalue.Value), (change.Key, keyvalue.Key));
                                     }
 
                                     break;
@@ -117,6 +117,9 @@ internal class InnerJoin<TLeft, TLeftKey, TRight, TRightKey, TDestination>(IObse
                     return joinedCache.CaptureChanges();
                 });
 
-                return new CompositeDisposable(leftLoader.Merge(rightLoader).SubscribeSafe(observer), leftCache, rightCache, rightShare.Connect());
+                lock (locker)
+                {
+                    return new CompositeDisposable(leftLoader.Merge(rightLoader).SubscribeSafe(observer), leftCache, rightCache, rightShare.Connect());
+                }
             });
 }

src/DynamicData/Cache/Internal/LeftJoin.cs

@@ -119,6 +119,9 @@ public IObservable<IChangeSet<TDestination, TLeftKey>> Run() => Observable.Creat
                         return joined.CaptureChanges();
                     });
 
-                return new CompositeDisposable(leftLoader.Merge(rightLoader).SubscribeSafe(observer), leftCache, rightCache);
+                lock (locker)
+                {
+                    return new CompositeDisposable(leftLoader.Merge(rightLoader).SubscribeSafe(observer), leftCache, rightCache);
+                }
             });
 }