diff --git a/src/activities/Elsa.Activities.Http/Extensions/HttpRequestExtensions.cs b/src/activities/Elsa.Activities.Http/Extensions/HttpRequestExtensions.cs index 3f55f1346..876e06d9c 100644 --- a/src/activities/Elsa.Activities.Http/Extensions/HttpRequestExtensions.cs +++ b/src/activities/Elsa.Activities.Http/Extensions/HttpRequestExtensions.cs @@ -54,5 +54,16 @@ namespace Elsa.Activities.Http.Extensions correlationId = null; return false; } + + public static bool GetUseDispatch(this HttpRequest request) + { + if (request.Query.ContainsKey("x-dispatch")) + return true; + + if (request.Headers.ContainsKey("X-Dispatch")) + return true; + + return false; + } } } \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Http/Middleware/HttpRequestMiddleware.cs b/src/activities/Elsa.Activities.Http/Middleware/HttpRequestMiddleware.cs index 3f5425d87..1f7e5febb 100644 --- a/src/activities/Elsa.Activities.Http/Middleware/HttpRequestMiddleware.cs +++ b/src/activities/Elsa.Activities.Http/Middleware/HttpRequestMiddleware.cs @@ -15,19 +15,27 @@ namespace Elsa.Activities.Http.Middleware public HttpEndpointMiddleware(RequestDelegate next) => _next = next; - public async Task InvokeAsync(HttpContext httpContext, IMediator mediator) + public async Task InvokeAsync(HttpContext httpContext, IMediator mediator, IWorkflowInstanceDispatcher workflowInstanceDispatcher) { var path = httpContext.Request.Path.Value.ToLowerInvariant(); var method = httpContext.Request.Method!.ToLowerInvariant(); var cancellationToken = httpContext.RequestAborted; httpContext.Request.TryGetCorrelationId(out var correlationId); + var useDispatch = httpContext.Request.GetUseDispatch(); const string activityType = nameof(HttpEndpoint); var trigger = new HttpEndpointBookmark(path, method, null); var bookmark = new HttpEndpointBookmark(path, method, correlationId?.ToLowerInvariant()); - var workflowInstanceCount = await mediator.Send(new TriggerWorkflowsRequest(activityType, bookmark, trigger, default, correlationId, default, TenantId), cancellationToken); + var triggerRequest = new TriggerWorkflowsRequest(activityType, bookmark, trigger, default, correlationId, default, TenantId, Execute: !useDispatch); + var response = await mediator.Send(triggerRequest, cancellationToken); - if (workflowInstanceCount == 0) + if (useDispatch) + { + // We created workflow instances but haven't scheduled them for execution yet. Use the dispatcher to schedule them for execution. + workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest()) + } + + if (response.WorkflowInstanceIds.Count == 0) { await _next(httpContext); } diff --git a/src/core/Elsa.Abstractions/Dispatch/Models.cs b/src/core/Elsa.Abstractions/Dispatch/Models.cs index 1647748c4..1e47e1bbc 100644 --- a/src/core/Elsa.Abstractions/Dispatch/Models.cs +++ b/src/core/Elsa.Abstractions/Dispatch/Models.cs @@ -1,10 +1,15 @@ -using Elsa.Bookmarks; +using System.Collections.Generic; +using Elsa.Bookmarks; using MediatR; namespace Elsa.Dispatch { - public record TriggerWorkflowsRequest(string ActivityType, IBookmark Bookmark, IBookmark Trigger, object? Input = default, string? CorrelationId = default, string? WorkflowInstanceId = default, string? ContextId = default, string? TenantId = default) : IRequest; - public record ExecuteWorkflowDefinitionRequest(string WorkflowDefinitionId, string? ActivityId = default, object? Input = default, string? CorrelationId = default, string? ContextId = default, string? TenantId = default) : IRequest; - public record ExecuteWorkflowInstanceRequest(string WorkflowInstanceId, string ActivityId, object? Input = default) : IRequest; + public record TriggerWorkflowsRequest(string ActivityType, IBookmark Bookmark, IBookmark Trigger, object? Input = default, string? CorrelationId = default, string? WorkflowInstanceId = default, string? ContextId = default, string? TenantId = default, bool Execute = true) : IRequest; + public record ExecuteWorkflowDefinitionRequest(string WorkflowDefinitionId, string? ActivityId = default, object? Input = default, string? CorrelationId = default, string? ContextId = default, string? TenantId = default, bool Execute = true) : IRequest; + public record ExecuteWorkflowInstanceRequest(string WorkflowInstanceId, string ActivityId, object? Input = default, bool Execute = true) : IRequest; + public record TriggerWorkflowsResponse(ICollection PendingWorkflows); + public record ExecuteWorkflowDefinitionResponse(PendingWorkflow? PendingWorkflow = default); + + public record PendingWorkflow(string WorkflowInstanceId, string ActivityId); } \ No newline at end of file diff --git a/src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowDefinition.cs b/src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowDefinition.cs index ef1a43b25..5461f5f62 100644 --- a/src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowDefinition.cs +++ b/src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowDefinition.cs @@ -14,7 +14,7 @@ using IDistributedLockProvider = Elsa.Services.IDistributedLockProvider; namespace Elsa.Dispatch.Handlers { - public class ExecuteWorkflowDefinition : IRequestHandler + public class ExecuteWorkflowDefinition : IRequestHandler { private readonly IStartsWorkflow _startsWorkflow; private readonly IWorkflowRegistry _workflowRegistry; @@ -39,19 +39,19 @@ namespace Elsa.Dispatch.Handlers _logger = logger; } - public async Task Handle(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken) + public async Task Handle(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken) { var workflowDefinitionId = request.WorkflowDefinitionId; var tenantId = request.TenantId; var workflowBlueprint = await _workflowRegistry.GetAsync(workflowDefinitionId, tenantId, VersionOptions.Published, cancellationToken); if (!ValidatePreconditions(workflowDefinitionId, workflowBlueprint)) - return Unit.Value; + return new ExecuteWorkflowDefinitionResponse(); - var lockKey = $"execute-workflow-definition:tenant:{tenantId}:workflow-definition:{workflowDefinitionId}"; var correlationId = request.CorrelationId; var correlationLockHandle = default(IDistributedSynchronizationHandle?); + // If we are creating a correlated workflow, make sure to acquire a lock on it to prevent duplicate workflow instances from being created. if (!string.IsNullOrWhiteSpace(correlationId)) { _logger.LogDebug("Acquiring lock on correlation ID {CorrelationId}", correlationId); @@ -63,13 +63,18 @@ namespace Elsa.Dispatch.Handlers try { + // Acquire a lock on the workflow definition so that we can ensure singleton-workflows never execute more than one instance. + var lockKey = $"execute-workflow-definition:tenant:{tenantId}:workflow-definition:{workflowDefinitionId}"; await using var handle = await _distributedLockProvider.AcquireLockAsync(lockKey, _elsaOptions.DistributedLockTimeout, cancellationToken); if (handle == null) throw new LockAcquisitionException($"Failed to acquire a lock on {lockKey}"); if (!workflowBlueprint!.IsSingleton || await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId) == false) - await _startsWorkflow.StartWorkflowAsync(workflowBlueprint, request.ActivityId, request.Input, request.CorrelationId, request.ContextId, cancellationToken); + { + var workflowInstance = await _startsWorkflow.StartWorkflowAsync(workflowBlueprint, request.ActivityId, request.Input, request.CorrelationId, request.ContextId, cancellationToken); + return new ExecuteWorkflowDefinitionResponse(workflowInstance.Id); + } } finally { @@ -77,7 +82,7 @@ namespace Elsa.Dispatch.Handlers await correlationLockHandle.DisposeAsync(); } - return Unit.Value; + return new ExecuteWorkflowDefinitionResponse(); } private bool ValidatePreconditions(string? workflowDefinitionId, IWorkflowBlueprint? workflowBlueprint) diff --git a/src/core/Elsa.Core/Dispatch/Handlers/TriggerWorkflows.cs b/src/core/Elsa.Core/Dispatch/Handlers/TriggerWorkflows.cs index 18fd7d527..aff5e7d3e 100644 --- a/src/core/Elsa.Core/Dispatch/Handlers/TriggerWorkflows.cs +++ b/src/core/Elsa.Core/Dispatch/Handlers/TriggerWorkflows.cs @@ -2,7 +2,6 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; -using Elsa.Activities.Signaling; using Elsa.Bookmarks; using Elsa.Exceptions; using Elsa.Models; @@ -17,7 +16,7 @@ using Open.Linq.AsyncExtensions; namespace Elsa.Dispatch.Handlers { - public class TriggerWorkflows : IRequestHandler + public class TriggerWorkflows : IRequestHandler { private readonly IWorkflowInstanceStore _workflowInstanceStore; private readonly IBookmarkFinder _bookmarkFinder; @@ -45,7 +44,7 @@ namespace Elsa.Dispatch.Handlers _logger = logger; } - public async Task Handle(TriggerWorkflowsRequest request, CancellationToken cancellationToken) + public async Task Handle(TriggerWorkflowsRequest request, CancellationToken cancellationToken) { var correlationId = request.CorrelationId; @@ -58,30 +57,30 @@ namespace Elsa.Dispatch.Handlers return await TriggerWorkflowsAsync(request, cancellationToken); } - private async Task TriggerWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken) + private async Task TriggerWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken) { var bookmarkResultsQuery = await _bookmarkFinder.FindBookmarksAsync(request.ActivityType, request.Bookmark, request.CorrelationId, request.TenantId, cancellationToken); var bookmarkResults = bookmarkResultsQuery.ToList(); - var triggeredCount = bookmarkResults.GroupBy(x => x.WorkflowInstanceId).Select(x => x.Key).Distinct().Count(); + var triggeredPendingWorkflows = bookmarkResults.Select(x => new PendingWorkflow(x.WorkflowInstanceId, x.ActivityId)).ToList(); + await ResumeWorkflowsAsync(bookmarkResults, request.Input, request.Execute, cancellationToken); + var startWorkflowsResponse = await StartWorkflowsAsync(request, cancellationToken); + var pendingWorkflows = triggeredPendingWorkflows.Concat(startWorkflowsResponse.PendingWorkflows).Distinct().ToList(); - await ResumeWorkflowsAsync(bookmarkResults, request.Input, cancellationToken); - var startedCount = await StartWorkflowsAsync(request, cancellationToken); - - return startedCount + triggeredCount; + return new TriggerWorkflowsResponse(pendingWorkflows); } - private async Task ResumeSpecificWorkflowInstanceAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken) + private async Task ResumeSpecificWorkflowInstanceAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken) { var bookmarkResultsQuery = await _bookmarkFinder.FindBookmarksAsync(request.ActivityType, request.Bookmark, request.CorrelationId, request.TenantId, cancellationToken); bookmarkResultsQuery = bookmarkResultsQuery.Where(x => x.WorkflowInstanceId == request.WorkflowInstanceId); var bookmarkResults = bookmarkResultsQuery.ToList(); - var triggeredCount = bookmarkResults.GroupBy(x => x.WorkflowInstanceId).Select(x => x.Key).Distinct().Count(); + await ResumeWorkflowsAsync(bookmarkResults, request.Input, request.Execute, cancellationToken); + var pendingWorkflows = bookmarkResults.Select(x => new PendingWorkflow(x.WorkflowInstanceId, x.ActivityId)).ToList(); - await ResumeWorkflowsAsync(bookmarkResults, request.Input, cancellationToken); - return triggeredCount; + return new TriggerWorkflowsResponse(pendingWorkflows); } - private async Task ResumeOrStartCorrelatedWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken) + private async Task ResumeOrStartCorrelatedWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken) { var correlationId = request.CorrelationId!; var lockKey = correlationId; @@ -102,37 +101,38 @@ namespace Elsa.Dispatch.Handlers { _logger.LogDebug("{WorkflowInstanceCount} existing workflows found with correlation ID '{CorrelationId}' will be queued for execution", correlatedWorkflowInstanceCount, correlationId); var bookmarkResults = await _bookmarkFinder.FindBookmarksAsync(request.ActivityType, request.Bookmark, correlationId, request.TenantId, cancellationToken).ToList(); - await ResumeWorkflowsAsync(bookmarkResults, request.Input, cancellationToken); - return correlatedWorkflowInstanceCount; + await ResumeWorkflowsAsync(bookmarkResults, request.Input, request.Execute, cancellationToken); + return new TriggerWorkflowsResponse(bookmarkResults.Select(x => new PendingWorkflow(x.WorkflowInstanceId, x.ActivityId)).ToList()); } } return await StartWorkflowsAsync(request, cancellationToken); } - private async Task StartWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken) + private async Task StartWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken) { _logger.LogDebug("Triggering workflows using {ActivityType}", request.ActivityType); var filter = request.Trigger; var triggers = (await _triggerFinder.FindTriggersAsync(request.ActivityType, filter, request.TenantId, cancellationToken)).ToList(); + var pendingWorkflows = new List(); foreach (var trigger in triggers) { var workflowBlueprint = trigger.WorkflowBlueprint; + var response = await _mediator.Send(new ExecuteWorkflowDefinitionRequest(workflowBlueprint.Id, trigger.ActivityId, request.Input, request.CorrelationId, request.ContextId, workflowBlueprint.TenantId, request.Execute), cancellationToken); - await _mediator.Send( - new ExecuteWorkflowDefinitionRequest(workflowBlueprint.Id, trigger.ActivityId, request.Input, request.CorrelationId, request.ContextId, workflowBlueprint.TenantId), - cancellationToken); + if (response.PendingWorkflow != null) + pendingWorkflows.Add(response.PendingWorkflow); } - return triggers.Count; + return new TriggerWorkflowsResponse(pendingWorkflows); } - private async Task ResumeWorkflowsAsync(IEnumerable results, object? input, CancellationToken cancellationToken) + private async Task ResumeWorkflowsAsync(IEnumerable results, object? input, bool execute, CancellationToken cancellationToken) { foreach (var result in results) - await _mediator.Send(new ExecuteWorkflowInstanceRequest(result.WorkflowInstanceId, result.ActivityId, input), cancellationToken); + await _mediator.Send(new ExecuteWorkflowInstanceRequest(result.WorkflowInstanceId, result.ActivityId, input, execute), cancellationToken); } } } \ No newline at end of file