diff --git a/src/modules/Elsa.Persistence.EFCore/Modules/Management/WorkflowInstanceStore.cs b/src/modules/Elsa.Persistence.EFCore/Modules/Management/WorkflowInstanceStore.cs
index 76359d8c7..42648e5c3 100644
--- a/src/modules/Elsa.Persistence.EFCore/Modules/Management/WorkflowInstanceStore.cs
+++ b/src/modules/Elsa.Persistence.EFCore/Modules/Management/WorkflowInstanceStore.cs
@@ -194,13 +194,14 @@ public class EFCoreWorkflowInstanceStore : IWorkflowInstanceStore
}
///
- public async ValueTask TryMarkInterruptedAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
+ public async ValueTask TryMarkInterruptedAsync(string workflowInstanceId, CancellationToken cancellationToken = default, bool allowFinishedCancelled = false)
{
await using var dbContext = await _store.CreateDbContextAsync(cancellationToken);
var updated = await dbContext.WorkflowInstances
- .Where(x => x.Id == workflowInstanceId && x.Status != WorkflowStatus.Finished)
+ .Where(x => x.Id == workflowInstanceId && (x.Status != WorkflowStatus.Finished || (allowFinishedCancelled && x.SubStatus == WorkflowSubStatus.Cancelled)))
.ExecuteUpdateAsync(
setters => setters
+ .SetProperty(x => x.Status, WorkflowStatus.Running)
.SetProperty(x => x.SubStatus, WorkflowSubStatus.Interrupted)
.SetProperty(x => x.IsExecuting, false),
cancellationToken);
diff --git a/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceStore.cs b/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceStore.cs
index 05a678a5a..f335d25ea 100644
--- a/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceStore.cs
+++ b/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceStore.cs
@@ -181,11 +181,23 @@ public interface IWorkflowInstanceStore
///
/// Sets to and
- /// to false only if the stored instance is still non-terminal.
+ /// to false only if the stored instance is still
+ /// .
///
+ /// The workflow instance to mark.
+ /// The cancellation token.
+ ///
+ /// Drain-only. When true, also accepts /
+ /// and promotes it to
+ /// + .
+ /// Default callers must leave this false so every
+ /// row is refused. Still refuses and
+ /// .
+ ///
///
/// true when the interrupt markers were applied; false when the instance is missing or already
- /// . Implementations must not overwrite a concurrent terminal commit.
+ /// (unless applies).
+ /// Implementations must not overwrite a concurrent naturally completed commit.
///
- ValueTask TryMarkInterruptedAsync(string workflowInstanceId, CancellationToken cancellationToken = default);
+ ValueTask TryMarkInterruptedAsync(string workflowInstanceId, CancellationToken cancellationToken = default, bool allowFinishedCancelled = false);
}
\ No newline at end of file
diff --git a/src/modules/Elsa.Workflows.Management/Stores/MemoryWorkflowInstanceStore.cs b/src/modules/Elsa.Workflows.Management/Stores/MemoryWorkflowInstanceStore.cs
index f40648a78..c892de0df 100644
--- a/src/modules/Elsa.Workflows.Management/Stores/MemoryWorkflowInstanceStore.cs
+++ b/src/modules/Elsa.Workflows.Management/Stores/MemoryWorkflowInstanceStore.cs
@@ -174,16 +174,23 @@ public class MemoryWorkflowInstanceStore : IWorkflowInstanceStore
}
///
- public ValueTask TryMarkInterruptedAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
+ public ValueTask TryMarkInterruptedAsync(string workflowInstanceId, CancellationToken cancellationToken = default, bool allowFinishedCancelled = false)
{
// Same lock as Save/Update so a runner's terminal persist cannot land between the
// non-terminal check and the Interrupted mutations.
lock (_sync)
{
var instance = _store.Find(x => x.Id == workflowInstanceId);
- if (instance is null || instance.Status == WorkflowStatus.Finished)
+ if (instance is null)
return ValueTask.FromResult(false);
+ if (instance.Status == WorkflowStatus.Finished)
+ {
+ if (!allowFinishedCancelled || instance.SubStatus != WorkflowSubStatus.Cancelled)
+ return ValueTask.FromResult(false);
+ }
+
+ instance.Status = WorkflowStatus.Running;
instance.SubStatus = WorkflowSubStatus.Interrupted;
instance.IsExecuting = false;
diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs b/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs
index 1a7b7a217..c66cb616f 100644
--- a/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs
+++ b/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs
@@ -1,6 +1,7 @@
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 +283,13 @@ 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.
+ ///
+ private static readonly TimeSpan PreCancelSnapshotTimeout = TimeSpan.FromMilliseconds(250);
+
private async Task<(int Count, IReadOnlyList Ids)> ForceCancelActiveCyclesAsync(DrainTrigger trigger, string generationId, CancellationToken cancellationToken)
{
var cap = _options.Value.MaxForceCancelledInstanceIdsReported;
@@ -299,6 +307,43 @@ 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
+ // so Phase A Cancel waits ~one per-find timeout, not N×timeout. Timeout/error:
+ // exclude that id (prefer preserving user-cancel / #8052 over promoting an
+ // unknown row).
+ var drainInducedInstanceIds = new HashSet(StringComparer.Ordinal);
+ var snapshotTasks = live.Select(async handle =>
+ {
+ 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 || snapshot.SubStatus != WorkflowSubStatus.Cancelled)
+ return handle.WorkflowInstanceId;
+ }
+ 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
@@ -323,12 +368,12 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
// 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
- // commits to land before we overwrite the sub-status with Interrupted — but we
- // bound the wait so a non-cancellable activity cannot block drain. PersistInterruptedAsync
- // applies Interrupted only via a store-level compare-and-set that refuses already-terminal
- // rows, so a late Finished commit is not overwritten (the recovery scan only requeues
- // Running+Interrupted).
+ // is after the workflow runner has finished its commit. We want that commit to land first
+ // so PersistInterruptedAsync can overwrite drain-induced Finished/Cancelled (the runner's
+ // reaction to force-cancel) with Running+Interrupted. User-cancelled rows snapshotted
+ // before Phase A, and naturally completed rows (Finished/Finished or Finished/Faulted),
+ // are refused so they are not requeued. Bound the wait so a non-cancellable activity
+ // cannot block drain.
// Total wall time for this phase is at most ForceCancelSettleTimeout regardless
// of N.
var settleTasks = live.Select(async handle =>
@@ -360,7 +405,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())
{
@@ -377,6 +422,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);
@@ -422,7 +468,7 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
return;
}
- if (instance.Status == WorkflowStatus.Finished)
+ if (ShouldSkipInterruptedPersist(instance, drainInducedInstanceIds))
{
_logger.LogInformation(
"Skipping Interrupted persist for instance {InstanceId}: already in terminal status {Status}/{SubStatus}.",
@@ -434,10 +480,13 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
try
{
- // Conditional write: do not SaveAsync the Find snapshot. A runner can commit Finished
- // between the read and a full-entity save, which would revert Status to Running and
- // let startup recovery requeue a completed instance.
- var marked = await instanceStore.TryMarkInterruptedAsync(instance.Id, cancellationToken);
+ // Conditional write: do not SaveAsync the Find snapshot. Default TryMark refuses
+ // every Finished row (#8052). Drain alone may set allowFinishedCancelled when
+ // this id is in the force-cancelled set and the runner committed Cancelled.
+ var allowFinishedCancelled = instance.Status == WorkflowStatus.Finished
+ && instance.SubStatus == WorkflowSubStatus.Cancelled
+ && drainInducedInstanceIds.Contains(instance.Id);
+ var marked = await instanceStore.TryMarkInterruptedAsync(instance.Id, cancellationToken, allowFinishedCancelled);
if (!marked)
{
_logger.LogInformation(
@@ -477,4 +526,20 @@ 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). Only ids that
+ /// snapshot clearly showed were not already 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/integration/Elsa.AI.IntegrationTests/AIRuntimeGroundingToolTests.cs b/test/integration/Elsa.AI.IntegrationTests/AIRuntimeGroundingToolTests.cs
index b566df328..3a9bcdd4c 100644
--- a/test/integration/Elsa.AI.IntegrationTests/AIRuntimeGroundingToolTests.cs
+++ b/test/integration/Elsa.AI.IntegrationTests/AIRuntimeGroundingToolTests.cs
@@ -116,7 +116,7 @@ public class AIRuntimeGroundingToolTests
public ValueTask DeleteAsync(WorkflowInstanceFilter filter, CancellationToken cancellationToken = default) => ValueTask.FromResult(0L);
public Task UpdateUpdatedTimestampAsync(string workflowInstanceId, DateTimeOffset value, CancellationToken cancellationToken = default) => Task.CompletedTask;
- public ValueTask TryMarkInterruptedAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
+ public ValueTask TryMarkInterruptedAsync(string workflowInstanceId, CancellationToken cancellationToken = default, bool allowFinishedCancelled = false)
{
var instance = _instances.FirstOrDefault(x => x.Id == workflowInstanceId);
if (instance is null || instance.Status == WorkflowStatus.Finished)
diff --git a/test/integration/Elsa.Workflows.IntegrationTests/GracefulShutdown/DeadlineBreachEndToEndTests.cs b/test/integration/Elsa.Workflows.IntegrationTests/GracefulShutdown/DeadlineBreachEndToEndTests.cs
index 7d012f56e..2c3676bfd 100644
--- a/test/integration/Elsa.Workflows.IntegrationTests/GracefulShutdown/DeadlineBreachEndToEndTests.cs
+++ b/test/integration/Elsa.Workflows.IntegrationTests/GracefulShutdown/DeadlineBreachEndToEndTests.cs
@@ -84,6 +84,7 @@ public class DeadlineBreachEndToEndTests
CancellationToken.None)).ToList();
Assert.NotEmpty(interruptedInstances);
+ Assert.Equal(WorkflowStatus.Running, interruptedInstances[0].Status);
Assert.False(interruptedInstances[0].IsExecuting,
"An Interrupted instance must have IsExecuting=false so the existing timeout-based crash recovery does not also pick it up.");
diff --git a/test/integration/Elsa.Workflows.IntegrationTests/GracefulShutdown/UserCancelDuringDrainTests.cs b/test/integration/Elsa.Workflows.IntegrationTests/GracefulShutdown/UserCancelDuringDrainTests.cs
new file mode 100644
index 000000000..c308bc75a
--- /dev/null
+++ b/test/integration/Elsa.Workflows.IntegrationTests/GracefulShutdown/UserCancelDuringDrainTests.cs
@@ -0,0 +1,121 @@
+using Elsa.Common;
+using Elsa.Common.Models;
+using Elsa.Extensions;
+using Elsa.Testing.Shared;
+using Elsa.Workflows.Management;
+using Elsa.Workflows.Management.Filters;
+using Elsa.Workflows.Models;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Messages;
+using Microsoft.Extensions.DependencyInjection;
+using Xunit.Abstractions;
+
+namespace Elsa.Workflows.IntegrationTests.GracefulShutdown;
+
+///
+/// A user cancellation that still has its original execution cycle live when drain snapshots
+/// must stay and must not be requeued.
+///
+public class UserCancelDuringDrainTests
+{
+ private readonly IServiceProvider _services;
+ private readonly IWorkflowRuntime _workflowRuntime;
+ private readonly IDrainOrchestrator _orchestrator;
+
+ public UserCancelDuringDrainTests(ITestOutputHelper testOutputHelper)
+ {
+ _services = new TestApplicationBuilder(testOutputHelper)
+ .AddActivitiesFrom()
+ .AddWorkflow()
+ .ConfigureElsa(elsa => elsa
+ .UseWorkflowRuntime(runtime => runtime.ConfigureGracefulShutdown(o =>
+ {
+ o.DrainDeadline = TimeSpan.FromMilliseconds(50);
+ o.IngressPauseTimeout = TimeSpan.FromMilliseconds(50);
+ })))
+ .Build();
+ _workflowRuntime = _services.GetRequiredService();
+ _orchestrator = _services.GetRequiredService();
+ }
+
+ [Fact(DisplayName = "Normal cancellation with a still-active execution cycle is not promoted to Interrupted or requeued after drain")]
+ public async Task NormalCancelThenDrainDoesNotRequeue()
+ {
+ await _services.PopulateRegistriesAsync();
+
+ var client = await _workflowRuntime.CreateClientAsync();
+ await client.CreateInstanceAsync(new CreateWorkflowInstanceRequest
+ {
+ WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(nameof(LongRunningObservableWorkflow), VersionOptions.Published)
+ });
+
+ var cycles = _services.GetRequiredService();
+ var runTask = Task.Run(() => client.RunInstanceAsync(RunWorkflowInstanceRequest.Empty));
+ await WaitUntilAsync(() => cycles.ActiveCount > 0, TimeSpan.FromSeconds(2));
+
+ await client.CancelAsync();
+
+ using var scope = _services.CreateScope();
+ var instanceStore = scope.ServiceProvider.GetRequiredService();
+ await WaitUntilAsync(async () =>
+ {
+ var current = await instanceStore.FindAsync(new WorkflowInstanceFilter { Id = client.WorkflowInstanceId });
+ return current is { Status: WorkflowStatus.Finished, SubStatus: WorkflowSubStatus.Cancelled };
+ }, TimeSpan.FromSeconds(2));
+
+ Assert.True(cycles.ActiveCount > 0, "The original execution cycle must still be active when drain starts.");
+
+ await _orchestrator.DrainAsync(DrainTrigger.HostStopSignal);
+
+ try { await runTask.WaitAsync(TimeSpan.FromSeconds(5)); }
+ catch (Exception ex) when (!ex.IsFatal()) { /* runner may complete normally or surface OCE */ }
+
+ var instance = await instanceStore.FindAsync(new WorkflowInstanceFilter { Id = client.WorkflowInstanceId });
+ Assert.NotNull(instance);
+ Assert.Equal(WorkflowStatus.Finished, instance.Status);
+ Assert.Equal(WorkflowSubStatus.Cancelled, instance.SubStatus);
+
+ var restarter = new RecordingRestarter();
+ var scanner = ActivatorUtilities.CreateInstance(scope.ServiceProvider, restarter);
+ var requeued = await scanner.ScanAndRequeueAsync(CancellationToken.None);
+
+ Assert.Equal(0, requeued);
+ Assert.Empty(restarter.RestartedIds);
+ }
+
+ private static async Task WaitUntilAsync(Func> condition, TimeSpan timeout)
+ {
+ var deadline = DateTime.UtcNow + timeout;
+ while (DateTime.UtcNow < deadline)
+ {
+ if (await condition())
+ return;
+ await Task.Delay(20);
+ }
+
+ throw new TimeoutException($"Condition was not met within {timeout}.");
+ }
+
+ private static Task WaitUntilAsync(Func condition, TimeSpan timeout) =>
+ WaitUntilAsync(() => Task.FromResult(condition()), timeout);
+
+ private sealed class RecordingRestarter : IWorkflowRestarter
+ {
+ public List RestartedIds { get; } = new();
+
+ public Task RestartWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
+ {
+ RestartedIds.Add(workflowInstanceId);
+ return Task.CompletedTask;
+ }
+ }
+}
+
+/// Published definition used by .
+public class LongRunningObservableWorkflow : WorkflowBase
+{
+ protected override void Build(IWorkflowBuilder builder)
+ {
+ builder.Root = new ObservableActivity { DelayMs = 2000 };
+ }
+}
diff --git a/test/unit/Elsa.Workflows.Management.UnitTests/Stores/MemoryWorkflowInstanceStoreTests.cs b/test/unit/Elsa.Workflows.Management.UnitTests/Stores/MemoryWorkflowInstanceStoreTests.cs
index 271f04a87..df8765c99 100644
--- a/test/unit/Elsa.Workflows.Management.UnitTests/Stores/MemoryWorkflowInstanceStoreTests.cs
+++ b/test/unit/Elsa.Workflows.Management.UnitTests/Stores/MemoryWorkflowInstanceStoreTests.cs
@@ -31,6 +31,76 @@ public class MemoryWorkflowInstanceStoreTests
Assert.False(instance.IsExecuting);
}
+ [Fact(DisplayName = "TryMarkInterruptedAsync refuses Finished/Cancelled unless allowFinishedCancelled is set")]
+ public async Task TryMarkInterrupted_DoesNotOverwriteCancelledInstanceByDefault()
+ {
+ var store = CreateStore(new WorkflowInstance
+ {
+ Id = "cancelled-1",
+ DefinitionId = "def-1",
+ DefinitionVersionId = "ver-1",
+ Version = 1,
+ Status = WorkflowStatus.Finished,
+ SubStatus = WorkflowSubStatus.Cancelled,
+ IsExecuting = false,
+ });
+
+ var marked = await store.TryMarkInterruptedAsync("cancelled-1");
+
+ Assert.False(marked);
+ var instance = await store.FindAsync(new() { Id = "cancelled-1" });
+ Assert.NotNull(instance);
+ Assert.Equal(WorkflowStatus.Finished, instance.Status);
+ Assert.Equal(WorkflowSubStatus.Cancelled, instance.SubStatus);
+ }
+
+ [Fact(DisplayName = "TryMarkInterruptedAsync promotes Finished/Cancelled only when allowFinishedCancelled is true")]
+ public async Task TryMarkInterrupted_PromotesCancelledWhenAllowed()
+ {
+ var store = CreateStore(new WorkflowInstance
+ {
+ Id = "cancelled-1",
+ DefinitionId = "def-1",
+ DefinitionVersionId = "ver-1",
+ Version = 1,
+ Status = WorkflowStatus.Finished,
+ SubStatus = WorkflowSubStatus.Cancelled,
+ IsExecuting = false,
+ });
+
+ var marked = await store.TryMarkInterruptedAsync("cancelled-1", allowFinishedCancelled: true);
+
+ Assert.True(marked);
+ var instance = await store.FindAsync(new() { Id = "cancelled-1" });
+ Assert.NotNull(instance);
+ Assert.Equal(WorkflowStatus.Running, instance.Status);
+ Assert.Equal(WorkflowSubStatus.Interrupted, instance.SubStatus);
+ Assert.False(instance.IsExecuting);
+ }
+
+ [Fact(DisplayName = "TryMarkInterruptedAsync still refuses Finished/Finished when allowFinishedCancelled is true")]
+ public async Task TryMarkInterrupted_DoesNotOverwriteFinishedEvenWhenCancelledAllowed()
+ {
+ var store = CreateStore(new WorkflowInstance
+ {
+ Id = "finished-1",
+ DefinitionId = "def-1",
+ DefinitionVersionId = "ver-1",
+ Version = 1,
+ Status = WorkflowStatus.Finished,
+ SubStatus = WorkflowSubStatus.Finished,
+ IsExecuting = false,
+ });
+
+ var marked = await store.TryMarkInterruptedAsync("finished-1", allowFinishedCancelled: true);
+
+ Assert.False(marked);
+ var instance = await store.FindAsync(new() { Id = "finished-1" });
+ Assert.NotNull(instance);
+ Assert.Equal(WorkflowStatus.Finished, instance.Status);
+ Assert.Equal(WorkflowSubStatus.Finished, instance.SubStatus);
+ }
+
[Fact(DisplayName = "TryMarkInterruptedAsync does not overwrite a Finished instance")]
public async Task TryMarkInterrupted_DoesNotOverwriteFinishedInstance()
{
diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs
index 490282d00..c005612d9 100644
--- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs
+++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs
@@ -1,3 +1,4 @@
+using System.Diagnostics;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Management.Filters;
using Elsa.Workflows.Runtime.HostedServices;
@@ -37,7 +38,7 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase
Version = 1,
IsExecuting = true,
}));
- InstanceStore.TryMarkInterruptedAsync("instance-1", Arg.Any()).Returns(new ValueTask(true));
+ InstanceStore.TryMarkInterruptedAsync("instance-1", Arg.Any(), false).Returns(new ValueTask(true));
var sut = BuildSut();
var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce);
@@ -46,7 +47,7 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase
Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount);
Assert.Contains("instance-1", outcome.ForceCancelledInstanceIds);
Assert.True(handle.CancellationToken.IsCancellationRequested);
- await InstanceStore.Received(1).TryMarkInterruptedAsync("instance-1", Arg.Any());
+ await InstanceStore.Received(1).TryMarkInterruptedAsync("instance-1", Arg.Any(), false);
await InstanceStore.DidNotReceive().SaveAsync(Arg.Any(), Arg.Any());
await LogStore.Received(1).AddAsync(Arg.Is(r => r.EventName == WorkflowInterruptedPayload.WorkflowInterruptedEventName), Arg.Any());
}
@@ -76,10 +77,232 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase
Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount);
Assert.Contains("instance-finished", outcome.ForceCancelledInstanceIds);
await InstanceStore.DidNotReceive().SaveAsync(Arg.Any(), Arg.Any());
- await InstanceStore.DidNotReceive().TryMarkInterruptedAsync(Arg.Any(), Arg.Any());
+ await InstanceStore.DidNotReceive().TryMarkInterruptedAsync(Arg.Any(), Arg.Any(), Arg.Any());
await LogStore.DidNotReceive().AddAsync(Arg.Any(), Arg.Any());
}
+ [Fact(DisplayName = "Force-cancel persists Interrupted when the runner commits Finished/Cancelled after drain cancel")]
+ public async Task PersistsInterruptedForCancelledInstance()
+ {
+ var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-cancelled", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
+ ExecutionCycleRegistry.ActiveCount.Returns(1);
+ ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle });
+ var running = new WorkflowInstance
+ {
+ Id = "instance-cancelled",
+ DefinitionId = "def-1",
+ DefinitionVersionId = "ver-1",
+ Version = 1,
+ Status = WorkflowStatus.Running,
+ SubStatus = WorkflowSubStatus.Executing,
+ IsExecuting = true,
+ };
+ var cancelled = new WorkflowInstance
+ {
+ Id = "instance-cancelled",
+ DefinitionId = "def-1",
+ DefinitionVersionId = "ver-1",
+ Version = 1,
+ Status = WorkflowStatus.Finished,
+ SubStatus = WorkflowSubStatus.Cancelled,
+ IsExecuting = false,
+ };
+ InstanceStore.FindAsync(Arg.Any(), Arg.Any())
+ .Returns(_ => new ValueTask(running), _ => new ValueTask(cancelled));
+ InstanceStore.TryMarkInterruptedAsync("instance-cancelled", Arg.Any(), true).Returns(new ValueTask(true));
+
+ var sut = BuildSut();
+ var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce);
+
+ Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+ Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount);
+ await InstanceStore.Received(1).TryMarkInterruptedAsync("instance-cancelled", Arg.Any(), true);
+ await LogStore.Received(1).AddAsync(Arg.Is(r => r.EventName == WorkflowInterruptedPayload.WorkflowInterruptedEventName), Arg.Any());
+ }
+
+ [Fact(DisplayName = "Force-cancel does not promote a user cancellation that was already Cancelled when drain snapshotted the live cycle")]
+ public async Task SkipsInterruptedPersistForAlreadyUserCancelledInstance()
+ {
+ var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-user-cancelled", 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(new WorkflowInstance
+ {
+ Id = "instance-user-cancelled",
+ DefinitionId = "def-1",
+ DefinitionVersionId = "ver-1",
+ Version = 1,
+ Status = WorkflowStatus.Finished,
+ SubStatus = WorkflowSubStatus.Cancelled,
+ IsExecuting = false,
+ }));
+
+ var sut = BuildSut();
+ var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce);
+
+ Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+ Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount);
+ await InstanceStore.DidNotReceive().TryMarkInterruptedAsync(Arg.Any(), Arg.Any(), Arg.Any());
+ await LogStore.DidNotReceive().AddAsync(Arg.Any(), Arg.Any());
+ }
+
+ [Fact(DisplayName = "Stalled pre-cancel FindAsync does not prevent force-cancel or drain completion")]
+ public async Task StalledSnapshotDoesNotBlockForceCancel()
+ {
+ var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-stalled", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
+ ExecutionCycleRegistry.ActiveCount.Returns(1);
+ ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle });
+ var stalled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var finds = 0;
+ InstanceStore.FindAsync(Arg.Any(), Arg.Any())
+ .Returns(_ =>
+ {
+ if (Interlocked.Increment(ref finds) == 1)
+ return new ValueTask(stalled.Task);
+
+ return new ValueTask(new WorkflowInstance
+ {
+ Id = "instance-stalled",
+ DefinitionId = "def-1",
+ DefinitionVersionId = "ver-1",
+ Version = 1,
+ Status = WorkflowStatus.Finished,
+ SubStatus = WorkflowSubStatus.Cancelled,
+ IsExecuting = false,
+ });
+ });
+
+ var sut = BuildSut();
+ var drainTask = sut.DrainAsync(DrainTrigger.OperatorForce).AsTask();
+
+ var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(1);
+ while (!handle.CancellationToken.IsCancellationRequested && DateTime.UtcNow < deadline)
+ await Task.Delay(20);
+
+ Assert.True(handle.CancellationToken.IsCancellationRequested, "Force-cancel must run even while the snapshot FindAsync is still stalled.");
+ Assert.False(stalled.Task.IsCompleted);
+
+ var outcome = await drainTask.WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+ Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount);
+ Assert.Contains("instance-stalled", outcome.ForceCancelledInstanceIds);
+ await InstanceStore.DidNotReceive().TryMarkInterruptedAsync(Arg.Any(), Arg.Any(), Arg.Any());
+ }
+
+ [Fact(DisplayName = "A stalled first snapshot does not exclude a later instance from drain-induced Interrupted promote")]
+ public async Task IndependentSnapshotTimeoutDoesNotStarveLaterFinds()
+ {
+ var stalledHandle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-stalled", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
+ var recoveredHandle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-recovered", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
+ ExecutionCycleRegistry.ActiveCount.Returns(2);
+ ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { stalledHandle, recoveredHandle });
+
+ var stalled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var stalledFinds = 0;
+ InstanceStore.FindAsync(Arg.Is(f => f.Id == "instance-stalled"), Arg.Any())
+ .Returns(_ =>
+ {
+ if (Interlocked.Increment(ref stalledFinds) == 1)
+ return new ValueTask(stalled.Task);
+
+ return new ValueTask(new WorkflowInstance
+ {
+ Id = "instance-stalled",
+ DefinitionId = "def-1",
+ DefinitionVersionId = "ver-1",
+ Version = 1,
+ Status = WorkflowStatus.Finished,
+ SubStatus = WorkflowSubStatus.Cancelled,
+ IsExecuting = false,
+ });
+ });
+ InstanceStore.FindAsync(Arg.Is(f => f.Id == "instance-recovered"), Arg.Any())
+ .Returns(_ => new ValueTask(new WorkflowInstance
+ {
+ Id = "instance-recovered",
+ DefinitionId = "def-1",
+ DefinitionVersionId = "ver-1",
+ Version = 1,
+ Status = WorkflowStatus.Running,
+ SubStatus = WorkflowSubStatus.Executing,
+ IsExecuting = true,
+ }));
+ InstanceStore.TryMarkInterruptedAsync("instance-recovered", Arg.Any(), false).Returns(new ValueTask(true));
+
+ var sut = BuildSut();
+ var drainTask = sut.DrainAsync(DrainTrigger.OperatorForce).AsTask();
+
+ var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(2);
+ while ((!stalledHandle.CancellationToken.IsCancellationRequested || !recoveredHandle.CancellationToken.IsCancellationRequested) && DateTime.UtcNow < deadline)
+ await Task.Delay(20);
+
+ Assert.True(stalledHandle.CancellationToken.IsCancellationRequested);
+ Assert.True(recoveredHandle.CancellationToken.IsCancellationRequested);
+ Assert.False(stalled.Task.IsCompleted);
+
+ var outcome = await drainTask.WaitAsync(TimeSpan.FromSeconds(8));
+ Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+ Assert.Equal(2, outcome.ExecutionCyclesForceCancelledCount);
+ await InstanceStore.DidNotReceive().TryMarkInterruptedAsync("instance-stalled", Arg.Any(), Arg.Any());
+ await InstanceStore.Received().TryMarkInterruptedAsync("instance-recovered", Arg.Any(), false);
+ }
+
+ [Fact(DisplayName = "Many stalled pre-cancel Finds do not serialize Phase A Cancel behind N snapshot timeouts")]
+ public async Task ParallelStalledSnapshotsDoNotSerializeForceCancel()
+ {
+ const int count = 8;
+ var handles = Enumerable.Range(0, count)
+ .Select(i => new ExecutionCycleHandle(Guid.NewGuid(), $"instance-stalled-{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);
+ InstanceStore.FindAsync(Arg.Any(), Arg.Any())
+ .Returns(ci =>
+ {
+ var id = ci.Arg().Id!;
+ lock (seen)
+ {
+ if (seen.Add(id))
+ return new ValueTask(hang.Task);
+ }
+
+ return new ValueTask(new WorkflowInstance
+ {
+ Id = id,
+ DefinitionId = "def-1",
+ DefinitionVersionId = "ver-1",
+ Version = 1,
+ Status = WorkflowStatus.Finished,
+ SubStatus = WorkflowSubStatus.Cancelled,
+ IsExecuting = false,
+ });
+ });
+
+ var sut = BuildSut();
+ var started = Stopwatch.StartNew();
+ var drainTask = sut.DrainAsync(DrainTrigger.OperatorForce).AsTask();
+
+ var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(1);
+ while (handles.Any(h => !h.CancellationToken.IsCancellationRequested) && DateTime.UtcNow < deadline)
+ await Task.Delay(10);
+
+ var cancelElapsed = started.Elapsed;
+ Assert.All(handles, h => Assert.True(h.CancellationToken.IsCancellationRequested));
+ Assert.True(
+ cancelElapsed < TimeSpan.FromSeconds(1),
+ $"Phase A Cancel waited {cancelElapsed}; concurrent snapshots must finish in ~one 250ms window, not {count}×250ms.");
+ Assert.False(hang.Task.IsCompleted);
+
+ var outcome = await drainTask.WaitAsync(TimeSpan.FromSeconds(8));
+ Assert.Equal(DrainResult.Forced, outcome.OverallResult);
+ Assert.Equal(count, outcome.ExecutionCyclesForceCancelledCount);
+ await InstanceStore.DidNotReceive().TryMarkInterruptedAsync(Arg.Any(), Arg.Any(), Arg.Any());
+ }
+
[Fact(DisplayName = "Force-cancel skips Interrupted persist when TryMarkInterrupted loses the terminal race")]
public async Task SkipsInterruptedPersistWhenMarkLosesTerminalRace()
{
@@ -96,13 +319,13 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase
Status = WorkflowStatus.Running,
IsExecuting = true,
}));
- InstanceStore.TryMarkInterruptedAsync("instance-raced", Arg.Any()).Returns(new ValueTask(false));
+ InstanceStore.TryMarkInterruptedAsync("instance-raced", Arg.Any(), false).Returns(new ValueTask(false));
var sut = BuildSut();
var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce);
Assert.Equal(DrainResult.Forced, outcome.OverallResult);
- await InstanceStore.Received(1).TryMarkInterruptedAsync("instance-raced", Arg.Any());
+ await InstanceStore.Received(1).TryMarkInterruptedAsync("instance-raced", Arg.Any(), false);
await InstanceStore.DidNotReceive().SaveAsync(Arg.Any(), Arg.Any());
await LogStore.DidNotReceive().AddAsync(Arg.Any(), Arg.Any());
}
@@ -122,7 +345,7 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase
Version = 1,
IsExecuting = true,
}));
- InstanceStore.TryMarkInterruptedAsync("instance-2", Arg.Any())
+ InstanceStore.TryMarkInterruptedAsync("instance-2", Arg.Any(), false)
.Returns(_ => ValueTask.FromException(new InvalidOperationException("db unavailable")));
var sut = BuildSut();