From 033e179c392409487e9e14ecb80afd59f8be067c Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Sun, 20 Sep 2026 22:06:08 -0700 Subject: [PATCH 1/2] Serialize Optional initialization with source notifications --- ...oObservableOptionalFixture.InitialValue.cs | 299 ++++++++++++++++++ .../Cache/ToObservableOptionalFixture.cs | 10 +- .../ObservableCacheEx.ToObservableOptional.cs | 21 +- 3 files changed, 321 insertions(+), 9 deletions(-) create mode 100644 src/DynamicData.Tests/Cache/ToObservableOptionalFixture.InitialValue.cs diff --git a/src/DynamicData.Tests/Cache/ToObservableOptionalFixture.InitialValue.cs b/src/DynamicData.Tests/Cache/ToObservableOptionalFixture.InitialValue.cs new file mode 100644 index 000000000..06cd45c28 --- /dev/null +++ b/src/DynamicData.Tests/Cache/ToObservableOptionalFixture.InitialValue.cs @@ -0,0 +1,299 @@ +using System; +using System.Linq; +using System.Reactive; +using System.Reactive.Disposables; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Threading; +using System.Threading.Tasks; +using Bogus; +using DynamicData.Kernel; +using DynamicData.Tests.Utilities; +using FluentAssertions; +using Xunit; +using Xunit.Abstractions; + +namespace DynamicData.Tests.Cache; + +public partial class ToObservableOptionalFixture +{ + private const int InitialValueSeed = 0x0F710; + + private readonly ITestOutputHelper _output; + + [Fact] + public async Task InitialOptionalNeverFollowsConcurrentSomeWithoutRemoval() + { + const int maximumIterations = 20_000; + const int maximumUnwatchedAdds = 16; + var value = CreateInitialValue(); + var expected = Optional.Some(value); + var randomizer = new Randomizer(InitialValueSeed); + var unwatchedValues = Enumerable.Range(0, maximumUnwatchedAdds) + .Select(index => Create($"{value.Key}/{index}", randomizer.AlphaNumeric(16))) + .ToArray(); + var changeSets = Enumerable.Range(0, maximumUnwatchedAdds + 1) + .Select(count => new ChangeSet(unwatchedValues.Take(count) + .Select(item => new Change(ChangeReason.Add, item.Key, item)) + .Append(new Change(ChangeReason.Add, value.Key, value)))) + .ToArray(); + var timeout = TimeSpan.FromSeconds(10); + using var cancellation = new CancellationTokenSource(); + using var barrier = new Barrier(2); + IObserver>? sourceObserver = null; + var completedIterations = 0; + var invalidSequences = 0; + var invalidFinalNones = 0; + string? firstFailure = null; + _output.WriteLine("Initial optional race: seed={0}, maximumIterations={1}, maximumUnwatchedAdds={2}", + InitialValueSeed, maximumIterations, maximumUnwatchedAdds); + + // Only the Add races subscription. Vary real changeset work (ignored keys before the watched + // key), not sleeps or spins. Every input contains exactly one Add for the watched key and no Remove. + var producer = Task.Factory.StartNew(() => + { + try + { + for (var iteration = 0; iteration < maximumIterations; iteration++) + { + Rendezvous(iteration, "subscription started"); + sourceObserver!.OnNext(changeSets[iteration % changeSets.Length]); + Rendezvous(iteration, "value delivered and subscription returned"); + } + } + catch (OperationCanceledException) when (cancellation.IsCancellationRequested) + { + // The subscriber stops the bounded run immediately after finding a violation. + } + catch + { + cancellation.Cancel(); + throw; + } + }, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default); + + try + { + for (var iteration = 0; iteration < maximumIterations; iteration++) + { + var source = Observable.Create>(observer => + { + sourceObserver = observer; + Rendezvous(iteration, "subscription started"); + return Disposable.Empty; + }); + using var subscription = source + .ToObservableOptional(value.Key, initialOptionalWhenMissing: true) + .ValidateSynchronization() + .RecordValues(out var results); + + Rendezvous(iteration, "value delivered and subscription returned"); + + // Both racing calls have returned. Complete only now, on this thread, so the probe + // does not also race completion against initialization. No further source delivery + // can occur in this round, and the synchronous terminal notification must be recorded. + sourceObserver!.OnCompleted(); + completedIterations++; + + var values = results.RecordedValues; + var validValues = (values.Count == 1 && values[0].Equals(expected)) + || (values.Count == 2 && !values[0].HasValue && values[1].Equals(expected)); + var validTermination = results.Error is null + && results.HasCompleted + && results.WhenFinalized.IsCompletedSuccessfully + && results.Notifications.Count == values.Count + 1 + && results.Notifications[^1].Value.Kind is NotificationKind.OnCompleted; + if (!validValues || !validTermination) + { + invalidSequences++; + if (values.Count != 0 && !values[^1].HasValue) + invalidFinalNones++; + + var sequence = string.Join(", ", values.Select(optional => optional.HasValue ? $"Some({optional.Value.Value})" : "None")); + firstFailure = $"iteration={iteration}, unwatchedAdds={iteration % changeSets.Length}, values=[{sequence}], completed={results.HasCompleted}, error={results.Error}"; + break; + } + } + } + finally + { + // There is never an unbounded worker left waiting for a round that the subscriber skipped. + cancellation.Cancel(); + await producer.WaitAsync(timeout); + _output.WriteLine("Initial optional race: completedIterations={0}/{1}, invalidSequences={2}, invalidFinalNones={3}, firstFailure={4}", + completedIterations, maximumIterations, invalidSequences, invalidFinalNones, firstFailure ?? "none"); + } + + invalidSequences.Should().Be(0, + "one Add for the watched key with no removal permits only [Some(value)] or [None, Some(value)], followed by completion; {0}", + firstFailure ?? "all iterations satisfied the contract"); + + void Rendezvous(int iteration, string phase) + { + if (!barrier.SignalAndWait(timeout, cancellation.Token)) + throw new TimeoutException($"Initial optional race stalled at iteration {iteration}, phase '{phase}'."); + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void InitialOptionalPreservesSynchronousEmptyCompletion(bool initialOptionalWhenMissing) + { + using var subscription = Observable.Empty>() + .ToObservableOptional(Key1, initialOptionalWhenMissing) + .ValidateSynchronization() + .RecordValues(out var results); + + var expected = initialOptionalWhenMissing + ? new[] { Optional.None() } + : Array.Empty>(); + results.RecordedValues.Should().Equal(expected); + results.Error.Should().BeNull(); + results.HasCompleted.Should().BeTrue(); + results.Notifications.Select(notification => notification.Value.Kind).Should().Equal( + initialOptionalWhenMissing + ? new[] { NotificationKind.OnNext, NotificationKind.OnCompleted } + : new[] { NotificationKind.OnCompleted }); + } + + [Fact] + public void InitialOptionalDoesNotEmitNoneAfterSynchronousError() + { + var error = new InvalidOperationException("The source failed during subscription."); + using var subscription = Observable.Throw>(error) + .ToObservableOptional(Key1, initialOptionalWhenMissing: true) + .ValidateSynchronization() + .RecordValues(out var results); + + results.RecordedValues.Should().BeEmpty(); + results.Error.Should().BeSameAs(error); + results.HasCompleted.Should().BeFalse(); + results.Notifications.Select(notification => notification.Value.Kind).Should().Equal(NotificationKind.OnError); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void InitialOptionalPreservesSynchronousValuesAndCompletion(bool removeBeforeCompletion) + { + var value = CreateInitialValue(); + var subscriptions = 0; + var disposals = 0; + var source = Observable.Create>(observer => + { + subscriptions++; + observer.OnNext(new ChangeSet { new(ChangeReason.Add, value.Key, value) }); + if (removeBeforeCompletion) + observer.OnNext(new ChangeSet { new(ChangeReason.Remove, value.Key, value) }); + observer.OnCompleted(); + return Disposable.Create(() => disposals++); + }); + using var subscription = source + .ToObservableOptional(value.Key, initialOptionalWhenMissing: true) + .ValidateSynchronization() + .RecordValues(out var results); + + var expected = removeBeforeCompletion + ? new[] { Optional.Some(value), Optional.None() } + : new[] { Optional.Some(value) }; + results.RecordedValues.Should().Equal(expected, "initialization must neither prepend nor append None to synchronous source values"); + results.Error.Should().BeNull(); + results.HasCompleted.Should().BeTrue(); + results.Notifications.Select(notification => notification.Value.Kind).Should().Equal( + removeBeforeCompletion + ? new[] { NotificationKind.OnNext, NotificationKind.OnNext, NotificationKind.OnCompleted } + : new[] { NotificationKind.OnNext, NotificationKind.OnCompleted }); + subscriptions.Should().Be(1); + disposals.Should().Be(1); + } + + [Fact] + public void InitialOptionalStateIsPerSubscription() + { + var value = CreateInitialValue(); + using var source = new TestSourceCache(item => item.Key); + var optional = source.Connect().ToObservableOptional(value.Key, initialOptionalWhenMissing: true); + source.AddOrUpdate(value); + using var firstSubscription = optional.ValidateSynchronization().RecordValues(out var firstResults); + + source.RemoveKey(value.Key); + using var secondSubscription = optional.ValidateSynchronization().RecordValues(out var secondResults); + secondResults.RecordedValues.Should().Equal(new[] { Optional.None() }, "the new subscriber starts while the key is absent"); + + source.AddOrUpdate(value); + firstSubscription.Dispose(); + source.RemoveKey(value.Key); + source.Complete(); + + firstResults.RecordedValues.Should().Equal(Optional.Some(value), Optional.None(), Optional.Some(value)); + firstResults.HasCompleted.Should().BeFalse(); + firstResults.Error.Should().BeNull(); + secondResults.RecordedValues.Should().Equal(Optional.None(), Optional.Some(value), Optional.None()); + secondResults.HasCompleted.Should().BeTrue(); + secondResults.Error.Should().BeNull(); + } + + [Fact] + public void InitialOptionalDisposalDuringInitialNoneReleasesSource() + { + var subscriptions = 0; + var disposals = 0; + var source = Observable.Create>(_ => + { + subscriptions++; + return Disposable.Create(() => disposals++); + }); + using var subscription = source + .ToObservableOptional(Key1, initialOptionalWhenMissing: true) + .Take(1) + .ValidateSynchronization() + .RecordValues(out var results); + + results.RecordedValues.Should().Equal(Optional.None()); + results.Error.Should().BeNull(); + results.HasCompleted.Should().BeTrue(); + results.Notifications.Select(notification => notification.Value.Kind) + .Should().Equal(NotificationKind.OnNext, NotificationKind.OnCompleted); + subscriptions.Should().Be(1); + disposals.Should().Be(1, "Take(1) must unsubscribe even though initialization ran before Subscribe returned"); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void InitialOptionalPropagatesAsynchronousTermination(bool failSource) + { + var value = CreateInitialValue(); + var error = new InvalidOperationException("The source failed after subscription."); + using var source = new Subject>(); + using var subscription = source + .ToObservableOptional(value.Key, initialOptionalWhenMissing: true) + .ValidateSynchronization() + .RecordValues(out var results); + + results.RecordedValues.Should().Equal(Optional.None()); + results.HasCompleted.Should().BeFalse(); + source.OnNext(new ChangeSet { new(ChangeReason.Add, value.Key, value) }); + if (failSource) + source.OnError(error); + else + source.OnCompleted(); + + results.RecordedValues.Should().Equal(Optional.None(), Optional.Some(value)); + results.Error.Should().BeSameAs(failSource ? error : null); + results.HasCompleted.Should().Be(!failSource); + results.Notifications.Select(notification => notification.Value.Kind).Should().Equal( + NotificationKind.OnNext, + NotificationKind.OnNext, + failSource ? NotificationKind.OnError : NotificationKind.OnCompleted); + source.HasObservers.Should().BeFalse(); + } + + private KeyValuePair CreateInitialValue() + { + var randomizer = new Randomizer(InitialValueSeed); + _output.WriteLine("Initial optional data: seed={0}", InitialValueSeed); + return Create(randomizer.AlphaNumeric(12), randomizer.AlphaNumeric(16)); + } +} diff --git a/src/DynamicData.Tests/Cache/ToObservableOptionalFixture.cs b/src/DynamicData.Tests/Cache/ToObservableOptionalFixture.cs index d4ea2fd7e..a9cb2212f 100644 --- a/src/DynamicData.Tests/Cache/ToObservableOptionalFixture.cs +++ b/src/DynamicData.Tests/Cache/ToObservableOptionalFixture.cs @@ -8,10 +8,11 @@ using FluentAssertions; using Xunit; +using Xunit.Abstractions; namespace DynamicData.Tests.Cache; -public class ToObservableOptionalFixture : IDisposable +public partial class ToObservableOptionalFixture : IDisposable { private const string Key1 = "Key1"; private const string Key2 = "Key2"; @@ -22,7 +23,11 @@ public class ToObservableOptionalFixture : IDisposable private readonly ISourceCache _source = new SourceCache(kvp => kvp.Key); private readonly ChangeSetAggregator _results; - public ToObservableOptionalFixture() => _results = _source.Connect().AsAggregator(); + public ToObservableOptionalFixture(ITestOutputHelper output) + { + _output = output; + _results = _source.Connect().AsAggregator(); + } public void Dispose() { @@ -306,4 +311,3 @@ private class KeyValuePair(string key, string value) public string Value { get; } = value; } } - diff --git a/src/DynamicData/Cache/ObservableCacheEx.ToObservableOptional.cs b/src/DynamicData/Cache/ObservableCacheEx.ToObservableOptional.cs index 43d2c39eb..89615148e 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.ToObservableOptional.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.ToObservableOptional.cs @@ -73,7 +73,9 @@ public static IObservable> ToObservableOptional /// An observable optional. /// source is null. /// - /// Worth noting: Uses lock-based coordination. If the key exists synchronously on Connect(), the initial None may or may not be emitted depending on timing. + /// An initial None is emitted only if no value from the source has been delivered first. + /// Synchronous initial values therefore suppress it; asynchronous values may be preceded by an initial + /// None, but are never followed by the initialization None. /// public static IObservable> ToObservableOptional(this IObservable> source, TKey key, bool initialOptionalWhenMissing, IEqualityComparer? equalityComparer = null) where TObject : notnull @@ -83,12 +85,19 @@ public static IObservable> ToObservableOptional { return Observable.Defer(() => { - var seenValue = false; + var isFirstNotification = true; return source.ToObservableOptional(key, equalityComparer) - .Do(_ => seenValue = true) - .Merge(Observable.Defer(() => seenValue - ? Observable.Empty>() - : Observable.Return(Optional.None()))); + .Select(static value => (Value: value, IsInitial: false)) + .Merge(Observable.Return((Value: Optional.None(), IsInitial: true))) + .Where(notification => + { + // Decide only after Merge serializes both inputs. Updating before delivery also + // handles reentrant input, without an unbounded notification index. + var shouldEmit = !notification.IsInitial || isFirstNotification; + isFirstNotification = false; + return shouldEmit; + }) + .Select(static notification => notification.Value); }); } From ea3e08d0d7dd5dbc35e89aa7f352f8eb51c49a51 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 21 Sep 2026 07:32:39 -0700 Subject: [PATCH 2/2] Document Optional initialization ordering --- .../dynamicdata-cache.instructions.md | 2 ++ .github/instructions/rx.instructions.md | 28 +++++++++++-------- 2 files changed, 19 insertions(+), 11 deletions(-) diff --git a/.github/instructions/dynamicdata-cache.instructions.md b/.github/instructions/dynamicdata-cache.instructions.md index 45b1874bb..1f9865616 100644 --- a/.github/instructions/dynamicdata-cache.instructions.md +++ b/.github/instructions/dynamicdata-cache.instructions.md @@ -701,6 +701,8 @@ Filters the stream to a single key. Watches a single key and emits `Optional` — `Some` when present, `None` when removed. +With `initialOptionalWhenMissing: true`, initial-value selection is serialized with source notifications. The synthetic `None` can only be the first notification. Later `None` values come from source removals, not initialization. Initialization state is independent for each subscription. + --- ### BatchIf diff --git a/.github/instructions/rx.instructions.md b/.github/instructions/rx.instructions.md index f5f3a21c8..991de8695 100644 --- a/.github/instructions/rx.instructions.md +++ b/.github/instructions/rx.instructions.md @@ -409,9 +409,9 @@ cd.Dispose(); // triggers cancellation ### Composition First — Observable.Create is a Last Resort -**The Rx contracts are axioms, not guidelines.** `Merge` subscribes sequentially. `Defer` evaluates at subscription time. `Do` fires synchronously during delivery. `Concat` subscribes to the second source only after the first completes. These guarantees are unconditional — they hold in every case, on every scheduler, under every threading model. If they didn't, nothing in Rx would work. +**Distinguish subscription order from notification order.** `Defer` evaluates at subscription time, `Do` runs during delivery, and `Concat` subscribes to its second source after the first completes. However, an already-subscribed asynchronous input can deliver while another input is being subscribed. -**Trust the contracts completely.** When you compose operators, you can reason about ordering, state, and lifecycle *because* the contracts are absolute. The moment you doubt them and add "safety" wrappers, you've abandoned the very thing that makes Rx code correct by construction. +**Use the Rx contracts at the actual shared-state boundary.** Per-subscription state is not automatically synchronized. A flag written in an input's `Do` and read by another input's `Defer` can race even when `Merge` subscribes to those inputs sequentially. Put decisions that depend on notification order after the operator that serializes those notifications. **Before reaching for `Observable.Create`, ask: can this be expressed as a composition of existing operators?** Rx operators already handle subscription lifecycle, error propagation, disposal, and serialization. Manual observer forwarding inside `Observable.Create` reimplements all of that — and introduces bugs that the operators would have prevented. @@ -443,20 +443,26 @@ return Observable.Create>(observer => // Each operator does one thing. The intent is immediately clear. return Observable.Defer(() => { - var seenValue = false; + var isFirstNotification = true; return source.ToObservableOptional(key) - .Do(_ => seenValue = true) - .Merge(Observable.Defer(() => seenValue - ? Observable.Empty>() - : Observable.Return(Optional.None()))); + .Select(value => (Value: value, IsInitial: false)) + .Merge(Observable.Return((Value: Optional.None(), IsInitial: true))) + .Where(notification => + { + var shouldEmit = !notification.IsInitial || isFirstNotification; + isFirstNotification = false; + return shouldEmit; + }) + .Select(notification => notification.Value); }); ``` **Why the composition wins:** -- `Defer` creates per-subscription state (the `seenValue` bool) — no shared mutable state -- `Do` captures a side effect without altering the stream — no manual forwarding -- `Merge` with inner `Defer` evaluates the condition *after* the synchronous subscription phase — the `Defer` factory runs when `Merge` subscribes to its second source, which happens after the first source's synchronous emissions -- Error propagation, completion, and disposal are all handled by the operators — zero manual wiring +- `Defer` creates independent first-notification state for every subscription. +- `Merge` serializes actual values and the initialization marker before the state is inspected or changed. +- The marker is emitted only when it wins that serialized order; it cannot overwrite a value that has already been delivered. +- The boolean is updated before downstream delivery, so reentrant input also sees the correct state without an overflowing notification counter. +- Rx operators retain ownership of error propagation, completion, and disposal. **When Observable.Create IS appropriate:** - You need to manage non-Rx resources (event handlers, timers, native resources) tied to subscription lifetime