diff --git a/src/modules/Elsa.Workflows.Runtime/Models/ExecutionCycleHandle.cs b/src/modules/Elsa.Workflows.Runtime/Models/ExecutionCycleHandle.cs index 6e5aa8b43..887254218 100644 --- a/src/modules/Elsa.Workflows.Runtime/Models/ExecutionCycleHandle.cs +++ b/src/modules/Elsa.Workflows.Runtime/Models/ExecutionCycleHandle.cs @@ -74,13 +74,27 @@ public sealed class ExecutionCycleHandle : IDisposable /// Invokes the cancel callback (when supplied at construction) to propagate cancellation into the workflow /// execution, then cancels the cycle's own linked CTS. Safe to call multiple times; idempotent. /// - public void Cancel() + public void Cancel() => TryCancel(); + + /// + /// 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. + /// + public bool TryCancel() { - if (_disposed != 0) return; + 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; + 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) + return false; // Propagate to the workflow execution first (this typically marks the workflow as Cancelled and clears its // schedule, so the runner stops scheduling new activities). The orchestrator's subsequent Interrupted @@ -90,6 +104,8 @@ public sealed class ExecutionCycleHandle : IDisposable try { _cycleCts.Cancel(); } catch (ObjectDisposedException) { /* Race with Dispose — acceptable. */ } + + return true; } /// Releases the linked CTS, notifies the registry, and signals . diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs b/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs index ae36ba0c2..b7be3f78b 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs @@ -1,6 +1,8 @@ +using System.Collections.Concurrent; using System.Diagnostics; using Elsa.Common; using Elsa.Workflows.Management; +using Elsa.Workflows.Management.Entities; using Elsa.Workflows.Management.Filters; using Elsa.Workflows.Runtime.Options; using Microsoft.Extensions.DependencyInjection; @@ -282,6 +284,22 @@ public sealed class DrainOrchestrator : IDrainOrchestrator /// private static readonly TimeSpan PersistInterruptedTimeout = TimeSpan.FromSeconds(5); + /// + /// Short shutdown budget for the pre-cancel instance snapshot. Force-cancel runs after the drain + /// deadline, so remaining deadline is already zero. WaitAsync unblocks even if the store ignores + /// its cancellation token. The budget starts when a snapshot slot is acquired — not while + /// waiting for . + /// + private static readonly TimeSpan PreCancelSnapshotTimeout = TimeSpan.FromMilliseconds(250); + + /// + /// Cap on concurrent pre-cancel snapshot Finds. Unbounded Task.WhenAll self-contends + /// under large live-cycle N: more 250ms timeouts, more drainInduced excludes, more + /// Interrupted misses. Queue wait uses the drain token only; each Find still gets its own + /// 250ms after it acquires a slot. + /// + internal const int MaxConcurrentPreCancelSnapshotFinds = 16; + private async Task<(int Count, IReadOnlyList Ids)> ForceCancelActiveCyclesAsync(DrainTrigger trigger, string generationId, CancellationToken cancellationToken) { var cap = _options.Value.MaxForceCancelledInstanceIdsReported; @@ -299,6 +317,64 @@ public sealed class DrainOrchestrator : IDrainOrchestrator ? WorkflowInterruptedPayload.ReasonOperatorForce : WorkflowInterruptedPayload.ReasonDeadlineBreach; + // Snapshot before Phase A. Each Find has its own 250ms budget and its own + // DI scope so concurrent reads do not share an EF DbContext (Phase C stays + // sequential on the outer scope for the same reason). Finds run concurrently + // under a semaphore so a large live-cycle set cannot self-contend into + // 250ms timeouts. Timeout/error: exclude that id (prefer preserving + // user-cancel / #8052). A successful null Find is not drain-induced yet — + // 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 missingPersistedRowIds = new ConcurrentBag(); + using var snapshotGate = new SemaphoreSlim(MaxConcurrentPreCancelSnapshotFinds); + var snapshotTasks = live.Select(async handle => + { + try + { + await snapshotGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + using var findScope = _scopeFactory.CreateScope(); + var findStore = findScope.ServiceProvider.GetRequiredService(); + using var perFindCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + perFindCts.CancelAfter(PreCancelSnapshotTimeout); + var snapshot = await findStore + .FindAsync(new WorkflowInstanceFilter { Id = handle.WorkflowInstanceId }, perFindCts.Token) + .AsTask() + .WaitAsync(perFindCts.Token) + .ConfigureAwait(false); + if (snapshot is null) + { + missingPersistedRowIds.Add(handle.WorkflowInstanceId); + return null; + } + + if (snapshot.SubStatus != WorkflowSubStatus.Cancelled) + return handle.WorkflowInstanceId; + } + finally + { + // Release after the await budget, not after the store call returns. + // A cancel-ignoring Find can outlive this slot (documented secondary; + // holding the slot until it completes would stall WhenAll / Phase A). + snapshotGate.Release(); + } + } + catch (Exception ex) when (!ex.IsFatal()) + { + _logger.LogWarning(ex, "Pre-cancel snapshot for instance {InstanceId} timed out or failed; excluding from drain-induced promote.", handle.WorkflowInstanceId); + } + + return null; + }); + + foreach (var instanceId in await Task.WhenAll(snapshotTasks).ConfigureAwait(false)) + { + if (instanceId is not null) + drainInducedInstanceIds.Add(instanceId); + } + // 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 // through the prior execution cycle's settle window — total wall time was O(N × settle @@ -307,12 +383,19 @@ public sealed class DrainOrchestrator : IDrainOrchestrator // Phase A — cancel every handle synchronously. CancellationTokenSource.Cancel is // cheap and we want every runner to observe cancellation simultaneously rather // than serialized behind preceding settle waits. + var cancelledInstanceIds = new HashSet(StringComparer.Ordinal); foreach (var handle in live) { try { - handle.Cancel(); + // Only a real transition counts. Cancel() is a no-op on an already-disposed handle + // (cycle finished during snapshot); treating that as drain-induced would rewrite a + // later Finished/Cancelled row the runner already committed. + if (!handle.TryCancel()) + continue; + totalCancelled++; + cancelledInstanceIds.Add(handle.WorkflowInstanceId); if (reportedIds.Count < cap) reportedIds.Add(handle.WorkflowInstanceId); } catch (Exception ex) when (!ex.IsFatal()) @@ -321,6 +404,15 @@ public sealed class DrainOrchestrator : IDrainOrchestrator } } + // 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 instanceId in missingPersistedRowIds) + { + if (cancelledInstanceIds.Contains(instanceId)) + drainInducedInstanceIds.Add(instanceId); + } + // Phase B — wait for every runner to settle in parallel under a shared deadline. // The handle disposes when ExecutionCycleTrackingMiddleware exits its `using` block, which // is after the workflow runner has finished its commit. We want runners' terminal @@ -358,7 +450,7 @@ public sealed class DrainOrchestrator : IDrainOrchestrator try { using var persistCts = new CancellationTokenSource(PersistInterruptedTimeout); - await PersistInterruptedAsync(instanceStore, logStore, handle, generationId, reason, persistCts.Token); + await PersistInterruptedAsync(instanceStore, logStore, handle, generationId, reason, drainInducedInstanceIds, persistCts.Token); } catch (Exception ex) when (!ex.IsFatal()) { @@ -375,6 +467,7 @@ public sealed class DrainOrchestrator : IDrainOrchestrator ExecutionCycleHandle handle, string generationId, string reason, + HashSet drainInducedInstanceIds, CancellationToken cancellationToken) { var instance = await instanceStore.FindAsync(new WorkflowInstanceFilter { Id = handle.WorkflowInstanceId }, cancellationToken); @@ -420,6 +513,14 @@ public sealed class DrainOrchestrator : IDrainOrchestrator return; } + if (ShouldSkipInterruptedPersist(instance, drainInducedInstanceIds)) + { + _logger.LogInformation( + "Skipping Interrupted persist for instance {InstanceId}: already in terminal status {Status}/{SubStatus}.", + instance.Id, instance.Status, instance.SubStatus); + return; + } + var actualReason = reason; try @@ -459,4 +560,21 @@ public sealed class DrainOrchestrator : IDrainOrchestrator .Select(s => new IngressSourceFinalState(s.Name, s.State, s.LastError, WasForceStopped: s.State == IngressSourceState.PauseFailed && s.LastError is not null)) .ToArray(); } + + /// + /// Skip persist when the row is already a real terminal outcome: natural completion, + /// fault, already Cancelled, or unknown pre-state (snapshot timeout/error). Ids that + /// snapshot showed were not Cancelled, or that had no row and that we ourselves + /// force-cancelled, may be promoted. + /// + private static bool ShouldSkipInterruptedPersist(WorkflowInstance instance, HashSet drainInducedInstanceIds) + { + if (instance.Status != WorkflowStatus.Finished) + return false; + + if (instance.SubStatus == WorkflowSubStatus.Cancelled) + return !drainInducedInstanceIds.Contains(instance.Id); + + return true; + } } diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs index 54b63636e..ddb163fdf 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs @@ -1,6 +1,7 @@ using Elsa.Workflows.Management.Entities; using Elsa.Workflows.Management.Filters; using Elsa.Workflows.Runtime.HostedServices; +using Elsa.Workflows.Runtime.Services; using Microsoft.Extensions.Logging; using NSubstitute; @@ -78,6 +79,225 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase Arg.Any()); } + [Fact(DisplayName = "Pre-cancel snapshot Finds never exceed MaxConcurrentPreCancelSnapshotFinds in flight")] + public async Task SnapshotFindsAreCapped() + { + const int extra = 8; + var cap = DrainOrchestrator.MaxConcurrentPreCancelSnapshotFinds; + var count = cap + extra; + var handles = Enumerable.Range(0, count) + .Select(i => new ExecutionCycleHandle(Guid.NewGuid(), $"instance-capped-{i}", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None)) + .ToArray(); + ExecutionCycleRegistry.ActiveCount.Returns(count); + ExecutionCycleRegistry.ListActiveCycles().Returns(handles); + + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var capReached = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var seen = new HashSet(StringComparer.Ordinal); + var inFlight = 0; + var maxInFlight = 0; + var maxLock = new object(); + + InstanceStore.FindAsync(Arg.Any(), Arg.Any()) + .Returns(ci => + { + var id = ci.Arg().Id!; + lock (seen) + { + if (!seen.Add(id)) + return new ValueTask(RunningInstance(id)); + } + + return new ValueTask(WaitForSnapshotSlotAsync(id)); + }); + + async Task WaitForSnapshotSlotAsync(string id) + { + var current = Interlocked.Increment(ref inFlight); + lock (maxLock) + { + if (current > maxInFlight) + maxInFlight = current; + if (maxInFlight >= cap) + capReached.TrySetResult(); + } + + try + { + await release.Task; + return RunningInstance(id); + } + finally + { + Interlocked.Decrement(ref inFlight); + } + } + + var sut = BuildSut(); + var drainTask = sut.DrainAsync(DrainTrigger.OperatorForce).AsTask(); + + await capReached.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await Task.Delay(50); + Assert.Equal(cap, maxInFlight); + + release.TrySetResult(); + + var outcome = await drainTask.WaitAsync(TimeSpan.FromSeconds(8)); + Assert.Equal(DrainResult.Forced, outcome.OverallResult); + Assert.Equal(count, outcome.ExecutionCyclesForceCancelledCount); + foreach (var handle in handles) + await InstanceStore.Received().SaveAsync(Arg.Is(i => i.Id == handle.WorkflowInstanceId && i.SubStatus == WorkflowSubStatus.Interrupted), Arg.Any()); + } + + [Fact(DisplayName = "A null pre-cancel snapshot still promotes a later drain-induced Finished/Cancelled row")] + public async Task NullSnapshotPromotesLaterCancelledInstance() + { + var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-missing", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None); + 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(CancelledInstance("instance-missing")); + }); + + var sut = BuildSut(); + var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce); + + Assert.Equal(DrainResult.Forced, outcome.OverallResult); + Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount); + await InstanceStore.Received().SaveAsync(Arg.Is(i => i.Id == "instance-missing" && i.SubStatus == WorkflowSubStatus.Interrupted), Arg.Any()); + } + + [Fact(DisplayName = "A confirmed Cancelled pre-cancel snapshot does not promote a later Finished/Cancelled row")] + public async Task CancelledSnapshotDoesNotPromoteLaterCancelledInstance() + { + var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-user-cancel", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None); + ExecutionCycleRegistry.ActiveCount.Returns(1); + ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle }); + InstanceStore.FindAsync(Arg.Any(), Arg.Any()) + .Returns(_ => new ValueTask(CancelledInstance("instance-user-cancel"))); + + var sut = BuildSut(); + var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce); + + Assert.Equal(DrainResult.Forced, outcome.OverallResult); + Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount); + await InstanceStore.DidNotReceive().SaveAsync(Arg.Any(), Arg.Any()); + await LogStore.DidNotReceive().AddAsync(Arg.Any(), Arg.Any()); + } + + [Fact(DisplayName = "A null snapshot of an already-disposed handle does not promote a later Finished/Cancelled row")] + public async Task DisposedHandleNullSnapshotDoesNotPromoteLaterCancelledInstance() + { + var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-already-done", 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(CancelledInstance("instance-already-done")); + }); + + 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 = "Waiting for a snapshot slot does not burn the per-Find 250ms budget")] + public async Task SnapshotQueueWaitDoesNotExcludeLaterFinds() + { + var cap = DrainOrchestrator.MaxConcurrentPreCancelSnapshotFinds; + const int extra = 4; + var count = cap + extra; + var handles = Enumerable.Range(0, count) + .Select(i => new ExecutionCycleHandle(Guid.NewGuid(), $"instance-queued-{i}", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None)) + .ToArray(); + ExecutionCycleRegistry.ActiveCount.Returns(count); + ExecutionCycleRegistry.ListActiveCycles().Returns(handles); + + var hang = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var seen = new HashSet(StringComparer.Ordinal); + var stalledIds = new HashSet(StringComparer.Ordinal); + var promotedIds = new HashSet(StringComparer.Ordinal); + var snapshotFinds = 0; + + InstanceStore.FindAsync(Arg.Any(), Arg.Any()) + .Returns(ci => + { + var id = ci.Arg().Id!; + lock (seen) + { + // Phase C: runner committed Finished/Cancelled after force-cancel. + // Only ids that snapshot successfully join drainInduced and may promote. + if (!seen.Add(id)) + return new ValueTask(CancelledInstance(id)); + } + + if (Interlocked.Increment(ref snapshotFinds) <= cap) + { + lock (stalledIds) + stalledIds.Add(id); + return new ValueTask(hang.Task); + } + + lock (promotedIds) + promotedIds.Add(id); + return new ValueTask(RunningInstance(id)); + }); + + var sut = BuildSut(); + var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce).AsTask().WaitAsync(TimeSpan.FromSeconds(8)); + + Assert.Equal(DrainResult.Forced, outcome.OverallResult); + Assert.Equal(count, outcome.ExecutionCyclesForceCancelledCount); + Assert.Equal(cap, stalledIds.Count); + Assert.Equal(extra, promotedIds.Count); + Assert.False(hang.Task.IsCompleted); + + foreach (var id in stalledIds) + await InstanceStore.DidNotReceive().SaveAsync(Arg.Is(i => i.Id == id), Arg.Any()); + foreach (var id in promotedIds) + await InstanceStore.Received().SaveAsync(Arg.Is(i => i.Id == id && i.SubStatus == WorkflowSubStatus.Interrupted), Arg.Any()); + } + + private static WorkflowInstance RunningInstance(string id) => new() + { + Id = id, + DefinitionId = "def-1", + DefinitionVersionId = "ver-1", + Version = 1, + Status = WorkflowStatus.Running, + SubStatus = WorkflowSubStatus.Executing, + IsExecuting = true, + }; + + private static WorkflowInstance CancelledInstance(string id) => new() + { + Id = id, + DefinitionId = "def-1", + DefinitionVersionId = "ver-1", + Version = 1, + Status = WorkflowStatus.Finished, + SubStatus = WorkflowSubStatus.Cancelled, + IsExecuting = false, + }; + [Fact(DisplayName = "Second non-force drain in same generation throws InvalidOperationException")] public async Task SecondNonForceDrainThrows() { diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/ExecutionCycleRegistryTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/ExecutionCycleRegistryTests.cs index bf3b58946..1c39dbe4b 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/ExecutionCycleRegistryTests.cs +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/ExecutionCycleRegistryTests.cs @@ -101,6 +101,23 @@ public class ExecutionCycleRegistryTests Assert.True(handle.CancellationToken.IsCancellationRequested); } + [Fact(DisplayName = "ExecutionCycleHandle.TryCancel reports whether this call transitioned the handle")] + public void ExecutionCycleHandleTryCancelReportsTransition() + { + var sut = new ExecutionCycleRegistry(_sources, _clock); + var handle = sut.BeginCycle("instance-1", null, CancellationToken.None); + + Assert.True(handle.TryCancel()); + Assert.False(handle.TryCancel()); + + handle.Dispose(); + Assert.False(handle.TryCancel()); + + var disposed = sut.BeginCycle("instance-2", null, CancellationToken.None); + disposed.Dispose(); + Assert.False(disposed.TryCancel()); + } + [Fact(DisplayName = "ExecutionCycleHandle.Cancel invokes the cancel callback supplied at registration")] public void CancelCallbackIsInvoked() {