From 4bfa6ae40fdfb015f60e868cfd02277009d2b43a Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 07:19:08 -0700 Subject: [PATCH 1/9] Make the cache operators honour the Rx observable contract Adds a conformance sweep that drives every operator overload with a source that completes or errors and records what the operator actually emits, then fixes what it found on the cache side. The sweep exists because Switch was found to silently drop OnCompleted, which raised the obvious question of how many others do the same. It reaches 476 of the 499 extension methods that hang off IObservable; the rest do not return an observable so there is nothing to assert. Overloads are driven individually, because they frequently do not share an implementation: Sort(comparer) completes correctly while Sort(observableComparer) does not. Terminal events are delivered both during and after subscription, since several operators only misbehave when the terminal arrives synchronously. The approved files record behaviour rather than intent, so any diff means behaviour moved: approve it as part of a fix, or treat it as a bug report. Four causes account for nearly all of it. Observable.Never was used as the default for control signals the caller did not supply, then merged with the data. Never says a signal might still arrive, so the merge could never complete. Empty is what is meant when nothing was supplied. This affected Sort, GroupOn and GroupOnImmutable. Several operators subscribed with only an OnNext handler and dropped terminal events outright: StatusMonitor, BatchIf and the combiners. BatchIf now also flushes anything held back by a pause before finishing, the way Buffer does. Internal side-effect subscriptions used a bare Subscribe(), and Rx rethrows an unhandled OnError out of the subscription. That is what threw from Group, Bind and DynamicCombiner. Terminal events do not survive an intermediate AsObservableCache, so TreeBuilder and SpecifiedGrouper never saw completion or failure at all. MergeMany discarded errors raised by its inner sequences, so a failing child was silently dropped. Rx's Merge propagates them, and MergeManyItems and DynamicCombiner inherit the fix. Three existing tests asserted the old behaviour and have been rewritten to the contract. And, Or, Except and Xor now complete once every source has and fail as soon as any one does, which is what they mean read as set operations over live sources. DeferUntilLoaded, SkipInitial, BufferInitial and the four JoinMany operators are built on the above and are fixed as a consequence. GroupOnProperty no longer completes while items are present, matching GroupOnPropertyWithImmutableState and AutoRefresh, because its regrouper watches live items. --- src/DynamicData.Tests/Cache/AndFixture.cs | 34 ++- src/DynamicData.Tests/Cache/BatchIfFixture.cs | 35 ++- .../Cache/CombinerCompletionFixture.cs | 73 +++++ .../Cache/DeferUntilLoadedFixture.cs | 32 ++- src/DynamicData.Tests/Cache/ExceptFixture.cs | 34 ++- src/DynamicData.Tests/Cache/GroupFixture.cs | 44 +++ .../Cache/GroupImmutableFixture.cs | 17 ++ .../Cache/InnerJoinFixture.cs | 23 +- .../Cache/MergeManyFixture.cs | 21 +- .../Cache/MergeManyInnerErrorFixture.cs | 47 +++ .../Cache/MergeManyItemsFixture.cs | 47 ++- .../Cache/MergeManyWithKeyOverloadFixture.cs | 10 +- .../Cache/MonitorStatusFixture.cs | 45 +++ .../Cache/OperatorCompletionFixture.cs | 268 ++++++++++++++++++ src/DynamicData.Tests/Cache/OrFixture.cs | 34 ++- src/DynamicData.Tests/Cache/SortFixture.cs | 30 +- .../Cache/TransformTreeFixture.cs | 29 ++ src/DynamicData.Tests/Cache/XorFixture.cs | 34 ++- src/DynamicData/Binding/BindPaged.cs | 46 ++- src/DynamicData/Binding/BindVirtualized.cs | 46 ++- src/DynamicData/Binding/SortAndBind.cs | 12 +- src/DynamicData/Cache/Internal/BatchIf.cs | 7 + src/DynamicData/Cache/Internal/Combiner.cs | 22 +- .../Cache/Internal/DynamicCombiner.cs | 14 +- src/DynamicData/Cache/Internal/GroupOn.cs | 11 +- .../Cache/Internal/GroupOnImmutable.cs | 4 +- src/DynamicData/Cache/Internal/MergeMany.cs | 2 +- .../Cache/Internal/MergeManyItems.cs | 7 +- src/DynamicData/Cache/Internal/Sort.cs | 5 +- .../Cache/Internal/SpecifiedGrouper.cs | 7 +- .../Cache/Internal/StatusMonitor.cs | 57 +--- src/DynamicData/Cache/Internal/TreeBuilder.cs | 5 +- .../Cache/ObservableCacheEx.Combine.cs | 6 +- 33 files changed, 1005 insertions(+), 103 deletions(-) create mode 100644 src/DynamicData.Tests/Cache/CombinerCompletionFixture.cs create mode 100644 src/DynamicData.Tests/Cache/MergeManyInnerErrorFixture.cs create mode 100644 src/DynamicData.Tests/Cache/OperatorCompletionFixture.cs diff --git a/src/DynamicData.Tests/Cache/AndFixture.cs b/src/DynamicData.Tests/Cache/AndFixture.cs index 0f0ec2ad0..18a7b84ba 100644 --- a/src/DynamicData.Tests/Cache/AndFixture.cs +++ b/src/DynamicData.Tests/Cache/AndFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; @@ -7,6 +7,8 @@ using FluentAssertions; using Xunit; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -106,4 +108,34 @@ public void UpdatingOneSourceOnlyProducesNoResults() } protected abstract IObservable> CreateObservable(); + + [Fact] + public void CompletesOnlyWhenEverySourceCompletes() + { + var completed = false; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = ObservableCacheEx.And(first, second).Subscribe(_ => { }, () => completed = true); + + first.OnCompleted(); + completed.Should().BeFalse("the second source is still live"); + + second.OnCompleted(); + completed.Should().BeTrue("every source has now finished"); + } + + [Fact] + public void DeliversAnErrorFromAnySource() + { + Exception? error = null; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = ObservableCacheEx.And(first, second).Subscribe(_ => { }, ex => error = ex, () => { }); + + second.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/BatchIfFixture.cs b/src/DynamicData.Tests/Cache/BatchIfFixture.cs index 87f47356e..f5064c39c 100644 --- a/src/DynamicData.Tests/Cache/BatchIfFixture.cs +++ b/src/DynamicData.Tests/Cache/BatchIfFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Reactive.Linq; using System.Reactive.Subjects; @@ -9,6 +9,9 @@ using Microsoft.Reactive.Testing; using Xunit; +using System.Collections.Generic; +using System.Reactive; +using System.Reactive.Concurrency; namespace DynamicData.Tests.Cache; @@ -140,4 +143,34 @@ public void ResultsWillBeReceivedIfNotPaused() _scheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); _results.Messages.Count.Should().Be(1, "Should be 1 update"); } + + [Fact] + public void CompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.BatchIf(Observable.Return(false), Scheduler.Immediate).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void FlushesHeldChangesBeforeCompleting() + { + var received = 0; + var completed = false; + + using var source = new Subject>(); + using var pause = new BehaviorSubject(true); + using var subscription = source.BatchIf(pause, Scheduler.Immediate).Subscribe(_ => received++, () => completed = true); + + source.OnNext(new ChangeSet { new(ChangeReason.Add, "a", new Person("a", 1)) }); + source.OnCompleted(); + + received.Should().Be(1, "changes held back by the pause would otherwise be lost"); + completed.Should().BeTrue(); + } } diff --git a/src/DynamicData.Tests/Cache/CombinerCompletionFixture.cs b/src/DynamicData.Tests/Cache/CombinerCompletionFixture.cs new file mode 100644 index 000000000..df5ea398f --- /dev/null +++ b/src/DynamicData.Tests/Cache/CombinerCompletionFixture.cs @@ -0,0 +1,73 @@ +using System; +using System.Reactive.Linq; +using System.Reactive.Subjects; + +using DynamicData.Tests.Domain; + +using FluentAssertions; + +using Xunit; + +namespace DynamicData.Tests.Cache; + +/// +/// Terminal event behaviour for the combining operators. +/// +public class CombinerCompletionFixture +{ + [Fact] + public void CombinersCompleteWhenEverySourceCompletes() + { + foreach (var combine in new Func>, IObservable>, IObservable>>[] + { + static (a, b) => ObservableCacheEx.And(a, b), + static (a, b) => a.Or(b), + static (a, b) => a.Except(b), + static (a, b) => a.Xor(b), + }) + { + using var first = new Subject>(); + using var second = new Subject>(); + var completed = false; + + using var subscription = combine(first, second).Subscribe(_ => { }, () => completed = true); + + first.OnCompleted(); + completed.Should().BeFalse("the second source is still live"); + + second.OnCompleted(); + completed.Should().BeTrue("every source has now finished"); + } + } + + [Fact] + public void CombinersDeliverErrorFromAnySource() + { + using var first = new Subject>(); + using var second = new Subject>(); + Exception? error = null; + + using var subscription = first.Or(second).Subscribe(_ => { }, ex => error = ex, () => { }); + + second.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } + + [Fact] + public void DynamicCombinerCompletesWhenEverySourceCompletes() + { + using var sources = new SourceList>>(); + using var first = new Subject>(); + var completed = false; + + sources.Add(first); + + using var subscription = sources.Or().Subscribe(_ => { }, () => completed = true); + + first.OnCompleted(); + sources.Dispose(); + + completed.Should().BeTrue(); + } +} diff --git a/src/DynamicData.Tests/Cache/DeferUntilLoadedFixture.cs b/src/DynamicData.Tests/Cache/DeferUntilLoadedFixture.cs index 1e19ba079..8e26af3bf 100644 --- a/src/DynamicData.Tests/Cache/DeferUntilLoadedFixture.cs +++ b/src/DynamicData.Tests/Cache/DeferUntilLoadedFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Linq; using DynamicData.Tests.Domain; @@ -6,6 +6,10 @@ using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -61,4 +65,30 @@ public void SkipInitialDoesNotReturnTheFirstBatchOfData() updateReceived.Should().BeTrue(); deferStream.Dispose(); } + + [Fact] + public void DeferUntilLoadedCompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.DeferUntilLoaded().Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void SkipInitialCompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.SkipInitial().Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("SkipInitial is built on DeferUntilLoaded"); + } } diff --git a/src/DynamicData.Tests/Cache/ExceptFixture.cs b/src/DynamicData.Tests/Cache/ExceptFixture.cs index 17c4996e8..18fbb4341 100644 --- a/src/DynamicData.Tests/Cache/ExceptFixture.cs +++ b/src/DynamicData.Tests/Cache/ExceptFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using DynamicData.Tests.Domain; @@ -6,6 +6,8 @@ using FluentAssertions; using Xunit; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -80,4 +82,34 @@ public void UpdatingOneSourceOnlyProducesResult() } protected abstract IObservable> CreateObservable(); + + [Fact] + public void CompletesOnlyWhenEverySourceCompletes() + { + var completed = false; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Except(second).Subscribe(_ => { }, () => completed = true); + + first.OnCompleted(); + completed.Should().BeFalse("the second source is still live"); + + second.OnCompleted(); + completed.Should().BeTrue("every source has now finished"); + } + + [Fact] + public void DeliversAnErrorFromAnySource() + { + Exception? error = null; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Except(second).Subscribe(_ => { }, ex => error = ex, () => { }); + + second.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/GroupFixture.cs b/src/DynamicData.Tests/Cache/GroupFixture.cs index c0139eed5..280d01841 100644 --- a/src/DynamicData.Tests/Cache/GroupFixture.cs +++ b/src/DynamicData.Tests/Cache/GroupFixture.cs @@ -9,6 +9,9 @@ using FluentAssertions; using Xunit; +using System.Reactive; +using System.Reactive.Concurrency; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -264,4 +267,45 @@ public class GroupViewModel public ReadOnlyObservableCollection Entries => _entries; } + + [Fact] + public void CompletesWhenNoRegrouperIsSupplied() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.Group(p => p.Age).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("an absent regrouper can never fire and so must not hold the result open"); + } + + [Fact] + public void DeliversTheErrorWithoutThrowing() + { + Exception? error = null; + + using var source = new Subject>(); + using var subscription = source.Group(p => p.Age).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } + + [Fact] + public void DeliversTheErrorWhenAResultGroupSourceIsSupplied() + { + Exception? error = null; + + using var source = new Subject>(); + using var subscription = source + .Group(p => p.Age, Observable.Never>()) + .Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/GroupImmutableFixture.cs b/src/DynamicData.Tests/Cache/GroupImmutableFixture.cs index 8640bccb7..05c5c0692 100644 --- a/src/DynamicData.Tests/Cache/GroupImmutableFixture.cs +++ b/src/DynamicData.Tests/Cache/GroupImmutableFixture.cs @@ -7,6 +7,10 @@ using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -177,4 +181,17 @@ public void UpdatesArePermissible() var group = _results.Data.Items[0]; group.Count.Should().Be(2); } + + [Fact] + public void CompletesWhenNoRegrouperIsSupplied() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.GroupWithImmutableState(p => p.Age).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("an absent regrouper can never fire and so must not hold the result open"); + } } diff --git a/src/DynamicData.Tests/Cache/InnerJoinFixture.cs b/src/DynamicData.Tests/Cache/InnerJoinFixture.cs index bd5ffc054..9d51f4048 100644 --- a/src/DynamicData.Tests/Cache/InnerJoinFixture.cs +++ b/src/DynamicData.Tests/Cache/InnerJoinFixture.cs @@ -1,10 +1,16 @@ -using System; +using System; using DynamicData.Tests.Utilities; using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.Cache; @@ -457,4 +463,19 @@ public override bool Equals(object? obj) public override string ToString() => $"{Key}: {Device} ({MetaData})"; } + + [Fact] + public void InnerJoinManyCompletesWhenBothSidesComplete() + { + var completed = false; + + using var left = new Subject>(); + using var subscription = left + .InnerJoinMany(Observable.Empty>(), p => p.Name, (_, person, _) => person) + .Subscribe(_ => { }, () => completed = true); + + left.OnCompleted(); + + completed.Should().BeTrue("the grouping it is built on must not hold the result open"); + } } diff --git a/src/DynamicData.Tests/Cache/MergeManyFixture.cs b/src/DynamicData.Tests/Cache/MergeManyFixture.cs index 553bf452e..3f6b962a0 100644 --- a/src/DynamicData.Tests/Cache/MergeManyFixture.cs +++ b/src/DynamicData.Tests/Cache/MergeManyFixture.cs @@ -1,10 +1,14 @@ -using System; +using System; using System.Reactive.Linq; using System.Reactive.Subjects; using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive; +using System.Reactive.Concurrency; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.Cache; @@ -83,4 +87,19 @@ public void InvokeObservable(bool value) _changed.OnNext(value); } } + + [Fact] + public void DeliversAnErrorRaisedByAChild() + { + Exception? error = null; + + using var source = new SourceCache(p => p.Name); + using var child = new Subject(); + using var subscription = source.Connect().MergeMany(_ => child).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.AddOrUpdate(new Person("a", 1)); + child.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType("Merge propagates a failure from any inner stream"); + } } diff --git a/src/DynamicData.Tests/Cache/MergeManyInnerErrorFixture.cs b/src/DynamicData.Tests/Cache/MergeManyInnerErrorFixture.cs new file mode 100644 index 000000000..8cc4c2301 --- /dev/null +++ b/src/DynamicData.Tests/Cache/MergeManyInnerErrorFixture.cs @@ -0,0 +1,47 @@ +using System; +using System.Reactive.Subjects; + +using DynamicData.Tests.Domain; + +using FluentAssertions; + +using Xunit; + +namespace DynamicData.Tests.Cache; + +/// +/// Merge propagates a failure from any inner stream, rather than discarding it. +/// +public class MergeManyInnerErrorFixture +{ + [Fact] + public void MergeManyDeliversErrorFromAChild() + { + using var source = new SourceCache(p => p.Name); + using var child = new Subject(); + Exception? error = null; + + using var subscription = source.Connect().MergeMany(_ => child).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.AddOrUpdate(new Person("a", 1)); + child.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType("a failing inner stream must not be silently discarded"); + } + + [Fact] + public void MergeManyItemsDeliversErrorFromAChild() + { + using var source = new SourceCache(p => p.Name); + using var child = new Subject(); + Exception? error = null; + + using var subscription = source.Connect().MergeManyItems(_ => child).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.AddOrUpdate(new Person("a", 1)); + child.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } + +} diff --git a/src/DynamicData.Tests/Cache/MergeManyItemsFixture.cs b/src/DynamicData.Tests/Cache/MergeManyItemsFixture.cs index e2a6be748..01ce186a8 100644 --- a/src/DynamicData.Tests/Cache/MergeManyItemsFixture.cs +++ b/src/DynamicData.Tests/Cache/MergeManyItemsFixture.cs @@ -1,10 +1,14 @@ -using System; +using System; using System.Reactive.Linq; using System.Reactive.Subjects; using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive; +using System.Reactive.Concurrency; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.Cache; @@ -95,4 +99,45 @@ public void InvokeObservable(bool value) _changed.OnNext(value); } } + + [Fact] + public void CompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.MergeManyItems(_ => Observable.Empty()).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void StaysOpenWhenOnlyAChildCompletes() + { + var completed = false; + + using var source = new SourceCache(p => p.Name); + using var subscription = source.Connect().MergeManyItems(_ => Observable.Return(1)).Subscribe(_ => { }, () => completed = true); + + source.AddOrUpdate(new Person("a", 1)); + + completed.Should().BeFalse("one child finishing does not finish the merge"); + } + + [Fact] + public void DeliversAnErrorRaisedByAChild() + { + Exception? error = null; + + using var source = new SourceCache(p => p.Name); + using var child = new Subject(); + using var subscription = source.Connect().MergeManyItems(_ => child).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.AddOrUpdate(new Person("a", 1)); + child.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/MergeManyWithKeyOverloadFixture.cs b/src/DynamicData.Tests/Cache/MergeManyWithKeyOverloadFixture.cs index e6987e846..abbc3750e 100644 --- a/src/DynamicData.Tests/Cache/MergeManyWithKeyOverloadFixture.cs +++ b/src/DynamicData.Tests/Cache/MergeManyWithKeyOverloadFixture.cs @@ -82,7 +82,7 @@ public void SingleItemCompleteWillNotMergedStream() } [Fact] - public void SingleItemFailWillNotFailMergedStream() + public void SingleItemFailWillFailMergedStream() { var failed = false; var stream = _source.Connect().MergeMany((o, key) => o.Observable).Subscribe(_ => { }, ex => failed = true); @@ -94,7 +94,7 @@ public void SingleItemFailWillNotFailMergedStream() stream.Dispose(); - failed.Should().BeFalse(); + failed.Should().BeTrue("Merge propagates a failure from any inner stream"); } /// @@ -144,7 +144,7 @@ public void MergedStreamCompletesWhenSourceAndItemsComplete() /// Stream completes even if one of the children fails. /// [Fact] - public void MergedStreamCompletesIfLastItemFails() + public void MergedStreamFailsIfLastItemFails() { var receivedError = default(Exception); var streamCompleted = false; @@ -159,9 +159,9 @@ public void MergedStreamCompletesIfLastItemFails() _source.Dispose(); item.FailObservable(new Exception("Test exception")); - receivedError.Should().Be(default); sourceCompleted.Should().BeTrue(); - streamCompleted.Should().BeTrue(); + receivedError.Should().NotBeNull("Merge propagates a failure from any inner stream"); + streamCompleted.Should().BeFalse("a failure and a completion are mutually exclusive"); } /// diff --git a/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs b/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs index 6a6a33a38..e127256e1 100644 --- a/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs +++ b/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs @@ -7,6 +7,9 @@ using FluentAssertions; using Xunit; +using System.Collections.Generic; +using System.Reactive.Concurrency; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.Cache; @@ -91,4 +94,46 @@ public void SetToLoaded() status.Should().Be(ConnectionStatus.Loaded, "Status should be ConnectionStatus.Loaded"); subscription.Dispose(); } + + [Fact] + public void CompletesWhenTheSourceCompletes() + { + var statuses = new List(); + var completed = false; + + using var source = new Subject(); + using var subscription = source.MonitorStatus().Subscribe(statuses.Add, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("the status stream is finished once the source is"); + statuses.Should().EndWith(ConnectionStatus.Completed); + } + + [Fact] + public void DeliversTheErrorAfterReportingIt() + { + var statuses = new List(); + Exception? error = null; + + using var source = new Subject(); + using var subscription = source.MonitorStatus().Subscribe(statuses.Add, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + statuses.Should().EndWith(ConnectionStatus.Errored); + } + + [Fact] + public void ReportsAStatusWhenTheSourceIsAlreadyFinished() + { + var statuses = new List(); + var completed = false; + + using var subscription = Observable.Empty().MonitorStatus().Subscribe(statuses.Add, () => completed = true); + + completed.Should().BeTrue("a terminal event arriving during subscription must not be lost"); + statuses.Should().Equal(ConnectionStatus.Pending, ConnectionStatus.Completed); + } } diff --git a/src/DynamicData.Tests/Cache/OperatorCompletionFixture.cs b/src/DynamicData.Tests/Cache/OperatorCompletionFixture.cs new file mode 100644 index 000000000..0cee74e73 --- /dev/null +++ b/src/DynamicData.Tests/Cache/OperatorCompletionFixture.cs @@ -0,0 +1,268 @@ +using System; +using System.Collections.Generic; +using System.Reactive; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using System.Reactive.Subjects; + +using DynamicData.Binding; +using DynamicData.Kernel; +using DynamicData.Tests.Domain; + +using FluentAssertions; + +using Xunit; + +namespace DynamicData.Tests.Cache; + +/// +/// Terminal event behaviour for operators which previously never delivered OnCompleted. +/// +public class OperatorCompletionFixture +{ + private static readonly IComparer ByName = SortExpressionComparer.Ascending(p => p.Name); + + [Fact] + public void MonitorStatusCompletesWhenSourceCompletes() + { + using var source = new Subject>(); + var statuses = new List(); + var completed = false; + + using var subscription = source.MonitorStatus().Subscribe(statuses.Add, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("the status stream is finished once the source is"); + statuses.Should().EndWith(ConnectionStatus.Completed); + } + + [Fact] + public void MonitorStatusReportsLoadedBeforeCompleted() + { + using var source = new Subject>(); + var statuses = new List(); + + using var subscription = source.MonitorStatus().Subscribe(statuses.Add); + + source.OnNext(new ChangeSet()); + source.OnCompleted(); + + statuses.Should().Equal(ConnectionStatus.Pending, ConnectionStatus.Loaded, ConnectionStatus.Completed); + } + + [Fact] + public void MonitorStatusDeliversErrorAfterReportingIt() + { + using var source = new Subject>(); + var statuses = new List(); + Exception? error = null; + + using var subscription = source.MonitorStatus().Subscribe(statuses.Add, ex => error = ex); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + statuses.Should().EndWith(ConnectionStatus.Errored); + } + + [Fact] + public void MonitorStatusCompletesWhenTheSourceIsAlreadyFinished() + { + var completed = false; + + using var subscription = Observable.Empty>() + .MonitorStatus() + .Subscribe(_ => { }, () => completed = true); + + completed.Should().BeTrue("a terminal event arriving during subscription must not be lost"); + } + + [Fact] + public void DeferUntilLoadedCompletesWhenSourceCompletes() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.DeferUntilLoaded().Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void SkipInitialCompletesWhenSourceCompletes() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.SkipInitial().Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void GroupWithImmutableStateCompletesWhenNoRegrouperIsSupplied() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.GroupWithImmutableState(p => p.Age).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("an absent regrouper can never fire and so must not hold the result open"); + } + + [Fact] + public void GroupOnCompletesWhenNoRegrouperIsSupplied() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.Group(p => p.Age).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void SortCompletesWhenGivenAComparerObservable() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.Sort(Observable.Return(ByName)).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("an absent resort signal can never fire and so must not hold the result open"); + } + + [Fact] + public void SortCompletesWhenGivenAResorter() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.Sort(ByName, Observable.Never().Take(0)).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void InnerJoinManyCompletesWhenBothSidesComplete() + { + using var left = new Subject>(); + var completed = false; + + using var subscription = left + .InnerJoinMany(Observable.Empty>(), p => p.Name, (_, person, _) => person) + .Subscribe(_ => { }, () => completed = true); + + left.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void BatchIfCompletesWhenSourceCompletes() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.BatchIf(Observable.Return(false), Scheduler.Immediate).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void BatchIfFlushesHeldChangesBeforeCompleting() + { + using var source = new Subject>(); + using var pause = new BehaviorSubject(true); + var received = 0; + var completed = false; + + using var subscription = source.BatchIf(pause, Scheduler.Immediate).Subscribe(_ => received++, () => completed = true); + + source.OnNext(new ChangeSet { new(ChangeReason.Add, "a", new Person("a", 1)) }); + source.OnCompleted(); + + received.Should().Be(1, "changes held back by the pause would otherwise be lost"); + completed.Should().BeTrue(); + } + + [Fact] + public void MergeManyItemsCompletesWhenSourceCompletes() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.MergeManyItems(_ => Observable.Empty()).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue(); + } + + [Fact] + public void MergeManyItemsStaysOpenWhenAChildCompletes() + { + using var source = new SourceCache(p => p.Name); + var completed = false; + + using var subscription = source.Connect().MergeManyItems(_ => Observable.Return(1)).Subscribe(_ => { }, () => completed = true); + + source.AddOrUpdate(new Person("a", 1)); + + completed.Should().BeFalse("one child finishing does not finish the merge"); + } + + [Fact] + public void GroupDeliversErrorWithoutThrowing() + { + using var source = new Subject>(); + Exception? error = null; + + using var subscription = source.Group(p => p.Age).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } + + [Fact] + public void TransformToTreeDeliversError() + { + Exception? error = null; + + using var subscription = Observable.Throw>(new InvalidOperationException("boom")) + .TransformToTree(p => p.Name) + .Subscribe(_ => { }, ex => error = ex, () => { }); + + error.Should().BeOfType("the intermediate caches must not swallow it"); + } + + [Fact] + public void TransformToTreeCompletesWhenSourceCompletes() + { + using var source = new Subject>(); + var completed = false; + + using var subscription = source.TransformToTree(p => p.Name).Subscribe(_ => { }, () => completed = true); + + source.OnNext(new ChangeSet { new(ChangeReason.Add, "a", new Person("a", 1)) }); + source.OnCompleted(); + + completed.Should().BeTrue(); + } +} diff --git a/src/DynamicData.Tests/Cache/OrFixture.cs b/src/DynamicData.Tests/Cache/OrFixture.cs index 720d3f0b4..da9d891f9 100644 --- a/src/DynamicData.Tests/Cache/OrFixture.cs +++ b/src/DynamicData.Tests/Cache/OrFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; @@ -7,6 +7,8 @@ using FluentAssertions; using Xunit; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -95,4 +97,34 @@ public void UpdatingOneSourceOnlyProducesResult() } protected abstract IObservable> CreateObservable(); + + [Fact] + public void CompletesOnlyWhenEverySourceCompletes() + { + var completed = false; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Or(second).Subscribe(_ => { }, () => completed = true); + + first.OnCompleted(); + completed.Should().BeFalse("the second source is still live"); + + second.OnCompleted(); + completed.Should().BeTrue("every source has now finished"); + } + + [Fact] + public void DeliversAnErrorFromAnySource() + { + Exception? error = null; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Or(second).Subscribe(_ => { }, ex => error = ex, () => { }); + + second.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/SortFixture.cs b/src/DynamicData.Tests/Cache/SortFixture.cs index 0f228b173..238beeed7 100644 --- a/src/DynamicData.Tests/Cache/SortFixture.cs +++ b/src/DynamicData.Tests/Cache/SortFixture.cs @@ -1,4 +1,4 @@ -#region +#region using System; using System.Collections.Generic; @@ -13,6 +13,8 @@ using FluentAssertions; using Xunit; +using System.Reactive.Concurrency; +using System.Reactive.Linq; #endregion @@ -1038,4 +1040,30 @@ public class Comparer : IComparer public int Compare(ViewModel? x, ViewModel? y) => StringComparer.OrdinalIgnoreCase.Compare(x?.Name, y?.Name); } } + + [Fact] + public void CompletesWhenGivenAComparerObservable() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.Sort(Observable.Return(_comparer)).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("an absent resort signal can never fire and so must not hold the result open"); + } + + [Fact] + public void CompletesWhenGivenAResorter() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.Sort(_comparer, Observable.Never().Take(0)).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("an absent comparer stream can never fire and so must not hold the result open"); + } } diff --git a/src/DynamicData.Tests/Cache/TransformTreeFixture.cs b/src/DynamicData.Tests/Cache/TransformTreeFixture.cs index 33bf44573..b50656b21 100644 --- a/src/DynamicData.Tests/Cache/TransformTreeFixture.cs +++ b/src/DynamicData.Tests/Cache/TransformTreeFixture.cs @@ -6,6 +6,9 @@ using FluentAssertions; using Xunit; +using System.Reactive.Concurrency; +using System.Reactive.Linq; +using DynamicData.Tests.Domain; namespace DynamicData.Tests.Cache; @@ -305,4 +308,30 @@ public override bool Equals(object? obj) public override string ToString() => $"Name: {Name}, Id: {Id}, BossId: {BossId}"; } + + [Fact] + public void CompletesWhenTheSourceCompletes() + { + var completed = false; + + using var source = new Subject>(); + using var subscription = source.TransformToTree(p => p.Name).Subscribe(_ => { }, () => completed = true); + + source.OnCompleted(); + + completed.Should().BeTrue("the intermediate caches must not swallow the terminal event"); + } + + [Fact] + public void DeliversTheError() + { + Exception? error = null; + + using var source = new Subject>(); + using var subscription = source.TransformToTree(p => p.Name).Subscribe(_ => { }, ex => error = ex, () => { }); + + source.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData.Tests/Cache/XorFixture.cs b/src/DynamicData.Tests/Cache/XorFixture.cs index 8a320430a..f76229aa5 100644 --- a/src/DynamicData.Tests/Cache/XorFixture.cs +++ b/src/DynamicData.Tests/Cache/XorFixture.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using DynamicData.Tests.Domain; @@ -6,6 +6,8 @@ using FluentAssertions; using Xunit; +using System.Reactive.Linq; +using System.Reactive.Subjects; namespace DynamicData.Tests.Cache; @@ -91,4 +93,34 @@ public void UpdatingOneSourceOnlyProducesResult() } protected abstract IObservable> CreateObservable(); + + [Fact] + public void CompletesOnlyWhenEverySourceCompletes() + { + var completed = false; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Xor(second).Subscribe(_ => { }, () => completed = true); + + first.OnCompleted(); + completed.Should().BeFalse("the second source is still live"); + + second.OnCompleted(); + completed.Should().BeTrue("every source has now finished"); + } + + [Fact] + public void DeliversAnErrorFromAnySource() + { + Exception? error = null; + + using var first = new Subject>(); + using var second = new Subject>(); + using var subscription = first.Xor(second).Subscribe(_ => { }, ex => error = ex, () => { }); + + second.OnError(new InvalidOperationException("boom")); + + error.Should().BeOfType(); + } } diff --git a/src/DynamicData/Binding/BindPaged.cs b/src/DynamicData/Binding/BindPaged.cs index 1cbe39d35..21e8a067f 100644 --- a/src/DynamicData/Binding/BindPaged.cs +++ b/src/DynamicData/Binding/BindPaged.cs @@ -46,15 +46,45 @@ private IObservable> UseContextSortOptions() => // making the comparedChanged observable to fire. Probably a deadlock var changesSubject = new Subject>(); var comparerSubject = new ReplaySubject>(1); + var bound = false; + + // Until the first element has supplied the binding options there is nothing standing between the + // source and the observer, so terminal events have to reach the observer directly. + void Fail(Exception error) + { + if (bound) + { + changesSubject.OnError(error); + } + else + { + observer.OnError(error); + } + } + + void Finish() + { + if (bound) + { + changesSubject.OnCompleted(); + } + else + { + observer.OnCompleted(); + } + } // once we have the initial values, publish as normal. var subsequent = shared .Skip(1) - .Subscribe(changesWithContext => - { - comparerSubject.OnNext(changesWithContext.Context.Comparer); - changesSubject.OnNext(changesWithContext); - }); + .Subscribe( + changesWithContext => + { + comparerSubject.OnNext(changesWithContext.Context.Comparer); + changesSubject.OnNext(changesWithContext); + }, + Fail, + Finish); // extract binding options from the page context var initial = shared @@ -72,9 +102,13 @@ private IObservable> UseContextSortOptions() => .SortAndBind(targetList, comparerSubject.DistinctUntilChanged(), extractedOptions) .SubscribeSafe(observer); + bound = true; + comparerSubject.OnNext(changesWithContext.Context.Comparer); changesSubject.OnNext(changesWithContext); - }); + }, + // 'subsequent' above relays the error. Without a handler here Rx would rethrow it as well. + static _ => { }); return new CompositeDisposable(initial, subscriber, subsequent, shared.Connect()); }); diff --git a/src/DynamicData/Binding/BindVirtualized.cs b/src/DynamicData/Binding/BindVirtualized.cs index 37a3e1739..b06e8a85c 100644 --- a/src/DynamicData/Binding/BindVirtualized.cs +++ b/src/DynamicData/Binding/BindVirtualized.cs @@ -44,15 +44,45 @@ private IObservable> UseVirtualSortOptions() => // making the comparedChanged observable to fire. Probably a deadlock var changesSubject = new Subject>(); var comparerSubject = new ReplaySubject>(1); + var bound = false; + + // Until the first element has supplied the binding options there is nothing standing between the + // source and the observer, so terminal events have to reach the observer directly. + void Fail(Exception error) + { + if (bound) + { + changesSubject.OnError(error); + } + else + { + observer.OnError(error); + } + } + + void Finish() + { + if (bound) + { + changesSubject.OnCompleted(); + } + else + { + observer.OnCompleted(); + } + } // once we have the initial values, publish as normal. var subsequent = shared .Skip(1) - .Subscribe(changesWithContext => - { - comparerSubject.OnNext(changesWithContext.Context.Comparer); - changesSubject.OnNext(changesWithContext); - }); + .Subscribe( + changesWithContext => + { + comparerSubject.OnNext(changesWithContext.Context.Comparer); + changesSubject.OnNext(changesWithContext); + }, + Fail, + Finish); // extract binding options from the virtual context var initial = shared @@ -70,9 +100,13 @@ private IObservable> UseVirtualSortOptions() => .SortAndBind(targetList, comparerSubject.DistinctUntilChanged(), extractedOptions) .SubscribeSafe(observer); + bound = true; + comparerSubject.OnNext(changesWithContext.Context.Comparer); changesSubject.OnNext(changesWithContext); - }); + }, + // 'subsequent' above relays the error. Without a handler here Rx would rethrow it as well. + static _ => { }); return new CompositeDisposable(initial, subscriber, subsequent, shared.Connect()); }); diff --git a/src/DynamicData/Binding/SortAndBind.cs b/src/DynamicData/Binding/SortAndBind.cs index 0bb5fe208..c5b230dea 100644 --- a/src/DynamicData/Binding/SortAndBind.cs +++ b/src/DynamicData/Binding/SortAndBind.cs @@ -70,11 +70,13 @@ public SortAndBind(IObservable> source, // Create a new sort applicator each time. var latestComparer = comparerChanged.SynchronizeSafe(queue) - .Subscribe(comparer => - { - sortApplicator = new SortApplicator(_cache, target, comparer, options); - sortApplicator.ApplySort(); - }); + .Subscribe( + comparer => + { + sortApplicator = new SortApplicator(_cache, target, comparer, options); + sortApplicator.ApplySort(); + }, + observer.OnError); // Listen to changes and apply the sorting var subscriber = source.SynchronizeSafe(queue) diff --git a/src/DynamicData/Cache/Internal/BatchIf.cs b/src/DynamicData/Cache/Internal/BatchIf.cs index 76af20a61..5242da7ba 100644 --- a/src/DynamicData/Cache/Internal/BatchIf.cs +++ b/src/DynamicData/Cache/Internal/BatchIf.cs @@ -97,6 +97,13 @@ IDisposable IntervalFunction() => { ResumeAction(); } + }, + observer.OnError, + () => + { + // Anything still held back would otherwise be lost, so flush before finishing. + ResumeAction(); + observer.OnCompleted(); }); return new CompositeDisposable(publisher, pausedHandler, timeoutDisposer, intervalTimerDisposer, queue); diff --git a/src/DynamicData/Cache/Internal/Combiner.cs b/src/DynamicData/Cache/Internal/Combiner.cs index c7734512d..97ee9458b 100644 --- a/src/DynamicData/Cache/Internal/Combiner.cs +++ b/src/DynamicData/Cache/Internal/Combiner.cs @@ -23,8 +23,16 @@ internal sealed class Combiner(CombineOperator type, Action> _sourceCaches = []; - public IDisposable Subscribe(IObservable>[] source) + public IDisposable Subscribe(IObservable>[] source, Action onError, Action onCompleted) { + // Merging semantics: the result finishes only once every source has. + var pending = source.Length; + if (pending == 0) + { + onCompleted(); + return Disposable.Empty; + } + // subscribe var disposable = new CompositeDisposable(); lock (_locker) @@ -34,7 +42,17 @@ public IDisposable Subscribe(IObservable>[] source) foreach (var pair in source.Zip(_sourceCaches, (item, cache) => new { Item = item, Cache = cache })) { - var subscription = pair.Item.Subscribe(updates => Update(pair.Cache, updates)); + var subscription = pair.Item.Subscribe( + updates => Update(pair.Cache, updates), + onError, + () => + { + if (Interlocked.Decrement(ref pending) == 0) + { + onCompleted(); + } + }); + disposable.Add(subscription); } } diff --git a/src/DynamicData/Cache/Internal/DynamicCombiner.cs b/src/DynamicData/Cache/Internal/DynamicCombiner.cs index 41c7e3a8d..8db64340f 100644 --- a/src/DynamicData/Cache/Internal/DynamicCombiner.cs +++ b/src/DynamicData/Cache/Internal/DynamicCombiner.cs @@ -39,7 +39,9 @@ public IObservable> Run() => Observable.Create> Run() => Observable.Create { }, static _ => { }); // when an list is added or removed, need to var sourceChanged = sharedLists.WhereReasonsAre(ListChangeReason.Add, ListChangeReason.AddRange).ForEachItemChange( @@ -77,7 +82,10 @@ public IObservable> Run() => Observable.Create { }, static _ => { }); return new CompositeDisposable(sourceLists, allChanges, removedItem, sourceChanged, sharedLists.Connect(), queue); }); diff --git a/src/DynamicData/Cache/Internal/GroupOn.cs b/src/DynamicData/Cache/Internal/GroupOn.cs index fd0936fe7..550b79e86 100644 --- a/src/DynamicData/Cache/Internal/GroupOn.cs +++ b/src/DynamicData/Cache/Internal/GroupOn.cs @@ -15,7 +15,9 @@ internal sealed class GroupOn(IObservable _groupSelectorKey = groupSelectorKey ?? throw new ArgumentNullException(nameof(groupSelectorKey)); - private readonly IObservable _regrouper = regrouper ?? Observable.Never(); + // An absent regrouper means no regroup signal will ever arrive. Never would say one still might, + // which leaves the merge below unable to complete when the source does. + private readonly IObservable _regrouper = regrouper ?? Observable.Empty(); private readonly IObservable> _source = source ?? throw new ArgumentNullException(nameof(source)); @@ -25,13 +27,16 @@ public IObservable> Run() => Observabl var queue = new SharedDeliveryQueue(); var grouper = new Grouper(_groupSelectorKey); - var groups = _source.SynchronizeSafe(queue).Finally(observer.OnCompleted).Select(grouper.Update).Where(changes => changes.Count != 0); + var groups = _source.SynchronizeSafe(queue).Select(grouper.Update).Where(changes => changes.Count != 0); var regroup = _regrouper.SynchronizeSafe(queue).Select(_ => grouper.Regroup()).Where(changes => changes.Count != 0); var published = groups.Merge(regroup).Publish(); var subscriber = published.SubscribeSafe(observer); - var disposer = published.DisposeMany().Subscribe(); + + // The observer above already receives any error. Without a handler here Rx would rethrow it + // out of the subscription instead. + var disposer = published.DisposeMany().Subscribe(static _ => { }, static _ => { }); var connected = published.Connect(); diff --git a/src/DynamicData/Cache/Internal/GroupOnImmutable.cs b/src/DynamicData/Cache/Internal/GroupOnImmutable.cs index ba33db031..3ee2ee0d2 100644 --- a/src/DynamicData/Cache/Internal/GroupOnImmutable.cs +++ b/src/DynamicData/Cache/Internal/GroupOnImmutable.cs @@ -15,7 +15,9 @@ internal sealed class GroupOnImmutable(IObservable _groupSelectorKey = groupSelectorKey ?? throw new ArgumentNullException(nameof(groupSelectorKey)); - private readonly IObservable _regrouper = regrouper ?? Observable.Never(); + // An absent regrouper means no regroup signal will ever arrive. Never would say one still might, + // which leaves the merge below unable to complete when the source does. + private readonly IObservable _regrouper = regrouper ?? Observable.Empty(); private readonly IObservable> _source = source ?? throw new ArgumentNullException(nameof(source)); diff --git a/src/DynamicData/Cache/Internal/MergeMany.cs b/src/DynamicData/Cache/Internal/MergeMany.cs index fe2eecc75..70d38cf27 100644 --- a/src/DynamicData/Cache/Internal/MergeMany.cs +++ b/src/DynamicData/Cache/Internal/MergeMany.cs @@ -46,7 +46,7 @@ public IObservable Run() => Observable.Create( Interlocked.Increment(ref counter.Value); return _observableSelector(t, key) .Finally(() => CheckCompleted(counter, queue)) - .Subscribe(queue.OnNext, static _ => { }); + .Subscribe(queue.OnNext, queue.OnError); }) .Subscribe(static _ => { }, observer.OnError)); }); diff --git a/src/DynamicData/Cache/Internal/MergeManyItems.cs b/src/DynamicData/Cache/Internal/MergeManyItems.cs index 26b85614b..ccf6050d3 100644 --- a/src/DynamicData/Cache/Internal/MergeManyItems.cs +++ b/src/DynamicData/Cache/Internal/MergeManyItems.cs @@ -31,5 +31,10 @@ public MergeManyItems(IObservable> source, Func observableSelector(t); } - public IObservable> Run() => Observable.Create>(observer => _source.SubscribeMany((t, v) => _observableSelector(t, v).Select(z => new ItemWithValue(t, z)).SubscribeSafe(observer)).Subscribe()); + // MergeMany already tracks the parent and every child subscription so that the result finishes only once + // all of them have. Reusing it keeps a child completing from terminating the whole stream. + public IObservable> Run() => + new MergeMany>( + _source, + (t, v) => _observableSelector(t, v).Select(z => new ItemWithValue(t, z))).Run(); } diff --git a/src/DynamicData/Cache/Internal/Sort.cs b/src/DynamicData/Cache/Internal/Sort.cs index 11b39035e..bbab31187 100644 --- a/src/DynamicData/Cache/Internal/Sort.cs +++ b/src/DynamicData/Cache/Internal/Sort.cs @@ -51,9 +51,10 @@ public IObservable> Run() => Observable.Create result is not null).Select(x => x!).SubscribeSafe(observer); } - var comparerChanged = (_comparerChangedObservable ?? Observable.Never>()).SynchronizeSafe(queue).Select(sorter.Sort); + // An absent comparer or resort signal will never fire, so it must not hold the merge open. + var comparerChanged = (_comparerChangedObservable ?? Observable.Empty>()).SynchronizeSafe(queue).Select(sorter.Sort); - var sortAgain = (_resorter ?? Observable.Never()).SynchronizeSafe(queue).Select(_ => sorter.Sort()); + var sortAgain = (_resorter ?? Observable.Empty()).SynchronizeSafe(queue).Select(_ => sorter.Sort()); var dataChanged = _source.SynchronizeSafe(queue).Select(sorter.Sort); diff --git a/src/DynamicData/Cache/Internal/SpecifiedGrouper.cs b/src/DynamicData/Cache/Internal/SpecifiedGrouper.cs index 863f2d6d6..493bb1c9a 100644 --- a/src/DynamicData/Cache/Internal/SpecifiedGrouper.cs +++ b/src/DynamicData/Cache/Internal/SpecifiedGrouper.cs @@ -23,8 +23,9 @@ public IObservable> Run() => Observabl { var queue = new SharedDeliveryQueue(); - // create source group cache - var sourceGroups = _source.SynchronizeSafe(queue).Group(_groupSelector).DisposeMany().AsObservableCache(); + // create source group cache. The observer is fed from the result group source below, so the + // source's own terminal events would otherwise never reach it. + var sourceGroups = _source.SynchronizeSafe(queue).Do(static _ => { }, observer.OnError, observer.OnCompleted).Group(_groupSelector).DisposeMany().AsObservableCache(); // create parent groups var parentGroups = _resultGroupSource.SynchronizeSafe(queue).Transform( @@ -53,7 +54,7 @@ public IObservable> Run() => Observabl { groupToUpdate.Value.Update(updater => updater.Clone(updates)); } - })).DisposeMany().Subscribe(); + })).DisposeMany().Subscribe(static _ => { }, static _ => { }); var notifier = parentGroups.Connect().Select( x => diff --git a/src/DynamicData/Cache/Internal/StatusMonitor.cs b/src/DynamicData/Cache/Internal/StatusMonitor.cs index 445074b63..84c9dfc0c 100644 --- a/src/DynamicData/Cache/Internal/StatusMonitor.cs +++ b/src/DynamicData/Cache/Internal/StatusMonitor.cs @@ -1,60 +1,17 @@ -// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved. +// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved. // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. -using System.Reactive.Disposables; using System.Reactive.Linq; -using System.Reactive.Subjects; namespace DynamicData.Cache.Internal; internal sealed class StatusMonitor(IObservable source) { - public IObservable Run() => Observable.Create( - observer => - { - var statusSubject = new Subject(); - var status = ConnectionStatus.Pending; - - void Error(Exception ex) - { - status = ConnectionStatus.Errored; - statusSubject.OnNext(status); - observer.OnError(ex); - } - - void Completion() - { - if (status == ConnectionStatus.Errored) - { - return; - } - - status = ConnectionStatus.Completed; - statusSubject.OnNext(status); - } - - void Updated() - { - if (status != ConnectionStatus.Pending) - { - return; - } - - status = ConnectionStatus.Loaded; - statusSubject.OnNext(status); - } - - var monitor = source.Subscribe(_ => Updated(), Error, Completion); - - var subscriber = statusSubject.StartWith(status).DistinctUntilChanged().SubscribeSafe(observer); - - return Disposable.Create( - () => - { - statusSubject.OnCompleted(); - monitor.Dispose(); - subscriber.Dispose(); - }); - }); + public IObservable Run() => + source.Select(static _ => ConnectionStatus.Loaded) + .Concat(Observable.Return(ConnectionStatus.Completed)) + .Catch(static error => Observable.Return(ConnectionStatus.Errored).Concat(Observable.Throw(error))) + .StartWith(ConnectionStatus.Pending) + .DistinctUntilChanged(); } diff --git a/src/DynamicData/Cache/Internal/TreeBuilder.cs b/src/DynamicData/Cache/Internal/TreeBuilder.cs index bf379ef99..c86134104 100644 --- a/src/DynamicData/Cache/Internal/TreeBuilder.cs +++ b/src/DynamicData/Cache/Internal/TreeBuilder.cs @@ -27,7 +27,8 @@ public IObservable, TKey>> Run() => Observable.Cr var queue = new SharedDeliveryQueue(); var reFilterObservable = new BehaviorSubject(Unit.Default); - var allData = _source.SynchronizeSafe(queue).AsObservableCache(); + // Terminal events do not survive the intermediate caches below, so relay them to the observer. + var allData = _source.SynchronizeSafe(queue).Do(static _ => { }, observer.OnError, observer.OnCompleted).AsObservableCache(); // for each object we need a node which provides // a structure to set the parent and children @@ -195,7 +196,7 @@ void UpdateChildren(Node parentNode) } reFilterObservable.OnNext(Unit.Default); - }).DisposeMany().Subscribe(); + }).DisposeMany().Subscribe(static _ => { }, static _ => { }); var filter = _predicateChanged.SynchronizeSafe(queue).CombineLatest(reFilterObservable, (predicate, _) => predicate); var result = allNodes.Connect().Filter(filter).SubscribeSafe(observer); diff --git a/src/DynamicData/Cache/ObservableCacheEx.Combine.cs b/src/DynamicData/Cache/ObservableCacheEx.Combine.cs index eb46b86b3..ca9206a28 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.Combine.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.Combine.cs @@ -1,4 +1,4 @@ -// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved. +// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved. // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. @@ -89,7 +89,7 @@ void UpdateAction(IChangeSet updates) try { var combiner = new Combiner(type, UpdateAction); - subscriber = combiner.Subscribe([.. sources]); + subscriber = combiner.Subscribe([.. sources], observer.OnError, observer.OnCompleted); } catch (Exception ex) { @@ -130,7 +130,7 @@ void UpdateAction(IChangeSet updates) list.Insert(0, source); var combiner = new Combiner(type, UpdateAction); - subscriber = combiner.Subscribe([.. list]); + subscriber = combiner.Subscribe([.. list], observer.OnError, observer.OnCompleted); } catch (Exception ex) { From dfa2a3ad4fd903d2a68c84fe1a5920a552fc6127 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 09:08:55 -0700 Subject: [PATCH 2/9] Test the contract per operator instead of with a sweep The conformance sweep was a lot of machinery to carry for what it asserts. Each behaviour it covered now sits in the fixture for the operator it belongs to, where someone changing that operator will actually see it. MonitorStatusFixture, DeferUntilLoadedFixture, SortFixture, GroupFixture, GroupImmutableFixture, BatchIfFixture, MergeManyItemsFixture, MergeManyFixture, TransformTreeFixture, InnerJoinFixture, and the four combiner fixtures. Group with a result group source completes as soon as that source completes, since no group can ever appear after that, so its error test uses a source that stays open. Worth knowing before reading the test. --- .../Cache/CombinerCompletionFixture.cs | 73 ----- .../Cache/MergeManyInnerErrorFixture.cs | 47 --- .../Cache/MonitorStatusFixture.cs | 3 +- .../Cache/OperatorCompletionFixture.cs | 268 ------------------ 4 files changed, 2 insertions(+), 389 deletions(-) delete mode 100644 src/DynamicData.Tests/Cache/CombinerCompletionFixture.cs delete mode 100644 src/DynamicData.Tests/Cache/MergeManyInnerErrorFixture.cs delete mode 100644 src/DynamicData.Tests/Cache/OperatorCompletionFixture.cs diff --git a/src/DynamicData.Tests/Cache/CombinerCompletionFixture.cs b/src/DynamicData.Tests/Cache/CombinerCompletionFixture.cs deleted file mode 100644 index df5ea398f..000000000 --- a/src/DynamicData.Tests/Cache/CombinerCompletionFixture.cs +++ /dev/null @@ -1,73 +0,0 @@ -using System; -using System.Reactive.Linq; -using System.Reactive.Subjects; - -using DynamicData.Tests.Domain; - -using FluentAssertions; - -using Xunit; - -namespace DynamicData.Tests.Cache; - -/// -/// Terminal event behaviour for the combining operators. -/// -public class CombinerCompletionFixture -{ - [Fact] - public void CombinersCompleteWhenEverySourceCompletes() - { - foreach (var combine in new Func>, IObservable>, IObservable>>[] - { - static (a, b) => ObservableCacheEx.And(a, b), - static (a, b) => a.Or(b), - static (a, b) => a.Except(b), - static (a, b) => a.Xor(b), - }) - { - using var first = new Subject>(); - using var second = new Subject>(); - var completed = false; - - using var subscription = combine(first, second).Subscribe(_ => { }, () => completed = true); - - first.OnCompleted(); - completed.Should().BeFalse("the second source is still live"); - - second.OnCompleted(); - completed.Should().BeTrue("every source has now finished"); - } - } - - [Fact] - public void CombinersDeliverErrorFromAnySource() - { - using var first = new Subject>(); - using var second = new Subject>(); - Exception? error = null; - - using var subscription = first.Or(second).Subscribe(_ => { }, ex => error = ex, () => { }); - - second.OnError(new InvalidOperationException("boom")); - - error.Should().BeOfType(); - } - - [Fact] - public void DynamicCombinerCompletesWhenEverySourceCompletes() - { - using var sources = new SourceList>>(); - using var first = new Subject>(); - var completed = false; - - sources.Add(first); - - using var subscription = sources.Or().Subscribe(_ => { }, () => completed = true); - - first.OnCompleted(); - sources.Dispose(); - - completed.Should().BeTrue(); - } -} diff --git a/src/DynamicData.Tests/Cache/MergeManyInnerErrorFixture.cs b/src/DynamicData.Tests/Cache/MergeManyInnerErrorFixture.cs deleted file mode 100644 index 8cc4c2301..000000000 --- a/src/DynamicData.Tests/Cache/MergeManyInnerErrorFixture.cs +++ /dev/null @@ -1,47 +0,0 @@ -using System; -using System.Reactive.Subjects; - -using DynamicData.Tests.Domain; - -using FluentAssertions; - -using Xunit; - -namespace DynamicData.Tests.Cache; - -/// -/// Merge propagates a failure from any inner stream, rather than discarding it. -/// -public class MergeManyInnerErrorFixture -{ - [Fact] - public void MergeManyDeliversErrorFromAChild() - { - using var source = new SourceCache(p => p.Name); - using var child = new Subject(); - Exception? error = null; - - using var subscription = source.Connect().MergeMany(_ => child).Subscribe(_ => { }, ex => error = ex, () => { }); - - source.AddOrUpdate(new Person("a", 1)); - child.OnError(new InvalidOperationException("boom")); - - error.Should().BeOfType("a failing inner stream must not be silently discarded"); - } - - [Fact] - public void MergeManyItemsDeliversErrorFromAChild() - { - using var source = new SourceCache(p => p.Name); - using var child = new Subject(); - Exception? error = null; - - using var subscription = source.Connect().MergeManyItems(_ => child).Subscribe(_ => { }, ex => error = ex, () => { }); - - source.AddOrUpdate(new Person("a", 1)); - child.OnError(new InvalidOperationException("boom")); - - error.Should().BeOfType(); - } - -} diff --git a/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs b/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs index e127256e1..ca31d171f 100644 --- a/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs +++ b/src/DynamicData.Tests/Cache/MonitorStatusFixture.cs @@ -104,10 +104,11 @@ public void CompletesWhenTheSourceCompletes() using var source = new Subject(); using var subscription = source.MonitorStatus().Subscribe(statuses.Add, () => completed = true); + source.OnNext(1); source.OnCompleted(); completed.Should().BeTrue("the status stream is finished once the source is"); - statuses.Should().EndWith(ConnectionStatus.Completed); + statuses.Should().Equal(ConnectionStatus.Pending, ConnectionStatus.Loaded, ConnectionStatus.Completed); } [Fact] diff --git a/src/DynamicData.Tests/Cache/OperatorCompletionFixture.cs b/src/DynamicData.Tests/Cache/OperatorCompletionFixture.cs deleted file mode 100644 index 0cee74e73..000000000 --- a/src/DynamicData.Tests/Cache/OperatorCompletionFixture.cs +++ /dev/null @@ -1,268 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Reactive; -using System.Reactive.Concurrency; -using System.Reactive.Linq; -using System.Reactive.Subjects; - -using DynamicData.Binding; -using DynamicData.Kernel; -using DynamicData.Tests.Domain; - -using FluentAssertions; - -using Xunit; - -namespace DynamicData.Tests.Cache; - -/// -/// Terminal event behaviour for operators which previously never delivered OnCompleted. -/// -public class OperatorCompletionFixture -{ - private static readonly IComparer ByName = SortExpressionComparer.Ascending(p => p.Name); - - [Fact] - public void MonitorStatusCompletesWhenSourceCompletes() - { - using var source = new Subject>(); - var statuses = new List(); - var completed = false; - - using var subscription = source.MonitorStatus().Subscribe(statuses.Add, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue("the status stream is finished once the source is"); - statuses.Should().EndWith(ConnectionStatus.Completed); - } - - [Fact] - public void MonitorStatusReportsLoadedBeforeCompleted() - { - using var source = new Subject>(); - var statuses = new List(); - - using var subscription = source.MonitorStatus().Subscribe(statuses.Add); - - source.OnNext(new ChangeSet()); - source.OnCompleted(); - - statuses.Should().Equal(ConnectionStatus.Pending, ConnectionStatus.Loaded, ConnectionStatus.Completed); - } - - [Fact] - public void MonitorStatusDeliversErrorAfterReportingIt() - { - using var source = new Subject>(); - var statuses = new List(); - Exception? error = null; - - using var subscription = source.MonitorStatus().Subscribe(statuses.Add, ex => error = ex); - - source.OnError(new InvalidOperationException("boom")); - - error.Should().BeOfType(); - statuses.Should().EndWith(ConnectionStatus.Errored); - } - - [Fact] - public void MonitorStatusCompletesWhenTheSourceIsAlreadyFinished() - { - var completed = false; - - using var subscription = Observable.Empty>() - .MonitorStatus() - .Subscribe(_ => { }, () => completed = true); - - completed.Should().BeTrue("a terminal event arriving during subscription must not be lost"); - } - - [Fact] - public void DeferUntilLoadedCompletesWhenSourceCompletes() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.DeferUntilLoaded().Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void SkipInitialCompletesWhenSourceCompletes() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.SkipInitial().Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void GroupWithImmutableStateCompletesWhenNoRegrouperIsSupplied() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.GroupWithImmutableState(p => p.Age).Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue("an absent regrouper can never fire and so must not hold the result open"); - } - - [Fact] - public void GroupOnCompletesWhenNoRegrouperIsSupplied() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.Group(p => p.Age).Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void SortCompletesWhenGivenAComparerObservable() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.Sort(Observable.Return(ByName)).Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue("an absent resort signal can never fire and so must not hold the result open"); - } - - [Fact] - public void SortCompletesWhenGivenAResorter() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.Sort(ByName, Observable.Never().Take(0)).Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void InnerJoinManyCompletesWhenBothSidesComplete() - { - using var left = new Subject>(); - var completed = false; - - using var subscription = left - .InnerJoinMany(Observable.Empty>(), p => p.Name, (_, person, _) => person) - .Subscribe(_ => { }, () => completed = true); - - left.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void BatchIfCompletesWhenSourceCompletes() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.BatchIf(Observable.Return(false), Scheduler.Immediate).Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void BatchIfFlushesHeldChangesBeforeCompleting() - { - using var source = new Subject>(); - using var pause = new BehaviorSubject(true); - var received = 0; - var completed = false; - - using var subscription = source.BatchIf(pause, Scheduler.Immediate).Subscribe(_ => received++, () => completed = true); - - source.OnNext(new ChangeSet { new(ChangeReason.Add, "a", new Person("a", 1)) }); - source.OnCompleted(); - - received.Should().Be(1, "changes held back by the pause would otherwise be lost"); - completed.Should().BeTrue(); - } - - [Fact] - public void MergeManyItemsCompletesWhenSourceCompletes() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.MergeManyItems(_ => Observable.Empty()).Subscribe(_ => { }, () => completed = true); - - source.OnCompleted(); - - completed.Should().BeTrue(); - } - - [Fact] - public void MergeManyItemsStaysOpenWhenAChildCompletes() - { - using var source = new SourceCache(p => p.Name); - var completed = false; - - using var subscription = source.Connect().MergeManyItems(_ => Observable.Return(1)).Subscribe(_ => { }, () => completed = true); - - source.AddOrUpdate(new Person("a", 1)); - - completed.Should().BeFalse("one child finishing does not finish the merge"); - } - - [Fact] - public void GroupDeliversErrorWithoutThrowing() - { - using var source = new Subject>(); - Exception? error = null; - - using var subscription = source.Group(p => p.Age).Subscribe(_ => { }, ex => error = ex, () => { }); - - source.OnError(new InvalidOperationException("boom")); - - error.Should().BeOfType(); - } - - [Fact] - public void TransformToTreeDeliversError() - { - Exception? error = null; - - using var subscription = Observable.Throw>(new InvalidOperationException("boom")) - .TransformToTree(p => p.Name) - .Subscribe(_ => { }, ex => error = ex, () => { }); - - error.Should().BeOfType("the intermediate caches must not swallow it"); - } - - [Fact] - public void TransformToTreeCompletesWhenSourceCompletes() - { - using var source = new Subject>(); - var completed = false; - - using var subscription = source.TransformToTree(p => p.Name).Subscribe(_ => { }, () => completed = true); - - source.OnNext(new ChangeSet { new(ChangeReason.Add, "a", new Person("a", 1)) }); - source.OnCompleted(); - - completed.Should().BeTrue(); - } -} From 5b469d3b4b1665222f5041404f4f36a8b6d876e6 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 15:32:34 -0700 Subject: [PATCH 3/9] Route cache terminal events through the observable channel consistently MergeMany sent parent-source failures straight to the observer while child failures went through the delivery queue, so a parent failure could overlap with queued child notifications or land ahead of them. It now uses the same queue as the children. Combine called OnCompleted immediately after OnError in its exception paths, which is a contract violation outright. The first overload's UpdateAction already had it right. BatchIf's interval, pause and timeout subscriptions had no error handler, so a failure of any of those control streams escaped rather than reaching the subscriber. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/DynamicData.Tests/Cache/BatchIfFixture.cs | 17 +++++++++++++++++ src/DynamicData/Cache/Internal/BatchIf.cs | 9 ++++++--- src/DynamicData/Cache/Internal/MergeMany.cs | 2 +- .../Cache/ObservableCacheEx.Combine.cs | 3 --- 4 files changed, 24 insertions(+), 7 deletions(-) diff --git a/src/DynamicData.Tests/Cache/BatchIfFixture.cs b/src/DynamicData.Tests/Cache/BatchIfFixture.cs index f5064c39c..d2d6c6e15 100644 --- a/src/DynamicData.Tests/Cache/BatchIfFixture.cs +++ b/src/DynamicData.Tests/Cache/BatchIfFixture.cs @@ -173,4 +173,21 @@ public void FlushesHeldChangesBeforeCompleting() received.Should().Be(1, "changes held back by the pause would otherwise be lost"); completed.Should().BeTrue(); } + + [Fact] + public void FailsWhenThePauseSelectorFails() + { + var expectedError = new Exception("Test Exception"); + var actualError = default(Exception); + var completed = false; + + using var source = new Subject>(); + using var pause = new Subject(); + using var subscription = source.BatchIf(pause, Scheduler.Immediate).Subscribe(_ => { }, error => actualError = error, () => completed = true); + + pause.OnError(expectedError); + + actualError.Should().BeSameAs(expectedError, "a failure of the pause selector belongs to the subscriber, not to whichever thread happened to raise it"); + completed.Should().BeFalse("the pause selector failed, it did not complete"); + } } diff --git a/src/DynamicData/Cache/Internal/BatchIf.cs b/src/DynamicData/Cache/Internal/BatchIf.cs index 5242da7ba..7988ba6e1 100644 --- a/src/DynamicData/Cache/Internal/BatchIf.cs +++ b/src/DynamicData/Cache/Internal/BatchIf.cs @@ -55,7 +55,8 @@ IDisposable IntervalFunction() => { paused = true; } - }); + }, + observer.OnError); if (intervalTimer is not null) { @@ -83,9 +84,11 @@ IDisposable IntervalFunction() => { paused = false; ResumeAction(); - }); + }, + observer.OnError); } - }); + }, + observer.OnError); var publisher = _source.SynchronizeSafe(queue).Subscribe( changes => diff --git a/src/DynamicData/Cache/Internal/MergeMany.cs b/src/DynamicData/Cache/Internal/MergeMany.cs index 70d38cf27..fa3a00ac3 100644 --- a/src/DynamicData/Cache/Internal/MergeMany.cs +++ b/src/DynamicData/Cache/Internal/MergeMany.cs @@ -48,7 +48,7 @@ public IObservable Run() => Observable.Create( .Finally(() => CheckCompleted(counter, queue)) .Subscribe(queue.OnNext, queue.OnError); }) - .Subscribe(static _ => { }, observer.OnError)); + .Subscribe(static _ => { }, queue.OnError)); }); private static void CheckCompleted(StrongBox counter, DeliveryQueue queue) diff --git a/src/DynamicData/Cache/ObservableCacheEx.Combine.cs b/src/DynamicData/Cache/ObservableCacheEx.Combine.cs index ca9206a28..450d5a5d4 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.Combine.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.Combine.cs @@ -94,7 +94,6 @@ void UpdateAction(IChangeSet updates) catch (Exception ex) { observer.OnError(ex); - observer.OnCompleted(); } return subscriber; @@ -119,7 +118,6 @@ void UpdateAction(IChangeSet updates) catch (Exception ex) { observer.OnError(ex); - observer.OnCompleted(); } } @@ -135,7 +133,6 @@ void UpdateAction(IChangeSet updates) catch (Exception ex) { observer.OnError(ex); - observer.OnCompleted(); } return subscriber; From bc02d0e25a4c1d5803ec4c5369a35b6aff795fac Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 16:15:37 -0700 Subject: [PATCH 4/9] Update cache operator docs for the changed terminal behaviour MergeMany no longer swallows child errors, and it now waits for every child before completing, so the documented behaviour was wrong on both counts. The MergeManyChangeSets entry drew a contrast with that swallowing, which no longer exists. BatchIf gained error forwarding from the pause selector. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/DynamicData/Cache/ObservableCacheEx.BatchIf.cs | 2 +- src/DynamicData/Cache/ObservableCacheEx.MergeMany.cs | 3 ++- src/DynamicData/Cache/ObservableCacheEx.MergeManyChangeSets.cs | 2 +- 3 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/DynamicData/Cache/ObservableCacheEx.BatchIf.cs b/src/DynamicData/Cache/ObservableCacheEx.BatchIf.cs index 99fb5ee13..843167418 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.BatchIf.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.BatchIf.cs @@ -66,7 +66,7 @@ public static IObservable> BatchIf(this /// UpdateBuffered while paused; forwarded immediately while active. /// RemoveBuffered while paused; forwarded immediately while active. /// RefreshBuffered while paused; forwarded immediately while active. - /// OnErrorBuffered data is lost. + /// OnErrorBuffered data is lost. An error from terminates the output the same way a source error does. /// OnCompletedAny remaining buffered data is flushed before completion. /// /// Worth noting: If the source completes while paused, buffered data IS flushed before OnCompleted. However, if the source errors while paused, buffered data is lost. diff --git a/src/DynamicData/Cache/ObservableCacheEx.MergeMany.cs b/src/DynamicData/Cache/ObservableCacheEx.MergeMany.cs index b3c290d34..90e0c88e4 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.MergeMany.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.MergeMany.cs @@ -49,7 +49,8 @@ public static partial class ObservableCacheEx /// UpdateDisposes the previous child subscription and creates a new one for the updated item. /// RemoveDisposes the child subscription for the removed item. /// RefreshNo effect on subscriptions. The child observable continues unchanged. - /// OnErrorErrors from child observables are silently swallowed (the child is unsubscribed). Errors from the source changeset stream terminate the merged output. + /// OnErrorAn error from a child observable, or from the source changeset stream, terminates the merged output. + /// OnCompletedThe output completes once the source changeset stream has completed and every active child observable has completed. A child completing on its own does not end the merged output. /// /// Worth noting: The output is a plain , not a changeset stream. If you need merged changesets, use instead. /// diff --git a/src/DynamicData/Cache/ObservableCacheEx.MergeManyChangeSets.cs b/src/DynamicData/Cache/ObservableCacheEx.MergeManyChangeSets.cs index 5fcbb9ed0..347f8ec32 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.MergeManyChangeSets.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.MergeManyChangeSets.cs @@ -148,7 +148,7 @@ public static IObservable> MergeManyCh /// /// /// EventBehavior - /// OnErrorAn error from the source (parent) stream or from any child changeset stream terminates the entire output. Unlike , child errors are NOT swallowed. + /// OnErrorAn error from the source (parent) stream or from any child changeset stream terminates the entire output. /// OnCompletedThe output completes when the source (parent) stream completes and all active child changeset streams have also completed. /// /// From 90081348cb63092a9e77c0d5a63813452c9194b6 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 16:50:40 -0700 Subject: [PATCH 5/9] Deliver failures through the observable channel in Watch and LimitSizeTo Both subscribed internally without an error handler and wrapped the source in Finally(observer.OnCompleted). A source failure was therefore reported to the subscriber as a successful completion, while the exception itself was rethrown wherever delivery happened to be. Watch is reachable through any cache built on a failing source. LimitSizeTo also had Finally sitting ahead of ObserveOn, so completion fired on the disposing thread and could arrive before notifications still queued on the scheduler. Completion now travels the same path as the data. TransformWithForcedTransform keeps a second subscription for its item cache, which had no error handler either, so an error reached the observer and was rethrown out of that subscription as well. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Cache/TerminalEventFixture.cs | 72 +++++++++++++++++++ .../Internal/TransformWithForcedTransform.cs | 4 +- src/DynamicData/Cache/ObservableCache.cs | 9 ++- .../Cache/ObservableCacheEx.LimitSizeTo.cs | 9 ++- 4 files changed, 89 insertions(+), 5 deletions(-) create mode 100644 src/DynamicData.Tests/Cache/TerminalEventFixture.cs diff --git a/src/DynamicData.Tests/Cache/TerminalEventFixture.cs b/src/DynamicData.Tests/Cache/TerminalEventFixture.cs new file mode 100644 index 000000000..46d670c2e --- /dev/null +++ b/src/DynamicData.Tests/Cache/TerminalEventFixture.cs @@ -0,0 +1,72 @@ +using System; +using System.Reactive.Concurrency; +using System.Reactive.Subjects; + +using DynamicData.Tests.Domain; + +using FluentAssertions; + +using Xunit; + +namespace DynamicData.Tests.Cache; + +/// +/// Terminal-event behaviour for operators that subscribe internally without exposing the +/// subscription. A missing error handler on one of those internal subscriptions does not fail +/// loudly: the exception is rethrown wherever delivery happened to be, and the subscriber is left +/// believing the sequence ended normally. +/// +public class TerminalEventFixture +{ + [Fact] + public void WatchFailsWhenTheSourceFails() + { + using var source = new Subject>(); + using var cache = new IntermediateCache(source); + + var actualError = default(Exception); + var isCompleted = false; + using var subscription = cache.Watch("Name1").Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true); + + var expectedError = new Exception("Test Exception"); + source.OnError(expectedError); + + actualError.Should().BeSameAs(expectedError, "a watcher should be told when the source fails"); + isCompleted.Should().BeFalse("the source failed, it did not complete"); + } + + [Fact] + public void WatchCompletesWhenTheSourceCompletes() + { + using var source = new Subject>(); + using var cache = new IntermediateCache(source); + + var actualError = default(Exception); + var isCompleted = false; + using var subscription = cache.Watch("Name1").Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true); + + source.OnCompleted(); + + isCompleted.Should().BeTrue("the source completed"); + actualError.Should().BeNull("no error occurred"); + } + + [Fact] + public void LimitSizeToCompletesWhenTheSourceIsDisposed() + { + // The eviction stream is driven by source.Connect(). A plain SourceCache has no upstream that + // can fail, so only the completion path is reachable here, but the handler that carries it is + // the same one that now carries a failure for any ISourceCache implementation that does fail. + // Completion also travels the scheduler now rather than firing ahead of whatever it has queued. + var source = new SourceCache(p => p.Key); + + var actualError = default(Exception); + var isCompleted = false; + using var subscription = source.LimitSizeTo(10, Scheduler.Immediate).Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true); + + source.Dispose(); + + isCompleted.Should().BeTrue("the source completed"); + actualError.Should().BeNull("no error occurred"); + } +} diff --git a/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs b/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs index c7c95aa66..5ab8a3f15 100644 --- a/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs +++ b/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs @@ -20,7 +20,9 @@ public IObservable> Run() => Observable.Create(); - var cacheLoader = shared.Subscribe(changes => cache.Clone(changes)); + // The transform subscription below reports errors to the observer. Without a handler here + // Rx would rethrow them out of this subscription as well. + var cacheLoader = shared.Subscribe(changes => cache.Clone(changes), static _ => { }); // create change set of items where force refresh is applied var refresher = forceTransform.SynchronizeSafe(queue).Select(selector => CaptureChanges(cache, selector)).Select(changes => new ChangeSet(changes)).NotEmpty(); diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index afce06cc6..f774178c6 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -276,7 +276,10 @@ private IObservable> CreateWatchObservable(TKey key) => ? _changes.SkipWhile(_ => Volatile.Read(ref _currentDeliveryVersion) <= snapshotVersion) : _changes; - return changes.Finally(observer.OnCompleted).Subscribe( + // Finally would fire on failure as well, reporting a source error to the watcher as a + // successful completion, and without an error handler the exception would be rethrown + // out of the delivery instead of reaching the observer. + return changes.Subscribe( changes => { foreach (var change in changes) @@ -287,7 +290,9 @@ private IObservable> CreateWatchObservable(TKey key) => observer.OnNext(change); } } - }); + }, + observer.OnError, + observer.OnCompleted); }); /// diff --git a/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs b/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs index 810f118d0..9afa8144c 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs @@ -88,7 +88,10 @@ public static IObservable>> LimitSizeTo< long orderItemWasAdded = -1; var sizeLimiter = new SizeLimiter(sizeLimit); - return source.Connect().Finally(observer.OnCompleted).ObserveOn(scheduler ?? GlobalConfig.DefaultScheduler).Transform((t, v) => new ExpirableItem(t, v, DateTime.Now, Interlocked.Increment(ref orderItemWasAdded))).Select(sizeLimiter.CloneAndReturnExpiredOnly).Where(expired => expired.Length != 0).Subscribe( + // Finally would fire on failure as well, reporting a source error downstream as a + // successful completion, and without an error handler the exception would be rethrown + // out of the delivery instead of reaching the observer. + return source.Connect().ObserveOn(scheduler ?? GlobalConfig.DefaultScheduler).Transform((t, v) => new ExpirableItem(t, v, DateTime.Now, Interlocked.Increment(ref orderItemWasAdded))).Select(sizeLimiter.CloneAndReturnExpiredOnly).Where(expired => expired.Length != 0).Subscribe( toRemove => { try @@ -100,7 +103,9 @@ public static IObservable>> LimitSizeTo< { observer.OnError(ex); } - }); + }, + observer.OnError, + observer.OnCompleted); }); } } From d9f370e1d5289c664368642957a96c824a151739 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 17:31:23 -0700 Subject: [PATCH 6/9] Serialize combined delivery in the cache Combiner Or, And, Xor and Except update shared state under a lock but delivered downstream after releasing it, so two sources written from different threads could both be inside the downstream callback at once. The state stayed intact, which is why this never showed up as a wrong value, but the delivery itself broke the contract every operator downstream relies on. The notification is now taken while the lock is held and drained after it is released, which keeps deliveries ordered and one at a time without a subscriber being able to block a producer. The new fixture drives every multi-source cache operator from several threads at once and checks both for overlapping delivery and for change set integrity. It fails on Or without this change. PopulateInto documented an error as merely terminating the subscription. It is a sink with no downstream observer, so Rx rethrows, which is worth saying. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Cache/MultiSourceSerializationFixture.cs | 279 ++++++++++++++++++ src/DynamicData/Cache/Internal/Combiner.cs | 39 ++- .../Cache/ObservableCacheEx.PopulateInto.cs | 2 +- 3 files changed, 306 insertions(+), 14 deletions(-) create mode 100644 src/DynamicData.Tests/Cache/MultiSourceSerializationFixture.cs diff --git a/src/DynamicData.Tests/Cache/MultiSourceSerializationFixture.cs b/src/DynamicData.Tests/Cache/MultiSourceSerializationFixture.cs new file mode 100644 index 000000000..bebeed4bd --- /dev/null +++ b/src/DynamicData.Tests/Cache/MultiSourceSerializationFixture.cs @@ -0,0 +1,279 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reactive; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Reactive.Threading.Tasks; +using System.Threading; +using System.Threading.Tasks; + +using DynamicData.Binding; +using DynamicData.Tests.Domain; +using DynamicData.Tests.Utilities; + +using FluentAssertions; + +using Xunit; + +using Randomizer = Bogus.Randomizer; + +namespace DynamicData.Tests.Cache; + +/// +/// Serialization coverage for the cache operators that take more than one input. A single input is +/// serialized by whatever feeds it, so these are the operators where two producers can reach the +/// same observer at once, and where a missing gate shows up as overlapping delivery rather than as +/// a wrong value. Every pipeline is checked both for overlap and for change set integrity, since a +/// torn read of shared operator state usually surfaces as a structurally invalid change set rather +/// than as a visible race. +/// +public class MultiSourceSerializationFixture +{ + private const int Seed = 0x1146; + + private static readonly TimeSpan DrainTimeout = TimeSpan.FromMinutes(2); + + [Fact] + public async Task OrDeliversSeriallyWhileBothSourcesAreWritten() + { + using var left = new SourceCache(p => p.Name); + using var right = new SourceCache(p => p.Name); + + await RunAsync( + left.Connect().Or(right.Connect()), + writers: + [ + randomizer => left.AddOrUpdate(new Person("L" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => right.AddOrUpdate(new Person("R" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => left.Remove("L" + randomizer.Int(1, 20)), + ], + complete: () => + { + left.Dispose(); + right.Dispose(); + }); + } + + [Fact] + public async Task SortDeliversSeriallyWhileTheComparerChanges() + { + using var source = new SourceCache(p => p.Name); + using var comparers = new Subject>(); + using var resort = new Subject(); + + var byAge = SortExpressionComparer.Ascending(p => p.Age); + var byName = SortExpressionComparer.Ascending(p => p.Name); + + await RunAsync( + source.Connect().Sort(comparers, resort).Transform(static person => person), + writers: + [ + randomizer => source.AddOrUpdate(new Person("P" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => comparers.OnNext(randomizer.Bool() ? byAge : byName), + _ => resort.OnNext(Unit.Default), + ], + complete: () => + { + comparers.OnCompleted(); + resort.OnCompleted(); + source.Dispose(); + }, + seedAction: () => comparers.OnNext(byAge)); + } + + [Fact] + public async Task GroupOnDeliversSeriallyWhileRegroupingConcurrently() + { + using var source = new SourceCache(p => p.Name); + using var regroup = new Subject(); + + await RunAsync( + source.Connect().Group(p => p.Age % 5, regroup).Transform(group => new Person(group.Key.ToString(), group.Key)).ChangeKey(static person => person.Name), + writers: + [ + randomizer => source.AddOrUpdate(new Person("P" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => source.Remove("P" + randomizer.Int(1, 20)), + _ => regroup.OnNext(Unit.Default), + ], + complete: () => + { + regroup.OnCompleted(); + source.Dispose(); + }); + } + + [Fact] + public async Task BatchIfDeliversSeriallyWhilePausingConcurrently() + { + using var source = new SourceCache(p => p.Name); + using var pause = new Subject(); + + await RunAsync( + source.Connect().BatchIf(pause, initialPauseState: false, timeOut: null), + writers: + [ + randomizer => source.AddOrUpdate(new Person("P" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => source.Remove("P" + randomizer.Int(1, 20)), + randomizer => pause.OnNext(randomizer.Bool()), + ], + complete: () => + { + // Leave the batch open, so anything held back has to be flushed on completion. + pause.OnNext(false); + pause.OnCompleted(); + source.Dispose(); + }); + } + + [Fact] + public async Task TransformWithForcedTransformDeliversSeriallyWhileForcing() + { + using var source = new SourceCache(p => p.Name); + using var force = new Subject>(); + + await RunAsync( + source.Connect().Transform(p => new PersonWithGender(p, p.Age % 2 == 0 ? "M" : "F"), forceTransform: force).Transform(p => new Person(p.Name, p.Age)), + writers: + [ + randomizer => source.AddOrUpdate(new Person("P" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => source.Remove("P" + randomizer.Int(1, 20)), + _ => force.OnNext(static _ => true), + ], + complete: () => + { + force.OnCompleted(); + source.Dispose(); + }); + } + + [Fact] + public async Task DynamicOrDeliversSeriallyWhileBothSourcesAreWritten() + { + using var left = new SourceCache(p => p.Name); + using var right = new SourceCache(p => p.Name); + using var sources = new SourceList>>(); + + sources.Add(left.Connect()); + sources.Add(right.Connect()); + + await RunAsync( + sources.Or(), + writers: + [ + randomizer => left.AddOrUpdate(new Person("L" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => right.AddOrUpdate(new Person("R" + randomizer.Int(1, 20), randomizer.Int(1, 80))), + randomizer => left.Remove("L" + randomizer.Int(1, 20)), + ], + complete: () => + { + // Children first, then the list of sources: the combined result ends only once the + // sources themselves and the list feeding them have all finished. + left.Dispose(); + right.Dispose(); + sources.Dispose(); + }); + } + + [Fact] + public async Task MergeManyChangeSetsDeliversSeriallyWhileChildrenAreWritten() + { + using var owners = new SourceCache(o => o.Id); + var created = new List(); + + for (var i = 0; i < 5; i++) + { + var owner = new AnimalOwner("Owner" + i); + created.Add(owner); + owners.AddOrUpdate(owner); + } + + // The child collections are lists, which allow duplicates, so names are handed out from a + // counter. Otherwise keying the merged result collides for reasons that have nothing to do + // with how the writers interleave. + var nextName = 0; + + await RunAsync( + owners.Connect().MergeManyChangeSets(o => o.Animals.Connect()).Transform(a => new Person(a.Name, a.Name.Length)).AddKey(static person => person.Name), + writers: + [ + randomizer => created[randomizer.Int(0, created.Count - 1)].Animals.Add(new Animal("A" + Interlocked.Increment(ref nextName), "Type", AnimalFamily.Mammal)), + randomizer => + { + var owner = created[randomizer.Int(0, created.Count - 1)]; + var animals = owner.Animals.Items.ToArray(); + if (animals.Length > 0) + { + owner.Animals.Remove(animals[randomizer.Int(0, animals.Length - 1)]); + } + }, + randomizer => created[randomizer.Int(0, created.Count - 1)].Animals.Add(new Animal("B" + Interlocked.Increment(ref nextName), "Type", AnimalFamily.Bird)), + ], + complete: () => + { + // The merged result completes only once the parent and every child have, so the child + // lists have to be disposed as well as the owner cache. + foreach (var owner in created) + { + owner.Dispose(); + } + + owners.Dispose(); + }); + } + + /// + /// Drives every writer from its own thread against a shared barrier, then waits for the pipeline + /// to drain and asserts that it ended cleanly. A serialization failure surfaces as an + /// UnsynchronizedNotificationException on the terminal notification rather than as a + /// wrong count, so the assertion is on how the sequence ended. + /// + private static async Task RunAsync( + IObservable> pipeline, + Action[] writers, + Action complete, + Action? seedAction = null) + { + var randomizer = new Randomizer(Seed); + var iterations = randomizer.Int(150, 250); + + var published = pipeline + .ValidateSynchronization() + .ValidateChangeSets(static person => person.Name) + + // Holding each notification briefly makes an overlapping delivery observable instead of + // something that has to be caught inside a window of a few instructions. + .Do(static _ => Thread.SpinWait(500)) + .Publish(); + + var terminal = published.Materialize().LastAsync().ToTask(); + using var subscription = published.Connect(); + + seedAction?.Invoke(); + + using var barrier = new Barrier(writers.Length + 1); + + var tasks = writers.Select((writer, index) => Task.Run(() => + { + var threadRandomizer = new Randomizer(Seed + index + 1); + barrier.SignalAndWait(); + + for (var i = 0; i < iterations; i++) + { + writer(threadRandomizer); + } + })).ToArray(); + + barrier.SignalAndWait(); + await Task.WhenAll(tasks); + + complete(); + + (await Task.WhenAny(terminal, Task.Delay(DrainTimeout))).Should().BeSameAs(terminal, "the pipeline should drain rather than deadlock"); + + var lastNotification = await terminal; + + lastNotification.Exception.Should().BeNull("notifications must not overlap, and every change set must be structurally valid, however the writers interleave"); + } +} + diff --git a/src/DynamicData/Cache/Internal/Combiner.cs b/src/DynamicData/Cache/Internal/Combiner.cs index 97ee9458b..aa3a7e295 100644 --- a/src/DynamicData/Cache/Internal/Combiner.cs +++ b/src/DynamicData/Cache/Internal/Combiner.cs @@ -2,8 +2,11 @@ // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. +using System.Reactive; using System.Reactive.Disposables; +using DynamicData.Internal; + namespace DynamicData.Cache.Internal; /// @@ -33,6 +36,15 @@ public IDisposable Subscribe(IObservable>[] source, Ac return Disposable.Empty; } + // Each source updates shared state under _locker, but delivery has to be serialized too: + // without this, two sources can compute their notifications, leave the lock, and then both + // be inside updatedCallback at the same time. The queue takes the notification while the + // lock is held and drains it after the lock is released, so deliveries stay ordered and + // one at a time without a subscriber being able to block a producer. + var queue = new DeliveryQueue>( + _locker, + Observer.Create(updatedCallback, onError, onCompleted)); + // subscribe var disposable = new CompositeDisposable(); lock (_locker) @@ -43,13 +55,13 @@ public IDisposable Subscribe(IObservable>[] source, Ac foreach (var pair in source.Zip(_sourceCaches, (item, cache) => new { Item = item, Cache = cache })) { var subscription = pair.Item.Subscribe( - updates => Update(pair.Cache, updates), - onError, + updates => Update(queue, pair.Cache, updates), + queue.OnError, () => { if (Interlocked.Decrement(ref pending) == 0) { - onCompleted(); + queue.OnCompleted(); } }); @@ -57,6 +69,10 @@ public IDisposable Subscribe(IObservable>[] source, Ac } } + // Queue last: the subscriptions are torn down first, so any terminal notification still in + // flight is delivered through a queue that is still running. + disposable.Add(queue); + return disposable; } @@ -91,22 +107,19 @@ private bool MatchesConstraint(TKey key) } } - private void Update(Cache cache, IChangeSet updates) + private void Update(DeliveryQueue> queue, Cache cache, IChangeSet updates) { - ChangeSet notifications; + using var scope = queue.AcquireLock(); - lock (_locker) - { - // update cache for the individual source - cache.Clone(updates); + // update cache for the individual source + cache.Clone(updates); - // update combined - notifications = UpdateCombined(updates); - } + // update combined + var notifications = UpdateCombined(updates); if (notifications.Count != 0) { - updatedCallback(notifications); + scope.EnqueueNext(notifications); } } diff --git a/src/DynamicData/Cache/ObservableCacheEx.PopulateInto.cs b/src/DynamicData/Cache/ObservableCacheEx.PopulateInto.cs index d5ce99d73..ce49086f4 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.PopulateInto.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.PopulateInto.cs @@ -43,7 +43,7 @@ public static partial class ObservableCacheEx /// UpdateThe item is updated in the destination cache via AddOrUpdate. /// RemoveThe item is removed from the destination cache. /// RefreshA Refresh is issued on the destination cache for the item. - /// OnErrorThe subscription is terminated. The destination cache is not rolled back. + /// OnErrorThe subscription ends and the destination cache is not rolled back. Since this method is a sink with no downstream observer, the exception is rethrown by Rx on whichever thread delivered it. Subscribe to directly if you need to handle the failure. /// OnCompletedThe subscription ends. The destination cache retains all items. /// /// From 4bcddacb107d1e817a84c356182be9a42384c71d Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Tue, 28 Jul 2026 08:10:32 -0700 Subject: [PATCH 7/9] Cover the terminal events of the result group source Raised in review as a possible gap. It is not one, but nothing was pinning the behaviour down, so this does. Group finishes when the result group source does, since no group can appear after that, and reports its failure. --- src/DynamicData.Tests/Cache/GroupFixture.cs | 26 +++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/src/DynamicData.Tests/Cache/GroupFixture.cs b/src/DynamicData.Tests/Cache/GroupFixture.cs index 280d01841..0054a920c 100644 --- a/src/DynamicData.Tests/Cache/GroupFixture.cs +++ b/src/DynamicData.Tests/Cache/GroupFixture.cs @@ -308,4 +308,30 @@ public void DeliversTheErrorWhenAResultGroupSourceIsSupplied() error.Should().BeOfType(); } + + [Fact] + public void DeliversTerminalEventsFromTheResultGroupSource() + { + var completed = false; + + using (var source = new Subject>()) + using (var groups = new Subject>()) + using (source.Group(p => p.Age, groups).Subscribe(_ => { }, () => completed = true)) + { + groups.OnCompleted(); + } + + completed.Should().BeTrue("no group can appear once the result group source is finished"); + + Exception? error = null; + + using (var source = new Subject>()) + using (var groups = new Subject>()) + using (source.Group(p => p.Age, groups).Subscribe(_ => { }, ex => error = ex, () => { })) + { + groups.OnError(new InvalidOperationException("boom")); + } + + error.Should().BeOfType(); + } } From a9b1858ccf29851cc7f08d05005a7482cae9e755 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Tue, 28 Jul 2026 10:47:39 -0700 Subject: [PATCH 8/9] Stop a parent update re-adding children that are already present MergeManyChangeSets with list children subscribed the replacement child before taking the previous one's items out, so every item went into the tracker twice and the emitted change set carried an Add for something already downstream, followed by the Remove that balanced it. Anything keying the result saw a duplicate key. This tracker appends without checking, unlike the keyed one, which is why the cache-children variants do not have the same problem and keep their existing order: having the new values present while the old are still there is what turns overlapping keys into Updates rather than Remove and Add churn. Found while chasing what looked like a concurrency fault. It is not: the regression test reproduces it single threaded with two animals and one parent update. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../MergeManyChangeSetsParentUpdateFixture.cs | 43 +++++++++++++++++++ .../Cache/Internal/MergeManyListChangeSets.cs | 8 +++- 2 files changed, 50 insertions(+), 1 deletion(-) create mode 100644 src/DynamicData.Tests/Cache/MergeManyChangeSetsParentUpdateFixture.cs diff --git a/src/DynamicData.Tests/Cache/MergeManyChangeSetsParentUpdateFixture.cs b/src/DynamicData.Tests/Cache/MergeManyChangeSetsParentUpdateFixture.cs new file mode 100644 index 000000000..608f0be64 --- /dev/null +++ b/src/DynamicData.Tests/Cache/MergeManyChangeSetsParentUpdateFixture.cs @@ -0,0 +1,43 @@ +using System; +using System.Linq; + +using DynamicData.Tests.Domain; +using DynamicData.Tests.Utilities; + +using FluentAssertions; + +using Xunit; + +namespace DynamicData.Tests.Cache; + +public class MergeManyChangeSetsParentUpdateFixture +{ + [Fact] + public void ParentUpdateDoesNotEmitDuplicateAdds() + { + using var owners = new SourceCache(o => o.Id); + + var owner = new AnimalOwner("Owner"); + owner.Animals.AddRange([ + new Animal("A1", "Type", AnimalFamily.Mammal), + new Animal("A2", "Type", AnimalFamily.Mammal), + ]); + owners.AddOrUpdate(owner); + + using var subscription = owners.Connect() + .MergeManyChangeSets(o => o.Animals.Connect()) + .Transform(a => new Person(a.Name, a.Name.Length)) + .AddKey(static person => person.Name) + .ValidateChangeSets(static person => person.Name) + .RecordCacheItems(out var results); + + results.Error.Should().BeNull("the initial subscription should be valid"); + + // Re-adding the same instance is an Update on the parent, which swaps the child subscription. + owners.AddOrUpdate(owner); + + results.Error.Should().BeNull("a parent update must not re-add children that are already present"); + results.RecordedItemsByKey.Should().HaveCount(2, "the owner still has exactly two animals"); + } +} + diff --git a/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs b/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs index bd7b18f4d..c97a8f28d 100644 --- a/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs +++ b/src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs @@ -46,11 +46,17 @@ protected override void ParentOnNext(IChangeSet Date: Tue, 28 Jul 2026 10:59:23 -0700 Subject: [PATCH 9/9] Drop the repeated comments on no-op error handlers Each of these sat above a Subscribe passing a discarding onError, saying the same thing about Rx rethrowing without a handler. The argument already says it. Also dropped the two notes about Finally in Watch and LimitSizeTo. They described code that is no longer there, and the terminal event tests are what stop it coming back. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/DynamicData/Binding/BindPaged.cs | 1 - src/DynamicData/Binding/BindVirtualized.cs | 1 - src/DynamicData/Cache/Internal/DynamicCombiner.cs | 4 ---- src/DynamicData/Cache/Internal/GroupOn.cs | 2 -- .../Cache/Internal/TransformWithForcedTransform.cs | 2 -- src/DynamicData/Cache/ObservableCache.cs | 3 --- src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs | 3 --- 7 files changed, 16 deletions(-) diff --git a/src/DynamicData/Binding/BindPaged.cs b/src/DynamicData/Binding/BindPaged.cs index 21e8a067f..5c02c4b04 100644 --- a/src/DynamicData/Binding/BindPaged.cs +++ b/src/DynamicData/Binding/BindPaged.cs @@ -107,7 +107,6 @@ void Finish() comparerSubject.OnNext(changesWithContext.Context.Comparer); changesSubject.OnNext(changesWithContext); }, - // 'subsequent' above relays the error. Without a handler here Rx would rethrow it as well. static _ => { }); return new CompositeDisposable(initial, subscriber, subsequent, shared.Connect()); diff --git a/src/DynamicData/Binding/BindVirtualized.cs b/src/DynamicData/Binding/BindVirtualized.cs index b06e8a85c..97d2b5d5b 100644 --- a/src/DynamicData/Binding/BindVirtualized.cs +++ b/src/DynamicData/Binding/BindVirtualized.cs @@ -105,7 +105,6 @@ void Finish() comparerSubject.OnNext(changesWithContext.Context.Comparer); changesSubject.OnNext(changesWithContext); }, - // 'subsequent' above relays the error. Without a handler here Rx would rethrow it as well. static _ => { }); return new CompositeDisposable(initial, subscriber, subsequent, shared.Connect()); diff --git a/src/DynamicData/Cache/Internal/DynamicCombiner.cs b/src/DynamicData/Cache/Internal/DynamicCombiner.cs index 8db64340f..2793efafd 100644 --- a/src/DynamicData/Cache/Internal/DynamicCombiner.cs +++ b/src/DynamicData/Cache/Internal/DynamicCombiner.cs @@ -62,8 +62,6 @@ public IObservable> Run() => Observable.Create { }, static _ => { }); // when an list is added or removed, need to @@ -83,8 +81,6 @@ public IObservable> Run() => Observable.Create { }, static _ => { }); return new CompositeDisposable(sourceLists, allChanges, removedItem, sourceChanged, sharedLists.Connect(), queue); diff --git a/src/DynamicData/Cache/Internal/GroupOn.cs b/src/DynamicData/Cache/Internal/GroupOn.cs index 550b79e86..075c9ab78 100644 --- a/src/DynamicData/Cache/Internal/GroupOn.cs +++ b/src/DynamicData/Cache/Internal/GroupOn.cs @@ -34,8 +34,6 @@ public IObservable> Run() => Observabl var published = groups.Merge(regroup).Publish(); var subscriber = published.SubscribeSafe(observer); - // The observer above already receives any error. Without a handler here Rx would rethrow it - // out of the subscription instead. var disposer = published.DisposeMany().Subscribe(static _ => { }, static _ => { }); var connected = published.Connect(); diff --git a/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs b/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs index 5ab8a3f15..5955ec0f8 100644 --- a/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs +++ b/src/DynamicData/Cache/Internal/TransformWithForcedTransform.cs @@ -20,8 +20,6 @@ public IObservable> Run() => Observable.Create(); - // The transform subscription below reports errors to the observer. Without a handler here - // Rx would rethrow them out of this subscription as well. var cacheLoader = shared.Subscribe(changes => cache.Clone(changes), static _ => { }); // create change set of items where force refresh is applied diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index f774178c6..42cc495fb 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -276,9 +276,6 @@ private IObservable> CreateWatchObservable(TKey key) => ? _changes.SkipWhile(_ => Volatile.Read(ref _currentDeliveryVersion) <= snapshotVersion) : _changes; - // Finally would fire on failure as well, reporting a source error to the watcher as a - // successful completion, and without an error handler the exception would be rethrown - // out of the delivery instead of reaching the observer. return changes.Subscribe( changes => { diff --git a/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs b/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs index 9afa8144c..ea832e226 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.LimitSizeTo.cs @@ -88,9 +88,6 @@ public static IObservable>> LimitSizeTo< long orderItemWasAdded = -1; var sizeLimiter = new SizeLimiter(sizeLimit); - // Finally would fire on failure as well, reporting a source error downstream as a - // successful completion, and without an error handler the exception would be rethrown - // out of the delivery instead of reaching the observer. return source.Connect().ObserveOn(scheduler ?? GlobalConfig.DefaultScheduler).Transform((t, v) => new ExpirableItem(t, v, DateTime.Now, Interlocked.Increment(ref orderItemWasAdded))).Select(sizeLimiter.CloneAndReturnExpiredOnly).Where(expired => expired.Length != 0).Subscribe( toRemove => {