diff --git a/src/modules/Elsa.Bpmn/Activities/BpmnProcess.cs b/src/modules/Elsa.Bpmn/Activities/BpmnProcess.cs new file mode 100644 index 000000000..69d03bd5d --- /dev/null +++ b/src/modules/Elsa.Bpmn/Activities/BpmnProcess.cs @@ -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; + +/// +/// Runs one BPMN process scope, driving the Bpmn.Semantics interpreter and applying what it returns onto this +/// activity's execution context. +/// +/// +/// +/// A scope owns its own execution state and its own record of the work it started, both held in +/// . A nested BPMN scope — an embedded subprocess, or an event +/// subprocess body — is another bound as work, so the scope hierarchy the interpreter has +/// no view of is exactly the activity hierarchy Elsa already maintains. +/// +/// +/// Like every container, this one never auto-completes: it completes when the interpreter returns a Complete +/// continuation, and its outcome is what a conditional sequence flow in the enclosing scope selects on. +/// +/// +[Activity("Elsa", "BPMN", "Executes a BPMN process scope.")] +[System.ComponentModel.Browsable(false)] +public class BpmnProcess : Container +{ + /// + public BpmnProcess([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line) + { + OnSignalReceived(OnScopeSignalledAsync); + OnSignalReceived(OnWorkFaultedAsync); + } + + /// + /// The BPMN process definition this scope executes. + /// + public BpmnProcessDefinition? Process { get; set; } + + /// + /// Maps each binding ref the definition declares to the id of the activity in + /// that runs it. + /// + /// + /// 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. + /// + public IDictionary WorkBindings { get; set; } = new Dictionary(StringComparer.Ordinal); + + /// + protected override ValueTask ScheduleChildrenAsync(ActivityExecutionContext context) => BpmnScopeHost.For(context).StartAsync(); + + /// + /// The activity bound to the given binding ref, or null when the definition declares a binding this + /// activity does not map. + /// + internal IActivity? FindWorkActivity(string bindingRef) => + WorkBindings.TryGetValue(bindingRef, out var activityId) + ? Activities.FirstOrDefault(activity => string.Equals(activity.Id, activityId, StringComparison.Ordinal)) + : null; + + /// + /// A unit of work completed. Named rather than a lambda, because completion callbacks are rehydrated by method name. + /// + 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); +} diff --git a/src/modules/Elsa.Bpmn/AssemblyInfo.cs b/src/modules/Elsa.Bpmn/AssemblyInfo.cs new file mode 100644 index 000000000..dc926c8f7 --- /dev/null +++ b/src/modules/Elsa.Bpmn/AssemblyInfo.cs @@ -0,0 +1,4 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("Elsa.Bpmn.UnitTests")] +[assembly: InternalsVisibleTo("Elsa.Bpmn.IntegrationTests")] diff --git a/src/modules/Elsa.Bpmn/Exceptions/BpmnScopeFaultException.cs b/src/modules/Elsa.Bpmn/Exceptions/BpmnScopeFaultException.cs new file mode 100644 index 000000000..c56fd6ed9 --- /dev/null +++ b/src/modules/Elsa.Bpmn/Exceptions/BpmnScopeFaultException.cs @@ -0,0 +1,29 @@ +namespace Elsa.Bpmn.Exceptions; + +/// +/// Thrown when the interpreter returns a Fault continuation for a BPMN scope: the scope failed +/// deterministically and cannot continue. +/// +/// +/// A propagated work fault does not surface this way. That case is handled where it arrives — inside the scope's +/// 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. +/// +public class BpmnScopeFaultException : Exception +{ + /// + /// Initializes a new instance of the class. + /// + /// The interpreter's stable fault code. + /// The interpreter's human-readable explanation. + public BpmnScopeFaultException(string code, string message) : base($"{code}: {message}") + { + Code = code; + } + + /// + /// The interpreter's stable fault code. + /// + public string Code { get; } +} diff --git a/src/modules/Elsa.Bpmn/Features/BpmnFeature.cs b/src/modules/Elsa.Bpmn/Features/BpmnFeature.cs index 0d6841542..0b57fa4dc 100644 --- a/src/modules/Elsa.Bpmn/Features/BpmnFeature.cs +++ b/src/modules/Elsa.Bpmn/Features/BpmnFeature.cs @@ -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; /// /// Provides BPMN execution support to the system. /// +[DependsOn(typeof(WorkflowManagementFeature))] public class BpmnFeature : FeatureBase { /// public BpmnFeature(IModule module) : base(module) { } + + /// + public override void Apply() + { + Module.AddActivity(); + } } diff --git a/src/modules/Elsa.Bpmn/Hosting/BpmnCommandApplier.cs b/src/modules/Elsa.Bpmn/Hosting/BpmnCommandApplier.cs new file mode 100644 index 000000000..82be12c83 --- /dev/null +++ b/src/modules/Elsa.Bpmn/Hosting/BpmnCommandApplier.cs @@ -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; + +/// +/// Translates the interpreter's three host commands onto . +/// +/// +/// +/// StartWork. +/// CancelWorkSubtreeCancelActivityAsync, which already walks the child subtree recursively. +/// SignalEnclosingScopeSendSignalAsync, which bubbles to ancestors. +/// +/// +internal sealed class BpmnCommandApplier(ActivityExecutionContext scopeContext, BpmnProcess process, BpmnScopeMemory memory) +{ + /// + /// Applies a command list in the order returned. + /// + /// + /// The ordering carries meaning and is not an implementation detail. An interrupting boundary event emits the + /// boundary path's StartWork before the teardown that retires the host it interrupted, and a host + /// that tidied up first would be applying a different process. + /// + public async ValueTask ApplyAsync(IReadOnlyList 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? 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() + }; +} diff --git a/src/modules/Elsa.Bpmn/Hosting/BpmnScopeDispatcher.cs b/src/modules/Elsa.Bpmn/Hosting/BpmnScopeDispatcher.cs new file mode 100644 index 000000000..affd97776 --- /dev/null +++ b/src/modules/Elsa.Bpmn/Hosting/BpmnScopeDispatcher.cs @@ -0,0 +1,60 @@ +namespace Elsa.Bpmn.Hosting; + +/// +/// Runs BPMN scope evaluations one at a time, in arrival order, for one workflow instance. +/// +/// +/// +/// 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. +/// +/// +/// 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. +/// +/// +internal sealed class BpmnScopeDispatcher +{ + private readonly Queue> _queue = new(); + + private bool _draining; + + /// Whether an evaluation is in flight, so a post will be queued rather than run. + public bool IsDraining => _draining; + + /// + /// Queues an evaluation and, unless one is already in flight, drains the queue to exhaustion. + /// + public async ValueTask PostAsync(Func 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; + } + } +} diff --git a/src/modules/Elsa.Bpmn/Hosting/BpmnScopeHost.cs b/src/modules/Elsa.Bpmn/Hosting/BpmnScopeHost.cs new file mode 100644 index 000000000..4b3774b5a --- /dev/null +++ b/src/modules/Elsa.Bpmn/Hosting/BpmnScopeHost.cs @@ -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; + +/// +/// The host side of the Bpmn.Semantics port for one BPMN scope: it feeds the interpreter's four entry points +/// and applies what comes back onto the scope's . +/// +/// +/// +/// 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. +/// +/// +/// A host instance is a view over one scope's context and is created per call. Everything durable lives in +/// , everything derived lives in the context's transient properties, and everything +/// ordering-related lives in the instance-wide . +/// +/// +internal sealed class BpmnScopeHost +{ + /// + /// What this host promises it can do. + /// + /// + /// is deliberately absent: reading container-scoped variables + /// through 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. + /// + public const BpmnHostCapabilities Capabilities = + BpmnHostCapabilities.SubtreeCancellation | BpmnHostCapabilities.ScopeSignalling | BpmnHostCapabilities.IterationScopes; + + /// + /// The property key under which a nested scope's invocation correlation is carried on its own context. + /// + 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 NoCorrelation = new Dictionary(StringComparer.Ordinal); + + private readonly ActivityExecutionContext _context; + private readonly BpmnProcess _process; + + private BpmnScopeHost(ActivityExecutionContext context) + { + _context = context; + _process = (BpmnProcess)context.Activity; + } + + /// Returns the host for the given BPMN scope context. + public static BpmnScopeHost For(ActivityExecutionContext context) => new(context); + + /// The evaluation queue shared by every BPMN scope in this workflow instance. + public static BpmnScopeDispatcher DispatcherOf(WorkflowExecutionContext context) => + context.TransientProperties.GetOrAdd(DispatcherKey, () => new BpmnScopeDispatcher()); + + /// The built graph for this scope. Derived from the definition, the bound work and the capabilities, none of which vary per instance. + public BpmnGraph Graph => _context.TransientProperties.GetOrAdd(GraphTransientPropertyKey, BuildGraph); + + // --- The interpreter's four entry points --------------------------------------------------------- + + /// The scope is beginning. + public ValueTask StartAsync() => EvaluateAsync(memory => + Interpreter.Start(new BpmnStartRequest(Graph, memory.State, Snapshot(memory)))); + + /// A unit of work finished, reporting zero or more outcome names. + 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)); + }); + + /// A nested scope this one invoked signalled outward. + 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)); + }); + } + + /// + /// A unit of work failed. Rides the seam: handle the signal, ask the interpreter what + /// BPMN makes of the fault, and claim it only when a catcher took it. + /// + /// + /// 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 Propagated disposition is left strictly alone — no + /// StopPropagation, 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. + /// + 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 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."); + } + } + + /// + /// Finds the unit of work this scope started that the failing activity belongs to, walking outward from the + /// failure. + /// + /// + /// 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. + /// + 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); + } + + /// + /// 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. + /// + private IReadOnlyDictionary InvocationCorrelation => + BpmnScopeMemory.Read>(_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); + } +} diff --git a/src/modules/Elsa.Bpmn/Hosting/BpmnScopeMemory.cs b/src/modules/Elsa.Bpmn/Hosting/BpmnScopeMemory.cs new file mode 100644 index 000000000..853ae31af --- /dev/null +++ b/src/modules/Elsa.Bpmn/Hosting/BpmnScopeMemory.cs @@ -0,0 +1,65 @@ +using System.Text.Json; +using Bpmn.Model.State; +using Elsa.Workflows; + +namespace Elsa.Bpmn.Hosting; + +/// +/// A BPMN scope's own memory: the interpreter's execution state, and the ledger of work this scope started. +/// +/// +/// Both live in the scope's , because that is where per-execution +/// state belongs and because the handle-to-context map must survive anything that rewrites +/// . Both are held as serialized JSON strings rather than as objects: +/// the property bag is written through PolymorphicObjectConverter, which silently drops the type of a value +/// whose type has no registered alias and hands back a loose ExpandoObject on the way in. +/// +internal sealed class BpmnScopeMemory +{ + /// The property key holding the interpreter's execution state. + public const string ExecutionStatePropertyKey = "Bpmn:ExecutionState"; + + /// The property key holding this scope's work ledger. + 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; + } + + /// The state the interpreter last returned, or null before the scope has been started. + public BpmnExecutionState? State { get; set; } + + /// The work this scope has started and not finished. + public BpmnWorkLedger Work { get; } + + /// Reads a scope's memory from its property bag. + public static BpmnScopeMemory Load(ActivityExecutionContext context) => + new(context, Read(context, ExecutionStatePropertyKey), Read(context, WorkLedgerPropertyKey) ?? new BpmnWorkLedger()); + + /// Reads a JSON string property, or returns null when it is absent. + public static T? Read(ActivityExecutionContext context, string key) where T : class => + context.Properties.TryGetValue(key, out var value) && value is string json && !string.IsNullOrWhiteSpace(json) + ? JsonSerializer.Deserialize(json, SerializerOptions) + : null; + + /// Writes a value as a JSON string property. + public static void Write(ActivityExecutionContext context, string key, T value) => + context.Properties[key] = JsonSerializer.Serialize(value, SerializerOptions); + + /// Persists the execution state. + public void SaveState() + { + if (State is not null) + Write(_context, ExecutionStatePropertyKey, State); + } + + /// Persists the work ledger. + public void SaveWork() => Write(_context, WorkLedgerPropertyKey, Work); +} diff --git a/src/modules/Elsa.Bpmn/Hosting/BpmnWorkLedger.cs b/src/modules/Elsa.Bpmn/Hosting/BpmnWorkLedger.cs new file mode 100644 index 000000000..f351bd93f --- /dev/null +++ b/src/modules/Elsa.Bpmn/Hosting/BpmnWorkLedger.cs @@ -0,0 +1,48 @@ +using Bpmn.Semantics; + +namespace Elsa.Bpmn.Hosting; + +/// +/// A BPMN scope's record of the work it has started and not finished — the handle-to-context map, and the source of +/// . +/// +/// +/// +/// 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 (binding ref, iteration id) 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. +/// +/// +/// The interpreter re-finds a parked token from (binding ref, iteration id) 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. +/// +/// +internal sealed class BpmnWorkLedger +{ + /// The handle counter. Handles are scope-local so a scope's trace reads independently of Elsa's id generator. + public int Sequence { get; set; } + + /// The live work, in start order. + public List Records { get; set; } = []; + + /// Mints the next scope-local handle. + public string NextHandle() => $"work-{++Sequence}"; + + /// The live work running in the given child activity execution, or null. + public BpmnWorkRecord? FindByChildContextId(string childContextId) => + Records.FirstOrDefault(x => string.Equals(x.ChildContextId, childContextId, StringComparison.Ordinal)); + + /// The live work with the given handle, or null. + public BpmnWorkRecord? FindByHandle(string handle) => + Records.FirstOrDefault(x => string.Equals(x.Handle, handle, StringComparison.Ordinal)); + + /// Drops a record, so the work is no longer reported as live. + public void Remove(BpmnWorkRecord record) => Records.Remove(record); + + /// The interpreter's view of this scope's live work. + public IReadOnlyCollection ToLiveWork() => + Records.Select(x => new BpmnLiveWork(x.BindingRef, x.IterationId, x.Handle)).ToArray(); +} diff --git a/src/modules/Elsa.Bpmn/Hosting/BpmnWorkRecord.cs b/src/modules/Elsa.Bpmn/Hosting/BpmnWorkRecord.cs new file mode 100644 index 000000000..59eb50fae --- /dev/null +++ b/src/modules/Elsa.Bpmn/Hosting/BpmnWorkRecord.cs @@ -0,0 +1,33 @@ +namespace Elsa.Bpmn.Hosting; + +/// +/// 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. +/// +/// +/// +/// is the whole point of this record. A unit of work is keyed on the child +/// 's id and never on ActivityExecutionContext.Tag. +/// Tag is shared mutable state: the completion-callback dispatch writes the popped callback entry's tag onto the +/// receiving 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. +/// +/// +internal sealed class BpmnWorkRecord +{ + /// The opaque host handle the interpreter echoes back. Scope-local and minted by . + public string Handle { get; set; } = null!; + + /// The binding ref the interpreter asked to start. + public string BindingRef { get; set; } = null!; + + /// The multi-instance iteration id this unit runs under, or null for ordinary single-run work. + public string? IterationId { get; set; } + + /// The element the work belongs to, for reporting. + public string ElementId { get; set; } = null!; + + /// The id of the child running this work. + public string ChildContextId { get; set; } = null!; +} diff --git a/src/modules/Elsa.Bpmn/Hosting/BpmnWorkTeardown.cs b/src/modules/Elsa.Bpmn/Hosting/BpmnWorkTeardown.cs new file mode 100644 index 000000000..f2cf445c7 --- /dev/null +++ b/src/modules/Elsa.Bpmn/Hosting/BpmnWorkTeardown.cs @@ -0,0 +1,58 @@ +using Elsa.Extensions; +using Elsa.Workflows; + +namespace Elsa.Bpmn.Hosting; + +/// +/// Stops a unit of work and everything it in turn started. +/// +/// +/// Both places that tear work down go through here: the interpreter's CancelWorkSubtree 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. +/// +internal static class BpmnWorkTeardown +{ + /// The activity execution with the given id, or null when it has already gone. + public static ActivityExecutionContext? FindContext(WorkflowExecutionContext workflowExecutionContext, string activityExecutionContextId) => + workflowExecutionContext.ActivityExecutionContexts.FirstOrDefault(x => string.Equals(x.Id, activityExecutionContextId, StringComparison.Ordinal)); + + /// + /// Tears down a unit of work and everything underneath it, or refuses when it cannot. + /// + /// + /// + /// Cancellation itself needs no new code: the public CancelActivityAsync extension already walks the child + /// subtree recursively, which is exactly what "and everything it in turn started" means. + /// + /// + /// 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 IActivityScheduler 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 FaultStrategy, which is the strategy that justifies it. It does not fail + /// loudly under ContinueWithIncidentsStrategy: 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. + /// + /// + 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(); + } +} diff --git a/src/modules/Elsa.Bpmn/Signals/BpmnScopeSignal.cs b/src/modules/Elsa.Bpmn/Signals/BpmnScopeSignal.cs new file mode 100644 index 000000000..2dcf55124 --- /dev/null +++ b/src/modules/Elsa.Bpmn/Signals/BpmnScopeSignal.cs @@ -0,0 +1,17 @@ +using System.Text.Json; + +namespace Elsa.Bpmn.Signals; + +/// +/// Carries the interpreter's SignalEnclosingScope command from a nested BPMN scope to the scope that started it. +/// +/// +/// 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 BpmnInterpreter.EscalationSignalCode with its identity in the payload. +/// +/// The signal code. +/// The signal payload. +public record BpmnScopeSignal(string Code, JsonElement? Payload); diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestBlockingWork.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestBlockingWork.cs new file mode 100644 index 000000000..08860ac81 --- /dev/null +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestBlockingWork.cs @@ -0,0 +1,32 @@ +using Elsa.Workflows; +using Elsa.Workflows.Signals; + +namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities; + +/// +/// 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. +/// +public class BpmnTestBlockingWork : Activity +{ + /// + public BpmnTestBlockingWork() => OnSignalReceived(OnCancelled); + + /// The log to record into. + public BpmnTestLog? Log { get; set; } + + /// + 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); +} diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestFaultingWork.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestFaultingWork.cs new file mode 100644 index 000000000..d359c6633 --- /dev/null +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestFaultingWork.cs @@ -0,0 +1,26 @@ +using Elsa.Workflows; +using Elsa.Workflows.Signals; + +namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities; + +/// +/// 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. +/// +public class BpmnTestFaultingWork : CodeActivity +{ + /// + public BpmnTestFaultingWork() => OnSignalReceived(OnCancelled); + + /// The log to record into. + public BpmnTestLog? Log { get; set; } + + /// + 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); +} diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestLog.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestLog.cs new file mode 100644 index 000000000..d393cd6f0 --- /dev/null +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestLog.cs @@ -0,0 +1,49 @@ +using Elsa.Bpmn.Hosting; +using Elsa.Workflows; + +namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities; + +/// +/// An ordered record of what the stand-in work activities did, plus snapshots taken at moments the applier's +/// invariants are observable. +/// +public sealed class BpmnTestLog +{ + private readonly List _entries = []; + private readonly Dictionary> _snapshots = new(StringComparer.Ordinal); + + /// Everything that happened, in order. + public IReadOnlyList Entries => _entries; + + /// Records that something happened. + public void Record(string entry) => _entries.Add(entry); + + /// The position of an entry, or -1. + public int PositionOf(string entry) => _entries.IndexOf(entry); + + /// How many times an entry was recorded. + public int Occurrences(string entry) => _entries.Count(x => string.Equals(x, entry, StringComparison.Ordinal)); + + /// A snapshot taken under the given key. + public IReadOnlyList Snapshot(string key) => + _snapshots.TryGetValue(key, out var snapshot) + ? snapshot + : throw new InvalidOperationException($"Nothing was captured under '{key}'. Captured keys: {string.Join(", ", _snapshots.Keys)}."); + + /// + /// 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 BpmnHostSnapshot.LiveWork. + /// + 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(scopeContext, BpmnScopeMemory.WorkLedgerPropertyKey) ?? new BpmnWorkLedger(); + + _snapshots[key] = ledger.Records.Select(x => x.BindingRef).OrderBy(x => x, StringComparer.Ordinal).ToList(); + } + + /// Captures the ids of the activities currently sitting in the workflow's scheduler. + public void CaptureScheduled(string key, ActivityExecutionContext context) => + _snapshots[key] = context.WorkflowExecutionContext.Scheduler.List().Select(x => x.Activity.Id).ToList(); +} diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestWork.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestWork.cs new file mode 100644 index 000000000..d544ec978 --- /dev/null +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestWork.cs @@ -0,0 +1,21 @@ +using Elsa.Workflows; +using Elsa.Workflows.Signals; + +namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities; + +/// +/// A stand-in for BPMN work that finishes as soon as it runs. +/// +public class BpmnTestWork : CodeActivity +{ + /// + public BpmnTestWork() => OnSignalReceived(OnCancelled); + + /// The log to record into. + public BpmnTestLog? Log { get; set; } + + /// + protected override void Execute(ActivityExecutionContext context) => BpmnTestWorkProbe.RecordExecution(this, Log, context); + + private void OnCancelled(CancelSignal signal, SignalContext context) => BpmnTestWorkProbe.RecordCancellation(this, Log, context); +} diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestWorkProbe.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestWorkProbe.cs new file mode 100644 index 000000000..942aedd2b --- /dev/null +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/Activities/BpmnTestWorkProbe.cs @@ -0,0 +1,36 @@ +using Elsa.Workflows; + +namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities; + +/// +/// The recording every stand-in work activity does, kept in one place so the activities differ only in what they do. +/// +internal static class BpmnTestWorkProbe +{ + /// Records that the work started, and what its scope reported as live at that moment. + public static void RecordExecution(IActivity activity, BpmnTestLog? log, ActivityExecutionContext context) + { + log?.Record($"executed:{activity.Id}"); + log?.CaptureLiveWork($"liveWork@{activity.Id}", context); + } + + /// + /// Records that the work was torn down, and what was in the scheduler at that moment. + /// + /// + /// The scheduler snapshot is what makes command ordering observable: a teardown applied after the boundary path's + /// StartWork sees the boundary path already scheduled, and one applied before it does not. + /// + 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); + } +} diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnHostInvariantTests.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnHostInvariantTests.cs new file mode 100644 index 000000000..fbcc1830c --- /dev/null +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnHostInvariantTests.cs @@ -0,0 +1,146 @@ +using Elsa.Common.Models; +using Elsa.Workflows; +using Elsa.Workflows.IncidentStrategies; +using Xunit.Abstractions; + +namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort; + +/// +/// 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. +/// +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); + } +} diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnProcessTests.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnProcessTests.cs new file mode 100644 index 000000000..0a13b4fe5 --- /dev/null +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnProcessTests.cs @@ -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; + +/// +/// The BPMN processes the host port is exercised against, end to end through IWorkflowRunner. +/// +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; +} diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestHost.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestHost.cs new file mode 100644 index 000000000..02131d3be --- /dev/null +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestHost.cs @@ -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; + +/// +/// Runs a BPMN scope through and lets a test finish blocked work by activity id. +/// +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() + .Build(); + + _workflowRunner = _services.GetRequiredService(); + _workflowBuilderFactory = _services.GetRequiredService(); + } + + /// What the stand-in work activities recorded. + public BpmnTestLog Log { get; } = new(); + + /// Runs the given root activity to quiescence. + public async Task 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)); + } + + /// + /// 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. + /// + public async Task 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 + })); + } + + /// + /// The work the named BPMN scope currently reports as live, read from its own persisted ledger — which is the sole + /// source of BpmnHostSnapshot.LiveWork. + /// + public IReadOnlyList<(string BindingRef, string? IterationId)> LiveWorkOf(string scopeActivityId) + { + var scopeContext = _result!.Journal.ActivityExecutionContexts.First(x => x.Activity.Id == scopeActivityId); + var ledger = BpmnScopeMemory.Read(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; + } +} diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestProcesses.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestProcesses.cs new file mode 100644 index 000000000..2e6303cb3 --- /dev/null +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestProcesses.cs @@ -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; + +/// +/// The processes the applier is exercised against, built in code and bound to stand-in work activities. +/// +/// +/// 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. +/// +internal static class BpmnTestProcesses +{ + /// An interrupting timer boundary event on a long-running task. + 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)); + } + + /// + /// 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. + /// + 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)); + } + + /// A parallel gateway split and join. + 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)); + } + + /// A task that fails, with an error boundary event that catches it. + 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)); + } + + /// + /// 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. + /// + 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)); + } + + /// A task that fails with nothing to catch it. + 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)); + } + + /// + /// A parallel multi-instance task: two concurrent instances of one binding, told apart only by their iteration id. + /// + 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)); + } + + /// A linear process: one task between a start and an end event. + 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)); + } + + /// The binding ref the given element's work is declared under. + 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(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 + }; +} diff --git a/test/unit/Elsa.Bpmn.UnitTests/BpmnScopeDispatcherTests.cs b/test/unit/Elsa.Bpmn.UnitTests/BpmnScopeDispatcherTests.cs new file mode 100644 index 000000000..b806f1594 --- /dev/null +++ b/test/unit/Elsa.Bpmn.UnitTests/BpmnScopeDispatcherTests.cs @@ -0,0 +1,67 @@ +using Elsa.Bpmn.Hosting; + +namespace Elsa.Bpmn.UnitTests; + +/// +/// The queue that keeps a scope evaluation raised while another one is applying its commands from running on top of it. +/// +public class BpmnScopeDispatcherTests +{ + private readonly BpmnScopeDispatcher _dispatcher = new(); + private readonly List _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(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; + } +} diff --git a/test/unit/Elsa.Bpmn.UnitTests/BpmnWorkTeardownTests.cs b/test/unit/Elsa.Bpmn.UnitTests/BpmnWorkTeardownTests.cs new file mode 100644 index 000000000..d2dfd360f --- /dev/null +++ b/test/unit/Elsa.Bpmn.UnitTests/BpmnWorkTeardownTests.cs @@ -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; + +/// +/// Covers the one refusal can make: a subtree that still has a scheduled-but-not-yet- +/// invoked activity in it. 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 +/// ContinueWithIncidentsStrategy — the caller applying the teardown must not leave the persisted ledger +/// claiming work it just tore down. +/// +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( + () => 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(() => 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(() => 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); + } + + /// + /// 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. + /// + 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()); + + 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); + } + + /// Records the subtree as the scope's live work, exactly as StartWork would have, and persists it. + 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); + } +}