diff --git a/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs b/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs
index fbc1398..9c8b598 100644
--- a/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs
+++ b/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs
@@ -60,6 +60,13 @@ internal sealed class Subscription : IDisposable
/// The parent observable that owns the source and property metadata.
private readonly PropertyChangingObservable _parent;
+ ///
+ /// Serializes the initial emit in the constructor with concurrent
+ /// invocations on other threads, so a racing handler emit and the constructor's initial emit do
+ /// not interleave on the downstream observer.
+ ///
+ private readonly Lock _gate = new();
+
/// The downstream observer. Set to on disposal.
private IObserver? _observer;
@@ -72,10 +79,7 @@ public Subscription(PropertyChangingObservable parent, IObserver observer)
_observer = observer;
parent._source.PropertyChanging += OnPropertyChanging;
-
- // Emit initial (StartWith) value
- var initial = parent._getter(parent._source);
- observer.OnNext(initial!);
+ EmitCurrent();
}
///
@@ -106,14 +110,28 @@ private void OnPropertyChanging(object? sender, PropertyChangingEventArgs e)
return;
}
- var observer = Volatile.Read(ref _observer);
- if (observer is null)
+ EmitCurrent();
+ }
+
+ ///
+ /// Reads the current property value under and forwards it to the downstream
+ /// observer. Holding across the read-emit pair ensures the constructor's
+ /// initial emit and any concurrent invocation cannot
+ /// interleave on the downstream observer.
+ ///
+ private void EmitCurrent()
+ {
+ lock (_gate)
{
- return;
+ var observer = Volatile.Read(ref _observer);
+ if (observer is null)
+ {
+ return;
+ }
+
+ var value = _parent._getter(_parent._source);
+ observer.OnNext(value!);
}
-
- var value = _parent._getter(_parent._source);
- observer.OnNext(value!);
}
}
}
diff --git a/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs b/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs
index cd73d66..7d440df 100644
--- a/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs
+++ b/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs
@@ -68,6 +68,13 @@ internal sealed class Subscription : IDisposable
/// The equality comparer used for distinct-until-changed filtering.
private readonly EqualityComparer _comparer;
+ ///
+ /// Serializes the initial emit in the constructor with concurrent
+ /// invocations on other threads, so the handler always sees a consistent
+ /// / snapshot regardless of timing.
+ ///
+ private readonly Lock _gate = new();
+
/// The downstream observer. Set to on disposal.
private IObserver? _observer;
@@ -90,12 +97,7 @@ public Subscription(PropertyObservable parent, IObserver observer)
_comparer = EqualityComparer.Default;
parent._source.PropertyChanged += OnPropertyChanged;
-
- // Emit initial (StartWith) value
- var initial = parent._getter(parent._source);
- _lastValue = initial;
- _hasValue = true;
- observer.OnNext(initial!);
+ EmitCurrent();
}
///
@@ -129,22 +131,37 @@ private void OnPropertyChanged(object? sender, PropertyChangedEventArgs e)
return;
}
- var observer = Volatile.Read(ref _observer);
- if (observer is null)
- {
- return;
- }
-
- var value = _parent._getter(_parent._source);
+ EmitCurrent();
+ }
- if (_parent._distinctUntilChanged && _hasValue && _comparer.Equals(value!, _lastValue!))
+ ///
+ /// Reads the current property value under and forwards it to the downstream
+ /// observer when the distinct-until-changed gate allows. Holding across the
+ /// read-decision-emit sequence ensures the constructor's initial emit and any concurrent
+ /// invocation cannot interleave on the downstream observer or
+ /// publish a duplicate when both see the same current value.
+ ///
+ private void EmitCurrent()
+ {
+ lock (_gate)
{
- return;
+ var observer = Volatile.Read(ref _observer);
+ if (observer is null)
+ {
+ return;
+ }
+
+ var value = _parent._getter(_parent._source);
+
+ if (_parent._distinctUntilChanged && _hasValue && _comparer.Equals(value!, _lastValue!))
+ {
+ return;
+ }
+
+ _lastValue = value;
+ _hasValue = true;
+ observer.OnNext(value!);
}
-
- _lastValue = value;
- _hasValue = true;
- observer.OnNext(value!);
}
}
}
diff --git a/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyChangingObservableInitialEmitSerializationTests.cs b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyChangingObservableInitialEmitSerializationTests.cs
new file mode 100644
index 0000000..e5ffd2e
--- /dev/null
+++ b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyChangingObservableInitialEmitSerializationTests.cs
@@ -0,0 +1,220 @@
+// Copyright (c) 2019-2026 ReactiveUI Association Incorporated. All rights reserved.
+// ReactiveUI Association Incorporated licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for full license information.
+
+using System.ComponentModel;
+using ReactiveUI.Binding.Observables;
+using ReactiveUI.Binding.Tests.TestModels;
+
+namespace ReactiveUI.Binding.Tests.Observables;
+
+///
+/// Tests that serializes the initial emit it performs while
+/// the subscription is still being built against
+/// notifications arriving at the same time.
+///
+///
+/// The window under test is the gap between the constructor reading the property and delivering that
+/// read downstream. A thread can be descheduled there, and these tests force that schedule by driving
+/// the competing writes from inside the property read itself.
+///
+public class PropertyChangingObservableInitialEmitSerializationTests
+{
+ /// The value the second competing write stores.
+ private const int SecondWrite = 2;
+
+ ///
+ /// A source that raises before it writes - the conventional shape - still lets a stale initial emit
+ /// land after a newer one once two writes pass through the constructor's read-to-emit gap. The
+ /// serialization is therefore load-bearing on this type, not merely a consistency measure.
+ ///
+ /// A representing the asynchronous unit test.
+ [Test]
+ public async Task Subscribe_ConventionalSourceWritesDuringInitialEmit_DoesNotDeliverAStaleValueLast()
+ {
+ var source = new RaiseThenWriteViewModel { Version = 0 };
+ var recorder = new EmissionRecorder();
+
+ using var competitor = new InitialEmitCompetitor(version => source.Version = version);
+
+ var observable = new PropertyChangingObservable(
+ source,
+ nameof(RaiseThenWriteViewModel.Version),
+ competitor.CreateContendedRead(() => source.Version));
+
+ using (observable.Subscribe(recorder))
+ {
+ competitor.WaitForCompletion();
+
+ // The competing emits are held behind the initial emit, so the initial 0 lands first and the
+ // pre-change values that follow it only ever move forward.
+ await AssertSequence(recorder.Snapshot(), 0, 0, 1);
+ }
+ }
+
+ ///
+ /// A source that writes before raising its before-change event - the shape the change calls
+ /// atypical - is the case the serialization is claimed to guard. Without it the handler emits both
+ /// written values before the initial emit lands, leaving the oldest value last.
+ ///
+ /// A representing the asynchronous unit test.
+ [Test]
+ public async Task Subscribe_AtypicalSourceWritesBeforeRaising_DoesNotDeliverAStaleValueLast()
+ {
+ var source = new WriteThenRaiseViewModel { Version = 0 };
+ var recorder = new EmissionRecorder();
+
+ using var competitor = new InitialEmitCompetitor(version => source.Version = version);
+
+ var observable = new PropertyChangingObservable(
+ source,
+ nameof(WriteThenRaiseViewModel.Version),
+ competitor.CreateContendedRead(() => source.Version));
+
+ using (observable.Subscribe(recorder))
+ {
+ competitor.WaitForCompletion();
+
+ await AssertSequence(recorder.Snapshot(), 0, 1, SecondWrite);
+ }
+ }
+
+ ///
+ /// The initial emit stays unconditional on an ordinary subscribe, including for a value equal to the
+ /// default for its type, and every subscriber receives its own.
+ ///
+ /// A representing the asynchronous unit test.
+ [Test]
+ public async Task Subscribe_NoConcurrentNotification_EmitsTheInitialValueToEverySubscriber()
+ {
+ var source = new RaiseThenWriteViewModel { Version = 0 };
+ var observable = new PropertyChangingObservable(
+ source,
+ nameof(RaiseThenWriteViewModel.Version),
+ static x => ((RaiseThenWriteViewModel)x).Version);
+
+ var first = new EmissionRecorder();
+ var second = new EmissionRecorder();
+
+ using (observable.Subscribe(first))
+ using (observable.Subscribe(second))
+ {
+ await AssertSequence(first.Snapshot(), 0);
+ await AssertSequence(second.Snapshot(), 0);
+ }
+ }
+
+ /// Asserts that a recorded emission sequence matches the expected one exactly.
+ /// The recorded emissions.
+ /// The emissions the subscription is required to produce, in order.
+ /// A representing the assertion.
+ private static async Task AssertSequence(IReadOnlyList actual, params int[] expected)
+ {
+ await Assert.That(actual).Count().IsEqualTo(expected.Length);
+
+ for (var index = 0; index < expected.Length; index++)
+ {
+ await Assert.That(actual[index]).IsEqualTo(expected[index]);
+ }
+ }
+
+ ///
+ /// Drives two competing property writes from inside the subscription's initial property read, so
+ /// both notifications fall in the constructor's read-to-emit gap.
+ ///
+ private sealed class InitialEmitCompetitor : IDisposable
+ {
+ ///
+ /// How long to give the competing thread to complete its emits while the initial read and emit
+ /// are still in progress. A serialized subscription blocks it for the whole window, so the wait
+ /// always expires; an unserialized one lets it run to completion in microseconds.
+ ///
+ private const int InterleaveWindowMilliseconds = 500;
+
+ /// Signals that the competing thread is running and about to write.
+ private readonly ManualResetEventSlim _started = new(false);
+
+ /// The thread performing the competing writes.
+ private readonly Thread _thread;
+
+ /// Whether the contention has been driven already, so later reads run plainly.
+ private bool _contended;
+
+ /// Initializes a new instance of the class.
+ /// Writes the observed property.
+ public InitialEmitCompetitor(Action write)
+ {
+ _thread = new(() =>
+ {
+ _started.Set();
+ write(1);
+ write(SecondWrite);
+ }) { IsBackground = true };
+ }
+
+ ///
+ /// Builds a property read that, the first time it runs, releases the competing thread and holds
+ /// for it before returning the value read on entry.
+ ///
+ /// Reads the current property value.
+ /// The property read to hand to the observable.
+ public Func CreateContendedRead(Func read) => source =>
+ {
+ var valueOnEntry = read();
+
+ if (_contended)
+ {
+ return valueOnEntry;
+ }
+
+ _contended = true;
+ _thread.Start();
+ _started.Wait();
+ _ = _thread.Join(InterleaveWindowMilliseconds);
+
+ return valueOnEntry;
+ };
+
+ /// Waits for the competing writes to finish.
+ public void WaitForCompletion() => _thread.Join();
+
+ ///
+ public void Dispose() => _started.Dispose();
+ }
+
+ /// A view model with the conventional before-change ordering: raise, then write.
+ private sealed class RaiseThenWriteViewModel : INotifyPropertyChanging
+ {
+ ///
+ public event PropertyChangingEventHandler? PropertyChanging;
+
+ /// Gets or sets the observed property, raising the event before the write lands.
+ public int Version
+ {
+ get;
+ set
+ {
+ PropertyChanging?.Invoke(this, new(nameof(Version)));
+ field = value;
+ }
+ }
+ }
+
+ /// A view model that writes before raising its before-change event, which is atypical.
+ private sealed class WriteThenRaiseViewModel : INotifyPropertyChanging
+ {
+ ///
+ public event PropertyChangingEventHandler? PropertyChanging;
+
+ /// Gets or sets the observed property, raising the event after the write has landed.
+ public int Version
+ {
+ get;
+ set
+ {
+ field = value;
+ PropertyChanging?.Invoke(this, new(nameof(Version)));
+ }
+ }
+ }
+}
diff --git a/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs
new file mode 100644
index 0000000..eb18798
--- /dev/null
+++ b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs
@@ -0,0 +1,522 @@
+// Copyright (c) 2019-2026 ReactiveUI Association Incorporated. All rights reserved.
+// ReactiveUI Association Incorporated licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for full license information.
+
+using System.ComponentModel;
+using ReactiveUI.Binding.Observables;
+using ReactiveUI.Binding.Tests.TestModels;
+
+namespace ReactiveUI.Binding.Tests.Observables;
+
+///
+/// Tests that serializes the initial emit it performs while the
+/// subscription is still being built against
+/// notifications arriving at the same time.
+///
+///
+/// Every interleaving here is forced rather than raced for: the source's event accessor, the property
+/// read and the downstream observer are each used as a hook to drive a notification into one specific
+/// point of the subscription's construction. Those are the same points a thread can be descheduled at,
+/// so the schedules are real ones - they are only made to happen every run instead of occasionally.
+///
+public class PropertyObservableInitialEmitSerializationTests
+{
+ /// The property value present before any competing write.
+ private const string InitialName = "Alice";
+
+ /// The property value a competing thread writes.
+ private const string UpdatedName = "Bob";
+
+ ///
+ /// How long to give a competing thread to complete its emit while the initial emit is still on the
+ /// stack. A serialized subscription blocks that thread for the whole window, so the wait always
+ /// expires; an unserialized one lets it through in microseconds.
+ ///
+ private const int InterleaveWindowMilliseconds = 500;
+
+ ///
+ /// Subscriptions the unforced sweep builds. Sized from measurement: against unserialized code this
+ /// count reports the defect on every target framework with a wide margin, where a tenth of it misses
+ /// on some of them because the early iterations run before the loop is fully optimized.
+ ///
+ private const int SweepIterations = 10_000;
+
+ ///
+ /// The reported defect: a notification landing after the handler is attached but before the
+ /// constructor has read and emitted delivers the same value twice, breaking the
+ /// no-consecutive-duplicates contract the caller asked for with distinctUntilChanged: true.
+ ///
+ /// A representing the asynchronous unit test.
+ [Test]
+ public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadWhileAttaching_EmitsTheValueOnce()
+ {
+ var source = new HookedViewModel { Name = InitialName };
+ var recorder = new EmissionRecorder();
+
+ // Drive a whole mutation through the source the instant the handler is attached, so the handler
+ // has already emitted by the time the constructor reads the property.
+ source.AfterHandlerAttached = () => RunToCompletionOnAnotherThread(() => source.Name = UpdatedName);
+
+ var observable = new PropertyObservable(
+ source,
+ nameof(HookedViewModel.Name),
+ static x => ((HookedViewModel)x).Name,
+ distinctUntilChanged: true);
+
+ using (observable.Subscribe(recorder))
+ {
+ await AssertNoErrors(recorder);
+ await AssertSequence(recorder.Snapshot(), UpdatedName);
+ }
+ }
+
+ ///
+ /// The same defect reached re-entrantly rather than across threads: the property read the
+ /// constructor performs for its initial emit itself raises
+ /// , so the handler runs part-way through
+ /// construction on the subscribing thread. This also pins that the serialization is re-entrant,
+ /// since a non-re-entrant gate would deadlock here rather than fail.
+ ///
+ /// A representing the asynchronous unit test.
+ [Test]
+ public async Task Subscribe_PropertyChangedRaisedReentrantlyDuringInitialRead_EmitsTheValueOnce()
+ {
+ var source = new HookedViewModel { Name = InitialName };
+ var recorder = new EmissionRecorder();
+ var raised = false;
+
+ string? ReadAndNotifyOnce(INotifyPropertyChanged instance)
+ {
+ if (!raised)
+ {
+ raised = true;
+ source.RaisePropertyChanged(nameof(HookedViewModel.Name));
+ }
+
+ return ((HookedViewModel)instance).Name;
+ }
+
+ var observable = new PropertyObservable(
+ source,
+ nameof(HookedViewModel.Name),
+ ReadAndNotifyOnce,
+ distinctUntilChanged: true);
+
+ using (observable.Subscribe(recorder))
+ {
+ await AssertNoErrors(recorder);
+ await AssertSequence(recorder.Snapshot(), InitialName);
+ }
+ }
+
+ ///
+ /// Pins the change's central claim - that a competing handler runs either wholly before or wholly
+ /// after the initial emit, never inside it - by holding a competing thread against the initial emit
+ /// while it is on the stack and recording whether its emit overlaps.
+ ///
+ /// A representing the asynchronous unit test.
+ [Test]
+ public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadDuringInitialEmit_DoesNotOverlapTheInitialEmit()
+ {
+ var source = new HookedViewModel { Name = InitialName };
+ using var competitorStarted = new ManualResetEventSlim(false);
+ Thread? competitor = null;
+
+ // Runs from inside the downstream call of the initial emit, which is the window the change keeps
+ // exclusive. The bounded join is what a blocked competing thread looks like from in here.
+ var recorder = new EmissionRecorder
+ {
+ OnFirstValue = () =>
+ {
+ competitor = new(() =>
+ {
+ competitorStarted.Set();
+ source.Name = UpdatedName;
+ }) { IsBackground = true };
+
+ competitor.Start();
+ competitorStarted.Wait();
+ _ = competitor.Join(InterleaveWindowMilliseconds);
+ },
+ };
+
+ var observable = new PropertyObservable(
+ source,
+ nameof(HookedViewModel.Name),
+ static x => ((HookedViewModel)x).Name,
+ distinctUntilChanged: true);
+
+ using (observable.Subscribe(recorder))
+ {
+ competitor!.Join();
+
+ await AssertNoErrors(recorder);
+ await Assert.That(recorder.MaxConcurrentEmissions).IsEqualTo(1);
+ await AssertSequence(recorder.Snapshot(), InitialName, UpdatedName);
+ }
+ }
+
+ ///
+ /// The initial emit stays unconditional on an ordinary subscribe, including when the value equals
+ /// the default for its type. The change applies the distinct-until-changed test to the initial emit
+ /// as well, so this guards the first emission against being swallowed by that new test.
+ ///
+ /// Whether the subscription suppresses consecutive duplicates.
+ /// A representing the asynchronous unit test.
+ [Test]
+ [Arguments(true)]
+ [Arguments(false)]
+ public async Task Subscribe_ValueIsTheTypeDefault_StillEmitsTheInitialValue(bool distinctUntilChanged)
+ {
+ var source = new HookedViewModel { Name = null, Count = 0 };
+
+ var nameRecorder = new EmissionRecorder();
+ var nameObservable = new PropertyObservable(
+ source,
+ nameof(HookedViewModel.Name),
+ static x => ((HookedViewModel)x).Name,
+ distinctUntilChanged);
+
+ var countRecorder = new EmissionRecorder();
+ var countObservable = new PropertyObservable(
+ source,
+ nameof(HookedViewModel.Count),
+ static x => ((HookedViewModel)x).Count,
+ distinctUntilChanged);
+
+ using (nameObservable.Subscribe(nameRecorder))
+ using (countObservable.Subscribe(countRecorder))
+ {
+ await Assert.That(nameRecorder.Snapshot()).Count().IsEqualTo(1);
+ await Assert.That(nameRecorder.Snapshot()[0]).IsNull();
+ await AssertSequence(countRecorder.Snapshot(), 0);
+ }
+ }
+
+ ///
+ /// Each subscription carries its own distinct-until-changed state, so a later subscriber still
+ /// receives the initial value even though an existing subscriber has already been given it, and a
+ /// re-subscription after disposal receives it again.
+ ///
+ /// Whether the subscription suppresses consecutive duplicates.
+ /// A representing the asynchronous unit test.
+ [Test]
+ [Arguments(true)]
+ [Arguments(false)]
+ public async Task Subscribe_ValueAlreadyDeliveredToAnotherSubscriber_StillEmitsTheInitialValue(bool distinctUntilChanged)
+ {
+ var source = new HookedViewModel { Name = InitialName };
+ var observable = new PropertyObservable(
+ source,
+ nameof(HookedViewModel.Name),
+ static x => ((HookedViewModel)x).Name,
+ distinctUntilChanged);
+
+ var first = new EmissionRecorder();
+ var second = new EmissionRecorder();
+ var afterDisposal = new EmissionRecorder();
+
+ using (observable.Subscribe(first))
+ using (observable.Subscribe(second))
+ {
+ await AssertSequence(first.Snapshot(), InitialName);
+ await AssertSequence(second.Snapshot(), InitialName);
+ }
+
+ using (observable.Subscribe(afterDisposal))
+ {
+ await AssertSequence(afterDisposal.Snapshot(), InitialName);
+ }
+ }
+
+ ///
+ /// An unforced sweep over the same window, reaching interleavings the forced tests do not enumerate.
+ /// A competing thread writes a fresh value continuously while a subscription is built against it, so
+ /// the subscribe always lands mid-storm, and no subscriber may ever see two equal values in a row.
+ ///
+ /// A representing the asynchronous unit test.
+ [Test]
+ public async Task Subscribe_ConcurrentWritesThroughoutSubscription_NeverEmitsConsecutiveDuplicates()
+ {
+ var iterationsThatEmitted = 0;
+ var iterationsWithDuplicate = 0;
+
+ for (var iteration = 0; iteration < SweepIterations; iteration++)
+ {
+ var source = new HookedViewModel { Count = 0 };
+ using var competitorStarted = new ManualResetEventSlim(false);
+ var stopped = 0;
+
+ var competitor = new Thread(() =>
+ {
+ var next = 0;
+ competitorStarted.Set();
+ while (Volatile.Read(ref stopped) == 0)
+ {
+ source.Count = ++next;
+ }
+ }) { IsBackground = true };
+
+ competitor.Start();
+ competitorStarted.Wait();
+
+ var recorder = new EmissionRecorder();
+ var observable = new PropertyObservable(
+ source,
+ nameof(HookedViewModel.Count),
+ static x => ((HookedViewModel)x).Count,
+ distinctUntilChanged: true);
+
+ using (observable.Subscribe(recorder))
+ {
+ Volatile.Write(ref stopped, 1);
+ competitor.Join();
+ }
+
+ var observed = recorder.Snapshot();
+ if (observed.Count > 0)
+ {
+ iterationsThatEmitted++;
+ }
+
+ for (var index = 1; index < observed.Count; index++)
+ {
+ if (observed[index] != observed[index - 1])
+ {
+ continue;
+ }
+
+ iterationsWithDuplicate++;
+ break;
+ }
+ }
+
+ await Assert.That(iterationsWithDuplicate).IsEqualTo(0);
+ await Assert.That(iterationsThatEmitted).IsEqualTo(SweepIterations);
+ }
+
+ ///
+ /// A source raises its event off a handler snapshot taken before the subscription attached, so the
+ /// notification for that write never reaches the new handler. On the conventional write-then-raise
+ /// setter the value is still not lost: the constructor reads the property after attaching, and that
+ /// read is the backstop.
+ ///
+ /// A representing the asynchronous unit test.
+ [Test]
+ public async Task Subscribe_HandlerAttachesAfterTheRaiserSnapshotsItsDelegate_StillObservesTheWrittenValue()
+ {
+ var source = new ConventionalRaiseOrderViewModel { Name = InitialName };
+ var recorder = new EmissionRecorder();
+ using var subscriptions = new GrowableCompositeDisposable();
+
+ source.BeforeRaise = () => subscriptions.Add(new PropertyObservable(
+ source,
+ nameof(ConventionalRaiseOrderViewModel.Name),
+ static x => ((ConventionalRaiseOrderViewModel)x).Name,
+ distinctUntilChanged: true).Subscribe(recorder));
+
+ source.Name = UpdatedName;
+
+ await Assert.That(source.HandlerWasInvoked).IsFalse();
+ await AssertSequence(recorder.Snapshot(), UpdatedName);
+ }
+
+ ///
+ /// The same missed notification against a source that raises before it writes. The backstop read
+ /// runs too early there, so the subscriber is left holding the pre-write value - a gap owned by the
+ /// source's ordering rather than by the subscription, and one no amount of serialization closes.
+ ///
+ /// A representing the asynchronous unit test.
+ [Test]
+ public async Task Subscribe_SourceRaisesPropertyChangedBeforeWritingTheField_ObservesThePreWriteValue()
+ {
+ var source = new RaiseBeforeWriteViewModel { Name = InitialName };
+ var recorder = new EmissionRecorder();
+ using var subscriptions = new GrowableCompositeDisposable();
+
+ source.BeforeRaise = () => subscriptions.Add(new PropertyObservable(
+ source,
+ nameof(RaiseBeforeWriteViewModel.Name),
+ static x => ((RaiseBeforeWriteViewModel)x).Name,
+ distinctUntilChanged: true).Subscribe(recorder));
+
+ source.Name = UpdatedName;
+
+ await Assert.That(source.HandlerWasInvoked).IsFalse();
+ await Assert.That(source.Name).IsEqualTo(UpdatedName);
+ await AssertSequence(recorder.Snapshot(), InitialName);
+ }
+
+ /// Asserts that nothing was pushed to the observer's error channel.
+ /// The emitted element type.
+ /// The recorder to inspect.
+ /// A representing the assertion.
+ private static async Task AssertNoErrors(EmissionRecorder recorder) =>
+ await Assert.That(recorder.ErrorSnapshot()).IsEmpty();
+
+ /// Asserts that a recorded emission sequence matches the expected one exactly.
+ /// The emitted element type.
+ /// The recorded emissions.
+ /// The emissions the subscription is required to produce, in order.
+ /// A representing the assertion.
+ private static async Task AssertSequence(IReadOnlyList actual, params T[] expected)
+ {
+ await Assert.That(actual).Count().IsEqualTo(expected.Length);
+
+ for (var index = 0; index < expected.Length; index++)
+ {
+ await Assert.That(actual[index]).IsEqualTo(expected[index]);
+ }
+ }
+
+ /// Runs an action on a dedicated thread and waits for it to finish.
+ /// The work to run away from the calling thread.
+ private static void RunToCompletionOnAnotherThread(Action action)
+ {
+ var thread = new Thread(action.Invoke) { IsBackground = true };
+ thread.Start();
+ thread.Join();
+ }
+
+ ///
+ /// A view model whose event accessor exposes the instant a handler becomes attached, which is where
+ /// the subscribe-time window opens.
+ ///
+ private sealed class HookedViewModel : INotifyPropertyChanged
+ {
+ /// Serializes handler registration against deregistration.
+ private readonly Lock _handlerGate = new();
+
+ /// The registered handlers.
+ private PropertyChangedEventHandler? _handlers;
+
+ ///
+ public event PropertyChangedEventHandler? PropertyChanged
+ {
+ add
+ {
+ lock (_handlerGate)
+ {
+ _handlers += value;
+ }
+
+ var hook = AfterHandlerAttached;
+ AfterHandlerAttached = null;
+ hook?.Invoke();
+ }
+
+ remove
+ {
+ lock (_handlerGate)
+ {
+ _handlers -= value;
+ }
+ }
+ }
+
+ /// Gets or sets a one-shot action run once a handler is added, from the adding thread.
+ public Action? AfterHandlerAttached { get; set; }
+
+ /// Gets or sets the observed reference-typed property, written before the event is raised.
+ public string? Name
+ {
+ get;
+ set
+ {
+ field = value;
+ RaisePropertyChanged(nameof(Name));
+ }
+ }
+
+ /// Gets or sets a value-typed property used to check the default-valued initial emit.
+ public int Count
+ {
+ get;
+ set
+ {
+ field = value;
+ RaisePropertyChanged(nameof(Count));
+ }
+ }
+
+ /// Raises without writing anything.
+ /// The property to report.
+ public void RaisePropertyChanged(string propertyName) => _handlers?.Invoke(this, new(propertyName));
+ }
+
+ ///
+ /// A view model with the conventional raise order - write the field, snapshot the handlers, raise -
+ /// exposing the point between the snapshot and the raise.
+ ///
+ private sealed class ConventionalRaiseOrderViewModel : INotifyPropertyChanged
+ {
+ ///
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ /// Gets or sets a one-shot action run after the handler snapshot is taken and before it is invoked.
+ public Action? BeforeRaise { get; set; }
+
+ /// Gets a value indicating whether the raise reached any handler.
+ public bool HandlerWasInvoked { get; private set; }
+
+ /// Gets or sets the observed property.
+ public string? Name
+ {
+ get;
+ set
+ {
+ field = value;
+
+ var snapshot = PropertyChanged;
+ var hook = BeforeRaise;
+ BeforeRaise = null;
+ hook?.Invoke();
+
+ if (snapshot is null)
+ {
+ return;
+ }
+
+ HandlerWasInvoked = true;
+ snapshot(this, new(nameof(Name)));
+ }
+ }
+ }
+
+ ///
+ /// A view model that raises before it writes, which is the ordering under which the subscription's
+ /// post-attach read cannot serve as a backstop.
+ ///
+ private sealed class RaiseBeforeWriteViewModel : INotifyPropertyChanged
+ {
+ ///
+ public event PropertyChangedEventHandler? PropertyChanged;
+
+ /// Gets or sets a one-shot action run after the handler snapshot is taken and before it is invoked.
+ public Action? BeforeRaise { get; set; }
+
+ /// Gets a value indicating whether the raise reached any handler.
+ public bool HandlerWasInvoked { get; private set; }
+
+ /// Gets or sets the observed property.
+ public string? Name
+ {
+ get;
+ set
+ {
+ var snapshot = PropertyChanged;
+ var hook = BeforeRaise;
+ BeforeRaise = null;
+ hook?.Invoke();
+
+ if (snapshot is not null)
+ {
+ HandlerWasInvoked = true;
+ snapshot(this, new(nameof(Name)));
+ }
+
+ field = value;
+ }
+ }
+ }
+}
diff --git a/src/tests/ReactiveUI.Binding.Tests/TestModels/EmissionRecorder.cs b/src/tests/ReactiveUI.Binding.Tests/TestModels/EmissionRecorder.cs
new file mode 100644
index 0000000..62f5618
--- /dev/null
+++ b/src/tests/ReactiveUI.Binding.Tests/TestModels/EmissionRecorder.cs
@@ -0,0 +1,116 @@
+// Copyright (c) 2019-2026 ReactiveUI Association Incorporated. All rights reserved.
+// ReactiveUI Association Incorporated licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for full license information.
+
+namespace ReactiveUI.Binding.Tests.TestModels;
+
+///
+/// An observer that records what it was given and how the calls overlapped, for tests that need to tell
+/// a serialized producer from one whose emissions can run concurrently on the downstream observer.
+///
+/// The emitted element type.
+internal sealed class EmissionRecorder : IObserver
+{
+ /// Serializes the recorded state against emissions arriving on several threads.
+ private readonly Lock _gate = new();
+
+ /// The values received so far, in arrival order.
+ private readonly List _values = [];
+
+ /// The errors received so far.
+ private readonly List _errors = [];
+
+ /// The number of emissions currently in flight.
+ private int _inFlight;
+
+ /// The high-water mark of .
+ private int _maxInFlight;
+
+ ///
+ /// Gets or sets a one-shot action run from inside the first emission, after the value has been
+ /// recorded and while that call still counts as in flight. It runs outside this recorder's own lock
+ /// so that what it measures is the producer's serialization rather than the recorder's.
+ ///
+ internal Action? OnFirstValue { get; set; }
+
+ ///
+ /// Gets the greatest number of emissions that were ever in flight at once. Above one means the
+ /// producer let two emissions overlap on the downstream observer.
+ ///
+ internal int MaxConcurrentEmissions
+ {
+ get
+ {
+ lock (_gate)
+ {
+ return _maxInFlight;
+ }
+ }
+ }
+
+ ///
+ public void OnNext(T value)
+ {
+ Action? hook;
+
+ lock (_gate)
+ {
+ _inFlight++;
+ if (_inFlight > _maxInFlight)
+ {
+ _maxInFlight = _inFlight;
+ }
+
+ _values.Add(value);
+
+ hook = _values.Count == 1 ? OnFirstValue : null;
+ OnFirstValue = null;
+ }
+
+ try
+ {
+ hook?.Invoke();
+ }
+ finally
+ {
+ lock (_gate)
+ {
+ _inFlight--;
+ }
+ }
+ }
+
+ ///
+ public void OnError(Exception error)
+ {
+ lock (_gate)
+ {
+ _errors.Add(error);
+ }
+ }
+
+ ///
+ public void OnCompleted()
+ {
+ }
+
+ /// Takes a copy of the values recorded so far.
+ /// The values received, in arrival order.
+ internal IReadOnlyList Snapshot()
+ {
+ lock (_gate)
+ {
+ return [.. _values];
+ }
+ }
+
+ /// Takes a copy of the errors recorded so far.
+ /// The errors received, in arrival order.
+ internal IReadOnlyList ErrorSnapshot()
+ {
+ lock (_gate)
+ {
+ return [.. _errors];
+ }
+ }
+}