Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 67 additions & 13 deletions TUnit.AspNetCore.Core/FlowSuppressingHostedService.cs
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,36 @@ namespace TUnit.AspNetCore;
/// Without this, wrapped services that own unmanaged resources leak silently because
/// the container only sees the non-disposable wrapper.
/// </para>
/// <para>
/// The stop-phase methods are additionally guarded to run the inner service's stop
/// exactly once. Minimal-hosting SUTs park their own <c>app.Run()</c> in
/// <c>WaitForShutdownAsync</c>, which calls <c>Host.StopAsync</c> when
/// <c>ApplicationStopping</c> fires — concurrently with the <c>Host.StopAsync</c> that
/// <c>WebApplicationFactory.DisposeAsync</c> already has in flight (the trigger of that
/// very <c>ApplicationStopping</c>). <c>Host.StopAsync</c> has no concurrency guard, so
/// every hosted service's <c>StopAsync</c> runs twice in parallel, breaking services
/// with non-thread-safe shutdown (e.g. Rebus' bus dispose throws
/// <see cref="ObjectDisposedException"/>). See https://github.com/thomhurst/TUnit/issues/6339.
/// All concurrent and repeated callers observe the single underlying stop task.
/// </para>
/// </remarks>
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);
public Task StopAsync(CancellationToken cancellationToken)
{
lock (_stopGate)
{
return _stopTask ??= InvokeOnce(inner.StopAsync, cancellationToken);

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve each StopAsync caller's cancellation token

When a second Host.StopAsync arrives while the first stop is still running, this returns the cached task that was created with the first caller's token, so the duplicate caller's token is never observed. In contexts that pass a shorter or already-canceling shutdown token, that caller can now wait for the original stop task and run past its own timeout instead of canceling; share the single inner invocation, but have duplicate waits respect their own cancellationToken.

Useful? React with 👍 / 👎.

}
}

public Task StartingAsync(CancellationToken cancellationToken) =>
inner is IHostedLifecycleService lifecycle
Expand All @@ -42,18 +64,36 @@ 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;
lock (_stopGate)
{
return _stoppingTask ??= InvokeOnce(lifecycle.StoppingAsync, cancellationToken);
}
}

public Task StoppedAsync(CancellationToken cancellationToken)
{
if (inner is not IHostedLifecycleService lifecycle)
{
return Task.CompletedTask;
}

lock (_stopGate)
{
return _stoppedTask ??= InvokeOnce(lifecycle.StoppedAsync, cancellationToken);
}
}

/// <inheritdoc />
public async ValueTask DisposeAsync()
Expand Down Expand Up @@ -83,6 +123,20 @@ public void Dispose()
}
}

// Captures a synchronous throw as a faulted task so it is cached in the once-guard —
// otherwise a second caller would re-invoke the inner stop method.
private static Task InvokeOnce(Func<CancellationToken, Task> op, CancellationToken ct)
{
try
{
return op(ct);
}
catch (Exception ex)
{
return Task.FromException(ex);
}
}

// 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
Expand Down
72 changes: 72 additions & 0 deletions TUnit.AspNetCore.Tests/HostStopExactlyOnceProbeTests.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using TUnit.AspNetCore;

namespace TUnit.AspNetCore.Tests;

/// <summary>
/// End-to-end regression test for https://github.com/thomhurst/TUnit/issues/6339.
/// <para>
/// With a minimal-hosting SUT, disposing the per-test factory races the SUT's own
/// <c>app.Run()</c> shutdown path: <c>WebApplicationFactory.DisposeAsync</c> calls
/// <c>Host.StopAsync</c>, whose <c>ApplicationStopping</c> signal wakes the app's parked
/// <c>WaitForShutdownAsync</c>, which calls <c>Host.StopAsync</c> again — concurrently.
/// Before the stop-once guard in <see cref="FlowSuppressingHostedService"/>, each hosted
/// service's <c>StopAsync</c> 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
/// <see cref="ObjectDisposedException"/>.
/// </para>
/// <para>
/// The probe throws — with both stack traces — if it ever observes a second
/// <c>StopAsync</c> entry, so a regression surfaces as loud test failures.
/// </para>
/// </summary>
public class HostStopExactlyOnceProbeTests : WebApplicationTest<TestWebAppFactory, Program>
{
protected override void ConfigureTestServices(IServiceCollection services)
{
services.AddSingleton<IHostedService, StopProbeHostedService>();
}

[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));
}
}

/// <summary>
/// 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.
/// </summary>
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);
}
}
150 changes: 150 additions & 0 deletions TUnit.AspNetCore.Tests/HostedServiceStopOnceTests.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
using Microsoft.Extensions.Hosting;
using TUnit.AspNetCore;

namespace TUnit.AspNetCore.Tests;

/// <summary>
/// Regression tests for https://github.com/thomhurst/TUnit/issues/6339.
/// <para>
/// Minimal-hosting SUTs park <c>app.Run()</c> in <c>WaitForShutdownAsync</c>, which calls
/// <c>Host.StopAsync</c> when <c>ApplicationStopping</c> fires — concurrently with the
/// <c>Host.StopAsync</c> from <c>WebApplicationFactory.DisposeAsync</c> that triggered it.
/// Every hosted service's stop then runs twice in parallel, breaking services with
/// non-thread-safe shutdown (Rebus' bus dispose throws <see cref="ObjectDisposedException"/>).
/// <see cref="FlowSuppressingHostedService"/> must absorb the duplicate calls.
/// </para>
/// </summary>
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 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<InvalidOperationException>();
await Assert.That(async () => await wrapper.StopAsync(CancellationToken.None))
.Throws<InvalidOperationException>();

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 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;
}
}
Loading