Merge pull request #310 from Vulthil/fix/outbox-relay-restart-race

Vulthil · web-flow · commit 041f40dfb5cf · 2026-07-09T21:49:54.000+02:00
fix: prevent restartable outbox relay cancellation from faulting the host on .NET 10
diff --git a/src/Vulthil.Extensions.Hosting/IRestartableHostedService.cs b/src/Vulthil.Extensions.Hosting/IRestartableHostedService.cs
@@ -7,6 +7,14 @@ namespace Vulthil.Extensions.Hosting;
 /// may pause it for the duration of an operation that must not run concurrently with it — for example a test harness
 /// resetting the database between tests, where a live database-polling relay would otherwise contend with the reset.
 /// Implement this only on services whose <see cref="IHostedService.StartAsync"/> / <see cref="IHostedService.StopAsync"/>
-/// are idempotent across repeated cycles (a <see cref="BackgroundService"/> whose <c>ExecuteAsync</c> re-runs cleanly).
+/// are idempotent across repeated cycles.
 /// </summary>
+/// <remarks>
+/// Prefer implementing <see cref="IHostedService"/> directly over inheriting <see cref="BackgroundService"/>: the
+/// host observes only the execute task created by a <see cref="BackgroundService"/>'s first start, and it treats a
+/// canceled execute task as a fault that stops the host whenever the application is not already shutting down. On
+/// .NET 10 that task is scheduled with a <c>Task.Run</c> gated on the stopping token, so a stop that races service
+/// startup can cancel the task before <c>ExecuteAsync</c> has run at all — no graceful handling inside
+/// <c>ExecuteAsync</c> can prevent the resulting host shutdown.
+/// </remarks>
 public interface IRestartableHostedService : IHostedService;
diff --git a/src/Vulthil.Extensions.Hosting/README.md b/src/Vulthil.Extensions.Hosting/README.md
@@ -16,10 +16,13 @@ around the per-test database reset, so a live database-polling relay (for exampl
 simply by implementing the marker — no test-only code, and the testing library never depends on the outbox engine.
 
 ```csharp
-internal sealed class OutboxBackgroundService(/* ... */) : BackgroundService, IRestartableHostedService
+internal sealed class OutboxBackgroundService(/* ... */) : IRestartableHostedService, IDisposable
 {
-    // ExecuteAsync is written so a Stop/Start cycle re-runs it cleanly.
+    // StartAsync/StopAsync manage the execute task so a Stop/Start cycle re-runs it cleanly.
 }
 ```
 
-Implement it only on services whose `StartAsync`/`StopAsync` are idempotent across repeated cycles.
+Implement it only on services whose `StartAsync`/`StopAsync` are idempotent across repeated cycles. Prefer
+implementing `IHostedService` directly over inheriting `BackgroundService`: the host observes only the execute task
+of a `BackgroundService`'s first start and stops the whole host when that task is canceled while the application is
+running — and on .NET 10 a stop racing service startup can cancel the task before `ExecuteAsync` has run at all.
diff --git a/src/Vulthil.Messaging.Inbox/InboxRetentionBackgroundService.cs b/src/Vulthil.Messaging.Inbox/InboxRetentionBackgroundService.cs
@@ -23,21 +23,34 @@ internal sealed class InboxRetentionBackgroundService(
     /// <inheritdoc />
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        using var timer = new PeriodicTimer(_options.SweepInterval, timeProvider);
-        do
+        try
+        {
+            using var timer = new PeriodicTimer(_options.SweepInterval, timeProvider);
+            do
+            {
+                await SweepSafelyAsync(stoppingToken);
+            }
+            while (await timer.WaitForNextTickAsync(stoppingToken));
+        }
+        catch (OperationCanceledException exception) when (stoppingToken.IsCancellationRequested)
         {
-            await SweepSafelyAsync(stoppingToken);
+            logger.LogDebug(exception, "Inbox retention sweep loop stopped");
         }
-        while (await timer.WaitForNextTickAsync(stoppingToken));
     }
 
+    /// <summary>
+    /// Runs one sweep, treating every failure as transient (logged, retried on the next tick). Only a cancellation
+    /// caused by the service stopping propagates; a foreign <see cref="OperationCanceledException"/> — for example a
+    /// store client surfacing a timeout as a cancellation — must not escape, because the host would treat the
+    /// canceled execute task of a still-running application as a fault and stop the host.
+    /// </summary>
     private async Task SweepSafelyAsync(CancellationToken cancellationToken)
     {
         try
         {
             await SweepAsync(cancellationToken);
         }
-        catch (Exception exception) when (exception is not OperationCanceledException)
+        catch (Exception exception) when (exception is not OperationCanceledException || !cancellationToken.IsCancellationRequested)
         {
             logger.LogError(exception, "Inbox retention sweep failed.");
         }
diff --git a/src/Vulthil.SharedKernel.Outbox/OutboxBackgroundService.cs b/src/Vulthil.SharedKernel.Outbox/OutboxBackgroundService.cs
@@ -6,15 +6,97 @@
 
 namespace Vulthil.SharedKernel.Outbox;
 
+/// <summary>
+/// Hosts the outbox relay loop, restartable via <see cref="IRestartableHostedService"/> so infrastructure (such as a
+/// test harness resetting the database) can pause it around operations that must not run concurrently with it.
+/// </summary>
+/// <remarks>
+/// The service implements <see cref="IHostedService"/> directly instead of inheriting
+/// <see cref="BackgroundService"/> because <see cref="BackgroundService"/> is not restart-safe: the host observes
+/// only the execute task created by the first <c>StartAsync</c>, and on .NET 10 that task is scheduled with
+/// <c>Task.Run(..., stoppingToken)</c> — a stop that lands before the thread pool has run the delegate transitions
+/// the observed task straight to canceled without <c>ExecuteAsync</c> ever running, which the host (whose default
+/// <c>BackgroundServiceExceptionBehavior</c> is <c>StopHost</c>) treats as a fault and stops the application while
+/// it is still serving. Owning the lifecycle removes both hazards: the execute task always runs the relay loop
+/// (cancellation is observed inside it and always ends in a graceful return), and the host never awaits the task.
+/// A genuine fault escaping the loop still stops the application — matching the <see cref="BackgroundService"/>
+/// default — via <see cref="IHostApplicationLifetime.StopApplication"/>.
+/// </remarks>
 internal sealed class OutboxBackgroundService(
     ILogger<OutboxBackgroundService> logger,
     IServiceScopeFactory serviceScopeFactory,
     IOutboxSignal signal,
     IEnumerable<IOutboxRelayGate> relayGates,
-    IOptions<OutboxProcessingOptions> options) : BackgroundService, IRestartableHostedService
+    IOptions<OutboxProcessingOptions> options,
+    IHostApplicationLifetime applicationLifetime) : IRestartableHostedService, IDisposable
 {
+    private CancellationTokenSource? _stoppingCts;
+
+    /// <summary>
+    /// Gets the task running the current relay loop, or <see langword="null"/> before the first start. The task
+    /// always completes successfully: cancellation ends it with a graceful return and a genuine fault is handled
+    /// inside it, so no caller ever observes an exception from a stopped relay.
+    /// </summary>
+    internal Task? ExecuteTask { get; private set; }
+
+    /// <inheritdoc />
+    public Task StartAsync(CancellationToken cancellationToken)
+    {
+        _stoppingCts?.Dispose();
+        _stoppingCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+        var stoppingToken = _stoppingCts.Token;
+        ExecuteTask = Task.Run(() => ExecuteAsync(stoppingToken), CancellationToken.None);
+        return Task.CompletedTask;
+    }
+
+    /// <inheritdoc />
+    public async Task StopAsync(CancellationToken cancellationToken)
+    {
+        if (ExecuteTask is null)
+        {
+            return;
+        }
+
+        try
+        {
+            await _stoppingCts!.CancelAsync();
+        }
+        finally
+        {
+            await ExecuteTask.WaitAsync(cancellationToken).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
+        }
+    }
+
     /// <inheritdoc />
-    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    public void Dispose()
+    {
+        _stoppingCts?.Cancel();
+        _stoppingCts?.Dispose();
+    }
+
+    /// <summary>
+    /// Runs the relay loop and guarantees the execute task completes successfully: a cancellation escaping the loop
+    /// after the service is stopped is swallowed, while any other exception is a genuine fault that stops the
+    /// application, mirroring the host's default behavior for faulted background services.
+    /// </summary>
+    private async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        try
+        {
+            await ProcessUntilStoppedAsync(stoppingToken);
+        }
+        catch (OperationCanceledException ex) when (stoppingToken.IsCancellationRequested)
+        {
+            logger.LogDebug(ex, "Outbox relay stopped before the processing loop observed the cancellation");
+        }
+        catch (Exception ex)
+        {
+            logger.LogCritical(ex, "Outbox relay faulted; stopping the application");
+            applicationLifetime.StopApplication();
+        }
+    }
+
+    private async Task ProcessUntilStoppedAsync(CancellationToken stoppingToken)
     {
         if (!await TryWaitForRelayGatesAsync(stoppingToken))
         {
diff --git a/src/Vulthil.SharedKernel.Outbox/OutboxRetentionBackgroundService.cs b/src/Vulthil.SharedKernel.Outbox/OutboxRetentionBackgroundService.cs
@@ -23,21 +23,34 @@ internal sealed class OutboxRetentionBackgroundService(
     /// <inheritdoc />
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        using var timer = new PeriodicTimer(_options.SweepInterval, timeProvider);
-        do
+        try
+        {
+            using var timer = new PeriodicTimer(_options.SweepInterval, timeProvider);
+            do
+            {
+                await SweepSafelyAsync(stoppingToken);
+            }
+            while (await timer.WaitForNextTickAsync(stoppingToken));
+        }
+        catch (OperationCanceledException exception) when (stoppingToken.IsCancellationRequested)
         {
-            await SweepSafelyAsync(stoppingToken);
+            logger.LogDebug(exception, "Outbox retention sweep loop stopped");
         }
-        while (await timer.WaitForNextTickAsync(stoppingToken));
     }
 
+    /// <summary>
+    /// Runs one sweep, treating every failure as transient (logged, retried on the next tick). Only a cancellation
+    /// caused by the service stopping propagates; a foreign <see cref="OperationCanceledException"/> — for example a
+    /// store client surfacing a timeout as a cancellation — must not escape, because the host would treat the
+    /// canceled execute task of a still-running application as a fault and stop the host.
+    /// </summary>
     private async Task SweepSafelyAsync(CancellationToken cancellationToken)
     {
         try
         {
             await SweepAsync(cancellationToken);
         }
-        catch (Exception exception) when (exception is not OperationCanceledException)
+        catch (Exception exception) when (exception is not OperationCanceledException || !cancellationToken.IsCancellationRequested)
         {
             logger.LogError(exception, "Outbox retention sweep failed.");
         }
diff --git a/tests/Vulthil.Messaging.Inbox.Tests/InboxRetentionBackgroundServiceTests.cs b/tests/Vulthil.Messaging.Inbox.Tests/InboxRetentionBackgroundServiceTests.cs
@@ -1,6 +1,7 @@
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
+using Vulthil.Messaging.Abstractions.Consumers;
 using Vulthil.xUnit;
 
 namespace Vulthil.Messaging.Inbox.Tests;
@@ -39,6 +40,39 @@ public async Task StoreWithoutRetentionSupportLogsExactlyOneWarningAcrossMultipl
         logger.WarningCount.ShouldBe(1);
     }
 
+    [Fact]
+    public async Task AForeignCancellationDuringASweepDoesNotEndTheExecuteTask()
+    {
+        // Arrange
+        Use<IOptions<InboxOptions>>(Options.Create(new InboxOptions()));
+        Use<IServiceScopeFactory>(new AutoMockerServiceScopeFactory(AutoMocker));
+        var store = new TimeoutThrowingRetentionStore();
+        Use<IIdempotencyStore>(store);
+
+        // Act
+        await Target.StartAsync(CancellationToken);
+        await store.FirstSweepAttempted;
+        await Target.StopAsync(CancellationToken);
+
+        // Assert
+        Target.ExecuteTask!.Status.ShouldBe(TaskStatus.RanToCompletion);
+    }
+
+    private sealed class TimeoutThrowingRetentionStore : IIdempotencyStore, IInboxRetentionStore
+    {
+        private readonly TaskCompletionSource _firstSweepAttempted = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public Task FirstSweepAttempted => _firstSweepAttempted.Task;
+
+        public Task<bool> ProcessAsync(string idempotencyKey, IMessageContext context, Func<CancellationToken, Task> process, CancellationToken cancellationToken) => Task.FromResult(false);
+
+        public Task<int> DeleteProcessedAsync(DateTimeOffset olderThanUtc, int batchSize, CancellationToken cancellationToken)
+        {
+            _firstSweepAttempted.TrySetResult();
+            throw new OperationCanceledException();
+        }
+    }
+
     private sealed class WarningCountingLogger : ILogger<InboxRetentionBackgroundService>
     {
         private readonly TaskCompletionSource _firstWarning = new(TaskCreationOptions.RunContinuationsAsynchronously);
diff --git a/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxBackgroundServiceTests.cs b/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxBackgroundServiceTests.cs
@@ -1,4 +1,6 @@
+using System.Threading.Channels;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Options;
 using Vulthil.xUnit;
 
@@ -23,7 +25,7 @@ public async Task StoppingWhileWaitingForRelayGatesCompletesGracefully()
         var gate = new BlockingRelayGate();
         Use<IEnumerable<IOutboxRelayGate>>([gate]);
         await Target.StartAsync(CancellationToken);
-        await gate.Entered;
+        await gate.WaitForEntryAsync(CancellationToken);
 
         // Act
         await Target.StopAsync(CancellationToken);
@@ -32,6 +34,74 @@ public async Task StoppingWhileWaitingForRelayGatesCompletesGracefully()
         Target.ExecuteTask!.Status.ShouldBe(TaskStatus.RanToCompletion);
     }
 
+    [Fact]
+    public async Task StoppingBeforeTheRelayLoopHasRunCompletesGracefully()
+    {
+        // Arrange
+        Use<IEnumerable<IOutboxRelayGate>>([new BlockingRelayGate()]);
+        await Target.StartAsync(CancellationToken);
+
+        // Act
+        await Target.StopAsync(CancellationToken);
+
+        // Assert
+        Target.ExecuteTask!.Status.ShouldBe(TaskStatus.RanToCompletion);
+    }
+
+    [Fact]
+    public async Task StartingWithAnAlreadyCanceledTokenCompletesTheExecuteTaskGracefully()
+    {
+        // Arrange
+        Use<IEnumerable<IOutboxRelayGate>>([new BlockingRelayGate()]);
+        using var canceledTokenSource = new CancellationTokenSource();
+        await canceledTokenSource.CancelAsync();
+
+        // Act
+        await Target.StartAsync(canceledTokenSource.Token);
+        await Target.ExecuteTask!;
+
+        // Assert
+        Target.ExecuteTask!.Status.ShouldBe(TaskStatus.RanToCompletion);
+    }
+
+    [Fact]
+    public async Task RestartingAfterStopRunsTheRelayAgain()
+    {
+        // Arrange
+        var gate = new BlockingRelayGate();
+        Use<IEnumerable<IOutboxRelayGate>>([gate]);
+        await Target.StartAsync(CancellationToken);
+        await gate.WaitForEntryAsync(CancellationToken);
+        await Target.StopAsync(CancellationToken);
+        var firstExecuteTask = Target.ExecuteTask;
+
+        // Act
+        await Target.StartAsync(CancellationToken);
+        await gate.WaitForEntryAsync(CancellationToken);
+        await Target.StopAsync(CancellationToken);
+
+        // Assert
+        Target.ExecuteTask.ShouldNotBeSameAs(firstExecuteTask);
+        Target.ExecuteTask!.Status.ShouldBe(TaskStatus.RanToCompletion);
+    }
+
+    [Fact]
+    public async Task AFaultOutsideTheProcessingLoopStopsTheApplication()
+    {
+        // Arrange
+        Use<IEnumerable<IOutboxRelayGate>>([]);
+        var failingOptions = new Mock<IOptions<OutboxProcessingOptions>>();
+        failingOptions.Setup(o => o.Value).Throws(new InvalidOperationException("Options failed"));
+        Use(failingOptions.Object);
+
+        // Act
+        await Target.StartAsync(CancellationToken);
+        await Target.ExecuteTask!;
+
+        // Assert
+        GetMock<IHostApplicationLifetime>().Verify(lifetime => lifetime.StopApplication(), Times.Once);
+    }
+
     [Fact]
     public async Task ABatchWithFewerSuccessesThanTheBatchSizeDelaysTheNextFetch()
     {
@@ -87,13 +157,13 @@ public async Task AFullySuccessfulFullBatchRefetchesImmediately()
 
     private sealed class BlockingRelayGate : IOutboxRelayGate
     {
-        private readonly TaskCompletionSource _entered = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly Channel<bool> _entries = Channel.CreateUnbounded<bool>();
 
-        public Task Entered => _entered.Task;
+        public async Task WaitForEntryAsync(CancellationToken cancellationToken) => await _entries.Reader.ReadAsync(cancellationToken);
 
         public Task WaitUntilReadyAsync(CancellationToken cancellationToken)
         {
-            _entered.TrySetResult();
+            _entries.Writer.TryWrite(true);
             return Task.Delay(Timeout.Infinite, cancellationToken);
         }
     }
diff --git a/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxRetentionBackgroundServiceTests.cs b/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxRetentionBackgroundServiceTests.cs

Original file line number	Diff line number	Diff line change
`@@ -23,21 +23,34 @@ internal sealed class InboxRetentionBackgroundService(`
`23`	`23`	`/// <inheritdoc />`
`24`	`24`	`protected override async Task ExecuteAsync(CancellationToken stoppingToken)`
`25`	`25`	`{`
`26`		`- using var timer = new PeriodicTimer(_options.SweepInterval, timeProvider);`
`27`		`- do`
	`26`	`+ try`
	`27`	`+ {`
	`28`	`+ using var timer = new PeriodicTimer(_options.SweepInterval, timeProvider);`
	`29`	`+ do`
	`30`	`+ {`
	`31`	`+ await SweepSafelyAsync(stoppingToken);`
	`32`	`+ }`
	`33`	`+ while (await timer.WaitForNextTickAsync(stoppingToken));`
	`34`	`+ }`
	`35`	`+ catch (OperationCanceledException exception) when (stoppingToken.IsCancellationRequested)`
`28`	`36`	`{`
`29`		`- await SweepSafelyAsync(stoppingToken);`
	`37`	`+ logger.LogDebug(exception, "Inbox retention sweep loop stopped");`
`30`	`38`	`}`
`31`		`- while (await timer.WaitForNextTickAsync(stoppingToken));`
`32`	`39`	`}`
`33`	`40`
	`41`	`+ /// <summary>`
	`42`	`+ /// Runs one sweep, treating every failure as transient (logged, retried on the next tick). Only a cancellation`
	`43`	`+ /// caused by the service stopping propagates; a foreign <see cref="OperationCanceledException"/> — for example a`
	`44`	`+ /// store client surfacing a timeout as a cancellation — must not escape, because the host would treat the`
	`45`	`+ /// canceled execute task of a still-running application as a fault and stop the host.`
	`46`	`+ /// </summary>`
`34`	`47`	`private async Task SweepSafelyAsync(CancellationToken cancellationToken)`
`35`	`48`	`{`
`36`	`49`	`try`
`37`	`50`	`{`
`38`	`51`	`await SweepAsync(cancellationToken);`
`39`	`52`	`}`
`40`		`- catch (Exception exception) when (exception is not OperationCanceledException)`
	`53`	`+ catch (Exception exception) when (exception is not OperationCanceledException \|\| !cancellationToken.IsCancellationRequested)`
`41`	`54`	`{`
`42`	`55`	`logger.LogError(exception, "Inbox retention sweep failed.");`
`43`	`56`	`}`