test: expand coverage for fused async helpers, subject base, partition, and sync operator branches

.github/workflows/sonarcloud.yml

@@ -19,7 +19,7 @@ jobs:
             minverMinimumMajorMinor: '2.3'
             sonarProjectKey: reactiveui_Extensions
             sonarOrganization: reactiveui
-            sonarExclusions: '**/tests/**,**/tools/**,**/benchmarks/**'
+            sonarExclusions: '**/tests/**,**/tools/**,**/benchmarks/**,**/TestResults/**'
             sonarCoverageExclusions: '**/tests/**,**/tools/**,**/benchmarks/**,**/*Tests/**,**/*Tests.cs,**/Generated/**'
             # CombineLatest{N}.cs are template-style arity-specific operator files. The remaining
             # duplication after the lifecycle / indexed-observer / base-class extractions is the

src/ReactiveUI.Extensions/Async/ObserverAsync.cs

@@ -157,18 +157,11 @@ public ValueTask OnErrorResumeAsync(Exception error, CancellationToken cancellat
             return default;
         }
 
-        ValueTask core;
-        try
-        {
-            core = OnErrorResumeAsync_Private(error, scope.Token);
-        }
-        catch (Exception e)
-        {
-            UnhandledExceptionHandler.OnUnhandledException(e);
-            scope.Dispose();
-            ExitOnSomethingCall();
-            return default;
-        }
+        // OnErrorResumeAsync_Private is an async ValueTask method — any sync or async exception
+        // it raises is captured into the returned ValueTask and surfaces through the await in
+        // OnErrorResumeAsyncSlow. A try/catch around the invocation expression itself would be
+        // dead code in modern C# async semantics.
+        var core = OnErrorResumeAsync_Private(error, scope.Token);
 
         if (core.IsCompletedSuccessfully)
         {

src/ReactiveUI.Extensions/Async/Operators/Merge.cs

@@ -228,6 +228,40 @@ internal void LinkExternalCancellation(CancellationToken external)
                 _disposeCts);
         }
 
+        /// <summary>
+        /// Re-checks the disposed flag inside the serialization gate and forwards the value to
+        /// downstream if still alive. Extracted as an <see langword="internal"/> method so the
+        /// inside-gate after-dispose decision is directly unit-testable without racing the gate.
+        /// </summary>
+        /// <param name="value">The value to forward.</param>
+        /// <returns>A task representing the asynchronous forward operation.</returns>
+        internal ValueTask ForwardOnNextLocked(T value)
+        {
+            if (DisposalHelper.IsDisposed(_disposed))
+            {
+                return default;
+            }
+
+            return _observer.OnNextAsync(value, DisposedCancellationToken);
+        }
+
+        /// <summary>
+        /// Re-checks the disposed flag inside the serialization gate and forwards the error to
+        /// downstream if still alive. Extracted as an <see langword="internal"/> method for
+        /// direct unit testing.
+        /// </summary>
+        /// <param name="exception">The error to forward.</param>
+        /// <returns>A task representing the asynchronous forward operation.</returns>
+        internal ValueTask ForwardOnErrorResumeLocked(Exception exception)
+        {
+            if (DisposalHelper.IsDisposed(_disposed))
+            {
+                return default;
+            }
+
+            return _observer.OnErrorResumeAsync(exception, DisposedCancellationToken);
+        }
+
         /// <summary>
         /// Forwards a value to the downstream observer under the serialization gate.
         /// </summary>
@@ -244,12 +278,7 @@ protected internal async ValueTask ForwardOnNext(T value, CancellationToken canc
 
             using (await _onSomethingGate.LockAsync(DisposedCancellationToken).ConfigureAwait(false))
             {
-                if (DisposalHelper.IsDisposed(_disposed))
-                {
-                    return;
-                }
-
-                await _observer.OnNextAsync(value, DisposedCancellationToken).ConfigureAwait(false);
+                await ForwardOnNextLocked(value).ConfigureAwait(false);
             }
         }
 
@@ -271,12 +300,7 @@ protected internal async ValueTask ForwardOnErrorResume(
 
             using (await _onSomethingGate.LockAsync(DisposedCancellationToken).ConfigureAwait(false))
             {
-                if (DisposalHelper.IsDisposed(_disposed))
-                {
-                    return;
-                }
-
-                await _observer.OnErrorResumeAsync(exception, DisposedCancellationToken).ConfigureAwait(false);
+                await ForwardOnErrorResumeLocked(exception).ConfigureAwait(false);
             }
         }
 
@@ -649,13 +673,25 @@ internal async ValueTask OnNextAsync(T value, CancellationToken token)
 
                 using (await _onSomethingGate.LockAsync(_disposedCancellationToken).ConfigureAwait(false))
                 {
-                    if (DisposalHelper.IsDisposed(_disposed))
-                    {
-                        return;
-                    }
+                    await OnNextAsyncLocked(value).ConfigureAwait(false);
+                }
+            }
 
-                    await _observer.OnNextAsync(value, _disposedCancellationToken).ConfigureAwait(false);
+            /// <summary>
+            /// Re-checks the disposed flag inside the serialization gate and forwards the value
+            /// to downstream if still alive. Extracted as an <see langword="internal"/> method for
+            /// direct unit testing of the inside-gate after-dispose decision.
+            /// </summary>
+            /// <param name="value">The value to forward.</param>
+            /// <returns>A task representing the asynchronous forward operation.</returns>
+            internal ValueTask OnNextAsyncLocked(T value)
+            {
+                if (DisposalHelper.IsDisposed(_disposed))
+                {
+                    return default;
                 }
+
+                return _observer.OnNextAsync(value, _disposedCancellationToken);
             }
 
             /// <summary>
@@ -674,13 +710,25 @@ internal async ValueTask OnErrorResumeAsync(Exception ex, CancellationToken toke
 
                 using (await _onSomethingGate.LockAsync(_disposedCancellationToken).ConfigureAwait(false))
                 {
-                    if (DisposalHelper.IsDisposed(_disposed))
-                    {
-                        return;
-                    }
+                    await OnErrorResumeAsyncLocked(ex).ConfigureAwait(false);
+                }
+            }
 
-                    await _observer.OnErrorResumeAsync(ex, _disposedCancellationToken).ConfigureAwait(false);
+            /// <summary>
+            /// Re-checks the disposed flag inside the serialization gate and forwards the error
+            /// to downstream if still alive. Extracted as an <see langword="internal"/> method
+            /// for direct unit testing of the inside-gate after-dispose decision.
+            /// </summary>
+            /// <param name="ex">The error to forward.</param>
+            /// <returns>A task representing the asynchronous forward operation.</returns>
+            internal ValueTask OnErrorResumeAsyncLocked(Exception ex)
+            {
+                if (DisposalHelper.IsDisposed(_disposed))
+                {
+                    return default;
                 }
+
+                return _observer.OnErrorResumeAsync(ex, _disposedCancellationToken);
             }
 
             /// <summary>

src/ReactiveUI.Extensions/Async/Operators/ObserveOnAsyncObservable.cs

@@ -34,6 +34,39 @@ protected override ValueTask<IAsyncDisposable> SubscribeAsyncCore(
     internal sealed class ObserveOnObserver(IObserverAsync<T> observer, AsyncContext asyncContext, bool forceYielding)
         : ObserverAsync<T>
     {
+        /// <summary>Slow path: switch to the target context then forward the value.
+        /// Exposed as <see langword="internal"/> so tests can invoke the slow-path body
+        /// directly without needing to race the current-context check.</summary>
+        /// <param name="value">The value to forward.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A task that completes after the context switch and downstream forward.</returns>
+        internal async ValueTask SwitchThenForwardAsync(T value, CancellationToken cancellationToken)
+        {
+            await asyncContext.SwitchContextAsync(forceYielding, cancellationToken);
+            await observer.OnNextAsync(value, cancellationToken).ConfigureAwait(false);
+        }
+
+        /// <summary>Slow path: switch to the target context then forward the error.
+        /// Exposed as <see langword="internal"/> for direct unit testing.</summary>
+        /// <param name="error">The error to forward.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A task that completes after the context switch and downstream forward.</returns>
+        internal async ValueTask SwitchThenErrorAsync(Exception error, CancellationToken cancellationToken)
+        {
+            await asyncContext.SwitchContextAsync(forceYielding, cancellationToken);
+            await observer.OnErrorResumeAsync(error, cancellationToken).ConfigureAwait(false);
+        }
+
+        /// <summary>Slow path: switch to the target context then forward completion.
+        /// Exposed as <see langword="internal"/> for direct unit testing.</summary>
+        /// <param name="result">The completion result.</param>
+        /// <returns>A task that completes after the context switch and downstream forward.</returns>
+        internal async ValueTask SwitchThenCompletedAsync(Result result)
+        {
+            await asyncContext.SwitchContextAsync(forceYielding, CancellationToken.None);
+            await observer.OnCompletedAsync(result).ConfigureAwait(false);
+        }
+
         /// <inheritdoc/>
         protected override ValueTask OnNextAsyncCore(T value, CancellationToken cancellationToken)
         {
@@ -68,34 +101,5 @@ protected override ValueTask OnCompletedAsyncCore(Result result)
 
             return SwitchThenCompletedAsync(result);
         }
-
-        /// <summary>Slow path: switch to the target context then forward the value.</summary>
-        /// <param name="value">The value to forward.</param>
-        /// <param name="cancellationToken">The cancellation token.</param>
-        /// <returns>A task that completes after the context switch and downstream forward.</returns>
-        private async ValueTask SwitchThenForwardAsync(T value, CancellationToken cancellationToken)
-        {
-            await asyncContext.SwitchContextAsync(forceYielding, cancellationToken);
-            await observer.OnNextAsync(value, cancellationToken).ConfigureAwait(false);
-        }
-
-        /// <summary>Slow path: switch to the target context then forward the error.</summary>
-        /// <param name="error">The error to forward.</param>
-        /// <param name="cancellationToken">The cancellation token.</param>
-        /// <returns>A task that completes after the context switch and downstream forward.</returns>
-        private async ValueTask SwitchThenErrorAsync(Exception error, CancellationToken cancellationToken)
-        {
-            await asyncContext.SwitchContextAsync(forceYielding, cancellationToken);
-            await observer.OnErrorResumeAsync(error, cancellationToken).ConfigureAwait(false);
-        }
-
-        /// <summary>Slow path: switch to the target context then forward completion.</summary>
-        /// <param name="result">The completion result.</param>
-        /// <returns>A task that completes after the context switch and downstream forward.</returns>
-        private async ValueTask SwitchThenCompletedAsync(Result result)
-        {
-            await asyncContext.SwitchContextAsync(forceYielding, CancellationToken.None);
-            await observer.OnCompletedAsync(result).ConfigureAwait(false);
-        }
     }
 }