diff --git a/src/common/Elsa.Testing.Shared/ServiceProviderExtensions.cs b/src/common/Elsa.Testing.Shared/ServiceProviderExtensions.cs index 22eadcff5..5217b8ea3 100644 --- a/src/common/Elsa.Testing.Shared/ServiceProviderExtensions.cs +++ b/src/common/Elsa.Testing.Shared/ServiceProviderExtensions.cs @@ -62,13 +62,13 @@ public static class ServiceProviderExtensions while (bookmarks.TryPop(out var bookmark)) { var resumeOptions = new ResumeWorkflowRuntimeOptions(BookmarkId: bookmark.Id); - var resumeResult = await workflowRuntime.ResumeWorkflowAsync(result.InstanceId, resumeOptions); + var resumeResult = await workflowRuntime.ResumeWorkflowAsync(result.WorkflowInstanceId, resumeOptions); foreach (var newBookmark in resumeResult.Bookmarks) bookmarks.Push(newBookmark); } // Return the workflow state. - return (await workflowRuntime.ExportWorkflowStateAsync(result.InstanceId))!; + return (await workflowRuntime.ExportWorkflowStateAsync(result.WorkflowInstanceId))!; } } \ No newline at end of file diff --git a/src/modules/Elsa.Http/Contracts/IHttpBookmarkProcessor.cs b/src/modules/Elsa.Http/Contracts/IHttpBookmarkProcessor.cs index a7250edfc..f3aece68c 100644 --- a/src/modules/Elsa.Http/Contracts/IHttpBookmarkProcessor.cs +++ b/src/modules/Elsa.Http/Contracts/IHttpBookmarkProcessor.cs @@ -1,3 +1,4 @@ +using Elsa.Workflows.Core.State; using Elsa.Workflows.Runtime.Contracts; namespace Elsa.Http.Contracts; @@ -10,7 +11,7 @@ public interface IHttpBookmarkProcessor /// /// Processes the specified by resuming each HTTP bookmark while we are in an HTTP context. /// - Task ProcessBookmarks( + Task> ProcessBookmarks( IEnumerable executionResults, string? correlationId, IDictionary? input, diff --git a/src/modules/Elsa.Http/Handlers/DefaultHttpEndpointWorkflowFaultHandler.cs b/src/modules/Elsa.Http/Handlers/DefaultHttpEndpointWorkflowFaultHandler.cs index 33ead3e13..6b6febefe 100644 --- a/src/modules/Elsa.Http/Handlers/DefaultHttpEndpointWorkflowFaultHandler.cs +++ b/src/modules/Elsa.Http/Handlers/DefaultHttpEndpointWorkflowFaultHandler.cs @@ -26,22 +26,16 @@ public class DefaultHttpEndpointWorkflowFaultHandler : IHttpEndpointWorkflowFaul public virtual async ValueTask HandleAsync(HttpEndpointFaultedWorkflowContext context) { var httpContext = context.HttpContext; - var workflowInstance = context.WorkflowInstance; - var fault = workflowInstance.WorkflowState.Fault!; + var workflowState = context.WorkflowState; + var fault = workflowState.Fault!; httpContext.Response.ContentType = MediaTypeNames.Application.Json; httpContext.Response.StatusCode = StatusCodes.Status500InternalServerError; var faultedResponse = _apiSerializer.Serialize(new { - errorMessage = $"Workflow faulted at {workflowInstance.FaultedAt!} with error: {fault.Message}", - exception = fault?.Exception, - workflow = new - { - name = workflowInstance.Name, - version = workflowInstance.Version, - instanceId = workflowInstance.Id - } + errorMessage = $"Workflow faulted with error: {fault.Message}", + workflowState = workflowState }); await httpContext.Response.WriteAsync(faultedResponse, context.CancellationToken); diff --git a/src/modules/Elsa.Http/Middleware/WorkflowsMiddleware.cs b/src/modules/Elsa.Http/Middleware/WorkflowsMiddleware.cs index b9bbb02e1..ad1f60e3a 100644 --- a/src/modules/Elsa.Http/Middleware/WorkflowsMiddleware.cs +++ b/src/modules/Elsa.Http/Middleware/WorkflowsMiddleware.cs @@ -13,6 +13,7 @@ using System.Net; using System.Net.Mime; using System.Text.Json; using Elsa.Extensions; +using Elsa.Workflows.Core.State; namespace Elsa.Http.Middleware; @@ -125,7 +126,13 @@ public class WorkflowsMiddleware return; // Process the trigger result by resuming each HTTP bookmark, if any. - await _httpBookmarkProcessor.ProcessBookmarks(new List { executionResult }, correlationId, input, cancellationToken); + var affectedWorkflowStates = await _httpBookmarkProcessor.ProcessBookmarks(new List { executionResult }, correlationId, input, cancellationToken); + + // Check if there were any errors. + var faultedWorkflowState = affectedWorkflowStates.FirstOrDefault(x => x.SubStatus == WorkflowSubStatus.Faulted); + + if (faultedWorkflowState != null) + await HandleWorkflowFaultAsync(httpContext, faultedWorkflowState, cancellationToken); } private string GetMatchingRoute(string path) @@ -202,18 +209,19 @@ public class WorkflowsMiddleware private async Task HandleWorkflowFaultAsync(HttpContext httpContext, WorkflowExecutionResult workflowExecutionResult, CancellationToken cancellationToken) { - var instanceFilter = new WorkflowInstanceFilter { Id = workflowExecutionResult.InstanceId }; - var workflowInstance = await _workflowInstanceStore.FindAsync(instanceFilter, cancellationToken); + var subStatus = workflowExecutionResult.SubStatus; - if (workflowInstance is not null - && workflowInstance.SubStatus == WorkflowSubStatus.Faulted - && !httpContext.Response.HasStarted) - { - await _httpEndpointWorkflowFaultHandler.HandleAsync(new HttpEndpointFaultedWorkflowContext(httpContext, workflowInstance, null, cancellationToken)); - return true; - } + if (subStatus != WorkflowSubStatus.Faulted || httpContext.Response.HasStarted) + return false; - return false; + var workflowState = (await _workflowRuntime.ExportWorkflowStateAsync(workflowExecutionResult.WorkflowInstanceId, cancellationToken))!; + return await HandleWorkflowFaultAsync(httpContext, workflowState, cancellationToken); + } + + private async Task HandleWorkflowFaultAsync(HttpContext httpContext, WorkflowState workflowState, CancellationToken cancellationToken) + { + await _httpEndpointWorkflowFaultHandler.HandleAsync(new HttpEndpointFaultedWorkflowContext(httpContext, workflowState, cancellationToken)); + return true; } private async Task AuthorizeAsync( diff --git a/src/modules/Elsa.Http/Models/HttpEndpointFaultedWorkflowContext.cs b/src/modules/Elsa.Http/Models/HttpEndpointFaultedWorkflowContext.cs index 605b85018..7f2721d8b 100644 --- a/src/modules/Elsa.Http/Models/HttpEndpointFaultedWorkflowContext.cs +++ b/src/modules/Elsa.Http/Models/HttpEndpointFaultedWorkflowContext.cs @@ -1,6 +1,12 @@ -using Elsa.Workflows.Management.Entities; +using Elsa.Workflows.Core.State; using Microsoft.AspNetCore.Http; namespace Elsa.Http.Models; -public record HttpEndpointFaultedWorkflowContext(HttpContext HttpContext, WorkflowInstance WorkflowInstance, Exception? Exception, CancellationToken CancellationToken); \ No newline at end of file +/// +/// Provides context about the faulted workflow. +/// +/// The HTTP context. +/// The faulted workflow state. +/// The cancellation token. +public record HttpEndpointFaultedWorkflowContext(HttpContext HttpContext, WorkflowState WorkflowState, CancellationToken CancellationToken); \ No newline at end of file diff --git a/src/modules/Elsa.Http/Services/HttpBookmarkProcessor.cs b/src/modules/Elsa.Http/Services/HttpBookmarkProcessor.cs index 3c323295d..c87761e65 100644 --- a/src/modules/Elsa.Http/Services/HttpBookmarkProcessor.cs +++ b/src/modules/Elsa.Http/Services/HttpBookmarkProcessor.cs @@ -1,6 +1,7 @@ using Elsa.Common.Models; using Elsa.Http.Contracts; using Elsa.Workflows.Core.Helpers; +using Elsa.Workflows.Core.State; using Elsa.Workflows.Runtime.Contracts; using Microsoft.AspNetCore.Http; @@ -30,7 +31,7 @@ public class HttpBookmarkProcessor : IHttpBookmarkProcessor } /// - public async Task ProcessBookmarks( + public async Task> ProcessBookmarks( IEnumerable executionResults, string? correlationId, IDictionary? input, @@ -51,9 +52,10 @@ public class HttpBookmarkProcessor : IHttpBookmarkProcessor from executionResult in executionResults from bookmark in executionResult.Bookmarks where bookmark.Name == writeHttpResponseTypeName || bookmark.Name == httpEndpointTypeName - select (executionResult.InstanceId, bookmark.Id); + select (InstanceId: executionResult.WorkflowInstanceId, bookmark.Id); var workflowExecutionResults = new Stack<(string InstanceId, string BookmarkId)>(query); + var workflowStates = new List(); while (workflowExecutionResults.TryPop(out var result)) { @@ -86,9 +88,13 @@ public class HttpBookmarkProcessor : IHttpBookmarkProcessor var workflowHost = await _workflowHostFactory.CreateAsync(workflow, workflowState, cancellationToken); var options = new ResumeWorkflowHostOptions(correlationId, result.BookmarkId, Input: input); await workflowHost.ResumeWorkflowAsync(options, cancellationToken); - + // Import the updated workflow state into the runtime. + workflowState = workflowHost.WorkflowState; await _workflowRuntime.ImportWorkflowStateAsync(workflowState, cancellationToken); + workflowStates.Add(workflowState); } + + return workflowStates; } } \ No newline at end of file diff --git a/src/modules/Elsa.ProtoActor/Extensions/ProtoStringExtensions.cs b/src/modules/Elsa.ProtoActor/Extensions/ProtoStringExtensions.cs index b182de34a..75feccca5 100644 --- a/src/modules/Elsa.ProtoActor/Extensions/ProtoStringExtensions.cs +++ b/src/modules/Elsa.ProtoActor/Extensions/ProtoStringExtensions.cs @@ -3,5 +3,5 @@ namespace Elsa.ProtoActor.Extensions; internal static class ProtoStringExtensions { public static string EmptyIfNull(this string? value) => value ?? ""; - public static string? NullIfEmpty(this string value) => value == "" ? default : value; + public static string? NullIfEmpty(this string? value) => value == "" ? default : value; } \ No newline at end of file diff --git a/src/modules/Elsa.ProtoActor/Features/ProtoActorFeature.cs b/src/modules/Elsa.ProtoActor/Features/ProtoActorFeature.cs index c33a290c9..9294788cc 100644 --- a/src/modules/Elsa.ProtoActor/Features/ProtoActorFeature.cs +++ b/src/modules/Elsa.ProtoActor/Features/ProtoActorFeature.cs @@ -4,6 +4,7 @@ using Elsa.Features.Attributes; using Elsa.Features.Services; using Elsa.ProtoActor.Grains; using Elsa.ProtoActor.HostedServices; +using Elsa.ProtoActor.Mappers; using Elsa.ProtoActor.Protos; using Elsa.ProtoActor.Services; using Elsa.Workflows.Core.Features; @@ -39,7 +40,7 @@ public class ProtoActorFeature : FeatureBase { // Configure default workflow execution pipeline suitable for Proto Actor. Module.UseWorkflows(workflows => workflows.WithProtoActorRuntimeWorkflowExecutionPipeline()); - + // Configure runtime with ProtoActor workflow runtime. Module.Configure().WorkflowRuntime = sp => ActivatorUtilities.CreateInstance(sp); } @@ -48,22 +49,22 @@ public class ProtoActorFeature : FeatureBase /// The name of the cluster to configure. /// public string ClusterName { get; set; } = "elsa-cluster"; - + /// /// A delegate that returns an instance of a concrete implementation of . /// public Func ClusterProvider { get; set; } = _ => new TestProvider(new TestProviderOptions(), new InMemAgent()); - + /// /// A delegate that configures an instance of . /// public Action ActorSystemConfig { get; set; } = SetupDefaultConfig; - + /// /// A delegate that configures an instance of an . /// public Action ActorSystem { get; set; } = (_, _) => { }; - + /// /// A delegate that returns an instance of to be used by the actor system. /// @@ -84,19 +85,19 @@ public class ProtoActorFeature : FeatureBase public override void Apply() { var services = Services; - + // Register ActorSystem. services.AddSingleton(sp => { var systemConfig = Proto.ActorSystemConfig .Setup() .WithMetrics(); - + var clusterProvider = ClusterProvider(sp); var system = new ActorSystem(systemConfig).WithServiceProvider(sp); var workflowGrainProps = system.DI().PropsFor(); var workflowRegistryGrainProps = system.DI().PropsFor(); - + var clusterConfig = ClusterConfig .Setup(ClusterName, clusterProvider, new PartitionIdentityLookup()) .WithHeartbeatExpiration(TimeSpan.FromDays(1)) @@ -114,17 +115,26 @@ public class ProtoActorFeature : FeatureBase system .WithRemote(remoteConfig) .WithCluster(clusterConfig); - - ActorSystem(sp, system); + + ActorSystem(sp, system); return system; }); // Logging. Log.SetLoggerFactory(LoggerFactory.Create(l => l.AddConsole().SetMinimumLevel(LogLevel.Warning))); - + // Persistence. services.AddSingleton(PersistenceProvider); - + + // Mappers. + services + .AddSingleton() + .AddSingleton() + .AddSingleton() + .AddSingleton() + .AddSingleton() + .AddSingleton(); + // Mediator handlers. services.AddHandlersFrom(); @@ -137,11 +147,11 @@ public class ProtoActorFeature : FeatureBase .AddTransient(sp => new RunningWorkflowsGrainActor((context, _) => ActivatorUtilities.CreateInstance(sp, context))) ; } - + private static void SetupDefaultConfig(IServiceProvider serviceProvider, ActorSystemConfig config) { } - + private static GrpcNetRemoteConfig CreateDefaultRemoteConfig(IServiceProvider serviceProvider) => GrpcNetRemoteConfig.BindToLocalhost() .WithProtoMessages(MessagesReflection.Descriptor) diff --git a/src/modules/Elsa.ProtoActor/Grains/WorkflowGrain.cs b/src/modules/Elsa.ProtoActor/Grains/WorkflowGrain.cs index d2304e12c..f6d0714b5 100644 --- a/src/modules/Elsa.ProtoActor/Grains/WorkflowGrain.cs +++ b/src/modules/Elsa.ProtoActor/Grains/WorkflowGrain.cs @@ -1,13 +1,16 @@ using Elsa.Common.Models; using Elsa.ProtoActor.Extensions; +using Elsa.ProtoActor.Mappers; using Elsa.ProtoActor.Protos; using Elsa.Workflows.Core.Contracts; -using Elsa.Workflows.Core.Models; using Elsa.Workflows.Core.State; using Elsa.Workflows.Runtime.Contracts; using Proto; using Proto.Cluster; using Proto.Persistence; +using Exception = System.Exception; +using WorkflowStatus = Elsa.Workflows.Core.Models.WorkflowStatus; +using ProtoBookmark = Elsa.ProtoActor.Protos.Bookmark; namespace Elsa.ProtoActor.Grains; @@ -19,8 +22,11 @@ public class WorkflowGrain : WorkflowGrainBase private const int MaxSnapshotsToKeep = 5; private readonly IWorkflowDefinitionService _workflowDefinitionService; private readonly IWorkflowHostFactory _workflowHostFactory; - private readonly IBookmarkPayloadSerializer _bookmarkPayloadSerializer; private readonly IWorkflowStateSerializer _workflowStateSerializer; + private readonly BookmarkMapper _bookmarkMapper; + private readonly WorkflowStatusMapper _workflowStatusMapper; + private readonly WorkflowSubStatusMapper _workflowSubStatusMapper; + private readonly WorkflowFaultStateMapper _workflowFaultStateMapper; private readonly Persistence _persistence; private string _definitionId = default!; @@ -34,15 +40,21 @@ public class WorkflowGrain : WorkflowGrainBase public WorkflowGrain( IWorkflowDefinitionService workflowDefinitionService, IWorkflowHostFactory workflowHostFactory, - IBookmarkPayloadSerializer bookmarkPayloadSerializer, IWorkflowStateSerializer workflowStateSerializer, IProvider provider, - IContext context) : base(context) + IContext context, + BookmarkMapper bookmarkMapper, + WorkflowStatusMapper workflowStatusMapper, + WorkflowSubStatusMapper workflowSubStatusMapper, + WorkflowFaultStateMapper workflowFaultStateMapper) : base(context) { _workflowDefinitionService = workflowDefinitionService; _workflowHostFactory = workflowHostFactory; - _bookmarkPayloadSerializer = bookmarkPayloadSerializer; _workflowStateSerializer = workflowStateSerializer; + _bookmarkMapper = bookmarkMapper; + _workflowStatusMapper = workflowStatusMapper; + _workflowSubStatusMapper = workflowSubStatusMapper; + _workflowFaultStateMapper = workflowFaultStateMapper; _persistence = Persistence.WithSnapshotting(provider, Context.ClusterIdentity()!.Identity, ApplySnapshot); } @@ -72,7 +84,6 @@ public class WorkflowGrain : WorkflowGrainBase { DefinitionId = workflow.Identity.DefinitionId, DefinitionVersion = workflow.Identity.Version, - //Bookmarks = _bookmarks }; } @@ -90,15 +101,15 @@ public class WorkflowGrain : WorkflowGrainBase var versionOptions = VersionOptions.FromString(request.VersionOptions); var cancellationToken = Context.CancellationToken; var startWorkflowOptions = new StartWorkflowHostOptions(instanceId, correlationId, input, request.TriggerActivityId); - + _workflowHost = await CreateWorkflowHostAsync(definitionId, versionOptions, cancellationToken); _version = _workflowHost.Workflow.Identity.Version; _definitionId = definitionId; _instanceId = instanceId; _input = input; - + var canStart = await _workflowHost.CanStartWorkflowAsync(startWorkflowOptions, cancellationToken); - + return new CanStartWorkflowResponse { CanStart = canStart @@ -106,7 +117,7 @@ public class WorkflowGrain : WorkflowGrainBase } /// - public override async Task Start(StartWorkflowRequest request) + public override async Task Start(StartWorkflowRequest request) { var definitionId = request.DefinitionId; var instanceId = request.InstanceId; @@ -134,15 +145,20 @@ public class WorkflowGrain : WorkflowGrainBase await SaveSnapshotAsync(); - return new StartWorkflowResponse + return new WorkflowExecutionResponse { Result = result, - Bookmarks = { Map(workflowState.Bookmarks) } + Bookmarks = { _bookmarkMapper.Map(workflowState.Bookmarks).ToList() }, + Status = _workflowStatusMapper.Map(workflowState.Status), + SubStatus = _workflowSubStatusMapper.Map(workflowState.SubStatus), + Fault = workflowState.Fault != null ? _workflowFaultStateMapper.Map(workflowState.Fault) : default, + TriggeredActivityId = string.Empty, + WorkflowInstanceId = instanceId }; } /// - public override async Task Resume(ResumeWorkflowRequest request) + public override async Task Resume(ResumeWorkflowRequest request) { _input = request.Input?.Deserialize(); var correlationId = request.CorrelationId; @@ -152,26 +168,26 @@ public class WorkflowGrain : WorkflowGrainBase var activityInstanceId = request.ActivityInstanceId.NullIfEmpty(); var activityHash = request.ActivityHash.NullIfEmpty(); var cancellationToken = Context.CancellationToken; - + var resumeWorkflowHostOptions = new ResumeWorkflowHostOptions( - correlationId, - bookmarkId, - activityId, + correlationId, + bookmarkId, + activityId, activityNodeId, activityInstanceId, activityHash, _input); - + var definitionId = _definitionId; var versionOptions = VersionOptions.SpecificVersion(_version); - + // Only need to reconstruct a workflow host if not already done so during CanStart. if (_workflowHost == null!) { _workflowHost = await CreateWorkflowHostAsync(definitionId, versionOptions, cancellationToken); _version = _workflowHost.Workflow.Identity.Version; } - + await _workflowHost.ResumeWorkflowAsync(resumeWorkflowHostOptions, cancellationToken); var finished = _workflowHost.WorkflowState.Status == WorkflowStatus.Finished; @@ -179,10 +195,15 @@ public class WorkflowGrain : WorkflowGrainBase await SaveSnapshotAsync(); - return new ResumeWorkflowResponse + return new WorkflowExecutionResponse { - Result = finished ? Protos.RunWorkflowResult.Finished : Protos.RunWorkflowResult.Suspended, - Bookmarks = { Map(_workflowHost.WorkflowState.Bookmarks) } + Result = finished ? RunWorkflowResult.Finished : RunWorkflowResult.Suspended, + Bookmarks = { _bookmarkMapper.Map(_workflowHost.WorkflowState.Bookmarks).ToList() }, + Fault = _workflowState.Fault != null ? _workflowFaultStateMapper.Map(_workflowState.Fault) : default, + TriggeredActivityId = string.Empty, + WorkflowInstanceId = _workflowState.Id, + Status = _workflowStatusMapper.Map(_workflowState.Status), + SubStatus = _workflowSubStatusMapper.Map(_workflowState.SubStatus) }; } @@ -218,6 +239,7 @@ public class WorkflowGrain : WorkflowGrainBase } private void ApplySnapshot(Snapshot snapshot) => (_definitionId, _instanceId, _version, _workflowState, _input) = (WorkflowSnapshot)snapshot.State; + private async Task SaveSnapshotAsync() { if (_workflowState.Status == WorkflowStatus.Finished) @@ -240,23 +262,4 @@ public class WorkflowGrain : WorkflowGrainBase var workflow = await _workflowDefinitionService.MaterializeWorkflowAsync(workflowDefinition, cancellationToken); return await _workflowHostFactory.CreateAsync(workflow, cancellationToken); } - - private IEnumerable Map(IEnumerable bookmarks) - { - return bookmarks.Select(x => - { - var payloadJson = x.Payload != null ? _bookmarkPayloadSerializer.Serialize(x.Payload) : ""; - return new BookmarkDto - { - Id = x.Id, - Name = x.Name, - ActivityNodeId = x.ActivityNodeId, - ActivityInstanceId = x.ActivityInstanceId, - Hash = x.Hash, - Data = payloadJson, - AutoBurn = x.AutoBurn, - CallbackMethodName = x.CallbackMethodName.EmptyIfNull() - }; - }); - } } \ No newline at end of file diff --git a/src/modules/Elsa.ProtoActor/Mappers/BookmarkMapper.cs b/src/modules/Elsa.ProtoActor/Mappers/BookmarkMapper.cs new file mode 100644 index 000000000..0a74ce8bb --- /dev/null +++ b/src/modules/Elsa.ProtoActor/Mappers/BookmarkMapper.cs @@ -0,0 +1,52 @@ +using Elsa.ProtoActor.Extensions; +using Elsa.Workflows.Core.Contracts; +using Elsa.Workflows.Core.Models; +using ProtoBookmark = Elsa.ProtoActor.Protos.Bookmark; + +namespace Elsa.ProtoActor.Mappers; + +/// +/// Maps between and . +/// +public class BookmarkMapper +{ + private readonly IBookmarkPayloadSerializer _bookmarkPayloadSerializer; + + /// + /// Initializes a new instance of . + /// + public BookmarkMapper(IBookmarkPayloadSerializer bookmarkPayloadSerializer) + { + _bookmarkPayloadSerializer = bookmarkPayloadSerializer; + } + + public Bookmark Map(ProtoBookmark bookmark) => + new(bookmark.Id, bookmark.Name, bookmark.Hash, bookmark.Payload, bookmark.ActivityNodeId, bookmark.ActivityInstanceId, bookmark.AutoBurn, bookmark.CallbackMethodName); + + + public IEnumerable Map(IEnumerable source) => + source.Select(x => + new ProtoBookmark + { + Id = x.Id, + Name = x.Name, + Hash = x.Hash, + Payload = x.Payload != null ? _bookmarkPayloadSerializer.Serialize(x.Payload) : string.Empty, + ActivityNodeId = x.ActivityNodeId, + ActivityInstanceId = x.ActivityInstanceId, + AutoBurn = x.AutoBurn, + CallbackMethodName = x.CallbackMethodName.NullIfEmpty() + }); + + public IEnumerable Map(IEnumerable source) => + source.Select(x => + new Bookmark( + x.Id, + x.Name, + x.Hash, + x.Payload.NullIfEmpty(), + x.ActivityNodeId, + x.ActivityInstanceId, + x.AutoBurn, + x.CallbackMethodName.NullIfEmpty())); +} \ No newline at end of file diff --git a/src/modules/Elsa.ProtoActor/Mappers/ExceptionMapper.cs b/src/modules/Elsa.ProtoActor/Mappers/ExceptionMapper.cs new file mode 100644 index 000000000..3983cddcf --- /dev/null +++ b/src/modules/Elsa.ProtoActor/Mappers/ExceptionMapper.cs @@ -0,0 +1,19 @@ +using Elsa.Workflows.Core.State; +using ProtoException = Elsa.ProtoActor.Protos.ExceptionState; + +namespace Elsa.ProtoActor.Mappers; + +public class ExceptionMapper +{ + public ExceptionState Map(ProtoException exception) => + new(Type.GetType(exception.Type)!, exception.Message, exception.StackTrace, exception.InnerException != null ? Map(exception.InnerException) : null); + + public ProtoException Map(ExceptionState exception) => + new() + { + Type = exception.Type.AssemblyQualifiedName, + Message = exception.Message, + StackTrace = exception.StackTrace, + InnerException = exception.InnerException != null ? Map(exception.InnerException) : null + }; +} \ No newline at end of file diff --git a/src/modules/Elsa.ProtoActor/Mappers/WorkflowExecutionResultMapper.cs b/src/modules/Elsa.ProtoActor/Mappers/WorkflowExecutionResultMapper.cs new file mode 100644 index 000000000..e1cf86be6 --- /dev/null +++ b/src/modules/Elsa.ProtoActor/Mappers/WorkflowExecutionResultMapper.cs @@ -0,0 +1,59 @@ +using Elsa.ProtoActor.Extensions; +using Elsa.Workflows.Runtime.Contracts; +using ProtoWorkflowExecutionResponse = Elsa.ProtoActor.Protos.WorkflowExecutionResponse; + +namespace Elsa.ProtoActor.Mappers; + +/// +/// Maps between and . +/// +public class WorkflowExecutionResultMapper +{ + private readonly WorkflowStatusMapper _workflowStatusMapper; + private readonly WorkflowSubStatusMapper _workflowSubStatusMapper; + private readonly BookmarkMapper _bookmarkMapper; + private readonly WorkflowFaultStateMapper _workflowFaultStateMapper; + + /// + /// Initializes a new instance of the class. + /// + public WorkflowExecutionResultMapper( + WorkflowStatusMapper workflowStatusMapper, + WorkflowSubStatusMapper workflowSubStatusMapper, + BookmarkMapper bookmarkMapper, + WorkflowFaultStateMapper workflowFaultStateMapper) + { + _workflowStatusMapper = workflowStatusMapper; + _workflowSubStatusMapper = workflowSubStatusMapper; + _bookmarkMapper = bookmarkMapper; + _workflowFaultStateMapper = workflowFaultStateMapper; + } + + /// + /// Maps a to a . + /// + /// The source. + /// The mapped . + public WorkflowExecutionResult Map(ProtoWorkflowExecutionResponse source) => new( + source.WorkflowInstanceId, + _workflowStatusMapper.Map(source.Status), + _workflowSubStatusMapper.Map(source.SubStatus), + _bookmarkMapper.Map(source.Bookmarks).ToList(), + source.TriggeredActivityId.NullIfEmpty(), + source.Fault != null ? _workflowFaultStateMapper.Map(source.Fault) : default); + + /// + /// Maps a to a . + /// + /// The source. + /// The mapped . + public ProtoWorkflowExecutionResponse Map(WorkflowExecutionResult source) => new() + { + WorkflowInstanceId = source.WorkflowInstanceId, + Status = _workflowStatusMapper.Map(source.Status), + SubStatus = _workflowSubStatusMapper.Map(source.SubStatus), + Bookmarks = {_bookmarkMapper.Map(source.Bookmarks)}, + TriggeredActivityId = source.TriggeredActivityId, + Fault = source.Fault != null ? _workflowFaultStateMapper.Map(source.Fault) : default + }; +} \ No newline at end of file diff --git a/src/modules/Elsa.ProtoActor/Mappers/WorkflowFaultStateMapper.cs b/src/modules/Elsa.ProtoActor/Mappers/WorkflowFaultStateMapper.cs new file mode 100644 index 000000000..4ec311a42 --- /dev/null +++ b/src/modules/Elsa.ProtoActor/Mappers/WorkflowFaultStateMapper.cs @@ -0,0 +1,41 @@ +using Elsa.Workflows.Core.State; +using ProtoWorkflowFault = Elsa.ProtoActor.Protos.WorkflowFault; + +namespace Elsa.ProtoActor.Mappers; + +/// +/// Maps between and . +/// +public class WorkflowFaultStateMapper +{ + private readonly ExceptionMapper _exceptionMapper; + + /// + /// Initializes a new instance of the class. + /// + public WorkflowFaultStateMapper(ExceptionMapper exceptionMapper) + { + _exceptionMapper = exceptionMapper; + } + + /// + /// Maps a to a . + /// + /// The source. + /// The mapped . + public WorkflowFaultState Map(ProtoWorkflowFault source) => + new(_exceptionMapper.Map(source.Exception), source.Message, source.FaultedActivityId); + + /// + /// Maps a to a . + /// + /// The source. + /// The mapped . + public ProtoWorkflowFault Map(WorkflowFaultState source) => + new() + { + Exception = source.Exception != null ? _exceptionMapper.Map(source.Exception) : default, + Message = source.Message, + FaultedActivityId = source.FaultedActivityId + }; +} \ No newline at end of file diff --git a/src/modules/Elsa.ProtoActor/Mappers/WorkflowStatusMapper.cs b/src/modules/Elsa.ProtoActor/Mappers/WorkflowStatusMapper.cs new file mode 100644 index 000000000..703a4bb82 --- /dev/null +++ b/src/modules/Elsa.ProtoActor/Mappers/WorkflowStatusMapper.cs @@ -0,0 +1,23 @@ +using Elsa.Workflows.Core.Models; +using ProtoWorkflowStatus = Elsa.ProtoActor.Protos.WorkflowStatus; + +namespace Elsa.ProtoActor.Mappers; + +public class WorkflowStatusMapper +{ + public WorkflowStatus Map(ProtoWorkflowStatus status) => + status switch + { + ProtoWorkflowStatus.Finished => WorkflowStatus.Finished, + ProtoWorkflowStatus.Running => WorkflowStatus.Running, + _ => throw new ArgumentOutOfRangeException(nameof(status), status, null) + }; + + public ProtoWorkflowStatus Map(WorkflowStatus status) => + status switch + { + WorkflowStatus.Finished => ProtoWorkflowStatus.Finished, + WorkflowStatus.Running => ProtoWorkflowStatus.Running, + _ => throw new ArgumentOutOfRangeException(nameof(status), status, null) + }; +} \ No newline at end of file diff --git a/src/modules/Elsa.ProtoActor/Mappers/WorkflowSubStatusMapper.cs b/src/modules/Elsa.ProtoActor/Mappers/WorkflowSubStatusMapper.cs new file mode 100644 index 000000000..4728e78e7 --- /dev/null +++ b/src/modules/Elsa.ProtoActor/Mappers/WorkflowSubStatusMapper.cs @@ -0,0 +1,29 @@ +using Elsa.Workflows.Core.Models; +using ProtoWorkflowSubStatus = Elsa.ProtoActor.Protos.WorkflowSubStatus; + +namespace Elsa.ProtoActor.Mappers; + +public class WorkflowSubStatusMapper +{ + public WorkflowSubStatus Map(ProtoWorkflowSubStatus subStatus) => + subStatus switch + { + ProtoWorkflowSubStatus.Faulted => WorkflowSubStatus.Faulted, + ProtoWorkflowSubStatus.Finished => WorkflowSubStatus.Finished, + ProtoWorkflowSubStatus.Cancelled => WorkflowSubStatus.Cancelled, + ProtoWorkflowSubStatus.Executing => WorkflowSubStatus.Executing, + ProtoWorkflowSubStatus.Suspended => WorkflowSubStatus.Suspended, + _ => throw new ArgumentOutOfRangeException(nameof(subStatus), subStatus, null) + }; + + public ProtoWorkflowSubStatus Map(WorkflowSubStatus subStatus) => + subStatus switch + { + WorkflowSubStatus.Faulted => ProtoWorkflowSubStatus.Faulted, + WorkflowSubStatus.Finished => ProtoWorkflowSubStatus.Finished, + WorkflowSubStatus.Cancelled => ProtoWorkflowSubStatus.Cancelled, + WorkflowSubStatus.Executing => ProtoWorkflowSubStatus.Executing, + WorkflowSubStatus.Suspended => ProtoWorkflowSubStatus.Suspended, + _ => throw new ArgumentOutOfRangeException(nameof(subStatus), subStatus, null) + }; +} \ No newline at end of file diff --git a/src/modules/Elsa.ProtoActor/Protos/Grains.proto b/src/modules/Elsa.ProtoActor/Protos/Grains.proto index a684f42af..bc6d269ee 100644 --- a/src/modules/Elsa.ProtoActor/Protos/Grains.proto +++ b/src/modules/Elsa.ProtoActor/Protos/Grains.proto @@ -7,8 +7,8 @@ import "Messages.proto"; service WorkflowGrain { rpc CanStart (StartWorkflowRequest) returns (CanStartWorkflowResponse); - rpc Start (StartWorkflowRequest) returns (StartWorkflowResponse); - rpc Resume (ResumeWorkflowRequest) returns (ResumeWorkflowResponse); + rpc Start (StartWorkflowRequest) returns (WorkflowExecutionResponse); + rpc Resume (ResumeWorkflowRequest) returns (WorkflowExecutionResponse); rpc ExportState(ExportWorkflowStateRequest) returns (ExportWorkflowStateResponse); rpc ImportState(ImportWorkflowStateRequest) returns (ImportWorkflowStateResponse); } diff --git a/src/modules/Elsa.ProtoActor/Protos/Messages.proto b/src/modules/Elsa.ProtoActor/Protos/Messages.proto index 23f7901a8..5d726d4a7 100644 --- a/src/modules/Elsa.ProtoActor/Protos/Messages.proto +++ b/src/modules/Elsa.ProtoActor/Protos/Messages.proto @@ -27,9 +27,40 @@ message StartWorkflowRequest { optional string TriggerActivityId = 6; } -message StartWorkflowResponse { - RunWorkflowResult Result = 1; - repeated BookmarkDto Bookmarks = 2; +message WorkflowExecutionResponse { + string WorkflowInstanceId = 1; + RunWorkflowResult Result = 2; + WorkflowStatus Status = 3; + WorkflowSubStatus SubStatus = 4; + repeated Bookmark Bookmarks = 5; + optional WorkflowFault Fault = 6; + optional string TriggeredActivityId = 7; +} + +message WorkflowFault { + optional ExceptionState Exception = 1; + string Message = 2; + optional string FaultedActivityId = 3; +} + +message ExceptionState { + string Type = 1; + string Message = 2; + optional string StackTrace = 3; + optional ExceptionState InnerException = 4; +} + +enum WorkflowStatus { + WorkflowStatusRunning = 0; + WorkflowStatusFinished = 1; +} + +enum WorkflowSubStatus { + WorkflowSubStatusExecuting = 0; + WorkflowSubStatusSuspended = 1; + WorkflowSubStatusFinished = 2; + WorkflowSubStatusCancelled = 3; + WorkflowSubStatusFaulted = 4; } message ResumeWorkflowRequest { @@ -43,11 +74,6 @@ message ResumeWorkflowRequest { optional Input input = 8; } -message ResumeWorkflowResponse { - RunWorkflowResult Result = 1; - repeated BookmarkDto Bookmarks = 2; -} - message ExportWorkflowStateRequest {} message ExportWorkflowStateResponse { @@ -65,11 +91,11 @@ enum RunWorkflowResult { RunWorkflowResultSuspended = 1; } -message BookmarkDto { +message Bookmark { string Id = 1; string Name = 2; string Hash = 3; - optional string Data = 4; + optional string Payload = 4; string ActivityNodeId = 5; string ActivityInstanceId = 6; optional bool AutoBurn = 7; diff --git a/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs b/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs index 26135929f..363a534df 100644 --- a/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs +++ b/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs @@ -1,13 +1,20 @@ using Elsa.Common.Models; using Elsa.Extensions; using Elsa.ProtoActor.Extensions; +using Elsa.ProtoActor.Mappers; using Elsa.ProtoActor.Protos; using Elsa.Workflows.Core.Contracts; -using Elsa.Workflows.Core.Models; using Elsa.Workflows.Core.State; using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Entities; using Proto.Cluster; +using Bookmark = Elsa.Workflows.Core.Models.Bookmark; +using ProtoWorkflowStatus = Elsa.ProtoActor.Protos.WorkflowStatus; +using ProtoWorkflowSubStatus = Elsa.ProtoActor.Protos.WorkflowSubStatus; +using ProtoWorkflowFault = Elsa.ProtoActor.Protos.WorkflowFault; +using ProtoException = Elsa.ProtoActor.Protos.ExceptionState; +using ProtoWorkflowExecutionResponse = Elsa.ProtoActor.Protos.WorkflowExecutionResponse; +using ProtoBookmark = Elsa.ProtoActor.Protos.Bookmark; namespace Elsa.ProtoActor.Services; @@ -24,6 +31,7 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime private readonly IBookmarkHasher _hasher; private readonly IWorkflowDefinitionService _workflowDefinitionService; private readonly IWorkflowInstanceFactory _workflowInstanceFactory; + private readonly WorkflowExecutionResultMapper _workflowExecutionResultMapper; /// /// Constructor. @@ -36,7 +44,8 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime IIdentityGenerator identityGenerator, IBookmarkHasher hasher, IWorkflowDefinitionService workflowDefinitionService, - IWorkflowInstanceFactory workflowInstanceFactory) + IWorkflowInstanceFactory workflowInstanceFactory, + WorkflowExecutionResultMapper workflowExecutionResultMapper) { _cluster = cluster; _workflowStateSerializer = workflowStateSerializer; @@ -46,6 +55,7 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime _hasher = hasher; _workflowDefinitionService = workflowDefinitionService; _workflowInstanceFactory = workflowInstanceFactory; + _workflowExecutionResultMapper = workflowExecutionResultMapper; } /// @@ -104,9 +114,8 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId); var response = await client.Start(request, cancellationToken); - var bookmarks = Map(response!.Bookmarks).ToList(); - return new WorkflowExecutionResult(workflowInstanceId, bookmarks); + return _workflowExecutionResultMapper.Map(response!); } /// @@ -128,15 +137,14 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime continue; var startResult = await StartWorkflowAsync(definitionId, startOptions, cancellationToken); - - results.Add(new WorkflowExecutionResult(startResult.InstanceId, startResult.Bookmarks)); + results.Add(startResult); } return results; } /// - public async Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default) + public async Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default) { var request = new ResumeWorkflowRequest { @@ -149,9 +157,8 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId); var response = await client.Resume(request, cancellationToken); - var bookmarks = Map(response!.Bookmarks).ToList(); - return new ResumeWorkflowResult(bookmarks); + return _workflowExecutionResultMapper.Map(response!); } /// @@ -164,7 +171,7 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime var bookmarks = await _bookmarkStore.FindManyAsync(filter, cancellationToken); return await ResumeWorkflowsAsync(bookmarks, new ResumeWorkflowRuntimeOptions(correlationId, Input: options.Input), cancellationToken); } - + /// public async Task TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsRuntimeOptions options, CancellationToken cancellationToken = default) { @@ -182,19 +189,16 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime { var startOptions = new StartWorkflowRuntimeOptions(collectedStartableWorkflow.CorrelationId, input, VersionOptions.Published, collectedStartableWorkflow.ActivityId, collectedStartableWorkflow.WorkflowInstanceId); - var startResult = await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions, cancellationToken); - return new WorkflowExecutionResult(startResult.InstanceId, startResult.Bookmarks); + return await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions, cancellationToken); } var collectedResumableWorkflow = (match as ResumableWorkflowMatch)!; var runtimeOptions = new ResumeWorkflowRuntimeOptions(collectedResumableWorkflow.CorrelationId, Input: input); - var resumeResult = await ResumeWorkflowAsync( + return await ResumeWorkflowAsync( match.WorkflowInstanceId, runtimeOptions with { BookmarkId = collectedResumableWorkflow.BookmarkId }, cancellationToken); - - return new WorkflowExecutionResult(collectedResumableWorkflow.WorkflowInstanceId, resumeResult.Bookmarks); } /// @@ -275,7 +279,7 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime runtimeOptions with { BookmarkId = bookmark.BookmarkId }, cancellationToken); - resumedWorkflows.Add(new WorkflowExecutionResult(workflowInstanceId, resumeResult.Bookmarks)); + resumedWorkflows.Add(resumeResult); } return resumedWorkflows; @@ -299,18 +303,6 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime } } - private static IEnumerable Map(IEnumerable source) => - source.Select(x => - new Bookmark( - x.Id, - x.Name, - x.Hash, - x.Data.NullIfEmpty(), - x.ActivityNodeId, - x.ActivityInstanceId, - x.AutoBurn, - x.CallbackMethodName.NullIfEmpty())); - private async Task> FindStartableWorkflowsAsync(WorkflowsFilter workflowsFilter, CancellationToken cancellationToken) { var hash = _hasher.Hash(workflowsFilter.ActivityTypeName, workflowsFilter.BookmarkPayload); diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Execute/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Execute/Endpoint.cs index f8271e814..dc02f607c 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Execute/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Execute/Endpoint.cs @@ -1,9 +1,14 @@ +using System.Net.Mime; using Elsa.Abstractions; using Elsa.Common.Models; using Elsa.Http.Contracts; +using Elsa.Workflows.Core.Contracts; +using Elsa.Workflows.Core.Models; +using Elsa.Workflows.Core.State; using Elsa.Workflows.Management.Contracts; using Elsa.Workflows.Runtime.Contracts; using JetBrains.Annotations; +using Microsoft.AspNetCore.Http; namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Execute; @@ -16,13 +21,15 @@ public class Execute : ElsaEndpoint private readonly IWorkflowDefinitionStore _store; private readonly IWorkflowRuntime _workflowRuntime; private readonly IHttpBookmarkProcessor _httpBookmarkProcessor; + private IApiSerializer _apiSerializer; /// - public Execute(IWorkflowDefinitionStore store, IWorkflowRuntime workflowRuntime, IHttpBookmarkProcessor httpBookmarkProcessor) + public Execute(IWorkflowDefinitionStore store, IWorkflowRuntime workflowRuntime, IHttpBookmarkProcessor httpBookmarkProcessor, IApiSerializer apiSerializer) { _store = store; _workflowRuntime = workflowRuntime; _httpBookmarkProcessor = httpBookmarkProcessor; + _apiSerializer = apiSerializer; } /// @@ -49,10 +56,39 @@ public class Execute : ElsaEndpoint var startWorkflowOptions = new StartWorkflowRuntimeOptions(correlationId, input, VersionOptions.Published); var result = await _workflowRuntime.StartWorkflowAsync(definitionId, startWorkflowOptions, cancellationToken); + // If a workflow fault occurred, respond appropriately with a 500 internal server error. + if (result.SubStatus == WorkflowSubStatus.Faulted) + { + var workflowState = await _workflowRuntime.ExportWorkflowStateAsync(result.WorkflowInstanceId, cancellationToken); + await HandleFaultAsync(workflowState!, cancellationToken); + return; + } + // Resume any HTTP bookmarks. await _httpBookmarkProcessor.ProcessBookmarks(new[] { result }, correlationId, default, cancellationToken); + + var workflowState2 = await _workflowRuntime.ExportWorkflowStateAsync(result.WorkflowInstanceId, cancellationToken); + + if (workflowState2!.SubStatus == WorkflowSubStatus.Faulted) + { + await HandleFaultAsync(workflowState2, cancellationToken); + return; + } if (!HttpContext.Response.HasStarted) - await SendOkAsync(new Response(result.InstanceId), cancellationToken); + await SendOkAsync(new Response(result.WorkflowInstanceId), cancellationToken); + } + + private async Task HandleFaultAsync(WorkflowState workflowState, CancellationToken cancellationToken) + { + var faultedResponse = _apiSerializer.Serialize(new + { + errorMessage = $"Workflow faulted with error: {workflowState.Fault!.Message}", + workflowState = workflowState + }); + + HttpContext.Response.ContentType = MediaTypeNames.Application.Json; + HttpContext.Response.StatusCode = StatusCodes.Status500InternalServerError; + await HttpContext.Response.WriteAsync(faultedResponse, cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs index adda24da9..6babb5e06 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs @@ -1,4 +1,5 @@ using Elsa.Common.Models; +using Elsa.Workflows.Core.Activities; using Elsa.Workflows.Core.Helpers; using Elsa.Workflows.Core.Models; using Elsa.Workflows.Core.State; @@ -46,7 +47,7 @@ public interface IWorkflowRuntime /// The ID of the workflow instance to resume. /// /// - Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default); + Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default); /// /// Resumes all workflows that are bookmarked on the specified activity type. @@ -117,13 +118,11 @@ public record ResumeWorkflowRuntimeOptions( public record CanStartWorkflowResult(string? InstanceId, bool CanStart); -public record ResumeWorkflowResult(ICollection Bookmarks); - public record TriggerWorkflowsRuntimeOptions(string? CorrelationId = default, string? WorkflowInstanceId = default, IDictionary? Input = default); public record TriggerWorkflowsResult(ICollection TriggeredWorkflows); -public record WorkflowExecutionResult(string InstanceId, ICollection Bookmarks, string? ActivityId = null); +public record WorkflowExecutionResult(string WorkflowInstanceId, WorkflowStatus Status, WorkflowSubStatus SubStatus, ICollection Bookmarks, string? TriggeredActivityId = null, WorkflowFaultState? Fault = default); public record UpdateBookmarksContext(string InstanceId, Diff Diff, string? CorrelationId); diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs index 8ae72249a..03825359f 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs @@ -75,7 +75,7 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime if (workflowDefinition == null) return null; - + return await StartWorkflowAsync(workflowDefinition, options, cancellationToken); } @@ -107,7 +107,7 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime continue; var startResult = await StartWorkflowAsync(definitionId, startOptions, cancellationToken); - results.Add(startResult with { ActivityId = trigger.ActivityId }); + results.Add(startResult with { TriggeredActivityId = trigger.ActivityId }); } } @@ -115,7 +115,7 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime } /// - public async Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default) + public async Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default) { await using (await _distributedLockProvider.AcquireLockAsync(workflowInstanceId, TimeSpan.FromMinutes(1), cancellationToken)) { @@ -134,8 +134,7 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime if (workflowDefinition == null) { - _logger.LogInformation("The workflow definition {DefinitionId} version {Version} was not found", definitionId, version); - return new ResumeWorkflowResult(Array.Empty()); + throw new Exception($"The workflow definition {definitionId} version {version} was not found"); } var workflow = await _workflowDefinitionService.MaterializeWorkflowAsync(workflowDefinition, cancellationToken); @@ -156,7 +155,7 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime await SaveWorkflowStateAsync(workflowState, cancellationToken); - return new ResumeWorkflowResult(workflowState.Bookmarks); + return new WorkflowExecutionResult(workflowState.Id, workflowState.Status, workflowState.SubStatus, workflowState.Bookmarks); } } @@ -188,17 +187,16 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime var startOptions = new StartWorkflowRuntimeOptions(collectedStartableWorkflow.CorrelationId, input, VersionOptions.Published, collectedStartableWorkflow.ActivityId, collectedStartableWorkflow.WorkflowInstanceId); var startResult = await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions, cancellationToken); - return startResult with { ActivityId = collectedStartableWorkflow.ActivityId }; + return startResult with { TriggeredActivityId = collectedStartableWorkflow.ActivityId }; } var collectedResumableWorkflow = (match as ResumableWorkflowMatch)!; var runtimeOptions = new ResumeWorkflowRuntimeOptions(collectedResumableWorkflow.CorrelationId, Input: input); - var resumeResult = await ResumeWorkflowAsync( + + return await ResumeWorkflowAsync( match.WorkflowInstanceId, runtimeOptions with { BookmarkId = collectedResumableWorkflow.BookmarkId }, cancellationToken); - - return new WorkflowExecutionResult(collectedResumableWorkflow.WorkflowInstanceId, resumeResult.Bookmarks); } /// @@ -248,14 +246,20 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime await SaveWorkflowStateAsync(workflowState, cancellationToken); - return new WorkflowExecutionResult(workflowState.Id, workflowState.Bookmarks); + return new WorkflowExecutionResult( + workflowState.Id, + workflowState.Status, + workflowState.SubStatus, + workflowState.Bookmarks, + default, + workflowState.Fault); } - + private async Task FindWorkflowDefinitionAsync(string definitionId, VersionOptions versionOptions, CancellationToken cancellationToken) { return await _workflowDefinitionService.FindAsync(definitionId, versionOptions, cancellationToken); } - + private async Task CreateWorkflowHostAsync(string definitionId, StartWorkflowRuntimeOptions options, CancellationToken cancellationToken) { var versionOptions = options.VersionOptions; @@ -263,10 +267,10 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime if (workflowDefinition == null) throw new Exception("Specified workflow definition and version does not exist"); - + return await CreateWorkflowHostAsync(workflowDefinition, cancellationToken); } - + private async Task CreateWorkflowHostAsync(WorkflowDefinition workflowDefinition, CancellationToken cancellationToken) { var workflow = await _workflowDefinitionService.MaterializeWorkflowAsync(workflowDefinition, cancellationToken); @@ -286,7 +290,7 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime runtimeOptions with { BookmarkId = bookmark.BookmarkId }, cancellationToken); - resumedWorkflows.Add(new WorkflowExecutionResult(workflowInstanceId, resumeResult.Bookmarks)); + resumedWorkflows.Add(new WorkflowExecutionResult(workflowInstanceId, resumeResult.Status, resumeResult.SubStatus, resumeResult.Bookmarks)); } return resumedWorkflows;