diff --git a/src/modules/Elsa.Common/Extensions/ExceptionExtensions.cs b/src/modules/Elsa.Common/Extensions/ExceptionExtensions.cs index 342201dde..cd44f1cdb 100644 --- a/src/modules/Elsa.Common/Extensions/ExceptionExtensions.cs +++ b/src/modules/Elsa.Common/Extensions/ExceptionExtensions.cs @@ -25,9 +25,10 @@ public static class ExceptionExtensions /// /// /// - /// Wrapper exceptions (, ) - /// are unwrapped before classification so that, for example, a - /// wrapping a is classified as fatal. + /// Wrapper exceptions (, , + /// ) are unwrapped before classification so that, + /// for example, an aggregate or wrapping a + /// is classified as fatal. /// /// /// Pattern: use as a filter on a generic catch 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: diff --git a/src/modules/Elsa.Workflows.Runtime/Models/ExecutionCycleHandle.cs b/src/modules/Elsa.Workflows.Runtime/Models/ExecutionCycleHandle.cs index 887254218..b7f1e1693 100644 --- a/src/modules/Elsa.Workflows.Runtime/Models/ExecutionCycleHandle.cs +++ b/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? _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; /// /// Creates a new handle. The owning supplies @@ -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; /// - /// Completes when runs — i.e., when the workflow runner finishes the cycle (cleanly or - /// via cancellation) and the middleware exits its using block. The drain orchestrator awaits this with - /// a timeout before persisting , ensuring its write happens AFTER - /// any commit the runner emits in response to . + /// Completes after 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 + /// using block. If cancellation callbacks are in flight, may return before this + /// cleanup completes. The drain orchestrator awaits this with a timeout before persisting + /// , ensuring its write happens AFTER any commit the runner emits in + /// response to . /// public Task Disposed => _disposedTcs.Task; @@ -78,22 +92,13 @@ public sealed class ExecutionCycleHandle : IDisposable /// /// Attempts to cancel the cycle. Returns true only when this call transitioned the handle from - /// not-cancelled to cancelled. Returns false 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 false 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. /// 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; } - /// Releases the linked CTS, notifies the registry, and signals . + /// + /// 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; is signaled afterwards. + /// 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(); } diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs b/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs index b7be3f78b..3865c3da5 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs +++ b/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(StringComparer.Ordinal); + var drainInducedCandidateIds = new ConcurrentDictionary(); + var activeSnapshotHandleIds = new ConcurrentDictionary(); var missingPersistedRowIds = new ConcurrentBag(); 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(StringComparer.Ordinal); + var cancelledHandles = new List(live.Count); + var handlesToPersist = new List(live.Count); + var activeSnapshotHandlesToRecover = new List(live.Count); + var disposedActiveSnapshotHandleIds = new HashSet(); 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 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). diff --git a/test/unit/Elsa.Common.UnitTests/ExceptionExtensionsTests.cs b/test/unit/Elsa.Common.UnitTests/ExceptionExtensionsTests.cs index 0947566c6..b8d539b6c 100644 --- a/test/unit/Elsa.Common.UnitTests/ExceptionExtensionsTests.cs +++ b/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() { diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs index ddb163fdf..1dd863929 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs @@ -50,6 +50,22 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase await LogStore.Received(1).AddAsync(Arg.Is(r => r.EventName == WorkflowInterruptedPayload.WorkflowInterruptedEventName), Arg.Any()); } + [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(), Arg.Any()) + .Returns(_ => new ValueTask(RunningInstance("instance-fatal-callback"))); + + var sut = BuildSut(); + var exception = await Assert.ThrowsAsync(() => 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(), Arg.Any()); } + [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(), Arg.Any()) + .Returns(_ => new ValueTask(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(i => i.Id == "instance-disposed-at-checkpoint" && i.SubStatus == WorkflowSubStatus.Interrupted && !i.IsExecuting), + Arg.Any()); + } + + [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(), Arg.Any()) + .Returns(_ => Interlocked.Increment(ref finds) == 1 + ? new ValueTask(RunningInstance("instance-suspended-after-snapshot")) + : new ValueTask(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(), Arg.Any()); + await LogStore.DidNotReceive().AddAsync(Arg.Any(), Arg.Any()); + } + + [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(), Arg.Any()) + .Returns(_ => new ValueTask(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(), Arg.Any()); + } + + [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? drainTask = null; + + try + { + await targetCallbackEntered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + ExecutionCycleRegistry.ActiveCount.Returns(2); + ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { target, blocker }); + InstanceStore.FindAsync(Arg.Any(), Arg.Any()) + .Returns(ci => new ValueTask(RunningInstance(ci.Arg().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(i => i.Id == target.WorkflowInstanceId && i.SubStatus == WorkflowSubStatus.Interrupted && !i.IsExecuting), + Arg.Any()); + } + 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(), Arg.Any()) + .Returns(ci => new ValueTask(RunningInstance(ci.Arg().Id!))); + + Task? 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(() => 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(i => i.Id == handle.WorkflowInstanceId && i.SubStatus == WorkflowSubStatus.Interrupted && !i.IsExecuting), + Arg.Any()); + } + 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(), Arg.Any()) + .Returns(_ => + { + if (Interlocked.Increment(ref finds) == 1) + return new ValueTask((WorkflowInstance?)null); + + return new ValueTask(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(), Arg.Any()); + await LogStore.DidNotReceive().AddAsync(Arg.Any(), Arg.Any()); + } + + [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(), Arg.Any()) + .Returns(_ => + { + if (Interlocked.Increment(ref finds) == 1) + return new ValueTask((WorkflowInstance?)null); + + return new ValueTask(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(), Arg.Any()); + await LogStore.DidNotReceive().AddAsync(Arg.Any(), Arg.Any()); + } + [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, diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/ExecutionCycleRegistryTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/ExecutionCycleRegistryTests.cs index 1c39dbe4b..b5acdb00a 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/ExecutionCycleRegistryTests.cs +++ b/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(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(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(() => 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() {