diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props
index d1615cfe..898a19f2 100644
--- a/src/Directory.Packages.props
+++ b/src/Directory.Packages.props
@@ -12,7 +12,7 @@
-
+
@@ -69,7 +69,7 @@
-
+
diff --git a/src/ReactiveUI.Primitives/Concurrency/SynchronizationContextSequencer.cs b/src/ReactiveUI.Primitives/Concurrency/SynchronizationContextSequencer.cs
index 12a06f08..2adb4901 100644
--- a/src/ReactiveUI.Primitives/Concurrency/SynchronizationContextSequencer.cs
+++ b/src/ReactiveUI.Primitives/Concurrency/SynchronizationContextSequencer.cs
@@ -9,11 +9,26 @@ namespace ReactiveUI.Primitives.Concurrency;
[System.Diagnostics.DebuggerDisplay("{DebuggerDisplay,nq}")]
public sealed class SynchronizationContextSequencer : ISequencer
{
+ /// Schedules delayed dispatch without owning the scheduler lifetime.
+ private readonly ISequencer _delaySequencer;
+
/// Initializes a new instance of the class.
/// The synchronization context used to schedule work.
/// is .
- public SynchronizationContextSequencer(SynchronizationContext context) =>
+ public SynchronizationContextSequencer(SynchronizationContext context)
+ : this(context, ThreadPoolSequencer.Instance)
+ {
+ }
+
+ /// Initializes a new instance of the class.
+ /// The synchronization context used to schedule work.
+ /// The scheduler used to wait before posting delayed work.
+ /// Either dependency is .
+ internal SynchronizationContextSequencer(SynchronizationContext context, ISequencer delaySequencer)
+ {
Context = context ?? throw new ArgumentNullException(nameof(context));
+ _delaySequencer = delaySequencer ?? throw new ArgumentNullException(nameof(delaySequencer));
+ }
/// Gets a sequencer for the current synchronization context.
/// There is no current synchronization context.
@@ -58,7 +73,7 @@ public void Schedule(IWorkItem item, long dueTimestamp)
return;
}
- ThreadPoolSequencer.Instance.Schedule(new DelayedPostWorkItem(this, item), dueTimestamp);
+ _delaySequencer.Schedule(new DelayedPostWorkItem(this, item), dueTimestamp);
}
/// Executes work when it has not already been cancelled.
diff --git a/src/tests/ReactiveUI.Primitives.Tests/ExpireCoordinatorTests.cs b/src/tests/ReactiveUI.Primitives.Tests/ExpireCoordinatorTests.cs
index f0a4d714..5585d565 100644
--- a/src/tests/ReactiveUI.Primitives.Tests/ExpireCoordinatorTests.cs
+++ b/src/tests/ReactiveUI.Primitives.Tests/ExpireCoordinatorTests.cs
@@ -2,6 +2,8 @@
// 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.Collections.Concurrent;
+using System.Runtime.CompilerServices;
using ReactiveUI.Primitives.Advanced;
using ReactiveUI.Primitives.Concurrency;
using ReactiveUI.Primitives.Signals;
@@ -30,10 +32,7 @@ public sealed class ExpireCoordinatorTests
private static readonly int[] ExpectedActiveValues = [0, 1, 2, 3, 4];
/// Timeout used while waiting for background work in this test.
- private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(5);
-
- /// How long the superseded timeout is given to reach the observer before the invariant is checked.
- private static readonly TimeSpan RaceSettleDelay = TimeSpan.FromMilliseconds(50);
+ private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(30);
/// Verifies the timeout re-arms on each value so an active source never expires.
/// A task representing the asynchronous operation.
@@ -174,29 +173,46 @@ public async Task ValueArrivingInsideTheWindowIsForwardedWhileTheTimeoutIsStillU
[Test]
public async Task TimeoutDoesNotEnterObserverWhileOnNextIsInFlight()
{
- VirtualClock clock = new(DateTimeOffset.UnixEpoch);
+ QueuedSequencer sequencer = new();
Signal source = new();
- BlockingObserver observer = new();
- using var subscription = source.Expire(TimeSpan.FromTicks(One), clock).Subscribe(observer);
+ using var observer = new BlockingObserver();
+ using var subscription = source.Expire(TimeSpan.FromTicks(One), sequencer).Subscribe(observer);
- // 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);
+ Task? timeoutFinished = null;
+ try
+ {
+ await observer.OnNextEntered.Task.WaitAsync(WaitTimeout).ConfigureAwait(false);
- var timeoutFinished = RunOnDedicatedThread(() => clock.AdvanceBy(TimeSpan.FromTicks(One)));
- await Task.Delay(RaceSettleDelay).ConfigureAwait(false);
+ // Queue the initial timeout for a dedicated worker while OnNext owns the coordinator gate. The assertion
+ // below is only about observer serialization: it must not report an error while the value callback is
+ // still active. Releasing OnNext lets the source re-arm its replacement timer and dispose this timer.
+ timeoutFinished = RunOnDedicatedThread(sequencer.ExecuteNext);
+ await sequencer.TimeoutExecutionStarted.Task.WaitAsync(WaitTimeout).ConfigureAwait(false);
- await Assert.That(observer.ErrorEnteredDuringOnNext).IsFalse();
-
- observer.ReleaseOnNext.Set();
- await onNextFinished.WaitAsync(WaitTimeout).ConfigureAwait(false);
- await timeoutFinished.WaitAsync(WaitTimeout).ConfigureAwait(false);
+ await Assert.That(observer.ErrorEnteredDuringOnNext).IsFalse();
+ }
+ finally
+ {
+ observer.ReleaseOnNext.Set();
+ if (timeoutFinished is not null)
+ {
+ await Task.WhenAll(onNextFinished, timeoutFinished).WaitAsync(WaitTimeout).ConfigureAwait(false);
+ }
+ else
+ {
+ await onNextFinished.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);
+ // The initial timer attempt must not terminate the sequence after the value wins the race. Executing the
+ // replacement proves that the successful value re-armed the inactivity timeout.
+ await Assert.That(observer.ErrorEnteredDuringOnNext).IsFalse();
+ await Assert.That(observer.Errors).IsEqualTo(0);
+ await Assert.That(observer.Values).IsEqualTo(One);
+ sequencer.ExecuteNext();
+ await Assert.That(observer.Errors).IsEqualTo(One);
+ await Assert.That(observer.TimeoutErrors).IsEqualTo(One);
await Assert.That(observer.Values).IsEqualTo(One);
}
@@ -265,14 +281,7 @@ private sealed class UndispatchedSequencer(DateTimeOffset start) : ISequencer
}
/// Observer that blocks source value handling so timeout serialization can be observed.
- [System.Diagnostics.CodeAnalysis.SuppressMessage(
- "Design",
- "SST2315:A type that owns a disposable should be disposable",
- Justification =
- "Test double that owns a ManualResetEventSlim used to gate OnNext so the test can observe timeout "
- + "serialization. Its lifetime is the test's; the test process owns and releases it, so it is deliberately "
- + "not IDisposable.")]
- private sealed class BlockingObserver : IObserver
+ private sealed class BlockingObserver : IObserver, IDisposable
{
/// Non-zero while is active. Written by the notifying thread and read by
/// the timeout thread, so the two must not race on a plain field.
@@ -288,6 +297,9 @@ private sealed class BlockingObserver : IObserver
/// The number of forwarded errors.
private int _errors;
+ /// The number of forwarded timeout errors.
+ private int _timeoutErrors;
+
/// Gets the task completed when is entered.
public TaskCompletionSource OnNextEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -300,9 +312,16 @@ private sealed class BlockingObserver : IObserver
/// Gets the number of forwarded errors.
public int Errors => Volatile.Read(ref _errors);
+ /// Gets the number of forwarded timeout errors.
+ public int TimeoutErrors => Volatile.Read(ref _timeoutErrors);
+
/// Gets a value indicating whether an error entered while was active.
public bool ErrorEnteredDuringOnNext => Volatile.Read(ref _errorEnteredDuringOnNext) != 0;
+ ///
+ [MethodImpl(MethodImplOptions.AggressiveInlining)]
+ public void Dispose() => ReleaseOnNext.Dispose();
+
///
public void OnCompleted()
{
@@ -316,6 +335,11 @@ public void OnError(Exception error)
Volatile.Write(ref _errorEnteredDuringOnNext, 1);
}
+ if (error is TimeoutException)
+ {
+ _ = Interlocked.Increment(ref _timeoutErrors);
+ }
+
_ = Interlocked.Increment(ref _errors);
}
@@ -324,9 +348,75 @@ public void OnNext(int value)
{
_ = Interlocked.Increment(ref _values);
Volatile.Write(ref _isInOnNext, 1);
- OnNextEntered.SetResult();
- _ = ReleaseOnNext.Wait(WaitTimeout);
- Volatile.Write(ref _isInOnNext, 0);
+ _ = OnNextEntered.TrySetResult();
+ try
+ {
+ if (!ReleaseOnNext.Wait(WaitTimeout))
+ {
+ throw new TimeoutException("The test did not release the in-flight OnNext callback.");
+ }
+ }
+ finally
+ {
+ Volatile.Write(ref _isInOnNext, 0);
+ }
+ }
+ }
+
+ /// Sequencer that queues work until this test explicitly executes it.
+ private sealed class QueuedSequencer : ISequencer
+ {
+ /// Scheduled work waiting for the test to execute it.
+ private readonly ConcurrentQueue<(IWorkItem Item, long DueTimestamp)> _items = new();
+
+ /// The current monotonic timestamp, updated before a queued item is invoked.
+ private long _timestamp;
+
+ /// Gets the task completed when a worker dequeues a queued timeout for execution.
+ public TaskCompletionSource TimeoutExecutionStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ ///
+ public DateTimeOffset Now => DateTimeOffset.UnixEpoch + Sequencer.ToTimeSpanDelta(Timestamp);
+
+ ///
+ public long Timestamp => Volatile.Read(ref _timestamp);
+
+ ///
+ [MethodImpl(MethodImplOptions.AggressiveInlining)]
+ public void Schedule(IWorkItem item) => _items.Enqueue((item, Timestamp));
+
+ ///
+ [MethodImpl(MethodImplOptions.AggressiveInlining)]
+ public void Schedule(IWorkItem item, long dueTimestamp) => _items.Enqueue((item, dueTimestamp));
+
+ /// Executes the next queued work item.
+ /// No timer was queued when execution was requested.
+ public void ExecuteNext()
+ {
+ if (!_items.TryDequeue(out var scheduled))
+ {
+ throw new InvalidOperationException("No queued timeout was available to execute.");
+ }
+
+ AdvanceTo(scheduled.DueTimestamp);
+ _ = TimeoutExecutionStarted.TrySetResult();
+ scheduled.Item.Execute();
+ }
+
+ /// Advances the clock to a scheduled due timestamp without moving it backwards.
+ /// The timestamp of the work about to execute.
+ private void AdvanceTo(long dueTimestamp)
+ {
+ long currentTimestamp;
+ do
+ {
+ currentTimestamp = Timestamp;
+ if (currentTimestamp >= dueTimestamp)
+ {
+ return;
+ }
+ }
+ while (Interlocked.CompareExchange(ref _timestamp, dueTimestamp, currentTimestamp) != currentTimestamp);
}
}
}
diff --git a/src/tests/ReactiveUI.Primitives.Tests/SequencerTests.Pools.cs b/src/tests/ReactiveUI.Primitives.Tests/SequencerTests.Pools.cs
index e1676ff6..0b74ebdf 100644
--- a/src/tests/ReactiveUI.Primitives.Tests/SequencerTests.Pools.cs
+++ b/src/tests/ReactiveUI.Primitives.Tests/SequencerTests.Pools.cs
@@ -160,24 +160,55 @@ public async Task SynchronizationContextSequencerPostsDelayedWorkOnceItIsDue()
await Assert.That(context.PostCount).IsEqualTo(1);
}
- /// Verifies delayed work cancelled before its due time never reaches the synchronization context.
+ /// Verifies cancellation after enqueue prevents delayed work from reaching the dispatcher.
/// A task representing the asynchronous operation.
[Test]
public async Task SynchronizationContextSequencerDropsDelayedWorkCancelledBeforeItIsDue()
{
RecordingSynchronizationContext context = new();
- SynchronizationContextSequencer sequencer = new(context);
+ ControlledDelaySequencer delay = new();
+ SynchronizationContextSequencer sequencer = new(context, delay);
CancellableWorkItem item = new();
- sequencer.Schedule(item, Sequencer.AddTimestamp(sequencer.Timestamp, CancelledDueTime));
- item.Dispose();
+ sequencer.Schedule(item, long.MaxValue);
+ await Assert.That(delay.ScheduledTimestamp).IsEqualTo(long.MaxValue);
+ await Assert.That(delay.Pending).IsNotNull();
+ await Assert.That(context.PostCount).IsEqualTo(0);
- await Task.Delay(CancelObservationWindow);
+ item.Dispose();
+ delay.ExecutePending();
await Assert.That(item.ExecuteCount).IsEqualTo(0);
await Assert.That(context.PostCount).IsEqualTo(0);
}
+ /// Verifies pending delayed work reaches the dispatcher when it has not been cancelled.
+ /// A task representing the asynchronous operation.
+ [Test]
+ public async Task SynchronizationContextSequencerPostsPendingDelayedWork()
+ {
+ RecordingSynchronizationContext context = new();
+ ControlledDelaySequencer delay = new();
+ SynchronizationContextSequencer sequencer = new(context, delay);
+ CancellableWorkItem item = new();
+
+ sequencer.Schedule(item, long.MaxValue);
+ await Assert.That(context.PostCount).IsEqualTo(0);
+ await Assert.That(item.ExecuteCount).IsEqualTo(0);
+
+ delay.ExecutePending();
+
+ await Assert.That(context.PostCount).IsEqualTo(1);
+ await Assert.That(item.ExecuteCount).IsEqualTo(1);
+ }
+
+ /// Verifies the internal delayed scheduler dependency cannot be absent.
+ /// A task representing the asynchronous operation.
+ [Test]
+ public async Task SynchronizationContextSequencerRejectsMissingDelayScheduler() =>
+ await Assert.That(static () => new SynchronizationContextSequencer(new RecordingSynchronizationContext(), null!))
+ .ThrowsExactly();
+
///
/// Verifies disposing a thread-pool sequencer twice releases its queued work exactly once and leaves the sequencer
/// closed. The second disposal must be a no-op rather than a second release of work the first disposal already
@@ -248,6 +279,42 @@ public async Task ThreadPoolSequencerDisposeDuringADrainStopsTheDrainRearmingThe
/// The isolated sequencer.
private static ThreadPoolSequencer CreateIsolatedThreadPoolSequencer() => new();
+ /// Holds a delayed item until the test explicitly dispatches it.
+ private sealed class ControlledDelaySequencer : ISequencer
+ {
+ /// Gets the pending delayed callback.
+ public IWorkItem? Pending { get; private set; }
+
+ /// Gets the timestamp forwarded by the dispatcher.
+ public long ScheduledTimestamp { get; private set; }
+
+ ///
+ public DateTimeOffset Now => DateTimeOffset.UnixEpoch;
+
+ ///
+ public long Timestamp => 0;
+
+ ///
+ [MethodImpl(MethodImplOptions.AggressiveInlining)]
+ public void Schedule(IWorkItem item) => item.Execute();
+
+ ///
+ public void Schedule(IWorkItem item, long dueTimestamp)
+ {
+ Pending = item;
+ ScheduledTimestamp = dueTimestamp;
+ }
+
+ /// Executes the item after the test has cancelled or inspected pending work.
+ /// No item is pending.
+ public void ExecutePending()
+ {
+ var pending = Pending ?? throw new InvalidOperationException("No delayed work was queued.");
+ Pending = null;
+ pending.Execute();
+ }
+ }
+
/// Work item that counts how many times a sequencer released it.
private sealed class DisposeCountingWorkItem : IWorkItem, IsDisposed
{
diff --git a/src/tests/ReactiveUI.Primitives.Tests/SignalOperatorMixinsTests.Deterministic.cs b/src/tests/ReactiveUI.Primitives.Tests/SignalOperatorMixinsTests.Deterministic.cs
index c523fbfd..d50f8f8b 100644
--- a/src/tests/ReactiveUI.Primitives.Tests/SignalOperatorMixinsTests.Deterministic.cs
+++ b/src/tests/ReactiveUI.Primitives.Tests/SignalOperatorMixinsTests.Deterministic.cs
@@ -48,6 +48,9 @@ public partial class SignalOperatorMixinsTests
/// The number of threads that rendezvous before the disposal race starts.
private const int RacingThreadCount = 2;
+ /// The completion guard for asynchronously scheduled enumeration on instrumented CI hosts.
+ private static readonly TimeSpan AsyncEnumerationCompletionTimeout = TimeSpan.FromSeconds(30);
+
/// A fixed deterministic timestamp used in place of the current time.
private static readonly DateTimeOffset FixedTimestamp = new(2024, 1, 1, 0, 0, 0, TimeSpan.Zero);
@@ -469,11 +472,11 @@ private static async Task VerifyAsyncEnumerableShiftAndExpireAsync()
List asyncValues = [];
TaskCompletionSource asyncCompleted = new(TaskCreationOptions.RunContinuationsAsynchronously);
using CancellationTokenSource asyncToken = new();
- _ = Signal.FromAsyncEnumerable(AsyncValues(Three), asyncToken.Token).Subscribe(
+ using var asyncSubscription = Signal.FromAsyncEnumerable(AsyncValues(Three), asyncToken.Token).Subscribe(
asyncValues.Add,
ex => asyncCompleted.TrySetException(ex),
() => asyncCompleted.TrySetResult(null));
- await asyncCompleted.Task.WaitAsync(TimeSpan.FromSeconds(Five)).ConfigureAwait(false);
+ await asyncCompleted.Task.WaitAsync(AsyncEnumerationCompletionTimeout).ConfigureAwait(false);
int[] expectedAsyncValues = [0, One, Two];
await Assert.That(asyncValues.SequenceEqual(expectedAsyncValues)).IsTrue();
var exact = await Signal.FromAsyncEnumerable(AsyncValues(Sixteen)).CollectArrayAsync().ConfigureAwait(false);
diff --git a/src/tests/ReactiveUI.Primitives.Tests/WasmSequencerTests.cs b/src/tests/ReactiveUI.Primitives.Tests/WasmSequencerTests.cs
index bd73cb81..1148d50f 100644
--- a/src/tests/ReactiveUI.Primitives.Tests/WasmSequencerTests.cs
+++ b/src/tests/ReactiveUI.Primitives.Tests/WasmSequencerTests.cs
@@ -14,8 +14,8 @@ public sealed class WasmSequencerTests
/// Expected values produced by an immediate burst, used to verify FIFO order.
private static readonly int[] ExpectedBurst = [1, 2, 3];
- /// Longest a test waits for scheduled work before failing.
- private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(5);
+ /// Completion guard allowing for shared timer and thread-pool contention on instrumented runners.
+ private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(30);
/// How far in the future delayed work is scheduled.
private static readonly TimeSpan ScheduleDelay = TimeSpan.FromMilliseconds(50);
@@ -104,13 +104,12 @@ public async Task DelayedScheduleExecutesAfterDue()
{
var sequencer = WasmSequencer.Default;
TaskCompletionSource executed = new(TaskCreationOptions.RunContinuationsAsynchronously);
- var start = sequencer.Timestamp;
- var due = Sequencer.AddTimestamp(start, ScheduleDelay);
+ var due = Sequencer.AddTimestamp(sequencer.Timestamp, ScheduleDelay);
sequencer.Schedule(new DelegateWorkItem(() => executed.TrySetResult(sequencer.Timestamp)), due);
var executedAt = await executed.Task.WaitAsync(WaitTimeout);
- await Assert.That(executedAt).IsGreaterThanOrEqualTo(start);
+ await Assert.That(executedAt).IsGreaterThanOrEqualTo(due);
}
/// Verifies a past-due timestamp executes promptly through the immediate path.