From 4aed8d9fa2aa76e2e574aae1418401155d910b33 Mon Sep 17 00:00:00 2001 From: xiangning Date: Thu, 23 Jul 2026 18:45:55 +0800 Subject: [PATCH] feat: process workflow timers with leases --- .../ProcessDueWorkflowTimersCommand.cs | 13 + .../ProcessDueWorkflowTimersCommandHandler.cs | 240 ++++++++++++++++++ .../Reliability/PostgresWorkflowCollection.cs | 13 + .../Reliability/PostgresWorkflowFixture.cs | 82 ++++++ .../WorkflowTimerProcessorTests.cs | 228 +++++++++++++++++ 5 files changed, 576 insertions(+) create mode 100644 src/modules/workflow/Application/Scheduler/Commands/ProcessDueWorkflowTimersCommand.cs create mode 100644 src/modules/workflow/Application/Scheduler/Handlers/ProcessDueWorkflowTimersCommandHandler.cs create mode 100644 tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowCollection.cs create mode 100644 tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowFixture.cs create mode 100644 tests/BuildFlow.Workflow.Tests/Reliability/WorkflowTimerProcessorTests.cs diff --git a/src/modules/workflow/Application/Scheduler/Commands/ProcessDueWorkflowTimersCommand.cs b/src/modules/workflow/Application/Scheduler/Commands/ProcessDueWorkflowTimersCommand.cs new file mode 100644 index 0000000..03e24a1 --- /dev/null +++ b/src/modules/workflow/Application/Scheduler/Commands/ProcessDueWorkflowTimersCommand.cs @@ -0,0 +1,13 @@ +using MediatR; + +namespace Workflow.Application.Scheduler.Commands; + +/// +/// 原子领取并处理一批到期工作流 Timer 的命令。 +/// +public sealed record ProcessDueWorkflowTimersCommand( + string LeaseOwner, + int BatchSize, + TimeSpan LeaseDuration, + int MaxAttempts, + TimeSpan RetryDelay) : IRequest; diff --git a/src/modules/workflow/Application/Scheduler/Handlers/ProcessDueWorkflowTimersCommandHandler.cs b/src/modules/workflow/Application/Scheduler/Handlers/ProcessDueWorkflowTimersCommandHandler.cs new file mode 100644 index 0000000..ff2fd03 --- /dev/null +++ b/src/modules/workflow/Application/Scheduler/Handlers/ProcessDueWorkflowTimersCommandHandler.cs @@ -0,0 +1,240 @@ +using System.Data.Common; +using MediatR; +using Microsoft.EntityFrameworkCore.Storage; +using Microsoft.Extensions.Logging; +using Workflow.Application.Engine; +using Workflow.Application.Scheduler.Commands; +using Workflow.Domain.Entities; +using Workflow.Domain.Enums; +using Workflow.Infrastructure.Persistence; + +namespace Workflow.Application.Scheduler.Handlers; + +/// +/// 使用 PostgreSQL 租约原子领取到期 Timer，并在事务内完成 token 推进。 +/// +public sealed class ProcessDueWorkflowTimersCommandHandler( + WorkflowDbContext db, + ProcessEngine engine, + ILogger logger) + : IRequestHandler +{ + public async Task Handle( + ProcessDueWorkflowTimersCommand request, + CancellationToken ct) + { + var claimedIds = await ClaimAsync(request, ct); + foreach (var jobId in claimedIds) + { + ct.ThrowIfCancellationRequested(); + try + { + await ProcessClaimedAsync(jobId, request.LeaseOwner, ct); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + logger.LogError(ex, "工作流 Timer 任务 {JobId} 处理失败", jobId); + await HandleFailureAsync(jobId, request, ex, ct); + } + } + + return claimedIds.Count; + } + + private async Task> ClaimAsync( + ProcessDueWorkflowTimersCommand request, + CancellationToken ct) + { + var now = DateTime.UtcNow; + var leaseExpiresAt = now.Add(request.LeaseDuration); + await using var transaction = await db.Database.BeginTransactionAsync(ct); + + await db.Database.ExecuteSqlInterpolatedAsync($""" + UPDATE b_workflow_timer_jobs + SET status = {(int)WorkflowTimerJobStatus.Failed}, + lease_owner = NULL, + lease_expires_at = NULL, + completed_at = {now}, + last_error = {"任务租约过期且已达到最大重试次数"}, + updated_by = {"system"}, + updated_at = {now} + WHERE status = {(int)WorkflowTimerJobStatus.Processing} + AND lease_expires_at <= {now} + AND attempt_count >= {request.MaxAttempts} + """, ct); + + await using var command = db.Database.GetDbConnection().CreateCommand(); + command.Transaction = transaction.GetDbTransaction(); + command.CommandText = """ + WITH candidates AS ( + SELECT id + FROM b_workflow_timer_jobs + WHERE ( + (status = @pending AND due_at <= @now) + OR (status = @processing AND lease_expires_at <= @now) + ) + AND attempt_count < @max_attempts + ORDER BY due_at, created_at + FOR UPDATE SKIP LOCKED + LIMIT @batch_size + ) + UPDATE b_workflow_timer_jobs AS job + SET status = @processing, + lease_owner = @lease_owner, + lease_expires_at = @lease_expires_at, + attempt_count = job.attempt_count + 1, + last_error = NULL, + updated_by = 'system', + updated_at = @now + FROM candidates + WHERE job.id = candidates.id + RETURNING job.id + """; + AddParameter(command, "pending", (int)WorkflowTimerJobStatus.Pending); + AddParameter(command, "processing", (int)WorkflowTimerJobStatus.Processing); + AddParameter(command, "now", now); + AddParameter(command, "max_attempts", request.MaxAttempts); + AddParameter(command, "batch_size", request.BatchSize); + AddParameter(command, "lease_owner", request.LeaseOwner); + AddParameter(command, "lease_expires_at", leaseExpiresAt); + + var claimedIds = new List(); + await using (var reader = await command.ExecuteReaderAsync(ct)) + { + while (await reader.ReadAsync(ct)) + { + claimedIds.Add(reader.GetGuid(0)); + } + } + + await transaction.CommitAsync(ct); + return claimedIds; + } + + private async Task ProcessClaimedAsync( + Guid jobId, + string leaseOwner, + CancellationToken ct) + { + await using var transaction = await db.Database.BeginTransactionAsync(ct); + var job = await db.WorkflowTimerJobs + .FromSqlInterpolated($"SELECT * FROM b_workflow_timer_jobs WHERE id = {jobId} FOR UPDATE") + .SingleOrDefaultAsync(ct); + if (job is null + || job.Status != WorkflowTimerJobStatus.Processing + || job.LeaseOwner != leaseOwner) + { + await transaction.RollbackAsync(ct); + return; + } + + var instance = await db.WorkflowInstances.FindAsync([job.InstanceId], ct); + var token = await db.WorkflowTokens.FindAsync([job.TokenId], ct); + if (instance is null + || token is null + || instance.Status != InstanceStatus.Running + || token.Status != TokenStatus.Active) + { + CompleteJob(job, WorkflowTimerJobStatus.Cancelled); + await db.SaveChangesAsync(ct); + await transaction.CommitAsync(ct); + return; + } + + await engine.EnsureGraphLoadedForAsync(instance); + var edge = (await engine.GetOutgoingEdgesPublicAsync(job.NodeId)).FirstOrDefault(); + token.Status = TokenStatus.Consumed; + token.CompletedAt = DateTime.UtcNow; + + WorkflowToken? downstreamToken = null; + if (edge is not null) + { + downstreamToken = new WorkflowToken + { + Id = Guid.NewGuid(), + InstanceId = instance.Id, + NodeId = edge.TargetNodeId, + Status = TokenStatus.Active, + }; + db.WorkflowTokens.Add(downstreamToken); + } + + CompleteJob(job, WorkflowTimerJobStatus.Completed); + await db.SaveChangesAsync(ct); + + if (downstreamToken is not null) + { + var targetNode = await engine.FindNodePublicAsync(downstreamToken.NodeId); + if (targetNode is not null) + { + await engine.ProcessNodeAsync(instance, downstreamToken, targetNode); + } + } + + await transaction.CommitAsync(ct); + } + + private async Task HandleFailureAsync( + Guid jobId, + ProcessDueWorkflowTimersCommand request, + Exception exception, + CancellationToken ct) + { + db.ChangeTracker.Clear(); + await using var transaction = await db.Database.BeginTransactionAsync(ct); + var job = await db.WorkflowTimerJobs + .FromSqlInterpolated($"SELECT * FROM b_workflow_timer_jobs WHERE id = {jobId} FOR UPDATE") + .SingleOrDefaultAsync(ct); + if (job is null + || job.Status != WorkflowTimerJobStatus.Processing + || job.LeaseOwner != request.LeaseOwner) + { + await transaction.RollbackAsync(ct); + return; + } + + job.LastError = Truncate(exception.Message, 2000); + job.LeaseOwner = null; + job.LeaseExpiresAt = null; + if (job.AttemptCount >= request.MaxAttempts) + { + job.Status = WorkflowTimerJobStatus.Failed; + job.CompletedAt = DateTime.UtcNow; + } + else + { + job.Status = WorkflowTimerJobStatus.Pending; + job.DueAt = DateTime.UtcNow.Add(request.RetryDelay); + job.CompletedAt = null; + } + + await db.SaveChangesAsync(ct); + await transaction.CommitAsync(ct); + } + + private static void CompleteJob( + WorkflowTimerJob job, + WorkflowTimerJobStatus status) + { + job.Status = status; + job.LeaseOwner = null; + job.LeaseExpiresAt = null; + job.LastError = null; + job.CompletedAt = DateTime.UtcNow; + } + + private static void AddParameter(DbCommand command, string name, object value) + { + var parameter = command.CreateParameter(); + parameter.ParameterName = name; + parameter.Value = value; + command.Parameters.Add(parameter); + } + + private static string Truncate(string value, int maxLength) + => value.Length <= maxLength ? value : value[..maxLength]; +} diff --git a/tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowCollection.cs b/tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowCollection.cs new file mode 100644 index 0000000..5e259c0 --- /dev/null +++ b/tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowCollection.cs @@ -0,0 +1,13 @@ +using Xunit; + +namespace BuildFlow.Workflow.Tests.Reliability; + +/// +/// 串行执行工作流 PostgreSQL 可靠性测试，避免共享测试服务发生数据库级竞争。 +/// +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class PostgresWorkflowCollection : ICollectionFixture +{ + /// 集合名称。 + public const string Name = "workflow-postgres"; +} diff --git a/tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowFixture.cs b/tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowFixture.cs new file mode 100644 index 0000000..28e0253 --- /dev/null +++ b/tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowFixture.cs @@ -0,0 +1,82 @@ +using Microsoft.EntityFrameworkCore; +using Npgsql; +using Xunit; + +namespace BuildFlow.Workflow.Tests.Reliability; + +/// +/// 为工作流可靠性测试创建并销毁独立 PostgreSQL 数据库。 +/// +public sealed class PostgresWorkflowFixture : IAsyncLifetime +{ + private const string DefaultConnection = + "Host=127.0.0.1;Port=55432;Username=buildflow;Password=buildflow;Database=postgres"; + + private string _databaseName = default!; + private string _adminConnectionString = default!; + + /// 当前测试数据库连接串。 + public string ConnectionString { get; private set; } = default!; + + public async Task InitializeAsync() + { + var configured = Environment.GetEnvironmentVariable("BUILDFLOW_WORKFLOW_TEST_CONNECTION") + ?? DefaultConnection; + var adminBuilder = new NpgsqlConnectionStringBuilder(configured) + { + Database = "postgres", + Pooling = false, + }; + _adminConnectionString = adminBuilder.ConnectionString; + _databaseName = $"workflow_reliability_{Guid.NewGuid():N}"; + + await using (var connection = new NpgsqlConnection(_adminConnectionString)) + { + await connection.OpenAsync(); + await using var command = connection.CreateCommand(); + command.CommandText = $"CREATE DATABASE \"{_databaseName}\""; + await command.ExecuteNonQueryAsync(); + } + + var databaseBuilder = new NpgsqlConnectionStringBuilder(configured) + { + Database = _databaseName, + Pooling = true, + }; + ConnectionString = databaseBuilder.ConnectionString; + + await using var db = CreateDbContext(); + await db.Database.EnsureCreatedAsync(); + } + + /// 创建指向独立测试数据库的 WorkflowDbContext。 + public WorkflowDbContext CreateDbContext() + { + var options = new DbContextOptionsBuilder() + .UseNpgsql(ConnectionString) + .UseSnakeCaseNamingConvention() + .Options; + var db = new WorkflowDbContext(options); + db.SavingChanges += AuditStub.OnSave; + return db; + } + + public async Task DisposeAsync() + { + NpgsqlConnection.ClearAllPools(); + await using var connection = new NpgsqlConnection(_adminConnectionString); + await connection.OpenAsync(); + + await using (var terminate = connection.CreateCommand()) + { + terminate.CommandText = + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = @database AND pid <> pg_backend_pid()"; + terminate.Parameters.AddWithValue("database", _databaseName); + await terminate.ExecuteNonQueryAsync(); + } + + await using var drop = connection.CreateCommand(); + drop.CommandText = $"DROP DATABASE IF EXISTS \"{_databaseName}\""; + await drop.ExecuteNonQueryAsync(); + } +} diff --git a/tests/BuildFlow.Workflow.Tests/Reliability/WorkflowTimerProcessorTests.cs b/tests/BuildFlow.Workflow.Tests/Reliability/WorkflowTimerProcessorTests.cs new file mode 100644 index 0000000..5a071ae --- /dev/null +++ b/tests/BuildFlow.Workflow.Tests/Reliability/WorkflowTimerProcessorTests.cs @@ -0,0 +1,228 @@ +using FluentAssertions; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Workflow.Application.Engine; +using Workflow.Application.Scheduler.Commands; +using Workflow.Application.Scheduler.Handlers; +using Workflow.Domain.Entities; +using Workflow.Domain.Enums; +using Workflow.Domain.Expressions; +using Xunit; + +namespace BuildFlow.Workflow.Tests.Reliability; + +/// +/// 工作流 Timer 的 PostgreSQL 原子领取、租约恢复和事务推进测试。 +/// +[Collection(PostgresWorkflowCollection.Name)] +public class WorkflowTimerProcessorTests(PostgresWorkflowFixture fixture) +{ + [Fact] + public async Task TwoWorkersClaimingOneDueTimerCreateOnlyOneDownstreamToken() + { + await ResetAsync(); + var seeded = await SeedTimerAsync(InstanceStatus.Running); + await using var dbA = fixture.CreateDbContext(); + await using var dbB = fixture.CreateDbContext(); + var handlerA = CreateHandler(dbA); + var handlerB = CreateHandler(dbB); + + var results = await Task.WhenAll( + handlerA.Handle(CreateCommand("worker-a"), CancellationToken.None), + handlerB.Handle(CreateCommand("worker-b"), CancellationToken.None)); + + results.Sum().Should().Be(1); + await using var verify = fixture.CreateDbContext(); + var job = await verify.WorkflowTimerJobs.SingleAsync(x => x.Id == seeded.JobId); + job.Status.Should().Be(WorkflowTimerJobStatus.Completed); + job.LeaseOwner.Should().BeNull(); + (await verify.WorkflowTokens.SingleAsync(x => x.Id == seeded.TokenId)) + .Status.Should().Be(TokenStatus.Consumed); + (await verify.WorkflowTokens.CountAsync(x => + x.InstanceId == seeded.InstanceId + && x.NodeId == seeded.TargetNodeId + && x.Status == TokenStatus.Active)).Should().Be(1); + } + + [Fact] + public async Task ExpiredProcessingLeaseCanBeRecoveredByAnotherWorker() + { + await ResetAsync(); + var seeded = await SeedTimerAsync( + InstanceStatus.Running, + WorkflowTimerJobStatus.Processing, + "dead-worker", + DateTime.UtcNow.AddMinutes(-1), + attemptCount: 1); + await using var db = fixture.CreateDbContext(); + var handler = CreateHandler(db); + + var processed = await handler.Handle(CreateCommand("recovery-worker"), CancellationToken.None); + + processed.Should().Be(1); + await using var verify = fixture.CreateDbContext(); + var job = await verify.WorkflowTimerJobs.SingleAsync(x => x.Id == seeded.JobId); + job.Status.Should().Be(WorkflowTimerJobStatus.Completed); + job.AttemptCount.Should().Be(2); + job.LeaseOwner.Should().BeNull(); + } + + [Fact] + public async Task NonRunningInstanceCancelsTimerWithoutCreatingDownstreamToken() + { + await ResetAsync(); + var seeded = await SeedTimerAsync(InstanceStatus.Terminated); + await using var db = fixture.CreateDbContext(); + var handler = CreateHandler(db); + + var processed = await handler.Handle(CreateCommand("worker-a"), CancellationToken.None); + + processed.Should().Be(1); + await using var verify = fixture.CreateDbContext(); + var job = await verify.WorkflowTimerJobs.SingleAsync(x => x.Id == seeded.JobId); + job.Status.Should().Be(WorkflowTimerJobStatus.Cancelled); + job.LeaseOwner.Should().BeNull(); + (await verify.WorkflowTokens.CountAsync(x => x.NodeId == seeded.TargetNodeId)).Should().Be(0); + } + + [Fact] + public async Task ExpiredLeaseAtMaxAttemptsBecomesFailed() + { + await ResetAsync(); + var seeded = await SeedTimerAsync( + InstanceStatus.Running, + WorkflowTimerJobStatus.Processing, + "dead-worker", + DateTime.UtcNow.AddMinutes(-1), + attemptCount: 5); + await using var db = fixture.CreateDbContext(); + var handler = CreateHandler(db); + + var processed = await handler.Handle(CreateCommand("worker-a"), CancellationToken.None); + + processed.Should().Be(0); + await using var verify = fixture.CreateDbContext(); + var job = await verify.WorkflowTimerJobs.SingleAsync(x => x.Id == seeded.JobId); + job.Status.Should().Be(WorkflowTimerJobStatus.Failed); + job.LastError.Should().Contain("最大重试次数"); + job.LeaseOwner.Should().BeNull(); + (await verify.WorkflowTokens.CountAsync(x => x.NodeId == seeded.TargetNodeId)).Should().Be(0); + } + + private static ProcessDueWorkflowTimersCommand CreateCommand(string owner) => new( + owner, + BatchSize: 10, + LeaseDuration: TimeSpan.FromMinutes(1), + MaxAttempts: 5, + RetryDelay: TimeSpan.FromSeconds(1)); + + private static ProcessDueWorkflowTimersCommandHandler CreateHandler(WorkflowDbContext db) + { + var services = new ServiceCollection().BuildServiceProvider(); + var engine = new ProcessEngine(db, services, new ConditionEvaluator()); + return new ProcessDueWorkflowTimersCommandHandler( + db, + engine, + NullLogger.Instance); + } + + private async Task ResetAsync() + { + await using var db = fixture.CreateDbContext(); + await db.WorkflowTimerJobs.ExecuteDeleteAsync(); + await db.WorkflowTasks.ExecuteDeleteAsync(); + await db.WorkflowTokens.ExecuteDeleteAsync(); + await db.WorkflowInstances.ExecuteDeleteAsync(); + await db.WorkflowEdges.ExecuteDeleteAsync(); + await db.WorkflowNodes.ExecuteDeleteAsync(); + await db.WorkflowDefinitionSnapshots.ExecuteDeleteAsync(); + await db.WorkflowDefinitions.ExecuteDeleteAsync(); + } + + private async Task SeedTimerAsync( + InstanceStatus instanceStatus, + WorkflowTimerJobStatus jobStatus = WorkflowTimerJobStatus.Pending, + string? leaseOwner = null, + DateTime? leaseExpiresAt = null, + int attemptCount = 0) + { + await using var db = fixture.CreateDbContext(); + var definitionId = Guid.NewGuid(); + var timerNodeId = Guid.NewGuid(); + var targetNodeId = Guid.NewGuid(); + var instanceId = Guid.NewGuid(); + var tokenId = Guid.NewGuid(); + var jobId = Guid.NewGuid(); + + db.WorkflowDefinitions.Add(new WorkflowDefinition + { + Id = definitionId, + Name = "Timer测试流程", + Code = $"timer-{definitionId:N}", + Version = 1, + Status = DefinitionStatus.Published, + }); + db.WorkflowNodes.AddRange( + new WorkflowNode + { + Id = timerNodeId, + DefinitionId = definitionId, + Name = "等待", + NodeType = NodeType.Timer, + }, + new WorkflowNode + { + Id = targetNodeId, + DefinitionId = definitionId, + Name = "审批", + NodeType = NodeType.Approval, + Config = """{ "assigneeRule": "role:user" }""", + }); + db.WorkflowEdges.Add(new WorkflowEdge + { + Id = Guid.NewGuid(), + DefinitionId = definitionId, + SourceNodeId = timerNodeId, + TargetNodeId = targetNodeId, + EdgeType = EdgeType.Normal, + }); + db.WorkflowInstances.Add(new WorkflowInstance + { + Id = instanceId, + DefinitionId = definitionId, + DefinitionVersion = 1, + Title = "Timer测试实例", + Status = instanceStatus, + Variables = "{}", + }); + db.WorkflowTokens.Add(new WorkflowToken + { + Id = tokenId, + InstanceId = instanceId, + NodeId = timerNodeId, + Status = TokenStatus.Active, + }); + db.WorkflowTimerJobs.Add(new WorkflowTimerJob + { + Id = jobId, + InstanceId = instanceId, + TokenId = tokenId, + NodeId = timerNodeId, + DueAt = DateTime.UtcNow.AddMinutes(-1), + Status = jobStatus, + LeaseOwner = leaseOwner, + LeaseExpiresAt = leaseExpiresAt, + AttemptCount = attemptCount, + }); + await db.SaveChangesAsync(); + + return new SeededTimer(jobId, instanceId, tokenId, targetNodeId); + } + + private sealed record SeededTimer( + Guid JobId, + Guid InstanceId, + Guid TokenId, + Guid TargetNodeId); +}