Fix error handling.

SebastianStehle · SebastianStehle · commit 6d4a2831ea58 · 2025-07-02T20:35:43.000+02:00
diff --git a/Directory.Build.props b/Directory.Build.props
@@ -11,7 +11,7 @@
     <PackageProjectUrl>https://github.com/squidex/squidex</PackageProjectUrl>
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
     <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-    <Version>7.24.0</Version>
+    <Version>7.25.0</Version>
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(GITHUB_ACTIONS)' == 'true'">
diff --git a/flows/Squidex.Flows.Tests/DefaultFlowExecutorTests.Should_abort_flow_when_preparation_fails.verified.txt b/flows/Squidex.Flows.Tests/DefaultFlowExecutorTests.Should_abort_flow_when_preparation_fails.verified.txt
@@ -0,0 +1,35 @@
+﻿{
+  InstanceId: Guid_1,
+  OwnerId: Guid_2,
+  DefinitionId: Guid_3,
+  Definition: {
+    InitialStepId: StepId1,
+    Steps: {
+      StepId1: {
+        IgnoreError: false,
+        Step: {}
+      }
+    }
+  },
+  Context: {},
+  Description: ,
+  ScheduleKey: ,
+  SchedulePartition: 75,
+  Steps: {
+    StepId1: {
+      Status: Failed,
+      IsPrepared: false,
+      Attempts: [
+        {
+          Started: DateTimeOffset_1,
+          Completed: DateTimeOffset_1,
+          Error: Step Error
+        }
+      ]
+    }
+  },
+  Created: DateTimeOffset_1,
+  Completed: DateTimeOffset_1,
+  Expires: DateTimeOffset_2,
+  Status: Failed
+}
diff --git a/flows/Squidex.Flows.Tests/DefaultFlowExecutorTests.Should_execute_to_end_when_simulating.verified.txt b/flows/Squidex.Flows.Tests/DefaultFlowExecutorTests.Should_execute_to_end_when_simulating.verified.txt
@@ -0,0 +1,55 @@
+﻿{
+  InstanceId: Guid_1,
+  OwnerId: Guid_2,
+  DefinitionId: Guid_3,
+  Definition: {
+    InitialStepId: StepId1,
+    Steps: {
+      StepId1: {
+        IgnoreError: false,
+        Step: {}
+      }
+    }
+  },
+  Context: {},
+  Description: ,
+  ScheduleKey: ,
+  SchedulePartition: 75,
+  Steps: {
+    StepId1: {
+      Status: Failed,
+      IsPrepared: true,
+      Attempts: [
+        {
+          Started: DateTimeOffset_1,
+          Completed: DateTimeOffset_1,
+          Error: Step Error
+        },
+        {
+          Started: DateTimeOffset_1,
+          Completed: DateTimeOffset_1,
+          Error: Step Error
+        },
+        {
+          Started: DateTimeOffset_1,
+          Completed: DateTimeOffset_1,
+          Error: Step Error
+        },
+        {
+          Started: DateTimeOffset_1,
+          Completed: DateTimeOffset_1,
+          Error: Step Error
+        },
+        {
+          Started: DateTimeOffset_1,
+          Completed: DateTimeOffset_1,
+          Error: Step Error
+        }
+      ]
+    }
+  },
+  Created: DateTimeOffset_1,
+  Completed: DateTimeOffset_1,
+  Expires: DateTimeOffset_2,
+  Status: Failed
+}
diff --git a/flows/Squidex.Flows.Tests/DefaultFlowExecutorTests.Should_not_execute_next_step_if_step_completes_flow.verified.txt b/flows/Squidex.Flows.Tests/DefaultFlowExecutorTests.Should_not_execute_next_step_if_step_completes_flow.verified.txt
@@ -0,0 +1,39 @@
+﻿{
+  InstanceId: Guid_1,
+  OwnerId: Guid_2,
+  DefinitionId: Guid_3,
+  Definition: {
+    InitialStepId: StepId1,
+    Steps: {
+      StepId1: {
+        NextStepId: StepId2,
+        IgnoreError: false,
+        Step: {}
+      },
+      StepId2: {
+        IgnoreError: false,
+        Step: {}
+      }
+    }
+  },
+  Context: {},
+  Description: ,
+  ScheduleKey: ,
+  SchedulePartition: 75,
+  Steps: {
+    StepId1: {
+      Status: Completed,
+      IsPrepared: true,
+      Attempts: [
+        {
+          Started: DateTimeOffset_1,
+          Completed: DateTimeOffset_1
+        }
+      ]
+    }
+  },
+  Created: DateTimeOffset_1,
+  Completed: DateTimeOffset_1,
+  Expires: DateTimeOffset_2,
+  Status: Completed
+}
diff --git a/flows/Squidex.Flows.Tests/DefaultFlowExecutorTests.cs b/flows/Squidex.Flows.Tests/DefaultFlowExecutorTests.cs
@@ -153,6 +153,38 @@ public async Task Should_execute_first_step_with_error()
         await Verify(state).AddNamedGuid(stepId1, "StepId1");
     }
 
+    [Fact]
+    public async Task Should_abort_flow_when_preparation_fails()
+    {
+        var step1 = A.Fake<FlowStep>();
+
+        A.CallTo(() => step1.PrepareAsync(A<FlowExecutionContext>._, A<CancellationToken>._))
+            .Throws(new InvalidOperationException("Step Error"));
+
+        var state = sut.CreateState(
+            new CreateFlowInstanceRequest<TestFlowContext>
+            {
+                Context = new TestFlowContext(),
+                DefinitionId = Guid.NewGuid().ToString(),
+                Definition = new FlowDefinition
+                {
+                    Steps = new Dictionary<Guid, FlowStepDefinition>
+                    {
+                        [stepId1] = new FlowStepDefinition { Step = step1 },
+                    },
+                    InitialStepId = stepId1,
+                },
+                OwnerId = Guid.NewGuid().ToString(),
+            });
+
+        await sut.ExecuteAsync(state, default);
+
+        AssertPrepared(step1, 1);
+        AssertExecuted(step1, 0);
+
+        await Verify(state).AddNamedGuid(stepId1, "StepId1");
+    }
+
     [Fact]
     public async Task Should_execute_first_step_again_after_error()
     {
@@ -226,7 +258,7 @@ public async Task Should_execute_first_step_again_after_attempts_exeeded()
     }
 
     [Fact]
-    public async Task Should_execute_too_end_when_simulating()
+    public async Task Should_execute_to_end_when_simulating()
     {
         var step1 = A.Fake<FlowStep>();
 
@@ -338,7 +370,7 @@ public async Task Should_execute_next_step_after_ignored_error()
     }
 
     [Fact]
-    public async Task Should_not_execute_next_step_if_steps_completes_flow()
+    public async Task Should_not_execute_next_step_if_step_completes_flow()
     {
         var step1 = A.Fake<FlowStep>();
         var step2 = A.Fake<FlowStep>();
diff --git a/flows/Squidex.Flows.Tests/PartionedSchedulerTests.cs b/flows/Squidex.Flows.Tests/PartionedSchedulerTests.cs
@@ -0,0 +1,93 @@
+﻿// ==========================================================================
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex UG (haftungsbeschraenkt)
+//  All rights reserved. Licensed under the MIT license.
+// ==========================================================================
+
+using System.Threading.Channels;
+using Squidex.Flows.Internal.Execution.Utils;
+
+namespace Squidex.Flows;
+
+public class PartionedSchedulerTests
+{
+    [Fact]
+    public async Task Should_handle_events_in_order_for_same_key()
+    {
+        var counter = 0;
+
+        var sut = new PartitionedScheduler<int>(async (_, ct) =>
+        {
+            var x = counter;
+            await Task.Delay(1, ct);
+            counter = x + 1;
+        }, 32, 2);
+
+        for (var i = 0; i < 100; i++)
+        {
+            await sut.ScheduleAsync(0, i);
+        }
+
+        await sut.CompleteAsync();
+
+        Assert.Equal(100, counter);
+    }
+
+    [Fact]
+    public async Task Should_not_handle_events_in_order_for_different_keys()
+    {
+        var counter = 0;
+
+        var sut = new PartitionedScheduler<int>(async (_, ct) =>
+        {
+            var x = counter;
+            await Task.Delay(1, ct);
+            counter = x + 1;
+        }, 32, 2);
+
+        for (var i = 0; i < 500; i++)
+        {
+            await sut.ScheduleAsync(i, i);
+        }
+
+        await sut.CompleteAsync();
+
+        Assert.True(counter < 500);
+    }
+
+    [Fact]
+    public async Task Should_not_swallow_exception()
+    {
+        var sut = new PartitionedScheduler<int>((_, ct) =>
+        {
+            throw new InvalidOperationException();
+        }, 32, 2);
+
+        var ex = await Assert.ThrowsAsync<ChannelClosedException>(async () => await ScheduleUntilFailed(sut, 0, 0));
+        Assert.IsType<InvalidOperationException>(ex.InnerException);
+    }
+
+    [Fact]
+    public async Task Should_not_swallow_exception_from_other_partition()
+    {
+        var sut = new PartitionedScheduler<int>((_, ct) =>
+        {
+            throw new InvalidOperationException();
+        }, 32, 2);
+
+        var ex1 = await Assert.ThrowsAsync<ChannelClosedException>(async () => await ScheduleUntilFailed(sut, 1, 1));
+        var ex2 = await Assert.ThrowsAsync<ChannelClosedException>(async () => await sut.ScheduleAsync(0, 0));
+
+        Assert.IsType<InvalidOperationException>(ex1.InnerException);
+        Assert.IsType<InvalidOperationException>(ex2.InnerException);
+    }
+
+    private static async Task ScheduleUntilFailed<T>(PartitionedScheduler<T> sut, object key, T value)
+    {
+        for (var i = 0; i < 100; i++)
+        {
+            await sut.ScheduleAsync(key, value);
+        }
+    }
+}
diff --git a/flows/Squidex.Flows/FlowOptions.cs b/flows/Squidex.Flows/FlowOptions.cs
@@ -26,6 +26,8 @@ public sealed class FlowOptions : IValidatableOptions
 
     public int NumPartitions { get; set; } = 120;
 
+    public int BufferSizePerWorker { get; set; } = 2;
+
     public int WorkerIndex { get; set; }
 
     public Func<Exception, bool>? IsSafeException { get; set; } = _ => true;
diff --git a/flows/Squidex.Flows/Internal/Execution/DefaultFlowExecutor.cs b/flows/Squidex.Flows/Internal/Execution/DefaultFlowExecutor.cs
@@ -181,20 +181,29 @@ public async Task ExecuteAsync(FlowExecutionState<TContext> state,
     {
         ArgumentNullException.ThrowIfNull(nameof(state));
 
-        while (true)
+        try
         {
-            if (state.Status is FlowExecutionStatus.Completed or FlowExecutionStatus.Failed)
+            while (true)
             {
-                break;
-            }
+                if (state.Status is FlowExecutionStatus.Completed or FlowExecutionStatus.Failed)
+                {
+                    break;
+                }
 
-            var now = Clock.GetCurrentInstant();
-            if (state.NextRun > now)
-            {
-                break;
-            }
+                var now = Clock.GetCurrentInstant();
+                if (state.NextRun > now)
+                {
+                    break;
+                }
 
-            await ExecuteCoreAsync(state, false, DefaultTimeout, ct);
+                await ExecuteCoreAsync(state, false, DefaultTimeout, ct);
+            }
+        }
+        catch
+        {
+            // This method should never fail, when used properly.
+            state.Failed(Clock.GetCurrentInstant());
+            throw;
         }
     }
 
@@ -246,9 +255,11 @@ void Log(string message, string? dump)
                 throw new InvalidOperationException("Flow step has already been completed.");
             }
 
-            await PrepareAsync(executionContext, stepDefinition.Step, stepState, cts.Token);
             try
             {
+                // Ensure that we also catch errors in the preparation.
+                await PrepareAsync(executionContext, stepDefinition.Step, stepState, cts.Token);
+
                 stepState.Status = FlowExecutionStatus.Running;
 
                 var result = await pipeline(executionContext, ct) ??
@@ -296,7 +307,12 @@ private void HandleError(
         // Ensure to take the time after the step execution and preparation.
         var now = Clock.GetCurrentInstant();
 
-        if (stepDefinition.IgnoreError)
+        if (!stepState.IsPrepared)
+        {
+            state.Failed(now);
+            stepState.Status = FlowExecutionStatus.Failed;
+        }
+        else if (stepDefinition.IgnoreError)
         {
             var nextId = state.GetNextStep(stepDefinition, default);
             if (nextId != null)
diff --git a/flows/Squidex.Flows/Internal/Execution/FlowExecutionWorker.cs b/flows/Squidex.Flows/Internal/Execution/FlowExecutionWorker.cs
diff --git a/flows/Squidex.Flows/Internal/Execution/Utils/PartitionedScheduler.cs b/flows/Squidex.Flows/Internal/Execution/Utils/PartitionedScheduler.cs
