build-flow.api/tests/BuildFlow.Workflow.Tests/Reliability/WorkflowTimerProcessorTests.cs

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;

/// <summary>
/// 工作流 Timer 的 PostgreSQL 原子领取、租约恢复和事务推进测试。
/// </summary>
[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<ProcessDueWorkflowTimersCommandHandler>.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<SeededTimer> 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);
}