Skip to content

Fix that the cache Switch operator never completes #1137

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dwcullop merged 5 commits into from
Jul 30, 2026
Merged

Fix that the cache Switch operator never completes #1137

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
638454a
Fix that the cache Switch operator never completes, without taking a …
dwcullop Jul 27, 2026
7516380
Ignore errors from a superseded source in the cache Switch operator
dwcullop Jul 27, 2026
8d9bdd9
Make the superseded-source test actually exercise the guard
dwcullop Jul 27, 2026
27b8a42
Address review feedback
dwcullop Jul 28, 2026
d6dc7ae
Merge upstream/main into the cache Switch fix
dwcullop Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
132 changes: 74 additions & 58 deletions src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -454,64 +454,80 @@ public void ReSuspendThenResumeDeliversAllInSingleBatch()
results.IsCompleted.Should().BeFalse();
}

[Fact]
public void OnErrorFiresIfCacheFailsWhileSuspended()
{
// A connection made while suspended is deferred, and must still be told when the source
// fails. Reporting the failure as a successful completion leaves the subscriber's error
// handling unrun and its data looking complete.
using var source = new Subject<IChangeSet<int, int>>();
using var cache = new IntermediateCache<int, int>(source);

using var suspend = cache.SuspendNotifications();
using var results = cache.Connect().AsAggregator();

var expectedError = new Exception("Test Exception");
source.OnError(expectedError);

results.Error.Should().Be(expectedError, "a connection deferred by a suspension should see the source fail");
results.IsCompleted.Should().BeFalse("the source failed, it did not complete");
}

[Fact]
public void OnErrorFiresIfCacheFailsWhileWatchIsSuspended()
{
// Watch defers the same way Connect does, and has the same obligation.
using var source = new Subject<IChangeSet<int, int>>();
using var cache = new IntermediateCache<int, int>(source);

using var suspend = cache.SuspendNotifications();
Exception? actualError = null;
var isCompleted = false;
using var subscription = cache.Watch(1).Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true);

var expectedError = new Exception("Test Exception");
source.OnError(expectedError);

actualError.Should().Be(expectedError, "a watch deferred by a suspension should see the source fail");
isCompleted.Should().BeFalse("the source failed, it did not complete");
}

[Fact]
public void OnErrorFiresIfCacheFailsAfterResumingWhileConnectionWasSuspended()
{
// The deferred connection has activated by the time the failure arrives, so this covers
// the path through the connection itself rather than through the suspension gate.
using var source = new Subject<IChangeSet<int, int>>();
using var cache = new IntermediateCache<int, int>(source);

var suspend = cache.SuspendNotifications();
using var results = cache.Connect().AsAggregator();
source.OnNext(new ChangeSet<int, int> { new(ChangeReason.Add, 1, 1) });

suspend.Dispose();
var expectedError = new Exception("Test Exception");
source.OnError(expectedError);

results.Error.Should().Be(expectedError, "an activated connection should still see the source fail");
results.Data.Count.Should().Be(1, "the data written before the failure should have arrived");
}

[Fact]
public void OnErrorFiresIfCacheFailsWhileSuspended()
{
// A connection made while suspended is deferred, and must still be told when the source
// fails. Reporting the failure as a successful completion leaves the subscriber's error
// handling unrun and its data looking complete.
using var source = new Subject<IChangeSet<int, int>>();
using var cache = new IntermediateCache<int, int>(source);

using var suspend = cache.SuspendNotifications();
using var results = cache.Connect().AsAggregator();

var expectedError = new Exception("Test Exception");
source.OnError(expectedError);

results.Error.Should().Be(expectedError, "a connection deferred by a suspension should see the source fail");
results.IsCompleted.Should().BeFalse("the source failed, it did not complete");
}

[Fact]
public void OnErrorFiresIfCacheFailsWhileWatchIsSuspended()
{
// Watch defers the same way Connect does, and has the same obligation.
using var source = new Subject<IChangeSet<int, int>>();
using var cache = new IntermediateCache<int, int>(source);

using var suspend = cache.SuspendNotifications();
Exception? actualError = null;
var isCompleted = false;
using var subscription = cache.Watch(1).Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true);

var expectedError = new Exception("Test Exception");
source.OnError(expectedError);

actualError.Should().Be(expectedError, "a watch deferred by a suspension should see the source fail");
isCompleted.Should().BeFalse("the source failed, it did not complete");
}

[Fact]
public void OnErrorFiresIfCacheFailsAfterResumingWhileConnectionWasSuspended()
{
// The deferred connection has activated by the time the failure arrives, so this covers
// the path through the connection itself rather than through the suspension gate.
using var source = new Subject<IChangeSet<int, int>>();
using var cache = new IntermediateCache<int, int>(source);

var suspend = cache.SuspendNotifications();
using var results = cache.Connect().AsAggregator();
source.OnNext(new ChangeSet<int, int> { new(ChangeReason.Add, 1, 1) });

suspend.Dispose();
var expectedError = new Exception("Test Exception");
source.OnError(expectedError);

results.Error.Should().Be(expectedError, "an activated connection should still see the source fail");
results.Data.Count.Should().Be(1, "the data written before the failure should have arrived");
}

[Fact]
public void OnCompletedFiresIfCacheDisposedAfterResumingWhileWatchWasSuspended()
{
// The tests above cover failure. Completion has to reach an activated watch too, and
// this covers the path through the watch itself rather than through the suspension gate.
var suspend = _source.SuspendNotifications();
var isCompleted = false;
using var subscription = _source.Watch(1).Subscribe(static _ => { }, () => isCompleted = true);
_source.AddOrUpdate(1);

suspend.Dispose();
_source.Dispose();

isCompleted.Should().BeTrue("a watch deferred by a suspension should still complete when the source does");
}

public void Dispose()
{
_source.Dispose();
Expand Down
182 changes: 179 additions & 3 deletions src/DynamicData.Tests/Cache/SwitchFixture.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
using System;
using System.Linq;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Subjects;

using System.Threading;
using DynamicData.Tests.Domain;

using DynamicData.Tests.Utilities;
using FluentAssertions;

using Xunit;

namespace DynamicData.Tests.Cache;
Expand Down Expand Up @@ -89,4 +90,179 @@ public void PropagatesInnerErrors()

results.Error.Should().Be(error);
}

[Fact]
public void CompletesWhenSourcesAndInnerComplete()
{
using var source = new SourceCache<Person, string>(p => p.Name);
using var switchable = new BehaviorSubject<IObservable<IChangeSet<Person, string>>>(source.Connect());
using var results = switchable.Switch().AsAggregator();

source.AddOrUpdate(Enumerable.Range(1, 100).Select(i => new Person("Person" + i, i)).ToArray());

switchable.OnCompleted();
results.IsCompleted.Should().BeFalse("the inner sequence is still running");

source.Dispose();

results.IsCompleted.Should().BeTrue("both the sources and the inner sequence have completed");
results.Error.Should().BeNull();
results.Data.Count.Should().Be(100, "all data should have been received before completion");
}

[Fact]
public void DoesNotCompleteWhileInnerIsStillRunning()
{
using var source = new SourceCache<Person, string>(p => p.Name);
using var switchable = new BehaviorSubject<IObservable<IChangeSet<Person, string>>>(source.Connect());
using var results = switchable.Switch().AsAggregator();

switchable.OnCompleted();
source.AddOrUpdate(new Person("Person1", 1));

results.IsCompleted.Should().BeFalse("the inner sequence has not completed");
results.Data.Count.Should().Be(1, "changes should still flow after the sources sequence completes");
}

[Fact]
public void DoesNotCompleteWhenOnlyASupersededInnerCompletes()
{
using var first = new SourceCache<Person, string>(p => p.Name);
using var second = new SourceCache<Person, string>(p => p.Name);
using var switchable = new BehaviorSubject<IObservable<IChangeSet<Person, string>>>(first.Connect());
using var results = switchable.Switch().AsAggregator();

switchable.OnNext(second.Connect());
switchable.OnCompleted();

first.Dispose();

results.IsCompleted.Should().BeFalse("the superseded sequence is not the current one");

second.AddOrUpdate(new Person("Person1", 1));
results.Data.Count.Should().Be(1, "the current sequence should still be delivering");

second.Dispose();
results.IsCompleted.Should().BeTrue("the current sequence has now completed");
}

[Fact]
public void CompletesWhenSourcesAndInnerCompleteSynchronously()
{
using var results = Observable.Return(Observable.Empty<IChangeSet<Person, string>>()).Switch().AsAggregator();

results.IsCompleted.Should().BeTrue("everything completed during subscription");
results.Error.Should().BeNull();
}

[Fact]
public void DeliversChangesEmittedBeforeSynchronousCompletion()
{
var change = new ChangeSet<Person, string> { new(ChangeReason.Add, "Person1", new Person("Person1", 1)) };
using var results = Observable.Return(Observable.Return((IChangeSet<Person, string>)change)).Switch().AsAggregator();

results.Data.Count.Should().Be(1, "changes emitted before a synchronous completion must not be lost");
results.IsCompleted.Should().BeTrue("the source completed");
results.Error.Should().BeNull();
}

[Fact]
public void IgnoresChangesFromASupersededSource()
{
using var first = new Subject<IChangeSet<Person, string>>();
using var second = new Subject<IChangeSet<Person, string>>();
using var switchable = new BehaviorSubject<IObservable<IChangeSet<Person, string>>>(first);
using var results = switchable.Switch().AsAggregator();

first.OnNext(new ChangeSet<Person, string> { new(ChangeReason.Add, "Person1", new Person("Person1", 1)) });
results.Data.Count.Should().Be(1);

switchable.OnNext(second);
results.Data.Count.Should().Be(0, "moving to a new source drops what the previous one contributed");

first.OnNext(new ChangeSet<Person, string> { new(ChangeReason.Add, "Person2", new Person("Person2", 2)) });

results.Data.Count.Should().Be(0, "a superseded source must not be able to write into the result");
results.Error.Should().BeNull();
}

[Fact]
public void PropagatesInnerErrorsRaisedSynchronously()
{
var error = new Exception("Test");
using var results = Observable.Return(Observable.Throw<IChangeSet<Person, string>>(error)).Switch().AsAggregator();

results.Error.Should().Be(error, "the error was raised during subscription");
}

[Fact]
public void DoesNotHoldALockWhileDeliveringDownstream()
{
// Observable.Switch holds its gate for the whole of downstream delivery, which is the shape that
// deadlocks when a pipeline crosses into another cache. Delivery has to go through the queue, which
// enqueues and returns, so a producer is never held up by whatever a subscriber is doing.
using var switchable = new Subject<IObservable<IChangeSet<Person, string>>>();
using var first = new Subject<IChangeSet<Person, string>>();

using var isDelivering = new ManualResetEventSlim(false);
using var release = new ManualResetEventSlim(false);

using var subscription = switchable.Switch().Subscribe(_ =>
{
isDelivering.Set();
release.Wait(TimeSpan.FromSeconds(10));
});

switchable.OnNext(first);

var deliverer = new Thread(() => first.OnNext(new ChangeSet<Person, string> { new(ChangeReason.Add, "a", new Person("a", 1)) })) { IsBackground = true };
deliverer.Start();

isDelivering.Wait(TimeSpan.FromSeconds(10)).Should().BeTrue("the subscriber should have been handed the change");

var producer = new Thread(() => switchable.OnNext(new Subject<IChangeSet<Person, string>>())) { IsBackground = true };
producer.Start();

var producerFinished = producer.Join(TimeSpan.FromSeconds(2));

release.Set();
deliverer.Join(TimeSpan.FromSeconds(10));
producer.Join(TimeSpan.FromSeconds(10));

producerFinished.Should().BeTrue("writing to the source must not block while a subscriber holds onto a notification");
}

[Fact]
public void IgnoresErrorsFromASupersededSource()
{
// Switching away from a source means anything it produces afterwards belongs to a source that
// is no longer selected, and that includes its failures. Ordinarily disposal stops a
// superseded source being heard from again, but disposal cannot reach a notification that is
// already in flight, so the operator has to discard it on arrival. The raw observable hands
// back the observer directly, which is how that in-flight failure is reproduced here without
// needing a race to land.
var supersededObserver = default(IObserver<IChangeSet<Person, string>>);
var superseded = RawAnonymousObservable.Create<IChangeSet<Person, string>>(observer =>
{
supersededObserver = observer;
return Disposable.Empty;
});

using var switchable = new Subject<IObservable<IChangeSet<Person, string>>>();
using var current = new Subject<IChangeSet<Person, string>>();

using var results = switchable.Switch().AsAggregator();

switchable.OnNext(superseded);
switchable.OnNext(current);

supersededObserver.Should().NotBeNull("the superseded source should have been subscribed");
supersededObserver!.OnError(new Exception("Test"));

results.Error.Should().BeNull("the failed source had already been switched away from");

current.OnNext(new ChangeSet<Person, string> { new(ChangeReason.Add, "a", new Person("a", 1)) });

results.Data.Count.Should().Be(1, "the selected source should still be delivering");
}
}
Loading
Loading