build-flow/build-flow.api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: atomically claim import jobs

Browse Source
This commit is contained in:
xiangning 2026-07-23 18:52:02 +08:00
parent 2684c2d5c4
commit e515136c11
5 changed files with 316 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/modules/workflow/Application/BuildFlow/Imports/Commands/ProcessAvailableImportJobsCommand.cs Normal file
View File

@ -0,0 +1,12 @@
using MediatR;
namespace Workflow.Application.BuildFlow.Imports.Commands;
/// <summary>
/// 原子领取并处理可执行导入任务的命令。
/// </summary>
public sealed record ProcessAvailableImportJobsCommand(
string LeaseOwner,
int BatchSize,
TimeSpan LeaseDuration,
int MaxAttempts) : IRequest<int>;

130
src/modules/workflow/Application/BuildFlow/Imports/Handlers/ProcessAvailableImportJobsCommandHandler.cs Normal file
View File

@ -0,0 +1,130 @@
using System.Data.Common;
using MediatR;
using Microsoft.EntityFrameworkCore.Storage;
using Microsoft.Extensions.Logging;
using Workflow.Application.BuildFlow.Imports.Commands;
using Workflow.Application.BuildFlow.Imports.Services;
using Workflow.Domain.Enums;
using Workflow.Infrastructure.Persistence;
namespace Workflow.Application.BuildFlow.Imports.Handlers;
/// <summary>
/// 使用 PostgreSQL 租约原子领取导入任务，并逐个交给模块内处理器执行。
/// </summary>
public sealed class ProcessAvailableImportJobsCommandHandler(
WorkflowDbContext db,
IImportJobProcessor processor,
ILogger<ProcessAvailableImportJobsCommandHandler> logger)
: IRequestHandler<ProcessAvailableImportJobsCommand, int>
{
public async Task<int> Handle(
ProcessAvailableImportJobsCommand request,
CancellationToken ct)
{
var claimedIds = await ClaimAsync(request, ct);
foreach (var jobId in claimedIds)
{
ct.ThrowIfCancellationRequested();
try
{
await processor.ProcessAsync(
jobId,
request.LeaseOwner,
request.LeaseDuration,
ct);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
throw;
}
catch (Exception ex)
{
logger.LogError(ex, "导入任务 {JobId} 处理失败，等待租约恢复", jobId);
}
}
return claimedIds.Count;
}
private async Task<List<Guid>> ClaimAsync(
ProcessAvailableImportJobsCommand 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_import_jobs
SET status = {(int)BuildFlowImportJobStatus.Failed},
lease_owner = NULL,
lease_expires_at = NULL,
finished_at = {now},
error_message = {"任务租约过期且已达到最大重试次数"},
updated_by = {"system"},
updated_at = {now}
WHERE status = {(int)BuildFlowImportJobStatus.Running}
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_import_jobs
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
)
UPDATE b_import_jobs AS job
SET status = @running,
lease_owner = @lease_owner,
lease_expires_at = @lease_expires_at,
attempt_count = job.attempt_count + 1,
last_heartbeat_at = @now,
started_at = COALESCE(job.started_at, @now),
finished_at = NULL,
error_message = NULL,
updated_by = 'system',
updated_at = @now
FROM candidates
WHERE job.id = candidates.id
RETURNING job.id
""";
AddParameter(command, "pending", (int)BuildFlowImportJobStatus.Pending);
AddParameter(command, "running", (int)BuildFlowImportJobStatus.Running);
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<Guid>();
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 static void AddParameter(DbCommand command, string name, object value)
{
var parameter = command.CreateParameter();
parameter.ParameterName = name;
parameter.Value = value;
command.Parameters.Add(parameter);
}
}

14
src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs
View File

@ -1,8 +1,16 @@
namespace Workflow.Application.BuildFlow.Imports.Services; namespace Workflow.Application.BuildFlow.Imports.Services;
/// <summary>导入任务核心处理器。由后台服务（ImportJobProcessorService）调用。</summary> /// <summary>导入任务核心处理器。仅处理已由 Application 命令原子领取的任务。</summary>
public interface IImportJobProcessor public interface IImportJobProcessor
{ {
/// <summary>处理一个 Pending 导入任务。处理器内部负责状态流转（Running → Success/Failed/PartialSuccess）。</summary> /// <summary>处理已持有指定租约的 Running 导入任务。</summary>
Task ProcessAsync(Guid jobId, CancellationToken ct); Task ProcessAsync(
Guid jobId,
string leaseOwner,
TimeSpan leaseDuration,
CancellationToken ct);
/// <summary>旧 Host 扫描器的临时兼容入口，Host 改为 MediatR 后删除。</summary>
Task ProcessAsync(Guid jobId, CancellationToken ct)
=> ProcessAsync(jobId, "legacy-import-host", TimeSpan.FromMinutes(5), ct);
} }

6
src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs
View File

@ -24,7 +24,11 @@ public class ImportJobProcessor(
WriteIndented = false, WriteIndented = false,
}; };
public async Task ProcessAsync(Guid jobId, CancellationToken ct) public async Task ProcessAsync(
Guid jobId,
string leaseOwner,
TimeSpan leaseDuration,
CancellationToken ct)
{ {
var job = await db.BuildFlowImportJobs.FindAsync([jobId], ct); var job = await db.BuildFlowImportJobs.FindAsync([jobId], ct);
if (job is null) if (job is null)

158
tests/BuildFlow.Workflow.Tests/Reliability/ImportJobClaimTests.cs Normal file
View File

@ -0,0 +1,158 @@
using System.Collections.Concurrent;
using FluentAssertions;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging.Abstractions;
using Workflow.Application.BuildFlow.Imports.Commands;
using Workflow.Application.BuildFlow.Imports.Handlers;
using Workflow.Application.BuildFlow.Imports.Services;
using Workflow.Domain.Entities;
using Workflow.Domain.Enums;
using Xunit;
namespace BuildFlow.Workflow.Tests.Reliability;
/// <summary>
/// 导入任务的 PostgreSQL 原子领取和过期租约恢复测试。
/// </summary>
[Collection(PostgresWorkflowCollection.Name)]
public class ImportJobClaimTests(PostgresWorkflowFixture fixture)
{
[Fact]
public async Task TwoWorkersCanClaimPendingImportOnlyOnce()
{
await ResetAsync();
var jobId = await SeedImportAsync(BuildFlowImportJobStatus.Pending);
var processor = new RecordingImportJobProcessor();
await using var dbA = fixture.CreateDbContext();
await using var dbB = fixture.CreateDbContext();
var handlerA = CreateHandler(dbA, processor);
var handlerB = CreateHandler(dbB, processor);
var results = await Task.WhenAll(
handlerA.Handle(CreateCommand("worker-a"), CancellationToken.None),
handlerB.Handle(CreateCommand("worker-b"), CancellationToken.None));
results.Sum().Should().Be(1);
processor.Calls.Should().ContainSingle(x => x.JobId == jobId);
await using var verify = fixture.CreateDbContext();
var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == jobId);
job.Status.Should().Be(BuildFlowImportJobStatus.Running);
job.AttemptCount.Should().Be(1);
job.LeaseOwner.Should().BeOneOf("worker-a", "worker-b");
job.LastHeartbeatAt.Should().NotBeNull();
}
[Fact]
public async Task ExpiredRunningImportCanBeReclaimed()
{
await ResetAsync();
var jobId = await SeedImportAsync(
BuildFlowImportJobStatus.Running,
"dead-worker",
DateTime.UtcNow.AddMinutes(-1),
attemptCount: 1);
var processor = new RecordingImportJobProcessor();
await using var db = fixture.CreateDbContext();
var handler = CreateHandler(db, processor);
var claimed = await handler.Handle(CreateCommand("recovery-worker"), CancellationToken.None);
claimed.Should().Be(1);
processor.Calls.Should().ContainSingle(x =>
x.JobId == jobId && x.LeaseOwner == "recovery-worker");
await using var verify = fixture.CreateDbContext();
var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == jobId);
job.AttemptCount.Should().Be(2);
job.LeaseOwner.Should().Be("recovery-worker");
job.LastHeartbeatAt.Should().NotBeNull();
}
[Fact]
public async Task ExpiredImportAtMaxAttemptsBecomesFailed()
{
await ResetAsync();
var jobId = await SeedImportAsync(
BuildFlowImportJobStatus.Running,
"dead-worker",
DateTime.UtcNow.AddMinutes(-1),
attemptCount: 5);
var processor = new RecordingImportJobProcessor();
await using var db = fixture.CreateDbContext();
var handler = CreateHandler(db, processor);
var claimed = await handler.Handle(CreateCommand("worker-a"), CancellationToken.None);
claimed.Should().Be(0);
processor.Calls.Should().BeEmpty();
await using var verify = fixture.CreateDbContext();
var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == jobId);
job.Status.Should().Be(BuildFlowImportJobStatus.Failed);
job.LeaseOwner.Should().BeNull();
job.LeaseExpiresAt.Should().BeNull();
job.FinishedAt.Should().NotBeNull();
job.ErrorMessage.Should().Contain("最大重试次数");
}
private static ProcessAvailableImportJobsCommand CreateCommand(string owner) => new(
owner,
BatchSize: 10,
LeaseDuration: TimeSpan.FromMinutes(5),
MaxAttempts: 5);
private static ProcessAvailableImportJobsCommandHandler CreateHandler(
WorkflowDbContext db,
IImportJobProcessor processor)
=> new(
db,
processor,
NullLogger<ProcessAvailableImportJobsCommandHandler>.Instance);
private async Task ResetAsync()
{
await using var db = fixture.CreateDbContext();
await db.BuildFlowImportJobs.ExecuteDeleteAsync();
}
private async Task<Guid> SeedImportAsync(
BuildFlowImportJobStatus status,
string? leaseOwner = null,
DateTime? leaseExpiresAt = null,
int attemptCount = 0)
{
await using var db = fixture.CreateDbContext();
var job = new BuildFlowImportJob
{
Id = Guid.NewGuid(),
FormDefinitionId = Guid.NewGuid(),
Status = status,
FileUrl = "unused.xlsx",
OriginalFileName = "unused.xlsx",
LeaseOwner = leaseOwner,
LeaseExpiresAt = leaseExpiresAt,
AttemptCount = attemptCount,
};
db.BuildFlowImportJobs.Add(job);
await db.SaveChangesAsync();
return job.Id;
}
private sealed class RecordingImportJobProcessor : IImportJobProcessor
{
public ConcurrentQueue<ImportJobCall> Calls { get; } = new();
public Task ProcessAsync(
Guid jobId,
string leaseOwner,
TimeSpan leaseDuration,
CancellationToken ct)
{
Calls.Enqueue(new ImportJobCall(jobId, leaseOwner, leaseDuration));
return Task.CompletedTask;
}
}
private sealed record ImportJobCall(
Guid JobId,
string LeaseOwner,
TimeSpan LeaseDuration);
}