Visitar URL original
Serialize static Or, And, Except and Xor delivery across sources by dwcullop · Pull Request #1224 · reactivemarbles/DynamicData · GitHub
Skip to content

Serialize static Or, And, Except and Xor delivery across sources - #1224

Open
dwcullop wants to merge 1 commit into
mainfrom
u/dacullop/main/combiner_serialized_delivery
Open

dwcullop wants to merge 1 commit into
mainfrom
u/dacullop/main/combiner_serialized_delivery

Conversation

@dwcullop

@dwcullop dwcullop commented Nov 2, 2026

Copy link
Copy Markdown
Member

Summary

The static combiner behind Or, And, Except and Xor (Combiner) can deliver two changesets to its observer at the same time, breaking the Rx serialization contract. Downstream operators that rely on that contract lose data.

Root cause

  • Combiner.Update applied each source's changes and computed the combined changeset under _locker, then released the lock and called the downstream callback.
  • When two sources changed on different threads, the second thread could compute its changeset and call OnNext while the first thread's OnNext was still running.
  • Rx operators assume serialized input. TakeUntil, for example, silently drops an OnNext that arrives while another is in flight.
  • CrossCacheDeadlockStressTest shows this as an intermittent Or = Merged mismatch: orCache (behind TakeUntil) occasionally ends up a few keys short of mergedCache (no TakeUntil).

Fix

Combiner routes every source through one SharedDeliveryQueue, the mechanism DynamicCombiner (the dynamic overloads) already uses.

  • Updates from all sources are applied and delivered one at a time.
  • No lock is held while the observer runs, so a second producer enqueues and returns instead of overlapping the delivery in flight or blocking behind it.
  • _locker is removed. The queue is disposed first, so nothing is delivered once disposal starts.

Regression test

OrFixture.IntegrationTests.ChangesFromDifferentThreads_AreDeliveredSerially covers both the params and the collection overloads. It is deterministic: no timers or retries.

  1. A dedicated thread changes source 1, and the observer parks inside that delivery.
  2. The test thread then changes source 2.
  3. The test asserts that the deliveries never overlapped and that both changes arrive, in order.

On unmodified main, both cases fail because the second change is delivered while the first is still in flight. The same interleaving through .TakeUntil(Observable.Never<Unit>()) loses the second change entirely. With the fix, both cases pass.

Behavior note

No public API changes. One thing does change: if another thread is mid-delivery, a write to a source can now return before its change reaches the combined subscriber, and the delivering thread delivers it. This matches the other queue-based operators. Before this change, such a write could be delivered concurrently with the one in flight.

Validation

  • Full suite (net9.0): 2633 passed, 0 failed.
  • CrossCacheDeadlockStressTest, 900 runs (300 in-process, 600 under CPU oversubscription): the original combiner failed once with the Or mismatch, the fixed one never did. The base rate is low, so the deterministic regression test is the real proof.

The static Combiner behind Or, And, Except and Xor updated its state under a lock but invoked the downstream
callback after releasing it. When two sources changed on different threads, both changesets could be delivered to
the observer at the same time, violating the Rx serialization contract. Operators downstream that rely on it can
lose data: TakeUntil drops a notification that arrives while another is in flight, which is the intermittent
"Or = Merged" mismatch seen in CrossCacheDeadlockStressTest.

Route every source through one SharedDeliveryQueue, as DynamicCombiner already does. Updates from all sources are
applied and delivered one at a time, and no lock is held while the observer runs, so a second producer enqueues and
returns instead of overlapping the delivery in flight or blocking behind it.

The regression test parks one thread inside the delivery of a change from the first source and changes the second
source from another thread. Before the fix both overloads deliver the second change alongside the first.

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant