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