Repository navigation
Conversation
Added a new test `FilterObservableError_PropagatesDownstream` to ensure error propagation in `FilterOnObservableFixture`. Introduced `AggregateManyExtensions` class to provide the `AggregateMany` operator for aggregating changesets. Refactored `FilterOnObservable` to utilize the new `AggregateMany` operator, removing the `FilterProxy` implementation. Adjusted copyright header for consistency. Replaced the old `AggregateManyExtensions` with a new implementation.
Renamed `onSource` to `onSourceChangeSet` in `FilterOnObservable` and `AggregateManyExtensions` for better clarity. Refactored `GroupOnObservable` and `TransformOnObservable` to use `Observable.Using` and `Observable.Defer` with `AggregateMany`, removing the `Subscription` class to simplify code and improve efficiency.
Replaces inheritance-based CacheParentSubscription with composition-based ICacheOrchestrator + Orchestrator runner. Per-subscription state is owned by orchestrator instances; runtime context (Track/Serialize/ScheduleEmit) is injected via Initialize on a fresh Subscription created inside each Observable.Create call, so multi-subscribe correctness is preserved. Primitive - ICacheOrchestrator / ICacheOrchestratorContext (Cache.Internal) - Orchestrator<TSource,TKey,TInner,TResult>.Run() wraps Observable.Create - IntObservableCacheEx.OrchestrateMany family of extension methods plus AsCacheOrchestrator lambda adapter - Specialized helpers OrchestrateManyChanges (ChangeAwareCache mirror), OrchestrateManyMerged (cache->cache merge), and OrchestrateManyMergedList (cache->list merge) Migrated operators - FilterOnObservable, TransformOnObservable, GroupOnObservable - MergeManyCacheChangeSets, MergeManyCacheChangeSetsSourceCompare, MergeManyListChangeSets (Cache/Internal/) - TransformManyAsync (showcase for Serialize hook) - AutoRefresh (uses ScheduleEmit for shared-grid buffered refresh batching; source pass-through preserved verbatim; refreshes suppressed for keys the source has touched in the same drain, fixes reactivemarbles#1099) Behavior changes (for next major version) - AutoRefresh and AutoRefreshOnObservable now propagate reevaluator exceptions (sync throw and OnError) instead of silently swallowing them - AutoRefresh now throws ArgumentNullException for null propertyAccessor / reevaluator at composition time instead of throwing NRE on first event - FilterOnObservable with non-null buffer now uses quiescence-based Pub+Buf+Throttle+Amb instead of fixed-grid Buffer; same coalescing, zero idle allocations - AutoRefresh no longer emits Refresh changes for keys touched by source in the same drain (eliminates redundant Add+Refresh and prevents Add+Remove+Refresh) Other fixes - Removed over-strict debug assertion in SuspendNotifications (functional state unaffected, only DEBUG-mode race window flagged) - Deterministic seed in TransformAsyncFixture.RemoveFlowsToTheEnd (replaces Random.Shared) - Cross-cache deadlock fixture ported to OrchestrateManyFixture Deletes - CacheParentSubscription.cs and CacheParentSubscriptionFixture.cs
Adds an optional CacheChangeHandlerBase abstract class that implements ICacheOrchestrator.OnSourceChangeSet by dispatching each Change to one of five protected virtual hooks (OnItemAdded/Updated/Removed/Refreshed + OnChangeSetProcessed). Orchestrators that inherit it stop hand-rolling the foreach+switch over change.Reason. Migrated to the base class: - ChangesOrchestrator (OrchestrateManyChanges) - MergedOrchestrator (OrchestrateManyMerged) - MergedListOrchestrator (OrchestrateManyMergedList) - TransformManyAsync.Orchestrator - AutoRefresh.UnbufferedOrchestrator + AutoRefresh.BufferedOrchestrator Removed ICacheOrchestratorContext.ScheduleEmit. Only AutoRefresh used it, and the buffered path now uses Subject<Change>.Buffer routed through Context.Serialize to keep all activity under the shared queue without needing a dedicated 'schedule a future drain' primitive. AutoRefresh.WithBuffering behavior change: - Now dedups Refresh changes within each buffer window (one Refresh per key per window), matching Jake's choice in the upstream WIP. - Source events and refreshes remain independent (refreshes for keys source touched in the same window are still emitted). AutoRefresh.WithoutBuffering preserves the reactivemarbles#1099 fix: refreshes for keys the source touched in the same drain cycle are suppressed. All 2386 tests pass, 0 skipped.
# Conflicts: # src/DynamicData/Cache/ObservableCacheEx.cs
Capture the downstream observer (as a DeliverySubQueue) at Initialize time instead of passing it to Emit per drain. The orchestrator now calls Emitter.OnNext/OnError/OnCompleted directly under the queue gate; serialization, post-termination drop, and reentrant drain are all handled by the sub-queue. ICacheOrchestrator becomes (Initialize(context, emitter), OnSourceChangeSet, OnInner, OnDrainComplete). Emit(observer) is gone. Renamed CacheChangeHandlerBase -> OrchestratorCacheChangeBase. Eliminated from the runtime: _hasTerminated, TerminalError helper, direct calls to the raw downstream observer. Eliminated from orchestrators: _pendingSource accumulator (both AutoRefresh orchestrators), _shouldFlushBuffered + _bufferedRefreshes (BufferedOrchestrator), OnChangeSetProcessed virtual (no remaining consumers). Net diff: +106 / -281 lines across the orchestrator surface. All 2386 tests pass.
Replace the quiescence-based throttle+amb scheme with a simpler leading-edge buffer: Publish + Buffer(() => Take(1).Delay(window, scheduler)) routed through the existing public FlattenBufferResult helper. Behaviour: idle streams allocate nothing (no timer scheduled until an actual emission arrives). Bursts still emit at most every 'window' interval. The throttle-based settle window is gone; if a caller relied on early flushing during quiet periods, they will now wait the full window. Net wire-output is the same set of merged changesets, just at slightly different timing.
Modified the buffering logic in the `OnInner` method of the `AutoRefresh` class. Changed from using `_refreshes.Buffer(_window, _scheduler)` to `_refreshes.Buffer(() => _refreshes.Take(1).Delay(_window, _scheduler))`. This update introduces a dynamic window for buffering events, potentially improving the timing and responsiveness of the refresh batching mechanism.
Collapse UnbufferedOrchestrator and BufferedOrchestrator into a single Orchestrator. Source changes still flow to the emitter immediately in both modes; refreshes accumulate in a per-key dictionary (latest value wins) and flush either at drain end (no buffer) or via a SerialDisposable-wrapped Timer armed by the first pending refresh (buffered). Pending refreshes are dropped when the source Updates or Removes the same key, so a Refresh carrying a value the source has already superseded is never emitted. The reactivemarbles#1099 source-touched suppression now applies to the buffered path as well, removing the sibling-bug between the two modes. Orchestration runtime now disposes orchestrators that implement IDisposable, fixing a pre-existing leak where the buffered orchestrator's Subject and buffer subscription survived subscription teardown. Adds 6 regression tests in AutoRefreshOnObservableFixture covering each scenario (Remove drops pending refresh, Update replaces it, multi-Update keeps only the latest) for both buffer modes.
Asserts that when an Update lands inside an armed window and the new instance's reevaluator never emits, no Refresh ever surfaces. Pins the 'pending empty disarms the timer' branch in isolation, separate from the existing test that follows up with a v2 emission.
The mid-test assertion in ChangeSetBufferIsGiven_UpdateDuringWindow_RefreshEmittedForNewInstance already proves the v1-armed timer is disarmed by the Update (asserts BeEmpty at the original v1 window boundary). A disarmed timer cannot spontaneously fire later, so the separate quiet-after-update case is transitively covered.
Both operators were hand-rolling the per-key inner subscription, completion counter, and queue plumbing that the Orchestrator runtime already provides. The orchestrator version eliminates the StrongBox<int> counter, the Concat(Observable.Never) hack to suppress source completion, and the manual Finally(CheckCompleted) bookkeeping on each inner subscription. Inner-observable errors now terminate the merged stream, matching the standard Rx contract. The previous behavior swallowed inner errors and treated them as completion. Two tests that asserted the swallow are updated to assert the propagation contract; consumers that want the old behavior can wrap their selector output with .Catch(Observable.Empty<...>()). OrchestratorCacheChangeBase.OnDrainComplete is now virtual with an empty default. Orchestrators that have no per-drain coalescing work no longer need to override it.
C1 (CRITICAL): TransformManyAsync and other aggregating orchestrators intermittently dropped concurrent inner emissions. When OnDrainComplete called tracker.EmitChanges(Emitter), the resulting Emitter.OnNext triggered a reentrant DrainPending that delivered the emitter item AND drained any concurrent inner items that had arrived during the brief lock release in DeliverStaged. Those items added to the tracker via OnInner but were never flushed: the outer DrainAll loop saw active bits empty and exited without firing _onDrainComplete again. Fix: ChangeSetMergeTracker.EmitChanges (and the List variant + DynamicGrouper) now return bool indicating whether they emitted. Every aggregating orchestrator loops in OnDrainComplete until the tracker reports nothing left to emit. C2 (CRITICAL): Buffered AutoRefresh dropped pending refreshes when the source completed before the buffer timer fired. The Timer subscription was registered via Context.Serialize, which does not participate in the orchestrator's completion counter; source completion could race ahead and fire Emitter.OnCompleted while a refresh sat in the pending dictionary unflushed. Fix: added ICacheOrchestratorContext.TrackAuxiliary<T>(observable, onNext) that subscribes through the queue AND contributes to completion accounting via Finally(DecrementSubscriptionCount). AutoRefresh.Orchestrator's buffered timer now uses TrackAuxiliary, so completion waits for the timer to fire and flush any pending refresh. H1 (HIGH, latent): The 3-lambda OrchestrateMany overload constructed a single LambdaCacheOrchestrator eagerly. Each subscription to the returned observable called Initialize on the same orchestrator instance, overwriting _context and _emitter from earlier subscribers. Current consumers wrapped the call site in Observable.Defer/Create so the bug was latent, but the primitive was a trap. Fix: wrapped the lambda overload in Observable.Defer so each subscription constructs a fresh orchestrator. Adds three regression tests covering the reentrant-drain hazard, the multi-subscribe trap, and the buffered completion-flush case. Full suite passes 2402/2402; the previously-flaky TransformManyAsyncFixture test now passes 30/30.
Three new tests in OrchestrateManyFixture: InnerError_PropagatesAndTerminatesStream - asserts that an OnError from a tracked inner observable terminates the merged stream with the same error and that completion does not also fire. SourceAlreadyCompleted_PropagatesCompletion - asserts that subscribing to an Observable.Empty source through the orchestrator immediately completes the downstream stream. SourceAlreadyErrored_PropagatesError - asserts that subscribing to an Observable.Throw source propagates the error through the orchestrator.
The previous fix added TrackAuxiliary to ICacheOrchestratorContext so that the buffered AutoRefresh timer subscription could participate in completion accounting. That polluted the orchestrator context with a method only one consumer needed. Replace it with a parameter on the existing OnDrainComplete hook: the runtime snapshots its completion latch BEFORE calling the orchestrator and passes the result as bool sourcesCompleted. AutoRefresh's buffered branch now flushes pending refreshes synchronously when sourcesCompleted is true, so the refresh surfaces before the runtime fires OnCompleted downstream and the timer subscription becomes purely opportunistic (no completion accounting needed). All ICacheOrchestrator implementations and OrchestratorCacheChangeBase updated for the new signature; consumers that did not need the flag simply ignore it. The lambda OrchestrateMany overload's onDrainComplete callback signature is unchanged (none of its current consumers need the flag). C2 regression test rewritten to match the new semantics: source + reevaluator completion now triggers an immediate synchronous flush rather than waiting for the buffer timer; the test verifies the refresh surfaces, completion happens immediately, and the timer is cancelled (no duplicate refresh at the original window boundary). Tests: 2405/2405 pass. TransformManyAsync flake check 30/30 pass.
Change OrchestrateMany to accept Func<context, emitter, ICacheOrchestrator> instead of a pre-constructed orchestrator instance. The runtime invokes the factory once per subscription, passing the per-subscription context and downstream emitter; the orchestrator captures both as readonly constructor params and is fully initialized when its first method is called. Eliminates: - Initialize method on ICacheOrchestrator (one fewer interface method) - The 'null!' partial-init zone in every orchestrator (_context, _emitter fields no longer needed) - The Observable.Defer wrapper around the lambda OrchestrateMany overload (factory pattern makes per-subscription instances structurally required, so H1's multi-subscribe corruption hazard cannot recur) - The redundant Observable.Create wrappers in MergeMany, MergeManyItems, TransformManyAsync, OrchestrateManyMerged, OrchestrateManyMergedList, OrchestrateManyChanges Run() methods (OrchestrateMany already creates per-subscription state) OrchestratorCacheChangeBase is now a primary-constructor abstract class that forwards context/emitter via constructor chaining. Derived orchestrators (Orchestrator inside AutoRefresh/MergeMany/MergeManyItems/TransformManyAsync, MergedOrchestrator, MergedListOrchestrator, ChangesOrchestrator, LambdaCacheOrchestrator, and the test TestOrchestrator) take context and emitter as their first two constructor parameters and chain to the base. Test orchestrator construction in OrchestrateManyFixture is wrapped by a Wire(...) helper that returns the observable plus a Func<TestOrchestrator> thunk; tests call Wire then subscribe to capture the per-subscription orchestrator instance. Tests: 2405/2405 pass. TransformManyAsync flake check 30/30 pass.
The C1 fix previously required every orchestrator's OnDrainComplete to loop until its tracker reported empty. The hazard was actually a queue-level lifecycle gap: when OnDrainComplete called Emitter.OnNext, ExitLockAndDrain took the reentrant path and drained items inline but left no trace; the outer DrainAll then checked _activeBits, saw it empty, and exited without re-firing OnDrainComplete. State accumulated during the reentrant drain (e.g. ChangeSetMergeTracker entries) was stranded until the next unrelated drain. Move the fix to SharedDeliveryQueue: set a _drainReentered flag in the reentrant path; reset it before each OnDrainComplete callback; loop back in DrainAll when either _activeBits.HasAny() OR _drainReentered. The flag is only touched by the drain thread, so no synchronization is needed. Removes: - bool return from ChangeSetMergeTracker.EmitChanges (Cache + List) - bool return from DynamicGrouper.EmitChanges - per-consumer while-loops in AutoRefresh, TransformManyAsync, OrchestrateManyMerged, OrchestrateManyMergedList, OrchestrateManyChanges, TransformOnObservable, GroupOnObservable Every orchestrator's OnDrainComplete now emits at most once per call; the queue re-calls them whenever a reentrant drain may have updated their state. Tests: 2405/2405 pass. TransformManyAsync flake check 30/30 pass.
Rename: OnDrainComplete(bool sourcesCompleted) -> OnDrainComplete(bool isFinal). The new name is action-oriented (signals to the orchestrator that this is the last chance to emit before downstream OnCompleted) rather than implementation-coupled (the old name referenced the runtime's _isCompleted latch mechanism). XML doc updated to make the meaning explicit. Add two missing contract tests: 1. OrchestrateManyFixture.OnDrainComplete_IsFinalIsFalseUntilSourceAndAllInnersComplete - Verifies isFinal=false on every call while source and tracked inners are still active, that source completion alone does not flip isFinal while an inner subscription remains, and that isFinal=true on at least one call after the last inner completes (followed by downstream OnCompleted). 2. SharedDeliveryQueueFixture.OnDrainComplete_RefiresAfterReentrantDrainTriggeredByCallback - Focused regression test for the queue-level C1 fix: when the onDrainComplete callback enqueues an item onto a sub-queue, the reentrant drain processes the item but the outer DrainAll must re-fire onDrainComplete because of the _drainReentered flag. Red-green verified: fails when _drainReentered check is removed; passes when restored. Tests: 2407/2407 pass (+2 new). TransformManyAsync flake check 30/30 pass.
Bundles eight design fixes from the orchestrator review: (1+2) Counter-based final check + CAS-latched completion: OnDrainComplete derives isFinal from the live subscription counter rather than a separate _isCompleted latch, and uses a one-shot CAS on _completionEmitted to guarantee exactly-one OnCompleted. Fixes a race where an orchestrator that called Track during OnDrainComplete(isFinal=true) would have the new inner's emissions silently dropped because the latch stayed set. (3) Factory exception safety: the factory call is wrapped in try/catch that disposes the already-allocated queue and emitter before rethrowing, so a faulting orchestrator constructor doesn't leak the runtime's resources. (4) Disposal order: source and inner subscriptions are disposed before the queue, so source pumps cannot keep firing into a terminating queue. (8) Documented Serialize lifetime semantics: a remarks block on ICacheOrchestratorContext.Serialize calls out explicitly that Serialize does NOT participate in completion accounting (unlike Track), preventing future TrackAuxiliary-style confusion. (9) Generic devirtualization: Orchestration and OrchestrateMany take a TOrch type parameter so dispatch sites (OnSourceChangeSet, OnInner, OnDrainComplete) call the concrete sealed class through a non-interface field and the JIT can devirtualize. All internal call sites updated to specify TOrch explicitly (older C# language modes can't infer it from the lambda body). (11) Track + Untrack split: the nullable observable parameter on Track is gone; Untrack(key) is now the explicit way to remove a registration. KeyedDisposable.Add doc clarified (Add has replace-or-add semantics). (13) wasReentrant flag on OnDrainComplete: the SharedDeliveryQueue callback now passes the captured _drainReentered flag through, and ICacheOrchestrator.OnDrainComplete(bool isFinal, bool wasReentrant) exposes it for advanced consumers. Most orchestrators ignore it. (14) Named SourceSubscriptionWeight constant replaces the magic 1 in _subscriptionCounter initialization. API rename: the 3-lambda convenience overload is now OrchestrateLambdas (not OrchestrateMany) to avoid generic inference conflict with the factory overload. GroupOnObservable and TransformOnObservable updated accordingly. Tests: 2407/2407 pass.
Adds the level-1 (in-isolation) test pattern complementing the existing level-2 (end-to-end) fixtures. Each orchestrator can now be tested two ways: via OrchestrateMany through the full SharedDeliveryQueue runtime (already covered by existing fixtures like AutoRefreshOnObservableFixture, TransformManyAsyncFixture), or directly using FakeOrchestratorContext to drive method calls and assert on the orchestrator's contract with its context (Track/Untrack calls) and emit policy. New isolation fixtures: - MergeManyOrchestratorFixture (4 tests): Add-tracks, Remove-untracks, OnInner-passthrough, OnDrainComplete-is-noop - MergeManyItemsOrchestratorFixture (2 tests): ItemWithValue pairing, Untrack on Remove - TransformManyAsyncOrchestratorFixture (4 tests): tracking, untrack, drain-emits-tracker, transformer-throws-routes-through-error-handler - MergedOrchestratorFixture (2 tests): Add-tracks-and-routes, Remove-untracks - MergedListOrchestratorFixture (2 tests): list parity - ChangesOrchestratorFixture (3 tests): onSourceChange per change, onInner routes, Untrack on Remove - LambdaCacheOrchestratorFixture (3 tests): forwarding contract for the 3-lambda overload - AutoRefreshOrchestratorFixture (5 tests): unbuffered drain flush, source-touched suppression, buffered defers to timer, buffered+isFinal flushes synchronously, Remove drops pending refresh Required visibility bumps from private to internal on 8 nested orchestrator classes (still inside internal enclosing types, so no expanded public surface). Supporting infrastructure: - FakeOrchestratorContext<TKey, TInner>: in-memory ICacheOrchestratorContext that records Track/Untrack calls without subscribing to observables (Serialize returns the observable as-is). Tests must call OnInner manually. - CollectingObserver<T>: minimal IObserver<T> that records values + terminal state. Tests: 2432/2432 pass (+25). TransformManyAsync flake check 30/30 pass.
Four changes: 1. SizeLimitFixture.OnCompleteIsInvokedWhenSourceIsDisposed: re-skip with the original message. The test is unrelated to this branch's changes (LimitSizeTo was not migrated to the orchestrator) and is racy under full-suite ThreadPool load. The previous commit un-skipped it speculatively; that was a mistake. 2. Orchestration.OrchestratorContext: extend the try/catch to cover all construction steps (CreateQueue, factory, source subscribe), not just the factory call. Without this, an exception from CreateQueue or the source subscribe leaks the queue/emitter/orchestrator/source-subscription because the ctor never completes and Dispose never runs. The catch block tears down everything that was successfully allocated. 3. AutoRefresh.Orchestrator.OnItemRefreshed: also DropPending(key) when the source forwards a Refresh. Without this, a pending refresh from a prior drain (queued by the reevaluator) and the source's Refresh were both emitted to the consumer for the same item, in quick succession. Symmetrical with OnItemUpdated which already drops pending. Regression test added in AutoRefreshOrchestratorFixture. 4. Rename OrchestrateLambdas to an OrchestrateMany overload and reshape the source-change callback. The (changes, track, untrack) shape was messy and could not expose the full ICacheOrchestratorContext (no Serialize, no Untrack-by-token). The new shape is (changes, context) passing the runtime context directly. GroupOnObservable and TransformOnObservable updated. Test fixtures updated.
Major internal refactor to unify orchestration logic for cache operators. Renamed OrchestrateMany to Orchestrate (and related types), updated all usages, and consolidated cache/list merging logic into new OrchestrateChangeSets/OrchestrateManyChangeSets methods. Lambda orchestrators now receive the emitter in onInner. Removed obsolete orchestrator classes and merged related logic. Updated tests and internal docs to match new structure. No public API changes.
dwcullop
changed the title
dwcullop
changed the title
There was a problem hiding this comment.
Pull request overview
This PR introduces a new internal “Orchestrate” composition primitive to standardize (and harden) the common DynamicData cache-operator pattern of maintaining 1:1 per-item inner subscriptions, routing all notifications through SharedDeliveryQueue, and performing correct completion/error/disposal accounting. It removes the older CacheParentSubscription base-class pattern and migrates multiple operators to the new orchestrator-driven implementation, alongside substantial new internal test coverage and several AutoRefresh correctness fixes.
Changes:
- Added internal orchestrator contracts + driver (
ICacheOrchestrator*,CacheOrchestration, andIntObservableCacheEx.Orchestrate*) and updatedSharedDeliveryQueueto surface reentrant-drain info. - Migrated per-item-subscription operators (e.g.,
MergeMany,FilterOnObservable,GroupOnObservable,TransformOnObservable,TransformManyAsync, MergeMany*ChangeSets) offCacheParentSubscription. - Expanded/adjusted tests to validate orchestration contracts and un-skipped/fixed prior AutoRefresh-related defects.
Reviewed changes
Copilot reviewed 41 out of 41 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| src/DynamicData/Internal/SharedDeliveryQueue.cs | Adds reentrant-drain tracking and passes reentrancy info into a drain-complete callback. |
| src/DynamicData/Internal/KeyedDisposable.cs | Clarifies Add semantics in XML docs (replace-or-add and disposal behavior). |
| src/DynamicData/Internal/CacheParentSubscription.cs | Removes the legacy parent/child subscription primitive. |
| src/DynamicData/Cache/ObservableCacheEx.MergeManyChangeSets.cs | Routes MergeManyChangeSets through the new orchestration primitives; inlines source-priority helper types. |
| src/DynamicData/Cache/ObservableCacheEx.AutoRefreshOnObservable.cs | Adds missing null-argument validation for the reevaluator overload. |
| src/DynamicData/Cache/ObservableCacheEx.AutoRefresh.cs | Adds missing null-argument validation for propertyAccessor. |
| src/DynamicData/Cache/ObservableCache.cs | Removes an assertion in SuspendNotifications. |
| src/DynamicData/Cache/Internal/TransformOnObservable.cs | Replaces CacheParentSubscription implementation with Orchestrate + per-drain ChangeAwareCache flush. |
| src/DynamicData/Cache/Internal/TransformManyAsync.cs | Replaces CacheParentSubscription implementation with an orchestrator-based implementation and merge tracker. |
| src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs | Switches to OrchestrateManyChangeSets (list-shape overload). |
| src/DynamicData/Cache/Internal/MergeManyItems.cs | Reimplements via Orchestrate (per-key tracking + forwarding). |
| src/DynamicData/Cache/Internal/MergeManyCacheChangeSetsSourceCompare.cs | Removes the legacy wrapper implementation (now handled via orchestrator path). |
| src/DynamicData/Cache/Internal/MergeManyCacheChangeSets.cs | Removes the legacy wrapper implementation (now handled via orchestrator path). |
| src/DynamicData/Cache/Internal/MergeMany.cs | Reimplements via Orchestrate (per-key tracking + forwarding). |
| src/DynamicData/Cache/Internal/IntObservableCacheEx.OrchestrateManyChangeSets.cs | New internal helper: orchestration specialization for merging child cache/list changesets. |
| src/DynamicData/Cache/Internal/IntObservableCacheEx.OrchestrateChangeSets.cs | New internal helper: orchestration specialization for “mirror manipulator” cache shapes via ChangeAwareCache. |
| src/DynamicData/Cache/Internal/IntObservableCacheEx.Orchestrate.cs | New internal entry points: typed-factory and lambda overloads for orchestrating source + inner observables. |
| src/DynamicData/Cache/Internal/ICacheOrchestratorContext.cs | New orchestrator runtime context API (Track, Untrack, Serialize). |
| src/DynamicData/Cache/Internal/ICacheOrchestrator.cs | New orchestrator contract interface (OnSourceChangeSet, OnInner, OnDrainComplete). |
| src/DynamicData/Cache/Internal/GroupOnObservable.cs | Migrates GroupOnObservable to Orchestrate (with DynamicGrouper lifetime managed via Observable.Using). |
| src/DynamicData/Cache/Internal/FilterOnObservable.cs | Migrates FilterOnObservable to OrchestrateChangeSets and updates buffering strategy. |
| src/DynamicData/Cache/Internal/CacheOrchestratorBase.cs | Adds a dispatching base class for per-ChangeReason orchestration. |
| src/DynamicData/Cache/Internal/CacheOrchestration.cs | New driver managing queue, per-key inner subscriptions, completion accounting, and construction cleanup. |
| src/DynamicData/Cache/Internal/AutoRefresh.cs | Rewrites AutoRefresh using an orchestrator with pending-refresh coalescing and buffering fixes. |
| src/DynamicData.Tests/Utilities/FakeOrchestratorContext.cs | New test utility for in-isolation orchestrator testing. |
| src/DynamicData.Tests/Utilities/CollectingObserver.cs | New test utility to collect values/terminal state for in-isolation tests. |
| src/DynamicData.Tests/Internal/TransformManyAsyncOrchestratorFixture.cs | New in-isolation tests for the TransformManyAsync orchestrator. |
| src/DynamicData.Tests/Internal/SharedDeliveryQueueFixture.cs | Adds regression coverage for reentrant drain + onDrainComplete behavior. |
| src/DynamicData.Tests/Internal/OrchestrateFixture.cs | New orchestration contract tests (serialization, completion counting, deadlock proof, reentrancy). |
| src/DynamicData.Tests/Internal/MergedOrchestratorFixture.cs | New in-isolation tests for the cache-shape merged orchestrator. |
| src/DynamicData.Tests/Internal/MergedListOrchestratorFixture.cs | New in-isolation tests for the list-shape merged orchestrator. |
| src/DynamicData.Tests/Internal/LambdaCacheOrchestratorFixture.cs | New in-isolation tests for the lambda orchestrator adapter. |
| src/DynamicData.Tests/Internal/ChangeSetOrchestratorFixture.cs | New in-isolation tests for the ChangeAwareCache-based orchestrator. |
| src/DynamicData.Tests/Internal/CacheParentSubscriptionFixture.cs | Removes legacy primitive tests (now replaced by OrchestrateFixture). |
| src/DynamicData.Tests/Internal/AutoRefreshOrchestratorFixture.cs | New in-isolation tests for AutoRefresh orchestrator buffering + suppression rules. |
| src/DynamicData.Tests/Cache/TransformAsyncFixture.cs | Makes delay randomness deterministic (Bogus Randomizer) and cleans up formatting. |
| src/DynamicData.Tests/Cache/MergeManyWithKeyOverloadFixture.cs | Updates expectations around child-observable error propagation. |
| src/DynamicData.Tests/Cache/FilterOnObservableFixture.cs | Adds coverage for propagating errors from per-item filter observables. |
| src/DynamicData.Tests/Cache/AutoRefreshOnObservableFixture.WithoutKey.cs | Un-skips reevaluator null-check regression test. |
| src/DynamicData.Tests/Cache/AutoRefreshOnObservableFixture.Base.cs | Un-skips/fixes multiple AutoRefreshOnObservable defect tests and adds new buffered-window regression tests. |
| src/DynamicData.Tests/Cache/AutoRefreshFixture.WithPropertyAccessor.cs | Un-skips propertyAccessor null-check regression test. |
Comments suppressed due to low confidence (1)
src/DynamicData/Cache/Internal/IntObservableCacheEx.OrchestrateManyChangeSets.cs:190
- Same double-queue pattern as the cache-shape overload: Serialize(...) is applied before creating ClonedListChangeSet, and then entry.Source is also passed to Track(...), causing nested SynchronizeSafe layers per key. This adds avoidable overhead in hot paths.
}
}
Five fixes from the test review: 1. OrchestrateFixture.Serialization_ParentAndChildDoNotInterleave: was vacuous (only one source push, no child emissions, callLog only contained one start/end pair). Rewritten to race source-side Refresh and child-side OnNext from two threads via Barrier, so the serialization gate actually has work to serialize. 2. OrchestrateFixture.ReentrantDrain: replaced await Task.Delay(50) with SpinWait.SpinUntil(predicate, 5s deadline). Deterministic completion check with a deadlock guard instead of a fixed wait. 3. TransformManyAsyncOrchestratorFixture.TransformerThrows: replaced await Task.Delay(20) with await trackedObs.DefaultIfEmpty(). LastOrDefaultAsync(). The deferred async lambda runs sync (throw + catch + handler-call + return Empty), so the inner completes deterministically. 4. ChangeSetOrchestratorFixture.OnDrainComplete_EmptyChangeSet_ NoEmission: new test verifying the orchestrator's empty-changeset suppression (the if (captured.Count != 0) guard). 5. MergedOrchestratorFixture / MergedListOrchestratorFixture: expanded from 2 tests each to 5/4. Added OnItemUpdated (re-tracks), MultipleKeys (independent tracking), and (cache only) OnItemRefreshed_WithoutReevalOnRefresh (negative path no-op). Did NOT add the positive reevalOnRefresh test: it can't pass in isolation because the orchestrator inspects the per-key entry's internal cache, which is only populated by the entry's source observable (driven by the real CacheOrchestration runtime, not the FakeOrchestratorContext). That path is covered by the existing integration tests in MergeManyChangeSetsCacheFixture. Also: AutoRefreshOrchestratorFixture now uses "using Microsoft.Reactive.Testing;" instead of the fully-qualified TestScheduler name.
dwcullop
changed the title
…avior change Two issues raised by the PR reviewer: 1. Double queue wrap in OrchestrateManyChangeSets / TransformManyAsync. ChangeSetCache (and ClonedListChangeSet) were being constructed over Context.Serialize(inner), and then entry.Source was passed to Context.Track, which also wraps with SynchronizeSafe. Each per-key inner emission was passing through two SharedDeliveryQueue sub-queues, doubling allocations and queue traffic. Drop the explicit Serialize and let Track do it once. 2. Stale XML docs on MergeMany. With the migration to the orchestrator, errors from per-item observables now terminate the merged stream rather than being swallowed. Updated the docs on MergeMany itself plus the four operators that route through it (WhenAnyPropertyChanged, WhenPropertyChanged, WhenValueChanged, AutoRefreshOnObservable) to describe the new behavior and call out the prior swallowing behavior.
The reviewer's double-wrap concern was a symptom of ChangeSetCache (and ClonedListChangeSet) doing .Do(Cache.Clone) inside their constructor. That forced every caller to pre-serialize the source observable so the mirror stayed coherent with whatever else read it, even when the caller already had its own serialization layer downstream. The orchestrator was having to do it twice: once via Context.Serialize to protect the mirror, and again via Context.Track to dispatch OnInner on the queue thread. Dropping just one wrap raced OnItemRemoved's enumeration of the mirror against concurrent inner emissions. Make both types passive: Source returns the raw observable, and a new Process(changes) method updates the mirror. Each caller now drives the mirror at the point in its pipeline where delivery is already serialized: - Cache and List MergeChangeSets call Process inside the MergeMany Do(). - MergeManyCacheChangeSets and MergeManyListChangeSets do the same. - CacheOrchestration-based operators (MergedOrchestrator, MergedListOrchestrator, TransformManyAsync) call Process in OnInner, on the queue thread, alongside the tracker call. They no longer need Context.Serialize. Single SynchronizeSafe wrap per per-key inner subscription. Mirror mutation happens on the same thread that reads it.
…o-op error forwarders
The list MergeMany was actively swallowing inner errors with an empty
onError handler. Forward the error to the merged observer so children
that fail terminate the merged stream, matching the cache MergeMany
behavior change.
The cache and list MergeChangeSets pipelines had
.Do(static _ => { }, observer.OnError) sitting after the MergeMany
projection. With Rx already propagating inner errors through MergeMany
to the downstream SubscribeSafe, that .Do was a no-op that just looked
like custom error handling. Remove it.
List-side docs updated for MergeMany, WhenAnyPropertyChanged,
WhenPropertyChanged, WhenValueChanged, AutoRefreshOnObservable, and
FilterOnObservable to describe the new propagation contract.
The list MergeManyFixture had a test asserting the old swallow
behavior (MergedStreamCompletesIfLastItemFails). Replaced with
MergedStreamFailsIfChildFails, matching the cache fixture's equivalent.
…hangeSets for list Completes the parity with cache operators from PR reactivemarbles#1114: - MergeManyItems: per-item observable emissions paired with the source item via ItemWithValue<T, TDest>. - TransformManyAsync: async transformer producing a child list changeset per source item; merged into a single output. Built on Task.FromAsync + SelectMany to integrate the await into the orchestrator pipeline. - MergeManyChangeSets (parent-comparer overload): conflict resolution when the same destination key appears in multiple parents' child changesets; the parent comparer picks the winning parent. Implemented via ParentChildEntry wrapper and reuses the existing MergeManyCacheChangeSetsOrchestrator. 15 new tests across 3 fixtures; 768/768 List tests pass. Public API additions: 3 new extension methods on ObservableListEx; all internal types remain internal.
Removed XML and inline comments across new and modified files that won't make sense after this PR merges or that narrate test/code flow without adding maintenance value: - Test fixture class summaries that just restated the SUT name - Inline test-flow narration (Arrange/Setup-style markers) - 'Prior to v9' release-note bullets on public docs (current behavior is already documented; release notes go in release notes) - 'Used by FooOperator, BarOperator' callouts in internal API docs - 'Unlike MergeMany, child errors are NOT swallowed' comparison on MergeManyChangeSets (no longer accurate now that MergeMany propagates) - The 'C1 fix' regression-test summary on SharedDeliveryQueueFixture - 'legacy CPS-subclass shape' and similar pre-orchestrator-refactor refs Kept comments that explain non-obvious WHY for the code at hand: lock ordering, race-condition rationale, subtle Rx behavior.
The cache-side MergeManyListChangeSets<TObject, TKey, TDestination> class was a one-line wrapper around OrchestrateManyChangeSets. Inline the call at the public API site and remove the file. The list-side MergeManyListChangeSets<TObject, TDestination> remains; it serves a list-source pipeline and has no orchestrator equivalent.
JakenVeina
left a comment
JakenVeina
left a comment
There was a problem hiding this comment.
Overall, I'm still pretty vehemently against the dual-interface implementation. To re-summarize:
- It's more noisy (code-wise) than just an abstract base class. You've got, what, 5 files? Versus just one for a base class.
- The biggest downside of an abstract base class is all the virtual method calls, which only occur for the abstract/virtual methods on the class. With two interfaces EVERY call is virtual. The performance is WORSE.
- It's not an established pattern that newcomers to the codebase are going to recognize.
| /// (an orchestrator emit triggered same-thread re-entry into the drain loop). Most | ||
| /// orchestrators ignore this. | ||
| /// </param> | ||
| void OnDrainComplete(bool isFinal, bool wasReentrant); |
There was a problem hiding this comment.
Is there a scenario in the other operators that needs this, cause it still doesn't make sense to me. .SynchronizeSafe() already guarantees synchronization of all the notifications coming out of it. Seems to me the only thing this accomplishes is a slight amount of potentially-free batching, which seems not worth the extra complexity to me, because it would only be "dumb" batching.
For example:
.AutoRefresh() desires to do "batching" in that any events that are received from a child stream during the .Subscribe() call to that stream should be ignored. The operator definitely DOESN'T want to emit a Refresh for an item before its Add, and after its Add a Refresh is redundant. OnDrainCompleted() doesn't help accomplish this at all. If .AutoRefresh() were to use OnDrainCompleted() to accomplish this, it'd have to keep track of a lot more potential temporary state (I.E. which refreshes came from source, and need to be emitted, and which refreshes came from children, and need to be ignored), THEN it'd have to iterate over that state a second time, to generate the final changeset. If it just does its own re-entrancy tracking, it can skip that entire process and just put changes directly into the result changeset, or not, as they come in.
Seems to me pretty much EVERY other operator that would use this system would want the same thing: know about re-entrancy WHEN the incoming notification arrives, to decide how to process it, not just queue it up to get processed later.
There was a problem hiding this comment.
Fair question, and I think the name is obscuring the answer. This hook is not really about reentrancy, it is the batch boundary. Three concrete things depend on it today.
1. Coalescing into a single changeset. ChangeSetOrchestrator.OnDrainComplete calls _cache.CaptureChanges() and emits once. Move that to per-notification and every inner value becomes its own changeset. AutoRefreshFixture.Base.ChangeSetBufferIsGiven_PropertyChangedNotificationsAreBufferedOnScheduler pins the current behaviour: two property change notifications raised in separate buffer windows arrive as one changeset with Count == 2 and Refreshes == 2. Emitting per notification turns that into two changesets of one, and the Count().Should().Be(1, "a buffer window expired") assertion no longer holds. The same shape drives OrchestrateManyChangeSets (both orchestrators) and TransformManyAsync, which all call _tracker.EmitChanges(Emitter) at the boundary.
2. isFinal cannot be derived at arrival time. It comes from _subscriptionCounter == 0 once the drain settles. AutoRefresh needs it to flush pending refreshes on the last drain, because stream termination cancels the buffer timer and anything still pending would never be delivered. No per-notification hook can know it is looking at the final one.
3. The de-dup window needs a place to reset. AutoRefresh uses _sourceTouched to suppress a refresh for any key the source itself already touched in the same batch. OnDrainComplete is where that set clears. Without a boundary there is nothing to scope it to.
On wasReentrant you have a point. I went through the implementations and none of them actually read it, so I am happy to drop that parameter and keep isFinal if you agree.
There was a problem hiding this comment.
Following up on the wasReentrant half of this: I am going to keep the parameter after all. Nothing reads it today, but it is cheap to carry and it is the kind of thing an operator may well want once we start adding more of them, so I would rather it be there than have to thread it back through later.
There was a problem hiding this comment.
TL;DR: I'm not sure what form the final DQ implementation will take, but I feel pretty strongly that it won't contain the OnDrainComplete API.
In my book, This level of de-duping is not the responsibility of the DQ, and not really desirable anyway.
We had a discussion a long time ago (I think when the re-entrancy thing came up for .MergeManyChangeSets()) of whether we wanted to implement, let's call it "aggressive batching". I.E. operators should generally publish a single notification "per-frame" of received notifications, where a frame equates roughly to a thread acquiring a lock. If that same thread publishes additional notifications during that lock window (I.E. via nested subscriptions), that's the same frame. The "aggressive batching" approach would have us also batch up additional frames (I.E. notifications that came in during the lock window, but on a different thread) by holding onto the batch if we detect that another thread is waiting by the time we're done, and letting that thread build on it.
We elected to not go for this, and I think that was the right call. Blurring the lines between changesets might have a small performance impact, but it would be VERY minimal, as we're not actually generating or processing fewer changes, just putting the same changes together into fewer collections. It also makes debugging more difficult, I think, when it's less obvious to trace where changes in a changeset came from, as they propagated from upstream.
Yet that's what the "de-duping" here is. It's merging together multiple upstream "frames" into a single downstream "frame". .AutoRefresh() I suppose does represent a unique exception to the above, as we CAN safely reduce the total number of changes, but this is ONLY a thing that can be done for Refresh changes. Really, that's an argument that .AutoRefresh() is SPECIAL and SHOULDN'T use the shared orchestrator logic.
And regardless of the merits of "aggressive batching", I see it as DEFINITELY not the responsibility of the DQ. The DQ's goal and responsibility is synchronization of notifications (without the chance of deadlocking), as made clear by the fact that the operator you wrote for it is .SynchronizeSafe(). It's made even clearer by the AsyncLock class you pointed out, which accomplishes that goal exactly, with a DEAD simple API.
There was a problem hiding this comment.
Agreed, and it's out. The queue on this branch is main's, with no callback, after #1162 and #1163. The orchestration tracks its own delivery frame the way CacheParentSubscription does after #1163. A frame is one upstream notification plus everything delivered synchronously beneath it on the same thread, and the hook is now OnFrameEnded(bool isFinal), called when the outermost frame closes. Work queued by another thread gets its own frame, so the cross-thread batching you objected to is gone. wasReentrant is gone too: nothing read it, and the queue no longer reports it.
On AutoRefresh being special: with frames, it does what you described, deciding at arrival. A refresh can only fall in the same frame as a source change if it was raised synchronously beneath that change, so _sourceTouched is exactly the re-entrancy check, and OnFrameEnded just clears it. Unbuffered, a refresh raised by a property change outside any source change is its own frame, so it goes out as soon as it arrives. The buffered path still needs isFinal: completion cancels the timer, so anything pending has to go out in the final frame.
One more data point: the two WhenPropertyChangedFixture integration tests that were skipped for "AutoRefresh has a separate concurrency bug" fail on main (3 of 3 runs) and pass on this branch (5 of 5), so they are un-skipped here.
| base.OnSourceChangeSet(changes); | ||
| if (changes.Count > 0) | ||
| { | ||
| Emitter.OnNext(changes); |
There was a problem hiding this comment.
Wait.... what? We're using the same drain queue for both upstream and downstream?
There was a problem hiding this comment.
Not quite in the sense it reads. There is one SharedDeliveryQueue, and the emitter is a sub-queue created from it by _queue.CreateQueue(observer). Upstream work (the source subscription, the per-key item sources, and this timer) goes through the gate, while downstream delivery goes through the sub-queue.
Sharing the parent buys two things. First, a single drain thread, so upstream processing cannot race downstream delivery while both are touching orchestrator state such as the ChangeAwareCache or _pendingRefreshes. Second, ordering: the emitter sub-queue is allocated first so it drains last, which is what makes "process all upstream work, then emit" hold. That ordering is what OnDrainComplete keys off, and two independent queues could not express the relationship.
For the timer specifically, routing it through the gate is what stops FlushPending running concurrently with a source notification that is mutating _pendingRefreshes.
There was a problem hiding this comment.
I've genuinely got no idea what this comment was about. There's no drain queue in sight here across the entire file. I guess there WAS some kind of permalink drift? Which is nonsense, cause that defeats the definition of a "permalink". Did you end up ripping the queue references out inot CacheOrchestration?
the emitter sub-queue is allocated first so it drains last, which is what makes "process all upstream work, then emit" hold
If that's important, then we definitely need to make it explicit, which this doesn't. The fact that something gets processed first just because it happens to be allocated first, compared to other things, is an implementation detail.
two independent queues could not express the relationship
Having a single queue doesn't express the relationship either.
There was a problem hiding this comment.
You were right on both counts, and it's gone. Downstream no longer goes through a sub-queue at all. Every hook already runs on the draining thread with the lock released, so the orchestration calls the observer directly. The upstream notifications (the source, the tracked inner observables, and anything passed to Serialize, such as AutoRefresh's buffer timer) go through the one SharedDeliveryQueue, which is now only a serializer.
"Process all upstream work, then emit" is explicit now: an orchestrator emits in OnFrameEnded, which runs when the outermost frame closes. It no longer depends on which sub-queue was allocated first, which #1162 would have broken anyway once delivery order became arrival order.
…esent The previous implementation subscribed the replacement child before taking the previous one's items out, so with list children every item went in twice and the emitted change set carried an Add for something already downstream. MergedListOrchestrator already removes before subscribing and does not have the problem, but nothing was holding it to that. Worth having here specifically because this PR deletes the file the equivalent fix landed in on reactivemarbles#1146, so the test is what carries the behaviour across. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Rename the orchestrator hooks to name what they respond to, per review feedback on reactivemarbles#1114: OnSourceChangeSet -> OnSourceNext OnInner -> OnItemSourceNext Orchestrate -> OrchestrateSubscriptions Also drops the Debug.Assert on the factory result, which nullable analysis already covers. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
The old name suggested a cache of individual changesets, which is not what it is. The type pairs a child changeset stream with a live mirror of the items that child currently holds, so merge operators can resolve a key that more than one child publishes. Also expands the summary to state that purpose directly. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
_queue -> _upstreamQueue _emitter -> _innerQueue Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Agreed on the layering, and I have taken a run at it in #1163. Worth flagging what I found on the way, because it changes the shape of the argument a bit. It is load-bearing. I also think it is not quite the thing you rejected. By your own description a frame is one thread's lock window including anything nested subscriptions publish inside it, and you said operators generally should publish once per frame. That is what the callback does in the common case. Where you are right is that it goes further: The frame-depth version in #1163 keeps the first and drops the second. A second thread's work now gets its own frame. Separately, #1162 fixes the queue delivering out of receipt order. Those two are independent, but the second one turns out to have been hiding the first: with everything in a drain cycle folded into one changeset, the wrong order mostly washed out before anything downstream saw it. Once emission is per frame, delivery order becomes changeset order. That also clears up the |
Takes main's SharedDeliveryQueue (reactivemarbles#1162, reactivemarbles#1163) wholesale, so the queue no longer has a drain-complete callback. Applies the global usings (reactivemarbles#1197) and Polyfills (reactivemarbles#1201) conventions to the files this branch adds or rewrites. The orchestration driver is reworked against the new queue in the next commit.
Replaces the two author-facing interfaces with a single base class and folds the context into the driver: - CacheOrchestrator (abstract) replaces ICacheOrchestrator and CacheOrchestratorBase. Per-change hooks stay virtual, OnItemSourceNext is abstract. - CacheOrchestration (sealed) is the per-subscription driver and is what orchestrators call Track, Untrack and Serialize on, replacing ICacheOrchestratorContext and the private OrchestratorContext. - The TOrch type parameter is gone. It never devirtualized anything for reference types, whose generic code is shared. Batching moves out of the delivery queue, matching reactivemarbles#1163. The driver tracks its own delivery frame and calls OnFrameEnded (was OnDrainComplete) when the outermost frame closes, so one upstream notification and everything it triggers synchronously still produce one emission, while another thread's work gets its own frame. wasReentrant is dropped: nothing read it and the queue no longer reports it. Downstream delivery no longer goes through a sub-queue of the same SharedDeliveryQueue. Hooks already run on the draining thread outside the lock, so the driver calls the observer directly, and "emit after all upstream work" is now explicit (frame end) rather than a consequence of sub-queue allocation order. The orchestrator fixtures now drive each orchestrator through the real driver instead of a fake context. The threaded OrchestrateFixture tests move to a nested IntegrationTests class and are deterministic: the cross-feeding deadlock test parks both writers inside delivery before they feed each other, and no test sleeps or polls. Un-skips the two WhenPropertyChangedFixture AutoRefresh concurrency tests: they fail on main and pass with the rewritten AutoRefresh.
Covers an inner emission reaching downstream (TransformOnObservable, MergeManyChangeSets, MergeMany), a property change becoming a refresh (AutoRefresh), and subscribing and unsubscribing inner observables as items come and go.
MergeMany sits under WhenPropertyChanged, WhenValueChanged and WhenAnyPropertyChanged, so its per-value cost matters. Main's version puts only the inner values through one DeliveryQueue<T>; the orchestrated one also serializes the source through a SharedDeliveryQueue and benchmarked about 1.8x slower per inner value. Restores main's implementation, keeping this branch's behavior change: inner errors now terminate the output, as the operator's docs say. Source errors go through the same queue, so the output cannot be errored twice.
|
I've reworked this along those lines. The two author-facing interfaces are gone:
The core is down from five files to three ( What I kept is the split between the driver and the operator logic. The driver owns the queue, the inner subscriptions, the completion counter and the frame depth; the orchestrator only sees On performance, I added
The branch's first round was noisy (standard deviations of 22 to 45 us on the rows where it trails); the second round is level on every row. Against
The |
What this is
A common pattern in cache operators: for each item in a source changeset, maintain a separate external subscription whose values feed into the operator's output.
MergeManysubscribes to one observable per item.AutoRefreshsubscribes to one re-evaluator per item.FilterOnObservablesubscribes to one bool stream per item.TransformManyAsync,GroupOnObservable, and several others do the same. The shape is universal: 1:1 between source items and external subscriptions, with results flowing back into a single output.Getting that right involves a lot of boilerplate: spin up a subscription on Add/Update, tear it down on Remove, route everything through the same serialization gate to avoid the kind of cross-cache deadlocks that motivated #1097, keep the downstream stream alive until both the source and every per-item subscription have completed, dispose cleanly on error or unsubscribe.
CacheParentSubscriptionwas the previous primitive for cache operators that needed parent/child subscription accounting. It worked, but every consumer had to subclass it and own per-subscription state as subclass fields. That made it easy to leak state across subscribers or forget to wire up the queue-drain machinery from #1097. It also didn't compose: operators wanting parent/child accounting AND aChangeAwareCache, or AND a merge tracker, had to inline both concerns.This PR replaces it with Orchestrator: one primitive that handles subscription accounting, queue routing, completion accounting, and disposal. Operator authors write a small implementation describing what their operator does. The runtime handles how.
MergeManyitself keeps its hand-rolled implementation for performance; see "Performance" below.One base class and a driver:
CacheOrchestrator<TSource, TKey, TInner, TResult>(abstract): what an operator author subclasses.CacheOrchestration<TSource, TKey, TInner, TResult>(sealed): the per-subscription driver. Built once per subscription, it owns the queue, the inner subscriptions and completion accounting, and is what the orchestrator callsTrack,UntrackandSerializeon.Architecture
flowchart TB subgraph public[Public extension methods] ops[AutoRefresh · MergeManyChangeSets ·<br/>FilterOnObservable · TransformManyAsync ·<br/>GroupOnObservable · TransformOnObservable · MergeManyItems ·<br/>MergeManyListChangeSets] end subgraph specializations[Internal specializations] oc[OrchestrateChangeSets<br/>mirror-manipulator with ChangeAwareCache] omc1[OrchestrateManyChangeSets cache<br/>merges child cache changesets] omc2[OrchestrateManyChangeSets list<br/>merges child list changesets] end subgraph base[Internal base entry points] ot[OrchestrateSubscriptions<br/>orchestrator factory] ol[OrchestrateSubscriptions lambdas<br/>convenience overload] end subgraph driver[Driver] co[CacheOrchestration<br/>per-subscription state owner<br/>Track · Untrack · Serialize] end subgraph contract[Contract] cor[CacheOrchestrator<br/>abstract class you subclass] end subgraph runtime[Runtime] sdq[SharedDeliveryQueue<br/>lock-released serialized delivery] end ops --> specializations ops --> ol ops --> ot ops -.subclass.-> cor specializations --> ot ot --> co ol --> co co --> sdq co -.calls.-> cor style public fill:#FFCDD2,stroke:#C62828,color:#000 style specializations fill:#FFE0B2,stroke:#E65100,color:#000 style base fill:#FFF9C4,stroke:#F57F17,color:#000 style driver fill:#C8E6C9,stroke:#2E7D32,color:#000 style contract fill:#BBDEFB,stroke:#1565C0,color:#000 style runtime fill:#E1BEE7,stroke:#6A1B9A,color:#000Types
CacheOrchestrator(you subclass)OnSourceNext(changes)(virtual)OnItemAdded/OnItemUpdated/OnItemRemoved/OnItemRefreshed. Override to handle the whole changeset.OnItemSourceNext(value, key)(abstract)Emitter.OnFrameEnded(isFinal)(virtual)isFinalmeans the source and every tracked inner have completed; emit anything held back now.A delivery frame is one upstream notification plus everything delivered synchronously beneath it on the same thread. Work another thread queues meanwhile gets a frame of its own.
CacheOrchestration(you call, viaContext)Track(key, observable)Untrack(key)Serialize(observable)Lifecycle
%%{init: {"themeVariables": {"sequenceRectStroke": "#000000", "sequenceRectStrokeWidth": "1px"}}}%% sequenceDiagram box rgba(255, 205, 210, 1) Consumer participant Sub as Subscriber end box rgba(255, 224, 178, 1) Driver participant Driver as CacheOrchestration end box rgba(255, 249, 196, 1) Runtime participant Q as SharedDeliveryQueue end box rgba(200, 230, 201, 1) Orchestrator participant Orch as Your orchestrator end box rgba(187, 222, 251, 1) Source participant Src as Source changeset end box rgba(225, 190, 231, 1) Inner participant Inn as Inner observable end Sub->>Driver: Subscribe Driver->>Q: new SharedDeliveryQueue Driver->>Orch: factory(orchestration, emitter) Driver->>Src: SynchronizeSafe(queue).SubscribeSafe rect rgba(245, 245, 245, 1) note over Q: delivery frame Src->>Q: OnNext(changeset) Q->>Orch: OnSourceNext(changeset) Orch->>Driver: Context.Track(key, innerObs) Driver->>Inn: SynchronizeSafe(queue).SubscribeSafe Inn->>Q: OnNext(value) Q->>Orch: OnItemSourceNext(value, key) Driver->>Orch: OnFrameEnded(isFinal) Orch->>Sub: emitter.OnNext(result) endThe Orchestrate family: five entry points
Two base entry points (orchestrator factory and lambda overload) plus three specializations that bake in common output shapes.
Base:
OrchestrateSubscriptionsSame
CacheOrchestrationdriver underneath. Use the factory overload when state and helpers want to live in aCacheOrchestratorsubclass. Use the lambda overload when closures suffice.OrchestrateChangeSets: cache changeset output viaChangeAwareCacheFor filter-like or per-key transform shapes: source is a keyed changeset, output is a keyed changeset, each source key contributes 0 or 1 items to the output. The runtime gives you a
ChangeAwareCachethat you mutate from both source and inner events. It captures and emits once per delivery frame. Used byFilterOnObservableandTransformOnObservable.OrchestrateManyChangeSets(cache): merge child cache changesetsThe shape
MergeManyChangeSetsproduces: per-source-key inner observables that are themselves cache changesets, merged into one cache changeset output. The runtime gives you aChangeSetMergeTrackerthat handles add/update/remove/refresh and conflict resolution via the supplied comparers. Used by bothMergeManyChangeSetsoverloads (vanilla and source-priority).OrchestrateManyChangeSets(list): merge child list changesetsSame shape but for list-typed inner changesets. Disambiguated by parameter type (no destination key). Used by
MergeManyListChangeSets.Picking an entry point
OrchestrateSubscriptionsfactory overload with aCacheOrchestratorsubclassOrchestrateSubscriptionslambda overloadChangeAwareCacheOrchestrateChangeSetsOrchestrateManyChangeSets(cache)OrchestrateManyChangeSets(list)Worked example:
TransformOnObservableThe
CacheParentSubscriptionversion was a nested subclass with its own fields, constructor, andParentOnNext/ChildOnNext/EmitChangesoverrides.Operators migrated
AutoRefreshCacheOrchestratorsubclass (timer state, pending-refresh map)FilterOnObservableOrchestrateChangeSetsGroupOnObservableOrchestrateSubscriptionslambda (closes over aDynamicGrouper)MergeManyMergeManyItemsOrchestrateSubscriptionslambdaMergeManyChangeSets(cache)OrchestrateManyChangeSets(cache)MergeManyChangeSets(source-compare)OrchestrateManyChangeSets(cache) with per-item parent-child boxingMergeManyListChangeSetsOrchestrateManyChangeSets(list)TransformManyAsyncCacheOrchestratorsubclass (per-key cache, merge tracker)TransformOnObservableOrchestrateSubscriptionslambda (closes over aChangeAwareCache)AutoRefresh correctness fixes
The rewrite fixes pre-existing defects that had previously-skipped tests. All of the tests are un-skipped here and pass:
PropertyAccessorIsNull_ThrowsException:propertyAccessoris now validated upfront. Previously threwNullReferenceExceptionon the first source notification.ReevaluatorIsNull_ThrowsException:reEvaluatoris now validated upfront. Same prior failure mode.ReevaluatorEmitsImmediately_ItemDoesNotRefresh([Bug]: AutoRefreshOnObservable produces refreshes for immediately-removed items #1099): a reevaluator that fires synchronously during item subscription no longer produces a redundant Refresh paired with the Add.ReevaluatorThrows_ExceptionPropagates: reevaluator exceptions now propagate toOnErrorinstead of being silently swallowed (Rx contract §6.4).WhenPropertyChangedFixtureintegration testsAutoRefreshThenFilter_ConcurrentAddsAndPropertyActivation_AllItemsObservedandAutoRefreshThenFilter_DualSubscribers_AllItemsObserved): previously skipped as "AutoRefresh has a separate concurrency bug". They fail on main and pass here.Other fixes
AutoRefreshdrops a pending refresh when the source forwards its own Refresh. A reevaluator-queued refresh from a prior frame was being emitted in addition to the source's Refresh for the same key. Symmetrical to the existing pending-drop logic on Update/Remove. Regression test added.isFinalis computed from the subscription counter, not a latched flag. A latched flag races when the orchestrator callsTrackduring theisFinalcallback: counter goes back to 1 but the flag still says true, soOnCompletedfires while a live inner exists. The counter is the source of truth.SubscribeSaferoutes a source that throws on subscribe toOnError, so the constructor has nothing to unwind.MergeManyCacheChangeSetsandMergeManyCacheChangeSetsSourceComparewrapper classes. Inlined their bodies into the publicMergeManyChangeSetsoverloads. The source-compare overload's supporting types now live as private nested generic types in the same partial-class file as the public API.Tests
OrchestrateFixture: the contract. Serialization, per-frame emission, completion accounting,isFinal, error propagation, per-subscription isolation. Its nestedIntegrationTestsclass (run outside the parallel collection) holds the threaded tests, all deterministic with no sleeps or polling: cross-feeding orchestrations that each park inside delivery before feeding the other (the deadlock proof), source and inner hooks never interleaving, and concurrent inner emissions all arriving in order.SharedDeliveryQueueFixture,DeliveryQueueFixture: queue contract.AutoRefreshOrchestratorFixture,ChangeSetOrchestratorFixture,LambdaCacheOrchestratorFixture,MergedOrchestratorFixture,MergedListOrchestratorFixture,TransformManyAsyncOrchestratorFixture. Each drives its orchestrator through the realCacheOrchestration.MergeManyerror-propagation tests inMergeManyWithKeyOverloadFixtureagainst the hand-rolled implementation.Performance
OrchestratedOperators(in the benchmarks project) against main, two interleaved rounds per side, 1,000 items, .NET 9:TransformOnObservableinner valueMergeManyChangeSetsinner valueAutoRefreshproperty changeTransformOnObservableadd 1,000 then clearMergeManyinner value (hand-rolled on both)The PR's first round was noisy; the second is level on every row. Against
CacheParentSubscription(the first two rows) the orchestration is level or faster.MergeManystays hand-rolled. On the orchestration it measured 63 to 76 us per 1,000 inner values against main's 34 to 39 us, about 30 ns more per value. Main's version puts only the inner values through oneDeliveryQueue<T>; the orchestration also serializes the source through aSharedDeliveryQueue.WhenPropertyChanged,WhenValueChangedandWhenAnyPropertyChangedall sit onMergeMany, so it keeps main's implementation. Its only change is the error handling below: inner errors terminate the output, and source errors go through the same queue so the output is never errored twice.Public surface
No new or changed public APIs.
Behavioral change: per-item observable errors now propagate
Several existing public operators previously swallowed errors from per-item observables (the offending item was silently unsubscribed and the merged stream continued). They now forward those errors, terminating the merged stream. This matches the contract of standard
Observable.Merge.Cache side:
MergeManyWhenAnyPropertyChanged(routes throughMergeMany)WhenPropertyChanged(routes throughMergeMany)WhenValueChanged(routes throughMergeMany)AutoRefreshOnObservableList side:
MergeManyWhenAnyPropertyChanged(routes throughMergeMany)WhenPropertyChanged(routes throughMergeMany)WhenValueChanged(routes throughMergeMany)AutoRefreshOnObservableFilterOnObservableXML docs on each operator now describe the new behavior and call out the prior swallowing.
If an application was relying on the old swallowing behavior, the workaround is to defensively
.Catch(...)inside the per-item observable.