feat(core): let a container withdraw work it scheduled but must not run (#7967)
A container that schedules a child and then decides the child must not run had no way to withdraw it. `IActivityScheduler` exposed no removal operation, and `CancelActivityAsync` no-opped on a context whose status was `Pending`, so a container could tear a branch down and still have an activity from that branch execute afterwards, side effects and all. Fixes #7943. - `IActivityScheduler.RemoveWhere` removes work items and keeps the order the survivors would have been taken in; implemented in both the FIFO and LIFO schedulers. - `CancelActivityAsync` (both the public extension and the internal one used when a container completes) cancels `Pending` contexts as well as running ones, and withdraws the work item that would have started the cancelled activity plus the items it had scheduled for children with no context yet. Withdrawal is a real removal rather than a terminal status honoured at dequeue time, because the scheduler is also read: `Flowchart.HasPendingWork` inspects it to decide whether it may complete, and the work item list is extracted into the persisted workflow state — a withdrawn-but-queued item would be persisted and rehydrated with a fresh context after a suspend/resume. `StateMachine` had hand-rolled the same operation to drop competing triggers by clearing the scheduler and re-scheduling everything else; it now calls `RemoveWhere`. `Elsa.Bpmn` no longer needs to refuse a teardown whose subtree still has queued work, so `BpmnWorkTeardown` drops the `NotSupportedException` and records the teardown reason on the torn-down activity's journal instead. BREAKING: `IActivityScheduler` gains a member; external implementations must add `RemoveWhere`. Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Sipke Schoorstra
GitHub
parent
0a86a4803e
commit
74fc891350
|
|
@ -84,6 +84,13 @@ Core scheduling is about which activity work item runs next. Key services:
|
|||
- [WorkflowExecutionContextSchedulerStrategy](../../src/modules/Elsa.Workflows.Core/Services/WorkflowExecutionContextSchedulerStrategy.cs)
|
||||
- [ActivityExecutionContextSchedulerStrategy](../../src/modules/Elsa.Workflows.Core/Services/ActivityExecutionContextSchedulerStrategy.cs)
|
||||
|
||||
Scheduling is reversible. A container that schedules a child and then decides the child must not run withdraws it by
|
||||
cancelling: `CancelActivityAsync` cancels contexts that are still `Pending` as well as running ones, and removes from
|
||||
the scheduler both the work item that would have started the cancelled activity and the work items it had scheduled for
|
||||
children that have no execution context yet. Withdrawal is a real removal (`IActivityScheduler.RemoveWhere`) rather than
|
||||
a flag honoured at dequeue time, because the scheduler is also read: `Flowchart` asks it whether the flowchart still has
|
||||
pending work, and the work item list is part of the persisted workflow state.
|
||||
|
||||
Runtime scheduling and external dispatch are separate and live in `Elsa.Workflows.Runtime`.
|
||||
|
||||
## Bookmarks And Triggers
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ namespace Elsa.Bpmn.Hosting;
|
|||
/// <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>CancelWorkSubtree</c></term><description><c>CancelActivityAsync</c>, which already walks the child subtree recursively and withdraws its not-yet-invoked work.</description></item>
|
||||
/// <item><term><c>SignalEnclosingScope</c></term><description><c>SendSignalAsync</c>, which bubbles to ancestors.</description></item>
|
||||
/// </list>
|
||||
/// </remarks>
|
||||
|
|
@ -135,12 +135,12 @@ internal sealed class BpmnCommandApplier(ActivityExecutionContext scopeContext,
|
|||
|
||||
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
|
||||
// Saved here rather than left to the end-of-command save in ApplyAsync: cancelling the subtree runs arbitrary
|
||||
// activity code (CancelSignal handlers, cancellation notifications), any of which can throw, and under the
|
||||
// continue-with-incidents strategy such a 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 teardown, keeps
|
||||
// the persisted ledger from claiming work this scope just tore down, and is what lets a completion callback
|
||||
// that arrives for that work anyway find no live record and be discarded (see
|
||||
// BpmnScopeHost.OnWorkCompletedAsync) instead of being handed to the interpreter as real work.
|
||||
memory.SaveWork();
|
||||
|
||||
|
|
|
|||
|
|
@ -8,8 +8,7 @@ namespace Elsa.Bpmn.Hosting;
|
|||
/// </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.
|
||||
/// terminalizing the unit of work whose fault it just claimed. The mechanism is the same in either case.
|
||||
/// </remarks>
|
||||
internal static class BpmnWorkTeardown
|
||||
{
|
||||
|
|
@ -18,41 +17,18 @@ internal static class BpmnWorkTeardown
|
|||
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.
|
||||
/// Tears down a unit of work and everything underneath it.
|
||||
/// </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>
|
||||
/// Cancellation itself needs no BPMN-specific code: the public <c>CancelActivityAsync</c> extension walks the
|
||||
/// child subtree recursively, which is exactly what "and everything it in turn started" means, and it withdraws
|
||||
/// the scheduled-but-not-yet-invoked work items in that subtree so nothing from the destroyed branch runs
|
||||
/// afterwards. The only thing this method adds is the reason, recorded on the torn-down activity's journal
|
||||
/// before it goes, so the workflow's execution log says which BPMN element destroyed the branch and why.
|
||||
/// </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.");
|
||||
}
|
||||
|
||||
childContext.AddExecutionLogEntry("Torn down by BPMN", reason);
|
||||
await childContext.CancelActivityAsync();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -230,15 +230,7 @@ public class StateMachine : Activity
|
|||
private static void RemoveScheduledCompetingTriggers(ActivityExecutionContext context, HashSet<string> competingTriggerIds)
|
||||
{
|
||||
var scheduler = context.WorkflowExecutionContext.Scheduler;
|
||||
var scheduledWorkItems = scheduler.List().ToList();
|
||||
|
||||
if (!scheduledWorkItems.Any(x => IsCompetingTriggerWorkItem(context, competingTriggerIds, x)))
|
||||
return;
|
||||
|
||||
scheduler.Clear();
|
||||
|
||||
foreach (var workItem in scheduledWorkItems.Where(x => !IsCompetingTriggerWorkItem(context, competingTriggerIds, x)))
|
||||
scheduler.Schedule(workItem);
|
||||
scheduler.RemoveWhere(x => IsCompetingTriggerWorkItem(context, competingTriggerIds, x));
|
||||
}
|
||||
|
||||
private static bool IsCompetingTriggerWorkItem(ActivityExecutionContext context, HashSet<string> competingTriggerIds, ActivityWorkItem workItem) =>
|
||||
|
|
|
|||
|
|
@ -14,11 +14,28 @@ public partial class ActivityExecutionContext
|
|||
return Status is not ActivityStatus.Canceled and not ActivityStatus.Completed;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Withdraws work that the engine has not invoked yet and that this activity's cancellation makes obsolete: the
|
||||
/// work item that would have started this activity, and the work items this activity scheduled for children whose
|
||||
/// execution contexts do not exist yet. Without this, a container could tear a branch down and still have an
|
||||
/// activity from that branch execute afterwards, side effects and all.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The work items are removed from the scheduler rather than left to be dequeued and discarded, because the
|
||||
/// scheduler is not write-only: containers such as <c>Flowchart</c> ask it whether they still have pending work,
|
||||
/// and a withdrawn item left in the queue would keep answering yes.
|
||||
/// </remarks>
|
||||
internal void WithdrawScheduledWork()
|
||||
{
|
||||
WorkflowExecutionContext.Scheduler.RemoveWhere(workItem => workItem.ExistingActivityExecutionContext == this || workItem.Owner == this);
|
||||
}
|
||||
|
||||
private async Task CancelActivityAsync()
|
||||
{
|
||||
if(!CanCancelActivity())
|
||||
return;
|
||||
|
||||
|
||||
WithdrawScheduledWork();
|
||||
TransitionTo(ActivityStatus.Canceled);
|
||||
ClearBookmarks();
|
||||
ClearCompletionCallbacks();
|
||||
|
|
|
|||
|
|
@ -41,6 +41,14 @@ public interface IActivityScheduler
|
|||
/// <param name="predicate">The predicate to match.</param>
|
||||
ActivityWorkItem? Find(Func<ActivityWorkItem, bool> predicate);
|
||||
|
||||
/// <summary>
|
||||
/// Removes every work item matching the specified predicate, preserving the order in which the remaining items
|
||||
/// will be taken. This is how a container withdraws work it scheduled but has since decided must not run.
|
||||
/// </summary>
|
||||
/// <param name="predicate">The predicate to match.</param>
|
||||
/// <returns>The number of work items removed.</returns>
|
||||
int RemoveWhere(Func<ActivityWorkItem, bool> predicate);
|
||||
|
||||
/// <summary>
|
||||
/// Clears all work items from the scheduler.
|
||||
/// </summary>
|
||||
|
|
|
|||
|
|
@ -298,12 +298,21 @@ public static partial class ActivityExecutionContextExtensions
|
|||
/// <summary>
|
||||
/// Cancel the activity. For blocking activities, it means their bookmarks will be removed. For job activities, the background work will be cancelled.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// An activity that is still <see cref="ActivityStatus.Pending"/> is cancelled too, rather than skipped: it has
|
||||
/// been scheduled and would otherwise be invoked after the container that scheduled it decided it must not run.
|
||||
/// Cancelling it also withdraws the work item that would have started it.
|
||||
/// </remarks>
|
||||
public async Task CancelActivityAsync()
|
||||
{
|
||||
// If the activity is not running, do nothing.
|
||||
if (activityExecutionContext.Status != ActivityStatus.Running && activityExecutionContext.Status != ActivityStatus.Faulted)
|
||||
// If the activity has already reached a terminal state, do nothing.
|
||||
if (activityExecutionContext.Status is ActivityStatus.Completed or ActivityStatus.Canceled)
|
||||
return;
|
||||
|
||||
// Withdraw work that has been scheduled but not yet invoked, so that cancelling a branch cannot leave an
|
||||
// activity from it running afterwards.
|
||||
activityExecutionContext.WithdrawScheduledWork();
|
||||
|
||||
// Select all child contexts.
|
||||
var childContexts = activityExecutionContext.Children.ToList();
|
||||
|
||||
|
|
|
|||
|
|
@ -29,6 +29,24 @@ public class QueueBasedActivityScheduler : IActivityScheduler
|
|||
/// <inheritdoc />
|
||||
public ActivityWorkItem? Find(Func<ActivityWorkItem, bool> predicate) => _queue.FirstOrDefault(predicate);
|
||||
|
||||
/// <inheritdoc />
|
||||
public int RemoveWhere(Func<ActivityWorkItem, bool> predicate)
|
||||
{
|
||||
// The queue enumerates front-first, so re-enqueueing what survives restores the original order.
|
||||
var remaining = _queue.Where(x => !predicate(x)).ToList();
|
||||
var removedCount = _queue.Count - remaining.Count;
|
||||
|
||||
if (removedCount == 0)
|
||||
return 0;
|
||||
|
||||
_queue.Clear();
|
||||
|
||||
foreach (var workItem in remaining)
|
||||
_queue.Enqueue(workItem);
|
||||
|
||||
return removedCount;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Clear() => _queue.Clear();
|
||||
}
|
||||
|
|
@ -29,6 +29,24 @@ public class StackBasedActivityScheduler : IActivityScheduler
|
|||
/// <inheritdoc />
|
||||
public ActivityWorkItem? Find(Func<ActivityWorkItem, bool> predicate) => _stack.FirstOrDefault(predicate);
|
||||
|
||||
/// <inheritdoc />
|
||||
public int RemoveWhere(Func<ActivityWorkItem, bool> predicate)
|
||||
{
|
||||
// The stack enumerates top-first, so what survives has to be pushed back in reverse to keep the same top.
|
||||
var remaining = _stack.Where(x => !predicate(x)).ToList();
|
||||
var removedCount = _stack.Count - remaining.Count;
|
||||
|
||||
if (removedCount == 0)
|
||||
return 0;
|
||||
|
||||
_stack.Clear();
|
||||
|
||||
for (var i = remaining.Count - 1; i >= 0; i--)
|
||||
_stack.Push(remaining[i]);
|
||||
|
||||
return removedCount;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Clear() => _stack.Clear();
|
||||
}
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
using Elsa.Extensions;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.WithdrawingScheduledWork.Activities;
|
||||
|
||||
/// <summary>
|
||||
/// A container that schedules a child the ordinary way — by activity, so the child has no execution context yet — and
|
||||
/// then cancels itself. There is no child context to cancel recursively, so withdrawing the queued work item is the
|
||||
/// only thing that can stop the child from running under a container that no longer exists.
|
||||
/// </summary>
|
||||
public class SelfCancellingScheduler : Activity
|
||||
{
|
||||
public IActivity Child { get; set; } = null!;
|
||||
|
||||
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
|
||||
{
|
||||
await context.ScheduleActivityAsync(Child);
|
||||
await context.CancelActivityAsync();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,42 @@
|
|||
using Elsa.Extensions;
|
||||
using Elsa.Workflows.Options;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.WithdrawingScheduledWork.Activities;
|
||||
|
||||
/// <summary>
|
||||
/// A container that schedules two children and then decides one of them must not run after all, before the engine has
|
||||
/// taken either work item. It holds a handle on the speculative child by creating that child's execution context up
|
||||
/// front and scheduling against it, which is how a container gets something to cancel later (Elsa.Bpmn does exactly
|
||||
/// this when it starts a unit of work).
|
||||
/// </summary>
|
||||
public class SpeculativeScheduler : Activity
|
||||
{
|
||||
/// <summary>The child that gets scheduled and then withdrawn.</summary>
|
||||
public IActivity Speculative { get; set; } = null!;
|
||||
|
||||
/// <summary>The child that gets scheduled and left alone, so the run proves it got as far as executing children.</summary>
|
||||
public IActivity Committed { get; set; } = null!;
|
||||
|
||||
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
|
||||
{
|
||||
var workflowExecutionContext = context.WorkflowExecutionContext;
|
||||
var speculativeContext = await workflowExecutionContext.CreateActivityExecutionContextAsync(Speculative, new ActivityInvocationOptions { Owner = context });
|
||||
workflowExecutionContext.AddActivityExecutionContext(speculativeContext);
|
||||
|
||||
await context.ScheduleActivityAsync(Speculative, new ScheduleWorkOptions
|
||||
{
|
||||
ExistingActivityExecutionContext = speculativeContext,
|
||||
CompletionCallback = OnChildCompletedAsync
|
||||
});
|
||||
await context.ScheduleActivityAsync(Committed, new ScheduleWorkOptions
|
||||
{
|
||||
CompletionCallback = OnChildCompletedAsync
|
||||
});
|
||||
|
||||
// The change of mind. Both work items are queued and neither has been invoked; the scheduler is FIFO, so the
|
||||
// speculative one is the very next thing the engine would take.
|
||||
await speculativeContext.CancelActivityAsync();
|
||||
}
|
||||
|
||||
private async ValueTask OnChildCompletedAsync(ActivityCompletedContext context) => await context.TargetContext.CompleteActivityAsync();
|
||||
}
|
||||
|
|
@ -0,0 +1,48 @@
|
|||
using Elsa.Testing.Shared;
|
||||
using Elsa.Workflows.IntegrationTests.Scenarios.WithdrawingScheduledWork.Activities;
|
||||
using Elsa.Workflows.IntegrationTests.Scenarios.WithdrawingScheduledWork.Workflows;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Xunit.Abstractions;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.WithdrawingScheduledWork;
|
||||
|
||||
/// <summary>
|
||||
/// A container that schedules a child and then decides the child must not run has to be able to withdraw it. Without
|
||||
/// that, the container tears a branch down and an activity from that branch executes anyway, side effects and all.
|
||||
/// </summary>
|
||||
public class WithdrawingScheduledWorkTests
|
||||
{
|
||||
private readonly CapturingTextWriter _capturingTextWriter = new();
|
||||
private readonly IServiceProvider _services;
|
||||
private readonly IWorkflowRunner _workflowRunner;
|
||||
|
||||
public WithdrawingScheduledWorkTests(ITestOutputHelper testOutputHelper)
|
||||
{
|
||||
_services = new TestApplicationBuilder(testOutputHelper)
|
||||
.WithCapturingTextWriter(_capturingTextWriter)
|
||||
.AddActivitiesFrom<SpeculativeScheduler>()
|
||||
.Build();
|
||||
|
||||
_workflowRunner = _services.GetRequiredService<IWorkflowRunner>();
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A child cancelled before it was invoked does not execute, and the container's other work still does.")]
|
||||
public async Task CancellingAScheduledChildBeforeInvocationWithdrawsIt()
|
||||
{
|
||||
await _services.PopulateRegistriesAsync();
|
||||
await _workflowRunner.RunAsync<SpeculativeSchedulingWorkflow>();
|
||||
|
||||
Assert.Equal(["Start", "Committed", "End"], _capturingTextWriter.Lines.ToList());
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Work scheduled by a container that then cancels itself does not execute.")]
|
||||
public async Task CancellingAContainerWithdrawsTheWorkItScheduled()
|
||||
{
|
||||
await _services.PopulateRegistriesAsync();
|
||||
await _workflowRunner.RunAsync<SelfCancellingSchedulingWorkflow>();
|
||||
|
||||
// "End" is absent because cancelling the container clears the completion callback the Sequence was waiting
|
||||
// on: the point of the assertion is that "Withdrawn" never ran.
|
||||
Assert.Equal(["Start"], _capturingTextWriter.Lines.ToList());
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.IntegrationTests.Scenarios.WithdrawingScheduledWork.Activities;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.WithdrawingScheduledWork.Workflows;
|
||||
|
||||
public class SelfCancellingSchedulingWorkflow : WorkflowBase
|
||||
{
|
||||
protected override void Build(IWorkflowBuilder builder)
|
||||
{
|
||||
builder.Root = new Sequence
|
||||
{
|
||||
Activities =
|
||||
{
|
||||
new WriteLine("Start"),
|
||||
new SelfCancellingScheduler
|
||||
{
|
||||
Child = new WriteLine("Withdrawn")
|
||||
},
|
||||
new WriteLine("End")
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,24 @@
|
|||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.IntegrationTests.Scenarios.WithdrawingScheduledWork.Activities;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.WithdrawingScheduledWork.Workflows;
|
||||
|
||||
public class SpeculativeSchedulingWorkflow : WorkflowBase
|
||||
{
|
||||
protected override void Build(IWorkflowBuilder builder)
|
||||
{
|
||||
builder.Root = new Sequence
|
||||
{
|
||||
Activities =
|
||||
{
|
||||
new WriteLine("Start"),
|
||||
new SpeculativeScheduler
|
||||
{
|
||||
Speculative = new WriteLine("Withdrawn"),
|
||||
Committed = new WriteLine("Committed")
|
||||
},
|
||||
new WriteLine("End")
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -11,12 +11,10 @@ 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.
|
||||
/// Covers the case <see cref="BpmnWorkTeardown"/> used to refuse: a subtree that still has a scheduled-but-not-yet-
|
||||
/// invoked activity in it. Cancelling now withdraws that work item, so the activity BPMN destroyed the branch of does
|
||||
/// not run afterwards. What remains this host's own contribution is the ledger removal, persisted before the teardown
|
||||
/// so that a completion arriving for torn-down work is discarded rather than fed to the interpreter as real work.
|
||||
/// </summary>
|
||||
public class BpmnWorkTeardownTests
|
||||
{
|
||||
|
|
@ -24,26 +22,43 @@ public class BpmnWorkTeardownTests
|
|||
private const string QueuedActivityId = "queued-activity";
|
||||
|
||||
[Fact]
|
||||
public async Task CancelSubtreeAsync_ThrowsNamingElementAndStrandedActivity_WhenSubtreeHasWorkStillQueued()
|
||||
public async Task CancelSubtreeAsync_WithdrawsTheQueuedWork_WhenSubtreeHasWorkStillQueued()
|
||||
{
|
||||
var (_, subtreeContext, _) = await BuildSubtreeWithQueuedWorkAsync();
|
||||
var workflowExecutionContext = subtreeContext.WorkflowExecutionContext;
|
||||
var queuedContext = workflowExecutionContext.ActivityExecutionContexts.Single(x => x.Activity.Id == QueuedActivityId);
|
||||
|
||||
var exception = await Assert.ThrowsAsync<NotSupportedException>(
|
||||
() => BpmnWorkTeardown.CancelSubtreeAsync(subtreeContext, "boundary interrupted").AsTask());
|
||||
await BpmnWorkTeardown.CancelSubtreeAsync(subtreeContext, "boundary interrupted");
|
||||
|
||||
Assert.Contains(SubtreeActivityId, exception.Message);
|
||||
Assert.Contains(QueuedActivityId, exception.Message);
|
||||
// Nothing is left for the engine to take, so the descendant never gets invoked...
|
||||
Assert.False(workflowExecutionContext.Scheduler.HasAny);
|
||||
|
||||
// ...and it is terminal rather than left Pending, so nothing can schedule it again either.
|
||||
Assert.Equal(ActivityStatus.Canceled, queuedContext.Status);
|
||||
Assert.Equal(ActivityStatus.Canceled, subtreeContext.Status);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ApplyAsync_RemovesLedgerRecordBeforeTheRefusalPropagates_WhenSubtreeHasWorkStillQueued()
|
||||
public async Task CancelSubtreeAsync_LeavesUnrelatedWorkScheduled_WhenSubtreeHasWorkStillQueued()
|
||||
{
|
||||
var (scopeContext, subtreeContext, _) = await BuildSubtreeWithQueuedWorkAsync();
|
||||
var workflowExecutionContext = subtreeContext.WorkflowExecutionContext;
|
||||
var siblingWorkItem = new ActivityWorkItem(new WriteLine("sibling") { Id = "sibling-activity" }, scopeContext);
|
||||
workflowExecutionContext.Scheduler.Schedule(siblingWorkItem);
|
||||
|
||||
await BpmnWorkTeardown.CancelSubtreeAsync(subtreeContext, "boundary interrupted");
|
||||
|
||||
Assert.Equal([siblingWorkItem], workflowExecutionContext.Scheduler.List());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ApplyAsync_RemovesLedgerRecord_WhenTearingDownWorkWithSomethingStillQueued()
|
||||
{
|
||||
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());
|
||||
await applier.ApplyAsync([new BpmnHostCommand.CancelWorkSubtree("work-1", SubtreeActivityId, "boundary interrupted")]);
|
||||
|
||||
// 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.
|
||||
|
|
@ -52,14 +67,13 @@ public class BpmnWorkTeardownTests
|
|||
}
|
||||
|
||||
[Fact]
|
||||
public async Task OnWorkCompletedAsync_DiscardsTheCallback_ForContextTornDownByARefusedCancellation()
|
||||
public async Task OnWorkCompletedAsync_DiscardsTheCallback_ForContextThatWasTornDown()
|
||||
{
|
||||
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());
|
||||
await applier.ApplyAsync([new BpmnHostCommand.CancelWorkSubtree("work-1", SubtreeActivityId, "boundary interrupted")]);
|
||||
|
||||
// `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
|
||||
|
|
|
|||
|
|
@ -0,0 +1,75 @@
|
|||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.Models;
|
||||
|
||||
namespace Elsa.Workflows.Core.UnitTests.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Covers <see cref="IActivityScheduler.RemoveWhere"/>, the operation a container uses to withdraw work it scheduled
|
||||
/// but has since decided must not run. What survives has to come back in the order it would have been taken in, which
|
||||
/// is the part a naive "clear and re-add" gets wrong for the LIFO scheduler.
|
||||
/// </summary>
|
||||
public class ActivitySchedulerTests
|
||||
{
|
||||
public static TheoryData<Func<IActivityScheduler>> Schedulers => new()
|
||||
{
|
||||
() => new QueueBasedActivityScheduler(),
|
||||
() => new StackBasedActivityScheduler()
|
||||
};
|
||||
|
||||
[Theory]
|
||||
[MemberData(nameof(Schedulers))]
|
||||
public void RemoveWhere_RemovesOnlyMatchingItems_AndReportsHowMany(Func<IActivityScheduler> createScheduler)
|
||||
{
|
||||
var scheduler = Schedule(createScheduler(), "a", "b", "c", "d");
|
||||
|
||||
var removedCount = scheduler.RemoveWhere(x => x.Activity.Id is "b" or "d");
|
||||
|
||||
Assert.Equal(2, removedCount);
|
||||
|
||||
// The survivors come out in the same order as a scheduler that only ever held them.
|
||||
Assert.Equal(TakeAll(Schedule(createScheduler(), "a", "c")), TakeAll(scheduler));
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[MemberData(nameof(Schedulers))]
|
||||
public void RemoveWhere_LeavesTheSchedulerUntouched_WhenNothingMatches(Func<IActivityScheduler> createScheduler)
|
||||
{
|
||||
var scheduler = Schedule(createScheduler(), "a", "b", "c");
|
||||
|
||||
var removedCount = scheduler.RemoveWhere(_ => false);
|
||||
|
||||
Assert.Equal(0, removedCount);
|
||||
Assert.Equal(TakeAll(Schedule(createScheduler(), "a", "b", "c")), TakeAll(scheduler));
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[MemberData(nameof(Schedulers))]
|
||||
public void RemoveWhere_EmptiesTheScheduler_WhenEverythingMatches(Func<IActivityScheduler> createScheduler)
|
||||
{
|
||||
var scheduler = Schedule(createScheduler(), "a", "b", "c");
|
||||
|
||||
var removedCount = scheduler.RemoveWhere(_ => true);
|
||||
|
||||
Assert.Equal(3, removedCount);
|
||||
Assert.False(scheduler.HasAny);
|
||||
}
|
||||
|
||||
private static IActivityScheduler Schedule(IActivityScheduler scheduler, params string[] activityIds)
|
||||
{
|
||||
foreach (var activityId in activityIds)
|
||||
scheduler.Schedule(new ActivityWorkItem(new WriteLine(activityId) { Id = activityId }));
|
||||
|
||||
return scheduler;
|
||||
}
|
||||
|
||||
/// <summary>Drains the scheduler, so assertions are about the order the work would actually have been taken in.</summary>
|
||||
private static List<string> TakeAll(IActivityScheduler scheduler)
|
||||
{
|
||||
var activityIds = new List<string>();
|
||||
|
||||
while (scheduler.HasAny)
|
||||
activityIds.Add(scheduler.Take().Activity.Id);
|
||||
|
||||
return activityIds;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue