fix: serialize PropertyObservable / PropertyChangingObservable initial-emit with concurrent handlers

src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs

@@ -60,6 +60,13 @@ internal sealed class Subscription : IDisposable
         /// <summary>The parent observable that owns the source and property metadata.</summary>
         private readonly PropertyChangingObservable<T> _parent;
 
+        /// <summary>
+        /// Serializes the initial emit in the constructor with concurrent <see cref="OnPropertyChanging"/>
+        /// invocations on other threads, so a racing handler emit and the constructor's initial emit do
+        /// not interleave on the downstream observer.
+        /// </summary>
+        private readonly Lock _gate = new();
+
         /// <summary>The downstream observer. Set to <see langword="null"/> on disposal.</summary>
         private IObserver<T>? _observer;
 
@@ -72,10 +79,7 @@ public Subscription(PropertyChangingObservable<T> parent, IObserver<T> observer)
             _observer = observer;
 
             parent._source.PropertyChanging += OnPropertyChanging;
-
-            // Emit initial (StartWith) value
-            var initial = parent._getter(parent._source);
-            observer.OnNext(initial!);
+            EmitCurrent();
         }
 
         /// <inheritdoc/>
@@ -106,14 +110,28 @@ private void OnPropertyChanging(object? sender, PropertyChangingEventArgs e)
                 return;
             }
 
-            var observer = Volatile.Read(ref _observer);
-            if (observer is null)
+            EmitCurrent();
+        }
+
+        /// <summary>
+        /// Reads the current property value under <see cref="_gate"/> and forwards it to the downstream
+        /// observer. Holding <see cref="_gate"/> across the read-emit pair ensures the constructor's
+        /// initial emit and any concurrent <see cref="OnPropertyChanging"/> invocation cannot
+        /// interleave on the downstream observer.
+        /// </summary>
+        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!);
         }
     }
 }

src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs

@@ -68,6 +68,13 @@ internal sealed class Subscription : IDisposable
         /// <summary>The equality comparer used for distinct-until-changed filtering.</summary>
         private readonly EqualityComparer<T> _comparer;
 
+        /// <summary>
+        /// Serializes the initial emit in the constructor with concurrent <see cref="OnPropertyChanged"/>
+        /// invocations on other threads, so the handler always sees a consistent
+        /// <see cref="_hasValue"/> / <see cref="_lastValue"/> snapshot regardless of timing.
+        /// </summary>
+        private readonly Lock _gate = new();
+
         /// <summary>The downstream observer. Set to <see langword="null"/> on disposal.</summary>
         private IObserver<T>? _observer;
 
@@ -90,12 +97,7 @@ public Subscription(PropertyObservable<T> parent, IObserver<T> observer)
             _comparer = EqualityComparer<T>.Default;
 
             parent._source.PropertyChanged += OnPropertyChanged;
-
-            // Emit initial (StartWith) value
-            var initial = parent._getter(parent._source);
-            _lastValue = initial;
-            _hasValue = true;
-            observer.OnNext(initial!);
+            EmitCurrent();
         }
 
         /// <inheritdoc/>
@@ -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!))
+        /// <summary>
+        /// Reads the current property value under <see cref="_gate"/> and forwards it to the downstream
+        /// observer when the distinct-until-changed gate allows. Holding <see cref="_gate"/> across the
+        /// read-decision-emit sequence ensures the constructor's initial emit and any concurrent
+        /// <see cref="OnPropertyChanged"/> invocation cannot interleave on the downstream observer or
+        /// publish a duplicate when both see the same current value.
+        /// </summary>
+        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!);
         }
     }
 }

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;
+
+/// <summary>
+/// Tests that <see cref="PropertyChangingObservable{T}"/> serializes the initial emit it performs while
+/// the subscription is still being built against <see cref="INotifyPropertyChanging.PropertyChanging"/>
+/// notifications arriving at the same time.
+/// </summary>
+/// <remarks>
+/// 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.
+/// </remarks>
+public class PropertyChangingObservableInitialEmitSerializationTests
+{
+    /// <summary>The value the second competing write stores.</summary>
+    private const int SecondWrite = 2;
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous unit test.</returns>
+    [Test]
+    public async Task Subscribe_ConventionalSourceWritesDuringInitialEmit_DoesNotDeliverAStaleValueLast()
+    {
+        var source = new RaiseThenWriteViewModel { Version = 0 };
+        var recorder = new EmissionRecorder<int>();
+
+        using var competitor = new InitialEmitCompetitor(version => source.Version = version);
+
+        var observable = new PropertyChangingObservable<int>(
+            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);
+        }
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous unit test.</returns>
+    [Test]
+    public async Task Subscribe_AtypicalSourceWritesBeforeRaising_DoesNotDeliverAStaleValueLast()
+    {
+        var source = new WriteThenRaiseViewModel { Version = 0 };
+        var recorder = new EmissionRecorder<int>();
+
+        using var competitor = new InitialEmitCompetitor(version => source.Version = version);
+
+        var observable = new PropertyChangingObservable<int>(
+            source,
+            nameof(WriteThenRaiseViewModel.Version),
+            competitor.CreateContendedRead(() => source.Version));
+
+        using (observable.Subscribe(recorder))
+        {
+            competitor.WaitForCompletion();
+
+            await AssertSequence(recorder.Snapshot(), 0, 1, SecondWrite);
+        }
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous unit test.</returns>
+    [Test]
+    public async Task Subscribe_NoConcurrentNotification_EmitsTheInitialValueToEverySubscriber()
+    {
+        var source = new RaiseThenWriteViewModel { Version = 0 };
+        var observable = new PropertyChangingObservable<int>(
+            source,
+            nameof(RaiseThenWriteViewModel.Version),
+            static x => ((RaiseThenWriteViewModel)x).Version);
+
+        var first = new EmissionRecorder<int>();
+        var second = new EmissionRecorder<int>();
+
+        using (observable.Subscribe(first))
+        using (observable.Subscribe(second))
+        {
+            await AssertSequence(first.Snapshot(), 0);
+            await AssertSequence(second.Snapshot(), 0);
+        }
+    }
+
+    /// <summary>Asserts that a recorded emission sequence matches the expected one exactly.</summary>
+    /// <param name="actual">The recorded emissions.</param>
+    /// <param name="expected">The emissions the subscription is required to produce, in order.</param>
+    /// <returns>A <see cref="Task"/> representing the assertion.</returns>
+    private static async Task AssertSequence(IReadOnlyList<int> 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]);
+        }
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    private sealed class InitialEmitCompetitor : IDisposable
+    {
+        /// <summary>
+        /// 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.
+        /// </summary>
+        private const int InterleaveWindowMilliseconds = 500;
+
+        /// <summary>Signals that the competing thread is running and about to write.</summary>
+        private readonly ManualResetEventSlim _started = new(false);
+
+        /// <summary>The thread performing the competing writes.</summary>
+        private readonly Thread _thread;
+
+        /// <summary>Whether the contention has been driven already, so later reads run plainly.</summary>
+        private bool _contended;
+
+        /// <summary>Initializes a new instance of the <see cref="InitialEmitCompetitor"/> class.</summary>
+        /// <param name="write">Writes the observed property.</param>
+        public InitialEmitCompetitor(Action<int> write)
+        {
+            _thread = new(() =>
+            {
+                _started.Set();
+                write(1);
+                write(SecondWrite);
+            }) { IsBackground = true };
+        }
+
+        /// <summary>
+        /// 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.
+        /// </summary>
+        /// <param name="read">Reads the current property value.</param>
+        /// <returns>The property read to hand to the observable.</returns>
+        public Func<INotifyPropertyChanging, int> CreateContendedRead(Func<int> read) => source =>
+        {
+            var valueOnEntry = read();
+
+            if (_contended)
+            {
+                return valueOnEntry;
+            }
+
+            _contended = true;
+            _thread.Start();
+            _started.Wait();
+            _ = _thread.Join(InterleaveWindowMilliseconds);
+
+            return valueOnEntry;
+        };
+
+        /// <summary>Waits for the competing writes to finish.</summary>
+        public void WaitForCompletion() => _thread.Join();
+
+        /// <inheritdoc/>
+        public void Dispose() => _started.Dispose();
+    }
+
+    /// <summary>A view model with the conventional before-change ordering: raise, then write.</summary>
+    private sealed class RaiseThenWriteViewModel : INotifyPropertyChanging
+    {
+        /// <inheritdoc/>
+        public event PropertyChangingEventHandler? PropertyChanging;
+
+        /// <summary>Gets or sets the observed property, raising the event before the write lands.</summary>
+        public int Version
+        {
+            get;
+            set
+            {
+                PropertyChanging?.Invoke(this, new(nameof(Version)));
+                field = value;
+            }
+        }
+    }
+
+    /// <summary>A view model that writes before raising its before-change event, which is atypical.</summary>
+    private sealed class WriteThenRaiseViewModel : INotifyPropertyChanging
+    {
+        /// <inheritdoc/>
+        public event PropertyChangingEventHandler? PropertyChanging;
+
+        /// <summary>Gets or sets the observed property, raising the event after the write has landed.</summary>
+        public int Version
+        {
+            get;
+            set
+            {
+                field = value;
+                PropertyChanging?.Invoke(this, new(nameof(Version)));
+            }
+        }
+    }
+}