Bound the shutdown drain so a stuck handler cannot stall the host

docs/configuration.md

@@ -89,6 +89,7 @@ A complete example covering all sections:
           "BatchSize": 100,
           "HeadRefreshInterval": "00:00:00.100",
           "HeadWindowSize": 2000,
+          "DrainTimeout": "00:00:05",
           "Retry": {
             "MaxRetries": 3,
             "RetryDelay": "00:00:01",
@@ -183,6 +184,17 @@ the same pattern — see [Custom backends](#custom-backends).
 | `BatchSize` | `int` | `100` | `ControlLoop:BatchSize` |
 | `HeadRefreshInterval` | `TimeSpan` | `00:00:00.100` (100 ms) | `ControlLoop:HeadRefreshInterval` |
 | `HeadWindowSize` | `int` | `2000` | `ControlLoop:HeadWindowSize` |
+| `DrainTimeout` | `TimeSpan` | `00:00:05` (5 s) | `ControlLoop:DrainTimeout` |
+
+`DrainTimeout` bounds how long shutdown waits for in-flight handlers. It applies to the control
+loop, the stable-head tracker and the dead-letter retry loop. A handler that ignores its
+`CancellationToken` would otherwise block `StopAsync` indefinitely — stalling host shutdown and,
+with leasing enabled, holding the processor lease until it expires. When the timeout elapses the
+wait is abandoned with a warning; nothing is lost, because a handler that never returns is never
+checkpointed, so its event is re-delivered on the next start.
+
+Size it above your slowest legitimate handler and below your orchestrator's termination grace
+period (Kubernetes `terminationGracePeriodSeconds`, 30 s by default).
 
 ### Retry options
 
@@ -397,7 +409,7 @@ surfacing them in one error message.
 | `ALB0001` | No backend was declared | Call `.WithPostgres(...)` or `.WithInMemory()` inside `AddAlberto` |
 | `ALB0002` | Two or more processors share the same id within one module | Add `[ProcessorId("...")]` to disambiguate |
 | `ALB0003` | `.WithTenancy()` declared but the backend does not support tenancy | Use `.WithPostgres(...)`, which supports tenancy, or remove `.WithTenancy()` |
-| `ALB0004` | A control loop duration or count is ≤ 0 (`PollingInterval`, `HeadRefreshInterval`, `BatchSize`, or `HeadWindowSize`) | Set a positive value in code or configuration |
+| `ALB0004` | A control loop duration or count is ≤ 0 (`PollingInterval`, `HeadRefreshInterval`, `DrainTimeout`, `BatchSize`, or `HeadWindowSize`) | Set a positive value in code or configuration |
 | `ALB0005` | `MaxConcurrency > 1` with `BatchingMode = Disabled` — concurrency only applies within a batch | Set `BatchingMode` to `IfSupported` or `Required`, or reduce `MaxConcurrency` to 1 |
 | `ALB0006` | A processor id is empty or contains whitespace | Use a non-empty identifier without whitespace |
 | `ALB0007` | `Retry.MaxRetries < 0` or `Retry.BackoffMultiplier < 1.0` | Use 0 to disable retries; use 1.0 for a constant backoff delay |

src/Alberto.Dcb/Configuration/AlbertoModuleValidator.cs

@@ -214,6 +214,14 @@ private static void ValidateControlLoop(AlbertoModuleDefinition definition, List
                 $"Set a positive window size via .WithControlLoop(o => o with {{ HeadWindowSize = ... }}) or '{path}:ControlLoop:HeadWindowSize'."));
         }
 
+        if (loop.DrainTimeout <= TimeSpan.Zero)
+        {
+            failures.Add(new AlbertoValidationFailure(
+                "ALB0004",
+                $"ControlLoop.DrainTimeout is {loop.DrainTimeout}, which is not a positive duration.",
+                $"Set a positive timeout via .WithControlLoop(o => o with {{ DrainTimeout = ... }}) or '{path}:ControlLoop:DrainTimeout'."));
+        }
+
         if (loop.Retry.MaxRetries < 0)
         {
             failures.Add(new AlbertoValidationFailure(

src/Alberto.Dcb/Configuration/ControlLoopOptions.cs

@@ -17,6 +17,18 @@ public sealed record ControlLoopOptions
     /// <summary>Size of the in-flight transaction window the head tracker keeps. Default 2000.</summary>
     public int HeadWindowSize { get; init; } = 2000;
 
+    /// <summary>
+    /// How long shutdown waits for an in-flight handler to drain before abandoning it. Default 5 s.
+    /// <para>
+    /// A handler that ignores its <see cref="CancellationToken"/> would otherwise block
+    /// <c>StopAsync</c> forever, stalling host shutdown and — under leasing — holding the
+    /// processor lease past its expiry. When the timeout elapses the loop stops waiting,
+    /// logs a warning and flushes the checkpoint at the last safely-completed position, so
+    /// abandoned events are re-delivered on the next start.
+    /// </para>
+    /// </summary>
+    public TimeSpan DrainTimeout { get; init; } = TimeSpan.FromSeconds(5);
+
     /// <summary>Retry behaviour for failing handlers.</summary>
     public RetryOptions Retry { get; init; } = new();
 
@@ -48,6 +60,9 @@ public sealed class ControlLoopOverrides : IAlbertoOverrides<ControlLoopOptions>
     /// <summary>Mirror of <see cref="ControlLoopOptions.HeadWindowSize"/>.</summary>
     public int? HeadWindowSize { get; set; }
 
+    /// <summary>Mirror of <see cref="ControlLoopOptions.DrainTimeout"/>.</summary>
+    public TimeSpan? DrainTimeout { get; set; }
+
     /// <summary>Mirror of <see cref="ControlLoopOptions.Retry"/>.</summary>
     public RetryOverrides? Retry { get; set; }
 
@@ -71,6 +86,7 @@ public ControlLoopOptions ApplyTo(ControlLoopOptions options)
             BatchSize = BatchSize ?? options.BatchSize,
             HeadRefreshInterval = HeadRefreshInterval ?? options.HeadRefreshInterval,
             HeadWindowSize = HeadWindowSize ?? options.HeadWindowSize,
+            DrainTimeout = DrainTimeout ?? options.DrainTimeout,
             Retry = Retry?.ApplyTo(options.Retry) ?? options.Retry,
             DeadLetterRetry = DeadLetterRetry?.ApplyTo(options.DeadLetterRetry) ?? options.DeadLetterRetry,
             Leases = Leases?.ApplyTo(options.Leases) ?? options.Leases,

src/Alberto.Dcb/ControlLoopRegistration.cs

@@ -51,7 +51,8 @@ static IEventStoreBackend Backend(IServiceProvider sp, string moduleKey) =>
                 options.HeadRefreshInterval,
                 options.HeadWindowSize,
                 sp.GetService<ILogger<EventStoreHead>>(),
-                signal);
+                signal,
+                options.DrainTimeout);
         });
 
         services.AddSingleton<IHostedService>(sp =>
@@ -108,7 +109,8 @@ static IEventStoreBackend Backend(IServiceProvider sp, string moduleKey) =>
                     executionOptionsByProcessorId.GetValueOrDefault(
                         p.ProcessorId,
                         ProcessorExecutionOptions.Default),
-                    logger))
+                    logger,
+                    options.DrainTimeout))
                 .ToList();
 
             if (!options.Leases.Enabled)
@@ -222,7 +224,8 @@ static IEventStoreBackend Backend(IServiceProvider sp, string moduleKey) =>
                     middlewares,
                     logger,
                     options.DeadLetterRetry.ClaimLease,
-                    replicaId))
+                    replicaId,
+                    drainTimeout: options.DrainTimeout))
                 .ToList();
 
             return new DeadLetterRetryLoopGroup(retryLoops);

src/Alberto.Dcb/DcbModuleBuilderExtensions.cs

@@ -639,7 +639,8 @@ static IEventStoreBackend Backend(IServiceProvider sp, string key) =>
                         processor, head, backend, checkpoints,
                         opts.PollingInterval, opts.BatchSize, moduleKey,
                         ProcessorExecutionOptions.Default,
-                        sp.GetService<ILogger<ControlLoop>>());
+                        sp.GetService<ILogger<ControlLoop>>(),
+                        opts.DrainTimeout);
                 });
             });
 

src/Alberto.Dcb/PublicAPI.Unshipped.txt

@@ -63,6 +63,8 @@ Alberto.Dcb.Configuration.ControlLoopOptions.BatchSize.get -> int
 Alberto.Dcb.Configuration.ControlLoopOptions.BatchSize.init -> void
 Alberto.Dcb.Configuration.ControlLoopOptions.DeadLetterRetry.get -> Alberto.Dcb.Configuration.DeadLetterRetryOptions!
 Alberto.Dcb.Configuration.ControlLoopOptions.DeadLetterRetry.init -> void
+Alberto.Dcb.Configuration.ControlLoopOptions.DrainTimeout.get -> System.TimeSpan
+Alberto.Dcb.Configuration.ControlLoopOptions.DrainTimeout.init -> void
 Alberto.Dcb.Configuration.ControlLoopOptions.HeadRefreshInterval.get -> System.TimeSpan
 Alberto.Dcb.Configuration.ControlLoopOptions.HeadRefreshInterval.init -> void
 Alberto.Dcb.Configuration.ControlLoopOptions.HeadWindowSize.get -> int
@@ -82,6 +84,8 @@ Alberto.Dcb.Configuration.ControlLoopOverrides.BatchSize.set -> void
 Alberto.Dcb.Configuration.ControlLoopOverrides.ControlLoopOverrides() -> void
 Alberto.Dcb.Configuration.ControlLoopOverrides.DeadLetterRetry.get -> Alberto.Dcb.Configuration.DeadLetterRetryOverrides?
 Alberto.Dcb.Configuration.ControlLoopOverrides.DeadLetterRetry.set -> void
+Alberto.Dcb.Configuration.ControlLoopOverrides.DrainTimeout.get -> System.TimeSpan?
+Alberto.Dcb.Configuration.ControlLoopOverrides.DrainTimeout.set -> void
 Alberto.Dcb.Configuration.ControlLoopOverrides.HeadRefreshInterval.get -> System.TimeSpan?
 Alberto.Dcb.Configuration.ControlLoopOverrides.HeadRefreshInterval.set -> void
 Alberto.Dcb.Configuration.ControlLoopOverrides.HeadWindowSize.get -> int?

src/Alberto.Dcb/Subscriptions/ControlLoop.cs

@@ -27,12 +27,14 @@ public sealed class ControlLoop : IHostedService, IAsyncDisposable
     private readonly IReadOnlyList<BatchConsumeMiddleware> _batchMiddlewares;
     private readonly bool _hasUnpairedPerEventMiddlewares;
     private readonly ProcessorExecutionOptions _executionOptions;
+    private readonly TimeSpan _drainTimeout;
     private readonly ILogger<ControlLoop>? _logger;
     // Pre-composed middleware chains built once at construction time (PERF-6).
     private readonly Func<ConsumeEventContext, Func<Task>, Task> _composedMiddleware;
     private readonly Func<BatchConsumeContext, Func<Task>, Task> _composedBatchMiddleware;
     private CancellationTokenSource? _cts;
     private Task? _loop;
+    private int _disposed;
 
     public bool IsFaulted { get; private set; }
     public string ProcessorId => _processor.ProcessorId;
@@ -49,7 +51,8 @@ internal ControlLoop(
         IReadOnlyList<BatchConsumeMiddleware>? batchMiddlewares = null,
         bool hasUnpairedPerEventMiddlewares = false,
         ProcessorExecutionOptions? executionOptions = null,
-        ILogger<ControlLoop>? logger = null)
+        ILogger<ControlLoop>? logger = null,
+        TimeSpan? drainTimeout = null)
     {
         _processor = processor;
         _head = head;
@@ -62,6 +65,7 @@ internal ControlLoop(
         _batchMiddlewares = batchMiddlewares ?? [];
         _hasUnpairedPerEventMiddlewares = hasUnpairedPerEventMiddlewares;
         _executionOptions = executionOptions ?? ProcessorExecutionOptions.Default;
+        _drainTimeout = drainTimeout ?? Configuration.ControlLoopOptions.Default.DrainTimeout;
         _logger = logger;
 
         // Pre-build the composed middleware chains once so per-event dispatch does not
@@ -112,26 +116,90 @@ public Task StartAsync(CancellationToken cancellationToken)
         return Task.CompletedTask;
     }
 
+    /// <summary>
+    /// Cancels the loop and waits for it to drain, bounded by the configured drain timeout.
+    /// <para>
+    /// A handler that ignores its <see cref="CancellationToken"/> cannot stall shutdown
+    /// indefinitely: once the timeout (or <paramref name="cancellationToken"/>) fires the
+    /// wait is abandoned and a warning is logged. Abandoning the wait never advances the
+    /// checkpoint past an unprocessed event — a worker that never returns also never calls
+    /// <c>MarkCompleted</c>, so the safe checkpoint stays behind it and the event is
+    /// re-delivered on the next start.
+    /// </para>
+    /// </summary>
     public async Task StopAsync(CancellationToken cancellationToken)
     {
         if (_cts is not null)
         {
             try { await _cts.CancelAsync(); }
             catch (ObjectDisposedException) { }
         }
-        if (_loop is not null) try { await _loop; } catch (OperationCanceledException) { }
+
+        if (_loop is null) return;
+
+        try
+        {
+            await _loop.WaitAsync(_drainTimeout, cancellationToken);
+        }
+        catch (OperationCanceledException) { }
+        catch (TimeoutException)
+        {
+            _logger?.LogWarning(
+                "ControlLoop {ProcessorId} did not drain within {DrainTimeout}; abandoning the wait. " +
+                "In-flight handlers are still running and were not checkpointed; their events will be " +
+                "re-delivered on the next start.",
+                ProcessorId, _drainTimeout);
+        }
     }
 
     public async ValueTask DisposeAsync()
     {
+        if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
+
+        var loop = _loop;
+        var abandoned = false;
+        CancellationTokenSource? cts = null;
+
         try
         {
             await StopAsync(CancellationToken.None);
         }
         finally
         {
-            Interlocked.Exchange(ref _cts, null)?.Dispose();
-            if (_processor is IAsyncDisposable d) await d.DisposeAsync();
+            abandoned = loop is not null && !loop.IsCompleted;
+            cts = Interlocked.Exchange(ref _cts, null);
+        }
+
+        if (abandoned)
+        {
+            // The loop is still running and its workers still hold tokens from _cts, so
+            // neither the CTS nor the processor may be torn down yet. Hand both off to a
+            // detached continuation that runs once the loop finally exits.
+            _ = ReleaseWhenLoopExitsAsync(loop!, cts, _processor as IAsyncDisposable);
+            return;
+        }
+
+        cts?.Dispose();
+        if (_processor is IAsyncDisposable d) await d.DisposeAsync();
+    }
+
+    /// <summary>
+    /// Deferred teardown for a loop that outlived its drain timeout: waits (unbounded, off
+    /// the shutdown path) for the abandoned loop to exit, then releases the resources its
+    /// workers were still using.
+    /// </summary>
+    private static async Task ReleaseWhenLoopExitsAsync(
+        Task loop, CancellationTokenSource? cts, IAsyncDisposable? processor)
+    {
+        try { await loop.ConfigureAwait(false); }
+        catch { /* the loop's own failure is logged where it happens */ }
+
+        cts?.Dispose();
+
+        if (processor is not null)
+        {
+            try { await processor.DisposeAsync().ConfigureAwait(false); }
+            catch { /* best-effort teardown after an abandoned drain */ }
         }
     }
 
@@ -226,7 +294,9 @@ private async Task RunPipelinedAsync(CancellationToken ct)
         var maxConcurrency = _executionOptions.MaxConcurrency;
         var initialCheckpoint = await _checkpointStore.GetAsync(ProcessorId, ct) ?? 0L;
         var watermark = new PositionWatermark(initialCheckpoint);
-        using var pipelineCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+        // Not `using`: when the worker drain times out the abandoned workers still hold this
+        // token, so disposal is deferred until they actually exit (see the finally block).
+        var pipelineCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
         var pipelineToken = pipelineCts.Token;
         Exception? pipelineFailure = null;
 
@@ -308,10 +378,40 @@ private async Task RunPipelinedAsync(CancellationToken ct)
         finally
         {
             channel.Writer.TryComplete();
-            await Task.WhenAll(workers);
-            // Final flush after all workers have drained
+
+            var drain = Task.WhenAll(workers);
+            var drained = true;
+
+            try
+            {
+                await drain.WaitAsync(_drainTimeout);
+            }
+            catch (TimeoutException)
+            {
+                drained = false;
+            }
+
+            // Final flush. Safe even when the drain timed out: a worker that never returned
+            // never called MarkCompleted, so SafeCheckpoint is still behind its position.
             await SaveWatermarkCheckpointAsync(watermark, CancellationToken.None);
 
+            if (drained)
+            {
+                pipelineCts.Dispose();
+            }
+            else
+            {
+                _logger?.LogWarning(
+                    "ControlLoop {ProcessorId} abandoned {WorkerCount} worker(s) that did not drain within " +
+                    "{DrainTimeout}. Checkpoint flushed at {SafeCheckpoint}; events in flight above that " +
+                    "position will be re-delivered on the next start.",
+                    ProcessorId, workers.Count(w => !w.IsCompleted), _drainTimeout, watermark.SafeCheckpoint);
+
+                // The abandoned workers still observe pipelineCts.Token — dispose only once
+                // they have actually exited.
+                _ = DisposeWhenDrainedAsync(drain, pipelineCts);
+            }
+
             if (pipelineFailure is not null)
             {
                 IsFaulted = true;
@@ -335,6 +435,18 @@ void ReportFailure(Exception failure)
         }
     }
 
+    /// <summary>
+    /// Disposes the pipeline's <see cref="CancellationTokenSource"/> once workers abandoned by
+    /// a drain timeout have finally exited, so their token registrations stay valid meanwhile.
+    /// </summary>
+    private static async Task DisposeWhenDrainedAsync(Task drain, CancellationTokenSource cts)
+    {
+        try { await drain.ConfigureAwait(false); }
+        catch { /* worker failures are already reported through ReportFailure */ }
+
+        cts.Dispose();
+    }
+
     private async Task RunWorkerAsync(
         ChannelReader<IEventEnvelope> reader,
         PositionWatermark watermark,

src/Alberto.Dcb/Subscriptions/ControlLoopAssembler.cs

@@ -107,15 +107,17 @@ public ControlLoop Create(
         int batchSize,
         string moduleKey,
         ProcessorExecutionOptions? executionOptions = null,
-        ILogger<ControlLoop>? logger = null)
+        ILogger<ControlLoop>? logger = null,
+        TimeSpan? drainTimeout = null)
     {
         var loop = new ControlLoop(
             processor, head, backend, checkpointStore,
             pollingInterval, batchSize, moduleKey,
             _middlewares, _batchMiddlewares,
             _hasUnpairedPerEventMiddlewares,
             executionOptions,
-            logger);
+            logger,
+            drainTimeout);
 
         // Wire a per-loop fence-violation handler through the interface rather than
         // via a concrete-type downcast. If the store is not fencable, this block is skipped