SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Merge pull request #5860 from elsa-workflows/enh/engine_ex

Browse source 
Add engine exception handling middleware
This commit is contained in:
raymonddenhaan 2024-08-07 11:37:48 +02:00 committed by GitHub
parent 83d9f03288 4f90ee71d9
commit 1437635903
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 54 additions and 147 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

153
src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs
View file

@ -16,18 +16,14 @@ using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Workflows;
/// <summary>
/// A delegate entry that is used by activities to be notified when the activities they scheduled are completed.
/// </summary>
/// <param name="Owner">The activity scheduling the <see cref="Child"/> activity.</param>
/// <param name="Child">The child <see cref="IActivity"/> being scheduled.</param>
/// <param name="CompletionCallback">The <see cref="ActivityCompletionCallback"/> delegate to invoke when the scheduled <see cref="Child"/> activity completes.</param>
/// <param name="Tag">An optional tag.</param>
public record ActivityCompletionCallbackEntry(ActivityExecutionContext Owner, ActivityNode Child, ActivityCompletionCallback? CompletionCallback, object? Tag = default);
/// <summary>
/// Provides context to the currently executing workflow.
/// </summary>
[PublicAPI]
public partial class WorkflowExecutionContext : IExecutionContext
{
@ -40,9 +36,7 @@ public partial class WorkflowExecutionContext : IExecutionContext
private readonly IHasher _hasher;
private readonly Action<WorkflowExecutionContext>? _statusUpdatedCallback;
/// <summary>
/// Initializes a new instance of <see cref="WorkflowExecutionContext"/>.
/// </summary>
private WorkflowExecutionContext(
IServiceProvider serviceProvider,
WorkflowGraph workflowGraph,
@ -87,9 +81,7 @@ public partial class WorkflowExecutionContext : IExecutionContext
_cancellationRegistrations.Add(sysSource.Token.Register(CancelWorkflow));
}
/// <summary>
/// Creates a new <see cref="WorkflowExecutionContext"/> for the specified workflow.
/// </summary>
public static async Task<WorkflowExecutionContext> CreateAsync(
IServiceProvider serviceProvider,
WorkflowGraph workflowGraph,
@ -122,9 +114,7 @@ public partial class WorkflowExecutionContext : IExecutionContext
);
}
/// <summary>
/// Creates a new <see cref="WorkflowExecutionContext"/> for the specified workflow.
/// </summary>
public static async Task<WorkflowExecutionContext> CreateAsync(
IServiceProvider serviceProvider,
WorkflowGraph workflowGraph,
@ -159,9 +149,7 @@ public partial class WorkflowExecutionContext : IExecutionContext
return workflowExecutionContext;
}
/// <summary>
/// Creates a new <see cref="WorkflowExecutionContext"/> for the specified workflow.
/// </summary>
public static async Task<WorkflowExecutionContext> CreateAsync(
IServiceProvider serviceProvider,
WorkflowGraph workflowGraph,
@ -202,9 +190,7 @@ public partial class WorkflowExecutionContext : IExecutionContext
return workflowExecutionContext;
}
/// <summary>
/// Assigns the specified workflow to this workflow execution context.
/// </summary>
/// <param name="workflowGraph">The workflow graph to assign.</param>
public async Task SetWorkflowGraphAsync(WorkflowGraph workflowGraph)
{
@ -220,247 +206,157 @@ public partial class WorkflowExecutionContext : IExecutionContext
activityExecutionContext.Activity = workflowGraph.NodeIdLookup[activityExecutionContext.Activity.NodeId].Activity;
}
/// <summary>
/// Gets the <see cref="IServiceProvider"/>.
/// </summary>
public IServiceProvider ServiceProvider { get; }
/// <summary>
/// Gets the <see cref="IActivityRegistry"/>.
/// </summary>
public IActivityRegistry ActivityRegistry { get; }
/// <summary>
/// Gets the <see cref="IActivityRegistryLookupService"/>.
/// </summary>
public IActivityRegistryLookupService ActivityRegistryLookup { get; }
/// <summary>
/// Gets the workflow graph.
/// </summary>
public WorkflowGraph WorkflowGraph { get; private set; }
/// <summary>
/// The <see cref="Workflow"/> associated with the execution context.
/// </summary>
public Workflow Workflow => WorkflowGraph.Workflow;
/// <summary>
/// A graph of the workflow structure.
/// </summary>
public ActivityNode Graph => WorkflowGraph.Root;
/// <summary>
/// The current status of the workflow.
/// </summary>
/// The current status of the workflow.
public WorkflowStatus Status => GetMainStatus(SubStatus);
/// <summary>
/// The current sub status of the workflow.
/// </summary>
public WorkflowSubStatus SubStatus { get; internal set; }
/// <summary>
/// The root <see cref="MemoryRegister"/> associated with the execution context.
/// </summary>
public MemoryRegister MemoryRegister { get; private set; } = default!;
/// <summary>
/// A unique ID of the execution context.
/// </summary>
public string Id { get; set; }
/// <inheritdoc />
public IActivity Activity => Workflow;
/// <summary>
/// An application-specific identifier associated with the execution context.
/// </summary>
public string? CorrelationId { get; set; }
/// <summary>
/// The ID of the workflow instance that triggered this instance.
/// </summary>
public string? ParentWorkflowInstanceId { get; set; }
/// <summary>
/// The date and time the workflow execution context was created.
/// </summary>
public DateTimeOffset CreatedAt { get; set; }
/// <summary>
/// The date and time the workflow execution context was last updated.
/// </summary>
public DateTimeOffset UpdatedAt { get; set; }
/// <summary>
/// The date and time the workflow execution context has finished.
/// </summary>
public DateTimeOffset? FinishedAt { get; set; }
/// <summary>
/// Gets the clock used to determine the current time.
/// </summary>
public ISystemClock SystemClock { get; }
/// <summary>
/// A flattened list of <see cref="ActivityNode"/>s from the <see cref="Graph"/>.
/// </summary>
public IReadOnlyCollection<ActivityNode> Nodes => WorkflowGraph.Nodes.ToList();
/// <summary>
/// A map between activity IDs and <see cref="ActivityNode"/>s in the workflow graph.
/// </summary>
public IDictionary<string, ActivityNode> NodeIdLookup => WorkflowGraph.NodeIdLookup;
/// <summary>
/// A map between hashed activity node IDs and <see cref="ActivityNode"/>s in the workflow graph.
/// </summary>
public IDictionary<string, ActivityNode> NodeHashLookup => WorkflowGraph.NodeHashLookup;
/// <summary>
/// A map between <see cref="IActivity"/>s and <see cref="ActivityNode"/>s in the workflow graph.
/// </summary>
public IDictionary<IActivity, ActivityNode> NodeActivityLookup => WorkflowGraph.NodeActivityLookup;
/// <summary>
/// The <see cref="IActivityScheduler"/> for the execution context.
/// </summary>
public IActivityScheduler Scheduler { get; }
/// <summary>
/// Gets the <see cref="IIdentityGenerator"/>.
/// </summary>
public IIdentityGenerator IdentityGenerator { get; }
/// <summary>
/// A collection of collected bookmarks during workflow execution.
/// </summary>
public ICollection<Bookmark> Bookmarks { get; set; } = new List<Bookmark>();
/// <summary>
/// A dictionary of inputs provided at the start of the current workflow execution.
/// </summary>
public IDictionary<string, object> Input { get; set; }
/// <summary>
/// A dictionary of outputs provided by the current workflow execution.
/// </summary>
public IDictionary<string, object> Output { get; set; } = new Dictionary<string, object>();
/// <inheritdoc />
public IDictionary<string, object> Properties { get; set; }
/// <summary>
/// A dictionary that can be used by application code and middleware to store information and even services. Values do not need to be serializable.
/// All data will be gone once workflow execution completes.
/// </summary>
public IDictionary<object, object> TransientProperties { get; set; } = new Dictionary<object, object>();
/// <summary>
/// A collection of incidents that may have occurred during execution.
/// </summary>
public ICollection<ActivityIncident> Incidents { get; set; }
/// <summary>
/// The current <see cref="ExecuteActivityDelegate"/> delegate to invoke when executing the next activity.
/// </summary>
public ExecuteActivityDelegate? ExecuteDelegate { get; set; }
/// <summary>
/// Provides context about the bookmark that was used to resume workflow execution, if any.
/// </summary>
public ResumedBookmarkContext? ResumedBookmarkContext { get; set; }
/// <summary>
/// The ID of the activity associated with the trigger that caused this workflow execution, if any.
/// </summary>
public string? TriggerActivityId { get; set; }
/// <summary>
/// A set of cancellation tokens that can be used to cancel the workflow execution without cancelling system-level operations.
/// </summary>
public CancellationTokens CancellationTokens { get; }
/// <summary>
/// A list of <see cref="ActivityCompletionCallbackEntry"/> callbacks that are invoked when the associated child activity completes.
/// </summary>
public ICollection<ActivityCompletionCallbackEntry> CompletionCallbacks => new ReadOnlyCollection<ActivityCompletionCallbackEntry>(_completionCallbackEntries);
/// <summary>
/// A list of <see cref="ActivityExecutionContext"/>s that are currently active.
/// </summary>
public IReadOnlyCollection<ActivityExecutionContext> ActivityExecutionContexts
{
get => _activityExecutionContexts.ToList();
internal set => _activityExecutionContexts = value.ToList();
}
/// <summary>
/// The last execution log sequence number. This number is incremented every time a new entry is added to the execution log and is persisted alongside the workflow instance and restored when the workflow is resumed.
/// </summary>
public long ExecutionLogSequence { get; set; }
/// <summary>
/// A collection of execution log entries. This collection is flushed when the workflow execution context ends.
/// </summary>
public ICollection<WorkflowExecutionLogEntry> ExecutionLog { get; } = new List<WorkflowExecutionLogEntry>();
/// <summary>
/// The expression execution context for the current workflow execution.
/// </summary>
public ExpressionExecutionContext? ExpressionExecutionContext { get; private set; }
/// <inheritdoc />
public IEnumerable<Variable> Variables => Workflow.Variables;
/// <summary>
/// Resolves the specified service type from the service provider.
/// </summary>
public T GetRequiredService<T>() where T : notnull => ServiceProvider.GetRequiredService<T>();
/// <summary>
/// Resolves the specified service type from the service provider.
/// </summary>
public object GetRequiredService(Type serviceType) => ServiceProvider.GetRequiredService(serviceType);
/// <summary>
/// Resolves the specified service type from the service provider, or creates a new instance if the service type was not found in the service container.
/// </summary>
public T GetOrCreateService<T>() where T : notnull => ActivatorUtilities.GetServiceOrCreateInstance<T>(ServiceProvider);
/// <summary>
/// Resolves the specified service type from the service provider, or creates a new instance if the service type was not found in the service container.
/// </summary>
public object GetOrCreateService(Type serviceType) => ActivatorUtilities.GetServiceOrCreateInstance(ServiceProvider, serviceType);
/// <summary>
/// Resolves the specified service type from the service provider.
/// </summary>
public T? GetService<T>() where T : notnull => ServiceProvider.GetService<T>();
/// <summary>
/// Resolves the specified service type from the service provider.
/// </summary>
public object? GetService(Type serviceType) => ServiceProvider.GetService(serviceType);
/// <summary>
/// Resolves multiple implementations of the specified service type from the service provider.
/// </summary>
public IEnumerable<T> GetServices<T>() where T : notnull => ServiceProvider.GetServices<T>();
/// <summary>
/// Registers a completion callback for the specified activity.
/// </summary>
internal void AddCompletionCallback(ActivityExecutionContext owner, ActivityNode child, ActivityCompletionCallback? completionCallback = default, object? tag = default)
{
var entry = new ActivityCompletionCallbackEntry(owner, child, completionCallback, tag);
_completionCallbackEntries.Add(entry);
}
/// <summary>
/// Unregisters the completion callback for the specified owner and child activity.
/// </summary>
internal ActivityCompletionCallbackEntry? PopCompletionCallback(ActivityExecutionContext owner, ActivityNode child)
{
var entry = _completionCallbackEntries.FirstOrDefault(x => x.Owner == owner && x.Child == child);
@ -480,61 +376,41 @@ public partial class WorkflowExecutionContext : IExecutionContext
_completionCallbackEntries.Remove(entry);
}
/// <summary>
/// Returns the <see cref="ActivityNode"/> with the specified activity ID from the workflow graph.
/// </summary>
public ActivityNode? FindNodeById(string nodeId) => NodeIdLookup.TryGetValue(nodeId, out var node) ? node : default;
/// <summary>
/// Returns the <see cref="ActivityNode"/> with the specified hash of the activity node ID from the workflow graph.
/// </summary>
/// <param name="hash">The hash of the activity node ID.</param>
/// <returns>The <see cref="ActivityNode"/> with the specified hash of the activity node ID.</returns>
public ActivityNode? FindNodeByHash(string hash) => NodeHashLookup.TryGetValue(hash, out var node) ? node : default;
/// <summary>
/// Returns the <see cref="ActivityNode"/> containing the specified activity from the workflow graph.
/// </summary>
public ActivityNode? FindNodeByActivity(IActivity activity)
{
return NodeActivityLookup.TryGetValue(activity, out var node) ? node : default;
}
/// <summary>
/// Returns the <see cref="ActivityNode"/> associated with the specified activity ID.
/// </summary>
public ActivityNode? FindNodeByActivityId(string activityId) => Nodes.FirstOrDefault(x => x.Activity.Id == activityId);
/// <summary>
/// Returns the <see cref="IActivity"/> with the specified ID from the workflow graph.
/// </summary>
public IActivity? FindActivityByNodeId(string nodeId) => FindNodeById(nodeId)?.Activity;
/// <summary>
/// Returns the <see cref="IActivity"/> with the specified ID from the workflow graph.
/// </summary>
public IActivity? FindActivityById(string activityId) => FindNodeById(NodeIdLookup.SingleOrDefault(n => n.Key.EndsWith(activityId)).Value.NodeId)?.Activity;
/// <summary>
/// Returns the <see cref="IActivity"/> with the specified hash of the activity node ID from the workflow graph.
/// </summary>
/// <param name="hash">The hash of the activity node ID.</param>
/// <returns>The <see cref="IActivity"/> with the specified hash of the activity node ID.</returns>
public IActivity? FindActivityByHash(string hash) => FindNodeByHash(hash)?.Activity;
/// <summary>
/// Returns a custom property with the specified key from the <see cref="Properties"/> dictionary.
/// </summary>
public T? GetProperty<T>(string key) => Properties.TryGetValue(key, out var value) ? value.ConvertTo<T>() : default;
/// <summary>
/// Sets a custom property with the specified key on the <see cref="Properties"/> dictionary.
/// </summary>
public void SetProperty<T>(string key, T value) => Properties[key] = value!;
/// <summary>
/// Updates a custom property with the specified key on the <see cref="Properties"/> dictionary.
/// </summary>
public T UpdateProperty<T>(string key, Func<T?, T> updater)
{
var value = GetProperty<T?>(key);
@ -543,16 +419,14 @@ public partial class WorkflowExecutionContext : IExecutionContext
return value;
}
/// <summary>
/// Returns true if the <see cref="Properties"/> dictionary contains the specified key.
/// </summary>
public bool HasProperty(string name) => Properties.ContainsKey(name);
internal bool CanTransitionTo(WorkflowSubStatus targetSubStatus) => ValidateStatusTransition(targetSubStatus);
internal bool CanTransitionTo(WorkflowSubStatus targetSubStatus) => ValidateStatusTransition();
internal void TransitionTo(WorkflowSubStatus subStatus)
{
if (!ValidateStatusTransition(SubStatus))
if (!ValidateStatusTransition())
throw new Exception($"Cannot transition from {SubStatus} to {subStatus}");
SubStatus = subStatus;
@ -561,7 +435,7 @@ public partial class WorkflowExecutionContext : IExecutionContext
if (Status == WorkflowStatus.Finished)
FinishedAt = UpdatedAt;
//For now only trigger on Cancelled, since the other statuses are handling via the host/runner
//For now only trigger on Cancelled, since the other statuses are handled via the host/runner
if (SubStatus == WorkflowSubStatus.Cancelled
&& _statusUpdatedCallback is not null)
_statusUpdatedCallback(this);
@ -576,9 +450,7 @@ public partial class WorkflowExecutionContext : IExecutionContext
}
}
/// <summary>
/// Creates a new <see cref="ActivityExecutionContext"/> for the specified activity.
/// </summary>
public async Task<ActivityExecutionContext> CreateActivityExecutionContextAsync(IActivity activity, ActivityInvocationOptions? options = default)
{
var activityDescriptor = await ActivityRegistryLookup.FindAsync(activity) ?? throw new ActivityNotFoundException(activity.Type);
@ -616,35 +488,23 @@ public partial class WorkflowExecutionContext : IExecutionContext
return activityExecutionContext;
}
/// <summary>
/// Returns a register of recorded activity output.
/// </summary>
public ActivityOutputRegister GetActivityOutputRegister() => TransientProperties.GetOrAdd(ActivityOutputRegistryKey, () => new ActivityOutputRegister());
/// <summary>
/// Returns the last activity result.
/// </summary>
public object? GetLastActivityResult() => TransientProperties.TryGetValue(LastActivityResultKey, out var value) ? value : default;
/// <summary>
/// Adds the specified <see cref="ActivityExecutionContext"/> to the workflow execution context.
/// </summary>
public void AddActivityExecutionContext(ActivityExecutionContext context) => _activityExecutionContexts.Add(context);
/// <summary>
/// Removes the specified <see cref="ActivityExecutionContext"/> from the workflow execution context.
/// </summary>
public void RemoveActivityExecutionContext(ActivityExecutionContext context) => _activityExecutionContexts.Remove(context);
/// <summary>
/// Removes the specified <see cref="ActivityExecutionContext"/> from the workflow execution context.
/// </summary>
/// <param name="predicate">The predicate used to filter the activity execution contexts to remove.</param>
public void RemoveActivityExecutionContext(Func<ActivityExecutionContext, bool> predicate) => _activityExecutionContexts.RemoveWhere(predicate);
/// <summary>
/// Records the output of the specified activity into the current workflow execution context.
/// </summary>
/// <param name="activityExecutionContext">The <see cref="ActivityExecutionContext"/> of the activity.</param>
/// <param name="outputName">The name of the output.</param>
/// <param name="value">The value of the output.</param>
@ -669,8 +529,9 @@ public partial class WorkflowExecutionContext : IExecutionContext
WorkflowSubStatus.Suspended => WorkflowStatus.Running,
_ => throw new ArgumentOutOfRangeException(nameof(subStatus), subStatus, null)
};
private bool ValidateStatusTransition(WorkflowSubStatus targetSubStatus)
// TODO: Check if we should not use the target subStatus here instead.
private bool ValidateStatusTransition()
{
var currentMainStatus = GetMainStatus(SubStatus);
return currentMainStatus != WorkflowStatus.Finished;

41
src/modules/Elsa.Workflows.Core/Middleware/Workflows/EngineExceptionHandlingMiddleware.cs Normal file
View file

@ -0,0 +1,41 @@
using Elsa.Common.Contracts;
using Elsa.Workflows.Contracts;
using Elsa.Workflows.Models;
using Elsa.Workflows.Pipelines.WorkflowExecution;
using Elsa.Workflows.State;
using Microsoft.Extensions.Logging;
namespace Elsa.Workflows.Middleware.Workflows;
/// Adds extension methods to <see cref="ExceptionHandlingMiddleware"/>.
public static class EngineExceptionHandlingMiddlewareExtensions
{
/// Installs the <see cref="ExceptionHandlingMiddleware"/> component in the activity execution pipeline.
public static IWorkflowExecutionPipelineBuilder UseEngineExceptionHandling(this IWorkflowExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.UseMiddleware<EngineExceptionHandlingMiddleware>();
}
/// Catches any exceptions thrown by downstream components and transitions the workflow into the faulted state.
public class EngineExceptionHandlingMiddleware(WorkflowMiddlewareDelegate next, ISystemClock systemClock, ILogger<EngineExceptionHandlingMiddleware> logger) : IWorkflowExecutionMiddleware
{
/// <inheritdoc />
public async ValueTask InvokeAsync(WorkflowExecutionContext context)
{
try
{
await next(context);
}
catch (Exception e)
{
logger.LogWarning(e, "An exception was caught from a downstream middleware component");
var exceptionState = ExceptionState.FromException(e);
var now = systemClock.UtcNow;
var activity = context.Workflow;
var incident = new ActivityIncident(activity.Id, activity.Type, e.Message, exceptionState, now);
// No state change as the workflow / activities status should be leading.
// We will however be adding an incident to make the issue visible.
context.Incidents.Add(incident);
context.AddExecutionLogEntry("Faulted", e.Message, exceptionState);
}
}
}

6
src/modules/Elsa.Workflows.Runtime/Elsa.Workflows.Runtime.csproj.DotSettings
View file

@ -1,2 +1,6 @@
<wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=constants/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=constants/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=contracts/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=extensions/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=models/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=services/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

1
src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowExecutionPipelineBuilderExtensions.cs
View file

@ -17,6 +17,7 @@ public static class WorkflowExecutionPipelineBuilderExtensions
public static IWorkflowExecutionPipelineBuilder UseDefaultPipeline(this IWorkflowExecutionPipelineBuilder pipelineBuilder) =>
pipelineBuilder
.Reset()
.UseEngineExceptionHandling()
.UseBackgroundActivities()
.UseBookmarkPersistence()
.UseActivityExecutionLogPersistence()