WIP Release 9.5 - #1165
Draft
JakenVeina wants to merge 23 commits into
Draft
WIP Release 9.5#1165JakenVeina wants to merge 23 commits into
JakenVeina wants to merge 23 commits into
Conversation
* 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-#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> (cherry picked from commit 8033135)
…1098) RandomPersonGenerator emits Person rows drawn from a finite name pool (~21 girls + ~30 boys cross-joined with 24 lastnames squared). Person.Key is Person.Name, so two independent .Take(10) calls can produce overlapping keys with non-trivial probability. When they collide, the second batch's AddOrUpdate produces 9 Adds + 1 Update instead of 10 Adds, breaking the per-message assertions in: - InvokeLimitSizeToWhenOverLimit - AddMoreThanLimitInBatched Both tests now draw 60 candidates up front, dedupe by Key, take the first 20, and split into two non-overlapping batches of 10. Verified: 50/50 consecutive runs of SizeLimitFixture pass with no failures. Co-authored-by: Darrin Cullop <dacullop@microsoft.com> (cherry picked from commit 87edfa9)
* Break ObservableListEx.cs into per-family partial classes Splits the 2900-line ObservableListEx.cs into 17 smaller partial-class files grouped by operator family. Each method (and all of its overloads) lives in exactly one file. The class declaration is changed to partial; no code, comments, or XML documentation is added, removed, or otherwise modified. All 2218 tests pass. * Rename Pagination to Virtualise and alphabetize list partial members Renames ObservableListEx.Pagination.cs to ObservableListEx.Virtualise.cs for closer parity with the cache equivalent (ObservableCacheEx.VirtualiseAndPage.cs). Sorts members alphabetically within each new partial file; overloads of the same name preserve their original declaration order. * Split ObservableListEx.cs partials into one file per operator (overload set) Mirrors the same convention applied to ObservableCacheEx (PR #1095): 1. ONE FILE PER OPERATOR NAME (one overload set per file). The previous 17 family files are replaced with 63 per-operator partial files. 2. BARE ObservableListEx.cs FILE restored to carry the canonical class-level XML documentation. All partials carry the same canonical class summary ('Extensions for ObservableList.') so SA1601 is satisfied and there are no divergent per-file class docs. 3. PRIVATE HELPERS placed AFTER all public members within their containing file. The 5 private 'Combine' overloads (used by And, Except, Or, Xor) are placed at the bottom of And.cs (alphabetically first caller). The byte content of every method body is preserved (verified programmatically). * Extract Combine private helpers into their own file Per Jake's review feedback, the five Combine private helpers (shared by And, Or, Except, Xor) move from ObservableListEx.And.cs to a dedicated ObservableListEx.Combine.cs, matching the per-operator pattern established for the public surface. Audit confirms Combine is the only multi-caller private helper in ObservableListEx partials. Byte-preserving move with no functional change. Library builds clean on all target frameworks. (cherry picked from commit e6d4e44)
* Break ObservableCacheEx.cs into per-family partial classes
Splits the 6800-line ObservableCacheEx.cs into 24 smaller partial-class files grouped by operator family. Each method (and all of its overloads) lives in exactly one file. No code, comments, or XML documentation is added, removed, or otherwise modified; this is a pure file reorganization. All 2218 tests pass.
* Break ObservableCacheEx.cs into per-family partial classes
Splits the monolithic ObservableCacheEx.cs into 19 smaller partial-class files grouped by operator family. The two pre-existing partials (ObservableCacheEx.SortAndBind.cs, ObservableCacheEx.VirtualiseAndPage.cs) are untouched. Each method (and all of its overloads) lives in exactly one file. No code, XML documentation, comments, preprocessor directives, or constants are added, removed, or otherwise modified. The split was generated programmatically with byte-level per-method equality checks against the original.
* Alphabetize members within new ObservableCacheEx partial files
Sorts members alphabetically by name within each new partial file. Overloads of the same name preserve their original declaration order. Constants sort before methods. Pre-existing partials (SortAndBind, VirtualiseAndPage) are not modified.
* Split ObservableCacheEx.cs partials into one file per operator (overload set)
Addresses PR review feedback:
1. ONE FILE PER OPERATOR NAME (one overload set per file). The previous split
into 19 family files is replaced with 103 per-operator partial files,
matching the existing convention set by ObservableCacheEx.SortAndBind.cs
and ObservableCacheEx.VirtualiseAndPage.cs.
2. BARE ObservableCacheEx.cs FILE restored to carry the canonical class-level
XML documentation. All partials carry the same canonical class summary
('Extensions for dynamic data.') so SA1601 is satisfied and there are no
divergent per-file class docs. SortAndBind.cs and VirtualiseAndPage.cs
were also updated for consistency.
3. PRIVATE HELPERS placed AFTER all public members within their containing
file. Each private helper lives in the alphabetically-first operator file
that calls it:
- Combine -> And.cs (also called by Except, Or, Xor)
- ForForced -> Transform.cs (also called by TransformSafe)
- AdaptSelector -> Group.cs (also called by GroupOnObservable)
- OnChangeAction -> OnItemAdded.cs (also called by OnItem* family)
- TrueFor -> TrueForAll.cs (also called by TrueForAny)
- CreateChangeSetTransformer -> TransformManyAsync.cs (also called by TransformManySafeAsync)
- DefaultResortOnSourceRefresh const -> MergeManyChangeSets.cs
- DefaultSortResetThreshold const -> Sort.cs
The byte content of every method body is preserved (verified programmatically).
#if/#endif preprocessor regions (SUPPORTS_BINDINGLIST in Bind.cs,
SUPPORTS_ASYNC_DISPOSABLE around AsyncDisposeMany) are reconstructed in the
new files.
* Extract shared private helpers into per-helper partial files
Per Jake's review feedback, private helpers used by multiple operators get their
own ObservableCacheEx.{HelperName}.cs file, matching the per-operator pattern
established for the public surface.
Combine -> ObservableCacheEx.Combine.cs (from And.cs)
AdaptSelector -> ObservableCacheEx.AdaptSelector.cs (from Group.cs)
OnChangeAction -> ObservableCacheEx.OnChangeAction.cs (from OnItemAdded.cs)
ForForced -> ObservableCacheEx.ForForced.cs (from Transform.cs)
CreateChangeSetTransformer -> ObservableCacheEx.CreateChangeSetTransformer.cs (from TransformManyAsync.cs)
TrueFor -> ObservableCacheEx.TrueFor.cs (from TrueForAll.cs)
DefaultSortResetThreshold const moved to ObservableCacheEx.cs (the core file).
Audit found it is used by both Sort and SortBy, contrary to the original PR body.
AsyncDisposeMany #if SUPPORTS_ASYNC_DISPOSABLE wrapping replaced with a project-level
Compile Remove. The file body is unconditionally compiled on supported platforms and
excluded entirely on unsupported ones (net4*).
All extractions are byte-preserving moves with no functional change. Builds clean on
all target frameworks (netstandard2.0, net462, net6-net10). Targeted tests pass.
(cherry picked from commit ab5bd6b)
* Fix TOCTOU race in WhenPropertyChanged/WhenValueChanged ObservablePropertyFactory used initial.Concat(events) for both the shallow and deep-chain forms. Concat subscribes to the second source (the PropertyChanged event handler) only AFTER the first (the initial value) completes. Any PropertyChanged notification that fired during that gap was silently dropped. The deep-chain form had an additional gap: Take(1).Repeat tore down all chain notifiers and then re-subscribed via GetNotifiers, losing any events that fired during the re-walk. Fix: 1. Shallow form: rewrite with Observable.Create. Attach the PropertyChanged event handler FIRST so no events are missed during the subscribe window. Use Interlocked.CompareExchange on initialClaimed to ensure exactly one first emission (either the initial or the first handler-fired event, whichever wins the race). A one-shot Interlocked-CAS dedup guard catches the rare setter-update-then-notify duplicate that the CAS cannot otherwise distinguish. 2. Deep-chain form: per-level SerialDisposable. ResubscribeFrom(level) atomically swaps each level's subscription slot to the new value's notifier (subscribe new before disposing old via SerialDisposable.Disposable=). At all times, every live chain level has an active notifier; no re-walk gap. Initial-emit uses the same CAS+dedup pattern as the shallow form. Both fixes are lock-free: only Interlocked.CompareExchange and Volatile read/write. The one-shot dedup guard uses EqualityComparer<TProperty>.Default exactly once per subscription, at the boundary between the initial and the first handler emission, not as a continuous DistinctUntilChanged. Regression tests in WhenPropertyChangedRaceFixture force the race deterministically by parking the observer's OnNext for the initial value while a separate thread mutates the property. Verified RED on main (3 of 4 tests fail), GREEN with fix (4 of 4 pass). Stability check 20/20. Tests: Binding suite 145/145 pass. Full suite 2339/2339 pass (excluding one pre-existing flake unrelated to this branch: SuspendNotificationsFixture.ConcurrentSuspendDuringResumeDoesNotCorrupt which fails on main too). * Deep chain: drainer-based re-walk eliminates concurrent-mutation race The per-level SerialDisposable approach still allowed events to be dropped when two threads concurrently mutated the same intermediate property. Both fired the same notifier; both ResubscribeFrom calls raced; whichever SerialDisposable.Disposable= swap landed last won, even if that thread's pre-walk had read a stale value. The slot would end up subscribed to the LOSER of the property setter race, and subsequent events on the actual current value were lost. Add a single-drainer pattern: notifier handlers signal _minDirtyLevel (Interlocked CAS loop on the minimum dirty level) and the winner of an Interlocked CAS on _drainerActive runs the actual re-walk. Others return immediately. The drainer loops until no signals remain, then re-checks once more to catch signals that arrived during the release. Initial subscription claims the drainer for the duration of ResubscribeFrom(0) + initial Emit so concurrent fires queue and process after. All work serialized through a single thread; no concurrent re-walks possible; the FINAL slot state always reflects the LATEST chain state because the drainer's last iteration always reads the current value. Lock-free. Only Interlocked, Volatile, and SerialDisposable's atomic swap. DeepChain_ConcurrentParentSwap_LeafEventOnWinnerNotDropped: statistical test (500 iterations) that triggered the race in ~10 percent of runs on the prior implementation; now 0 of 500. Stability check 10/10. DeepChain_FiveLevels_MidChainSwap_DeeperLevelsRetargetCorrectly: structural test for the depth-5 mid-chain re-attach case. Tests: Binding suite 147/147 pass. * Simplify deep-chain via recursive Switch composition The drainer pattern was overengineered. The Rx-idiomatic shape for `observe a property chain where each level can be reassigned` is a recursive composition: each level is an ObserveLevel emitting current-then-changes, and the chain is built with .Select(child => deeper).Switch(). When a parent fires, Switch atomically subscribes to the new deeper chain and disposes the old; no SerialDisposable bookkeeping, no min-dirty-level signaling, no CAS-claimed drainer. ObserveLevel attaches the PropertyChanged handler BEFORE reading the initial value (same shallow-form fix), so events fired during the per-level subscribe window are not missed. The outer subscriber still applies the CAS-based first-emission-wins and one-shot dedup at the boundary to handle the initial-emit race. Trade-off: concurrent mutations of the SAME observed property from multiple threads (which is an Rx contract violation by the caller) can leave Switch's lock-acquisition order out of sync with the user's setter-completion order. Well-behaved INPC usage serializes mutations on observed properties; the simplified design relies on that contract. Removed the DeepChain_ConcurrentParentSwap_LeafEventOnWinnerNotDropped test and the block-observer-during-initial-emit deep-chain test (the latter deadlocked against Switch's internal lock by design). Net diff: -242 lines. ObservablePropertyFactory shrank from ~330 lines to ~175. Binding suite 145/145 pass. * Serialize deep-chain via SharedDeliveryQueue Replace the recursive Switch composition with a single SharedDeliveryQueue that funnels two sub-queues: a high-index signal queue carrying level-change notifications, and a low-index emission queue for the user observer. The drainer processes signals first (LIFO), running ResubscribeFrom and Emit serialized against itself, then delivers user emissions last so they observe the latest chain layout. An InitialSetupSignal sentinel funnels the initial chain attachment through the same drainer, closing the subscribe gap without taking a separate lock. Switch is removed entirely: its internal gate held during downstream OnNext deadlocked any observer that blocked synchronously, and adding DeliveryQueue downstream of Switch could not break the cycle. Re-adds the two concurrent regression tests that previously deadlocked or relied on the drainer: - DeepChain_ConcurrentLeafMutationDuringInitialEmit_NotDropped - DeepChain_ConcurrentParentSwap_LeafEventOnWinnerNotDropped (500 iterations) * Address PR feedback: dedup gating, exception routing, test hygiene Production: - Dedup window is now armed only when notifyInitial is true. When the caller didn't ask for an initial value, two consecutive same-valued PropertyChanged events are both legitimate and must both be delivered; the previous code silently dropped the second one. - Wrap the value accessor / chain walk in try/catch and route exceptions to userSub/queue.OnError. The earlier Rx pipeline got this from Select; the new direct invocation needs it explicitly so a throwing property getter doesn't escape the drainer / PropertyChanged invocation thread. Tests: - Add NotifyInitialFalse_DoesNotDedupSameValuedEvents (shallow + deep) covering the dedup gating fix. - Add timeouts to ManualResetEventSlim.Wait so a failed assertion can't park the observer thread indefinitely. Release observerCanContinue in finally. - Replace Thread.Sleep with bounded SpinWait.SpinUntil(condition, timeout) via a WaitForCondition helper. - Capture and dispose the IDisposable returned by Subscribe inside Task.Run so the PropertyChanged handler is detached at test end. - Remove unused subscribeCompleted local. * Tighten regression-test budgets for CI runners Two follow-ups after CI flaked on heavily-loaded shared runners: - Flatten the deep-chain disposable from nested CompositeDisposable to a single composite via collection-expression spread (avoids the redundant inner CompositeDisposable allocation around levelSlots). - Bump the default WaitForCondition timeout from 5s to 30s and route all ManualResetEventSlim / subscribeTask.Wait calls through it. Locally these waits return in <1ms; the larger budget only matters when CI is under heavy load. - Reduce DeepChain_ConcurrentParentSwap_LeafEventOnWinnerNotDropped from 500 to 50 iterations. With SharedDeliveryQueue the outcome is deterministic, so a single iteration proves correctness; 50 is defence in depth. Also drop the unnecessary intermediate WaitForCondition since Task.WaitAll already implies the drainer has fully drained both queued signals. Local: 7/7 race tests pass in ~60ms; 10/10 stability runs clean. * Refactor ObservablePropertyFactory: extract Emitter and DeepChainSubscription Three improvements: - Extract the dedup state machine into a private Emitter : IObserver<T> class. Both factories now wrap their downstream queue in an Emitter; the initialClaimed / dedupArmed / seedValue trio and the PropertyValuesEqual helper live in one place instead of being copy-pasted across two constructors. - Encapsulate the deep-chain runtime in a private DeepChainSubscription : IDisposable class. Fields are default-initialized before the constructor body runs and assigned in well-defined order, which eliminates the DeliverySubQueue<int>? signalSub = null bootstrap (the field is always assigned before any code path that could read it). The InitialSetupSignal sentinel + drainer flow is unchanged. - Reduce the shallow factory to the single-property hot path with a small EmitCurrent helper for the accessor try/catch, removing the second copy of the dedup state machine. Behaviour is unchanged. 148/148 Binding tests pass; race fixture 10/10 stable. * Symmetric SinglePropertySubscription parallel to DeepChainSubscription Extract the shallow-form runtime into a SinglePropertySubscription : IDisposable class with the same shape as DeepChainSubscription: constructor takes (observer, source, [chain-or-name], notifyInitial) and assigns all fields in well-defined order; Dispose tears down handler + queue. The two factories now each become a one-liner Observable.Create that constructs the appropriate subscription. EmitCurrent and OnPropertyChanged become instance methods on SinglePropertySubscription, removing the last shared static helper and keeping all per-subscription state contained. Behaviour unchanged. 148/148 Binding tests pass; race fixture 10/10 stable. * SinglePropertySubscription: route downstream OnNext throws to OnError Match DeepChainSubscription.ProcessSignal's pattern: wrap both the value read AND the emission in try/catch. The downstream observer's OnNext is invoked synchronously by the DeliveryQueue drain, so if it throws, the exception was escaping back out through OnPropertyChanged and into the property setter that fired the event. Route the throw to OnError instead. * Collapse ProcessSignal branches via shared isInitial computation The initial-setup case and the level-fire case only differ in two scalar derivations: (a) where to start the rewalk (0 vs level+1) and (b) whether to emit (always vs only when _notifyInitial). Compute both up front and let the rest of the method be linear. No behaviour change; 148/148 Binding tests pass. * Add multi-threaded torture and AutoRefresh integration tests Two new race fixture tests: 1. DeepChain_FiveLevels_AllLevelsMutatedConcurrently_FinalEmissionMatchesActual Five worker threads each mutate at one level of a depth-5 chain (root subtree swap, mid-level swaps, leaf-int mutations). Many mutations land on detached subtrees and are correctly ignored; mutations on the live chain are processed by the SharedDeliveryQueue drainer in order. After Task.WhenAll the drainer continues until empty; the final emission must equal ReadCurrent() because the last queued signal's ReadCurrent runs against the now-frozen chain state. 50 iterations, 200 mutations per thread, 0 mismatches on every run. 2. AutoRefreshThenFilter_ConcurrentPropertyMutationsOnAddedItems_AllFinalStatesObserved End-to-end: SourceCache + AutoRefresh(IsActive) + Filter(IsActive). Cache pre-populated, then four worker threads concurrently set Activated on every item to a per-item randomized final value. Multiple threads writing the same final value generate many concurrent PropertyChanged invocations per item, exercising SinglePropertySubscription's DeliveryQueue under contention. After the storm the filter contents must match the per-item finalActive map. Deliberately not testing 'mutate while adding' against AutoRefresh: ObservableCache.CreateConnectObservable has the same initial.Concat(_changes) TOCTOU subscribe-window bug as the WhenPropertyChanged shape this PR fixes, and a during-add test would detect that separate cache-side bug as noise unrelated to this PR. Local: 150/150 Binding tests pass; new tests 10/10 stable. * Strengthen deep-chain torture invariants Last-emission-equals-current proves the drainer reached the end of the queue without corruption, but doesn't catch garbage values or Rx contract violations along the way. Add three additional invariants per iteration: 1. ValidateSynchronization() on the subscription chain. Any concurrent OnNext to the user observer (which would indicate a SharedDeliveryQueue serialization bug) throws UnsynchronizedNotificationException during the test instead of silently producing wrong data. 2. Build the set of values any thread could legitimately have written (initial leaf, the leaf-int range, and each subtree-swap range), then assert every emission is in that set. Catches torn reads or stale-detached-subtree mis-reads. 3. First emission must equal the initial value when notifyInitial=true. Catches initial-emit-dropped bugs that the final-state check could mask if the final state happens to equal the initial. What this test still does NOT verify: that every mutation which landed on the live chain produced an emission. That requires causal-history reconstruction which isn't tractable from outside the operator. Local 10/10 stable, ~325ms per run. * Use AsAggregator in the AutoRefresh integration test Replace the manual HashSet + Subscribe(changes => switch on Reason / Add / Remove) plumbing with .AsAggregator(). The aggregator provides Data (IObservableCache) for current contents and Error for terminal exception state, both thread-safe to read. Net effect: ~25 lines of manual change tracking collapse to one line plus assertions against results.Data.Keys. * Remove dedup; route PropertyChanged events without equality guard Drops the one-shot equality dedup in the Emitter and removes the Emitter class entirely. SinglePropertySubscription and DeepChainSubscription now forward every emission through their DeliveryQueue / DeliverySubQueue directly. Same-valued PropertyChanged events that follow the initial emission are delivered as legitimate events; nothing in the property pipeline drops events for equality reasons. Other fixes in the same pass: - TryOnError helpers wrap both EmitCurrent and ProcessSignal so a downstream observer that throws from OnError cannot propagate the secondary exception back into the PropertyChanged setter (shallow) or the SharedDeliveryQueue drainer (deep). - DeepChainSubscription pre-allocates one notifier callback per level in the constructor; ResubscribeFrom indexes into _levelCallbacks instead of allocating a fresh closure per re-walk. - Renamed the existing notifyInitial=false dedup test to PropertyChangedEventsAreNeverDropped_RegardlessOfNotifyInitial and extended it to also cover notifyInitial=true on shallow and deep chains. - Class summary, in-test commentary, and production rationales rewritten to present-tense contracts; removed migration narrative, PR references, and past-bug descriptors per repo comment instructions. 150/150 Binding tests pass; race fixture 10/10 stable. * Address Jake's PR feedback: simplify race tests, split single-threaded tests, let observer throws propagate Production: - EmitCurrent (SinglePropertySubscription) and ProcessSignal (DeepChainSubscription) no longer wrap the downstream OnNext in try/catch. Per the Rx contract, if the user observer throws, the exception propagates back to whoever invoked the PropertyChanged setter (shallow) or back through the SharedDeliveryQueue drainer (deep), matching what a plain Subject<T> would do. The try/catch around the chain walk and accessor stays - those are user code whose throws route to OnError. - TryOnError helpers removed; their swallow-secondary-throw behaviour was non-standard. Tests: - Split WhenPropertyChangedRaceFixture into two fixtures. RaceFixture now contains only the truly multi-threaded tests (5 tests: shallow concurrent mutation during initial emit, deep concurrent leaf mutation during initial emit, deep concurrent parent swap, deep 5-level torture, AutoRefresh integration). The single-threaded contract tests move to a new WhenPropertyChangedBehaviorFixture (7 tests: handler-attach ordering, four no-dedup scenarios split into individual [Fact]s, deep post-swap leaf capture, deep mid-chain swap re-targeting). - The two concurrent initial-emit tests adopt Jake's symmetric Task.WhenAll(subscribe, mutate) shape: observer's OnNext signals + waits, mutator waits then mutates and releases. Removes the manual try/finally + subscribeTask.Result + WaitForCondition plumbing. 153/153 Binding tests pass; the property-changed fixtures run 10/10 stable. * Adopt Jake's exact implementation for Shallow_ConcurrentMutationDuringInitialEmit_NotDropped Replaces the existing test body with Jake's verbatim code from the PR review: named-argument style with column-aligned colons, Item class with Id and Value, observedValues / propertyValue naming, BeEquivalentTo with WithStrictOrdering and the original because string. Removes TestModel from the race fixture (no longer used). * Drop cache-side TOCTOU rationale from integration test comment The cache-side observation is unrelated to this PR. The integration test pre-populates the cache to keep what's being verified focused on the WhenPropertyChanged path under multi-threaded property contention; that's what the comment should say. * Skip AutoRefresh+Filter integration tests; add dual-subscriber variant Both AutoRefresh+Filter integration variants reproduce a race that lives in AutoRefresh's internal Publish multicast: the Filter path reads the property value before MergeMany subscribes the per-item refresh handler, so a concurrent property mutation in that gap is dropped. AutoRefresh calls WhenPropertyChanged with notifyInitial=false, so the per-item subscribe is not the source of the race. Both tests fail equally on upstream main and on this branch; mark them [Fact(Skip)] so the scenarios are preserved without breaking the build, and track the AutoRefresh fix separately. Also: KeyedActivable now only raises PropertyChanged on actual value change (standard MVVM semantics), so a dropped transition is unrecoverable, matching real consumer patterns. (cherry picked from commit 5f44d05)
JakenVeina
force-pushed
the
release-wip/9.5
branch
from
August 8, 2026 04:38
5eb6787 to
fc6c09e
Compare
…rent (#1113) When an Update changeset entry has Previous matching the predicate and Current not matching, FilterImmutable emits a Remove. Previously this Remove carried the new (non-matching) value as Current, violating the Change<T,K> contract that Remove.Current is the item being removed (the item that just left downstream). Since the new value never reached downstream, only the previous value can satisfy this contract. Consumers that read Current on Remove (e.g. composition with TransformImmutable, side-effect handlers like DisposeMany or OnItemRemoved equivalents) received the wrong reference, silently producing incorrect results or InvalidCastException. (cherry picked from commit 6d2144c)
* test(sum): add more sum tests * chore: test renames * test(sum): split `SumFixture` into separate partial classes for cache and list sources (cherry picked from commit b1cb9a1)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> (cherry picked from commit 5610bb4)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> (cherry picked from commit aaa4ed1)
- Replace the glennawatson/ChangeLog action with the GitReleaseNoteGenerator global tool (git-release-notes) to produce release notes, matching the reactiveui pipeline. Resolves #1093. - Swap the stale dotnet/nbgv JS action for the nbgv global tool, stamping cloud variables via `nbgv cloud -a` and exposing SemVer2/PrereleaseVersion as step outputs. - Keep Nerdbank.GitVersioning (version.json) as the version source; all branch/version policy checks are unchanged. (cherry picked from commit f933ae5)
* Add RemoveKey tests showing issue with Refresh and Filter * Fix index out of range issue with static List Filter * Remove unused variable in RemoveKeyFixture * Remove unused variable in RemoveKeyFixture (really) * Revert formatting to previous in Filter.Static.cs * Add unit test changing order or RemoveKey call * Update RemoveKey test names based on PR feedback * Remove extraneous comment per PR feedback * Move Filter-related RemoveKey tests to FilterFixture --------- Co-authored-by: John Cummings <jcummings2sf@gmail.com> (cherry picked from commit d26b63c)
(cherry picked from commit dfef239)
…esumeNotifications()` (#1132) * Fixed that `ObservableCache.SuspendNotifications()` uses a `lock` to synchronize updates to internal state for tracking notification suspensions, but `.ResumeNotifications()` did not use any synchronization when updating that same state, allowing for corruption of state if the two methods happen to overlap, concurrently. Also fixed that the test for this behavior was intermittent, specifically that it tended to falsely pass when running in parallel with many other tests. The test has now been moved to its own `IntegrationTests` fixture, to guarantee it is never parallelized with other tests. Resolves #1131. * Deliver resumed notifications off the lock, closing the #1131 race without holding it ResumeNotifications() was wrapped in lock(_locker), which via lock reentrancy held the cache lock across the DeliveryQueue drain, delivering accumulated changes to subscribers while the lock was held. Restore off-lock delivery and close the suspend/resume state-divergence race a different way: - SuspendNotifications() no longer wraps the resume callback in the cache lock; queued changes drain outside the lock, as the DeliveryQueue intends. - ResumeNotifications() re-checks the suspend count under the lock before emitting the resume signal, so a concurrent SuspendNotifications() in the decrement/emit window wins and the count and the subject cannot diverge. - SuspensionTracker.SuspendNotifications() gates the suspended signal on the subject's own value rather than the count, keeping it monotonic while a resume's signal is still in flight. Tests: - Add StaleResumeSignalIsSuppressedByConcurrentReSuspend: deterministic proof that a connection made during a racing re-suspend does not activate on a stale resume signal (fails without the count re-check). - Add ResumeDeliversPendingChangesWithoutHoldingTheLock: proves a blocked subscriber does not stall a concurrent lock-requiring operation during resume delivery. - Remove ResumeSignalUnderLockPreventsStaleSnapshotFromReSuspend: redundant coverage that only passed by timing out a deadlock. * Adjusted SuspendNotificationsFixture.UnitTests, for consistent styling. * Moved an additional test exercising concurrency, that got missed, from SuspendNotificationsFixture.UnitTests to SuspendNotificationsFixture.IntegrationTests. * Close the suspend/resume race by emitting the resume signal in a single lock acquisition ResumeNotifications() was acquiring the lock twice: once to decrement the suspend count and queue the accumulated changes, and again to emit the resume signal. The gap between those two acquisitions was the race window. A SuspendNotifications() landing in it would see a zero count and skip emitting 'true', and the resume would then emit 'false' on top of it, leaving the count and the subject in disagreement. Emitting the signal inside the same scope that cleared the count removes the window entirely, rather than detecting and compensating for it after the fact. Delivery is still performed off the lock: ScopedAccess.Dispose() calls ExitLockAndDeliver(), which releases the lock before draining the queue. The signal now fires before the pending changes are delivered instead of after, which a deferred subscriber handles via the existing HasPending/_currentDeliveryVersion guard in CreateConnectObservable. Also guard EmitResumeNotification() against a disposed subject, since SuspensionTracker.Dispose() runs from the delivery callbacks off the lock and can tear the subject down underneath a concurrent resume. With the window closed, ResumeSignalUnderLockPreventsStaleSnapshotFromReSuspend is restored and passes alongside StaleResumeSignalIsSuppressedByConcurrentReSuspend. --------- Co-authored-by: Darrin W. Cullop <Darrin.Cullop@microsoft.com> (cherry picked from commit 4f0b457)
* Make every BatchIf overload shape resolve source.BatchIf(pause) did not compile. Nor did source.BatchIf(pause, true). Five overloads, and the parameters that tell them apart all carried defaults, so a short call was applicable to several at once and none was better. Dropped the defaults from those parameters, which is what the list side's BufferIf already does with the same five shapes. Not a breaking change in practice: every call form that leaned on those defaults is ambiguous today, so nothing that compiles can depend on them. It also sorts out the named forms, BatchIf(pause, initialPauseState: true) among them. The cast workaround this forced is already in the test suite, in CrossCacheDeadlockStressTest and DeadlockTortureTest, both written as BatchIf(pause, false, (TimeSpan?)null) because the natural call does not build. Fixes #1152. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Move the BatchIf overload tests into BatchIfFixture They belong with the rest of the operator's coverage rather than in a fixture of their own, and they can use the source and scheduler already set up there. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Add two-parameter BatchIf overload and restore API snapshot encoding Adds an explicit BatchIf(source, pauseIfTrueSelector) overload so the shortest call form binds to a dedicated method rather than relying on the scheduler overload's default. Removes the overload-shape test. Asserting API shape by hand is the wrong tool; the API approval snapshot already covers it. The API snapshots had picked up a stripped UTF-8 BOM, LF to CRLF conversion, and an added trailing newline from being regenerated by hand. Rebuilt from main so the diff is only the signature changes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> (cherry picked from commit 7408226)
A Connect() or Watch() made while notifications are suspended is deferred until the suspension lifts. When the cache's source failed, that deferral reported the failure as a successful completion, so the subscriber's error handling never ran and its data looked complete. Two things caused it. The suspension tracker was disposed on failure, which completes the subject that the deferral is waiting on, and the deferral mapped that completion straight onto observer.OnCompleted. Faulting the subject instead carries the exception, and dropping the Do lets the terminal event travel through the chain rather than bypassing it. SelectMany replaces Select followed by Switch. Take(1) means there is only ever one inner sequence, so the two are equivalent for delivery, but SelectMany propagates the gate's terminal event by itself. That leaves the deferral self-contained rather than dependent on the behaviour of another operator. (cherry picked from commit 6562004)
* Fix that the cache Switch operator never completes, without taking a lock Switch could never complete, could retain data from a source it had already switched away from, and routed delivery through a lock. It relayed changes through a private LockFreeObservableCache. That cache only ends when it is disposed, so the terminal event of the source had nowhere to go. Errors were carried across by hand through a merged subject; completion had no equivalent path and was silently dropped. A consumer never received OnCompleted, ever. Switching is now explicit rather than delegated to Observable.Switch, which holds its gate for the whole of the downstream OnNext call. A pipeline that crosses into another cache runs that work under the gate, and a producer on another thread blocks behind it, which is the cross cache deadlock shape the delivery queue exists to avoid. Measured against a subscriber that blocks inside OnNext, writing to the source from another thread took 748ms through Observable.Switch and 1ms through the queue, which enqueues and returns. A SerialDisposable holds the current inner subscription, and each one carries an identity, so a superseded source still delivering concurrently is dropped rather than applied on top of the state its replacement has already established. That is the second defect above. All state changes and all delivery happen through the queue lock, which is released before anything is handed downstream. The result completes once the sources and the current inner have both completed, and fails as soon as either does. * Ignore errors from a superseded source in the cache Switch operator OnNext and OnCompleted both check the captured id against the active source, because a source that has been switched away from may still be mid-delivery. OnError went straight to the queue without that check, so a late failure from a source no longer selected could terminate the output. Rx's own Switch discards it, so the hand-rolled version was a regression on that point. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Make the superseded-source test actually exercise the guard The first version subscribed a Subject and failed it after the switch, but SerialDisposable had already disposed that subscription, so Rx suppressed the notification and the test passed with or without the guard. Disposal cannot reach a notification already in flight, which is the case the guard exists for. RawAnonymousObservable hands back the observer directly, so the failure can be delivered after the switch without needing a race to land. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Address review feedback Rename active to activeSourceId and id to sourceId, which say what they are. Use SubscribeSafe throughout, so a source that throws out of its subscribe call is reported through the queue rather than escaping to whoever happened to be subscribing. Replace the CompositeDisposable with an explicit closure. Disposal order matters here, the queue has to drain before the subscriptions feeding it go away, and CompositeDisposable does not specify an order. Drop OnCompletedFiresIfCacheDisposedAfterConnectingWhileSuspended, which is covered more thoroughly by the test in #1145. --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> (cherry picked from commit 41b4d9d)
* feat: add Sum operator Cache benchmarks * feat: add Sum operator List benchmarks * fix: actually add multiple counts for tests * fix: consolidate seed creation (cherry picked from commit be60da6)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
6 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Backporting fixes and non-breaking enhancements from main to 9.x, for a 9.5 release.