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);
+}