diff --git a/src/modules/Elsa.Persistence.EFCore/Modules/Management/WorkflowInstanceStore.cs b/src/modules/Elsa.Persistence.EFCore/Modules/Management/WorkflowInstanceStore.cs index 4ed4d0e47..76359d8c7 100644 --- a/src/modules/Elsa.Persistence.EFCore/Modules/Management/WorkflowInstanceStore.cs +++ b/src/modules/Elsa.Persistence.EFCore/Modules/Management/WorkflowInstanceStore.cs @@ -193,6 +193,21 @@ public class EFCoreWorkflowInstanceStore : IWorkflowInstanceStore } } + /// + public async ValueTask TryMarkInterruptedAsync(string workflowInstanceId, CancellationToken cancellationToken = default) + { + await using var dbContext = await _store.CreateDbContextAsync(cancellationToken); + var updated = await dbContext.WorkflowInstances + .Where(x => x.Id == workflowInstanceId && x.Status != WorkflowStatus.Finished) + .ExecuteUpdateAsync( + setters => setters + .SetProperty(x => x.SubStatus, WorkflowSubStatus.Interrupted) + .SetProperty(x => x.IsExecuting, false), + cancellationToken); + + return updated > 0; + } + /// [RequiresUnreferencedCode("Calls Elsa.Workflows.Contracts.IWorkflowStateSerializer.SerializeAsync(WorkflowState, CancellationToken)")] public async ValueTask SaveAsync(WorkflowInstance instance, CancellationToken cancellationToken = default) diff --git a/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceStore.cs b/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceStore.cs index 7ba1c3647..05a678a5a 100644 --- a/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceStore.cs +++ b/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceStore.cs @@ -178,4 +178,14 @@ public interface IWorkflowInstanceStore /// The new timestamp value to set. /// The cancellation token to observe during the operation. Task UpdateUpdatedTimestampAsync(string workflowInstanceId, DateTimeOffset value, CancellationToken cancellationToken = default); + + /// + /// Sets to and + /// to false only if the stored instance is still non-terminal. + /// + /// + /// true when the interrupt markers were applied; false when the instance is missing or already + /// . Implementations must not overwrite a concurrent terminal commit. + /// + ValueTask TryMarkInterruptedAsync(string workflowInstanceId, CancellationToken cancellationToken = default); } \ 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 09125daed..f40648a78 100644 --- a/src/modules/Elsa.Workflows.Management/Stores/MemoryWorkflowInstanceStore.cs +++ b/src/modules/Elsa.Workflows.Management/Stores/MemoryWorkflowInstanceStore.cs @@ -14,6 +14,7 @@ namespace Elsa.Workflows.Management.Stores; public class MemoryWorkflowInstanceStore : IWorkflowInstanceStore { private readonly MemoryStore _store; + private readonly object _sync = new(); /// /// Constructor. @@ -123,26 +124,30 @@ public class MemoryWorkflowInstanceStore : IWorkflowInstanceStore /// public ValueTask SaveAsync(WorkflowInstance instance, CancellationToken cancellationToken = default) { - _store.Save(instance, x => x.Id); + lock (_sync) + _store.Save(instance, x => x.Id); return ValueTask.CompletedTask; } public ValueTask AddAsync(WorkflowInstance instance, CancellationToken cancellationToken = default) { - _store.Add(instance, GetId); + lock (_sync) + _store.Add(instance, GetId); return ValueTask.CompletedTask; } public ValueTask UpdateAsync(WorkflowInstance instance, CancellationToken cancellationToken = default) { - _store.Update(instance, GetId); + lock (_sync) + _store.Update(instance, GetId); return ValueTask.CompletedTask; } /// public ValueTask SaveManyAsync(IEnumerable instances, CancellationToken cancellationToken = default) { - _store.SaveMany(instances, GetId); + lock (_sync) + _store.SaveMany(instances, GetId); return ValueTask.CompletedTask; } @@ -168,6 +173,33 @@ public class MemoryWorkflowInstanceStore : IWorkflowInstanceStore workflowInstance.UpdatedAt = value; } + /// + public ValueTask TryMarkInterruptedAsync(string workflowInstanceId, CancellationToken cancellationToken = default) + { + // 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) + return ValueTask.FromResult(false); + + instance.SubStatus = WorkflowSubStatus.Interrupted; + instance.IsExecuting = false; + + // In-place completion on the same object does not take this lock. If Status became + // Finished, do not keep Interrupted or report success. + if (instance.Status == WorkflowStatus.Finished) + { + instance.SubStatus = WorkflowSubStatus.Finished; + instance.IsExecuting = false; + return ValueTask.FromResult(false); + } + + return ValueTask.FromResult(true); + } + } + private static string GetId(WorkflowInstance workflowInstance) => workflowInstance.Id; [RequiresUnreferencedCode("Calls Elsa.Workflows.Management.Filters.WorkflowInstanceFilter.Apply(IQueryable)")] diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IInterruptedRecoveryScanner.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IInterruptedRecoveryScanner.cs index ef294cf16..5fa94966e 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IInterruptedRecoveryScanner.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IInterruptedRecoveryScanner.cs @@ -8,13 +8,14 @@ namespace Elsa.Workflows.Runtime; /// /// Disjoint from the timeout-based RestartInterruptedWorkflowsTask recurring task: that task filters /// IsExecuting = true with a stale UpdatedAt; this scan filters -/// (which has IsExecuting = false). The two filters never overlap, so an instance is recovered by exactly -/// one mechanism — see FR-022 and research R4. +/// and (Interrupted instances have IsExecuting = false). The two +/// filters never overlap, so an instance is recovered by exactly one mechanism — see FR-022 and research R4. +/// Terminal Finished+Interrupted rows from a drain/runner race are excluded so they are not requeued. /// public interface IInterruptedRecoveryScanner { /// - /// Enumerates instances in the sub-status, requeues each via + /// Enumerates running instances in the sub-status, requeues each via /// IWorkflowRestarter, and returns the count successfully requeued. /// ValueTask ScanAndRequeueAsync(CancellationToken cancellationToken); diff --git a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs index 8bf9c7c87..d4c70e72a 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs @@ -299,7 +299,8 @@ public class WorkflowRuntimeFeature(IModule module) : FeatureBase(module) // transition rule (SC-007) so transports stay thin. Scoped because INotificationSender is scoped. .AddScoped() // Interrupted-workflow recovery on shell activation (US3). Disjoint from the timeout-based - // RestartInterruptedWorkflowsTask: filter is SubStatus = Interrupted; that task's filter is IsExecuting=true. + // RestartInterruptedWorkflowsTask: filter is SubStatus = Interrupted AND Status = Running; + // that task's filter is IsExecuting=true. .AddScoped() .AddStartupTask() // Internal bookmark-queue processor surfaced as an ingress source for diagnostic visibility (FR-006). diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs b/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs index ae36ba0c2..1a7b7a217 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DrainOrchestrator.cs @@ -325,8 +325,10 @@ public sealed class DrainOrchestrator : IDrainOrchestrator // 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, accepting the - // runner-clobber race for that one instance (the recovery scan picks it up). + // 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). // Total wall time for this phase is at most ForceCancelSettleTimeout regardless // of N. var settleTasks = live.Select(async handle => @@ -420,13 +422,29 @@ public sealed class DrainOrchestrator : IDrainOrchestrator return; } + if (instance.Status == WorkflowStatus.Finished) + { + _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 { - instance.SubStatus = WorkflowSubStatus.Interrupted; - instance.IsExecuting = false; - await instanceStore.SaveAsync(instance, cancellationToken); + // 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); + if (!marked) + { + _logger.LogInformation( + "Skipping Interrupted persist for instance {InstanceId}: a concurrent persist already left it in a terminal status.", + instance.Id); + return; + } } catch (Exception ex) when (!ex.IsFatal()) { diff --git a/src/modules/Elsa.Workflows.Runtime/Services/InterruptedRecoveryScanner.cs b/src/modules/Elsa.Workflows.Runtime/Services/InterruptedRecoveryScanner.cs index b5d1f253c..1a88765e6 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/InterruptedRecoveryScanner.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/InterruptedRecoveryScanner.cs @@ -39,7 +39,11 @@ public sealed class InterruptedRecoveryScanner : IInterruptedRecoveryScanner /// public async ValueTask ScanAndRequeueAsync(CancellationToken cancellationToken) { - var filter = new WorkflowInstanceFilter { WorkflowSubStatus = WorkflowSubStatus.Interrupted }; + var filter = new WorkflowInstanceFilter + { + WorkflowSubStatus = WorkflowSubStatus.Interrupted, + WorkflowStatus = WorkflowStatus.Running, + }; var batchSize = _runtimeOptions.Value.RestartInterruptedWorkflowsBatchSize; var instances = _instanceStore.EnumerateSummariesAsync(filter, batchSize, cancellationToken); var requeued = 0; diff --git a/src/modules/Elsa.Workflows.Runtime/StartupTasks/RecoverInterruptedWorkflowsStartupTask.cs b/src/modules/Elsa.Workflows.Runtime/StartupTasks/RecoverInterruptedWorkflowsStartupTask.cs index f2005407f..db684c11b 100644 --- a/src/modules/Elsa.Workflows.Runtime/StartupTasks/RecoverInterruptedWorkflowsStartupTask.cs +++ b/src/modules/Elsa.Workflows.Runtime/StartupTasks/RecoverInterruptedWorkflowsStartupTask.cs @@ -5,9 +5,9 @@ using JetBrains.Annotations; namespace Elsa.Workflows.Runtime.StartupTasks; /// -/// Startup task that scans for workflow instances in the sub-status +/// Startup task that scans for running workflow instances in the sub-status /// and requeues each one immediately, bypassing the timeout-based RestartInterruptedWorkflowsTask recurring -/// cadence. See FR-021 and research R4. +/// cadence. See FR-021 and research R4. Terminal Finished+Interrupted rows are left untouched. /// [UsedImplicitly] [SingleNodeTask] diff --git a/test/integration/Elsa.AI.IntegrationTests/AIRuntimeGroundingToolTests.cs b/test/integration/Elsa.AI.IntegrationTests/AIRuntimeGroundingToolTests.cs index 954725144..b566df328 100644 --- a/test/integration/Elsa.AI.IntegrationTests/AIRuntimeGroundingToolTests.cs +++ b/test/integration/Elsa.AI.IntegrationTests/AIRuntimeGroundingToolTests.cs @@ -116,6 +116,17 @@ 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) + { + var instance = _instances.FirstOrDefault(x => x.Id == workflowInstanceId); + if (instance is null || instance.Status == WorkflowStatus.Finished) + return ValueTask.FromResult(false); + + instance.SubStatus = WorkflowSubStatus.Interrupted; + instance.IsExecuting = false; + return ValueTask.FromResult(true); + } + private IEnumerable Apply(WorkflowInstanceFilter filter) { var query = _instances.AsEnumerable(); diff --git a/test/integration/Elsa.Workflows.IntegrationTests/GracefulShutdown/InterruptedRecoveryIntegrationTests.cs b/test/integration/Elsa.Workflows.IntegrationTests/GracefulShutdown/InterruptedRecoveryIntegrationTests.cs index 6e88a7812..a5c64a167 100644 --- a/test/integration/Elsa.Workflows.IntegrationTests/GracefulShutdown/InterruptedRecoveryIntegrationTests.cs +++ b/test/integration/Elsa.Workflows.IntegrationTests/GracefulShutdown/InterruptedRecoveryIntegrationTests.cs @@ -73,8 +73,8 @@ public class InterruptedRecoveryIntegrationTests var instanceStore = scope.ServiceProvider.GetRequiredService(); // The two recovery paths use disjoint filters: - // - RestartInterruptedWorkflowsTask (recurring): IsExecuting=true AND UpdatedAt < threshold - // - RecoverInterruptedWorkflowsStartupTask (this scan): SubStatus = Interrupted (which has IsExecuting=false) + // - RestartInterruptedWorkflowsTask (recurring): IsExecuting=true AND UpdatedAt < threshold AND Status=Running + // - RecoverInterruptedWorkflowsStartupTask (this scan): SubStatus = Interrupted AND Status = Running (IsExecuting=false) await SeedInstancesAsync(instanceStore, 2, WorkflowSubStatus.Interrupted, isExecuting: false, idPrefix: "graceful-"); await SeedInstancesAsync(instanceStore, 2, WorkflowSubStatus.Executing, isExecuting: true, idPrefix: "ungraceful-"); @@ -89,7 +89,39 @@ public class InterruptedRecoveryIntegrationTests Assert.Equal(2, stillExecuting.Count()); } - private static async Task SeedInstancesAsync(IWorkflowInstanceStore store, int count, WorkflowSubStatus subStatus, bool isExecuting, string idPrefix = "instance-") + [Fact(DisplayName = "Scan does NOT requeue Finished+Interrupted instances (issue #8052)")] + public async Task DoesNotRequeueFinishedInterruptedInstances() + { + var fakeRestarter = new RecordingRestarter(); + using var scope = _services.CreateScope(); + var instanceStore = scope.ServiceProvider.GetRequiredService(); + + await SeedInstancesAsync(instanceStore, 2, WorkflowSubStatus.Interrupted, isExecuting: false, idPrefix: "running-", status: WorkflowStatus.Running); + await SeedInstancesAsync(instanceStore, 3, WorkflowSubStatus.Interrupted, isExecuting: false, idPrefix: "finished-", status: WorkflowStatus.Finished); + + var scanner = ActivatorUtilities.CreateInstance(scope.ServiceProvider, fakeRestarter); + var requeued = await scanner.ScanAndRequeueAsync(CancellationToken.None); + + Assert.Equal(2, requeued); + Assert.All(fakeRestarter.RestartedIds, id => Assert.StartsWith("running-", id)); + + var stillFinishedInterrupted = await instanceStore.FindManyAsync( + new WorkflowInstanceFilter + { + WorkflowSubStatus = WorkflowSubStatus.Interrupted, + WorkflowStatus = WorkflowStatus.Finished, + }, + CancellationToken.None); + Assert.Equal(3, stillFinishedInterrupted.Count()); + } + + private static async Task SeedInstancesAsync( + IWorkflowInstanceStore store, + int count, + WorkflowSubStatus subStatus, + bool isExecuting, + string idPrefix = "instance-", + WorkflowStatus status = WorkflowStatus.Running) { for (var i = 0; i < count; i++) { @@ -99,7 +131,7 @@ public class InterruptedRecoveryIntegrationTests DefinitionId = "def-1", DefinitionVersionId = "ver-1", Version = 1, - Status = WorkflowStatus.Running, + Status = status, SubStatus = subStatus, IsExecuting = isExecuting, CreatedAt = DateTimeOffset.UtcNow, @@ -109,7 +141,7 @@ public class InterruptedRecoveryIntegrationTests Id = $"{idPrefix}{i}", DefinitionId = "def-1", DefinitionVersionId = "ver-1", - Status = WorkflowStatus.Running, + Status = status, SubStatus = subStatus, }, }, CancellationToken.None); diff --git a/test/unit/Elsa.Workflows.Management.UnitTests/Stores/MemoryWorkflowInstanceStoreTests.cs b/test/unit/Elsa.Workflows.Management.UnitTests/Stores/MemoryWorkflowInstanceStoreTests.cs new file mode 100644 index 000000000..271f04a87 --- /dev/null +++ b/test/unit/Elsa.Workflows.Management.UnitTests/Stores/MemoryWorkflowInstanceStoreTests.cs @@ -0,0 +1,145 @@ +using Elsa.Common.Services; +using Elsa.Workflows.Management.Entities; +using Elsa.Workflows.Management.Filters; +using Elsa.Workflows.Management.Stores; + +namespace Elsa.Workflows.Management.UnitTests.Stores; + +public class MemoryWorkflowInstanceStoreTests +{ + [Fact(DisplayName = "TryMarkInterruptedAsync applies Interrupted only when the instance is still Running")] + public async Task TryMarkInterrupted_MarksRunningInstance() + { + var store = CreateStore(new WorkflowInstance + { + Id = "running-1", + DefinitionId = "def-1", + DefinitionVersionId = "ver-1", + Version = 1, + Status = WorkflowStatus.Running, + SubStatus = WorkflowSubStatus.Executing, + IsExecuting = true, + }); + + var marked = await store.TryMarkInterruptedAsync("running-1"); + + Assert.True(marked); + var instance = await store.FindAsync(new() { Id = "running-1" }); + Assert.NotNull(instance); + Assert.Equal(WorkflowStatus.Running, instance.Status); + Assert.Equal(WorkflowSubStatus.Interrupted, instance.SubStatus); + Assert.False(instance.IsExecuting); + } + + [Fact(DisplayName = "TryMarkInterruptedAsync does not overwrite a Finished instance")] + public async Task TryMarkInterrupted_DoesNotOverwriteFinishedInstance() + { + 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"); + + 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 replace a concurrent Finished commit with a stale Running snapshot")] + public async Task TryMarkInterrupted_DoesNotClobberConcurrentFinishedCommit() + { + var memory = new MemoryStore(); + var store = new MemoryWorkflowInstanceStore(memory); + await store.SaveAsync(new WorkflowInstance + { + Id = "raced-1", + DefinitionId = "def-1", + DefinitionVersionId = "ver-1", + Version = 1, + Status = WorkflowStatus.Running, + SubStatus = WorkflowSubStatus.Executing, + IsExecuting = true, + }); + + var staleRunning = await store.FindAsync(new() { Id = "raced-1" }); + Assert.NotNull(staleRunning); + + await store.SaveAsync(new WorkflowInstance + { + Id = "raced-1", + DefinitionId = "def-1", + DefinitionVersionId = "ver-1", + Version = 1, + Status = WorkflowStatus.Finished, + SubStatus = WorkflowSubStatus.Finished, + IsExecuting = false, + }); + + staleRunning.SubStatus = WorkflowSubStatus.Interrupted; + staleRunning.IsExecuting = false; + + var marked = await store.TryMarkInterruptedAsync("raced-1"); + + Assert.False(marked); + var instance = await store.FindAsync(new() { Id = "raced-1" }); + Assert.NotNull(instance); + Assert.Equal(WorkflowStatus.Finished, instance.Status); + Assert.Equal(WorkflowSubStatus.Finished, instance.SubStatus); + Assert.NotSame(staleRunning, instance); + } + + [Fact(DisplayName = "TryMarkInterruptedAsync does not leave Finished+Interrupted when SaveAsync completes concurrently")] + public async Task TryMarkInterrupted_DoesNotLeaveFinishedInterruptedWhenSaveCompletes() + { + for (var i = 0; i < 200; i++) + { + var store = new MemoryWorkflowInstanceStore(new MemoryStore()); + await store.SaveAsync(new WorkflowInstance + { + Id = "save-race-1", + DefinitionId = "def-1", + DefinitionVersionId = "ver-1", + Version = 1, + Status = WorkflowStatus.Running, + SubStatus = WorkflowSubStatus.Executing, + IsExecuting = true, + }); + + var mark = Task.Run(() => store.TryMarkInterruptedAsync("save-race-1").AsTask()); + var complete = Task.Run(() => store.SaveAsync(new WorkflowInstance + { + Id = "save-race-1", + DefinitionId = "def-1", + DefinitionVersionId = "ver-1", + Version = 1, + Status = WorkflowStatus.Finished, + SubStatus = WorkflowSubStatus.Finished, + IsExecuting = false, + }).AsTask()); + + await Task.WhenAll(mark, complete); + + var stored = await store.FindAsync(new WorkflowInstanceFilter { Id = "save-race-1" }); + Assert.NotNull(stored); + Assert.False( + stored.Status == WorkflowStatus.Finished && stored.SubStatus == WorkflowSubStatus.Interrupted, + $"Finished+Interrupted after SaveAsync completion race (iteration {i})."); + } + } + + private static MemoryWorkflowInstanceStore CreateStore(WorkflowInstance instance) + { + var store = new MemoryWorkflowInstanceStore(new MemoryStore()); + store.SaveAsync(instance).AsTask().GetAwaiter().GetResult(); + return store; + } +} diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs index 54b63636e..490282d00 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs @@ -37,6 +37,7 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase Version = 1, IsExecuting = true, })); + InstanceStore.TryMarkInterruptedAsync("instance-1", Arg.Any()).Returns(new ValueTask(true)); var sut = BuildSut(); var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce); @@ -45,10 +46,67 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount); Assert.Contains("instance-1", outcome.ForceCancelledInstanceIds); Assert.True(handle.CancellationToken.IsCancellationRequested); - await InstanceStore.Received(1).SaveAsync(Arg.Is(i => i.SubStatus == WorkflowSubStatus.Interrupted && !i.IsExecuting), Arg.Any()); + await InstanceStore.Received(1).TryMarkInterruptedAsync("instance-1", Arg.Any()); + await InstanceStore.DidNotReceive().SaveAsync(Arg.Any(), Arg.Any()); await LogStore.Received(1).AddAsync(Arg.Is(r => r.EventName == WorkflowInterruptedPayload.WorkflowInterruptedEventName), Arg.Any()); } + [Fact(DisplayName = "Force-cancel skips Interrupted persist when the instance is already Finished")] + public async Task SkipsInterruptedPersistForFinishedInstance() + { + var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-finished", 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-finished", + DefinitionId = "def-1", + DefinitionVersionId = "ver-1", + Version = 1, + Status = WorkflowStatus.Finished, + SubStatus = WorkflowSubStatus.Finished, + IsExecuting = false, + })); + + var sut = BuildSut(); + var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce); + + Assert.Equal(DrainResult.Forced, outcome.OverallResult); + 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 LogStore.DidNotReceive().AddAsync(Arg.Any(), Arg.Any()); + } + + [Fact(DisplayName = "Force-cancel skips Interrupted persist when TryMarkInterrupted loses the terminal race")] + public async Task SkipsInterruptedPersistWhenMarkLosesTerminalRace() + { + var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-raced", 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-raced", + DefinitionId = "def-1", + DefinitionVersionId = "ver-1", + Version = 1, + Status = WorkflowStatus.Running, + IsExecuting = true, + })); + InstanceStore.TryMarkInterruptedAsync("instance-raced", Arg.Any()).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.DidNotReceive().SaveAsync(Arg.Any(), Arg.Any()); + await LogStore.DidNotReceive().AddAsync(Arg.Any(), Arg.Any()); + } + [Fact(DisplayName = "Persistence failure during drain produces Reason=PersistenceFailure in payload")] public async Task PersistenceFailureRecordsReason() { @@ -64,8 +122,8 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase Version = 1, IsExecuting = true, })); - InstanceStore.SaveAsync(Arg.Any(), Arg.Any()) - .Returns(_ => throw new InvalidOperationException("db unavailable")); + InstanceStore.TryMarkInterruptedAsync("instance-2", Arg.Any()) + .Returns(_ => ValueTask.FromException(new InvalidOperationException("db unavailable"))); var sut = BuildSut(); var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce); diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Recovery/InterruptedRecoveryScannerTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Recovery/InterruptedRecoveryScannerTests.cs index 4056db2df..0fc5b5984 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Recovery/InterruptedRecoveryScannerTests.cs +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Recovery/InterruptedRecoveryScannerTests.cs @@ -17,8 +17,8 @@ public class InterruptedRecoveryScannerTests private readonly ILogger _logger = Substitute.For>(); private readonly RuntimeOptions _runtimeOptions = new() { RestartInterruptedWorkflowsBatchSize = 10 }; - [Fact(DisplayName = "Scan filters by SubStatus = Interrupted")] - public async Task FiltersBySubStatus() + [Fact(DisplayName = "Scan filters by SubStatus = Interrupted and Status = Running")] + public async Task FiltersByInterruptedAndRunning() { StubInstances(new[] { Summary("a") }); var sut = BuildSut(); @@ -26,7 +26,9 @@ public class InterruptedRecoveryScannerTests await sut.ScanAndRequeueAsync(CancellationToken.None); await _instanceStore.Received().SummarizeManyAsync( - Arg.Is(f => f.WorkflowSubStatus == WorkflowSubStatus.Interrupted), + Arg.Is(f => + f.WorkflowSubStatus == WorkflowSubStatus.Interrupted + && f.WorkflowStatus == WorkflowStatus.Running), Arg.Any(), Arg.Any()); }