Wait for quiescence in MergeManyChangeSets stress tests (reactivemarbles#1100)

* Make MergeManyChangeSetsCacheSourceCompare stress test deterministic

MultiThreadedStressTest(10, 50) fails intermittently in CI with two prices
present in market.PricesCache.Items but missing from the live aggregator.
The two affected prices have the latest timestamps in the batch, which is
the signature of a race during high-contention production.

Bogus.Randomizer wraps System.Random. When constructed with a seed, the
randomizer stores the random in a protected localSeed field and bypasses
its internal Locker on every generator call. The test shares one seeded
Randomizer across many parallel producer threads:

- Directly via _randomizer.Number / .Bool / .TimeSpan / .Interval
- Indirectly via _marketFaker.WithSeed(_randomizer), since every
  Faker<T>.Generate call routes through the same randomizer

Concurrent calls into the underlying System.Random corrupt its internal
state, producing values inconsistent with what a serialized run would
produce. That is sufficient to explain the observed asymmetry between
the post-hoc PricesCache snapshot and the live aggregator stream.

Introduce SynchronizedRandomizer, a Randomizer subclass that replaces
the protected localSeed field with a LockedRandom (a Random subclass that
serializes every virtual method on an internal lock). The seed and method
contracts are unchanged; the wrapper only adds synchronization.

Apply it to the failing fixture. Other Randomizer uses across the test
project remain unchanged for now; they are either single-threaded or
have not exhibited flake symptoms.

Verified: 20 consecutive runs of the fixture pass at MaxParallelThreads=16,
zero failures.

* Wait for quiescence in MergeManyChangeSets stress tests

The post-reactivemarbles#1079 cache delivery model decouples mutation from notification:
AddOrUpdate enqueues a notification and returns; the actual delivery to
subscribers runs later on whichever thread wins the drain. That removed
the cross-cache deadlock the old Synchronize(lock) shape produced, but it
opened a small window between mutation and observed delivery. Tests that
compare a live aggregator's view against the cache's current Items at
assert time can see disagreement during that window.

The source-compare fixture already adopted the right shape:

    var merged = source.MergeManyChangeSets(...).Publish();
    var cacheCompleted = merged.LastOrDefaultAsync().ToTask();
    using var local = merged.AsAggregator();
    using var connect = merged.Connect();
    ...
    await cacheCompleted;
    CheckResultContents(..., local);

Port the same pattern to the cache and list MergeManyChangeSets stress
fixtures. The local aggregator now sits on the Publish chain so it shares
the completion task; the await before CheckResultContents pins the
quiescence point.

Also delete the SynchronizedRandomizer change made earlier on this branch.
Bogus.Randomizer takes a process-wide lock on Locker.Value for every
generator call regardless of whether localSeed is set, so the wrapper was
addressing a non-problem.

---------

Co-authored-by: Darrin Cullop <dacullop@microsoft.com>

Author: dwcullop

src/DynamicData.Tests/Cache/MergeManyChangeSetsCacheFixture.cs

@@ -5,6 +5,7 @@
 using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
+using System.Reactive.Threading.Tasks;
 using System.Threading.Tasks;
 using Bogus;
 using DynamicData.Kernel;
@@ -90,10 +91,11 @@ IObservable<MarketPrice> AddRemovePrices(Market market, int priceCount, int para
                 .Parallelize(priceCount, parallel, obs => obs.StressAddRemove(market.PricesCache, _ => GetRemoveTime(), scheduler))
                 .Finally(market.PricesCache.Dispose);
 
-        var merged = _marketCache.Connect().MergeManyChangeSets(market => market.LatestPrices);
-        using var priceResults = merged.AsAggregator();
-
+        var merged = _marketCache.Connect().MergeManyChangeSets(market => market.LatestPrices).Publish();
         var adding = true;
+        var cacheCompleted = merged.LastOrDefaultAsync().ToTask();
+        using var priceResults = merged.AsAggregator();
+        using var connect = merged.Connect();
 
         // Start asynchrononously modifying the parent list and the child lists
         using var addingSub = AddRemoveStress(marketCount, priceCount, Environment.ProcessorCount, TaskPoolScheduler.Default)
@@ -119,6 +121,9 @@ IObservable<MarketPrice> AddRemovePrices(Market market, int priceCount, int para
         }
         while (adding);
 
+        // Wait for the source cache to finish delivering all notifications.
+        await cacheCompleted;
+
         // Verify the results
         CheckResultContents(_marketCacheResults, priceResults);
     }

src/DynamicData.Tests/Cache/MergeManyChangeSetsListFixture.cs

@@ -5,6 +5,7 @@
 using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
+using System.Reactive.Threading.Tasks;
 using System.Threading.Tasks;
 using Bogus;
 using DynamicData.Kernel;
@@ -86,9 +87,11 @@ IObservable<Animal> AddRemoveAnimals(AnimalOwner owner, int animalCount, int par
                 .Parallelize(animalCount, parallel, obs => obs.StressAddRemove(owner.Animals, _ => GetRemoveTime(), scheduler))
                 .Finally(owner.Animals.Dispose);
 
-        var mergeAnimals = _animalOwners.Connect().MergeManyChangeSets(owner => owner.Animals.Connect());
-
+        var mergeAnimals = _animalOwners.Connect().MergeManyChangeSets(owner => owner.Animals.Connect()).Publish();
         var addingAnimals = true;
+        var cacheCompleted = mergeAnimals.LastOrDefaultAsync().ToTask();
+        using var animalResults = mergeAnimals.AsAggregator();
+        using var connect = mergeAnimals.Connect();
 
         // Start asynchrononously modifying the parent list and the child lists
         using var addAnimals = AddRemoveAnimalsStress(ownerCount, animalCount, Environment.ProcessorCount, TaskPoolScheduler.Default)
@@ -114,8 +117,12 @@ IObservable<Animal> AddRemoveAnimals(AnimalOwner owner, int animalCount, int par
         }
         while (addingAnimals);
 
-        // Verify the results
-        CheckResultContents();
+        // Wait for the source cache to finish delivering all notifications.
+        await cacheCompleted;
+
+        // Verify the results against the aggregator wired into the same Publish chain
+        // that cacheCompleted observes.
+        CheckResultContents(_animalOwners.Items, _animalOwnerResults, animalResults);
     }
 
     [Fact]