diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnPersistenceTests.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnPersistenceTests.cs new file mode 100644 index 000000000..1b5ea6dd5 --- /dev/null +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnPersistenceTests.cs @@ -0,0 +1,128 @@ +using Bpmn.Model.State; +using Elsa.Workflows; +using Xunit.Abstractions; + +namespace Elsa.Bpmn.IntegrationTests.Scenarios.HostPort; + +/// +/// What survives a suspend and a resume: the pruned and the host-side work ledger, +/// both persisted as JSON strings in . +/// +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); + } +} diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestHost.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestHost.cs index 02131d3be..615be8ff8 100644 --- a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestHost.cs +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestHost.cs @@ -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(); } + /// + /// Replaces the current with what a round trip through Elsa's own + /// hands back — what a real persistence store would return on load, + /// rather than the exact same in-memory object the previous run produced. + /// + public void RoundTripStateThroughJson() + { + var state = _state ?? throw new InvalidOperationException("The workflow has not been run yet."); + var serializer = _services.GetRequiredService(); + + _state = serializer.Deserialize(serializer.Serialize(state)); + } + + /// + /// A BPMN scope's own memory, read straight off the current the way a resumed scope + /// itself sees it, rather than off any live 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 + /// tells them apart by call-stack depth -- the nested scope's context is always + /// deeper than the root's. + /// + 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(executionStateJson), + JsonSerializer.Deserialize(workLedgerJson) ?? new BpmnWorkLedger()); + } + private RunWorkflowResult Record(RunWorkflowResult result) { _result = result; diff --git a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestProcesses.cs b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestProcesses.cs index 2ea34b603..c0950c93f 100644 --- a/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestProcesses.cs +++ b/test/integration/Elsa.Bpmn.IntegrationTests/Scenarios/HostPort/BpmnTestProcesses.cs @@ -71,26 +71,8 @@ internal static class BpmnTestProcesses } /// 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)); - } + public static BpmnProcess ParallelSplitAndJoin(BpmnTestLog log) => + ParallelSplitAndJoinTopology("parallel-split-and-join", Immediate("left", log), Immediate("right", log), log); /// A task that fails, with an error boundary event that catches it. public static BpmnProcess ErrorBoundaryCaught(BpmnTestLog log) @@ -168,6 +150,56 @@ internal static class BpmnTestProcesses return Scope("scope", definition, Blocking("each", log), Immediate("after", log)); } + /// + /// 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. + /// + 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)); + } + + /// + /// 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. + /// + public static BpmnProcess ParallelSplitAndJoinBlocking(BpmnTestLog log) => + ParallelSplitAndJoinTopology("parallel-split-and-join-blocking", Blocking("left", log), Blocking("right", log), log); + + /// + /// The start/split/left/right/join/after/end graph shared by and + /// , parameterised by the work the two branches run. + /// + 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)); + } + /// /// A collection-mode multi-instance task: one instance per item of a container-scoped variable, which the /// interpreter reads back through IBpmnVariableReader while it evaluates. @@ -250,6 +282,42 @@ internal static class BpmnTestProcesses return Scope("scope", definition, Scope("sub", body, Immediate("subOnly", log)), Immediate("after", log)); } + /// + /// A parallel split and join, blocking on both branches, nested inside an embedded subprocess. Used to prove + /// a nested 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. + /// + 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)); + } + /// A linear process: one task between a start and an end event. public static BpmnProcess LinearTask(BpmnTestLog log) {