fix(runtime): close drain cancel-dispose race (#8123)

* fix(runtime): close cancel-dispose race

* fix(runtime): recover disposed active checkpoints

* fix(runtime): guard disposed checkpoint recovery

* test(runtime): share suspended instance fixture

* fix(runtime): serialize cycle CTS lifecycle

* fix(runtime): serialize cycle CTS cleanup

* test(runtime): update lifecycle comment

* fix(runtime): avoid CTS callback disposal deadlock

* fix(runtime): preserve cancellation recovery races

* fix(runtime): preserve cancel state after callback errors

* test(runtime): clean up blocked drain callbacks

* fix(runtime): guard linked token cancellation cleanup

* fix(runtime): preserve fatal cancellation callback failures

* fix(common): classify fatal exceptions inside aggregates

* fix(runtime): await deferred recovery after drain cancellation

* fix(runtime): clarify deferred recovery timeout

* test(runtime): await cancellation cleanup before disposal assertion

src/modules/Elsa.Common/Extensions/ExceptionExtensions.cs

@@ -25,9 +25,10 @@ public static class ExceptionExtensions
     ///   <item><see cref="ThreadAbortException"/></item>
     /// </list>
     /// <para>
-    /// Wrapper exceptions (<see cref="TypeInitializationException"/>, <see cref="System.Reflection.TargetInvocationException"/>)
-    /// are unwrapped before classification so that, for example, a <see cref="TypeInitializationException"/>
-    /// wrapping a <see cref="StackOverflowException"/> is classified as fatal.
+    /// Wrapper exceptions (<see cref="AggregateException"/>, <see cref="TypeInitializationException"/>,
+    /// <see cref="System.Reflection.TargetInvocationException"/>) are unwrapped before classification so that,
+    /// for example, an aggregate or <see cref="TypeInitializationException"/> wrapping a
+    /// <see cref="StackOverflowException"/> is classified as fatal.
     /// </para>
     /// <para>
     /// Pattern: use as a filter on a generic <c>catch</c> where the surrounding logic must remain best-effort
@@ -45,6 +46,9 @@ public static class ExceptionExtensions
     {
         while (exception is not null)
         {
+            if (exception is AggregateException aggregateException)
+                return aggregateException.Flatten().InnerExceptions.Any(inner => inner.IsFatal());
+
             switch (exception)
             {
                 case StackOverflowException:

src/modules/Elsa.Workflows.Runtime/Models/ExecutionCycleHandle.cs

@@ -10,11 +10,20 @@ namespace Elsa.Workflows.Runtime;
 public sealed class ExecutionCycleHandle : IDisposable
 {
     private readonly CancellationTokenSource _cycleCts;
+    private readonly CancellationTokenRegistration _linkedTokenRegistration;
     private readonly Action<ExecutionCycleHandle>? _onDisposed;
     private readonly Action? _cancelCallback;
     private readonly TaskCompletionSource _disposedTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
-    private int _cancelled;
-    private int _disposed;
+    private readonly object _cycleCtsGate = new();
+    private int _cycleCtsCancellationInProgress;
+    private bool _cycleCtsDisposeRequested;
+    private bool _cycleCtsDisposed;
+    private int _lifecycleState;
+
+    private const int ActiveState = 0;
+    private const int CancellingState = 1;
+    private const int CancelledState = 2;
+    private const int DisposedState = 3;
 
     /// <summary>
     /// Creates a new handle. The owning <see cref="IExecutionCycleRegistry"/> supplies <paramref name="onDisposed"/>
@@ -39,7 +48,10 @@ public sealed class ExecutionCycleHandle : IDisposable
         WorkflowInstanceId = workflowInstanceId;
         IngressSourceName = ingressSourceName;
         StartedAt = startedAt;
-        _cycleCts = CancellationTokenSource.CreateLinkedTokenSource(linkedToken);
+        _cycleCts = new CancellationTokenSource();
+        _linkedTokenRegistration = linkedToken.UnsafeRegister(
+            static state => ((ExecutionCycleHandle)state!).PropagateLinkedCancellation(),
+            this);
         _onDisposed = onDisposed;
         _cancelCallback = cancelCallback;
     }
@@ -62,10 +74,12 @@ public sealed class ExecutionCycleHandle : IDisposable
     public CancellationToken CancellationToken => _cycleCts.Token;
 
     /// <summary>
-    /// Completes when <see cref="Dispose"/> runs — i.e., when the workflow runner finishes the cycle (cleanly or
-    /// via cancellation) and the middleware exits its <c>using</c> block. The drain orchestrator awaits this with
-    /// a timeout before persisting <see cref="WorkflowSubStatus.Interrupted"/>, ensuring its write happens AFTER
-    /// any commit the runner emits in response to <see cref="Cancel"/>.
+    /// Completes after <see cref="Dispose"/> logically releases the handle and physically cleans up its linked CTS —
+    /// i.e., when the workflow runner finishes the cycle (cleanly or via cancellation) and the middleware exits its
+    /// <c>using</c> block. If cancellation callbacks are in flight, <see cref="Dispose"/> may return before this
+    /// cleanup completes. The drain orchestrator awaits this with a timeout before persisting
+    /// <see cref="WorkflowSubStatus.Interrupted"/>, ensuring its write happens AFTER any commit the runner emits in
+    /// response to <see cref="Cancel"/>.
     /// </summary>
     public Task Disposed => _disposedTcs.Task;
 
@@ -78,22 +92,13 @@ public sealed class ExecutionCycleHandle : IDisposable
 
     /// <summary>
     /// Attempts to cancel the cycle. Returns <c>true</c> only when this call transitioned the handle from
-    /// not-cancelled to cancelled. Returns <c>false</c> when the handle was already disposed or already cancelled,
-    /// so drain can avoid treating a finished cycle as a force-cancel.
+    /// active to cancelled. Returns <c>false</c> when the handle was already disposed or already cancelling/cancelled,
+    /// so drain can avoid treating a finished cycle as a force-cancel. Disposal wins if it races with the cancellation
+    /// callback, so a cycle that completes while cancellation is in flight is not reported as drain-cancelled.
     /// </summary>
     public bool TryCancel()
     {
-        if (Volatile.Read(ref _disposed) != 0)
-            return false;
-
-        // Idempotent guard: ensures the cancel callback and CTS cancellation run AT MOST once even if Cancel() is
-        // called repeatedly before disposal. Without this the drain orchestrator (or any other future caller) could
-        // accidentally trigger a non-idempotent cancellation side effect multiple times.
-        if (Interlocked.Exchange(ref _cancelled, 1) != 0)
-            return false;
-
-        // Disposed after we claimed cancel: the cycle already finished; do not treat as our force-cancel.
-        if (Volatile.Read(ref _disposed) != 0)
+        if (Interlocked.CompareExchange(ref _lifecycleState, CancellingState, ActiveState) != ActiveState)
             return false;
 
         // Propagate to the workflow execution first (this typically marks the workflow as Cancelled and clears its
@@ -102,17 +107,97 @@ public sealed class ExecutionCycleHandle : IDisposable
         try { _cancelCallback?.Invoke(); }
         catch (Exception ex) when (!ex.IsFatal()) { /* Cancellation is best-effort; non-fatal failures here must not break the drain. */ }
 
-        try { _cycleCts.Cancel(); }
-        catch (ObjectDisposedException) { /* Race with Dispose — acceptable. */ }
+        PropagateCycleCtsCancellation();
 
-        return true;
+        // Publish cancellation only after its effects complete. Dispose can transition CancellingState directly to
+        // DisposedState, making this CAS fail when the cycle completed during the callback or CTS cancellation.
+        return Interlocked.CompareExchange(ref _lifecycleState, CancelledState, CancellingState) == CancellingState;
     }
 
-    /// <summary>Releases the linked CTS, notifies the registry, and signals <see cref="Disposed"/>.</summary>
+    /// <summary>
+    /// Logically releases the handle and notifies the registry. If cancellation callbacks are in flight, this method
+    /// may return before physical cleanup of the linked CTS completes; <see cref="Disposed"/> is signaled afterwards.
+    /// </summary>
     public void Dispose()
     {
-        if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
+        while (true)
+        {
+            var state = Volatile.Read(ref _lifecycleState);
+            if (state == DisposedState) return;
+            if (Interlocked.CompareExchange(ref _lifecycleState, DisposedState, state) == state) break;
+        }
+
         _onDisposed?.Invoke(this);
+        RequestCycleCtsDisposal();
+    }
+
+    private void PropagateLinkedCancellation()
+    {
+        PropagateCycleCtsCancellation();
+    }
+
+    private void PropagateCycleCtsCancellation()
+    {
+        lock (_cycleCtsGate)
+        {
+            if (_cycleCtsDisposed)
+                return;
+
+            _cycleCtsCancellationInProgress++;
+        }
+
+        try
+        {
+            _cycleCts.Cancel();
+        }
+        catch (ObjectDisposedException)
+        {
+            // Dispose may have won before cancellation propagation started.
+        }
+        catch (Exception ex) when (!ex.IsFatal())
+        {
+            // CTS callbacks are best-effort; preserve the lifecycle transition even when one reports a non-fatal error.
+        }
+        finally
+        {
+            var dispose = false;
+            lock (_cycleCtsGate)
+            {
+                _cycleCtsCancellationInProgress--;
+                if (_cycleCtsDisposeRequested && _cycleCtsCancellationInProgress == 0 && !_cycleCtsDisposed)
+                {
+                    _cycleCtsDisposed = true;
+                    dispose = true;
+                }
+            }
+
+            if (dispose)
+                DisposeCycleCts();
+        }
+    }
+
+    private void RequestCycleCtsDisposal()
+    {
+        var dispose = false;
+        lock (_cycleCtsGate)
+        {
+            _cycleCtsDisposeRequested = true;
+            if (_cycleCtsCancellationInProgress == 0 && !_cycleCtsDisposed)
+            {
+                _cycleCtsDisposed = true;
+                dispose = true;
+            }
+        }
+
+        if (dispose)
+            DisposeCycleCts();
+    }
+
+    private void DisposeCycleCts()
+    {
+        // CancellationTokenRegistration.Dispose is self-unregister-safe when this is called from the linked
+        // token callback, and waits for a callback running on another thread before releasing the registration.
+        _linkedTokenRegistration.Dispose();
         _cycleCts.Dispose();
         _disposedTcs.TrySetResult();
     }

src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs

@@ -326,6 +326,8 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
         // no persisted user-cancel exists, but we only promote if Phase A actually
         // cancels that live handle (DeadlineBreachPersistsInterrupted).
         var drainInducedInstanceIds = new HashSet<string>(StringComparer.Ordinal);
+        var drainInducedCandidateIds = new ConcurrentDictionary<Guid, string>();
+        var activeSnapshotHandleIds = new ConcurrentDictionary<Guid, byte>();
         var missingPersistedRowIds = new ConcurrentBag<string>();
         using var snapshotGate = new SemaphoreSlim(MaxConcurrentPreCancelSnapshotFinds);
         var snapshotTasks = live.Select(async handle =>
@@ -351,7 +353,12 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
                     }
 
                     if (snapshot.SubStatus != WorkflowSubStatus.Cancelled)
+                    {
+                        drainInducedCandidateIds.TryAdd(handle.Id, handle.WorkflowInstanceId);
+                        if (snapshot.IsExecuting)
+                            activeSnapshotHandleIds.TryAdd(handle.Id, 0);
                         return handle.WorkflowInstanceId;
+                    }
                 }
                 finally
                 {
@@ -369,11 +376,7 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
             return null;
         });
 
-        foreach (var instanceId in await Task.WhenAll(snapshotTasks).ConfigureAwait(false))
-        {
-            if (instanceId is not null)
-                drainInducedInstanceIds.Add(instanceId);
-        }
+        await Task.WhenAll(snapshotTasks).ConfigureAwait(false);
 
         // Force-cancel proceeds in three phases. The split exists because cancelling and
         // awaiting in the same loop made every execution cycle after the first run at full speed
@@ -384,6 +387,10 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
         // cheap and we want every runner to observe cancellation simultaneously rather
         // than serialized behind preceding settle waits.
         var cancelledInstanceIds = new HashSet<string>(StringComparer.Ordinal);
+        var cancelledHandles = new List<ExecutionCycleHandle>(live.Count);
+        var handlesToPersist = new List<ExecutionCycleHandle>(live.Count);
+        var activeSnapshotHandlesToRecover = new List<ExecutionCycleHandle>(live.Count);
+        var disposedActiveSnapshotHandleIds = new HashSet<Guid>();
         foreach (var handle in live)
         {
             try
@@ -392,9 +399,15 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
                 // (cycle finished during snapshot); treating that as drain-induced would rewrite a
                 // later Finished/Cancelled row the runner already committed.
                 if (!handle.TryCancel())
+                {
+                    if (activeSnapshotHandleIds.ContainsKey(handle.Id))
+                        activeSnapshotHandlesToRecover.Add(handle);
                     continue;
+                }
 
                 totalCancelled++;
+                cancelledHandles.Add(handle);
+                handlesToPersist.Add(handle);
                 cancelledInstanceIds.Add(handle.WorkflowInstanceId);
                 if (reportedIds.Count < cap) reportedIds.Add(handle.WorkflowInstanceId);
             }
@@ -404,9 +417,17 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
             }
         }
 
+        var recoveryHandleIds = activeSnapshotHandlesToRecover.Select(handle => handle.Id).ToHashSet();
+
         // A live handle we ourselves cancelled whose snapshot found no row is drain-induced:
         // there was no persisted user-cancel to preserve. Timeout/error and disposed no-ops stay excluded.
         // Do not use reportedIds here — that list is capped by MaxForceCancelledInstanceIdsReported.
+        foreach (var handle in cancelledHandles)
+        {
+            if (drainInducedCandidateIds.TryGetValue(handle.Id, out var instanceId))
+                drainInducedInstanceIds.Add(instanceId);
+        }
+
         foreach (var instanceId in missingPersistedRowIds)
         {
             if (cancelledInstanceIds.Contains(instanceId))
@@ -420,23 +441,41 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
         // bound the wait so a non-cancellable activity cannot block drain, accepting the
         // runner-clobber race for that one instance (the recovery scan picks it up).
         // Total wall time for this phase is at most ForceCancelSettleTimeout regardless
-        // of N.
+        // of N. Retained active-snapshot recovery candidates use an independent token so
+        // a canceled drain still observes their deferred disposal within that same bound.
         var settleTasks = live.Select(async handle =>
         {
             try
             {
-                await handle.Disposed.WaitAsync(ForceCancelSettleTimeout, cancellationToken).ConfigureAwait(false);
+                var settleCancellationToken = recoveryHandleIds.Contains(handle.Id) ? CancellationToken.None : cancellationToken;
+                await handle.Disposed.WaitAsync(ForceCancelSettleTimeout, settleCancellationToken).ConfigureAwait(false);
             }
             catch (TimeoutException)
             {
-                _logger.LogWarning("Execution cycle {ExecutionCycleId} (instance={InstanceId}) did not settle within {Timeout}; persisting Interrupted now (the runner may overwrite — recovery scan picks it up).", handle.Id, handle.WorkflowInstanceId, ForceCancelSettleTimeout);
+                if (recoveryHandleIds.Contains(handle.Id))
+                {
+                    _logger.LogWarning("Retained active-snapshot recovery candidate {ExecutionCycleId} (instance={InstanceId}) did not dispose within {Timeout}; Interrupted recovery persistence is skipped unless it settles before the recovery check.", handle.Id, handle.WorkflowInstanceId, ForceCancelSettleTimeout);
+                }
+                else
+                {
+                    _logger.LogWarning("Execution cycle {ExecutionCycleId} (instance={InstanceId}) did not settle within {Timeout}; persisting Interrupted now (the runner may overwrite — recovery scan picks it up).", handle.Id, handle.WorkflowInstanceId, ForceCancelSettleTimeout);
+                }
             }
             catch (OperationCanceledException) { /* drain CT fired — proceed to persist anyway */ }
         });
         await Task.WhenAll(settleTasks).ConfigureAwait(false);
 
-        // Phase C — persist Interrupted for every handle. Sequential to keep DbContext
-        // usage single-threaded; per-handle persistence is small.
+        foreach (var handle in activeSnapshotHandlesToRecover)
+        {
+            if (!handle.Disposed.IsCompleted)
+                continue;
+
+            disposedActiveSnapshotHandleIds.Add(handle.Id);
+            handlesToPersist.Add(handle);
+        }
+
+        // Phase C — persist Interrupted for every cancelled handle and disposed active checkpoint.
+        // Sequential to keep DbContext usage single-threaded; per-handle persistence is small.
         //
         // Each persist runs under its own bounded token that is NOT linked to the drain CT.
         // Phase B's catch on OperationCanceledException explicitly comments "drain CT fired —
@@ -445,12 +484,12 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
         // cancelled token and throw. Result: every execution cycle would be left in an unrecovered
         // executing state on host shutdown. The bounded non-drain token preserves the
         // forensic write while preventing a stuck DB from hanging shutdown indefinitely.
-        foreach (var handle in live)
+        foreach (var handle in handlesToPersist)
         {
             try
             {
                 using var persistCts = new CancellationTokenSource(PersistInterruptedTimeout);
-                await PersistInterruptedAsync(instanceStore, logStore, handle, generationId, reason, drainInducedInstanceIds, persistCts.Token);
+                await PersistInterruptedAsync(instanceStore, logStore, handle, generationId, reason, drainInducedInstanceIds, disposedActiveSnapshotHandleIds.Contains(handle.Id), persistCts.Token);
             }
             catch (Exception ex) when (!ex.IsFatal())
             {
@@ -468,10 +507,16 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
         string generationId,
         string reason,
         HashSet<string> drainInducedInstanceIds,
+        bool requireExecuting,
         CancellationToken cancellationToken)
     {
         var instance = await instanceStore.FindAsync(new WorkflowInstanceFilter { Id = handle.WorkflowInstanceId }, cancellationToken);
 
+        // A disposed checkpoint is only recoverable while its row still shows execution; a later
+        // natural suspension or completion must remain untouched.
+        if (requireExecuting && (instance is null || instance.Status == WorkflowStatus.Finished || !instance.IsExecuting))
+            return;
+
         if (instance is null)
         {
             // The instance row was never persisted (e.g., a execution cycle whose runner never reached commitStateHandler).

test/unit/Elsa.Common.UnitTests/ExceptionExtensionsTests.cs

@@ -49,6 +49,26 @@ public class ExceptionExtensionsTests
         Assert.True(outer.IsFatal());
     }
 
+    [Fact(DisplayName = "AggregateException containing a fatal cause is fatal")]
+    public void AggregateWrappingFatalIsFatal()
+    {
+        var outer = new AggregateException(
+            new InvalidOperationException("recoverable"),
+            new AggregateException(new OutOfMemoryException("fatal")));
+
+        Assert.True(outer.IsFatal());
+    }
+
+    [Fact(DisplayName = "AggregateException containing only recoverable causes is not fatal")]
+    public void AggregateWrappingRecoverableIsNotFatal()
+    {
+        var outer = new AggregateException(
+            new InvalidOperationException("recoverable"),
+            new TimeoutException("recoverable"));
+
+        Assert.False(outer.IsFatal());
+    }
+
     [Fact(DisplayName = "TargetInvocationException wrapping a recoverable cause is NOT fatal")]
     public void WrappedRecoverableIsNotFatal()
     {

test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs

@@ -50,6 +50,22 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase
         await LogStore.Received(1).AddAsync(Arg.Is<Entities.WorkflowExecutionLogRecord>(r => r.EventName == WorkflowInterruptedPayload.WorkflowInterruptedEventName), Arg.Any<CancellationToken>());
     }
 
+    [Fact(DisplayName = "Force drain propagates fatal CTS callback exceptions wrapped in an aggregate")]
+    public async Task ForceDrainPropagatesFatalCtsCallbackExceptions()
+    {
+        using var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-fatal-callback", ingressSourceName: null, startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
+        using var registration = handle.CancellationToken.Register(() => throw new OutOfMemoryException("fatal callback failure"));
+        ExecutionCycleRegistry.ActiveCount.Returns(1);
+        ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle });
+        InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
+            .Returns(_ => new ValueTask<WorkflowInstance?>(RunningInstance("instance-fatal-callback")));
+
+        var sut = BuildSut();
+        var exception = await Assert.ThrowsAsync<AggregateException>(() => sut.DrainAsync(DrainTrigger.OperatorForce).AsTask());
+
+        Assert.Contains(exception.Flatten().InnerExceptions, inner => inner is OutOfMemoryException);
+    }
+
     [Fact(DisplayName = "Persistence failure during drain produces Reason=PersistenceFailure in payload")]
     public async Task PersistenceFailureRecordsReason()
     {
@@ -219,6 +235,282 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase
         await LogStore.DidNotReceive().AddAsync(Arg.Any<Entities.WorkflowExecutionLogRecord>(), Arg.Any<CancellationToken>());
     }
 
+    [Fact(DisplayName = "A disposed handle with an executing snapshot is persisted as Interrupted")]
+    public async Task DisposedHandleWithExecutingSnapshotIsPersisted()
+    {
+        var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-disposed-at-checkpoint", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
+        handle.Dispose();
+        ExecutionCycleRegistry.ActiveCount.Returns(1);
+        ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle });
+        InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
+            .Returns(_ => new ValueTask<WorkflowInstance?>(RunningInstance("instance-disposed-at-checkpoint")));
+
+        var sut = BuildSut();
+        var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce);
+
+        Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+        Assert.Equal(0, outcome.ExecutionCyclesForceCancelledCount);
+        await InstanceStore.Received(1).SaveAsync(
+            Arg.Is<WorkflowInstance>(i => i.Id == "instance-disposed-at-checkpoint" && i.SubStatus == WorkflowSubStatus.Interrupted && !i.IsExecuting),
+            Arg.Any<CancellationToken>());
+    }
+
+    [Fact(DisplayName = "A disposed handle with an active snapshot is not persisted after the row suspends")]
+    public async Task DisposedHandleWithActiveSnapshotDoesNotPersistLaterSuspendedInstance()
+    {
+        var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-suspended-after-snapshot", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
+        handle.Dispose();
+        ExecutionCycleRegistry.ActiveCount.Returns(1);
+        ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle });
+
+        var finds = 0;
+        InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
+            .Returns(_ => Interlocked.Increment(ref finds) == 1
+                ? new ValueTask<WorkflowInstance?>(RunningInstance("instance-suspended-after-snapshot"))
+                : new ValueTask<WorkflowInstance?>(SuspendedInstance("instance-suspended-after-snapshot")));
+
+        var sut = BuildSut();
+        var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce);
+
+        Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+        Assert.Equal(0, outcome.ExecutionCyclesForceCancelledCount);
+        await InstanceStore.DidNotReceive().SaveAsync(Arg.Any<WorkflowInstance>(), Arg.Any<CancellationToken>());
+        await LogStore.DidNotReceive().AddAsync(Arg.Any<Entities.WorkflowExecutionLogRecord>(), Arg.Any<CancellationToken>());
+    }
+
+    [Fact(DisplayName = "A disposed handle with a suspended snapshot is not persisted as Interrupted")]
+    public async Task DisposedHandleWithSuspendedSnapshotIsNotPersisted()
+    {
+        var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-suspended", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
+        handle.Dispose();
+        ExecutionCycleRegistry.ActiveCount.Returns(1);
+        ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle });
+        InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
+            .Returns(_ => new ValueTask<WorkflowInstance?>(SuspendedInstance("instance-suspended")));
+
+        var sut = BuildSut();
+        var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce);
+
+        Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+        Assert.Equal(0, outcome.ExecutionCyclesForceCancelledCount);
+        await InstanceStore.DidNotReceive().SaveAsync(Arg.Any<WorkflowInstance>(), Arg.Any<CancellationToken>());
+    }
+
+    [Fact(DisplayName = "An active snapshot disposed after a failed cancel is recovered after settling")]
+    public async Task ActiveSnapshotDisposedAfterFailedCancelIsRecoveredAfterSettling()
+    {
+        var targetCallbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseTargetCallback = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        using var target = new ExecutionCycleHandle(
+            Guid.NewGuid(),
+            "instance-disposed-during-settle",
+            ingressSourceName: "http.trigger",
+            startedAt: DateTimeOffset.UtcNow,
+            linkedToken: CancellationToken.None,
+            cancelCallback: () =>
+            {
+                targetCallbackEntered.SetResult();
+                releaseTargetCallback.Task.GetAwaiter().GetResult();
+            });
+        var blockerCallbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseBlockerCallback = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        using var blocker = new ExecutionCycleHandle(
+            Guid.NewGuid(),
+            "instance-phase-a-blocker",
+            ingressSourceName: "http.trigger",
+            startedAt: DateTimeOffset.UtcNow,
+            linkedToken: CancellationToken.None,
+            cancelCallback: () =>
+            {
+                blockerCallbackEntered.SetResult();
+                releaseBlockerCallback.Task.GetAwaiter().GetResult();
+            });
+        var preCancelTask = Task.Run(target.TryCancel);
+        Task<DrainOutcome>? drainTask = null;
+
+        try
+        {
+            await targetCallbackEntered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+            ExecutionCycleRegistry.ActiveCount.Returns(2);
+            ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { target, blocker });
+            InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
+                .Returns(ci => new ValueTask<WorkflowInstance?>(RunningInstance(ci.Arg<WorkflowInstanceFilter>().Id!)));
+
+            var sut = BuildSut();
+            drainTask = Task.Run(async () => await sut.DrainAsync(DrainTrigger.OperatorForce));
+            await blockerCallbackEntered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+            Assert.False(target.Disposed.IsCompleted);
+
+            releaseBlockerCallback.TrySetResult();
+            await Task.Yield();
+            target.Dispose();
+            releaseTargetCallback.TrySetResult();
+
+            Assert.False(await preCancelTask.WaitAsync(TimeSpan.FromSeconds(5)));
+            var outcome = await drainTask.WaitAsync(TimeSpan.FromSeconds(5));
+
+            Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+            Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount);
+            await InstanceStore.Received(1).SaveAsync(
+                Arg.Is<WorkflowInstance>(i => i.Id == target.WorkflowInstanceId && i.SubStatus == WorkflowSubStatus.Interrupted && !i.IsExecuting),
+                Arg.Any<CancellationToken>());
+        }
+        finally
+        {
+            // Always release synchronous callback gates so an assertion or timeout cannot strand the test host.
+            releaseBlockerCallback.TrySetResult();
+            releaseTargetCallback.TrySetResult();
+
+            await ObserveCleanupAsync(preCancelTask);
+
+            if (drainTask is not null)
+                await ObserveCleanupAsync(drainTask);
+        }
+    }
+
+    [Fact(DisplayName = "A canceled drain still observes deferred disposal of a failed-cancel candidate")]
+    public async Task CanceledDrainObservesDeferredCandidateDisposal()
+    {
+        using var drainCts = new CancellationTokenSource();
+        var disposalEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseDisposal = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        using var handle = new ExecutionCycleHandle(
+            Guid.NewGuid(),
+            "instance-canceled-drain-disposal",
+            ingressSourceName: "http.trigger",
+            startedAt: DateTimeOffset.UtcNow,
+            linkedToken: CancellationToken.None,
+            onDisposed: _ =>
+            {
+                disposalEntered.SetResult();
+                releaseDisposal.Task.GetAwaiter().GetResult();
+            });
+        var disposeTask = Task.Run(handle.Dispose);
+        var blockerCallbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseBlockerCallback = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        using var blocker = new ExecutionCycleHandle(
+            Guid.NewGuid(),
+            "instance-canceled-drain-blocker",
+            ingressSourceName: "http.trigger",
+            startedAt: DateTimeOffset.UtcNow,
+            linkedToken: CancellationToken.None,
+            cancelCallback: () =>
+            {
+                blockerCallbackEntered.SetResult();
+                releaseBlockerCallback.Task.GetAwaiter().GetResult();
+            });
+        ExecutionCycleRegistry.ActiveCount.Returns(2);
+        ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle, blocker });
+        InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
+            .Returns(ci => new ValueTask<WorkflowInstance?>(RunningInstance(ci.Arg<WorkflowInstanceFilter>().Id!)));
+
+        Task<DrainOutcome>? drainTask = null;
+        try
+        {
+            await disposalEntered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+            var sut = BuildSut();
+            // Force-cancel invokes synchronous callbacks in Phase A. Keep the test thread available to
+            // release the blocker and cancel the drain while that callback is intentionally suspended.
+            drainTask = Task.Run(async () => await sut.DrainAsync(DrainTrigger.OperatorForce, drainCts.Token));
+            await blockerCallbackEntered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+            await drainCts.CancelAsync();
+            releaseBlockerCallback.TrySetResult();
+
+            await Assert.ThrowsAsync<TimeoutException>(() => drainTask.WaitAsync(TimeSpan.FromSeconds(1)));
+            releaseDisposal.TrySetResult();
+
+            var outcome = await drainTask.WaitAsync(TimeSpan.FromSeconds(5));
+            await disposeTask.WaitAsync(TimeSpan.FromSeconds(5));
+
+            Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+            Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount);
+            Assert.DoesNotContain(handle.WorkflowInstanceId, outcome.ForceCancelledInstanceIds);
+            await InstanceStore.Received(1).SaveAsync(
+                Arg.Is<WorkflowInstance>(i => i.Id == handle.WorkflowInstanceId && i.SubStatus == WorkflowSubStatus.Interrupted && !i.IsExecuting),
+                Arg.Any<CancellationToken>());
+        }
+        finally
+        {
+            releaseBlockerCallback.TrySetResult();
+            releaseDisposal.TrySetResult();
+            await ObserveCleanupAsync(disposeTask);
+
+            if (drainTask is not null)
+                await ObserveCleanupAsync(drainTask);
+        }
+    }
+
+    [Fact(DisplayName = "A disposed handle is not persisted as Interrupted when its later row is still running")]
+    public async Task DisposedHandleDoesNotPersistLaterRunningInstance()
+    {
+        var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-already-running", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
+        handle.Dispose();
+        ExecutionCycleRegistry.ActiveCount.Returns(1);
+        ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle });
+
+        var finds = 0;
+        InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
+            .Returns(_ =>
+            {
+                if (Interlocked.Increment(ref finds) == 1)
+                    return new ValueTask<WorkflowInstance?>((WorkflowInstance?)null);
+
+                return new ValueTask<WorkflowInstance?>(RunningInstance("instance-already-running"));
+            });
+
+        var sut = BuildSut();
+        var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce);
+
+        Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+        Assert.Equal(0, outcome.ExecutionCyclesForceCancelledCount);
+        await InstanceStore.DidNotReceive().SaveAsync(Arg.Any<WorkflowInstance>(), Arg.Any<CancellationToken>());
+        await LogStore.DidNotReceive().AddAsync(Arg.Any<Entities.WorkflowExecutionLogRecord>(), Arg.Any<CancellationToken>());
+    }
+
+    [Fact(DisplayName = "A cycle disposed during cancellation is not counted or persisted as drain-cancelled")]
+    public async Task CycleDisposedDuringCancellationIsNotCountedOrPersisted()
+    {
+        var callbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseCallback = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var handle = new ExecutionCycleHandle(
+            Guid.NewGuid(),
+            "instance-completed-during-cancel",
+            ingressSourceName: "http.trigger",
+            startedAt: DateTimeOffset.UtcNow,
+            linkedToken: CancellationToken.None,
+            cancelCallback: () =>
+            {
+                callbackEntered.SetResult();
+                releaseCallback.Task.GetAwaiter().GetResult();
+            });
+        ExecutionCycleRegistry.ActiveCount.Returns(1);
+        ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle });
+
+        var finds = 0;
+        InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
+            .Returns(_ =>
+            {
+                if (Interlocked.Increment(ref finds) == 1)
+                    return new ValueTask<WorkflowInstance?>((WorkflowInstance?)null);
+
+                return new ValueTask<WorkflowInstance?>(RunningInstance("instance-completed-during-cancel"));
+            });
+
+        var sut = BuildSut();
+        var drainTask = Task.Run(async () => await sut.DrainAsync(DrainTrigger.OperatorForce));
+        await callbackEntered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        handle.Dispose();
+        releaseCallback.SetResult();
+
+        var outcome = await drainTask.WaitAsync(TimeSpan.FromSeconds(5));
+        Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+        Assert.Equal(0, outcome.ExecutionCyclesForceCancelledCount);
+        await InstanceStore.DidNotReceive().SaveAsync(Arg.Any<WorkflowInstance>(), Arg.Any<CancellationToken>());
+        await LogStore.DidNotReceive().AddAsync(Arg.Any<Entities.WorkflowExecutionLogRecord>(), Arg.Any<CancellationToken>());
+    }
+
     [Fact(DisplayName = "Waiting for a snapshot slot does not burn the per-Find 250ms budget")]
     public async Task SnapshotQueueWaitDoesNotExcludeLaterFinds()
     {
@@ -287,6 +579,33 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase
         IsExecuting = true,
     };
 
+    private static async Task ObserveCleanupAsync(Task task)
+    {
+        try
+        {
+            await task.WaitAsync(TimeSpan.FromSeconds(5));
+        }
+        catch (TimeoutException)
+        {
+            // Preserve the original assertion/timeout while observing the cleanup task.
+        }
+        catch (OperationCanceledException)
+        {
+            // Preserve the original assertion/timeout while observing the cleanup task.
+        }
+    }
+
+    private static WorkflowInstance SuspendedInstance(string id) => new()
+    {
+        Id = id,
+        DefinitionId = "def-1",
+        DefinitionVersionId = "ver-1",
+        Version = 1,
+        Status = WorkflowStatus.Running,
+        SubStatus = WorkflowSubStatus.Suspended,
+        IsExecuting = false,
+    };
+
     private static WorkflowInstance CancelledInstance(string id) => new()
     {
         Id = id,

test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/ExecutionCycleRegistryTests.cs

@@ -118,6 +118,123 @@ public class ExecutionCycleRegistryTests
         Assert.False(disposed.TryCancel());
     }
 
+    [Fact(DisplayName = "ExecutionCycleHandle.TryCancel reports false when disposed during the cancellation callback")]
+    public async Task TryCancelReportsFalseWhenDisposedDuringCancellationCallback()
+    {
+        var sut = new ExecutionCycleRegistry(_sources, _clock);
+        var callbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseCallback = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var handle = sut.BeginCycle(
+            "instance-1",
+            ingressSourceName: null,
+            linkedToken: CancellationToken.None,
+            cancelCallback: () =>
+            {
+                callbackEntered.SetResult();
+                releaseCallback.Task.GetAwaiter().GetResult();
+            });
+
+        var cancelTask = Task.Run(handle.TryCancel);
+        await callbackEntered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        handle.Dispose();
+        releaseCallback.SetResult();
+
+        Assert.False(await cancelTask.WaitAsync(TimeSpan.FromSeconds(5)));
+    }
+
+    [Fact(DisplayName = "ExecutionCycleHandle.Dispose completes while a CTS callback waits for it")]
+    public async Task DisposeCompletesWhileCtsCallbackWaitsForIt()
+    {
+        var cancellationCallbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var disposalCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var callbackObservedDisposal = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var handle = new ExecutionCycleHandle(
+            Guid.NewGuid(),
+            "instance-1",
+            ingressSourceName: null,
+            startedAt: DateTimeOffset.UtcNow,
+            linkedToken: CancellationToken.None,
+            onDisposed: null);
+        using var registration = handle.CancellationToken.Register(() =>
+        {
+            cancellationCallbackEntered.SetResult();
+            callbackObservedDisposal.SetResult(disposalCompleted.Task.Wait(TimeSpan.FromSeconds(5)));
+        });
+
+        var cancelTask = Task.Run(handle.TryCancel);
+        await cancellationCallbackEntered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        var disposeTask = Task.Run(() =>
+        {
+            handle.Dispose();
+            disposalCompleted.TrySetResult();
+        });
+
+        Assert.True(await callbackObservedDisposal.Task.WaitAsync(TimeSpan.FromSeconds(5)));
+        await disposeTask.WaitAsync(TimeSpan.FromSeconds(5));
+        Assert.False(await cancelTask.WaitAsync(TimeSpan.FromSeconds(5)));
+        Assert.True(handle.Disposed.IsCompletedSuccessfully);
+    }
+
+    [Fact(DisplayName = "ExecutionCycleHandle defers CTS disposal while linked-token cancellation is in progress")]
+    public async Task DisposeCompletesWhileLinkedTokenCancellationWaitsForIt()
+    {
+        using var linkedCts = new CancellationTokenSource();
+        var cancellationCallbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var disposalCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var callbackObservedDisposal = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        using var handle = new ExecutionCycleHandle(
+            Guid.NewGuid(),
+            "instance-linked-cancellation",
+            ingressSourceName: null,
+            startedAt: DateTimeOffset.UtcNow,
+            linkedToken: linkedCts.Token);
+        var cycleToken = handle.CancellationToken;
+        using var registration = cycleToken.Register(() =>
+        {
+            cancellationCallbackEntered.SetResult();
+            callbackObservedDisposal.SetResult(disposalCompleted.Task.Wait(TimeSpan.FromSeconds(5)));
+        });
+
+        var cancelTask = Task.Run(() => linkedCts.Cancel());
+        await cancellationCallbackEntered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        var disposeTask = Task.Run(() =>
+        {
+            handle.Dispose();
+            disposalCompleted.TrySetResult();
+        });
+
+        Assert.True(await callbackObservedDisposal.Task.WaitAsync(TimeSpan.FromSeconds(5)));
+        await disposeTask.WaitAsync(TimeSpan.FromSeconds(5));
+        await cancelTask.WaitAsync(TimeSpan.FromSeconds(5));
+        Assert.True(handle.Disposed.IsCompletedSuccessfully);
+        Assert.True(cycleToken.IsCancellationRequested);
+    }
+
+    [Fact(DisplayName = "ExecutionCycleHandle defers CTS disposal from a cancellation callback")]
+    public void DisposeDefersCtsDisposalUntilCancellationPropagationExits()
+    {
+        var disposedDuringCancellation = false;
+        var handle = new ExecutionCycleHandle(
+            Guid.NewGuid(),
+            "instance-1",
+            ingressSourceName: null,
+            startedAt: DateTimeOffset.UtcNow,
+            linkedToken: CancellationToken.None);
+
+        using var registration = handle.CancellationToken.Register(() =>
+        {
+            handle.Dispose();
+            disposedDuringCancellation = handle.Disposed.IsCompleted;
+        });
+
+        Assert.False(handle.TryCancel());
+        Assert.False(disposedDuringCancellation);
+        Assert.True(handle.Disposed.IsCompletedSuccessfully);
+    }
+
     [Fact(DisplayName = "ExecutionCycleHandle.Cancel invokes the cancel callback supplied at registration")]
     public void CancelCallbackIsInvoked()
     {
@@ -133,7 +250,7 @@ public class ExecutionCycleRegistryTests
         Assert.Equal(1, callbackInvocations);
 
         // Truly idempotent: a second Cancel() before Dispose() must NOT re-invoke the callback. The handle uses an
-        // Interlocked _cancelled flag so callers can't accidentally trigger non-idempotent cancellation side effects.
+        // Interlocked lifecycle state so callers can't accidentally trigger non-idempotent cancellation side effects.
         handle.Cancel();
         Assert.Equal(1, callbackInvocations);
 
@@ -158,6 +275,42 @@ public class ExecutionCycleRegistryTests
         Assert.True(handle.CancellationToken.IsCancellationRequested);
     }
 
+    [Fact(DisplayName = "ExecutionCycleHandle.TryCancel swallows non-fatal CTS callback exceptions")]
+    public void TryCancelSwallowsNonFatalCtsCallbackExceptions()
+    {
+        var handle = new ExecutionCycleHandle(
+            Guid.NewGuid(),
+            "instance-1",
+            ingressSourceName: null,
+            startedAt: DateTimeOffset.UtcNow,
+            linkedToken: CancellationToken.None);
+        using var registration = handle.CancellationToken.Register(() => throw new InvalidOperationException("callback refused to cancel"));
+
+        Assert.True(handle.TryCancel());
+        Assert.False(handle.TryCancel());
+
+        handle.Dispose();
+        Assert.True(handle.Disposed.IsCompletedSuccessfully);
+    }
+
+    [Fact(DisplayName = "ExecutionCycleHandle.TryCancel propagates fatal CTS callback exceptions wrapped in an aggregate")]
+    public void TryCancelPropagatesFatalCtsCallbackExceptions()
+    {
+        var handle = new ExecutionCycleHandle(
+            Guid.NewGuid(),
+            "instance-1",
+            ingressSourceName: null,
+            startedAt: DateTimeOffset.UtcNow,
+            linkedToken: CancellationToken.None);
+        using var registration = handle.CancellationToken.Register(() => throw new OutOfMemoryException("fatal callback failure"));
+
+        var exception = Assert.Throws<AggregateException>(() => handle.TryCancel());
+
+        Assert.Contains(exception.Flatten().InnerExceptions, inner => inner is OutOfMemoryException);
+        handle.Dispose();
+        Assert.True(handle.Disposed.IsCompletedSuccessfully);
+    }
+
     [Fact(DisplayName = "ExecutionCycleHandle.Disposed completes when the handle is disposed")]
     public async Task DisposedTaskCompletesOnDispose()
     {