diff --git a/src/clients/Elsa.Api.Client/Resources/WorkflowInstances/Enums/WorkflowSubStatus.cs b/src/clients/Elsa.Api.Client/Resources/WorkflowInstances/Enums/WorkflowSubStatus.cs index ccb9a48533..41f174cbf3 100644 --- a/src/clients/Elsa.Api.Client/Resources/WorkflowInstances/Enums/WorkflowSubStatus.cs +++ b/src/clients/Elsa.Api.Client/Resources/WorkflowInstances/Enums/WorkflowSubStatus.cs @@ -19,7 +19,7 @@ public enum WorkflowSubStatus /// The workflow is currently suspended and waiting for external stimuli to resume. /// Suspended, - + /// /// The workflow completed successfully. /// @@ -40,4 +40,9 @@ public enum WorkflowSubStatus /// The instance is resumable and will be picked up by the runtime's shell-activation recovery scan. /// Interrupted, -} \ No newline at end of file + + /// + /// Cancellation was requested and is being processed. + /// + Cancelling, +} diff --git a/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs b/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs index 19373143f7..2ec4fd366d 100644 --- a/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs @@ -703,6 +703,7 @@ private WorkflowStatus GetMainStatus(WorkflowSubStatus subStatus) => WorkflowSubStatus.Faulted => WorkflowStatus.Finished, WorkflowSubStatus.Finished => WorkflowStatus.Finished, WorkflowSubStatus.Suspended => WorkflowStatus.Running, + WorkflowSubStatus.Cancelling => WorkflowStatus.Running, _ => throw new ArgumentOutOfRangeException(nameof(subStatus), subStatus, null) }; diff --git a/src/modules/Elsa.Workflows.Core/Enums/WorkflowSubStatus.cs b/src/modules/Elsa.Workflows.Core/Enums/WorkflowSubStatus.cs index 2c96e8ba7c..4d86a8499c 100644 --- a/src/modules/Elsa.Workflows.Core/Enums/WorkflowSubStatus.cs +++ b/src/modules/Elsa.Workflows.Core/Enums/WorkflowSubStatus.cs @@ -19,7 +19,7 @@ public enum WorkflowSubStatus /// The workflow is currently suspended and waiting for external stimuli to resume. /// Suspended, - + /// /// The workflow completed successfully. /// @@ -43,4 +43,9 @@ public enum WorkflowSubStatus /// sub-status with a stale liveness timestamp. /// Interrupted, -} \ No newline at end of file + + /// + /// Cancellation was requested and is being processed. + /// + Cancelling, +} diff --git a/src/modules/Elsa.Workflows.Runtime/Services/WorkflowCancellationService.cs b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowCancellationService.cs index 552b75323a..6f7d3f68d9 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/WorkflowCancellationService.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowCancellationService.cs @@ -1,3 +1,4 @@ +using Elsa.Common; using Elsa.Common.Models; using Elsa.Workflows.Management; using Elsa.Workflows.Management.Entities; @@ -10,6 +11,7 @@ namespace Elsa.Workflows.Runtime; public class WorkflowCancellationService( IWorkflowDefinitionService workflowDefinitionService, IWorkflowInstanceStore workflowInstanceStore, + ISystemClock systemClock, IWorkflowCancellationDispatcher dispatcher) : IWorkflowCancellationService { @@ -67,19 +69,47 @@ public async Task CancelWorkflowByDefinitionAsync(string definitionId, Vers private async Task CancelWorkflows(IList workflowInstances, CancellationToken cancellationToken) { - var tasks = workflowInstances.Where(i => i.Status != WorkflowStatus.Finished) + var cancellableWorkflowInstances = workflowInstances.Where(i => i.Status != WorkflowStatus.Finished).ToList(); + var instanceIds = workflowInstances.Select(i => i.Id).ToList(); + + await MarkWorkflowInstancesAsCancellingAsync(cancellableWorkflowInstances, cancellationToken); + + var tasks = cancellableWorkflowInstances .Select(i => dispatcher.DispatchAsync(new DispatchCancelWorkflowRequest { WorkflowInstanceId = i.Id }, cancellationToken)).ToList(); - var instanceIds = workflowInstances.Select(i => i.Id).ToList(); await CancelChildWorkflowInstances(instanceIds, cancellationToken); await Task.WhenAll(tasks); return tasks.Count; } + private async Task MarkWorkflowInstancesAsCancellingAsync(ICollection workflowInstances, CancellationToken cancellationToken) + { + if (workflowInstances.Count == 0) + return; + + var now = systemClock.UtcNow; + + foreach (var workflowInstance in workflowInstances) + { + workflowInstance.Status = WorkflowStatus.Running; + workflowInstance.SubStatus = WorkflowSubStatus.Cancelling; + workflowInstance.UpdatedAt = now; + + if (workflowInstance.WorkflowState is not null) + { + workflowInstance.WorkflowState.Status = WorkflowStatus.Running; + workflowInstance.WorkflowState.SubStatus = WorkflowSubStatus.Cancelling; + workflowInstance.WorkflowState.UpdatedAt = now; + } + } + + await workflowInstanceStore.SaveManyAsync(workflowInstances, cancellationToken); + } + private async Task CancelChildWorkflowInstances(IEnumerable workflowInstanceIds, CancellationToken cancellationToken) { var tasks = new List>(); @@ -101,4 +131,4 @@ private async Task CancelChildWorkflowInstances(IEnumerable workflowInst await Task.WhenAll(tasks); } -} \ No newline at end of file +} diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/WorkflowCancellationServiceTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/WorkflowCancellationServiceTests.cs new file mode 100644 index 0000000000..6df3348b7e --- /dev/null +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/WorkflowCancellationServiceTests.cs @@ -0,0 +1,113 @@ +using Elsa.Common; +using Elsa.Common.Models; +using Elsa.Workflows.Management; +using Elsa.Workflows.Management.Entities; +using Elsa.Workflows.Management.Filters; +using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; +using Elsa.Workflows.State; +using NSubstitute; + +namespace Elsa.Workflows.Runtime.UnitTests.Services; + +public class WorkflowCancellationServiceTests +{ + private readonly IWorkflowDefinitionService _workflowDefinitionService = Substitute.For(); + private readonly IWorkflowInstanceStore _workflowInstanceStore = Substitute.For(); + private readonly ISystemClock _systemClock = Substitute.For(); + private readonly IWorkflowCancellationDispatcher _dispatcher = Substitute.For(); + + [Fact] + public async Task CancelWorkflowsAsync_MarksNonFinishedInstancesAsCancellingBeforeDispatchingCancellation() + { + var now = new DateTimeOffset(2026, 5, 22, 17, 0, 0, TimeSpan.Zero); + var events = new List(); + var suspendedInstance = CreateInstance("workflow-1", WorkflowStatus.Running, WorkflowSubStatus.Suspended); + var finishedInstance = CreateInstance("workflow-2", WorkflowStatus.Finished, WorkflowSubStatus.Finished); + var instances = new[] { suspendedInstance, finishedInstance }; + var service = CreateService(); + + _systemClock.UtcNow.Returns(now); + _workflowInstanceStore + .FindManyAsync(Arg.Any(), Arg.Any()) + .Returns(callInfo => + { + var filter = callInfo.ArgAt(0); + var matchesRequestedIds = filter.Ids?.Any() == true; + IEnumerable result = matchesRequestedIds ? instances : Array.Empty(); + return new ValueTask>(result); + }); + _workflowInstanceStore + .SaveManyAsync(Arg.Any>(), Arg.Any()) + .Returns(_ => + { + events.Add("save"); + return ValueTask.CompletedTask; + }); + _dispatcher + .DispatchAsync(Arg.Any(), Arg.Any()) + .Returns(_ => + { + events.Add("dispatch"); + return Task.FromResult(new DispatchCancelWorkflowsResponse()); + }); + + var count = await service.CancelWorkflowsAsync(new[] { "workflow-1", "workflow-2" }); + + Assert.Equal(1, count); + Assert.Equal(new[] { "save", "dispatch" }, events); + Assert.Equal(WorkflowStatus.Running, suspendedInstance.Status); + Assert.Equal(WorkflowSubStatus.Cancelling, suspendedInstance.SubStatus); + Assert.Equal(now, suspendedInstance.UpdatedAt); + Assert.Equal(WorkflowStatus.Running, suspendedInstance.WorkflowState.Status); + Assert.Equal(WorkflowSubStatus.Cancelling, suspendedInstance.WorkflowState.SubStatus); + Assert.Equal(now, suspendedInstance.WorkflowState.UpdatedAt); + Assert.Equal(WorkflowSubStatus.Finished, finishedInstance.SubStatus); + + await _workflowInstanceStore.Received(1).SaveManyAsync( + Arg.Is>(items => + items.Count() == 1 && + items.Single().Id == suspendedInstance.Id), + Arg.Any()); + await _dispatcher.Received(1).DispatchAsync( + Arg.Is(request => request.WorkflowInstanceId == suspendedInstance.Id), + Arg.Any()); + } + + [Fact] + public async Task CancelWorkflowAsync_ReturnsFalseWithoutDispatching_WhenInstanceDoesNotExist() + { + var service = CreateService(); + _workflowInstanceStore + .FindAsync(Arg.Any(), Arg.Any()) + .Returns(new ValueTask((WorkflowInstance?)null)); + + var result = await service.CancelWorkflowAsync("missing-workflow"); + + Assert.False(result); + await _workflowInstanceStore.DidNotReceive().SaveManyAsync(Arg.Any>(), Arg.Any()); + await _dispatcher.DidNotReceive().DispatchAsync(Arg.Any(), Arg.Any()); + } + + private WorkflowCancellationService CreateService() => new(_workflowDefinitionService, _workflowInstanceStore, _systemClock, _dispatcher); + + private static WorkflowInstance CreateInstance(string id, WorkflowStatus status, WorkflowSubStatus subStatus) + { + var workflowState = new WorkflowState + { + Id = id, + Status = status, + SubStatus = subStatus + }; + + return new() + { + Id = id, + DefinitionId = "definition-1", + DefinitionVersionId = "definition-version-1", + Status = status, + SubStatus = subStatus, + WorkflowState = workflowState + }; + } +}