feat!: deliver notifications outside the lock, one sink per operator, no runtime-async

[glennawatson]

What kind of change does this PR introduce?

A feature, with breaking changes to a few advanced types.

What is the new behavior?

A subject no longer holds a lock while it calls your code. Your OnNext runs after the lock is released, so a subscriber that blocks can no longer freeze the producer.

Here is the deadlock that motivated it, from the ReactiveUI.Binding work. A property changes on a worker thread. The subscriber handles it by marshalling to the UI thread, the way Dispatcher.Invoke and Control.Invoke do. At the same moment the UI thread sets that same property:

// worker thread: holds the subject's lock, waits for the UI thread
source.Subscribe(value => dispatcher.Invoke(() => label.Text = value));

// UI thread: inside Invoke, sets the property, waits for the same lock
viewModel.Name = "Ada";

The worker holds the lock and waits for the UI thread; the UI thread waits for the lock. Every lock-based variant hung on this, and fixing only the leaf was not enough: chains through CombineLatest and Merge still deadlocked, because those coordinators also called downstream inside their own lock.

How delivery works now

A producer takes the lock only to record the value, then releases it and delivers. One thread delivers at a time. A second producer waits a short, bounded time for that delivery to finish; past the budget it hands its value to the delivering thread and returns, and the delivering thread drains the handover before it leaves. The thread that is delivering yields to a producer that is waiting, so no producer is starved.

Values still arrive one at a time and in the order they were sent. What you give up is the guarantee that a value is always delivered on the thread that raised it: serialized delivery, delivery on the raising thread, and never blocking a producer cannot all hold at once. This keeps the first two, and only gives up the third past the wait budget - which is precisely the case that used to deadlock.

Where the design comes from

• Latest-value conflation in CurrentValueDelivery<T> follows Kotlin's MutableStateFlow: hold one value slot, mark that it changed, read the current value when it is delivered, and skip a value equal to the last one. Kotlin's own reason for resuming collectors outside its lock is the same as ours, "to avoid dead-locks when using unconfined coroutines".
• The hand-off, where a producer that cannot deliver leaves its work for the thread already delivering, is the emitter-loop from RxJava's SerializedObserver and Rx.NET's AsyncLock. The FIFO queue in SerializedDelivery<T> is that shape too.
• The bounded wait before handing off is closest to Reactor's busy-looping emit handler, which spins for a fixed time before giving up. We hand the work off rather than failing the emission.
• The owner yielding to a waiting producer is not borrowed; it is what keeps a producer from being starved once waiting is bounded.

Kotlin's StateFlow never has a producer deliver to a collector at all, because each collector runs its own loop. Pushing to an IObserver<T> has no such loop, so something has to call the observer - which is what the gate decides.

What it costs and what it buys

All figures pinned, .NET 10, one subscriber.

Against the types that make the same promise - one notification at a time, from any thread - for 1000 notifications:

Subject	Mean
Signal.Serialized<T>()	5.33 us
SerializeWitness<T>	5.73 us
SynchronizeWitness<T>, lock-based	6.22 us
Subject.Synchronize, System.Reactive	7.28 us
Observable.Synchronize, System.Reactive	9.52 us

Replay, late subscriber over a 16-value buffer:

Subject	Mean	Allocated
ReplaySignal<T>	251 ns	520 B
ReplaySubject<T>, R3	280 ns	688 B
ReplaySubject<T>, System.Reactive	464 ns	696 B

Latest value, 1024 notifications, against subjects that do not serialize at all:

Subject	Mean	Allocated
BehaviorSubject<T>, System.Reactive	7.22 us	200 B
BehaviorSubject<T>, R3	8.36 us	192 B
StateSignal<T>	12.22 us	248 B

That last table is the one to read carefully, because it compares different guarantees. BehaviorSubject in both libraries lets two producers call one observer at the same time; StateSignal does not. The price of that is 6.2 ns per notification and one 80-byte allocation per subscription - two interlocked operations, with nothing allocated per notification. Every library that does offer serialization is slower than this one.

Serializing by holding a lock across the observer call is the approach that deadlocks, and it is what Observable.Synchronize and Subject.Synchronize do. This PR serializes without holding anything while your code runs.

What it buys, measured on the Binding side against the same probes that used to fail:

• No overlapping calls and no stale values in 2000 two-writer rounds. Delivering without serializing left 15 stale finals in the same probe.
• A deep path such as x => x.Child.Name delivered every final change in 5000 rounds. Unserialized delivery silently dropped 27 of them.
• A UI thread that is already delivering when a background burst starts is held for its own delivery only - 1, 10 or 50 ms for a 1, 10 or 50 ms subscriber - against 2.5 s when the hand-off had no bound.
• A producer's worst case under contention is the wait budget, 21 ms, against 101 to 116 ms for the lock under the same load.
• Binding's three property observables ported onto this API pass all 21 behaviour checks, including the ordering ones.

New things you can use

• Serialize() hands you values one at a time from a source several threads push into.

  IObservable<int> safe = noisySource.Serialize();
  safe.Subscribe(value => Console.WriteLine(value));   // never two at once

• Signal.Serialized<T>() is a subject with that same promise, for when you own the subject.

  SerializedSignal<int> subject = Signal.Serialized<int>();
  subject.Subscribe(value => Console.WriteLine(value));
  
  Parallel.For(0, 100, i => subject.OnNext(i));        // safe from many threads

• Signal.Serialized<T>(signal) wraps a subject you already have.

  SerializedSignal<int> guarded = Signal.Serialized(existingSubject);

• CurrentValueSignal<T> turns anything with a current value into an observable. You give it a way to read the value and a way to hear that it changed, and each subscriber reads the value as it subscribes. Changes raised while a delivery is running conflate to the latest value.

  var person = new Person { Name = "Ada" };
  
  IObservable<string> names = new CurrentValueSignal<string>(
      () => person.Name,
      onChanged =>
      {
          void Handler(object? sender, PropertyChangedEventArgs e) => onChanged();
          person.PropertyChanged += Handler;
          return new ActionDisposable(() => person.PropertyChanged -= Handler);
      });
  
  names.Subscribe(name => Console.WriteLine(name));    // prints Ada, then every change

  Pass an IEqualityComparer<T> as the third argument and a value equal to the last one is skipped.
• DisposableSet holds a few disposables inline, without allocating a list for the usual two or three.

  var subscriptions = new DisposableSet(first, second);
  subscriptions.Add(third);
  subscriptions.Dispose();                             // disposes all three

• SyncLatestSlot and SubscribeToSlotAsync let you write your own combine-latest style operator. You say which slot a source fills; the library tracks which sources have reported and when the group is done.

  // inside your ISyncLatestCoordinator<int>
  public ValueTask<IAsyncDisposable> SubscribeAtAsync(int index, CancellationToken token) =>
      index == 0
          ? first.SubscribeToSlotAsync(this, index, value => _first = value, token)
          : second.SubscribeToSlotAsync(this, index, value => _second = value, token);

Operator and disposable corrections

• Switch completes when an inner sequence finishes while it is still subscribing.
• The async SubscribeAsync sink calls your handler after releasing its queue lock.
• Buffer(count, skip) opens a window every skip values. A skip below the count overlaps windows, a skip above it leaves a gap, and completion flushes every window still filling.
• Timeout(DateTimeOffset) and its sequencer overload arm one window at subscription, so arriving values do not push the deadline back.
• HandleCancellation on an observable passes the token into the wait, so cancelling part-way through ends it.
• Exception.Throw and Exception.Rethrow go through ExceptionDispatchInfo on every target, keeping the stack trace from the original throw site.
• AsObservable returns a read-only view, so a caller cannot cast it back to the subject and push values in.
• Retry(n) counts total runs, matching the System.Reactive operator of that name. Reattempt(n) counts extra tries.
• Probe and Sample send a value still waiting when the source completes, ahead of the completion, as Calm and the time-based Buffer already do.
• OnErrorRetry<TException> retries only that exception type. Any other failure terminates the sequence rather than retrying forever.
• Async SwitchTo ignores a superseded inner sequence's outcome, so an inner still running when the next arrives no longer deadlocks the producer.
• SignalAsync.Use disposes its resource exactly once.
• Async Interval counts from zero, matching Every, Pulse, Timer and the synchronous operator of that name.
• Async TakeUntil(predicate) emits the element that matched before completing, matching the synchronous helper.
• Async LogErrors reports terminal failures to the logger, not only resumable errors.
• AnyAsync takes a predicate without a cancellation token, matching the other terminals.
• Awaiting an empty AsyncSignal reports the same message as ToTask, FirstAsync and LastAsync.
• A single-assignment slot runs its action on disposal whether or not a value was ever assigned. A replaceable slot disposes a value assigned after disposal without running the action a second time.

The async library gains Probe and Sample

IObservableAsync<T> carries the same sampling operator as the synchronous side, with an optional TimeProvider.

IObservableAsync<int> sampled = source.Probe(TimeSpan.FromMilliseconds(100));

The types behind the fused operators are public

Calling the operator stays the normal path. When you want the type itself, it is there to construct.

// the operator
IObservable<int> a = source.Unique();

// the type behind it
IObservable<int> b = new UniqueSignal<int>(source, EqualityComparer<int>.Default);

The signals behind Fold, Reduce, Unique, Zip, CombineLatest, Calm, Shift, Probe, Latch, KeepNotNull, KeepType, Reattempt and absolute-time Expire all live in Advanced and all take their sources through the constructor.

CreateSignal<T>, CreateSignal<T, TState>, CreateSafeSignal<T>, DeferSignal<T>, WitnessOnSignal<T>, CatchSignal<T>, CallbackSignalAsync<T>, OnErrorResumeNextSignal<T>, LatchCoordinator, CombineLatestCoordinator, ReattemptCoordinator and CalmCoordinator join them, each with the constructor and entry point a caller needs.

Every operator is one sink

An operator never builds its behaviour by chaining other operators, so a value passes through one layer per operator you wrote and no more. Where an operator has two jobs, one sink does both: the TakeUntil overloads that take a cancellation token watch their stop condition and the token together, rather than stacking a second sink on top.

SelectManyThen runs both projection stages against one active count, LastOrDefaultAsync resolves its fallback inside the task terminal, and ScheduledSignal builds its observed view once at construction.

The readme documents the whole surface

Every operator is listed with what it does, and the ones you reach for daily carry a short example. The advanced types sit at the bottom, for readers who want to build a sink rather than call an extension method. Each example compiles against the built assemblies.

The net11.0 targets build without runtime-async

Runtime-async is unsupported on Mono, and Blazor WebAssembly runs on Mono. A net11.0 package asset compiled with it fails for those consumers at their runtime, with nothing to warn them at build time.

The .NET shared framework enables the feature because it ships a separate Mono build and the publish step picks one. A NuGet package resolves a single lib/net11.0 asset for every runtime, so it has no equivalent lever. ASP.NET Core draws the same line, enabling runtime-async only for libraries that never ship as standalone packages.

What is the current behavior?

Operators, sinks and subjects call your code while holding a lock, so a subscriber that marshals to another thread can deadlock the producer. Switch swallows completion when an inner sequence completes during its own subscribe. Shared behavior is inherited from public abstract base classes.

What might this PR break?

Six public types are removed. Each was a base class you inherited from. Implement the interface instead, and hold the state struct as a field.

Removed	Use instead
WitnessAsync<T>	IWitnessAsync<T> plus a WitnessAsyncState field
SyncLatestCoordinatorBase<TResult>	ISyncLatestCoordinator<TResult> plus SyncLatestLifecycle<TResult>
TaskResultWitnessAsyncBase<T, TTaskValue>	ITaskSignalJob<T> plus TaskSignalState
ForwardingWitnessAsync<T>	IWitnessAsync<T>
CoalescingDispatchScheduler	the platform sequencer for your UI framework
IReentrantAsyncDisposable	IAsyncDisposable

// Before
public sealed class MyObserver : WitnessAsync<int>
{
    protected override ValueTask OnNextAsyncCore(int value, CancellationToken token) => default;
}

// After
public sealed class MyObserver : IWitnessAsync<int>
{
    private WitnessAsyncState _witness;

    ref WitnessAsyncState IWitnessState.Witness => ref _witness;

    ValueTask IWitnessAsync<int>.OnNextAsyncCore(int value, CancellationToken token) => default;
}

A value is no longer guaranteed to be delivered on the thread that raised it, as described above. Broadcaster<T> is unchanged.

These operators change what they emit.

Operator	Before	After
Retry(n)	runs the source n times after the first	runs the source n times in total; Retry(0) completes without running it
Probe(period), Sample(interval)	drops a value still waiting when the source completes	sends it ahead of the completion
Buffer(count, skip)	emits one window per count values	opens a window every skip values, so windows overlap or leave a gap; completion flushes every window still filling
OnErrorRetry<TException>	retries every failure	retries only TException; any other failure terminates the sequence
Schedule(dueTime, sequencer) on a plain value	shared a name with the source operator, so a concrete signal type bound it and emitted the signal object	named ScheduleValue; the source operator is what a signal type binds
async Interval(period)	counts from 1	counts from 0, matching Every, Pulse, Timer and the synchronous operator
async TakeUntil(predicate)	drops the element that matched	emits it, then completes, matching the synchronous helper
async LogErrors(logger)	logged resumable errors only	logs terminal failures too
SelectManyThen	completed once per projection stage	completes once

A call that relied on Retry(n) making n attempts beyond the first should use Reattempt(n).

The Android targets no longer generate the resource designer, so the generated Resource type leaves the Android public surface.

Checklist

• I have read the Contribute guide
• Tests have been added or updated (for bug fixes / features)
• Docs have been added or updated (for bug fixes / features)
• Changes target the main branch
• PR title follows Conventional Commits

Additional information

Most of the diff is mechanical: regenerated API baselines under src/*/PublicAPI/**, and operator conversions that all follow one pattern. The files worth reading are DeliveryGate.cs, DeliveryGateState.cs, SerializedDelivery{T}.cs, SerializedWitness{T}.cs, SerializedBroadcaster{T}.cs, CurrentValueDelivery{T}.cs, SwitchToSignal.cs, SubscribeAsyncObservable.cs and SyncLatestSlot.cs.

The Avalonia and Blazor benchmarks sit in their own assemblies, ReactiveUI.Primitives.Avalonia.Benchmarks and ReactiveUI.Primitives.Blazor.Benchmarks, because those packages start at net8.0. WPF and WinForms benchmark assemblies follow the same shape. Neither benchmark project builds a .NET Framework leg any more: the generator harness needs AppContext.GetData and a generic Enum.GetValues, and the comparison benchmarks need GeneratedRegex.

Two workflow_dispatch workflows run the benchmark projects: Benchmarks for a suite run, and Benchmarks A/B to compare two commits.

The platform dispatcher sequencers (WPF, WinForms, WinUI, MAUI, Android, Apple) have no benchmarks, because each needs its own host. That is tracked in #216.

For advanced users: building your own sink

Everything the operators are built from is public, so you can build a sink with the same guarantee instead of reaching for an extension method.

• DeliveryGate and DeliveryGateState decide who delivers. One thread delivers; another waits briefly, then hands its work over.

  private DeliveryGateState _gate;
  
  if (DeliveryGate.TryEnter(ref _gate))
  {
      observer.OnNext(value);                 // no lock is held here
      DeliveryGate.Exit(ref _gate, drain);    // delivers anything handed over meanwhile
  }

• SerializedDelivery<T> is the whole mechanism in one struct: the gate, the queue and the terminal notification. Hold it as a field and call it in place.

  private SerializedDelivery<int> _delivery;
  
  public void OnNext(int value) => _delivery.OnNext(_observer, value, new Drain(this));

• IDrainTarget is how the gate drains your queue. Its Drain() calls DrainTo with the same observer.
• SerializedWitness<T> is that struct already wrapped around one observer.

  var witness = new SerializedWitness<int>(observer);
  witness.OnNext(1);                          // safe from any thread

• SerializedBroadcaster<T> and SerializedBroadcast<T> fan one notification out to many subscribers. Post under your lock, then flush after releasing it, so every subscriber sees the same order.

  SerializedBroadcast<int> batch;
  lock (_gate)
  {
      _lastValue = value;
      batch = _broadcaster.PostNext(value);
  }
  
  batch.Flush();                              // observers run here, with no lock held

• CurrentValueDelivery<T> and ICurrentValueReader<T> deliver a value that is pulled rather than pushed. Start reads and delivers the first value inside the delivery, so a change raised during that read arrives after it, never before.
• CurrentValueWitness<T> is that struct wrapped around one observer and reader, with Start, Changed, Complete and Fault.
• WitnessSubscription subscribes an operator's witness to its source and links the two teardowns, so a single-source operator's subscribe is one call.

  ValueTask<IAsyncDisposable> IObservableAsync<T>.SubscribeAsync(
      IObserverAsync<T> observer,
      CancellationToken token) =>
      WitnessSubscription.SubscribeAsync(source, new MyWitness(observer, token), observer, token);

• TaskResultCompletionSource<T>.CompleteAndDisposeAsync publishes a terminal result, or its exception, and disposes the witness in one call.
• CoalescingDispatchState, DispatchWorkState<TState> and IDispatchHost are how the UI sequencers batch work onto a dispatcher. Implement IDispatchHost on a struct that reaches your platform's dispatcher and embed CoalescingDispatchState to write a sequencer for a toolkit this repository does not ship.
• IWitnessAsync<T>, IWitnessState and WitnessAsyncState are the async equivalents: implement the interface, hold the state struct.

[codecov]

Codecov Report

❌ Patch coverage is 99.71158% with 7 lines in your changes missing coverage. Please review.
✅ Project coverage is 99.80%. Comparing base (8aba4af) to head (fe53444).

Files with missing lines	Patch %	Lines
...Primitives.Shared/Advanced/ExpireCoordinator{T}.cs	96.15%	2 Missing ⚠️
...ons.Shared/Operators/RetryWithBackoffObservable.cs	95.65%	1 Missing ⚠️
...sions.Shared/Operators/RetryWithDelayObservable.cs	95.23%	1 Missing ⚠️
...rimitives.Shared/Advanced/AsObservableSignal{T}.cs	85.71%	1 Missing ⚠️
...rc/Primitives.Shared/Signals/Signal{GetAwaiter}.cs	93.33%	1 Missing ⚠️
...mitives.Async.Core/Advanced/LogErrorsWitness{T}.cs	87.50%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #217      +/-   ##
==========================================
+ Coverage   99.77%   99.80%   +0.03%     
==========================================
  Files         714      757      +43     
  Lines       22221    23650    +1429     
  Branches     2735     2774      +39     
==========================================
+ Hits        22171    23605    +1434     
+ Misses         49       45       -4     
+ Partials        1        0       -1     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
98.6% Coverage on New Code
0.6% Duplication on New Code

See analysis details on SonarQube Cloud