diff --git a/TUnit.AspNetCore.Core/FlowSuppressingHostedService.cs b/TUnit.AspNetCore.Core/FlowSuppressingHostedService.cs index 0f8f05a9ce..32ffc8867b 100644 --- a/TUnit.AspNetCore.Core/FlowSuppressingHostedService.cs +++ b/TUnit.AspNetCore.Core/FlowSuppressingHostedService.cs @@ -23,14 +23,31 @@ namespace TUnit.AspNetCore; /// Without this, wrapped services that own unmanaged resources leak silently because /// the container only sees the non-disposable wrapper. /// +/// +/// The stop-phase methods are additionally guarded to run the inner service's stop +/// exactly once. Minimal-hosting SUTs park their own app.Run() in +/// WaitForShutdownAsync, which calls Host.StopAsync when +/// ApplicationStopping fires — concurrently with the Host.StopAsync that +/// WebApplicationFactory.DisposeAsync already has in flight (the trigger of that +/// very ApplicationStopping). Host.StopAsync has no concurrency guard, so +/// every hosted service's StopAsync runs twice in parallel, breaking services +/// with non-thread-safe shutdown (e.g. Rebus' bus dispose throws +/// ). See https://github.com/thomhurst/TUnit/issues/6339. +/// All concurrent and repeated callers observe the single underlying stop task. +/// /// internal sealed class FlowSuppressingHostedService(IHostedService inner) : IHostedLifecycleService, IAsyncDisposable, IDisposable { + private readonly object _stopGate = new(); + private Task? _stopTask; + private Task? _stoppingTask; + private Task? _stoppedTask; + public Task StartAsync(CancellationToken cancellationToken) => RunOnCleanContext(inner.StartAsync, cancellationToken); public Task StopAsync(CancellationToken cancellationToken) => - inner.StopAsync(cancellationToken); + InvokeOnce(ref _stopTask, inner.StopAsync, cancellationToken); public Task StartingAsync(CancellationToken cancellationToken) => inner is IHostedLifecycleService lifecycle @@ -42,18 +59,30 @@ inner is IHostedLifecycleService lifecycle ? RunOnCleanContext(lifecycle.StartedAsync, cancellationToken) : Task.CompletedTask; - // Stop lifecycle is intentionally not wrapped: stop methods typically signal - // cancellation and await shutdown rather than spawning new long-running background - // work, so context capture during Stop is not the span-leak vector that Start is. - public Task StoppingAsync(CancellationToken cancellationToken) => - inner is IHostedLifecycleService lifecycle - ? lifecycle.StoppingAsync(cancellationToken) - : Task.CompletedTask; + // Stop lifecycle is not flow-suppressed: stop methods typically signal cancellation + // and await shutdown rather than spawning new long-running background work, so + // context capture during Stop is not the span-leak vector that Start is. They are + // once-guarded, though — concurrent Host.StopAsync calls invoke these hooks twice + // (see the class remarks). + public Task StoppingAsync(CancellationToken cancellationToken) + { + if (inner is not IHostedLifecycleService lifecycle) + { + return Task.CompletedTask; + } - public Task StoppedAsync(CancellationToken cancellationToken) => - inner is IHostedLifecycleService lifecycle - ? lifecycle.StoppedAsync(cancellationToken) - : Task.CompletedTask; + return InvokeOnce(ref _stoppingTask, lifecycle.StoppingAsync, cancellationToken); + } + + public Task StoppedAsync(CancellationToken cancellationToken) + { + if (inner is not IHostedLifecycleService lifecycle) + { + return Task.CompletedTask; + } + + return InvokeOnce(ref _stoppedTask, lifecycle.StoppedAsync, cancellationToken); + } /// public async ValueTask DisposeAsync() @@ -83,6 +112,61 @@ public void Dispose() } } + // Publish a placeholder before invoking the operation outside the lock. Duplicate callers + // can immediately wait with their own cancellation token, even when the operation blocks + // synchronously before returning its task. + private Task InvokeOnce( + ref Task? cachedTask, + Func operation, + CancellationToken cancellationToken) + { + TaskCompletionSource? completionSource = null; + Task task; + var ownsInvocation = false; + + lock (_stopGate) + { + if (cachedTask is null) + { + completionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + cachedTask = completionSource.Task; + ownsInvocation = true; + } + + task = cachedTask; + } + + if (completionSource is not null) + { + _ = ExecuteAndCompleteAsync(completionSource, operation, cancellationToken); + } + + return ownsInvocation ? task : WaitWithCancellation(task, cancellationToken); + } + + private static async Task ExecuteAndCompleteAsync( + TaskCompletionSource completionSource, + Func operation, + CancellationToken cancellationToken) + { + try + { + await operation(cancellationToken).ConfigureAwait(false); + completionSource.SetResult(); + } + catch (OperationCanceledException exception) + { + completionSource.SetCanceled(exception.CancellationToken); + } + catch (Exception exception) + { + completionSource.SetException(exception); + } + } + + private static Task WaitWithCancellation(Task task, CancellationToken cancellationToken) => + cancellationToken.CanBeCanceled ? task.WaitAsync(cancellationToken) : task; + // Dispatch onto a thread-pool worker with a clean captured ExecutionContext by // combining SuppressFlow + Task.Run. Unlike wrapping `using (SuppressFlow()) return op(ct);` // which only suppresses during the synchronous body, this keeps the inner operation diff --git a/TUnit.AspNetCore.Tests/HostStopExactlyOnceProbeTests.cs b/TUnit.AspNetCore.Tests/HostStopExactlyOnceProbeTests.cs new file mode 100644 index 0000000000..21ca344a95 --- /dev/null +++ b/TUnit.AspNetCore.Tests/HostStopExactlyOnceProbeTests.cs @@ -0,0 +1,72 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using TUnit.AspNetCore; + +namespace TUnit.AspNetCore.Tests; + +/// +/// End-to-end regression test for https://github.com/thomhurst/TUnit/issues/6339. +/// +/// With a minimal-hosting SUT, disposing the per-test factory races the SUT's own +/// app.Run() shutdown path: WebApplicationFactory.DisposeAsync calls +/// Host.StopAsync, whose ApplicationStopping signal wakes the app's parked +/// WaitForShutdownAsync, which calls Host.StopAsync again — concurrently. +/// Before the stop-once guard in , each hosted +/// service's StopAsync ran twice in parallel (observed here as 4–8 failures per +/// 50 tests), which is how Rebus' non-thread-safe bus dispose produced the reporter's +/// . +/// +/// +/// The probe throws — with both stack traces — if it ever observes a second +/// StopAsync entry, so a regression surfaces as loud test failures. +/// +/// +public class HostStopExactlyOnceProbeTests : WebApplicationTest +{ + protected override void ConfigureTestServices(IServiceCollection services) + { + services.AddSingleton(); + } + + [Test] + [Repeat(50)] + public async Task Host_Hosted_Service_Stops_Exactly_Once_Under_Parallel_Churn() + { + var client = Factory.CreateClient(); + using var response = await client.GetAsync("/"); + + // Small jitter so test completions (and After-hook disposals) overlap heavily. + await Task.Delay(Random.Shared.Next(0, 20)); + } +} + +/// +/// Throws on a second StopAsync entry — sequential re-stop or concurrent overlap — so the +/// offending call site's stack appears in the test output alongside the first stop's stack. +/// +internal sealed class StopProbeHostedService : IHostedService +{ + private int _stopEntries; + private volatile string? _firstStopStack; + + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + public async Task StopAsync(CancellationToken cancellationToken) + { + var entry = Interlocked.Increment(ref _stopEntries); + + if (entry > 1) + { + throw new InvalidOperationException( + $"StopAsync entered {entry} times on probe {GetHashCode():x8}.\n" + + $"--- first stop stack ---\n{_firstStopStack}\n" + + $"--- this stop stack ---\n{Environment.StackTrace}"); + } + + _firstStopStack = Environment.StackTrace; + + // Widen the window like a real bus shutdown (Rebus stops workers, waits on its + // cleanup task for up to 5s). A concurrent second stop lands inside this delay. + await Task.Delay(100, CancellationToken.None); + } +} diff --git a/TUnit.AspNetCore.Tests/HostedServiceStopOnceTests.cs b/TUnit.AspNetCore.Tests/HostedServiceStopOnceTests.cs new file mode 100644 index 0000000000..d95f7212f9 --- /dev/null +++ b/TUnit.AspNetCore.Tests/HostedServiceStopOnceTests.cs @@ -0,0 +1,275 @@ +using Microsoft.Extensions.Hosting; +using TUnit.AspNetCore; + +namespace TUnit.AspNetCore.Tests; + +/// +/// Regression tests for https://github.com/thomhurst/TUnit/issues/6339. +/// +/// Minimal-hosting SUTs park app.Run() in WaitForShutdownAsync, which calls +/// Host.StopAsync when ApplicationStopping fires — concurrently with the +/// Host.StopAsync from WebApplicationFactory.DisposeAsync that triggered it. +/// Every hosted service's stop then runs twice in parallel, breaking services with +/// non-thread-safe shutdown (Rebus' bus dispose throws ). +/// must absorb the duplicate calls. +/// +/// +public class HostedServiceStopOnceTests +{ + [Test] + public async Task Concurrent_StopAsync_Invokes_Inner_Once() + { + var inner = new CountingHostedService(); + var wrapper = new FlowSuppressingHostedService(inner); + + var gate = new TaskCompletionSource(); + var callers = Enumerable.Range(0, 16) + .Select(_ => Task.Run(async () => + { + await gate.Task; + await wrapper.StopAsync(CancellationToken.None); + })) + .ToArray(); + gate.SetResult(); + await Task.WhenAll(callers); + + await Assert.That(inner.StopCalls).IsEqualTo(1); + } + + [Test] + public async Task Sequential_StopAsync_Invokes_Inner_Once() + { + var inner = new CountingHostedService(); + var wrapper = new FlowSuppressingHostedService(inner); + + await wrapper.StopAsync(CancellationToken.None); + await wrapper.StopAsync(CancellationToken.None); + + await Assert.That(inner.StopCalls).IsEqualTo(1); + } + + [Test] + public async Task Duplicate_Stop_Callers_Observe_The_Same_Task() + { + var inner = new CountingHostedService(); + var wrapper = new FlowSuppressingHostedService(inner); + + var first = wrapper.StopAsync(CancellationToken.None); + var second = wrapper.StopAsync(CancellationToken.None); + + await Assert.That(ReferenceEquals(first, second)).IsTrue(); + } + + [Test] + public async Task Duplicate_Stop_Caller_Observes_Its_Own_Cancellation_Token() + { + var inner = new BlockingStopHostedService(); + var wrapper = new FlowSuppressingHostedService(inner); + + var first = wrapper.StopAsync(CancellationToken.None); + await inner.Entered; + + using var cancellationTokenSource = new CancellationTokenSource(); + var second = wrapper.StopAsync(cancellationTokenSource.Token); + cancellationTokenSource.Cancel(); + + await Assert.That(async () => await second).Throws(); + await Assert.That(inner.StopCalls).IsEqualTo(1); + + inner.Release(); + await first; + } + + [Test] + public async Task Owning_Stop_Caller_Awaits_Inner_After_Token_Cancellation() + { + var inner = new BlockingStopHostedService(); + var wrapper = new FlowSuppressingHostedService(inner); + + using var cancellationTokenSource = new CancellationTokenSource(); + var stop = wrapper.StopAsync(cancellationTokenSource.Token); + await inner.Entered; + + try + { + cancellationTokenSource.Cancel(); + await Assert.That(stop.IsCompleted).IsFalse(); + } + finally + { + inner.Release(); + } + + await stop; + } + + [Test] + public async Task Duplicate_Stop_Caller_Does_Not_Wait_For_Synchronous_Startup() + { + using var inner = new SynchronouslyBlockingStopHostedService(); + var wrapper = new FlowSuppressingHostedService(inner); + + var first = Task.Run(() => wrapper.StopAsync(CancellationToken.None)); + inner.WaitUntilEntered(); + + using var cancellationTokenSource = new CancellationTokenSource(); + cancellationTokenSource.Cancel(); + + try + { + var second = Task.Run(() => wrapper.StopAsync(cancellationTokenSource.Token)); + await Assert.That(async () => await second.WaitAsync(TimeSpan.FromSeconds(1))) + .Throws(); + } + finally + { + inner.Release(); + } + + await first; + await Assert.That(inner.StopCalls).IsEqualTo(1); + } + + [Test] + public async Task Synchronously_Throwing_Stop_Is_Cached_Not_Reinvoked() + { + var inner = new SyncThrowingStopHostedService(); + var wrapper = new FlowSuppressingHostedService(inner); + + await Assert.That(async () => await wrapper.StopAsync(CancellationToken.None)) + .Throws(); + await Assert.That(async () => await wrapper.StopAsync(CancellationToken.None)) + .Throws(); + + await Assert.That(inner.StopCalls).IsEqualTo(1); + } + + [Test] + public async Task Lifecycle_Stopping_And_Stopped_Are_Once_Guarded() + { + var inner = new CountingLifecycleHostedService(); + var wrapper = new FlowSuppressingHostedService(inner); + + await wrapper.StoppingAsync(CancellationToken.None); + await wrapper.StoppingAsync(CancellationToken.None); + await wrapper.StoppedAsync(CancellationToken.None); + await wrapper.StoppedAsync(CancellationToken.None); + + await Assert.That(inner.StoppingCalls).IsEqualTo(1); + await Assert.That(inner.StoppedCalls).IsEqualTo(1); + } +} + +internal class CountingHostedService : IHostedService +{ + private int _stopCalls; + + public int StopCalls => _stopCalls; + + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + public async Task StopAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _stopCalls); + + // Keep the stop in flight briefly so concurrent duplicate callers would + // overlap it rather than arrive after completion. + await Task.Delay(50, CancellationToken.None); + } +} + +internal sealed class SyncThrowingStopHostedService : IHostedService +{ + private int _stopCalls; + + public int StopCalls => _stopCalls; + + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + public Task StopAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _stopCalls); + throw new InvalidOperationException("stop failed synchronously"); + } +} + +internal sealed class BlockingStopHostedService : IHostedService +{ + private readonly TaskCompletionSource _entered = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _stopCalls; + + public Task Entered => _entered.Task; + public int StopCalls => _stopCalls; + + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + public async Task StopAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _stopCalls); + _entered.SetResult(); + await _release.Task; + } + + public void Release() => _release.SetResult(); +} + +internal sealed class SynchronouslyBlockingStopHostedService : IHostedService, IDisposable +{ + private readonly ManualResetEventSlim _entered = new(); + private readonly ManualResetEventSlim _release = new(); + private int _stopCalls; + + public int StopCalls => _stopCalls; + + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + public Task StopAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _stopCalls); + _entered.Set(); + _release.Wait(); + return Task.CompletedTask; + } + + public void WaitUntilEntered() + { + if (!_entered.Wait(TimeSpan.FromSeconds(5))) + { + throw new TimeoutException("StopAsync was not entered within five seconds."); + } + } + public void Release() => _release.Set(); + + public void Dispose() + { + _entered.Dispose(); + _release.Dispose(); + } +} + +internal sealed class CountingLifecycleHostedService : IHostedService, IHostedLifecycleService +{ + private int _stoppingCalls; + private int _stoppedCalls; + + public int StoppingCalls => _stoppingCalls; + public int StoppedCalls => _stoppedCalls; + + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public Task StartingAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public Task StartedAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + public Task StoppingAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _stoppingCalls); + return Task.CompletedTask; + } + + public Task StoppedAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _stoppedCalls); + return Task.CompletedTask; + } +}