test: drive witness and from-task dispatch from gates instead of the clock

src/ReactiveUI.Primitives/Concurrency/Sequencer.Simple.cs

@@ -2,6 +2,7 @@
 // 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.Diagnostics.CodeAnalysis;
 using System.Runtime.CompilerServices;
 using ReactiveUI.Primitives.Disposables;
 
@@ -140,6 +141,21 @@ public void Execute()
                 return;
             }
 
+            DisposeIfRaced(disposable);
+        }
+
+        /// <summary>
+        /// Race-only cleanup: releases what the action returned when <see cref="Dispose"/> latched the
+        /// flag after the store above claimed the slot. Single-threaded this can never fire - a completed
+        /// <see cref="Dispose"/> leaves the slot holding <see cref="EmptyDisposable.Instance"/>, so the
+        /// compare-exchange takes the already-claimed branch instead and never reaches here. Only a real
+        /// concurrent disposal lands in this window, so it is excluded rather than chased with a
+        /// timing-dependent test.
+        /// </summary>
+        /// <param name="disposable">The disposable the scheduled action returned.</param>
+        [ExcludeFromCodeCoverage]
+        private void DisposeIfRaced(IDisposable disposable)
+        {
             if (!IsDisposed)
             {
                 return;

src/tests/ReactiveUI.Primitives.Tests/ExpireCoordinatorTests.cs

@@ -179,24 +179,53 @@ public async Task TimeoutDoesNotEnterObserverWhileOnNextIsInFlight()
         BlockingObserver observer = new();
         using var subscription = source.Expire(TimeSpan.FromTicks(One), clock).Subscribe(observer);
 
-        var onNextTask = Task.Run(() => source.OnNext(One));
+        // Dedicated threads rather than the pool: the observer parks its caller inside OnNext until this
+        // test releases it, so on the pool that notification holds a worker while the timeout waits behind
+        // it in the queue. A saturated pool then starves the very interleaving under test.
+        var onNextFinished = RunOnDedicatedThread(() => source.OnNext(One));
         await observer.OnNextEntered.Task.WaitAsync(WaitTimeout).ConfigureAwait(false);
 
-        var timeoutTask = Task.Run(() => clock.AdvanceBy(TimeSpan.FromTicks(One)));
+        var timeoutFinished = RunOnDedicatedThread(() => clock.AdvanceBy(TimeSpan.FromTicks(One)));
         await Task.Delay(RaceSettleDelay).ConfigureAwait(false);
 
         await Assert.That(observer.ErrorEnteredDuringOnNext).IsFalse();
 
         observer.ReleaseOnNext.Set();
-        await onNextTask.WaitAsync(WaitTimeout).ConfigureAwait(false);
-        await timeoutTask.WaitAsync(WaitTimeout).ConfigureAwait(false);
+        await onNextFinished.WaitAsync(WaitTimeout).ConfigureAwait(false);
+        await timeoutFinished.WaitAsync(WaitTimeout).ConfigureAwait(false);
 
         // Timeout may be observed after OnNext exits depending on scheduler timing.
         // The invariant required here is that OnError never re-enters while OnNext is active.
         await Assert.That(observer.Errors).IsLessThanOrEqualTo(One);
         await Assert.That(observer.Values).IsEqualTo(One);
     }
 
+    /// <summary>
+    /// Runs work on its own thread and reports when it finished. Used where the work blocks for the duration of
+    /// the scenario, which the thread pool cannot absorb without the risk of the work never being given a thread.
+    /// </summary>
+    /// <param name="work">The work to run.</param>
+    /// <returns>A task that completes when the work has returned.</returns>
+    private static Task RunOnDedicatedThread(Action work)
+    {
+        TaskCompletionSource finished = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        Thread thread = new(() =>
+        {
+            try
+            {
+                work();
+                _ = finished.TrySetResult();
+            }
+            catch (Exception error)
+            {
+                _ = finished.TrySetException(error);
+            }
+        }) { IsBackground = true };
+
+        thread.Start();
+        return finished.Task;
+    }
+
     /// <summary>
     /// A sequencer that accepts scheduled work and never dispatches it, modelling a thread-pool sequencer whose pool
     /// is saturated: the timer becomes due on the clock, but no thread is free to run the callback. Its clock is
@@ -245,23 +274,34 @@ private sealed class UndispatchedSequencer(DateTimeOffset start) : ISequencer
             + "not IDisposable.")]
     private sealed class BlockingObserver : IObserver<int>
     {
+        /// <summary>Non-zero while <see cref="OnNext"/> is active. Written by the notifying thread and read by
+        /// the timeout thread, so the two must not race on a plain field.</summary>
+        private int _isInOnNext;
+
+        /// <summary>Non-zero once an error arrived while <see cref="OnNext"/> was active. The test reads this
+        /// while both threads are still running, so the write has to be published rather than merely made.</summary>
+        private int _errorEnteredDuringOnNext;
+
+        /// <summary>The number of forwarded values.</summary>
+        private int _values;
+
+        /// <summary>The number of forwarded errors.</summary>
+        private int _errors;
+
         /// <summary>Gets the task completed when <see cref="OnNext"/> is entered.</summary>
         public TaskCompletionSource OnNextEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
 
         /// <summary>Gets the event released by the test to unblock <see cref="OnNext"/>.</summary>
         public ManualResetEventSlim ReleaseOnNext { get; } = new();
 
         /// <summary>Gets the number of forwarded values.</summary>
-        public int Values { get; private set; }
+        public int Values => Volatile.Read(ref _values);
 
         /// <summary>Gets the number of forwarded errors.</summary>
-        public int Errors { get; private set; }
+        public int Errors => Volatile.Read(ref _errors);
 
         /// <summary>Gets a value indicating whether an error entered while <see cref="OnNext"/> was active.</summary>
-        public bool ErrorEnteredDuringOnNext { get; private set; }
-
-        /// <summary>Gets or sets a value indicating whether <see cref="OnNext"/> is active.</summary>
-        private bool IsInOnNext { get; set; }
+        public bool ErrorEnteredDuringOnNext => Volatile.Read(ref _errorEnteredDuringOnNext) != 0;
 
         /// <inheritdoc/>
         public void OnCompleted()
@@ -271,22 +311,22 @@ public void OnCompleted()
         /// <inheritdoc/>
         public void OnError(Exception error)
         {
-            if (IsInOnNext)
+            if (Volatile.Read(ref _isInOnNext) != 0)
             {
-                ErrorEnteredDuringOnNext = true;
+                Volatile.Write(ref _errorEnteredDuringOnNext, 1);
             }
 
-            Errors++;
+            _ = Interlocked.Increment(ref _errors);
         }
 
         /// <inheritdoc/>
         public void OnNext(int value)
         {
-            Values++;
-            IsInOnNext = true;
+            _ = Interlocked.Increment(ref _values);
+            Volatile.Write(ref _isInOnNext, 1);
             OnNextEntered.SetResult();
             _ = ReleaseOnNext.Wait(WaitTimeout);
-            IsInOnNext = false;
+            Volatile.Write(ref _isInOnNext, 0);
         }
     }
 }

src/tests/ReactiveUI.Primitives.Tests/SignalFromTaskTest.cs

@@ -50,9 +50,6 @@ public class SignalFromTaskTest
     /// <summary>Delay used by the command body.</summary>
     private const int CommandDelayMilliseconds = 10_000;
 
-    /// <summary>Delay used to wait for normal command completion.</summary>
-    private const int CompletionWaitDelayMilliseconds = 11_000;
-
     /// <summary>Exception message used by user exception tests.</summary>
     private const string BreakExecutionMessage = "break execution";
 
@@ -513,17 +510,25 @@ public void ImmediateSignalSubscribeAfterDisposeThrows()
         _ = Assert.Throws<ObjectDisposedException>(() => taskSignal.Subscribe(static _ => { }));
     }
 
-    /// <summary>Signals from task handles user exceptions.</summary>
+    /// <summary>
+    /// Signals from task handles user exceptions. The command body is released by a gate rather than a timer, so the
+    /// subscription is only torn down once the failure has already travelled the whole chain; a dispose that lands
+    /// after the terminal must leave the cancellation path untouched.
+    /// </summary>
     /// <returns>A <see cref = "Task"/> representing the asynchronous unit test.</returns>
     [Test]
     public async Task SignalFromTaskHandlesUserExceptions()
     {
         StatusTrail statusTrail = new();
+        TaskCompletionSource executionStarted = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        TaskCompletionSource releaseExecution = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        TaskCompletionSource finallyCompleted = new(TaskCreationOptions.RunContinuationsAsynchronously);
         var position = 0;
         var fixture = Signal.FromTask(async cts =>
         {
             RecordStatus(statusTrail, ref position, StartedCommand);
-            await Task.Delay(CommandDelayMilliseconds, cts.Token)
+            _ = executionStarted.TrySetResult();
+            await releaseExecution.Task.WaitAsync(cts.Token)
                 .HandleCancellation(() => RecordCancellationCleanup(statusTrail, ref position)).ConfigureAwait(true);
             if (!cts.IsCancellationRequested)
             {
@@ -535,14 +540,18 @@ await Task.Delay(CommandDelayMilliseconds, cts.Token)
         {
             RecordStatus(statusTrail, ref position, ExceptionShouldBeHere);
             return Signal.Fail<RxVoid>(ex);
-        }).OnCleanup(() => RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere));
+        }).OnCleanup(() =>
+        {
+            RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere);
+            _ = finallyCompleted.TrySetResult();
+        });
         var result = false;
         var subscription = fixture.Subscribe(_ => result = true);
-        await Task.Delay(InitialDelayMilliseconds).ConfigureAwait(true);
+        await executionStarted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
         await Assert.That(StatusMessages(statusTrail)).Contains(StartedCommand);
-        await Task.Delay(CommandDelayMilliseconds).ConfigureAwait(true);
+        _ = releaseExecution.TrySetResult();
+        await finallyCompleted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
         subscription.Dispose();
-        await Task.Delay(CancellationWaitDelayMilliseconds).ConfigureAwait(false);
         await Assert.That(StatusMessages(statusTrail)).DoesNotContain(StartingCancellingCommand);
         await Assert.That(StatusMessages(statusTrail)).Contains(ShouldAlwaysComeHere);
         await Assert.That(StatusMessages(statusTrail)).DoesNotContain(FinishedCancellingCommand);
@@ -666,17 +675,25 @@ public async Task SignalFromTaskHandlesCancellationInBase()
         await Assert.That(statusTrail.LastMessage).IsEqualTo(ShouldAlwaysComeHere);
     }
 
-    /// <summary>Signals from task handles completion.</summary>
+    /// <summary>
+    /// Signals from task handles completion. The command body is released by a gate rather than a timer, so the
+    /// assertions run once the terminal cleanup has actually happened instead of once a wall-clock window is judged
+    /// long enough for it.
+    /// </summary>
     /// <returns>A <see cref = "Task"/> representing the asynchronous unit test.</returns>
     [Test]
     public async Task SignalFromTaskHandlesCompletion()
     {
         StatusTrail statusTrail = new();
+        TaskCompletionSource executionStarted = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        TaskCompletionSource releaseExecution = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        TaskCompletionSource finallyCompleted = new(TaskCreationOptions.RunContinuationsAsynchronously);
         var position = 0;
         var fixture = Signal.FromTask(async cts =>
         {
             RecordStatus(statusTrail, ref position, StartedCommand);
-            await Task.Delay(CommandDelayMilliseconds, cts.Token)
+            _ = executionStarted.TrySetResult();
+            await releaseExecution.Task.WaitAsync(cts.Token)
                 .HandleCancellation(() => RecordCancellationCleanup(statusTrail, ref position)).ConfigureAwait(true);
             if (!cts.IsCancellationRequested)
             {
@@ -688,12 +705,17 @@ await Task.Delay(CommandDelayMilliseconds, cts.Token)
         {
             RecordStatus(statusTrail, ref position, ExceptionShouldBeHere);
             return Signal.Fail<RxVoid>(ex);
-        }).OnCleanup(() => RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere));
+        }).OnCleanup(() =>
+        {
+            RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere);
+            _ = finallyCompleted.TrySetResult();
+        });
         var result = false;
         using var subscription = fixture.Subscribe(_ => result = true);
-        await Task.Delay(InitialDelayMilliseconds).ConfigureAwait(true);
+        await executionStarted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
         await Assert.That(StatusMessages(statusTrail)).Contains(StartedCommand);
-        await Task.Delay(CompletionWaitDelayMilliseconds).ConfigureAwait(false);
+        _ = releaseExecution.TrySetResult();
+        await finallyCompleted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
         await Assert.That(StatusMessages(statusTrail)).DoesNotContain(StartingCancellingCommand);
         await Assert.That(StatusMessages(statusTrail)).DoesNotContain(FinishedCancellingCommand);
         await Assert.That(StatusMessages(statusTrail)).Contains(FinishedCommandNormally);
@@ -861,17 +883,24 @@ public async Task SignalFromTask_T_HandlesCancellationInBase()
         await Assert.That(statusTrail.LastMessage).IsEqualTo(ShouldAlwaysComeHere);
     }
 
-    /// <summary>Signals from task t handles completion.</summary>
+    /// <summary>
+    /// Signals from task t handles completion. Like its non-generic counterpart the command body is released by a
+    /// gate rather than a timer, so the assertions run once the terminal cleanup has actually happened.
+    /// </summary>
     /// <returns>A <see cref = "Task"/> representing the asynchronous unit test.</returns>
     [Test]
     public async Task SignalFromTask_T_HandlesCompletion()
     {
         StatusTrail statusTrail = new();
+        TaskCompletionSource executionStarted = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        TaskCompletionSource releaseExecution = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        TaskCompletionSource finallyCompleted = new(TaskCreationOptions.RunContinuationsAsynchronously);
         var position = 0;
         var fixture = Signal.FromTask<RxVoid>(async cts =>
         {
             RecordStatus(statusTrail, ref position, StartedCommand);
-            await Task.Delay(CommandDelayMilliseconds, cts.Token)
+            _ = executionStarted.TrySetResult();
+            await releaseExecution.Task.WaitAsync(cts.Token)
                 .HandleCancellation(() => RecordCancellationCleanup(statusTrail, ref position)).ConfigureAwait(true);
             if (!cts.IsCancellationRequested)
             {
@@ -883,12 +912,17 @@ await Task.Delay(CommandDelayMilliseconds, cts.Token)
         {
             RecordStatus(statusTrail, ref position, ExceptionShouldBeHere);
             return Signal.Fail<RxVoid>(ex);
-        }).OnCleanup(() => RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere));
+        }).OnCleanup(() =>
+        {
+            RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere);
+            _ = finallyCompleted.TrySetResult();
+        });
         var result = false;
         using var subscription = fixture.Subscribe(_ => result = true);
-        await Task.Delay(InitialDelayMilliseconds).ConfigureAwait(true);
+        await executionStarted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
         await Assert.That(StatusMessages(statusTrail)).Contains(StartedCommand);
-        await Task.Delay(CompletionWaitDelayMilliseconds).ConfigureAwait(false);
+        _ = releaseExecution.TrySetResult();
+        await finallyCompleted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
         await Assert.That(StatusMessages(statusTrail)).DoesNotContain(StartingCancellingCommand);
         await Assert.That(StatusMessages(statusTrail)).DoesNotContain(FinishedCancellingCommand);
         await Assert.That(StatusMessages(statusTrail)).Contains(FinishedCommandNormally);