From 64636a362b62011ef0fee18398db416cecc58a16 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Wed, 8 Sep 2021 22:37:42 +0200 Subject: [PATCH] Initial work on workflow instance cancellation --- .../Handlers/RemoveScheduledTriggers.cs | 14 ++++-- .../Events/WorkflowInstanceCancelled.cs | 19 ++++++++ .../Workflows/IWorkflowInstanceCanceller.cs | 20 ++++++++ .../LockingWorkflowInstanceCanceller.cs | 45 ++++++++++++++++++ .../ElsaServiceCollectionExtensions.cs | 2 + .../Workflows/WorkflowInstanceCanceller.cs | 43 +++++++++++++++++ .../Services/Workflows/WorkflowRunner.cs | 4 +- .../elsa-workflows-studio/src/index.html | 7 ++- .../Endpoints/WorkflowInstances/Cancel.cs | 47 +++++++++++++++++++ 9 files changed, 195 insertions(+), 6 deletions(-) create mode 100644 src/core/Elsa.Abstractions/Events/WorkflowInstanceCancelled.cs create mode 100644 src/core/Elsa.Abstractions/Services/Workflows/IWorkflowInstanceCanceller.cs create mode 100644 src/core/Elsa.Core/Decorators/LockingWorkflowInstanceCanceller.cs create mode 100644 src/core/Elsa.Core/Services/Workflows/WorkflowInstanceCanceller.cs create mode 100644 src/server/Elsa.Server.Api/Endpoints/WorkflowInstances/Cancel.cs diff --git a/src/activities/Elsa.Activities.Temporal.Common/Handlers/RemoveScheduledTriggers.cs b/src/activities/Elsa.Activities.Temporal.Common/Handlers/RemoveScheduledTriggers.cs index 69b6d9032..bef066dfe 100644 --- a/src/activities/Elsa.Activities.Temporal.Common/Handlers/RemoveScheduledTriggers.cs +++ b/src/activities/Elsa.Activities.Temporal.Common/Handlers/RemoveScheduledTriggers.cs @@ -6,7 +6,13 @@ using MediatR; namespace Elsa.Activities.Temporal.Common.Handlers { - public class RemoveScheduledTriggers : INotificationHandler, INotificationHandler, INotificationHandler, INotificationHandler + public class RemoveScheduledTriggers : + INotificationHandler, + INotificationHandler, + INotificationHandler, + INotificationHandler, + INotificationHandler, + INotificationHandler { private readonly IWorkflowDefinitionScheduler _workflowDefinitionScheduler; private readonly IWorkflowInstanceScheduler _workflowInstanceScheduler; @@ -20,14 +26,16 @@ namespace Elsa.Activities.Temporal.Common.Handlers public async Task Handle(BlockingActivityRemoved notification, CancellationToken cancellationToken) { // TODO: Consider introducing a "stereotype" field for activities to exit early in case they are not stereotyped as "temporal". - + await _workflowInstanceScheduler.UnscheduleAsync( notification.WorkflowExecutionContext.WorkflowInstance.Id, notification.BlockingActivity.ActivityId, cancellationToken); } - public Task Handle(WorkflowDefinitionPublished notification, CancellationToken cancellationToken) => _workflowDefinitionScheduler.UnscheduleAsync(notification.WorkflowDefinition.DefinitionId , cancellationToken); + public async Task Handle(WorkflowCancelled notification, CancellationToken cancellationToken) => await _workflowInstanceScheduler.UnscheduleAsync(notification.WorkflowExecutionContext.WorkflowInstance.Id, cancellationToken); + public async Task Handle(WorkflowInstanceCancelled notification, CancellationToken cancellationToken) => await _workflowInstanceScheduler.UnscheduleAsync(notification.WorkflowInstance.Id, cancellationToken); + public Task Handle(WorkflowDefinitionPublished notification, CancellationToken cancellationToken) => _workflowDefinitionScheduler.UnscheduleAsync(notification.WorkflowDefinition.DefinitionId, cancellationToken); public Task Handle(WorkflowDefinitionRetracted notification, CancellationToken cancellationToken) => _workflowDefinitionScheduler.UnscheduleAsync(notification.WorkflowDefinition.DefinitionId, cancellationToken); public Task Handle(WorkflowDefinitionDeleted notification, CancellationToken cancellationToken) => _workflowDefinitionScheduler.UnscheduleAsync(notification.WorkflowDefinition.DefinitionId, cancellationToken); } diff --git a/src/core/Elsa.Abstractions/Events/WorkflowInstanceCancelled.cs b/src/core/Elsa.Abstractions/Events/WorkflowInstanceCancelled.cs new file mode 100644 index 000000000..abc9b9e30 --- /dev/null +++ b/src/core/Elsa.Abstractions/Events/WorkflowInstanceCancelled.cs @@ -0,0 +1,19 @@ +using Elsa.Models; +using Elsa.Services.Models; +using MediatR; + +namespace Elsa.Events +{ + /// + /// Published when a workflow instance was cancelled + /// + public class WorkflowInstanceCancelled : INotification + { + public WorkflowInstanceCancelled(WorkflowInstance workflowInstance) + { + WorkflowInstance = workflowInstance; + } + + public WorkflowInstance WorkflowInstance { get; } + } +} \ No newline at end of file diff --git a/src/core/Elsa.Abstractions/Services/Workflows/IWorkflowInstanceCanceller.cs b/src/core/Elsa.Abstractions/Services/Workflows/IWorkflowInstanceCanceller.cs new file mode 100644 index 000000000..f271163ad --- /dev/null +++ b/src/core/Elsa.Abstractions/Services/Workflows/IWorkflowInstanceCanceller.cs @@ -0,0 +1,20 @@ +using System.Threading; +using System.Threading.Tasks; +using Elsa.Models; + +namespace Elsa.Services +{ + public interface IWorkflowInstanceCanceller + { + Task CancelAsync(string workflowInstanceId, CancellationToken cancellationToken = default); + } + + public record CancelWorkflowInstanceResult(CancelWorkflowInstanceResultStatus Status, WorkflowInstance? WorkflowInstance); + + public enum CancelWorkflowInstanceResultStatus + { + Ok, + NotFound, + InvalidStatus + } +} \ No newline at end of file diff --git a/src/core/Elsa.Core/Decorators/LockingWorkflowInstanceCanceller.cs b/src/core/Elsa.Core/Decorators/LockingWorkflowInstanceCanceller.cs new file mode 100644 index 000000000..518e2b218 --- /dev/null +++ b/src/core/Elsa.Core/Decorators/LockingWorkflowInstanceCanceller.cs @@ -0,0 +1,45 @@ +using System.Threading; +using System.Threading.Tasks; +using Elsa.Exceptions; +using Elsa.Options; +using Elsa.Services; +using Elsa.Services.Models; + +namespace Elsa.Decorators +{ + public class LockingWorkflowInstanceCanceller : IWorkflowInstanceCanceller + { + private readonly IWorkflowInstanceCanceller _workflowInstanceCanceller; + private readonly IDistributedLockProvider _distributedLockProvider; + private readonly ElsaOptions _elsaOptions; + + public LockingWorkflowInstanceCanceller(IWorkflowInstanceCanceller workflowInstanceCanceller, IDistributedLockProvider distributedLockProvider, ElsaOptions elsaOptions) + { + _workflowInstanceCanceller = workflowInstanceCanceller; + _distributedLockProvider = distributedLockProvider; + _elsaOptions = elsaOptions; + } + + public async Task CancelAsync(string workflowInstanceId, CancellationToken cancellationToken = default) + { + var workflowInstanceLockKey = $"workflow-instance:{workflowInstanceId}"; + var currentWorkflowInstanceLockHandle = AmbientLockContext.GetCurrentWorkflowInstanceLock(workflowInstanceId); + var workflowInstanceLockHandle = currentWorkflowInstanceLockHandle ?? await _distributedLockProvider.AcquireLockAsync(workflowInstanceLockKey, _elsaOptions.DistributedLockTimeout, cancellationToken); + + if (workflowInstanceLockHandle == null) + throw new LockAcquisitionException("Could not acquire a lock within the configured amount of time"); + + try + { + AmbientLockContext.SetCurrentWorkflowInstanceLock(workflowInstanceId, workflowInstanceLockHandle); + + return await _workflowInstanceCanceller.CancelAsync(workflowInstanceId, cancellationToken); + } + finally + { + AmbientLockContext.DeleteCurrentWorkflowInstanceLock(workflowInstanceId); + await workflowInstanceLockHandle.DisposeAsync(); + } + } + } +} \ No newline at end of file diff --git a/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs b/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs index a1fd5958a..4bbab3ece 100644 --- a/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs +++ b/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs @@ -95,6 +95,7 @@ namespace Microsoft.Extensions.DependencyInjection services.Decorate(); services.Decorate(); services.Decorate(); + services.Decorate(); //TenantId default source services.TryAddScoped(); @@ -174,6 +175,7 @@ namespace Microsoft.Extensions.DependencyInjection .AddScoped() .AddScoped() .AddScoped() + .AddScoped() .AddSingleton() .AddTransient() .AddSingleton() diff --git a/src/core/Elsa.Core/Services/Workflows/WorkflowInstanceCanceller.cs b/src/core/Elsa.Core/Services/Workflows/WorkflowInstanceCanceller.cs new file mode 100644 index 000000000..2fc97ccfd --- /dev/null +++ b/src/core/Elsa.Core/Services/Workflows/WorkflowInstanceCanceller.cs @@ -0,0 +1,43 @@ +using System.Threading; +using System.Threading.Tasks; +using Elsa.Events; +using Elsa.Models; +using Elsa.Persistence; +using MediatR; +using Microsoft.Extensions.Logging; +using NodaTime; + +namespace Elsa.Services.Workflows +{ + public class WorkflowInstanceCanceller : IWorkflowInstanceCanceller + { + private readonly IWorkflowInstanceStore _workflowInstanceStore; + private readonly IClock _clock; + private readonly IMediator _mediator; + + public WorkflowInstanceCanceller(IWorkflowInstanceStore workflowInstanceStore, IClock clock, IMediator mediator) + { + _workflowInstanceStore = workflowInstanceStore; + _clock = clock; + _mediator = mediator; + } + + public async Task CancelAsync(string workflowInstanceId, CancellationToken cancellationToken = default) + { + var workflowInstance = await _workflowInstanceStore.FindByIdAsync(workflowInstanceId, cancellationToken); + + if (workflowInstance == null) + return new CancelWorkflowInstanceResult(CancelWorkflowInstanceResultStatus.NotFound, null); + + if (workflowInstance.WorkflowStatus != WorkflowStatus.Idle && workflowInstance.WorkflowStatus != WorkflowStatus.Running && workflowInstance.WorkflowStatus != WorkflowStatus.Suspended) + return new CancelWorkflowInstanceResult(CancelWorkflowInstanceResultStatus.InvalidStatus, workflowInstance); + + workflowInstance.WorkflowStatus = WorkflowStatus.Cancelled; + workflowInstance.CancelledAt = _clock.GetCurrentInstant(); + + await _workflowInstanceStore.SaveAsync(workflowInstance, cancellationToken); + + return new CancelWorkflowInstanceResult(CancelWorkflowInstanceResultStatus.Ok, workflowInstance); + } + } +} \ No newline at end of file diff --git a/src/core/Elsa.Core/Services/Workflows/WorkflowRunner.cs b/src/core/Elsa.Core/Services/Workflows/WorkflowRunner.cs index 0bb635918..a19f1319e 100644 --- a/src/core/Elsa.Core/Services/Workflows/WorkflowRunner.cs +++ b/src/core/Elsa.Core/Services/Workflows/WorkflowRunner.cs @@ -128,12 +128,12 @@ namespace Elsa.Services.Workflows default: throw new ArgumentOutOfRangeException(); } - + await _mediator.Publish(new WorkflowExecuted(workflowExecutionContext), cancellationToken); var statusEvent = workflowExecutionContext.Status switch { - WorkflowStatus.Cancelled => new WorkflowCancelled(workflowExecutionContext), + WorkflowStatus.Cancelled => new WorkflowCancelled(workflowExecutionContext), // TODO: Publish WorkflowInstanceCancelled event also WorkflowStatus.Finished => new WorkflowCompleted(workflowExecutionContext), WorkflowStatus.Faulted => new WorkflowFaulted(workflowExecutionContext), WorkflowStatus.Suspended => new WorkflowSuspended(workflowExecutionContext), diff --git a/src/designer/elsa-workflows-studio/src/index.html b/src/designer/elsa-workflows-studio/src/index.html index d7aa22996..2d3cc97f6 100644 --- a/src/designer/elsa-workflows-studio/src/index.html +++ b/src/designer/elsa-workflows-studio/src/index.html @@ -11,10 +11,15 @@ + - + diff --git a/src/server/Elsa.Server.Api/Endpoints/WorkflowInstances/Cancel.cs b/src/server/Elsa.Server.Api/Endpoints/WorkflowInstances/Cancel.cs new file mode 100644 index 000000000..74efa8bf6 --- /dev/null +++ b/src/server/Elsa.Server.Api/Endpoints/WorkflowInstances/Cancel.cs @@ -0,0 +1,47 @@ +using System.Threading; +using System.Threading.Tasks; +using Elsa.Models; +using Elsa.Persistence; +using Elsa.Services; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; +using Swashbuckle.AspNetCore.Annotations; + +namespace Elsa.Server.Api.Endpoints.WorkflowInstances +{ + [ApiController] + [ApiVersion("1")] + [Route("v{apiVersion:apiVersion}/workflow-instances/{id}/cancel")] + [Produces("application/json")] + public class Cancel : Controller + { + private readonly IWorkflowInstanceCanceller _canceller; + + public Cancel(IWorkflowInstanceCanceller canceller) + { + _canceller = canceller; + } + + [HttpPost] + [ProducesResponseType(StatusCodes.Status200OK)] + [ProducesResponseType(StatusCodes.Status404NotFound)] + [ProducesResponseType(StatusCodes.Status400BadRequest)] + [SwaggerOperation( + Summary = "Cancels a workflow instance.", + Description = "Retries a workflow instance.", + OperationId = "WorkflowInstances.Retry", + Tags = new[] { "WorkflowInstances" }) + ] + public async Task Handle(string id, CancellationToken cancellationToken = default) + { + var result = await _canceller.CancelAsync(id, cancellationToken); + + return result.Status switch + { + CancelWorkflowInstanceResultStatus.NotFound => NotFound(), + CancelWorkflowInstanceResultStatus.InvalidStatus => BadRequest($"Cannot cancel a workflow instance with status {result.WorkflowInstance!.WorkflowStatus}"), + _ => Ok() + }; + } + } +} \ No newline at end of file