From 9ee53abdaaa2f732fb622bd5a49815fc33d3562c Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Mon, 14 Sep 2026 14:00:20 +0200 Subject: [PATCH] fix(runtime): execution log default order Timestamp+Sequence; GetLastEntry Sequence-aware (#8152) * fix(runtime): order execution log Find by Timestamp then Sequence Align Memory and EF default Find/FindMany with Timestamp then Sequence so same-ms batches are stable across stores. Make Journal/GetLastEntry order by Sequence descending so the last event is not arbitrary. Co-authored-by: Sipke Schoorstra * fix(runtime): tie default execution-log order with Id Sequence is per execution context, so distinct instances can share Timestamp+Sequence. Add Id as the unique default sort key so Memory and EF offset pages stay deterministic. Co-authored-by: Sipke Schoorstra --------- Co-authored-by: Cursor Agent --- .../Runtime/WorkflowExecutionLogStore.cs | 4 +- .../Journal/GetLastEntry/Endpoint.cs | 6 +- .../Contracts/IWorkflowExecutionLogStore.cs | 6 +- ...owExecutionLogRecordQueryableExtensions.cs | 25 +++++++ .../Stores/MemoryWorkflowExecutionLogStore.cs | 4 +- .../WorkflowStoreConformanceTests.cs | 69 ++++++++++++++++++- 6 files changed, 103 insertions(+), 11 deletions(-) create mode 100644 src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowExecutionLogRecordQueryableExtensions.cs diff --git a/src/modules/Elsa.Persistence.EFCore/Modules/Runtime/WorkflowExecutionLogStore.cs b/src/modules/Elsa.Persistence.EFCore/Modules/Runtime/WorkflowExecutionLogStore.cs index ee914012f..6b10d3312 100644 --- a/src/modules/Elsa.Persistence.EFCore/Modules/Runtime/WorkflowExecutionLogStore.cs +++ b/src/modules/Elsa.Persistence.EFCore/Modules/Runtime/WorkflowExecutionLogStore.cs @@ -40,7 +40,7 @@ public class EFCoreWorkflowExecutionLogStore(EntityStore public async Task FindAsync(WorkflowExecutionLogRecordFilter filter, CancellationToken cancellationToken = default) { - return await store.QueryAsync(queryable => Filter(queryable, filter), OnLoadAsync, cancellationToken).FirstOrDefault(); + return await store.QueryAsync(queryable => Filter(queryable, filter).OrderByTimestampThenSequence(), OnLoadAsync, cancellationToken).FirstOrDefault(); } /// @@ -53,7 +53,7 @@ public class EFCoreWorkflowExecutionLogStore(EntityStore> FindManyAsync(WorkflowExecutionLogRecordFilter filter, PageArgs pageArgs, CancellationToken cancellationToken = default) { var count = await store.QueryAsync(queryable => Filter(queryable, filter), cancellationToken).LongCount(); - var results = await store.QueryAsync(queryable => Filter(queryable, filter).OrderBy(x => x.Timestamp).Paginate(pageArgs), OnLoadAsync, cancellationToken).ToList(); + var results = await store.QueryAsync(queryable => Filter(queryable, filter).OrderByTimestampThenSequence().Paginate(pageArgs), OnLoadAsync, cancellationToken).ToList(); return new(results, count); } diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/Journal/GetLastEntry/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/Journal/GetLastEntry/Endpoint.cs index 3b57e07a6..bfc10f397 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/Journal/GetLastEntry/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/Journal/GetLastEntry/Endpoint.cs @@ -32,8 +32,10 @@ public class Get(IWorkflowExecutionLogStore store) : ElsaEndpoint( - x => x.Timestamp, + // Sequence is the instance-monotonic event cursor. Same-timestamp batches must not + // pick an arbitrary "last" entry when only Timestamp descending is applied. + var sort = new WorkflowExecutionLogRecordOrder( + x => x.Sequence, OrderDirection.Descending ); diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowExecutionLogStore.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowExecutionLogStore.cs index 50a4767a0..02634c336 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowExecutionLogStore.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowExecutionLogStore.cs @@ -24,7 +24,8 @@ public interface IWorkflowExecutionLogStore : ILogRecordStore - /// Returns the first workflow execution log record matching the specified filter. + /// Returns the first workflow execution log record matching the specified filter, + /// ordered by Timestamp, Sequence, then Id ascending. /// Task FindAsync(WorkflowExecutionLogRecordFilter filter, CancellationToken cancellationToken = default); @@ -34,7 +35,8 @@ public interface IWorkflowExecutionLogStore : ILogRecordStore FindAsync(WorkflowExecutionLogRecordFilter filter, WorkflowExecutionLogRecordOrder order, CancellationToken cancellationToken = default); /// - /// Returns a set of workflow execution log records matching the specified filter. + /// Returns a set of workflow execution log records matching the specified filter, + /// ordered by Timestamp, Sequence, then Id ascending. /// Task> FindManyAsync(WorkflowExecutionLogRecordFilter filter, PageArgs pageArgs, CancellationToken cancellationToken = default); diff --git a/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowExecutionLogRecordQueryableExtensions.cs b/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowExecutionLogRecordQueryableExtensions.cs new file mode 100644 index 000000000..ea10f82ef --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowExecutionLogRecordQueryableExtensions.cs @@ -0,0 +1,25 @@ +using Elsa.Workflows.Runtime; +using Elsa.Workflows.Runtime.Entities; + +// ReSharper disable once CheckNamespace +namespace Elsa.Extensions; + +/// +/// Canonical default order for queries that omit an explicit order. +/// +public static class WorkflowExecutionLogRecordQueryableExtensions +{ + /// + /// Timestamp ascending, then Sequence ascending, then Id ascending. + /// + /// + /// Preserves EF's historical Timestamp default and uses Sequence as the same-timestamp + /// tiebreaker — the purpose of . + /// Sequence is per execution context, so distinct instances can share Timestamp+Sequence; + /// Id is the unique key that keeps offset pagination stable across Memory and EF. + /// Journal list APIs still pass Sequence-primary order explicitly because Sequence is + /// the instance-monotonic event cursor (ExecutionLogSequence). + /// + public static IQueryable OrderByTimestampThenSequence(this IQueryable queryable) => + queryable.OrderBy(x => x.Timestamp).ThenBy(x => x.Sequence).ThenBy(x => x.Id); +} diff --git a/src/modules/Elsa.Workflows.Runtime/Stores/MemoryWorkflowExecutionLogStore.cs b/src/modules/Elsa.Workflows.Runtime/Stores/MemoryWorkflowExecutionLogStore.cs index 73b7a1795..95194b48d 100644 --- a/src/modules/Elsa.Workflows.Runtime/Stores/MemoryWorkflowExecutionLogStore.cs +++ b/src/modules/Elsa.Workflows.Runtime/Stores/MemoryWorkflowExecutionLogStore.cs @@ -51,7 +51,7 @@ public class MemoryWorkflowExecutionLogStore : IWorkflowExecutionLogStore /// public Task FindAsync(WorkflowExecutionLogRecordFilter filter, CancellationToken cancellationToken = default) { - var result = _store.Query(query => Filter(query, filter)).FirstOrDefault(); + var result = _store.Query(query => Filter(query, filter).OrderByTimestampThenSequence()).FirstOrDefault(); return Task.FromResult(result); } @@ -66,7 +66,7 @@ public class MemoryWorkflowExecutionLogStore : IWorkflowExecutionLogStore public Task> FindManyAsync(WorkflowExecutionLogRecordFilter filter, PageArgs pageArgs, CancellationToken cancellationToken = default) { var count = _store.Query(query => Filter(query, filter)).LongCount(); - var result = _store.Query(query => Filter(query, filter).Paginate(pageArgs)).ToList(); + var result = _store.Query(query => Filter(query, filter).OrderByTimestampThenSequence().Paginate(pageArgs)).ToList(); return Task.FromResult(Page.Of(result, count)); } diff --git a/test/integration/Elsa.Workflows.Persistence.ConformanceTests/WorkflowStoreConformanceTests.cs b/test/integration/Elsa.Workflows.Persistence.ConformanceTests/WorkflowStoreConformanceTests.cs index 4a25b6fe5..1402a1862 100644 --- a/test/integration/Elsa.Workflows.Persistence.ConformanceTests/WorkflowStoreConformanceTests.cs +++ b/test/integration/Elsa.Workflows.Persistence.ConformanceTests/WorkflowStoreConformanceTests.cs @@ -1,3 +1,4 @@ +using Elsa.Common.Entities; using Elsa.Common.Models; using Elsa.Common.Multitenancy; using Elsa.Workflows; @@ -5,6 +6,7 @@ using Elsa.Workflows.Management.Entities; using Elsa.Workflows.Management.Filters; using Elsa.Workflows.Runtime.Entities; using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.OrderDefinitions; namespace Elsa.Workflows.Persistence.ConformanceTests; @@ -316,6 +318,65 @@ public abstract class WorkflowStoreConformanceTests Assert.Equal("el-3", Assert.Single((await scenario.ExecutionLogs.FindManyAsync(new WorkflowExecutionLogRecordFilter(), PageArgs.All)).Items).Id); } + [Fact] + public async Task ExecutionLogDefaultOrderIsTimestampThenSequenceAndLastEntryUsesSequence() + { + await using var scenario = await CreateScenarioAsync(); + var sameTimestamp = StartedAt; + var laterTimestamp = StartedAt.AddMinutes(1); + + // Insert later Sequence first so dictionary/heap order cannot accidentally match the contract. + await scenario.ExecutionLogs.SaveAsync(ExecutionLog("el-seq-2", "instance-1", "activity-a", "Completed", timestamp: sameTimestamp, sequence: 2)); + await scenario.ExecutionLogs.SaveAsync(ExecutionLog("el-seq-3", "instance-1", "activity-a", "Faulted", timestamp: sameTimestamp, sequence: 3)); + await scenario.ExecutionLogs.SaveAsync(ExecutionLog("el-seq-1", "instance-1", "activity-a", "Started", timestamp: sameTimestamp, sequence: 1)); + await scenario.ExecutionLogs.SaveAsync(ExecutionLog("el-later", "instance-1", "activity-b", "Started", timestamp: laterTimestamp, sequence: 0)); + + var filter = new WorkflowExecutionLogRecordFilter { WorkflowInstanceId = "instance-1" }; + var page = await scenario.ExecutionLogs.FindManyAsync(filter, PageArgs.All); + Assert.Equal(["el-seq-1", "el-seq-2", "el-seq-3", "el-later"], page.Items.Select(x => x.Id).ToArray()); + + var firstPage = await scenario.ExecutionLogs.FindManyAsync(filter, PageArgs.FromRange(0, 2)); + Assert.Equal(4, firstPage.TotalCount); + Assert.Equal(["el-seq-1", "el-seq-2"], firstPage.Items.Select(x => x.Id).ToArray()); + + var first = await scenario.ExecutionLogs.FindAsync(filter); + Assert.Equal("el-seq-1", first!.Id); + + var sameTimestampFilter = new WorkflowExecutionLogRecordFilter + { + WorkflowInstanceId = "instance-1", + ActivityId = "activity-a", + EventNames = ["Started", "Completed", "Faulted"] + }; + var lastEntryOrder = new WorkflowExecutionLogRecordOrder(x => x.Sequence, OrderDirection.Descending); + var last = await scenario.ExecutionLogs.FindAsync(sameTimestampFilter, lastEntryOrder); + Assert.Equal("el-seq-3", last!.Id); + } + + [Fact] + public async Task ExecutionLogDefaultOrderUsesIdWhenTimestampAndSequenceTieAcrossInstances() + { + await using var scenario = await CreateScenarioAsync(); + var timestamp = StartedAt; + const long sequence = 1; + + // Reverse Id insert order so dictionary/heap order cannot accidentally match Id ascending. + await scenario.ExecutionLogs.SaveAsync(ExecutionLog("el-d", "instance-4", "activity-a", "Started", timestamp: timestamp, sequence: sequence)); + await scenario.ExecutionLogs.SaveAsync(ExecutionLog("el-b", "instance-2", "activity-a", "Started", timestamp: timestamp, sequence: sequence)); + await scenario.ExecutionLogs.SaveAsync(ExecutionLog("el-c", "instance-3", "activity-a", "Started", timestamp: timestamp, sequence: sequence)); + await scenario.ExecutionLogs.SaveAsync(ExecutionLog("el-a", "instance-1", "activity-a", "Started", timestamp: timestamp, sequence: sequence)); + + var filter = new WorkflowExecutionLogRecordFilter(); + var all = await scenario.ExecutionLogs.FindManyAsync(filter, PageArgs.All); + Assert.Equal(["el-a", "el-b", "el-c", "el-d"], all.Items.Select(x => x.Id).ToArray()); + + var firstPage = await scenario.ExecutionLogs.FindManyAsync(filter, PageArgs.FromRange(0, 2)); + var secondPage = await scenario.ExecutionLogs.FindManyAsync(filter, PageArgs.FromRange(2, 2)); + Assert.Equal(4, firstPage.TotalCount); + Assert.Equal(["el-a", "el-b"], firstPage.Items.Select(x => x.Id).ToArray()); + Assert.Equal(["el-c", "el-d"], secondPage.Items.Select(x => x.Id).ToArray()); + } + private static async Task SeedMixedTriggersAsync(WorkflowStoreScenario scenario) { await scenario.Triggers.SaveAsync(Trigger("id-a", hash: "hash-a", tenantId: "tenant-a")); @@ -411,7 +472,9 @@ public abstract class WorkflowStoreConformanceTests string workflowInstanceId, string activityId, string eventName, - string activityType = "Elsa.WriteLine") => + string activityType = "Elsa.WriteLine", + DateTimeOffset? timestamp = null, + long sequence = 0) => new() { Id = id, @@ -425,8 +488,8 @@ public abstract class WorkflowStoreConformanceTests ActivityType = activityType, ActivityTypeVersion = 1, ActivityNodeId = $"node-{activityId}", - Timestamp = StartedAt, - Sequence = 0, + Timestamp = timestamp ?? StartedAt, + Sequence = sequence, EventName = eventName };