feat(bpmn): host port and command applier (#7942)
* feat(bpmn): host-side applier for the Bpmn.Semantics port Translates the interpreter's three host commands onto ActivityExecutionContext and feeds its four entry points, plus the minimum BpmnProcess container needed to exercise them end to end through IWorkflowRunner. StartWork schedules the bound activity, CancelWorkSubtree calls the public CancelActivityAsync extension (already recursive), and SignalEnclosingScope sends a BpmnScopeSignal up the ancestor chain. OnWorkFaulted rides the FaultSignal seam: it asks the interpreter what BPMN made of the fault and calls StopPropagation only on a Caught disposition, leaving a Propagated one strictly alone so an enclosing scope or the incident strategy takes it. A unit of work is keyed on the child ActivityExecutionContext.Id, recorded in the scope's own persisted ledger, never on Tag: the completion-callback dispatch rewrites the receiving context's Tag, so a nested scope wears a different tag than its parent remembers it by. Interpreter correlation travels on the child's context rather than on the shared activity instance. Evaluations go through one queue per workflow instance, so a scope signalled mid-apply is drained after the command list rather than re-entering the interpreter. Commands are applied in the order returned. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * test(bpmn): cover the teardown refusal path Adds a focused unit test that drives BpmnWorkTeardown.CancelSubtreeAsync into the NotSupportedException branch by constructing a real context tree with a scheduled-but-not-invoked descendant, so a regression that silently drops the detection is caught. Also records why BpmnWorkLedger's append-only, handle-keyed Records list cannot strand a context on a duplicate StartWork for a live (BindingRef, IterationId) slot, a case the port's own guarantee makes unreachable from this applier. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(bpmn): keep a refused teardown from stranding ledger state A subtree cancellation refused with NotSupportedException is absorbed into an incident under ContinueWithIncidentsStrategy rather than crashing, so the end-of-command ledger save was being skipped and the persisted ledger kept claiming work BPMN had just torn down. Save the ledger removal before the possible throw instead of after, so a later completion callback for the stranded activity finds no live record and is discarded instead of being fed to the interpreter. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Sipke Schoorstra
GitHub
parent
7389e0a674
commit
f99f37407c
77
src/modules/Elsa.Bpmn/Activities/BpmnProcess.cs
Normal file
77
src/modules/Elsa.Bpmn/Activities/BpmnProcess.cs
Normal file
|
|
@ -0,0 +1,77 @@
|
|||
using System.Runtime.CompilerServices;
|
||||
using Bpmn.Model;
|
||||
using Elsa.Bpmn.Hosting;
|
||||
using Elsa.Bpmn.Signals;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.Attributes;
|
||||
using Elsa.Workflows.Signals;
|
||||
|
||||
namespace Elsa.Bpmn.Activities;
|
||||
|
||||
/// <summary>
|
||||
/// Runs one BPMN process scope, driving the <c>Bpmn.Semantics</c> interpreter and applying what it returns onto this
|
||||
/// activity's execution context.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// A scope owns its own execution state and its own record of the work it started, both held in
|
||||
/// <see cref="ActivityExecutionContext.Properties"/>. A nested BPMN scope — an embedded subprocess, or an event
|
||||
/// subprocess body — is another <see cref="BpmnProcess"/> bound as work, so the scope hierarchy the interpreter has
|
||||
/// no view of is exactly the activity hierarchy Elsa already maintains.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Like every container, this one never auto-completes: it completes when the interpreter returns a <c>Complete</c>
|
||||
/// continuation, and its outcome is what a conditional sequence flow in the enclosing scope selects on.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
[Activity("Elsa", "BPMN", "Executes a BPMN process scope.")]
|
||||
[System.ComponentModel.Browsable(false)]
|
||||
public class BpmnProcess : Container
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public BpmnProcess([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line)
|
||||
{
|
||||
OnSignalReceived<BpmnScopeSignal>(OnScopeSignalledAsync);
|
||||
OnSignalReceived<FaultSignal>(OnWorkFaultedAsync);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The BPMN process definition this scope executes.
|
||||
/// </summary>
|
||||
public BpmnProcessDefinition? Process { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Maps each binding ref the definition declares to the id of the activity in <see cref="Container.Activities"/>
|
||||
/// that runs it.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The interpreter never parses a binding ref — it compares and echoes it — so resolving one to an actual timer,
|
||||
/// work item, HTTP call or nested process is entirely the host's.
|
||||
/// </remarks>
|
||||
public IDictionary<string, string> WorkBindings { get; set; } = new Dictionary<string, string>(StringComparer.Ordinal);
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override ValueTask ScheduleChildrenAsync(ActivityExecutionContext context) => BpmnScopeHost.For(context).StartAsync();
|
||||
|
||||
/// <summary>
|
||||
/// The activity bound to the given binding ref, or <c>null</c> when the definition declares a binding this
|
||||
/// activity does not map.
|
||||
/// </summary>
|
||||
internal IActivity? FindWorkActivity(string bindingRef) =>
|
||||
WorkBindings.TryGetValue(bindingRef, out var activityId)
|
||||
? Activities.FirstOrDefault(activity => string.Equals(activity.Id, activityId, StringComparison.Ordinal))
|
||||
: null;
|
||||
|
||||
/// <summary>
|
||||
/// A unit of work completed. Named rather than a lambda, because completion callbacks are rehydrated by method name.
|
||||
/// </summary>
|
||||
internal ValueTask OnWorkCompletedAsync(ActivityCompletedContext context) =>
|
||||
BpmnScopeHost.For(context.TargetContext).OnWorkCompletedAsync(context.ChildContext, context.Result);
|
||||
|
||||
private ValueTask OnScopeSignalledAsync(BpmnScopeSignal signal, SignalContext context) =>
|
||||
BpmnScopeHost.For(context.ReceiverActivityExecutionContext).OnScopeSignalledAsync(signal, context);
|
||||
|
||||
private ValueTask OnWorkFaultedAsync(FaultSignal signal, SignalContext context) =>
|
||||
BpmnScopeHost.For(context.ReceiverActivityExecutionContext).OnWorkFaultedAsync(signal, context);
|
||||
}
|
||||
4
src/modules/Elsa.Bpmn/AssemblyInfo.cs
Normal file
4
src/modules/Elsa.Bpmn/AssemblyInfo.cs
Normal file
|
|
@ -0,0 +1,4 @@
|
|||
using System.Runtime.CompilerServices;
|
||||
|
||||
[assembly: InternalsVisibleTo("Elsa.Bpmn.UnitTests")]
|
||||
[assembly: InternalsVisibleTo("Elsa.Bpmn.IntegrationTests")]
|
||||
29
src/modules/Elsa.Bpmn/Exceptions/BpmnScopeFaultException.cs
Normal file
29
src/modules/Elsa.Bpmn/Exceptions/BpmnScopeFaultException.cs
Normal file
|
|
@ -0,0 +1,29 @@
|
|||
namespace Elsa.Bpmn.Exceptions;
|
||||
|
||||
/// <summary>
|
||||
/// Thrown when the interpreter returns a <c>Fault</c> continuation for a BPMN scope: the scope failed
|
||||
/// deterministically and cannot continue.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// A propagated work fault does not surface this way. That case is handled where it arrives — inside the scope's
|
||||
/// <see cref="Elsa.Workflows.Signals.FaultSignal"/> handler, which leaves the signal alone so an enclosing scope or
|
||||
/// the incident strategy takes it. This exception covers the remaining ways a scope can fault, such as a deadlock
|
||||
/// the interpreter detects while routing a completion, where there is no signal in flight to leave alone.
|
||||
/// </remarks>
|
||||
public class BpmnScopeFaultException : Exception
|
||||
{
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="BpmnScopeFaultException"/> class.
|
||||
/// </summary>
|
||||
/// <param name="code">The interpreter's stable fault code.</param>
|
||||
/// <param name="message">The interpreter's human-readable explanation.</param>
|
||||
public BpmnScopeFaultException(string code, string message) : base($"{code}: {message}")
|
||||
{
|
||||
Code = code;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The interpreter's stable fault code.
|
||||
/// </summary>
|
||||
public string Code { get; }
|
||||
}
|
||||
|
|
@ -1,15 +1,26 @@
|
|||
using Elsa.Bpmn.Activities;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Features.Abstractions;
|
||||
using Elsa.Features.Attributes;
|
||||
using Elsa.Features.Services;
|
||||
using Elsa.Workflows.Management.Features;
|
||||
|
||||
namespace Elsa.Bpmn.Features;
|
||||
|
||||
/// <summary>
|
||||
/// Provides BPMN execution support to the system.
|
||||
/// </summary>
|
||||
[DependsOn(typeof(WorkflowManagementFeature))]
|
||||
public class BpmnFeature : FeatureBase
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public BpmnFeature(IModule module) : base(module)
|
||||
{
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public override void Apply()
|
||||
{
|
||||
Module.AddActivity<BpmnProcess>();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
152
src/modules/Elsa.Bpmn/Hosting/BpmnCommandApplier.cs
Normal file
152
src/modules/Elsa.Bpmn/Hosting/BpmnCommandApplier.cs
Normal file
|
|
@ -0,0 +1,152 @@
|
|||
using System.Text.Json;
|
||||
using Bpmn.Model;
|
||||
using Bpmn.Semantics;
|
||||
using Elsa.Bpmn.Activities;
|
||||
using Elsa.Bpmn.Signals;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Memory;
|
||||
using Elsa.Workflows.Options;
|
||||
|
||||
namespace Elsa.Bpmn.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// Translates the interpreter's three host commands onto <see cref="ActivityExecutionContext"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <list type="table">
|
||||
/// <item><term><c>StartWork</c></term><description><see cref="ActivityExecutionContext.ScheduleActivityAsync(IActivity?, ScheduleWorkOptions?)"/>.</description></item>
|
||||
/// <item><term><c>CancelWorkSubtree</c></term><description><c>CancelActivityAsync</c>, which already walks the child subtree recursively.</description></item>
|
||||
/// <item><term><c>SignalEnclosingScope</c></term><description><c>SendSignalAsync</c>, which bubbles to ancestors.</description></item>
|
||||
/// </list>
|
||||
/// </remarks>
|
||||
internal sealed class BpmnCommandApplier(ActivityExecutionContext scopeContext, BpmnProcess process, BpmnScopeMemory memory)
|
||||
{
|
||||
/// <summary>
|
||||
/// Applies a command list <b>in the order returned</b>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The ordering carries meaning and is not an implementation detail. An interrupting boundary event emits the
|
||||
/// boundary path's <c>StartWork</c> <i>before</i> the teardown that retires the host it interrupted, and a host
|
||||
/// that tidied up first would be applying a different process.
|
||||
/// </remarks>
|
||||
public async ValueTask ApplyAsync(IReadOnlyList<BpmnHostCommand> commands)
|
||||
{
|
||||
foreach (var command in commands)
|
||||
{
|
||||
switch (command)
|
||||
{
|
||||
case BpmnHostCommand.StartWork start:
|
||||
await StartWorkAsync(start);
|
||||
break;
|
||||
case BpmnHostCommand.CancelWorkSubtree cancel:
|
||||
await CancelWorkSubtreeAsync(cancel);
|
||||
break;
|
||||
case BpmnHostCommand.SignalEnclosingScope signal:
|
||||
await SignalEnclosingScopeAsync(signal);
|
||||
break;
|
||||
default:
|
||||
// The command hierarchy is closed, so this can only be reached by a library version that added a
|
||||
// command this host has never heard of. Refusing is the only honest answer: silently skipping it
|
||||
// would run a different process than the one the interpreter decided on.
|
||||
throw new NotSupportedException($"The BPMN host command '{command.GetType().Name}' is not supported by this host.");
|
||||
}
|
||||
|
||||
memory.SaveWork();
|
||||
}
|
||||
}
|
||||
|
||||
private async ValueTask StartWorkAsync(BpmnHostCommand.StartWork start)
|
||||
{
|
||||
var activity = process.FindWorkActivity(start.BindingRef)
|
||||
?? throw new InvalidOperationException(
|
||||
$"BPMN element '{start.ElementId}' binds work '{start.BindingRef}', which activity '{process.Id}' does not map to a child activity.");
|
||||
|
||||
var workflowExecutionContext = scopeContext.WorkflowExecutionContext;
|
||||
|
||||
// The child's context is created up front so that this scope has its id before the child ever runs, and can
|
||||
// key the unit of work on it. The alternative — recognising the child by ActivityExecutionContext.Tag — is
|
||||
// unsound across nested scopes, because the completion-callback dispatch rewrites the receiving context's Tag.
|
||||
var childContext = await workflowExecutionContext.CreateActivityExecutionContextAsync(activity, new ActivityInvocationOptions
|
||||
{
|
||||
Owner = scopeContext,
|
||||
Variables = BuildIterationVariables(start.IterationScope),
|
||||
SchedulingActivityExecutionId = scopeContext.Id
|
||||
});
|
||||
|
||||
// The correlation is opaque interpreter state that must travel with the work and, when the work is a nested
|
||||
// BPMN process, arrive there as its InvocationCorrelation. It goes on the child's own context rather than on
|
||||
// the activity instance: an activity object is shared by every concurrent execution of one definition, so
|
||||
// writing per-invocation state onto it corrupts as soon as two instances run at once.
|
||||
BpmnScopeMemory.Write(childContext, BpmnScopeHost.InvocationCorrelationPropertyKey, start.Correlation);
|
||||
|
||||
childContext.Taint();
|
||||
workflowExecutionContext.AddActivityExecutionContext(childContext);
|
||||
|
||||
// Recorded before scheduling, so the work is live from the moment anything could report against it. This
|
||||
// never checks for an existing record on the same (BindingRef, IterationId): the port guarantees the
|
||||
// interpreter never issues a second StartWork for a slot it already holds live, so a duplicate here would be
|
||||
// an interpreter contract breach, not a host-side race. Were it to happen anyway, Records is append-only and
|
||||
// keyed by handle rather than by slot, so nothing gets overwritten or stranded — the older record, and the
|
||||
// context behind it, stay exactly as reachable as before. What would go wrong is the snapshot then reporting
|
||||
// two live entries for one slot, which is the interpreter's invariant to keep, not this host's to enforce.
|
||||
memory.Work.Records.Add(new BpmnWorkRecord
|
||||
{
|
||||
Handle = memory.Work.NextHandle(),
|
||||
BindingRef = start.BindingRef,
|
||||
IterationId = start.IterationScope?.IterationId,
|
||||
ElementId = start.ElementId,
|
||||
ChildContextId = childContext.Id
|
||||
});
|
||||
|
||||
// A named instance method, not a lambda: completion callbacks are rehydrated by method name.
|
||||
await scopeContext.ScheduleActivityAsync(activity, new ScheduleWorkOptions
|
||||
{
|
||||
CompletionCallback = process.OnWorkCompletedAsync,
|
||||
ExistingActivityExecutionContext = childContext,
|
||||
SchedulingActivityExecutionId = scopeContext.Id
|
||||
});
|
||||
}
|
||||
|
||||
private async ValueTask CancelWorkSubtreeAsync(BpmnHostCommand.CancelWorkSubtree cancel)
|
||||
{
|
||||
// The interpreter only ever names a handle this scope reported as live, so a miss means the work has already
|
||||
// gone — a race BPMN produces routinely, and one the interpreter absorbs on the way back in.
|
||||
if (memory.Work.FindByHandle(cancel.Handle) is not { } record)
|
||||
return;
|
||||
|
||||
memory.Work.Remove(record);
|
||||
|
||||
// Saved here rather than left to the end-of-command save in ApplyAsync: CancelSubtreeAsync can refuse with a
|
||||
// NotSupportedException when the subtree still has scheduled-but-not-invoked work, and under the
|
||||
// continue-with-incidents strategy that throw is absorbed into an incident rather than left to crash the
|
||||
// process — so the end-of-command save would never run. Saving the removal now, before the possible throw,
|
||||
// keeps the persisted ledger from claiming work this scope just tore down, and is what lets a completion
|
||||
// callback that later arrives for the stranded activity find no live record and be discarded (see
|
||||
// BpmnScopeHost.OnWorkCompletedAsync) instead of being handed to the interpreter as real work.
|
||||
memory.SaveWork();
|
||||
|
||||
if (BpmnWorkTeardown.FindContext(scopeContext.WorkflowExecutionContext, record.ChildContextId) is not { } childContext)
|
||||
return;
|
||||
|
||||
await BpmnWorkTeardown.CancelSubtreeAsync(childContext, $"element '{cancel.ElementId}', {cancel.Reason}");
|
||||
}
|
||||
|
||||
private ValueTask SignalEnclosingScopeAsync(BpmnHostCommand.SignalEnclosingScope signal) =>
|
||||
scopeContext.SendSignalAsync(new BpmnScopeSignal(signal.Code, signal.Payload));
|
||||
|
||||
private static ICollection<Variable>? BuildIterationVariables(BpmnIterationScope? iterationScope) =>
|
||||
iterationScope?.Values.Select(value => new Variable(value.Key, ToClrValue(value.Value))).ToList();
|
||||
|
||||
private static object? ToClrValue(BpmnValue value) => value.Json is not { } json
|
||||
? null
|
||||
: json.ValueKind switch
|
||||
{
|
||||
JsonValueKind.String => json.GetString(),
|
||||
JsonValueKind.Number => json.TryGetInt64(out var integer) ? integer : json.GetDouble(),
|
||||
JsonValueKind.True => true,
|
||||
JsonValueKind.False => false,
|
||||
JsonValueKind.Null or JsonValueKind.Undefined => null,
|
||||
_ => json.GetRawText()
|
||||
};
|
||||
}
|
||||
60
src/modules/Elsa.Bpmn/Hosting/BpmnScopeDispatcher.cs
Normal file
60
src/modules/Elsa.Bpmn/Hosting/BpmnScopeDispatcher.cs
Normal file
|
|
@ -0,0 +1,60 @@
|
|||
namespace Elsa.Bpmn.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// Runs BPMN scope evaluations one at a time, in arrival order, for one workflow instance.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// A nested scope can terminalize while its parent is still applying a command list, and a scope raising an
|
||||
/// escalation delivers it to its parent from inside its own command loop. Either way the parent must not be
|
||||
/// evaluated on the spot: doing so re-enters the interpreter with a half-applied live-work set and an evaluation
|
||||
/// already in flight. Posting instead of calling is what keeps that from happening — a post made while the queue is
|
||||
/// draining is appended and picked up once the evaluation in flight has fully applied.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// One queue serves every scope in the instance, which is what makes the ordering total rather than per-scope.
|
||||
/// It lives in the workflow execution context's transient properties: it coordinates a single burst of execution
|
||||
/// and has nothing to persist.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal sealed class BpmnScopeDispatcher
|
||||
{
|
||||
private readonly Queue<Func<ValueTask>> _queue = new();
|
||||
|
||||
private bool _draining;
|
||||
|
||||
/// <summary>Whether an evaluation is in flight, so a post will be queued rather than run.</summary>
|
||||
public bool IsDraining => _draining;
|
||||
|
||||
/// <summary>
|
||||
/// Queues an evaluation and, unless one is already in flight, drains the queue to exhaustion.
|
||||
/// </summary>
|
||||
public async ValueTask PostAsync(Func<ValueTask> evaluation)
|
||||
{
|
||||
_queue.Enqueue(evaluation);
|
||||
|
||||
if (_draining)
|
||||
return;
|
||||
|
||||
_draining = true;
|
||||
|
||||
try
|
||||
{
|
||||
while (_queue.Count > 0)
|
||||
await _queue.Dequeue()();
|
||||
}
|
||||
catch
|
||||
{
|
||||
// An evaluation that threw may have applied part of a command list, so everything queued behind it was
|
||||
// computed against a state that no longer describes this instance. Dropping the rest is the conservative
|
||||
// direction: the failure is on its way to the incident strategy, and resuming half-planned work would
|
||||
// bury it under activity nobody asked for.
|
||||
_queue.Clear();
|
||||
throw;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_draining = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
266
src/modules/Elsa.Bpmn/Hosting/BpmnScopeHost.cs
Normal file
266
src/modules/Elsa.Bpmn/Hosting/BpmnScopeHost.cs
Normal file
|
|
@ -0,0 +1,266 @@
|
|||
using Bpmn.Model;
|
||||
using Bpmn.Semantics;
|
||||
using Elsa.Bpmn.Activities;
|
||||
using Elsa.Bpmn.Exceptions;
|
||||
using Elsa.Bpmn.Signals;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Activities.Flowchart.Models;
|
||||
using Elsa.Workflows.Signals;
|
||||
|
||||
namespace Elsa.Bpmn.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// The host side of the <c>Bpmn.Semantics</c> port for one BPMN scope: it feeds the interpreter's four entry points
|
||||
/// and applies what comes back onto the scope's <see cref="ActivityExecutionContext"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Every entry point is synchronous and returns a value; the interpreter never calls back. The host's job is to say
|
||||
/// what its world looks like — a snapshot — and then to do what it is told, in the order it is told.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// A host instance is a view over one scope's context and is created per call. Everything durable lives in
|
||||
/// <see cref="BpmnScopeMemory"/>, everything derived lives in the context's transient properties, and everything
|
||||
/// ordering-related lives in the instance-wide <see cref="BpmnScopeDispatcher"/>.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal sealed class BpmnScopeHost
|
||||
{
|
||||
/// <summary>
|
||||
/// What this host promises it can do.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <see cref="BpmnHostCapabilities.ScopeVariables"/> is deliberately absent: reading container-scoped variables
|
||||
/// through <see cref="IBpmnVariableReader"/> needs the three-valued reader that distinguishes "no such variable"
|
||||
/// from "stored externally, cannot tell you", which this host does not have yet. Declaring it anyway would buy a
|
||||
/// definition that quietly iterates zero times instead of a refusal naming the element, and a capability is a
|
||||
/// claim rather than a wish.
|
||||
/// </remarks>
|
||||
public const BpmnHostCapabilities Capabilities =
|
||||
BpmnHostCapabilities.SubtreeCancellation | BpmnHostCapabilities.ScopeSignalling | BpmnHostCapabilities.IterationScopes;
|
||||
|
||||
/// <summary>
|
||||
/// The property key under which a nested scope's invocation correlation is carried on its own context.
|
||||
/// </summary>
|
||||
public const string InvocationCorrelationPropertyKey = "Bpmn:InvocationCorrelation";
|
||||
|
||||
private const string GraphTransientPropertyKey = "Bpmn:Graph";
|
||||
|
||||
// The interpreter is a pure function of its request: it holds no per-instance state, so one instance serves the
|
||||
// whole process. Creating one per evaluation would only re-register the built-in element behaviors.
|
||||
private static readonly BpmnInterpreter Interpreter = BpmnInterpreter.CreateDefault();
|
||||
|
||||
private static readonly object DispatcherKey = new();
|
||||
private static readonly IReadOnlyDictionary<string, string> NoCorrelation = new Dictionary<string, string>(StringComparer.Ordinal);
|
||||
|
||||
private readonly ActivityExecutionContext _context;
|
||||
private readonly BpmnProcess _process;
|
||||
|
||||
private BpmnScopeHost(ActivityExecutionContext context)
|
||||
{
|
||||
_context = context;
|
||||
_process = (BpmnProcess)context.Activity;
|
||||
}
|
||||
|
||||
/// <summary>Returns the host for the given BPMN scope context.</summary>
|
||||
public static BpmnScopeHost For(ActivityExecutionContext context) => new(context);
|
||||
|
||||
/// <summary>The evaluation queue shared by every BPMN scope in this workflow instance.</summary>
|
||||
public static BpmnScopeDispatcher DispatcherOf(WorkflowExecutionContext context) =>
|
||||
context.TransientProperties.GetOrAdd(DispatcherKey, () => new BpmnScopeDispatcher());
|
||||
|
||||
/// <summary>The built graph for this scope. Derived from the definition, the bound work and the capabilities, none of which vary per instance.</summary>
|
||||
public BpmnGraph Graph => _context.TransientProperties.GetOrAdd(GraphTransientPropertyKey, BuildGraph);
|
||||
|
||||
// --- The interpreter's four entry points ---------------------------------------------------------
|
||||
|
||||
/// <summary>The scope is beginning.</summary>
|
||||
public ValueTask StartAsync() => EvaluateAsync(memory =>
|
||||
Interpreter.Start(new BpmnStartRequest(Graph, memory.State, Snapshot(memory))));
|
||||
|
||||
/// <summary>A unit of work finished, reporting zero or more outcome names.</summary>
|
||||
public ValueTask OnWorkCompletedAsync(ActivityExecutionContext childContext, object? result) => EvaluateAsync(memory =>
|
||||
{
|
||||
// Keyed on the child's own context id. A completion for work this scope no longer holds is absorbed rather
|
||||
// than faulted: an interrupting boundary tears its host down while the host's work is still in flight, and a
|
||||
// late completion for work that was torn down is an ordinary BPMN race.
|
||||
if (memory.Work.FindByChildContextId(childContext.Id) is not { } record)
|
||||
return null;
|
||||
|
||||
// The completing work must ALREADY be gone from LiveWork when the interpreter is asked.
|
||||
memory.Work.Remove(record);
|
||||
memory.SaveWork();
|
||||
|
||||
var outcomeNames = result is Outcomes outcomes ? outcomes.Names : [];
|
||||
|
||||
return Interpreter.OnWorkCompleted(new BpmnWorkCompletedRequest(
|
||||
Graph, memory.State, Snapshot(memory), record.BindingRef, record.Handle, outcomeNames, record.IterationId));
|
||||
});
|
||||
|
||||
/// <summary>A nested scope this one invoked signalled outward.</summary>
|
||||
public ValueTask OnScopeSignalledAsync(BpmnScopeSignal signal, SignalContext signalContext)
|
||||
{
|
||||
var sender = signalContext.SenderActivityExecutionContext;
|
||||
|
||||
// The channel delivers to the sender before walking its ancestors, and a scope never signals itself.
|
||||
if (string.Equals(sender.Id, _context.Id, StringComparison.Ordinal))
|
||||
return default;
|
||||
|
||||
// A scope signal is for the immediate enclosing scope, which is the one that started the sender's work. Any
|
||||
// other receiver lets it keep bubbling; that is also how an unrelated container in between composes.
|
||||
if (BpmnScopeMemory.Load(_context).Work.FindByChildContextId(sender.Id) is not { } signalling)
|
||||
return default;
|
||||
|
||||
signalContext.StopPropagation();
|
||||
|
||||
return EvaluateAsync(memory =>
|
||||
{
|
||||
// Unlike a completion, the signalling work stays in the ledger: an escalating activity keeps running, and
|
||||
// removing it makes the interpreter believe it has already gone.
|
||||
if (memory.Work.FindByHandle(signalling.Handle) is not { } record)
|
||||
return null;
|
||||
|
||||
return Interpreter.OnWorkSignalled(new BpmnWorkSignalledRequest(
|
||||
Graph, memory.State, Snapshot(memory), record.BindingRef, record.Handle, signal.Code, signal.Payload, record.IterationId));
|
||||
});
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A unit of work failed. Rides the <see cref="FaultSignal"/> seam: handle the signal, ask the interpreter what
|
||||
/// BPMN makes of the fault, and claim it only when a catcher took it.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The disposition has to be decided before this handler returns, so the interpreter is asked inline; only the
|
||||
/// commands are applied through the dispatcher. A <c>Propagated</c> disposition is left strictly alone — no
|
||||
/// <c>StopPropagation</c>, nothing terminalized — so the fault reaches the enclosing scope, which is how BPMN
|
||||
/// error propagation crosses a scope boundary, or the incident strategy, which is how it surfaces at the root.
|
||||
/// </remarks>
|
||||
public async ValueTask OnWorkFaultedAsync(FaultSignal signal, SignalContext signalContext)
|
||||
{
|
||||
var memory = BpmnScopeMemory.Load(_context);
|
||||
|
||||
if (ResolveFaultedWork(memory, signal.FaultedContext) is not { } record)
|
||||
return;
|
||||
|
||||
// As with a completion, the failed work must ALREADY be removed before the interpreter is asked.
|
||||
memory.Work.Remove(record);
|
||||
memory.SaveWork();
|
||||
|
||||
var evaluation = Interpreter.OnWorkFaulted(new BpmnWorkFaultedRequest(
|
||||
Graph, memory.State, Snapshot(memory), record.BindingRef, record.Handle, signal.Exception.Message));
|
||||
|
||||
memory.State = evaluation.State.Prune();
|
||||
memory.SaveState();
|
||||
|
||||
if (evaluation.Disposition is not BpmnErrorDisposition.Caught)
|
||||
return;
|
||||
|
||||
signalContext.StopPropagation();
|
||||
|
||||
// A handler that claims a fault owns terminalizing the failed activity, and BPMN terminalizes the whole unit of
|
||||
// work rather than the one activity that threw: when the failure came from inside a nested scope, this scope's
|
||||
// failing work is that scope. Cancelling it recursively covers the activity that actually threw. The interpreter
|
||||
// issues no teardown for failed work — it treats it as already terminal — so this is the host's own doing and
|
||||
// not a command out of order. RecoverFromFault stays the middleware's alone: it decrements every ancestor's
|
||||
// fault count, so a second call drives them negative.
|
||||
if (BpmnWorkTeardown.FindContext(_context.WorkflowExecutionContext, record.ChildContextId) is { } failedWorkContext)
|
||||
await BpmnWorkTeardown.CancelSubtreeAsync(failedWorkContext, $"element '{record.ElementId}' failed");
|
||||
|
||||
await DispatcherOf(_context.WorkflowExecutionContext).PostAsync(() => ApplyAsync(memory, evaluation));
|
||||
}
|
||||
|
||||
// --- Plumbing ------------------------------------------------------------------------------------
|
||||
|
||||
private ValueTask EvaluateAsync(Func<BpmnScopeMemory, BpmnEvaluation?> evaluate) =>
|
||||
DispatcherOf(_context.WorkflowExecutionContext).PostAsync(async () =>
|
||||
{
|
||||
var memory = BpmnScopeMemory.Load(_context);
|
||||
var evaluation = evaluate(memory);
|
||||
|
||||
if (evaluation is null)
|
||||
return;
|
||||
|
||||
// Persist the state before acting on the commands: a command applied against a state that was never
|
||||
// recorded is how a crash produces work with no token behind it.
|
||||
memory.State = evaluation.State.Prune();
|
||||
memory.SaveState();
|
||||
|
||||
await ApplyAsync(memory, evaluation);
|
||||
});
|
||||
|
||||
private async ValueTask ApplyAsync(BpmnScopeMemory memory, BpmnEvaluation evaluation)
|
||||
{
|
||||
await new BpmnCommandApplier(_context, _process, memory).ApplyAsync(evaluation.Commands);
|
||||
|
||||
switch (evaluation.Continuation)
|
||||
{
|
||||
case BpmnContinuation.Complete complete:
|
||||
// The scope completes because the interpreter said so, never because it ran out of children.
|
||||
await _context.CompleteActivityAsync(new Outcomes(complete.Outcome));
|
||||
break;
|
||||
case BpmnContinuation.Defer:
|
||||
break;
|
||||
case BpmnContinuation.Fault fault:
|
||||
throw new BpmnScopeFaultException(fault.Code, fault.Message);
|
||||
default:
|
||||
throw new NotSupportedException($"The BPMN continuation '{evaluation.Continuation.GetType().Name}' is not supported by this host.");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Finds the unit of work this scope started that the failing activity belongs to, walking outward from the
|
||||
/// failure.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// A fault raised deep inside a nested scope is, to this scope, its own subprocess work failing. The nested scope
|
||||
/// sees the signal first and claims it if it has a catcher; if it does not, the signal arrives here and this walk
|
||||
/// is what turns "some activity failed" into "the work I started failed", which is exactly what BPMN error
|
||||
/// propagation across a scope boundary means.
|
||||
/// </remarks>
|
||||
private BpmnWorkRecord? ResolveFaultedWork(BpmnScopeMemory memory, ActivityExecutionContext faultedContext)
|
||||
{
|
||||
for (var current = faultedContext; current is not null && !string.Equals(current.Id, _context.Id, StringComparison.Ordinal); current = current.ParentActivityExecutionContext)
|
||||
{
|
||||
if (memory.Work.FindByChildContextId(current.Id) is { } record)
|
||||
return record;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
private BpmnHostSnapshot Snapshot(BpmnScopeMemory memory)
|
||||
{
|
||||
var invocationCorrelation = InvocationCorrelation;
|
||||
|
||||
return new BpmnHostSnapshot(
|
||||
ScopeInstanceId: _context.Id,
|
||||
// A scope has an enclosing one exactly when another scope started it, which is what the carried
|
||||
// correlation records. A root process has none, so an unhandled escalation is a documented no-op.
|
||||
HasEnclosingScope: invocationCorrelation.Count > 0,
|
||||
LiveWork: memory.Work.ToLiveWork(),
|
||||
InvocationCorrelation: invocationCorrelation,
|
||||
Variables: BpmnNoVariables.Instance,
|
||||
Capabilities: Capabilities);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The correlation of the work that started this scope. It belongs to the scope and is fixed for its lifetime;
|
||||
/// a completing unit of work's correlation is never written here, because the event-subprocess start hint is read
|
||||
/// from this same dictionary.
|
||||
/// </summary>
|
||||
private IReadOnlyDictionary<string, string> InvocationCorrelation =>
|
||||
BpmnScopeMemory.Read<Dictionary<string, string>>(_context, InvocationCorrelationPropertyKey) ?? NoCorrelation;
|
||||
|
||||
private BpmnGraph BuildGraph()
|
||||
{
|
||||
var definition = _process.Process
|
||||
?? throw new InvalidOperationException($"BPMN process activity '{_process.Id}' has no process definition to execute.");
|
||||
|
||||
// Every binding the definition declares, with the nested definition attached where the bound activity is
|
||||
// itself a BPMN scope. The graph validator reads that for an event subprocess body's start trigger.
|
||||
var boundWork = BpmnBoundWork.Derive(definition, bindingRef => (_process.FindWorkActivity(bindingRef) as BpmnProcess)?.Process);
|
||||
|
||||
return BpmnGraph.Build(definition, boundWork, Capabilities);
|
||||
}
|
||||
}
|
||||
65
src/modules/Elsa.Bpmn/Hosting/BpmnScopeMemory.cs
Normal file
65
src/modules/Elsa.Bpmn/Hosting/BpmnScopeMemory.cs
Normal file
|
|
@ -0,0 +1,65 @@
|
|||
using System.Text.Json;
|
||||
using Bpmn.Model.State;
|
||||
using Elsa.Workflows;
|
||||
|
||||
namespace Elsa.Bpmn.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// A BPMN scope's own memory: the interpreter's execution state, and the ledger of work this scope started.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Both live in the scope's <see cref="ActivityExecutionContext.Properties"/>, because that is where per-execution
|
||||
/// state belongs and because the handle-to-context map must survive anything that rewrites
|
||||
/// <see cref="ActivityExecutionContext.Tag"/>. Both are held as serialized JSON strings rather than as objects:
|
||||
/// the property bag is written through <c>PolymorphicObjectConverter</c>, which silently drops the type of a value
|
||||
/// whose type has no registered alias and hands back a loose <c>ExpandoObject</c> on the way in.
|
||||
/// </remarks>
|
||||
internal sealed class BpmnScopeMemory
|
||||
{
|
||||
/// <summary>The property key holding the interpreter's execution state.</summary>
|
||||
public const string ExecutionStatePropertyKey = "Bpmn:ExecutionState";
|
||||
|
||||
/// <summary>The property key holding this scope's work ledger.</summary>
|
||||
public const string WorkLedgerPropertyKey = "Bpmn:WorkLedger";
|
||||
|
||||
private static readonly JsonSerializerOptions SerializerOptions = new(JsonSerializerDefaults.General);
|
||||
|
||||
private readonly ActivityExecutionContext _context;
|
||||
|
||||
private BpmnScopeMemory(ActivityExecutionContext context, BpmnExecutionState? state, BpmnWorkLedger work)
|
||||
{
|
||||
_context = context;
|
||||
State = state;
|
||||
Work = work;
|
||||
}
|
||||
|
||||
/// <summary>The state the interpreter last returned, or <c>null</c> before the scope has been started.</summary>
|
||||
public BpmnExecutionState? State { get; set; }
|
||||
|
||||
/// <summary>The work this scope has started and not finished.</summary>
|
||||
public BpmnWorkLedger Work { get; }
|
||||
|
||||
/// <summary>Reads a scope's memory from its property bag.</summary>
|
||||
public static BpmnScopeMemory Load(ActivityExecutionContext context) =>
|
||||
new(context, Read<BpmnExecutionState>(context, ExecutionStatePropertyKey), Read<BpmnWorkLedger>(context, WorkLedgerPropertyKey) ?? new BpmnWorkLedger());
|
||||
|
||||
/// <summary>Reads a JSON string property, or returns <c>null</c> when it is absent.</summary>
|
||||
public static T? Read<T>(ActivityExecutionContext context, string key) where T : class =>
|
||||
context.Properties.TryGetValue(key, out var value) && value is string json && !string.IsNullOrWhiteSpace(json)
|
||||
? JsonSerializer.Deserialize<T>(json, SerializerOptions)
|
||||
: null;
|
||||
|
||||
/// <summary>Writes a value as a JSON string property.</summary>
|
||||
public static void Write<T>(ActivityExecutionContext context, string key, T value) =>
|
||||
context.Properties[key] = JsonSerializer.Serialize(value, SerializerOptions);
|
||||
|
||||
/// <summary>Persists the execution state.</summary>
|
||||
public void SaveState()
|
||||
{
|
||||
if (State is not null)
|
||||
Write(_context, ExecutionStatePropertyKey, State);
|
||||
}
|
||||
|
||||
/// <summary>Persists the work ledger.</summary>
|
||||
public void SaveWork() => Write(_context, WorkLedgerPropertyKey, Work);
|
||||
}
|
||||
48
src/modules/Elsa.Bpmn/Hosting/BpmnWorkLedger.cs
Normal file
48
src/modules/Elsa.Bpmn/Hosting/BpmnWorkLedger.cs
Normal file
|
|
@ -0,0 +1,48 @@
|
|||
using Bpmn.Semantics;
|
||||
|
||||
namespace Elsa.Bpmn.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// A BPMN scope's record of the work it has started and not finished — the handle-to-context map, and the source of
|
||||
/// <see cref="BpmnHostSnapshot.LiveWork"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// What belongs in here depends on which callback the scope is about to make, and the difference is load-bearing:
|
||||
/// completing and faulting work must already have been removed, because leaving it lets a re-armed non-interrupting
|
||||
/// listener key onto the same <c>(binding ref, iteration id)</c> slot and a teardown then targets the work that just
|
||||
/// finished; signalling work must still be present, because an escalating activity keeps running and removing it
|
||||
/// makes the interpreter believe it has already gone.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The interpreter re-finds a parked token from <c>(binding ref, iteration id)</c> alone, so a scope must never hold
|
||||
/// two live units of work under one such pair. Multi-instance instances share a binding ref and are told apart by
|
||||
/// their iteration id, which is what it is for.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal sealed class BpmnWorkLedger
|
||||
{
|
||||
/// <summary>The handle counter. Handles are scope-local so a scope's trace reads independently of Elsa's id generator.</summary>
|
||||
public int Sequence { get; set; }
|
||||
|
||||
/// <summary>The live work, in start order.</summary>
|
||||
public List<BpmnWorkRecord> Records { get; set; } = [];
|
||||
|
||||
/// <summary>Mints the next scope-local handle.</summary>
|
||||
public string NextHandle() => $"work-{++Sequence}";
|
||||
|
||||
/// <summary>The live work running in the given child activity execution, or <c>null</c>.</summary>
|
||||
public BpmnWorkRecord? FindByChildContextId(string childContextId) =>
|
||||
Records.FirstOrDefault(x => string.Equals(x.ChildContextId, childContextId, StringComparison.Ordinal));
|
||||
|
||||
/// <summary>The live work with the given handle, or <c>null</c>.</summary>
|
||||
public BpmnWorkRecord? FindByHandle(string handle) =>
|
||||
Records.FirstOrDefault(x => string.Equals(x.Handle, handle, StringComparison.Ordinal));
|
||||
|
||||
/// <summary>Drops a record, so the work is no longer reported as live.</summary>
|
||||
public void Remove(BpmnWorkRecord record) => Records.Remove(record);
|
||||
|
||||
/// <summary>The interpreter's view of this scope's live work.</summary>
|
||||
public IReadOnlyCollection<BpmnLiveWork> ToLiveWork() =>
|
||||
Records.Select(x => new BpmnLiveWork(x.BindingRef, x.IterationId, x.Handle)).ToArray();
|
||||
}
|
||||
33
src/modules/Elsa.Bpmn/Hosting/BpmnWorkRecord.cs
Normal file
33
src/modules/Elsa.Bpmn/Hosting/BpmnWorkRecord.cs
Normal file
|
|
@ -0,0 +1,33 @@
|
|||
namespace Elsa.Bpmn.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// One unit of work a BPMN scope started and has not finished: the interpreter's view of it, plus the identity of
|
||||
/// the Elsa activity execution that runs it.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <see cref="ChildContextId"/> is the whole point of this record. A unit of work is keyed on the child
|
||||
/// <see cref="Elsa.Workflows.ActivityExecutionContext"/>'s id and never on <c>ActivityExecutionContext.Tag</c>.
|
||||
/// Tag is shared mutable state: the completion-callback dispatch writes the popped callback entry's tag onto the
|
||||
/// <i>receiving</i> context, so a nested scope's own tag is rewritten by its own children, and a parent that
|
||||
/// remembered a child as one tag finds the child wearing another. Making the handles globally unique does not fix
|
||||
/// that, because the two values are written by different owners at different times.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal sealed class BpmnWorkRecord
|
||||
{
|
||||
/// <summary>The opaque host handle the interpreter echoes back. Scope-local and minted by <see cref="BpmnWorkLedger"/>.</summary>
|
||||
public string Handle { get; set; } = null!;
|
||||
|
||||
/// <summary>The binding ref the interpreter asked to start.</summary>
|
||||
public string BindingRef { get; set; } = null!;
|
||||
|
||||
/// <summary>The multi-instance iteration id this unit runs under, or <c>null</c> for ordinary single-run work.</summary>
|
||||
public string? IterationId { get; set; }
|
||||
|
||||
/// <summary>The element the work belongs to, for reporting.</summary>
|
||||
public string ElementId { get; set; } = null!;
|
||||
|
||||
/// <summary>The id of the child <see cref="Elsa.Workflows.ActivityExecutionContext"/> running this work.</summary>
|
||||
public string ChildContextId { get; set; } = null!;
|
||||
}
|
||||
58
src/modules/Elsa.Bpmn/Hosting/BpmnWorkTeardown.cs
Normal file
58
src/modules/Elsa.Bpmn/Hosting/BpmnWorkTeardown.cs
Normal file
|
|
@ -0,0 +1,58 @@
|
|||
using Elsa.Extensions;
|
||||
using Elsa.Workflows;
|
||||
|
||||
namespace Elsa.Bpmn.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// Stops a unit of work and everything it in turn started.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Both places that tear work down go through here: the interpreter's <c>CancelWorkSubtree</c> command, and a scope
|
||||
/// terminalizing the unit of work whose fault it just claimed. The mechanism is the same in either case, and so is the
|
||||
/// one thing this host cannot do.
|
||||
/// </remarks>
|
||||
internal static class BpmnWorkTeardown
|
||||
{
|
||||
/// <summary>The activity execution with the given id, or <c>null</c> when it has already gone.</summary>
|
||||
public static ActivityExecutionContext? FindContext(WorkflowExecutionContext workflowExecutionContext, string activityExecutionContextId) =>
|
||||
workflowExecutionContext.ActivityExecutionContexts.FirstOrDefault(x => string.Equals(x.Id, activityExecutionContextId, StringComparison.Ordinal));
|
||||
|
||||
/// <summary>
|
||||
/// Tears down a unit of work and everything underneath it, or refuses when it cannot.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Cancellation itself needs no new code: the public <c>CancelActivityAsync</c> extension already walks the child
|
||||
/// subtree recursively, which is exactly what "and everything it in turn started" means.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// What it cannot do is withdraw work whose activity is scheduled but has not been invoked yet. Such work has no
|
||||
/// running execution to cancel, and <c>IActivityScheduler</c> offers no way to remove a queued work item, so the
|
||||
/// activity runs after BPMN destroyed the branch it belongs to regardless of what this method does. Throwing
|
||||
/// still fails loudly under <c>FaultStrategy</c>, which is the strategy that justifies it. It does not fail
|
||||
/// loudly under <c>ContinueWithIncidentsStrategy</c>: there the throw is absorbed into an incident, execution
|
||||
/// continues, and the stranded activity still runs. What limits the damage in that case is the caller's own
|
||||
/// doing, not this method's: the caller removes the work's ledger record and persists that removal before
|
||||
/// invoking this method, so an absorbed throw cannot leave the persisted ledger claiming work that was just torn
|
||||
/// down, and the stranded activity's eventual completion callback finds no live record and is discarded rather
|
||||
/// than handed to the interpreter as real work. That does not stop the stray activity from running, and it does
|
||||
/// not undo whatever side effects it has — it only stops its result from being believed.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public static async ValueTask CancelSubtreeAsync(ActivityExecutionContext childContext, string reason)
|
||||
{
|
||||
var scheduler = childContext.WorkflowExecutionContext.Scheduler;
|
||||
var subtree = new[] { childContext }.Concat(childContext.GetDescendants());
|
||||
var queued = subtree.FirstOrDefault(context => scheduler.Any(item => item.ExistingActivityExecutionContext?.Id == context.Id));
|
||||
|
||||
if (queued is not null)
|
||||
{
|
||||
throw new NotSupportedException(
|
||||
$"BPMN asked to tear down the work of activity '{childContext.Activity.Id}' ({reason}), but activity "
|
||||
+ $"'{queued.Activity.Id}' inside that subtree is scheduled and has not started, and a scheduled work "
|
||||
+ "item cannot be withdrawn. Running it anyway would leave a branch alive that BPMN destroyed.");
|
||||
}
|
||||
|
||||
await childContext.CancelActivityAsync();
|
||||
}
|
||||
}
|
||||
17
src/modules/Elsa.Bpmn/Signals/BpmnScopeSignal.cs
Normal file
17
src/modules/Elsa.Bpmn/Signals/BpmnScopeSignal.cs
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
using System.Text.Json;
|
||||
|
||||
namespace Elsa.Bpmn.Signals;
|
||||
|
||||
/// <summary>
|
||||
/// Carries the interpreter's <c>SignalEnclosingScope</c> command from a nested BPMN scope to the scope that started it.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The signal travels the ordinary Elsa channel, so it is delivered to the sending scope first and then to each
|
||||
/// ancestor in turn. Only the scope that started the sending scope's unit of work claims it; every other receiver
|
||||
/// lets it keep bubbling, which is also how a scope that cannot match an escalation re-signals it one hop further
|
||||
/// out. There is deliberately one code per kind of scope signal rather than a growing namespace: BPMN escalation
|
||||
/// travels under <c>BpmnInterpreter.EscalationSignalCode</c> with its identity in the payload.
|
||||
/// </remarks>
|
||||
/// <param name="Code">The signal code.</param>
|
||||
/// <param name="Payload">The signal payload.</param>
|
||||
public record BpmnScopeSignal(string Code, JsonElement? Payload);
|
||||
|
|
@ -0,0 +1,32 @@
|
|||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Signals;
|
||||
|
||||
namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities;
|
||||
|
||||
/// <summary>
|
||||
/// A stand-in for BPMN work that runs until something else finishes it — a long-running task, or an armed listener
|
||||
/// waiting for its trigger. It blocks on a bookmark, so a test resumes it by name.
|
||||
/// </summary>
|
||||
public class BpmnTestBlockingWork : Activity
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public BpmnTestBlockingWork() => OnSignalReceived<CancelSignal>(OnCancelled);
|
||||
|
||||
/// <summary>The log to record into.</summary>
|
||||
public BpmnTestLog? Log { get; set; }
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void Execute(ActivityExecutionContext context)
|
||||
{
|
||||
BpmnTestWorkProbe.RecordExecution(this, Log, context);
|
||||
context.CreateBookmark(ResumeAsync);
|
||||
}
|
||||
|
||||
private async ValueTask ResumeAsync(ActivityExecutionContext context)
|
||||
{
|
||||
Log?.Record($"resumed:{Id}");
|
||||
await context.CompleteActivityAsync();
|
||||
}
|
||||
|
||||
private void OnCancelled(CancelSignal signal, SignalContext context) => BpmnTestWorkProbe.RecordCancellation(this, Log, context);
|
||||
}
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Signals;
|
||||
|
||||
namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities;
|
||||
|
||||
/// <summary>
|
||||
/// A stand-in for BPMN work that fails. The host reports that work failed and describes why in human terms; which
|
||||
/// BPMN error a catcher matches is a property of the model, not of the report.
|
||||
/// </summary>
|
||||
public class BpmnTestFaultingWork : CodeActivity
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public BpmnTestFaultingWork() => OnSignalReceived<CancelSignal>(OnCancelled);
|
||||
|
||||
/// <summary>The log to record into.</summary>
|
||||
public BpmnTestLog? Log { get; set; }
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void Execute(ActivityExecutionContext context)
|
||||
{
|
||||
BpmnTestWorkProbe.RecordExecution(this, Log, context);
|
||||
throw new InvalidOperationException($"The work bound to '{Id}' failed.");
|
||||
}
|
||||
|
||||
private void OnCancelled(CancelSignal signal, SignalContext context) => BpmnTestWorkProbe.RecordCancellation(this, Log, context);
|
||||
}
|
||||
|
|
@ -0,0 +1,49 @@
|
|||
using Elsa.Bpmn.Hosting;
|
||||
using Elsa.Workflows;
|
||||
|
||||
namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities;
|
||||
|
||||
/// <summary>
|
||||
/// An ordered record of what the stand-in work activities did, plus snapshots taken at moments the applier's
|
||||
/// invariants are observable.
|
||||
/// </summary>
|
||||
public sealed class BpmnTestLog
|
||||
{
|
||||
private readonly List<string> _entries = [];
|
||||
private readonly Dictionary<string, IReadOnlyList<string>> _snapshots = new(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>Everything that happened, in order.</summary>
|
||||
public IReadOnlyList<string> Entries => _entries;
|
||||
|
||||
/// <summary>Records that something happened.</summary>
|
||||
public void Record(string entry) => _entries.Add(entry);
|
||||
|
||||
/// <summary>The position of an entry, or <c>-1</c>.</summary>
|
||||
public int PositionOf(string entry) => _entries.IndexOf(entry);
|
||||
|
||||
/// <summary>How many times an entry was recorded.</summary>
|
||||
public int Occurrences(string entry) => _entries.Count(x => string.Equals(x, entry, StringComparison.Ordinal));
|
||||
|
||||
/// <summary>A snapshot taken under the given key.</summary>
|
||||
public IReadOnlyList<string> Snapshot(string key) =>
|
||||
_snapshots.TryGetValue(key, out var snapshot)
|
||||
? snapshot
|
||||
: throw new InvalidOperationException($"Nothing was captured under '{key}'. Captured keys: {string.Join(", ", _snapshots.Keys)}.");
|
||||
|
||||
/// <summary>
|
||||
/// Captures the binding refs the enclosing BPMN scope currently reports as live work, read from the scope's own
|
||||
/// persisted ledger — which is what the host hands the interpreter as <c>BpmnHostSnapshot.LiveWork</c>.
|
||||
/// </summary>
|
||||
public void CaptureLiveWork(string key, ActivityExecutionContext context)
|
||||
{
|
||||
var scopeContext = context.ParentActivityExecutionContext
|
||||
?? throw new InvalidOperationException("Expected the work activity to be owned by a BPMN scope.");
|
||||
var ledger = BpmnScopeMemory.Read<BpmnWorkLedger>(scopeContext, BpmnScopeMemory.WorkLedgerPropertyKey) ?? new BpmnWorkLedger();
|
||||
|
||||
_snapshots[key] = ledger.Records.Select(x => x.BindingRef).OrderBy(x => x, StringComparer.Ordinal).ToList();
|
||||
}
|
||||
|
||||
/// <summary>Captures the ids of the activities currently sitting in the workflow's scheduler.</summary>
|
||||
public void CaptureScheduled(string key, ActivityExecutionContext context) =>
|
||||
_snapshots[key] = context.WorkflowExecutionContext.Scheduler.List().Select(x => x.Activity.Id).ToList();
|
||||
}
|
||||
|
|
@ -0,0 +1,21 @@
|
|||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Signals;
|
||||
|
||||
namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities;
|
||||
|
||||
/// <summary>
|
||||
/// A stand-in for BPMN work that finishes as soon as it runs.
|
||||
/// </summary>
|
||||
public class BpmnTestWork : CodeActivity
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public BpmnTestWork() => OnSignalReceived<CancelSignal>(OnCancelled);
|
||||
|
||||
/// <summary>The log to record into.</summary>
|
||||
public BpmnTestLog? Log { get; set; }
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void Execute(ActivityExecutionContext context) => BpmnTestWorkProbe.RecordExecution(this, Log, context);
|
||||
|
||||
private void OnCancelled(CancelSignal signal, SignalContext context) => BpmnTestWorkProbe.RecordCancellation(this, Log, context);
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
using Elsa.Workflows;
|
||||
|
||||
namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities;
|
||||
|
||||
/// <summary>
|
||||
/// The recording every stand-in work activity does, kept in one place so the activities differ only in what they do.
|
||||
/// </summary>
|
||||
internal static class BpmnTestWorkProbe
|
||||
{
|
||||
/// <summary>Records that the work started, and what its scope reported as live at that moment.</summary>
|
||||
public static void RecordExecution(IActivity activity, BpmnTestLog? log, ActivityExecutionContext context)
|
||||
{
|
||||
log?.Record($"executed:{activity.Id}");
|
||||
log?.CaptureLiveWork($"liveWork@{activity.Id}", context);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Records that the work was torn down, and what was in the scheduler at that moment.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The scheduler snapshot is what makes command ordering observable: a teardown applied after the boundary path's
|
||||
/// <c>StartWork</c> sees the boundary path already scheduled, and one applied before it does not.
|
||||
/// </remarks>
|
||||
public static void RecordCancellation(IActivity activity, BpmnTestLog? log, SignalContext context)
|
||||
{
|
||||
var receiver = context.ReceiverActivityExecutionContext;
|
||||
|
||||
// CancelSignal is delivered to the cancelled activity and then to each of its ancestors; only the first is
|
||||
// this activity's own cancellation.
|
||||
if (receiver.Activity != activity || !string.Equals(receiver.Id, context.SenderActivityExecutionContext.Id, StringComparison.Ordinal))
|
||||
return;
|
||||
|
||||
log?.Record($"cancelled:{activity.Id}");
|
||||
log?.CaptureScheduled($"scheduled@cancel:{activity.Id}", receiver);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,146 @@
|
|||
using Elsa.Common.Models;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.IncidentStrategies;
|
||||
using Xunit.Abstractions;
|
||||
|
||||
namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort;
|
||||
|
||||
/// <summary>
|
||||
/// The invariants the interpreter relies on its host to uphold. Each of these is a rule a host can break while every
|
||||
/// process still appears to run, which is exactly why they are pinned separately from the processes.
|
||||
/// </summary>
|
||||
public class BpmnHostInvariantTests(ITestOutputHelper testOutputHelper)
|
||||
{
|
||||
private readonly BpmnTestHost _host = new(testOutputHelper);
|
||||
|
||||
[Fact(DisplayName = "Commands are applied in the order returned")]
|
||||
public async Task Commands_AreAppliedInTheOrderReturned()
|
||||
{
|
||||
// An interrupting boundary event returns the boundary path's StartWork *before* the teardown that retires the
|
||||
// host it interrupted. The interrupted task records what was in the scheduler at the moment it was torn down,
|
||||
// so applying the teardown first is visible: the boundary path would not be there yet.
|
||||
|
||||
// Arrange
|
||||
await _host.RunAsync(BpmnTestProcesses.InterruptingTimerBoundary(_host.Log));
|
||||
|
||||
// Act
|
||||
await _host.FinishWorkAsync("timeout");
|
||||
|
||||
// Assert
|
||||
Assert.Contains("onTimeout", _host.Log.Snapshot("scheduled@cancel:task"));
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Completed work is gone from LiveWork")]
|
||||
public async Task CompletedWork_IsRemovedFromLiveWork()
|
||||
{
|
||||
// The next activity the evaluation schedules reads the scope's ledger, which is the sole source of
|
||||
// BpmnHostSnapshot.LiveWork. Both the completed listener and the torn-down host must be gone from it:
|
||||
// completion is terminal, and leaving it lets a later teardown resolve the handle of work that already
|
||||
// finished.
|
||||
//
|
||||
// The applier removes the record before asking the interpreter, which is what the port requires. That stricter
|
||||
// ordering is not asserted here because it is not observable for these constructs: the interpreter reads
|
||||
// LiveWork only to resolve teardown handles by (binding ref, iteration id), and none of the processes in scope
|
||||
// tears down a slot a just-completed unit of work shares. It becomes observable with multi-instance work and
|
||||
// re-armed scope listeners, which arrive with the issues that add them.
|
||||
|
||||
// Arrange
|
||||
await _host.RunAsync(BpmnTestProcesses.InterruptingTimerBoundary(_host.Log));
|
||||
|
||||
// Act
|
||||
await _host.FinishWorkAsync("timeout");
|
||||
|
||||
// Assert
|
||||
Assert.Equal([BpmnTestProcesses.BindingRef("onTimeout")], _host.Log.Snapshot("liveWork@onTimeout"));
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Faulted work is gone from LiveWork")]
|
||||
public async Task FaultedWork_IsRemovedFromLiveWork()
|
||||
{
|
||||
// Act
|
||||
await _host.RunAsync(BpmnTestProcesses.ErrorBoundaryCaught(_host.Log), typeof(FaultStrategy));
|
||||
|
||||
// Assert: only the boundary path's work is live by the time it runs; the failed work is not.
|
||||
Assert.Equal([BpmnTestProcesses.BindingRef("recover")], _host.Log.Snapshot("liveWork@recover"));
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Signalling work is still in LiveWork when the interpreter is asked about the signal")]
|
||||
public async Task SignallingWork_StaysInLiveWork()
|
||||
{
|
||||
// A signal is not terminal: the escalating subprocess keeps running. Removing it would make the interpreter
|
||||
// believe the boundary event's host had already gone, so the escalation would be recorded as late and the
|
||||
// boundary path below would never be scheduled at all.
|
||||
|
||||
// Arrange
|
||||
await _host.RunAsync(BpmnTestProcesses.EscalationOutOfSubprocess(_host.Log));
|
||||
|
||||
// Act
|
||||
await _host.FinishWorkAsync("subWork");
|
||||
|
||||
// Assert: the boundary path ran, and the subprocess that raised the escalation was still live when it did.
|
||||
Assert.Contains("executed:notify", _host.Log.Entries);
|
||||
Assert.Contains(BpmnTestProcesses.BindingRef("sub"), _host.Log.Snapshot("liveWork@notify"));
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A parent evaluation raised mid-apply is queued and drained, not recursed")]
|
||||
public async Task ParentEvaluationRaisedMidApply_IsQueuedNotRecursed()
|
||||
{
|
||||
// The subprocess's evaluation returns [SignalEnclosingScope, StartWork(subMore)]. Delivering the signal reaches
|
||||
// the parent scope synchronously, and the parent's own evaluation must wait until the subprocess has finished
|
||||
// applying its command list rather than running on top of it.
|
||||
//
|
||||
// The observable difference is the order in which the two branches' work is scheduled. Queued, the subprocess
|
||||
// schedules subMore first and the parent schedules notify second; recursing swaps them, because the parent runs
|
||||
// between the signal and the StartWork that follows it. Elsa's scheduler is FIFO, so scheduling order is also
|
||||
// execution order.
|
||||
|
||||
// Arrange
|
||||
await _host.RunAsync(BpmnTestProcesses.EscalationOutOfSubprocess(_host.Log));
|
||||
|
||||
// Act
|
||||
await _host.FinishWorkAsync("subWork");
|
||||
|
||||
// Assert
|
||||
Assert.True(
|
||||
_host.Log.PositionOf("executed:subMore") < _host.Log.PositionOf("executed:notify"),
|
||||
$"Expected the subprocess to finish applying its command list before the parent's escalation path was scheduled, but the log was: {string.Join(", ", _host.Log.Entries)}.");
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Concurrent instances of one binding are told apart by their iteration id")]
|
||||
public async Task ConcurrentInstancesOfOneBinding_AreToldApartByIterationId()
|
||||
{
|
||||
// The interpreter re-finds a parked token from (binding ref, iteration id) alone, so a scope may never hold two
|
||||
// live units of work under one such pair. Both instances of this task share a binding ref and one activity, so
|
||||
// nothing but the iteration id — and, on the host side, the child activity execution — separates them.
|
||||
|
||||
// Arrange
|
||||
await _host.RunAsync(BpmnTestProcesses.ParallelMultiInstanceTask(_host.Log));
|
||||
|
||||
// Assert: two live instances, distinguished only by the iteration id.
|
||||
var liveWork = _host.LiveWorkOf("scope");
|
||||
Assert.Equal(2, liveWork.Count);
|
||||
Assert.All(liveWork, work => Assert.Equal(BpmnTestProcesses.BindingRef("each"), work.BindingRef));
|
||||
Assert.Equal(2, liveWork.Select(work => work.IterationId).Distinct().Count());
|
||||
Assert.DoesNotContain(liveWork, work => work.IterationId is null);
|
||||
|
||||
// Act: finish both instances.
|
||||
await _host.FinishWorkAsync("each");
|
||||
var result = await _host.FinishWorkAsync("each");
|
||||
|
||||
// Assert
|
||||
Assert.Equal(2, _host.Log.Occurrences("executed:each"));
|
||||
Assert.Equal(1, _host.Log.Occurrences("executed:after"));
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A scope completes because the interpreter said so")]
|
||||
public async Task Scope_CompletesOnTheInterpretersContinuation()
|
||||
{
|
||||
// Act
|
||||
var result = await _host.RunAsync(BpmnTestProcesses.LinearTask(_host.Log));
|
||||
|
||||
// Assert
|
||||
Assert.Equal(ActivityStatus.Completed, result.Journal.ActivityExecutionContexts.First(x => x.Activity.Id == "scope").Status);
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,128 @@
|
|||
using Elsa.Common.Models;
|
||||
using Elsa.Testing.Shared;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.IncidentStrategies;
|
||||
using Elsa.Workflows.Models;
|
||||
using Xunit.Abstractions;
|
||||
|
||||
namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort;
|
||||
|
||||
/// <summary>
|
||||
/// The BPMN processes the host port is exercised against, end to end through <c>IWorkflowRunner</c>.
|
||||
/// </summary>
|
||||
public class BpmnProcessTests(ITestOutputHelper testOutputHelper)
|
||||
{
|
||||
private readonly BpmnTestHost _host = new(testOutputHelper);
|
||||
|
||||
[Fact(DisplayName = "An interrupting timer boundary event tears its host down and routes the boundary path")]
|
||||
public async Task InterruptingTimerBoundary_TearsTheHostDownAndRoutesTheBoundaryPath()
|
||||
{
|
||||
// Arrange
|
||||
await _host.RunAsync(BpmnTestProcesses.InterruptingTimerBoundary(_host.Log));
|
||||
|
||||
// Act: the timer fires while the task is still running.
|
||||
var result = await _host.FinishWorkAsync("timeout");
|
||||
|
||||
// Assert
|
||||
Assert.Contains("cancelled:task", _host.Log.Entries);
|
||||
Assert.Contains("executed:onTimeout", _host.Log.Entries);
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "An escalation out of an embedded subprocess is caught by a non-interrupting boundary event, and the subprocess keeps running")]
|
||||
public async Task EscalationOutOfSubprocess_IsCaughtWithoutInterruptingTheSubprocess()
|
||||
{
|
||||
// This is the nested-scope case. It only passes when the scope keys its work on the child activity execution
|
||||
// context rather than on ActivityExecutionContext.Tag, because a nested scope's tag is rewritten by its own
|
||||
// children's completion callbacks: the parent would be looking for a tag the child no longer wears.
|
||||
|
||||
// Arrange
|
||||
await _host.RunAsync(BpmnTestProcesses.EscalationOutOfSubprocess(_host.Log));
|
||||
|
||||
// Act: the subprocess reaches its escalation throw event.
|
||||
await _host.FinishWorkAsync("subWork");
|
||||
|
||||
// Assert: the escalation crossed the scope boundary and the boundary path ran...
|
||||
Assert.Contains("executed:notify", _host.Log.Entries);
|
||||
|
||||
// ...while the subprocess carried on past the throw event rather than being torn down.
|
||||
Assert.Contains("executed:subMore", _host.Log.Entries);
|
||||
Assert.DoesNotContain("cancelled:subMore", _host.Log.Entries);
|
||||
|
||||
// And the subprocess still completes normally, so the main path continues.
|
||||
var result = await _host.FinishWorkAsync("subMore");
|
||||
|
||||
Assert.Contains("executed:after", _host.Log.Entries);
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A parallel split and join fires the join exactly once, after both branches")]
|
||||
public async Task ParallelSplitAndJoin_FiresTheJoinOnce()
|
||||
{
|
||||
// Act
|
||||
var result = await _host.RunAsync(BpmnTestProcesses.ParallelSplitAndJoin(_host.Log));
|
||||
|
||||
// Assert
|
||||
Assert.Equal(1, _host.Log.Occurrences("executed:after"));
|
||||
Assert.True(_host.Log.PositionOf("executed:left") < _host.Log.PositionOf("executed:after"));
|
||||
Assert.True(_host.Log.PositionOf("executed:right") < _host.Log.PositionOf("executed:after"));
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A caught error routes the boundary path and is not an incident")]
|
||||
public async Task ErrorBoundaryCaught_ContinuesDownTheBoundaryPath()
|
||||
{
|
||||
// Act
|
||||
var result = await _host.RunAsync(BpmnTestProcesses.ErrorBoundaryCaught(_host.Log), typeof(FaultStrategy));
|
||||
|
||||
// Assert: the disposition was Caught, so the scope claimed the fault and terminalized the failed work itself.
|
||||
Assert.Contains("executed:recover", _host.Log.Entries);
|
||||
Assert.Equal(ActivityStatus.Canceled, StatusOf(result, "risky"));
|
||||
|
||||
// A fault a container claimed is not an incident: the middleware recovered the bookkeeping because
|
||||
// propagation was stopped.
|
||||
Assert.Empty(result.WorkflowState.Incidents);
|
||||
Assert.Equal(WorkflowStatus.Finished, result.WorkflowState.Status);
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "An error nothing in a subprocess catches propagates to a boundary event on the subprocess")]
|
||||
public async Task ErrorPropagatedOutOfSubprocess_IsCaughtByTheEnclosingScope()
|
||||
{
|
||||
// The nested scope reports Propagated and leaves the signal alone, so it reaches the enclosing scope, which
|
||||
// resolves it to *its* failing unit of work — the subprocess — rather than to the activity that threw.
|
||||
|
||||
// Act
|
||||
var result = await _host.RunAsync(BpmnTestProcesses.ErrorPropagatedOutOfSubprocess(_host.Log), typeof(FaultStrategy));
|
||||
|
||||
// Assert
|
||||
Assert.Contains("executed:subRecover", _host.Log.Entries);
|
||||
Assert.DoesNotContain("executed:after", _host.Log.Entries);
|
||||
|
||||
// The whole failing unit of work is terminal, not just the activity that threw.
|
||||
Assert.Equal(ActivityStatus.Canceled, StatusOf(result, "sub"));
|
||||
Assert.Equal(ActivityStatus.Canceled, StatusOf(result, "subRisky"));
|
||||
|
||||
Assert.Empty(result.WorkflowState.Incidents);
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "An uncaught error propagates and is left to the incident strategy")]
|
||||
public async Task UncaughtError_PropagatesToTheIncidentStrategy()
|
||||
{
|
||||
// Act
|
||||
var result = await _host.RunAsync(BpmnTestProcesses.UncaughtError(_host.Log), typeof(FaultStrategy));
|
||||
|
||||
// Assert: the disposition was Propagated, so the scope did not stop propagation and the incident strategy ran
|
||||
// exactly as it would with no handler present.
|
||||
Assert.Equal(WorkflowStatus.Finished, result.WorkflowState.Status);
|
||||
Assert.Equal(WorkflowSubStatus.Faulted, result.WorkflowState.SubStatus);
|
||||
|
||||
var incident = Assert.Single(result.WorkflowState.Incidents);
|
||||
Assert.Equal("risky", incident.ActivityId);
|
||||
Assert.Equal(ActivityStatus.Faulted, StatusOf(result, "risky"));
|
||||
}
|
||||
|
||||
private static ActivityStatus? StatusOf(RunWorkflowResult result, string activityId) =>
|
||||
result.Journal.ActivityExecutionContexts.FirstOrDefault(x => x.Activity.Id == activityId)?.Status;
|
||||
}
|
||||
|
|
@ -0,0 +1,92 @@
|
|||
using Elsa.Bpmn.Hosting;
|
||||
using Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Testing.Shared;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.Models;
|
||||
using Elsa.Workflows.Options;
|
||||
using Elsa.Workflows.State;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Xunit.Abstractions;
|
||||
|
||||
namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort;
|
||||
|
||||
/// <summary>
|
||||
/// Runs a BPMN scope through <see cref="IWorkflowRunner"/> and lets a test finish blocked work by activity id.
|
||||
/// </summary>
|
||||
public sealed class BpmnTestHost
|
||||
{
|
||||
private readonly IServiceProvider _services;
|
||||
private readonly IWorkflowRunner _workflowRunner;
|
||||
private readonly IWorkflowBuilderFactory _workflowBuilderFactory;
|
||||
|
||||
private Workflow? _workflow;
|
||||
private WorkflowState? _state;
|
||||
private RunWorkflowResult? _result;
|
||||
|
||||
public BpmnTestHost(ITestOutputHelper testOutputHelper)
|
||||
{
|
||||
_services = new TestApplicationBuilder(testOutputHelper)
|
||||
.ConfigureElsa(elsa => elsa.UseBpmn())
|
||||
.AddActivitiesFrom<BpmnTestWork>()
|
||||
.Build();
|
||||
|
||||
_workflowRunner = _services.GetRequiredService<IWorkflowRunner>();
|
||||
_workflowBuilderFactory = _services.GetRequiredService<IWorkflowBuilderFactory>();
|
||||
}
|
||||
|
||||
/// <summary>What the stand-in work activities recorded.</summary>
|
||||
public BpmnTestLog Log { get; } = new();
|
||||
|
||||
/// <summary>Runs the given root activity to quiescence.</summary>
|
||||
public async Task<RunWorkflowResult> RunAsync(IActivity root, Type? incidentStrategyType = null)
|
||||
{
|
||||
await _services.PopulateRegistriesAsync();
|
||||
|
||||
_workflow = await _workflowBuilderFactory.CreateBuilder().BuildWorkflowAsync(new TestWorkflow(builder =>
|
||||
{
|
||||
builder.WorkflowOptions.IncidentStrategyType = incidentStrategyType;
|
||||
builder.Root = root;
|
||||
}));
|
||||
|
||||
return Record(await _workflowRunner.RunAsync(_workflow));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Finishes the blocked work of the named activity, which is how a test fires a timer, delivers a message, or
|
||||
/// completes a long-running task.
|
||||
/// </summary>
|
||||
public async Task<RunWorkflowResult> FinishWorkAsync(string activityId)
|
||||
{
|
||||
var state = _state ?? throw new InvalidOperationException("The workflow has not been run yet.");
|
||||
|
||||
var bookmark = state.Bookmarks.FirstOrDefault(x => x.ActivityId == activityId)
|
||||
?? throw new InvalidOperationException(
|
||||
$"Activity '{activityId}' is not blocked. Blocked activities: {(state.Bookmarks.Count == 0 ? "(none)" : string.Join(", ", state.Bookmarks.Select(x => x.ActivityId)))}.");
|
||||
|
||||
return Record(await _workflowRunner.RunAsync(_workflow!, state, new RunWorkflowOptions
|
||||
{
|
||||
BookmarkId = bookmark.Id
|
||||
}));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The work the named BPMN scope currently reports as live, read from its own persisted ledger — which is the sole
|
||||
/// source of <c>BpmnHostSnapshot.LiveWork</c>.
|
||||
/// </summary>
|
||||
public IReadOnlyList<(string BindingRef, string? IterationId)> LiveWorkOf(string scopeActivityId)
|
||||
{
|
||||
var scopeContext = _result!.Journal.ActivityExecutionContexts.First(x => x.Activity.Id == scopeActivityId);
|
||||
var ledger = BpmnScopeMemory.Read<BpmnWorkLedger>(scopeContext, BpmnScopeMemory.WorkLedgerPropertyKey) ?? new BpmnWorkLedger();
|
||||
|
||||
return ledger.Records.Select(record => (record.BindingRef, record.IterationId)).ToList();
|
||||
}
|
||||
|
||||
private RunWorkflowResult Record(RunWorkflowResult result)
|
||||
{
|
||||
_result = result;
|
||||
_state = result.WorkflowState;
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,216 @@
|
|||
using Bpmn.Model;
|
||||
using Elsa.Bpmn.Activities;
|
||||
using Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities;
|
||||
using Elsa.Workflows;
|
||||
|
||||
namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort;
|
||||
|
||||
/// <summary>
|
||||
/// The processes the applier is exercised against, built in code and bound to stand-in work activities.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Every activity's id is the BPMN element id it runs, and its binding ref is that id prefixed, so a process reads
|
||||
/// the same way in the model and in the assertions.
|
||||
/// </remarks>
|
||||
internal static class BpmnTestProcesses
|
||||
{
|
||||
/// <summary>An interrupting timer boundary event on a long-running task.</summary>
|
||||
public static BpmnProcess InterruptingTimerBoundary(BpmnTestLog log)
|
||||
{
|
||||
var definition = new BpmnProcessBuilder("interrupting-timer-boundary")
|
||||
.StartEvent("start")
|
||||
.Task("task", bindingRef: BindingRef("task"))
|
||||
.EndEvent("end")
|
||||
.BoundaryEvent("timeout", attachedTo: "task", eventDefinition: Timer(), interrupting: true, bindingRef: BindingRef("timeout"))
|
||||
.Task("onTimeout", bindingRef: BindingRef("onTimeout"))
|
||||
.EndEvent("timedOut")
|
||||
.ConnectSequence("start", "task", "end")
|
||||
.ConnectSequence("timeout", "onTimeout", "timedOut")
|
||||
.Build();
|
||||
|
||||
return Scope("scope", definition, Blocking("task", log), Blocking("timeout", log), Immediate("onTimeout", log));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// An escalation thrown out of an embedded subprocess, caught by a non-interrupting escalation boundary event on
|
||||
/// the subprocess. The nested scope keeps running, which is what "non-interrupting" means.
|
||||
/// </summary>
|
||||
public static BpmnProcess EscalationOutOfSubprocess(BpmnTestLog log)
|
||||
{
|
||||
// subFirst runs before anything the parent could be confused with, deliberately: it puts the nested scope's
|
||||
// handle counter ahead of the parent's. A host that recognised its work by a shared, rewritable key rather than
|
||||
// by the child activity execution would otherwise be rescued by two independent counters happening to agree.
|
||||
var body = new BpmnProcessBuilder("subprocess-body")
|
||||
.StartEvent("subStart")
|
||||
.Task("subFirst", bindingRef: BindingRef("subFirst"))
|
||||
.Task("subWork", bindingRef: BindingRef("subWork"))
|
||||
.IntermediateThrowEvent("subEscalate", Escalation("REVIEW"))
|
||||
.Task("subMore", bindingRef: BindingRef("subMore"))
|
||||
.EndEvent("subEnd")
|
||||
.ConnectSequence("subStart", "subFirst", "subWork", "subEscalate", "subMore", "subEnd")
|
||||
.Build();
|
||||
|
||||
var definition = new BpmnProcessBuilder("escalation-out-of-subprocess")
|
||||
.StartEvent("start")
|
||||
.SubProcess("sub", bindingRef: BindingRef("sub"))
|
||||
.Task("after", bindingRef: BindingRef("after"))
|
||||
.EndEvent("end")
|
||||
.BoundaryEvent("escalated", attachedTo: "sub", eventDefinition: Escalation("REVIEW"), interrupting: false)
|
||||
.Task("notify", bindingRef: BindingRef("notify"))
|
||||
.EndEvent("notified")
|
||||
.ConnectSequence("start", "sub", "after", "end")
|
||||
.ConnectSequence("escalated", "notify", "notified")
|
||||
.Build();
|
||||
|
||||
// subMore blocks so the subprocess is still running when the escalation path executes, which is what makes
|
||||
// "the escalating work is still live" observable rather than merely asserted.
|
||||
var nested = Scope("sub", body, Immediate("subFirst", log), Blocking("subWork", log), Blocking("subMore", log));
|
||||
|
||||
return Scope("scope", definition, nested, Immediate("after", log), Immediate("notify", log));
|
||||
}
|
||||
|
||||
/// <summary>A parallel gateway split and join.</summary>
|
||||
public static BpmnProcess ParallelSplitAndJoin(BpmnTestLog log)
|
||||
{
|
||||
var definition = new BpmnProcessBuilder("parallel-split-and-join")
|
||||
.StartEvent("start")
|
||||
.ParallelGateway("split")
|
||||
.Task("left", bindingRef: BindingRef("left"))
|
||||
.Task("right", bindingRef: BindingRef("right"))
|
||||
.ParallelGateway("join")
|
||||
.Task("after", bindingRef: BindingRef("after"))
|
||||
.EndEvent("end")
|
||||
.ConnectSequence("start", "split")
|
||||
.Connect("split", "left")
|
||||
.Connect("split", "right")
|
||||
.Connect("left", "join")
|
||||
.Connect("right", "join")
|
||||
.ConnectSequence("join", "after", "end")
|
||||
.Build();
|
||||
|
||||
return Scope("scope", definition, Immediate("left", log), Immediate("right", log), Immediate("after", log));
|
||||
}
|
||||
|
||||
/// <summary>A task that fails, with an error boundary event that catches it.</summary>
|
||||
public static BpmnProcess ErrorBoundaryCaught(BpmnTestLog log)
|
||||
{
|
||||
var definition = new BpmnProcessBuilder("error-boundary-caught")
|
||||
.StartEvent("start")
|
||||
.Task("risky", bindingRef: BindingRef("risky"))
|
||||
.EndEvent("end")
|
||||
.BoundaryEvent("oops", attachedTo: "risky", eventDefinition: new BpmnEventDefinition(BpmnEventDefinitionTypes.Error))
|
||||
.Task("recover", bindingRef: BindingRef("recover"))
|
||||
.EndEvent("recovered")
|
||||
.ConnectSequence("start", "risky", "end")
|
||||
.ConnectSequence("oops", "recover", "recovered")
|
||||
.Build();
|
||||
|
||||
return Scope("scope", definition, Faulting("risky", log), Immediate("recover", log));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A task that fails inside an embedded subprocess with nothing there to catch it, and an error boundary event on
|
||||
/// the subprocess in the enclosing scope that does.
|
||||
/// </summary>
|
||||
public static BpmnProcess ErrorPropagatedOutOfSubprocess(BpmnTestLog log)
|
||||
{
|
||||
var body = new BpmnProcessBuilder("failing-subprocess-body")
|
||||
.StartEvent("subStart")
|
||||
.Task("subRisky", bindingRef: BindingRef("subRisky"))
|
||||
.EndEvent("subEnd")
|
||||
.ConnectSequence("subStart", "subRisky", "subEnd")
|
||||
.Build();
|
||||
|
||||
var definition = new BpmnProcessBuilder("error-propagated-out-of-subprocess")
|
||||
.StartEvent("start")
|
||||
.SubProcess("sub", bindingRef: BindingRef("sub"))
|
||||
.Task("after", bindingRef: BindingRef("after"))
|
||||
.EndEvent("end")
|
||||
.BoundaryEvent("subOops", attachedTo: "sub", eventDefinition: new BpmnEventDefinition(BpmnEventDefinitionTypes.Error))
|
||||
.Task("subRecover", bindingRef: BindingRef("subRecover"))
|
||||
.EndEvent("subRecovered")
|
||||
.ConnectSequence("start", "sub", "after", "end")
|
||||
.ConnectSequence("subOops", "subRecover", "subRecovered")
|
||||
.Build();
|
||||
|
||||
var nested = Scope("sub", body, Faulting("subRisky", log));
|
||||
|
||||
return Scope("scope", definition, nested, Immediate("after", log), Immediate("subRecover", log));
|
||||
}
|
||||
|
||||
/// <summary>A task that fails with nothing to catch it.</summary>
|
||||
public static BpmnProcess UncaughtError(BpmnTestLog log)
|
||||
{
|
||||
var definition = new BpmnProcessBuilder("uncaught-error")
|
||||
.StartEvent("start")
|
||||
.Task("risky", bindingRef: BindingRef("risky"))
|
||||
.EndEvent("end")
|
||||
.ConnectSequence("start", "risky", "end")
|
||||
.Build();
|
||||
|
||||
return Scope("scope", definition, Faulting("risky", log));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A parallel multi-instance task: two concurrent instances of one binding, told apart only by their iteration id.
|
||||
/// </summary>
|
||||
public static BpmnProcess ParallelMultiInstanceTask(BpmnTestLog log)
|
||||
{
|
||||
var definition = new BpmnProcessBuilder("parallel-multi-instance-task")
|
||||
.StartEvent("start")
|
||||
.Task(BpmnElementTypes.Task, "each", bindingRef: BindingRef("each"), loopCharacteristics: new BpmnLoopCharacteristics(isSequential: false, cardinality: 2))
|
||||
.Task("after", bindingRef: BindingRef("after"))
|
||||
.EndEvent("end")
|
||||
.ConnectSequence("start", "each", "after", "end")
|
||||
.Build();
|
||||
|
||||
return Scope("scope", definition, Blocking("each", log), Immediate("after", log));
|
||||
}
|
||||
|
||||
/// <summary>A linear process: one task between a start and an end event.</summary>
|
||||
public static BpmnProcess LinearTask(BpmnTestLog log)
|
||||
{
|
||||
var definition = new BpmnProcessBuilder("linear-task")
|
||||
.StartEvent("start")
|
||||
.Task("only", bindingRef: BindingRef("only"))
|
||||
.EndEvent("end")
|
||||
.ConnectSequence("start", "only", "end")
|
||||
.Build();
|
||||
|
||||
return Scope("scope", definition, Immediate("only", log));
|
||||
}
|
||||
|
||||
/// <summary>The binding ref the given element's work is declared under.</summary>
|
||||
public static string BindingRef(string elementId) => $"node-{elementId}";
|
||||
|
||||
private static BpmnEventDefinition Timer() => new(BpmnEventDefinitionTypes.Timer);
|
||||
|
||||
private static BpmnEventDefinition Escalation(string code) =>
|
||||
new(BpmnEventDefinitionTypes.Escalation, new Dictionary<string, string>(StringComparer.Ordinal) { [BpmnEventDefinitionProperties.Code] = code });
|
||||
|
||||
private static BpmnProcess Scope(string id, BpmnProcessDefinition definition, params IActivity[] work) => new()
|
||||
{
|
||||
Id = id,
|
||||
Process = definition,
|
||||
Activities = work.ToList(),
|
||||
WorkBindings = work.ToDictionary(activity => BindingRef(activity.Id), activity => activity.Id, StringComparer.Ordinal)
|
||||
};
|
||||
|
||||
private static BpmnTestWork Immediate(string id, BpmnTestLog log) => new()
|
||||
{
|
||||
Id = id,
|
||||
Log = log
|
||||
};
|
||||
|
||||
private static BpmnTestBlockingWork Blocking(string id, BpmnTestLog log) => new()
|
||||
{
|
||||
Id = id,
|
||||
Log = log
|
||||
};
|
||||
|
||||
private static BpmnTestFaultingWork Faulting(string id, BpmnTestLog log) => new()
|
||||
{
|
||||
Id = id,
|
||||
Log = log
|
||||
};
|
||||
}
|
||||
67
test/unit/Elsa.Bpmn.UnitTests/BpmnScopeDispatcherTests.cs
Normal file
67
test/unit/Elsa.Bpmn.UnitTests/BpmnScopeDispatcherTests.cs
Normal file
|
|
@ -0,0 +1,67 @@
|
|||
using Elsa.Bpmn.Hosting;
|
||||
|
||||
namespace Elsa.Bpmn.UnitTests;
|
||||
|
||||
/// <summary>
|
||||
/// The queue that keeps a scope evaluation raised while another one is applying its commands from running on top of it.
|
||||
/// </summary>
|
||||
public class BpmnScopeDispatcherTests
|
||||
{
|
||||
private readonly BpmnScopeDispatcher _dispatcher = new();
|
||||
private readonly List<string> _log = [];
|
||||
|
||||
[Fact(DisplayName = "An evaluation posted while one is in flight runs after it, not inside it")]
|
||||
public async Task PostFromInsideARunningEvaluation_IsQueued()
|
||||
{
|
||||
await _dispatcher.PostAsync(async () =>
|
||||
{
|
||||
_log.Add("outer:begin");
|
||||
await _dispatcher.PostAsync(() =>
|
||||
{
|
||||
_log.Add("inner");
|
||||
return default;
|
||||
});
|
||||
_log.Add("outer:end");
|
||||
});
|
||||
|
||||
Assert.Equal(["outer:begin", "outer:end", "inner"], _log);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Evaluations run in the order they were posted")]
|
||||
public async Task QueuedEvaluations_RunInPostOrder()
|
||||
{
|
||||
await _dispatcher.PostAsync(async () =>
|
||||
{
|
||||
await _dispatcher.PostAsync(() => Add("first"));
|
||||
await _dispatcher.PostAsync(() => Add("second"));
|
||||
await Add("root");
|
||||
});
|
||||
|
||||
Assert.Equal(["root", "first", "second"], _log);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "An evaluation that throws drops what was queued behind it")]
|
||||
public async Task FailedEvaluation_DropsTheRest()
|
||||
{
|
||||
// Everything queued behind a failed evaluation was planned against a state that no longer describes the
|
||||
// instance, and the failure is on its way to the incident strategy.
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(async () => await _dispatcher.PostAsync(async () =>
|
||||
{
|
||||
await _dispatcher.PostAsync(() => Add("queued"));
|
||||
throw new InvalidOperationException("boom");
|
||||
}));
|
||||
|
||||
Assert.Empty(_log);
|
||||
Assert.False(_dispatcher.IsDraining);
|
||||
|
||||
// The queue is usable again, and does not resurrect what was dropped.
|
||||
await _dispatcher.PostAsync(() => Add("after"));
|
||||
Assert.Equal(["after"], _log);
|
||||
}
|
||||
|
||||
private ValueTask Add(string entry)
|
||||
{
|
||||
_log.Add(entry);
|
||||
return default;
|
||||
}
|
||||
}
|
||||
128
test/unit/Elsa.Bpmn.UnitTests/BpmnWorkTeardownTests.cs
Normal file
128
test/unit/Elsa.Bpmn.UnitTests/BpmnWorkTeardownTests.cs
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
using Bpmn.Semantics;
|
||||
using Elsa.Bpmn.Activities;
|
||||
using Elsa.Bpmn.Hosting;
|
||||
using Elsa.Testing.Shared;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.Models;
|
||||
using Elsa.Workflows.Options;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace Elsa.Bpmn.UnitTests;
|
||||
|
||||
/// <summary>
|
||||
/// Covers the one refusal <see cref="BpmnWorkTeardown"/> can make: a subtree that still has a scheduled-but-not-yet-
|
||||
/// invoked activity in it. <see cref="IActivityScheduler"/> offers no way to withdraw a queued work item, so running
|
||||
/// it after BPMN destroyed its branch would leave a stray live branch behind. The host must throw rather than let
|
||||
/// that happen silently, and — since that throw is absorbed rather than propagated under
|
||||
/// <c>ContinueWithIncidentsStrategy</c> — the caller applying the teardown must not leave the persisted ledger
|
||||
/// claiming work it just tore down.
|
||||
/// </summary>
|
||||
public class BpmnWorkTeardownTests
|
||||
{
|
||||
private const string SubtreeActivityId = "subtree-activity";
|
||||
private const string QueuedActivityId = "queued-activity";
|
||||
|
||||
[Fact]
|
||||
public async Task CancelSubtreeAsync_ThrowsNamingElementAndStrandedActivity_WhenSubtreeHasWorkStillQueued()
|
||||
{
|
||||
var (_, subtreeContext, _) = await BuildSubtreeWithQueuedWorkAsync();
|
||||
|
||||
var exception = await Assert.ThrowsAsync<NotSupportedException>(
|
||||
() => BpmnWorkTeardown.CancelSubtreeAsync(subtreeContext, "boundary interrupted").AsTask());
|
||||
|
||||
Assert.Contains(SubtreeActivityId, exception.Message);
|
||||
Assert.Contains(QueuedActivityId, exception.Message);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ApplyAsync_RemovesLedgerRecordBeforeTheRefusalPropagates_WhenSubtreeHasWorkStillQueued()
|
||||
{
|
||||
var (scopeContext, subtreeContext, process) = await BuildSubtreeWithQueuedWorkAsync();
|
||||
var memory = SeedLedgerRecord(scopeContext, subtreeContext);
|
||||
var applier = new BpmnCommandApplier(scopeContext, process, memory);
|
||||
|
||||
await Assert.ThrowsAsync<NotSupportedException>(() => applier.ApplyAsync(
|
||||
[new BpmnHostCommand.CancelWorkSubtree("work-1", SubtreeActivityId, "boundary interrupted")]).AsTask());
|
||||
|
||||
// The ledger property is reloaded from scratch, from what was actually persisted onto the context, rather
|
||||
// than read off the in-memory `memory` instance the applier already mutated.
|
||||
var reloaded = BpmnScopeMemory.Load(scopeContext);
|
||||
Assert.Null(reloaded.Work.FindByChildContextId(subtreeContext.Id));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task OnWorkCompletedAsync_DiscardsTheCallback_ForContextTornDownByARefusedCancellation()
|
||||
{
|
||||
var (scopeContext, subtreeContext, process) = await BuildSubtreeWithQueuedWorkAsync();
|
||||
var memory = SeedLedgerRecord(scopeContext, subtreeContext);
|
||||
var applier = new BpmnCommandApplier(scopeContext, process, memory);
|
||||
|
||||
await Assert.ThrowsAsync<NotSupportedException>(() => applier.ApplyAsync(
|
||||
[new BpmnHostCommand.CancelWorkSubtree("work-1", SubtreeActivityId, "boundary interrupted")]).AsTask());
|
||||
|
||||
// `process.Process` is deliberately left unset. A completion that is fed to the interpreter rather than
|
||||
// discarded reaches BpmnScopeHost.Graph, which throws InvalidOperationException for want of a process
|
||||
// definition. So this call succeeding is itself the assertion: the callback for the torn-down context must
|
||||
// never get that far.
|
||||
await BpmnScopeHost.For(scopeContext).OnWorkCompletedAsync(subtreeContext, null);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A three-level activity tree, built as the actual workflow definition, so every context created below has a
|
||||
/// real ActivityNode: the scope is the BPMN process the ledger belongs to, the subtree is the unit of work being
|
||||
/// torn down, and the queued activity is the work BPMN scheduled but the engine has not invoked yet.
|
||||
/// </summary>
|
||||
private static async Task<(ActivityExecutionContext ScopeContext, ActivityExecutionContext SubtreeContext, BpmnProcess Process)> BuildSubtreeWithQueuedWorkAsync()
|
||||
{
|
||||
var queuedActivity = new WriteLine("queued") { Id = QueuedActivityId };
|
||||
var subtreeActivity = new Sequence { Id = SubtreeActivityId, Activities = { queuedActivity } };
|
||||
var process = new BpmnProcess { Id = "process-activity", Activities = { subtreeActivity } };
|
||||
|
||||
var fixture = new ActivityTestFixture(process);
|
||||
|
||||
// The fixture's default IIdentityGenerator is an unconfigured substitute that hands out the same (null)
|
||||
// id to every context; each context created below needs its own real one.
|
||||
fixture.ConfigureServices(services => services.AddSingleton<IIdentityGenerator, GuidIdentityGenerator>());
|
||||
|
||||
var scopeContext = await fixture.BuildAsync();
|
||||
var workflowExecutionContext = scopeContext.WorkflowExecutionContext;
|
||||
|
||||
// Only the root's own type is registered by the fixture; the nested activities need registering too so
|
||||
// their descriptors can be resolved when their contexts are created below.
|
||||
await workflowExecutionContext.ActivityRegistry.RegisterAsync(typeof(Sequence));
|
||||
await workflowExecutionContext.ActivityRegistry.RegisterAsync(typeof(WriteLine));
|
||||
|
||||
// The subtree root being torn down: already running, standing in for the BPMN unit of work a boundary event
|
||||
// just interrupted.
|
||||
var subtreeContext = await workflowExecutionContext.CreateActivityExecutionContextAsync(subtreeActivity, new ActivityInvocationOptions { Owner = scopeContext });
|
||||
workflowExecutionContext.AddActivityExecutionContext(subtreeContext);
|
||||
subtreeContext.TransitionTo(ActivityStatus.Running);
|
||||
|
||||
// A child of that subtree BPMN has already scheduled but the engine has not yet invoked: its context exists
|
||||
// and is still Pending, and a matching work item sits in the scheduler.
|
||||
var queuedContext = await workflowExecutionContext.CreateActivityExecutionContextAsync(queuedActivity, new ActivityInvocationOptions { Owner = subtreeContext });
|
||||
workflowExecutionContext.AddActivityExecutionContext(queuedContext);
|
||||
workflowExecutionContext.Scheduler.Schedule(new ActivityWorkItem(queuedActivity, existingActivityExecutionContext: queuedContext));
|
||||
|
||||
return (scopeContext, subtreeContext, process);
|
||||
}
|
||||
|
||||
/// <summary>Records the subtree as the scope's live work, exactly as StartWork would have, and persists it.</summary>
|
||||
private static BpmnScopeMemory SeedLedgerRecord(ActivityExecutionContext scopeContext, ActivityExecutionContext subtreeContext)
|
||||
{
|
||||
var memory = BpmnScopeMemory.Load(scopeContext);
|
||||
|
||||
memory.Work.Records.Add(new BpmnWorkRecord
|
||||
{
|
||||
Handle = "work-1",
|
||||
BindingRef = "subtree-binding",
|
||||
ElementId = SubtreeActivityId,
|
||||
ChildContextId = subtreeContext.Id
|
||||
});
|
||||
|
||||
memory.SaveWork();
|
||||
|
||||
return BpmnScopeMemory.Load(scopeContext);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue