build-flow/build-flow.api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
build-flow.api/docs/superpowers/plans/2026-07-23-buildflow-workflow-job-reliability.md
xiangning e453ba1969 docs: plan workflow job reliability
2026-07-23 18:35:44 +08:00

36 KiB
Raw Blame History

BuildFlow Workflow Job Reliability Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Replace process-local workflow timers and non-atomic import polling with PostgreSQL-backed jobs that are safe across multiple Host instances, process restarts, lease expiry, and retries.

Architecture: Workflow owns all job state and processing in Domain/Application/Infrastructure. BuildFlow.Host background services only create scopes and send MediatR commands. PostgreSQL performs atomic claiming with FOR UPDATE SKIP LOCKED; Timer completion and import row progress use database transactions so committed work is never duplicated after recovery.

Tech Stack: .NET 10, EF Core 10, Npgsql/PostgreSQL 17, MediatR 12, xUnit, FluentAssertions, FastEndpoints Host background services.

Scope And Contracts

  • Existing HTTP routes, request DTOs, response DTOs, permissions, and status values remain compatible.
  • Workflow schema changes:
    • add b_workflow_timer_jobs;
    • add lease/attempt/heartbeat columns and composite indexes to b_import_jobs.
  • Timer jobs use Pending, Processing, Completed, Failed, and Cancelled.
  • Import jobs keep the existing Pending, Running, Success, Failed, and PartialSuccess API-visible states.
  • LeaseOwner is a per-Host-service-instance identifier and is never returned by the API.
  • A graceful import cancellation releases the lease and returns the job to Pending; a hard crash leaves Running until lease expiry.
  • Import recovery uses SuccessRows + FailedRows as the committed row cursor. Each row and its cursor update commit in the same PostgreSQL transaction. Third batch will replace per-row commits with bounded batches without changing this recovery contract.
  • No new cross-module service bridge is allowed. Both Host services call workflow Application through MediatR.

Fixed Configuration

Add Workflow:Jobs with these defaults:

"Workflow": {
  "Jobs": {
    "Timer": {
      "PollIntervalSeconds": 5,
      "StartupDelaySeconds": 5,
      "BatchSize": 20,
      "LeaseSeconds": 60,
      "MaxAttempts": 5,
      "RetryDelaySeconds": 30
    },
    "Import": {
      "PollIntervalSeconds": 5,
      "StartupDelaySeconds": 5,
      "BatchSize": 5,
      "LeaseSeconds": 300,
      "MaxAttempts": 5
    }
  }
}

BatchSize, interval, lease, retry delay, and max attempts must all be positive. Host option validation must fail startup for invalid values.

Task 1: Add Workflow Job State And EF Model

Files:

  • Create: src/modules/workflow/Domain/Enums/WorkflowTimerJobStatus.cs

  • Create: src/modules/workflow/Domain/Entities/WorkflowTimerJob.cs

  • Create: src/modules/workflow/Infrastructure/Persistence/Configurations/WorkflowTimerJobConfiguration.cs

  • Create: src/modules/workflow/Application/Reliability/WorkflowJobOptions.cs

  • Modify: src/modules/workflow/Domain/Entities/BuildFlowImportJob.cs

  • Modify: src/modules/workflow/Infrastructure/Persistence/Configurations/BuildFlowImportJobConfiguration.cs

  • Modify: src/modules/workflow/Infrastructure/Persistence/WorkflowDbContext.cs

  • Create: tests/BuildFlow.Workflow.Tests/Reliability/WorkflowJobModelTests.cs

  • Step 1: Write the failing model test

Create WorkflowJobModelTests.cs:

using FluentAssertions;
using Microsoft.EntityFrameworkCore;
using Workflow.Domain.Entities;
using Xunit;

namespace BuildFlow.Workflow.Tests.Reliability;

/// <summary>
/// 工作流持久化任务的 EF 模型约束测试。
/// </summary>
public class WorkflowJobModelTests
{
    [Fact]
    public void WorkflowTimerJobShouldHaveUniqueTokenAndLeaseIndexes()
    {
        using var db = CreateDb();
        var entity = db.Model.FindEntityType(typeof(WorkflowTimerJob));

        entity.Should().NotBeNull();
        entity!.GetTableName().Should().Be("b_workflow_timer_jobs");
        entity.GetIndexes().Single(x => x.Properties.Select(p => p.Name)
            .SequenceEqual([nameof(WorkflowTimerJob.TokenId)])).IsUnique.Should().BeTrue();
        entity.GetIndexes().Should().Contain(x => x.Properties.Select(p => p.Name)
            .SequenceEqual([nameof(WorkflowTimerJob.Status), nameof(WorkflowTimerJob.DueAt)]));
        entity.GetIndexes().Should().Contain(x => x.Properties.Select(p => p.Name)
            .SequenceEqual([nameof(WorkflowTimerJob.Status), nameof(WorkflowTimerJob.LeaseExpiresAt)]));
    }

    [Fact]
    public void ImportJobShouldHaveLeaseFieldsAndCompositeIndexes()
    {
        using var db = CreateDb();
        var entity = db.Model.FindEntityType(typeof(BuildFlowImportJob))!;

        entity.FindProperty(nameof(BuildFlowImportJob.LeaseOwner)).Should().NotBeNull();
        entity.FindProperty(nameof(BuildFlowImportJob.LeaseExpiresAt)).Should().NotBeNull();
        entity.FindProperty(nameof(BuildFlowImportJob.AttemptCount)).Should().NotBeNull();
        entity.FindProperty(nameof(BuildFlowImportJob.LastHeartbeatAt)).Should().NotBeNull();
        entity.GetIndexes().Should().Contain(x => x.Properties.Select(p => p.Name)
            .SequenceEqual([nameof(BuildFlowImportJob.Status), nameof(BuildFlowImportJob.CreatedAt)]));
        entity.GetIndexes().Should().Contain(x => x.Properties.Select(p => p.Name)
            .SequenceEqual([nameof(BuildFlowImportJob.Status), nameof(BuildFlowImportJob.LeaseExpiresAt)]));
    }

    private static WorkflowDbContext CreateDb()
    {
        var options = new DbContextOptionsBuilder<WorkflowDbContext>()
            .UseInMemoryDatabase($"workflow-job-model-{Guid.NewGuid():N}")
            .Options;
        return new WorkflowDbContext(options);
    }
}
  • Step 2: Run the test and verify RED

Run:

dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowJobModelTests

Expected: compilation fails because WorkflowTimerJob and the import lease properties do not exist.

  • Step 3: Add the Timer status and entity

Create WorkflowTimerJobStatus.cs with exactly five values:

namespace Workflow.Domain.Enums;

/// <summary>
/// 工作流定时任务状态。Processing 必须持有有效租约，Completed/Failed/Cancelled 为终态。
/// </summary>
public enum WorkflowTimerJobStatus
{
    /// <summary>等待到期领取。</summary>
    Pending = 0,

    /// <summary>已被某个 Host 实例领取并持有租约。</summary>
    Processing = 1,

    /// <summary>已成功消费源 token 并完成下游投递。</summary>
    Completed = 2,

    /// <summary>达到最大重试次数后永久失败。</summary>
    Failed = 3,

    /// <summary>实例或 token 已不可继续推进。</summary>
    Cancelled = 4,
}

Create WorkflowTimerJob.cs:

using BuildFlow.Shared.Domain.Common;
using Workflow.Domain.Enums;

namespace Workflow.Domain.Entities;

/// <summary>
/// 持久化工作流 Timer 任务。租约保证多实例只处理一次，TokenId 唯一保证同一令牌只建一个任务。
/// </summary>
public class WorkflowTimerJob : BaseEntity, IAuditable
{
    /// <summary>所属工作流实例 ID。</summary>
    public Guid InstanceId { get; set; }

    /// <summary>等待到期的源 token ID，同一 token 只允许一个 Timer 任务。</summary>
    public Guid TokenId { get; set; }

    /// <summary>Timer 节点 ID。</summary>
    public Guid NodeId { get; set; }

    /// <summary>计划触发时间（UTC）。</summary>
    public DateTime DueAt { get; set; }

    /// <summary>任务状态。</summary>
    public WorkflowTimerJobStatus Status { get; set; } = WorkflowTimerJobStatus.Pending;

    /// <summary>当前租约持有者。</summary>
    public string? LeaseOwner { get; set; }

    /// <summary>租约到期时间（UTC）。</summary>
    public DateTime? LeaseExpiresAt { get; set; }

    /// <summary>累计领取次数。</summary>
    public int AttemptCount { get; set; }

    /// <summary>最近一次失败原因。</summary>
    public string? LastError { get; set; }

    /// <summary>进入终态的时间（UTC）。</summary>
    public DateTime? CompletedAt { get; set; }

    /// <summary>创建人。</summary>
    public string CreatedBy { get; set; } = default!;

    /// <summary>创建时间（UTC）。</summary>
    public DateTime CreatedAt { get; set; }

    /// <summary>最后修改人。</summary>
    public string UpdatedBy { get; set; } = default!;

    /// <summary>最后修改时间（UTC）。</summary>
    public DateTime UpdatedAt { get; set; }
}
  • Step 4: Configure the Timer table and import lease columns

WorkflowTimerJobConfiguration must map every field with a Chinese .HasComment(), use ValueGeneratedNever(), set LeaseOwner to HasMaxLength(200).IsConcurrencyToken(), set LastError to 2000 characters, add the unique TokenId index, and add (Status, DueAt) plus (Status, LeaseExpiresAt) indexes. Add cascade foreign keys to WorkflowInstance and WorkflowToken; do not add a foreign key to WorkflowNode because published JSON snapshots may contain nodes not present in the legacy relation table.

Add these properties to BuildFlowImportJob:

/// <summary>当前处理租约的 Host 实例标识。</summary>
public string? LeaseOwner { get; set; }

/// <summary>租约到期时间（UTC），到期后允许其他实例恢复。</summary>
public DateTime? LeaseExpiresAt { get; set; }

/// <summary>任务被领取的累计次数，包含租约恢复。</summary>
public int AttemptCount { get; set; }

/// <summary>最近一次处理心跳时间（UTC）。</summary>
public DateTime? LastHeartbeatAt { get; set; }

Map them with comments, LeaseOwner length 200 and concurrency-token semantics. Replace the single Status index with (Status, CreatedAt) and (Status, LeaseExpiresAt). Add DbSet<WorkflowTimerJob> WorkflowTimerJobs to WorkflowDbContext.

  • Step 5: Add validated options type

Create WorkflowJobOptions.cs with SectionName = "Workflow:Jobs", nested TimerJobSection and ImportJobSection, and the exact defaults from the configuration section above. Keep one top-level type; nested option-section classes are allowed by the repository file rule.

  • Step 6: Run the model tests and workflow build
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowJobModelTests
dotnet build src/modules/workflow/Application/Workflow.Application.csproj --no-restore

Expected: model tests pass; build has 0 warnings and 0 errors.

  • Step 7: Commit the model
git add src/modules/workflow/Domain/Enums/WorkflowTimerJobStatus.cs src/modules/workflow/Domain/Entities/WorkflowTimerJob.cs src/modules/workflow/Infrastructure/Persistence/Configurations/WorkflowTimerJobConfiguration.cs src/modules/workflow/Application/Reliability/WorkflowJobOptions.cs src/modules/workflow/Domain/Entities/BuildFlowImportJob.cs src/modules/workflow/Infrastructure/Persistence/Configurations/BuildFlowImportJobConfiguration.cs src/modules/workflow/Infrastructure/Persistence/WorkflowDbContext.cs tests/BuildFlow.Workflow.Tests/Reliability/WorkflowJobModelTests.cs
git commit -m "feat: add workflow job lease model"

Task 2: Persist Future Timer Nodes Instead Of Using Task.Run

Files:

  • Modify: src/modules/workflow/Application/Engine/ProcessEngine.cs:1005-1135

  • Modify: tests/BuildFlow.Workflow.Tests/Engine/ProcessEngineTests.cs

  • Create: tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs

  • Step 1: Write failing Timer persistence tests

Add these tests to ProcessEngineTests:

[Fact]
public async Task ProcessTimerNode_WithFutureDueAt_PersistsSinglePendingJob()
{
    var timerNode = CreateNode(NodeType.Timer);
    timerNode.Config = """{ "delayMinutes": 10 }""";
    var nextNode = CreateNode(NodeType.Approval);
    var (_, instance) = PersistSetup([timerNode, nextNode], [CreateEdge(timerNode, nextNode)]);
    instance.Status = InstanceStatus.Running;
    var token = PersistToken(instance, timerNode);
    var variablesBefore = instance.Variables;

    await _engine.ProcessNodeAsync(instance, token, timerNode);

    var job = await _dbContext.WorkflowTimerJobs.SingleAsync();
    job.InstanceId.Should().Be(instance.Id);
    job.TokenId.Should().Be(token.Id);
    job.NodeId.Should().Be(timerNode.Id);
    job.Status.Should().Be(WorkflowTimerJobStatus.Pending);
    job.DueAt.Should().BeAfter(DateTime.UtcNow.AddMinutes(9));
    token.Status.Should().Be(TokenStatus.Active);
    instance.Variables.Should().Be(variablesBefore);
}

[Fact]
public async Task ProcessTimerNode_ReenteredForSameToken_DoesNotCreateDuplicateJob()
{
    var timerNode = CreateNode(NodeType.Timer);
    timerNode.Config = """{ "delayMinutes": 10 }""";
    var (_, instance) = PersistSetup([timerNode], []);
    instance.Status = InstanceStatus.Running;
    var token = PersistToken(instance, timerNode);

    await _engine.ProcessNodeAsync(instance, token, timerNode);
    await _engine.ProcessNodeAsync(instance, token, timerNode);

    (await _dbContext.WorkflowTimerJobs.CountAsync()).Should().Be(1);
}

Create an architecture test that reads ProcessEngine.cs and asserts Task.Run and timer:{ are absent.

  • Step 2: Run and verify RED
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter "FullyQualifiedName~ProcessTimerNode_"
dotnet test tests/BuildFlow.Architecture.Tests/BuildFlow.Architecture.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowJobArchitectureTests

Expected: behavior tests fail because no Timer job is stored; architecture test fails on Task.Run and the variables key.

  • Step 3: Replace the future-Timer branch

Keep the immediate branch unchanged. Replace the future branch with:

var existingJob = await _dbContext.WorkflowTimerJobs
    .FirstOrDefaultAsync(x => x.TokenId == token.Id);
if (existingJob is null)
{
    _dbContext.WorkflowTimerJobs.Add(new WorkflowTimerJob
    {
        Id = Guid.NewGuid(),
        InstanceId = instance.Id,
        TokenId = token.Id,
        NodeId = node.Id,
        DueAt = fireAt.Value,
        Status = WorkflowTimerJobStatus.Pending,
    });
    await _dbContext.SaveChangesAsync();
}

Delete the variables marker and the complete Task.Run block. Update the XML comment to state that future timers are resumed only by the persistent scheduler.

  • Step 4: Verify GREEN and commit
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter "FullyQualifiedName~ProcessTimerNode_"
dotnet test tests/BuildFlow.Architecture.Tests/BuildFlow.Architecture.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowJobArchitectureTests
git add src/modules/workflow/Application/Engine/ProcessEngine.cs tests/BuildFlow.Workflow.Tests/Engine/ProcessEngineTests.cs tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs
git commit -m "fix: persist workflow timer jobs"

Task 3: Add PostgreSQL Timer Claiming And Transactional Completion

Files:

  • Create: src/modules/workflow/Application/Scheduler/Commands/ProcessDueWorkflowTimersCommand.cs

  • Create: src/modules/workflow/Application/Scheduler/Handlers/ProcessDueWorkflowTimersCommandHandler.cs

  • Create: tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowFixture.cs

  • Create: tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowCollection.cs

  • Create: tests/BuildFlow.Workflow.Tests/Reliability/WorkflowTimerProcessorTests.cs

  • Step 1: Add the PostgreSQL fixture

The fixture must read BUILDFLOW_WORKFLOW_TEST_CONNECTION, defaulting to:

Host=127.0.0.1;Port=55432;Username=buildflow;Password=buildflow;Database=postgres

At initialization it creates a unique workflow_timer_test_<guid> database, builds a snake-case WorkflowDbContext, and calls EnsureCreatedAsync(). At disposal it clears Npgsql pools, terminates connections to that database, drops it, and leaves the shared PostgreSQL container running. Put the xUnit collection definition in its own file.

  • Step 2: Write failing concurrency and recovery tests

WorkflowTimerProcessorTests must use two independently created DbContexts and cover:

[Fact]
public async Task TwoWorkersClaimingOneDueTimerCreateOnlyOneDownstreamToken()

[Fact]
public async Task ExpiredProcessingLeaseCanBeRecoveredByAnotherWorker()

[Fact]
public async Task NonRunningInstanceCancelsTimerWithoutCreatingDownstreamToken()

[Fact]
public async Task ExpiredLeaseAtMaxAttemptsBecomesFailed()

Seed a relation-table graph containing Timer -> Approval, one running instance, one active source token, and one due Timer job. Run two handlers concurrently with owners worker-a and worker-b. Assert the sum of returned processed counts is one, the job is Completed, the source token is Consumed, and exactly one active downstream token exists. Recovery must increment AttemptCount; cancellation must clear lease fields; exhausted work must preserve a Chinese LastError and create no token.

  • Step 3: Run and verify RED

First check whether the isolated PostgreSQL port is already available:

docker ps --filter publish=55432 --format '{{.Names}} {{.Status}}'

If there is no healthy PostgreSQL container on that port, start the dedicated test instance:

docker run -d --rm --name buildflow-workflow-test-postgres -p 55432:5432 -e POSTGRES_USER=buildflow -e POSTGRES_PASSWORD=buildflow -e POSTGRES_DB=postgres pgvector/pgvector:0.8.0-pg17

Then run:

dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowTimerProcessorTests

Expected: compilation fails because the command and handler do not exist.

  • Step 4: Add the command contract
using MediatR;

namespace Workflow.Application.Scheduler.Commands;

/// <summary>
/// 原子领取并处理一批到期工作流 Timer 的命令。
/// </summary>
public sealed record ProcessDueWorkflowTimersCommand(
    string LeaseOwner,
    int BatchSize,
    TimeSpan LeaseDuration,
    int MaxAttempts,
    TimeSpan RetryDelay) : IRequest<int>;
  • Step 5: Implement atomic claim

The handler must start a short transaction, first mark expired Processing rows whose AttemptCount >= MaxAttempts as Failed, then execute this claim shape with Npgsql parameters and return claimed IDs:

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;

The claim transaction must commit before processing starts so a process crash leaves a recoverable lease.

  • Step 6: Implement each Timer completion transaction

For every claimed ID:

  1. Begin a new transaction.
  2. Load the Timer job with SELECT * FROM b_workflow_timer_jobs WHERE id = @id FOR UPDATE.
  3. Require Status == Processing and matching LeaseOwner; otherwise roll back and skip.
  4. Reload instance and token.
  5. If instance is not Running or token is not Active, set Cancelled, clear lease, set CompletedAt, save, and commit.
  6. Load the graph through ProcessEngine.EnsureGraphLoadedForAsync.
  7. Consume the source token, create at most one downstream token, set the Timer job to Completed, clear lease, and set CompletedAt.
  8. Call ProcessEngine.ProcessNodeAsync for the downstream node before committing so token propagation and Timer completion share the transaction.
  9. On a non-cancellation exception, roll back, clear the change tracker, then conditionally update the still-owned job. If attempts are exhausted set Failed; otherwise set Pending, move DueAt to UtcNow + RetryDelay, clear lease, and store LastError truncated to 2000 characters.
  10. On cancellation, leave the committed claim untouched so lease expiry provides crash-safe recovery.
  • Step 7: Verify Timer concurrency and commit
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowTimerProcessorTests
git add src/modules/workflow/Application/Scheduler/Commands/ProcessDueWorkflowTimersCommand.cs src/modules/workflow/Application/Scheduler/Handlers/ProcessDueWorkflowTimersCommandHandler.cs tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowFixture.cs tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowCollection.cs tests/BuildFlow.Workflow.Tests/Reliability/WorkflowTimerProcessorTests.cs
git commit -m "feat: process workflow timers with leases"

Task 4: Add The Timer Host Scheduler Adapter

Files:

  • Create: src/BuildFlow.Host/Services/BackgroundServices/WorkflowTimerSchedulerService.cs

  • Modify: src/BuildFlow.Host/BuildFlowHostModule.cs

  • Modify: src/BuildFlow.Host/appsettings.json

  • Modify: tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs

  • Step 1: Extend the failing architecture test

Assert WorkflowTimerSchedulerService.cs contains IMediator, mediator.Send, and ProcessDueWorkflowTimersCommand, and does not contain WorkflowDbContext or ProcessEngine.

  • Step 2: Run and verify RED
dotnet test tests/BuildFlow.Architecture.Tests/BuildFlow.Architecture.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowJobArchitectureTests

Expected: failure because the service file and registration do not exist.

  • Step 3: Implement the Host adapter

WorkflowTimerSchedulerService must:

  • create one owner string in the constructor: timer:{Environment.MachineName}:{Environment.ProcessId}:{Guid.NewGuid():N};
  • wait StartupDelaySeconds;
  • use PeriodicTimer with PollIntervalSeconds;
  • create one DI scope per scan;
  • resolve IMediator only;
  • send ProcessDueWorkflowTimersCommand using the configured batch, lease, attempts, and retry delay;
  • log only owner, claimed count, and timing metadata;
  • catch normal stop cancellation without stopping the Host.

Register validated WorkflowJobOptions, register the hosted service, and add the exact Workflow:Jobs JSON shown at the start of this plan.

  • Step 4: Verify and commit
dotnet test tests/BuildFlow.Architecture.Tests/BuildFlow.Architecture.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowJobArchitectureTests
dotnet build src/BuildFlow.Host/BuildFlow.Host.csproj --no-restore
git add src/BuildFlow.Host/Services/BackgroundServices/WorkflowTimerSchedulerService.cs src/BuildFlow.Host/BuildFlowHostModule.cs src/BuildFlow.Host/appsettings.json tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs
git commit -m "feat: schedule persistent workflow timers"

Task 5: Atomically Claim Import Jobs Through MediatR

Files:

  • Create: src/modules/workflow/Application/BuildFlow/Imports/Commands/ProcessAvailableImportJobsCommand.cs

  • Create: src/modules/workflow/Application/BuildFlow/Imports/Handlers/ProcessAvailableImportJobsCommandHandler.cs

  • Modify: src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs

  • Create: tests/BuildFlow.Workflow.Tests/Reliability/ImportJobClaimTests.cs

  • Step 1: Write failing PostgreSQL claim tests

Cover these behaviors with two DbContexts and a recording IImportJobProcessor stub:

[Fact]
public async Task TwoWorkersCanClaimPendingImportOnlyOnce()

[Fact]
public async Task ExpiredRunningImportCanBeReclaimed()

[Fact]
public async Task ExpiredImportAtMaxAttemptsBecomesFailed()

The processor stub records (jobId, leaseOwner, leaseDuration). Assert one callback for one Pending job, owner replacement plus attempt increment for an expired Running job, and no callback for exhausted work.

  • Step 2: Run and verify RED
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~ImportJobClaimTests

Expected: compilation fails because the command and handler do not exist and the processor signature has no lease parameters.

  • Step 3: Add the command and processor contract
using MediatR;

namespace Workflow.Application.BuildFlow.Imports.Commands;

/// <summary>
/// 原子领取并处理可执行导入任务的命令。
/// </summary>
public sealed record ProcessAvailableImportJobsCommand(
    string LeaseOwner,
    int BatchSize,
    TimeSpan LeaseDuration,
    int MaxAttempts) : IRequest<int>;

Change IImportJobProcessor to:

Task ProcessAsync(
    Guid jobId,
    string leaseOwner,
    TimeSpan leaseDuration,
    CancellationToken ct);
  • Step 4: Implement import claim SQL

Use the same short-transaction pattern as Timer. Candidates are:

WHERE (
    status = @pending
    OR (status = @running AND lease_expires_at <= @now)
)
AND attempt_count < @max_attempts
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT @batch_size

The update must set Running, owner, expiry, increment attempts, set LastHeartbeatAt, preserve the first StartedAt with COALESCE, and return IDs. Before claiming, expired Running jobs already at max attempts must become Failed, receive FinishedAt, clear lease, and receive the Chinese error 任务租约过期且已达到最大重试次数.

After the claim transaction commits, invoke IImportJobProcessor.ProcessAsync once per claimed ID. One job failure must be logged by the Application handler through ILogger and must not prevent processing other claimed IDs.

  • Step 5: Verify and commit
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~ImportJobClaimTests
git add src/modules/workflow/Application/BuildFlow/Imports/Commands/ProcessAvailableImportJobsCommand.cs src/modules/workflow/Application/BuildFlow/Imports/Handlers/ProcessAvailableImportJobsCommandHandler.cs src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs tests/BuildFlow.Workflow.Tests/Reliability/ImportJobClaimTests.cs
git commit -m "feat: atomically claim import jobs"

Task 6: Make Import Progress Transactional And Recoverable

Files:

  • Modify: src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs

  • Create: tests/BuildFlow.Workflow.Tests/Reliability/ImportJobRecoveryTests.cs

  • Step 1: Write failing recovery tests

Use real PostgreSQL, a temporary file, a real enabled FormDefinition, and a stub IFormSchemaExcelService. Cover:

[Fact]
public async Task RecoveredImportStartsAtCommittedCursorWithoutDuplicatingRows()

[Fact]
public async Task SuccessfulImportClearsLeaseAndPersistsHeartbeat()

[Fact]
public async Task GracefulCancellationRequeuesJobAndKeepsCommittedProgress()

For recovery, seed a Running job with SuccessRows = 1, one existing FormData row, and two parsed Excel rows. Assert only the second row is submitted, final FormData count is two, final success count is two, and the lease is cleared.

  • Step 2: Run and verify RED
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~ImportJobRecoveryTests

Expected: failures because the current processor skips Running jobs and does not support a lease owner or committed cursor.

  • Step 3: Refactor processing around the claimed lease

The first job query must require all of:

x.Id == jobId
&& x.Status == BuildFlowImportJobStatus.Running
&& x.LeaseOwner == leaseOwner

Do not change status to Running in the processor. Parse the file, set TotalRows, and set nextRowIndex = job.SuccessRows + job.FailedRows. Reject a cursor greater than parsed row count as a task-level failure.

For each row from nextRowIndex:

  1. Begin a PostgreSQL transaction.
  2. Reload the job with FOR UPDATE.
  3. Verify it is still Running and owned by leaseOwner.
  4. For invalid Excel rows, increment FailedRows, append ImportRowError, extend lease, update heartbeat, save, and commit.
  5. For valid rows, call the existing SubmitFormDataCommandHandler inside the same transaction, increment SuccessRows, extend lease, update heartbeat, save, and commit.
  6. If row submission throws, roll back that row, begin a new transaction, lock and revalidate the job, record the row error and failed count, update heartbeat, and commit.

Because FormData and the row cursor share one transaction, a hard crash leaves either both committed or both absent. This is the required no-duplicate recovery guarantee.

After the final row, lock the job one last time, calculate Success/Failed/PartialSuccess, set FinishedAt, clear LeaseOwner and LeaseExpiresAt, persist ErrorJson, and commit. Task-level exceptions set Failed and clear lease. Graceful cancellation uses a best-effort non-cancelled cleanup update that returns the still-owned job to Pending without changing committed counts.

  • Step 4: Verify recovery and commit
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~ImportJobRecoveryTests
git add src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs tests/BuildFlow.Workflow.Tests/Reliability/ImportJobRecoveryTests.cs
git commit -m "fix: make import jobs recoverable"

Task 7: Reduce Import Host Service To A MediatR Adapter

Files:

  • Modify: src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs

  • Modify: src/BuildFlow.Host/BuildFlowHostModule.cs

  • Modify: tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs

  • Step 1: Add the failing architecture assertions

Assert ImportJobProcessorService.cs contains IMediator, mediator.Send, and ProcessAvailableImportJobsCommand, and does not contain WorkflowDbContext, IImportJobProcessor, BuildFlowImportJobStatus, or a .Where( scan.

  • Step 2: Run and verify RED
dotnet test tests/BuildFlow.Architecture.Tests/BuildFlow.Architecture.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowJobArchitectureTests

Expected: failure because Host still queries Pending IDs directly.

  • Step 3: Replace the Host scan

Keep the BackgroundService lifecycle and cancellation handling, but replace RunScanAsync with a scope that resolves IMediator and sends:

new ProcessAvailableImportJobsCommand(
    _leaseOwner,
    options.Import.BatchSize,
    TimeSpan.FromSeconds(options.Import.LeaseSeconds),
    options.Import.MaxAttempts)

Use one constructor-created owner string: import:{Environment.MachineName}:{Environment.ProcessId}:{Guid.NewGuid():N}. Use the dedicated import interval/startup delay instead of NotificationOptions.

  • Step 4: Verify and commit
dotnet test tests/BuildFlow.Architecture.Tests/BuildFlow.Architecture.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowJobArchitectureTests
dotnet build src/BuildFlow.Host/BuildFlow.Host.csproj --no-restore
git add src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs src/BuildFlow.Host/BuildFlowHostModule.cs tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs
git commit -m "refactor: dispatch import jobs through MediatR"

Task 8: Generate And Verify The Workflow Reliability Migration

Files:

  • Create: generated src/modules/workflow/Infrastructure/Persistence/Migrations/*_AddWorkflowJobReliability.cs

  • Create: generated src/modules/workflow/Infrastructure/Persistence/Migrations/*_AddWorkflowJobReliability.Designer.cs

  • Modify: src/modules/workflow/Infrastructure/Persistence/Migrations/WorkflowDbContextModelSnapshot.cs

  • Create: tests/BuildFlow.Workflow.Tests/Reliability/WorkflowJobMigrationTests.cs

  • Step 1: Generate the migration

dotnet ef migrations add AddWorkflowJobReliability --project src/modules/workflow/Infrastructure --startup-project src/BuildFlow.Host --context WorkflowDbContext
  • Step 2: Inspect generated operations before testing

The migration must contain:

  • CreateTable("b_workflow_timer_jobs");
  • four AddColumn operations on b_import_jobs;
  • Timer unique/composite indexes;
  • import composite indexes;
  • no DropTable, no unrelated DropColumn, and no table rebuild.

Run:

rg -n "CreateTable|AddColumn|CreateIndex|DropTable|DropColumn" src/modules/workflow/Infrastructure/Persistence/Migrations/*AddWorkflowJobReliability*.cs
  • Step 3: Add empty-database and upgrade tests

WorkflowJobMigrationTests must create two unique PostgreSQL databases:

  1. Empty path: migrate directly to latest and assert b_workflow_timer_jobs plus all lease columns exist.
  2. Upgrade path: migrate to 20260719041812_FixWorkflowTaskAttachmentsColumn, insert one Pending row into b_import_jobs, migrate to latest, and assert the original ID/status/file path remain while attempt_count = 0 and lease columns are null.

Use EF IMigrator for target migration selection. Drop both databases in finally.

  • Step 4: Run migration tests and commit
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowJobMigrationTests
git add src/modules/workflow/Infrastructure/Persistence/Migrations tests/BuildFlow.Workflow.Tests/Reliability/WorkflowJobMigrationTests.cs
git commit -m "feat: migrate workflow job reliability state"

Task 9: Run The Second-Batch Verification Gate

Files:

  • Verify only.

  • Step 1: Run focused reliability suites

dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter "FullyQualifiedName~Reliability|FullyQualifiedName~ProcessTimerNode_"
dotnet test tests/BuildFlow.Architecture.Tests/BuildFlow.Architecture.Tests.csproj --no-restore --filter FullyQualifiedName~WorkflowJobArchitectureTests
  • Step 2: Run the complete Workflow and Architecture projects
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore
dotnet test tests/BuildFlow.Architecture.Tests/BuildFlow.Architecture.Tests.csproj --no-restore
  • Step 3: Run full solution tests and build
dotnet test BuildFlow.slnx --no-restore
dotnet build BuildFlow.slnx --no-restore

Expected: every project passes, zero skipped, build has 0 warnings and 0 errors.

  • Step 4: Run mandatory static scans
rg -n "Task\.Run|timer:\{" src/modules/workflow/Application/Engine/ProcessEngine.cs
rg -n "WorkflowDbContext|ProcessEngine" src/BuildFlow.Host/Services/BackgroundServices/WorkflowTimerSchedulerService.cs
rg -n "WorkflowDbContext|IImportJobProcessor|BuildFlowImportJobStatus|\.Where\(" src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs
rg -n "FindAsync\(\[jobId\].*Status.*Pending|Status == BuildFlowImportJobStatus.Pending" src/BuildFlow.Host src/modules/workflow/Application/BuildFlow/Imports
git diff --check
git status --short

Expected: all rg scans have no output, git diff --check has no output, and no second-batch files remain uncommitted.

  • Step 5: Record batch evidence

Report:

  • exact pass/fail/skip totals;
  • Timer two-worker claim result;
  • Timer lease recovery and max-attempt result;
  • import two-worker claim result;
  • import committed-cursor recovery row counts;
  • empty and upgrade migration results;
  • build warning/error totals;
  • static scan results;
  • all second-batch commit hashes.

Do not begin the third-batch database query/performance plan until every command above is green.

Self-Review Results

  • Spec coverage: Timer model, persistent scheduling, atomic claim, lease recovery, max attempts, cancellation, import claim, heartbeat, and no-duplicate recovery all have explicit tasks and tests.
  • Scope boundary: per-row import transactions intentionally remain; bounded 200-row batching, streaming Excel, full deterministic hash, and aggregate-refresh optimization remain third-batch work.
  • Type consistency: Host commands, processor signatures, status names, option names, and configuration paths are consistent across tasks.
  • Placeholder scan: passed; every implementation step names its concrete types and behavior.