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 <sipkeschoorstra@outlook.com> * 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 <sipkeschoorstra@outlook.com> --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com>
This commit is contained in:
Sipke Schoorstra
GitHub
parent
90c29fb589
commit
9ee53abdaa
|
|
@ -40,7 +40,7 @@ public class EFCoreWorkflowExecutionLogStore(EntityStore<RuntimeElsaDbContext, W
|
|||
/// <inheritdoc />
|
||||
public async Task<WorkflowExecutionLogRecord?> 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();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
|
@ -53,7 +53,7 @@ public class EFCoreWorkflowExecutionLogStore(EntityStore<RuntimeElsaDbContext, W
|
|||
public async Task<Page<WorkflowExecutionLogRecord>> 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);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -32,8 +32,10 @@ public class Get(IWorkflowExecutionLogStore store) : ElsaEndpoint<Request, Workf
|
|||
EventNames = ["Started", "Completed", "Faulted"]
|
||||
};
|
||||
|
||||
var sort = new WorkflowExecutionLogRecordOrder<DateTimeOffset>(
|
||||
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<long>(
|
||||
x => x.Sequence,
|
||||
OrderDirection.Descending
|
||||
);
|
||||
|
||||
|
|
|
|||
|
|
@ -24,7 +24,8 @@ public interface IWorkflowExecutionLogStore : ILogRecordStore<WorkflowExecutionL
|
|||
Task SaveAsync(WorkflowExecutionLogRecord record, CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
Task<WorkflowExecutionLogRecord?> FindAsync(WorkflowExecutionLogRecordFilter filter, CancellationToken cancellationToken = default);
|
||||
|
||||
|
|
@ -34,7 +35,8 @@ public interface IWorkflowExecutionLogStore : ILogRecordStore<WorkflowExecutionL
|
|||
Task<WorkflowExecutionLogRecord?> FindAsync<TOrderBy>(WorkflowExecutionLogRecordFilter filter, WorkflowExecutionLogRecordOrder<TOrderBy> order, CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
Task<Page<WorkflowExecutionLogRecord>> FindManyAsync(WorkflowExecutionLogRecordFilter filter, PageArgs pageArgs, CancellationToken cancellationToken = default);
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,25 @@
|
|||
using Elsa.Workflows.Runtime;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
|
||||
// ReSharper disable once CheckNamespace
|
||||
namespace Elsa.Extensions;
|
||||
|
||||
/// <summary>
|
||||
/// Canonical default order for <see cref="IWorkflowExecutionLogStore"/> queries that omit an explicit order.
|
||||
/// </summary>
|
||||
public static class WorkflowExecutionLogRecordQueryableExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Timestamp ascending, then Sequence ascending, then Id ascending.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Preserves EF's historical Timestamp default and uses Sequence as the same-timestamp
|
||||
/// tiebreaker — the purpose of <see cref="WorkflowExecutionLogRecord.Sequence"/>.
|
||||
/// 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 (<c>ExecutionLogSequence</c>).
|
||||
/// </remarks>
|
||||
public static IQueryable<WorkflowExecutionLogRecord> OrderByTimestampThenSequence(this IQueryable<WorkflowExecutionLogRecord> queryable) =>
|
||||
queryable.OrderBy(x => x.Timestamp).ThenBy(x => x.Sequence).ThenBy(x => x.Id);
|
||||
}
|
||||
|
|
@ -51,7 +51,7 @@ public class MemoryWorkflowExecutionLogStore : IWorkflowExecutionLogStore
|
|||
/// <inheritdoc />
|
||||
public Task<WorkflowExecutionLogRecord?> 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<Page<WorkflowExecutionLogRecord>> 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));
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<long>(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
|
||||
};
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue