test(bpmn): prove execution state and the work ledger survive real persistence (#7947)
* test(bpmn): prove BpmnExecutionState pruning and ledger rehydration survive a real suspend/resume Prune() was already being called before every persisted write in BpmnScopeHost, but nothing proved it, and no BPMN test had ever exercised a genuine rehydration: every existing scenario used the in-memory WorkflowState object straight from the previous run. Add tests that round-trip a suspended scope's state through Elsa's own IWorkflowStateSerializer -- the boundary that mangled values before -- and assert the persisted BpmnExecutionState stays bounded across many evaluations and that a resumed scope with two live units of work matches each completion back to its binding through the rehydrated BpmnWorkLedger. Both tests were confirmed red by mutation-testing away Prune() and by returning an empty ledger from BpmnScopeMemory.Load. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * test(bpmn): prove a nested scope's ledger survives real persistence Both new tests in the prior commit suspended a root scope with live work outstanding, but neither crossed a nested scope's own ledger through Elsa's real IWorkflowStateSerializer -- the intersection of the two things that have actually broken here: the handle-to-context map, and the serializer boundary. Add a nested parallel split/join, blocking on both branches inside an embedded subprocess, round-tripped through the serializer between each branch's completion, and confirmed red by returning an empty ledger from BpmnScopeMemory.Load and green with it restored. Also extract the start/split/left/right/join/after/end topology shared verbatim by ParallelSplitAndJoin and ParallelSplitAndJoinBlocking into one private builder parameterised by the branches' work, keeping both public factories unchanged. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Sipke Schoorstra
GitHub
parent
6a3d65ff7e
commit
7eaf056d65
|
|
@ -0,0 +1,128 @@
|
|||
using Bpmn.Model.State;
|
||||
using Elsa.Workflows;
|
||||
using Xunit.Abstractions;
|
||||
|
||||
namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort;
|
||||
|
||||
/// <summary>
|
||||
/// What survives a suspend and a resume: the pruned <see cref="BpmnExecutionState"/> and the host-side work ledger,
|
||||
/// both persisted as JSON strings in <see cref="ActivityExecutionContext.Properties"/>.
|
||||
/// </summary>
|
||||
public class BpmnPersistenceTests(ITestOutputHelper testOutputHelper)
|
||||
{
|
||||
private readonly BpmnTestHost _host = new(testOutputHelper);
|
||||
|
||||
[Fact(DisplayName = "The persisted execution state stays bounded across many evaluations of one scope")]
|
||||
public async Task PersistedExecutionState_StaysBoundedAcrossManyEvaluations()
|
||||
{
|
||||
// A sequential multi-instance loop drives one evaluation per iteration, all on the same scope. Every
|
||||
// iteration but the last consumes a token that nothing afterward can ever reference again: an interpreter
|
||||
// state that is written without being pruned first re-serializes every one of them, so the token count
|
||||
// grows with the iteration count instead of staying flat.
|
||||
|
||||
const int cardinality = 12;
|
||||
|
||||
// Arrange & Act: run every iteration but the last, capturing the persisted token count while the scope is
|
||||
// still suspended and mid-loop after each one. The scope's own context stops being persisted the moment it
|
||||
// completes, so the last iteration -- which also completes "after" and the scope itself -- is run separately.
|
||||
await _host.RunAsync(BpmnTestProcesses.SequentialMultiInstanceTask(_host.Log, cardinality));
|
||||
|
||||
var maxTokenCount = 0;
|
||||
|
||||
for (var i = 0; i < cardinality - 1; i++)
|
||||
{
|
||||
await _host.FinishWorkAsync("each");
|
||||
|
||||
var (state, _) = _host.PersistedScopeMemory();
|
||||
Assert.NotNull(state);
|
||||
maxTokenCount = Math.Max(maxTokenCount, state.Tokens.Count);
|
||||
}
|
||||
|
||||
await _host.FinishWorkAsync("each");
|
||||
|
||||
// Assert: the loop actually ran to completion...
|
||||
Assert.Equal(cardinality, _host.Log.Occurrences("executed:each"));
|
||||
Assert.Equal(1, _host.Log.Occurrences("executed:after"));
|
||||
|
||||
// ...and the persisted state's token count stayed flat rather than growing with the iteration count. Without
|
||||
// pruning, a consumed token from every prior iteration would still be there by the time the loop is nearly
|
||||
// done, so the count would climb toward the iteration count instead of staying near-constant.
|
||||
Assert.True(
|
||||
maxTokenCount < cardinality,
|
||||
$"Expected the persisted token count to stay well below the iteration count ({cardinality}) because Prune() drops consumed tokens no active work references, but the highest observed count was {maxTokenCount}.");
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A scope resumed from a JSON round trip matches each completion back to its binding through the rehydrated ledger")]
|
||||
public async Task ResumedScope_WithTwoUnitsOfLiveWork_MatchesEachCompletionToItsBindingThroughTheRehydratedLedger()
|
||||
{
|
||||
// Arrange: suspend with two live units of work outstanding, both bound under the same scope.
|
||||
await _host.RunAsync(BpmnTestProcesses.ParallelSplitAndJoinBlocking(_host.Log));
|
||||
|
||||
var (_, ledgerBeforeResume) = _host.PersistedScopeMemory();
|
||||
Assert.Equal(2, ledgerBeforeResume.Records.Count);
|
||||
Assert.Contains(ledgerBeforeResume.Records, x => x.BindingRef == BpmnTestProcesses.BindingRef("left"));
|
||||
Assert.Contains(ledgerBeforeResume.Records, x => x.BindingRef == BpmnTestProcesses.BindingRef("right"));
|
||||
|
||||
// Act: round-trip the workflow state through Elsa's own serializer -- what a real persistence store would
|
||||
// do -- and resume each branch by name, one at a time, round-tripping again between them.
|
||||
_host.RoundTripStateThroughJson();
|
||||
|
||||
// Assert: the ledger the resumed scope would read from is intact before either branch is even resumed.
|
||||
var (_, ledgerAfterResume) = _host.PersistedScopeMemory();
|
||||
Assert.Equal(2, ledgerAfterResume.Records.Count);
|
||||
Assert.Contains(ledgerAfterResume.Records, x => x.BindingRef == BpmnTestProcesses.BindingRef("left"));
|
||||
Assert.Contains(ledgerAfterResume.Records, x => x.BindingRef == BpmnTestProcesses.BindingRef("right"));
|
||||
|
||||
await _host.FinishWorkAsync("left");
|
||||
_host.RoundTripStateThroughJson();
|
||||
var result = await _host.FinishWorkAsync("right");
|
||||
|
||||
// Assert: both branches ran, and the join fired exactly once. Neither is possible unless the resumed scope's
|
||||
// ledger still mapped each completing child context back to its own binding: a scope with a lost or shared
|
||||
// handle map either drops a completion outright (the join never fires and the workflow never finishes) or
|
||||
// cannot tell the two branches apart (the join fires more than once).
|
||||
Assert.Contains("resumed:left", _host.Log.Entries);
|
||||
Assert.Contains("resumed:right", _host.Log.Entries);
|
||||
Assert.Equal(1, _host.Log.Occurrences("executed:after"));
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A nested scope resumed from a JSON round trip matches each completion back to its binding through its own rehydrated ledger")]
|
||||
public async Task ResumedNestedScope_WithTwoUnitsOfLiveWork_MatchesEachCompletionToItsBindingThroughTheRehydratedLedger()
|
||||
{
|
||||
// Arrange: suspend with two live units of work outstanding inside the nested scope -- the embedded
|
||||
// subprocess -- while the root scope's own ledger holds only the subprocess itself.
|
||||
await _host.RunAsync(BpmnTestProcesses.NestedParallelSplitAndJoinBlocking(_host.Log));
|
||||
|
||||
var (_, nestedLedgerBeforeResume) = _host.PersistedScopeMemory(nested: true);
|
||||
Assert.Equal(2, nestedLedgerBeforeResume.Records.Count);
|
||||
Assert.Contains(nestedLedgerBeforeResume.Records, x => x.BindingRef == BpmnTestProcesses.BindingRef("subLeft"));
|
||||
Assert.Contains(nestedLedgerBeforeResume.Records, x => x.BindingRef == BpmnTestProcesses.BindingRef("subRight"));
|
||||
|
||||
// Act: round-trip the workflow state through Elsa's own serializer, then resume each branch by name, one at
|
||||
// a time, round-tripping again between them -- exactly as the root-scope test above does, but crossing into
|
||||
// the nested scope's own ledger instead of the root's.
|
||||
_host.RoundTripStateThroughJson();
|
||||
|
||||
// Assert: the nested scope's own ledger is intact before either branch is even resumed.
|
||||
var (_, nestedLedgerAfterResume) = _host.PersistedScopeMemory(nested: true);
|
||||
Assert.Equal(2, nestedLedgerAfterResume.Records.Count);
|
||||
Assert.Contains(nestedLedgerAfterResume.Records, x => x.BindingRef == BpmnTestProcesses.BindingRef("subLeft"));
|
||||
Assert.Contains(nestedLedgerAfterResume.Records, x => x.BindingRef == BpmnTestProcesses.BindingRef("subRight"));
|
||||
|
||||
await _host.FinishWorkAsync("subLeft");
|
||||
_host.RoundTripStateThroughJson();
|
||||
var result = await _host.FinishWorkAsync("subRight");
|
||||
|
||||
// Assert: both branches ran, the nested join fired exactly once, and the process ran all the way out to the
|
||||
// root scope's own "after" step and completion. None of this is possible unless the resumed nested scope's
|
||||
// ledger still mapped each completing child context back to its own binding: a nested scope with a lost or
|
||||
// shared handle map either drops a completion outright (the nested join never fires and the process never
|
||||
// finishes) or cannot tell the two branches apart (the nested join fires more than once).
|
||||
Assert.Contains("resumed:subLeft", _host.Log.Entries);
|
||||
Assert.Contains("resumed:subRight", _host.Log.Entries);
|
||||
Assert.Equal(1, _host.Log.Occurrences("executed:subAfter"));
|
||||
Assert.Equal(1, _host.Log.Occurrences("executed:after"));
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,3 +1,5 @@
|
|||
using System.Text.Json;
|
||||
using Bpmn.Model.State;
|
||||
using Elsa.Bpmn.Hosting;
|
||||
using Elsa.Bpmn.IntegrationTests.Scenarios.HostPort.Activities;
|
||||
using Elsa.Extensions;
|
||||
|
|
@ -83,6 +85,42 @@ public sealed class BpmnTestHost
|
|||
return ledger.Records.Select(record => (record.BindingRef, record.IterationId)).ToList();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Replaces the current <see cref="WorkflowState"/> with what a round trip through Elsa's own
|
||||
/// <see cref="IWorkflowStateSerializer"/> hands back — what a real persistence store would return on load,
|
||||
/// rather than the exact same in-memory object the previous run produced.
|
||||
/// </summary>
|
||||
public void RoundTripStateThroughJson()
|
||||
{
|
||||
var state = _state ?? throw new InvalidOperationException("The workflow has not been run yet.");
|
||||
var serializer = _services.GetRequiredService<IWorkflowStateSerializer>();
|
||||
|
||||
_state = serializer.Deserialize(serializer.Serialize(state));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A BPMN scope's own memory, read straight off the current <see cref="WorkflowState"/> the way a resumed scope
|
||||
/// itself sees it, rather than off any live <see cref="ActivityExecutionContext"/> the current process happens
|
||||
/// to still hold. Sound for any process with exactly one BPMN scope; a process with a nested scope has two, so
|
||||
/// <paramref name="nested"/> tells them apart by call-stack depth -- the nested scope's context is always
|
||||
/// deeper than the root's.
|
||||
/// </summary>
|
||||
internal (BpmnExecutionState? State, BpmnWorkLedger Work) PersistedScopeMemory(bool nested = false)
|
||||
{
|
||||
var state = _state ?? throw new InvalidOperationException("The workflow has not been run yet.");
|
||||
var scopeStates = state.ActivityExecutionContexts.Where(x => x.Properties.ContainsKey(BpmnScopeMemory.WorkLedgerPropertyKey));
|
||||
var scopeState = nested
|
||||
? scopeStates.OrderByDescending(x => x.CallStackDepth).First()
|
||||
: scopeStates.OrderBy(x => x.CallStackDepth).First();
|
||||
|
||||
var executionStateJson = scopeState.Properties.TryGetValue(BpmnScopeMemory.ExecutionStatePropertyKey, out var value) ? value as string : null;
|
||||
var workLedgerJson = (string)scopeState.Properties[BpmnScopeMemory.WorkLedgerPropertyKey];
|
||||
|
||||
return (
|
||||
executionStateJson is null ? null : JsonSerializer.Deserialize<BpmnExecutionState>(executionStateJson),
|
||||
JsonSerializer.Deserialize<BpmnWorkLedger>(workLedgerJson) ?? new BpmnWorkLedger());
|
||||
}
|
||||
|
||||
private RunWorkflowResult Record(RunWorkflowResult result)
|
||||
{
|
||||
_result = result;
|
||||
|
|
|
|||
|
|
@ -71,26 +71,8 @@ internal static class BpmnTestProcesses
|
|||
}
|
||||
|
||||
/// <summary>A parallel gateway split and join.</summary>
|
||||
public static BpmnProcess ParallelSplitAndJoin(BpmnTestLog log)
|
||||
{
|
||||
var definition = new BpmnProcessBuilder("parallel-split-and-join")
|
||||
.StartEvent("start")
|
||||
.ParallelGateway("split")
|
||||
.Task("left", bindingRef: BindingRef("left"))
|
||||
.Task("right", bindingRef: BindingRef("right"))
|
||||
.ParallelGateway("join")
|
||||
.Task("after", bindingRef: BindingRef("after"))
|
||||
.EndEvent("end")
|
||||
.ConnectSequence("start", "split")
|
||||
.Connect("split", "left")
|
||||
.Connect("split", "right")
|
||||
.Connect("left", "join")
|
||||
.Connect("right", "join")
|
||||
.ConnectSequence("join", "after", "end")
|
||||
.Build();
|
||||
|
||||
return Scope("scope", definition, Immediate("left", log), Immediate("right", log), Immediate("after", log));
|
||||
}
|
||||
public static BpmnProcess ParallelSplitAndJoin(BpmnTestLog log) =>
|
||||
ParallelSplitAndJoinTopology("parallel-split-and-join", Immediate("left", log), Immediate("right", log), log);
|
||||
|
||||
/// <summary>A task that fails, with an error boundary event that catches it.</summary>
|
||||
public static BpmnProcess ErrorBoundaryCaught(BpmnTestLog log)
|
||||
|
|
@ -168,6 +150,56 @@ internal static class BpmnTestProcesses
|
|||
return Scope("scope", definition, Blocking("each", log), Immediate("after", log));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A sequential multi-instance task: one instance at a time, each blocking until a test finishes it. Used to
|
||||
/// drive many evaluations of one scope so a persisted blob that is not pruned is observable as unbounded growth.
|
||||
/// </summary>
|
||||
public static BpmnProcess SequentialMultiInstanceTask(BpmnTestLog log, int cardinality)
|
||||
{
|
||||
var definition = new BpmnProcessBuilder("sequential-multi-instance-task")
|
||||
.StartEvent("start")
|
||||
.Task(BpmnElementTypes.Task, "each", bindingRef: BindingRef("each"), loopCharacteristics: new BpmnLoopCharacteristics(isSequential: true, cardinality: cardinality))
|
||||
.Task("after", bindingRef: BindingRef("after"))
|
||||
.EndEvent("end")
|
||||
.ConnectSequence("start", "each", "after", "end")
|
||||
.Build();
|
||||
|
||||
return Scope("scope", definition, Blocking("each", log), Immediate("after", log));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A parallel split into two branches that both block, and a join. Used to prove a scope suspends with two live
|
||||
/// units of work outstanding and, once resumed, matches each completion back to its own binding through the
|
||||
/// rehydrated ledger.
|
||||
/// </summary>
|
||||
public static BpmnProcess ParallelSplitAndJoinBlocking(BpmnTestLog log) =>
|
||||
ParallelSplitAndJoinTopology("parallel-split-and-join-blocking", Blocking("left", log), Blocking("right", log), log);
|
||||
|
||||
/// <summary>
|
||||
/// The start/split/left/right/join/after/end graph shared by <see cref="ParallelSplitAndJoin"/> and
|
||||
/// <see cref="ParallelSplitAndJoinBlocking"/>, parameterised by the work the two branches run.
|
||||
/// </summary>
|
||||
private static BpmnProcess ParallelSplitAndJoinTopology(string processId, IActivity leftWork, IActivity rightWork, BpmnTestLog log)
|
||||
{
|
||||
var definition = new BpmnProcessBuilder(processId)
|
||||
.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, leftWork, rightWork, Immediate("after", log));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A collection-mode multi-instance task: one instance per item of a container-scoped variable, which the
|
||||
/// interpreter reads back through <c>IBpmnVariableReader</c> while it evaluates.
|
||||
|
|
@ -250,6 +282,42 @@ internal static class BpmnTestProcesses
|
|||
return Scope("scope", definition, Scope("sub", body, Immediate("subOnly", log)), Immediate("after", log));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A parallel split and join, blocking on both branches, nested inside an embedded subprocess. Used to prove
|
||||
/// a <em>nested</em> scope's own ledger -- not just a root scope's -- matches each completion back to its
|
||||
/// binding after a round trip through Elsa's own serializer.
|
||||
/// </summary>
|
||||
public static BpmnProcess NestedParallelSplitAndJoinBlocking(BpmnTestLog log)
|
||||
{
|
||||
var body = new BpmnProcessBuilder("nested-parallel-split-and-join-body")
|
||||
.StartEvent("subStart")
|
||||
.ParallelGateway("subSplit")
|
||||
.Task("subLeft", bindingRef: BindingRef("subLeft"))
|
||||
.Task("subRight", bindingRef: BindingRef("subRight"))
|
||||
.ParallelGateway("subJoin")
|
||||
.Task("subAfter", bindingRef: BindingRef("subAfter"))
|
||||
.EndEvent("subEnd")
|
||||
.ConnectSequence("subStart", "subSplit")
|
||||
.Connect("subSplit", "subLeft")
|
||||
.Connect("subSplit", "subRight")
|
||||
.Connect("subLeft", "subJoin")
|
||||
.Connect("subRight", "subJoin")
|
||||
.ConnectSequence("subJoin", "subAfter", "subEnd")
|
||||
.Build();
|
||||
|
||||
var definition = new BpmnProcessBuilder("nested-parallel-split-and-join-blocking")
|
||||
.StartEvent("start")
|
||||
.SubProcess("sub", bindingRef: BindingRef("sub"))
|
||||
.Task("after", bindingRef: BindingRef("after"))
|
||||
.EndEvent("end")
|
||||
.ConnectSequence("start", "sub", "after", "end")
|
||||
.Build();
|
||||
|
||||
var nested = Scope("sub", body, Blocking("subLeft", log), Blocking("subRight", log), Immediate("subAfter", log));
|
||||
|
||||
return Scope("scope", definition, nested, Immediate("after", log));
|
||||
}
|
||||
|
||||
/// <summary>A linear process: one task between a start and an end event.</summary>
|
||||
public static BpmnProcess LinearTask(BpmnTestLog log)
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in a new issue