From 65566dc112c77e665ce6d5daf1113b67b91d02e0 Mon Sep 17 00:00:00 2001 From: xiangning Date: Fri, 24 Jul 2026 09:10:49 +0800 Subject: [PATCH] perf: persist imports in bounded batches --- src/BuildFlow.Host/BuildFlowHostModule.cs | 5 +- .../ImportJobProcessorService.cs | 3 +- src/BuildFlow.Host/appsettings.json | 3 +- .../ProcessAvailableImportJobsCommand.cs | 3 +- ...rocessAvailableImportJobsCommandHandler.cs | 1 + .../Imports/Services/IImportJobProcessor.cs | 1 + .../Imports/Services/ImportJobProcessor.cs | 217 +++++++----------- .../Reliability/WorkflowJobOptions.cs | 3 + .../Performance/CancellingExcelRows.cs | 22 ++ .../CountingTransactionInterceptor.cs | 25 ++ .../EnumerableFormSchemaExcelService.cs | 17 ++ .../Performance/ImportJobBatchingTests.cs | 203 ++++++++++++++++ .../Reliability/ImportJobClaimTests.cs | 9 +- .../Reliability/ImportJobRecoveryTests.cs | 29 ++- 14 files changed, 391 insertions(+), 150 deletions(-) create mode 100644 tests/BuildFlow.Workflow.Tests/Performance/CancellingExcelRows.cs create mode 100644 tests/BuildFlow.Workflow.Tests/Performance/CountingTransactionInterceptor.cs create mode 100644 tests/BuildFlow.Workflow.Tests/Performance/EnumerableFormSchemaExcelService.cs create mode 100644 tests/BuildFlow.Workflow.Tests/Performance/ImportJobBatchingTests.cs diff --git a/src/BuildFlow.Host/BuildFlowHostModule.cs b/src/BuildFlow.Host/BuildFlowHostModule.cs index 96bd843..20f8697 100644 --- a/src/BuildFlow.Host/BuildFlowHostModule.cs +++ b/src/BuildFlow.Host/BuildFlowHostModule.cs @@ -117,8 +117,9 @@ public class BuildFlowHostModule : AbpModule && options.Import.StartupDelaySeconds > 0 && options.Import.BatchSize > 0 && options.Import.LeaseSeconds > 0 - && options.Import.MaxAttempts > 0, - "Workflow:Jobs 的间隔、批量、租约、重试延迟和最大尝试次数必须为正数。") + && options.Import.MaxAttempts > 0 + && options.Import.RowsPerBatch is > 0 and <= 1000, + "Workflow:Jobs 的间隔、批量、租约、重试延迟和最大尝试次数必须为正数，导入事务行数不得超过 1000。") .ValidateOnStart(); services.AddNotifications(config); services.AddHostedService(); diff --git a/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs b/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs index 278ccaf..03a8579 100644 --- a/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs +++ b/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs @@ -68,7 +68,8 @@ public sealed class ImportJobProcessorService( _leaseOwner, importOptions.BatchSize, TimeSpan.FromSeconds(importOptions.LeaseSeconds), - importOptions.MaxAttempts), + importOptions.MaxAttempts, + importOptions.RowsPerBatch), ct); if (processed > 0) diff --git a/src/BuildFlow.Host/appsettings.json b/src/BuildFlow.Host/appsettings.json index 13ee571..a054206 100644 --- a/src/BuildFlow.Host/appsettings.json +++ b/src/BuildFlow.Host/appsettings.json @@ -107,7 +107,8 @@ "StartupDelaySeconds": 5, "BatchSize": 5, "LeaseSeconds": 300, - "MaxAttempts": 5 + "MaxAttempts": 5, + "RowsPerBatch": 200 } } }, diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Commands/ProcessAvailableImportJobsCommand.cs b/src/modules/workflow/Application/BuildFlow/Imports/Commands/ProcessAvailableImportJobsCommand.cs index e511c2b..083b5b8 100644 --- a/src/modules/workflow/Application/BuildFlow/Imports/Commands/ProcessAvailableImportJobsCommand.cs +++ b/src/modules/workflow/Application/BuildFlow/Imports/Commands/ProcessAvailableImportJobsCommand.cs @@ -9,4 +9,5 @@ public sealed record ProcessAvailableImportJobsCommand( string LeaseOwner, int BatchSize, TimeSpan LeaseDuration, - int MaxAttempts) : IRequest; + int MaxAttempts, + int RowsPerBatch) : IRequest; diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Handlers/ProcessAvailableImportJobsCommandHandler.cs b/src/modules/workflow/Application/BuildFlow/Imports/Handlers/ProcessAvailableImportJobsCommandHandler.cs index d80bb10..8baccf7 100644 --- a/src/modules/workflow/Application/BuildFlow/Imports/Handlers/ProcessAvailableImportJobsCommandHandler.cs +++ b/src/modules/workflow/Application/BuildFlow/Imports/Handlers/ProcessAvailableImportJobsCommandHandler.cs @@ -32,6 +32,7 @@ public sealed class ProcessAvailableImportJobsCommandHandler( jobId, request.LeaseOwner, request.LeaseDuration, + request.RowsPerBatch, ct); } catch (OperationCanceledException) when (ct.IsCancellationRequested) diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs b/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs index daffd74..84420f0 100644 --- a/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs +++ b/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs @@ -8,5 +8,6 @@ public interface IImportJobProcessor Guid jobId, string leaseOwner, TimeSpan leaseDuration, + int rowsPerBatch, CancellationToken ct); } diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs b/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs index e329c27..9725912 100644 --- a/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs +++ b/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs @@ -1,10 +1,7 @@ using System.Text.Json; using Workflow.Application.BuildFlow.Common.Services; using Workflow.Application.BuildFlow.Imports.Services; -using Workflow.Application.Form.Aggregates; -using Workflow.Application.Form.Derivation; -using Workflow.Application.Form.FormData.Commands; -using Workflow.Application.Form.FormData.Handlers; +using Workflow.Application.Form.Schema; using Workflow.Domain.Entities; using Workflow.Domain.Enums; using Workflow.Infrastructure.Persistence; @@ -12,14 +9,15 @@ using Workflow.Infrastructure.Persistence; namespace Workflow.Application.BuildFlow.Imports.Handlers; /// -/// 导入任务核心处理器。 -/// 流程：Pending → Running → 解析 xlsx → 逐行提交 FormData → 写回 Success/Failed/PartialSuccess + ErrorJson。 -/// 异常会被捕获并写回 ErrorMessage + Status=Failed，避免单个任务失败拖垮后台服务。 +/// 导入任务核心处理器。惰性枚举 Excel 行并按有界批次提交表单数据、进度和租约心跳。 /// public class ImportJobProcessor( WorkflowDbContext db, - IFormSchemaExcelService excelService) : IImportJobProcessor + IFormSchemaExcelService excelService, + ImportBatchWriter batchWriter) : IImportJobProcessor { + private const int MaxRowsPerBatch = 1000; + private static readonly JsonSerializerOptions JsonOptions = new() { WriteIndented = false, @@ -29,10 +27,18 @@ public class ImportJobProcessor( Guid jobId, string leaseOwner, TimeSpan leaseDuration, + int rowsPerBatch, CancellationToken ct) { try { + if (rowsPerBatch is <= 0 or > MaxRowsPerBatch) + { + throw new BusinessException( + ErrorCode.InvalidParam, + $"导入批次行数必须在 1 到 {MaxRowsPerBatch} 之间"); + } + var job = await db.BuildFlowImportJobs .AsNoTracking() .FirstOrDefaultAsync(x => @@ -40,99 +46,88 @@ public class ImportJobProcessor( && x.Status == BuildFlowImportJobStatus.Running && x.LeaseOwner == leaseOwner, ct); - if (job is null) return; + if (job is null) + { + return; + } var form = await db.FormDefinitions .AsNoTracking() - .Where(f => f.Id == job.FormDefinitionId) - .Select(f => new { f.SchemaJson }) + .Where(x => x.Id == job.FormDefinitionId) + .Select(x => new { x.SchemaJson }) .FirstOrDefaultAsync(ct); if (form is null) { - throw new NotFoundException(ErrorCode.NotFound, $"表单定义 {job.FormDefinitionId} 不存在"); + throw new NotFoundException( + ErrorCode.NotFound, + $"表单定义 {job.FormDefinitionId} 不存在"); } + + var schemaJson = form.SchemaJson + ?? """{"type":"object","properties":{}}"""; + if (!File.Exists(job.FileUrl)) { - throw new BusinessException(ErrorCode.Business, $"上传文件不存在：{job.FileUrl}"); + throw new BusinessException( + ErrorCode.Business, + $"上传文件不存在：{job.FileUrl}"); } + var preparedSchema = PreparedFormSchema.Create(schemaJson); var fileBytes = await File.ReadAllBytesAsync(job.FileUrl, ct); - var rows = excelService.Parse(form.SchemaJson ?? "{}", fileBytes); - var nextRowIndex = await InitializeProgressAsync( - jobId, - leaseOwner, - leaseDuration, - ct); - if (nextRowIndex is null) return; - - var submitHandler = new SubmitFormDataCommandHandler( - db, - new FormValueDerivationService([], [], db), - new FormAggregateRefreshService(db)); - + var committedCursor = job.SuccessRows + job.FailedRows; + var batch = new List(rowsPerBatch); var rowIndex = 0; - foreach (var row in rows) + + foreach (var row in excelService.Parse(schemaJson, fileBytes)) { ct.ThrowIfCancellationRequested(); - if (rowIndex < nextRowIndex.Value) + if (rowIndex >= committedCursor) { - rowIndex++; - continue; + batch.Add(new ImportBatchItem(rowIndex, row)); } - if (row.Errors.Count > 0) + rowIndex++; + if (batch.Count == rowsPerBatch) { - var recorded = await CommitFailedRowAsync( - jobId, - leaseOwner, - leaseDuration, - row.RowNumber, - row.Errors.Select(e => e.Message).ToList(), - rowIndex + 1, - ct); - if (!recorded) return; - rowIndex++; - continue; - } - - try - { - var committed = await CommitSuccessfulRowAsync( + var committed = await CommitBatchAsync( jobId, leaseOwner, leaseDuration, job.FormDefinitionId, - rowIndex, - row.JsonObject, - submitHandler, + preparedSchema, + batch, ct); - if (!committed) return; - } - catch (OperationCanceledException) when (ct.IsCancellationRequested) - { - throw; - } - catch (Exception ex) - { - var recorded = await CommitFailedRowAsync( - jobId, - leaseOwner, - leaseDuration, - row.RowNumber, - [ex.Message], - rowIndex + 1, - ct); - if (!recorded) return; - } + if (!committed) + { + return; + } - rowIndex++; + batch.Clear(); + } } - if (nextRowIndex.Value > rowIndex) + if (committedCursor > rowIndex) { throw new BusinessException( ErrorCode.Business, - $"导入任务进度游标 {nextRowIndex.Value} 超过总行数 {rowIndex}"); + $"导入任务进度游标 {committedCursor} 超过总行数 {rowIndex}"); + } + + if (batch.Count > 0) + { + var committed = await CommitBatchAsync( + jobId, + leaseOwner, + leaseDuration, + job.FormDefinitionId, + preparedSchema, + batch, + ct); + if (!committed) + { + return; + } } await CompleteJobAsync(jobId, leaseOwner, rowIndex, ct); @@ -148,35 +143,13 @@ public class ImportJobProcessor( } } - private async Task InitializeProgressAsync( - Guid jobId, - string leaseOwner, - TimeSpan leaseDuration, - CancellationToken ct) - { - db.ChangeTracker.Clear(); - await using var transaction = await db.Database.BeginTransactionAsync(ct); - var job = await LockOwnedJobAsync(jobId, leaseOwner, ct); - if (job is null) - { - await transaction.RollbackAsync(ct); - return null; - } - - RenewLease(job, leaseDuration); - await db.SaveChangesAsync(ct); - await transaction.CommitAsync(ct); - return job.SuccessRows + job.FailedRows; - } - - private async Task CommitSuccessfulRowAsync( + private async Task CommitBatchAsync( Guid jobId, string leaseOwner, TimeSpan leaseDuration, Guid formDefinitionId, - int rowIndex, - string dataJson, - SubmitFormDataCommandHandler submitHandler, + PreparedFormSchema schema, + IReadOnlyList batch, CancellationToken ct) { db.ChangeTracker.Clear(); @@ -188,41 +161,22 @@ public class ImportJobProcessor( return false; } - var instanceId = DeterministicInstanceId(jobId, rowIndex); - await submitHandler.Handle( - new SubmitFormDataCommand(formDefinitionId, instanceId, dataJson), + var result = await batchWriter.WriteAsync( + jobId, + formDefinitionId, + schema, + batch, ct); - job.SuccessRows++; - job.TotalRows = Math.Max(job.TotalRows, rowIndex + 1); - RenewLease(job, leaseDuration); - await db.SaveChangesAsync(ct); - await transaction.CommitAsync(ct); - return true; - } - - private async Task CommitFailedRowAsync( - Guid jobId, - string leaseOwner, - TimeSpan leaseDuration, - int rowNumber, - IReadOnlyList messages, - int processedRows, - CancellationToken ct) - { - db.ChangeTracker.Clear(); - await using var transaction = await db.Database.BeginTransactionAsync(ct); - var job = await LockOwnedJobAsync(jobId, leaseOwner, ct); - if (job is null) + job.SuccessRows += result.SuccessCount; + job.FailedRows += result.Errors.Count; + job.TotalRows = Math.Max(job.TotalRows, batch[^1].RowIndex + 1); + if (result.Errors.Count > 0) { - await transaction.RollbackAsync(ct); - return false; + var errors = DeserializeErrors(job.ErrorJson); + errors.AddRange(result.Errors); + job.ErrorJson = JsonSerializer.Serialize(errors, JsonOptions); } - var errors = DeserializeErrors(job.ErrorJson); - errors.Add(new ImportRowError(rowNumber, messages)); - job.FailedRows++; - job.TotalRows = Math.Max(job.TotalRows, processedRows); - job.ErrorJson = JsonSerializer.Serialize(errors, JsonOptions); RenewLease(job, leaseDuration); await db.SaveChangesAsync(ct); await transaction.CommitAsync(ct); @@ -329,13 +283,4 @@ public class ImportJobProcessor( private static string Truncate(string value, int maxLength) => value.Length <= maxLength ? value : value[..maxLength]; - - private static Guid DeterministicInstanceId(Guid jobId, int rowIndex) - { - // 组合 jobId 与 rowIndex 生成确定性 Guid，便于行级聚合。 - var bytes = jobId.ToByteArray(); - bytes[0] = (byte)(rowIndex & 0xFF); - bytes[1] = (byte)((rowIndex >> 8) & 0xFF); - return new Guid(bytes); - } } diff --git a/src/modules/workflow/Application/Reliability/WorkflowJobOptions.cs b/src/modules/workflow/Application/Reliability/WorkflowJobOptions.cs index f2148f5..e91bc2c 100644 --- a/src/modules/workflow/Application/Reliability/WorkflowJobOptions.cs +++ b/src/modules/workflow/Application/Reliability/WorkflowJobOptions.cs @@ -53,5 +53,8 @@ public class WorkflowJobOptions /// 最大领取尝试次数。 public int MaxAttempts { get; set; } = 5; + + /// 单个导入事务最多处理的 Excel 行数。 + public int RowsPerBatch { get; set; } = 200; } } diff --git a/tests/BuildFlow.Workflow.Tests/Performance/CancellingExcelRows.cs b/tests/BuildFlow.Workflow.Tests/Performance/CancellingExcelRows.cs new file mode 100644 index 0000000..83fea4c --- /dev/null +++ b/tests/BuildFlow.Workflow.Tests/Performance/CancellingExcelRows.cs @@ -0,0 +1,22 @@ +using Workflow.Application.BuildFlow.Common.Services; + +namespace BuildFlow.Workflow.Tests.Performance; + +/// 枚举到指定索引时触发取消的 Excel 行序列。 +public sealed class CancellingExcelRows( + IReadOnlyList rows, + int cancelAtIndex, + Action cancel) : IEnumerable +{ + public IEnumerator GetEnumerator() + { + for (var index = 0; index < rows.Count; index++) + { + if (index == cancelAtIndex) cancel(); + yield return rows[index]; + } + } + + System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() + => GetEnumerator(); +} diff --git a/tests/BuildFlow.Workflow.Tests/Performance/CountingTransactionInterceptor.cs b/tests/BuildFlow.Workflow.Tests/Performance/CountingTransactionInterceptor.cs new file mode 100644 index 0000000..b64bfdd --- /dev/null +++ b/tests/BuildFlow.Workflow.Tests/Performance/CountingTransactionInterceptor.cs @@ -0,0 +1,25 @@ +using System.Data.Common; +using Microsoft.EntityFrameworkCore.Diagnostics; + +namespace BuildFlow.Workflow.Tests.Performance; + +/// 统计 EF 显式事务提交次数的测试拦截器。 +public sealed class CountingTransactionInterceptor : DbTransactionInterceptor +{ + /// 已提交事务数量。 + public int CommitCount { get; private set; } + + public override void TransactionCommitted( + DbTransaction transaction, + TransactionEndEventData eventData) + => CommitCount++; + + public override Task TransactionCommittedAsync( + DbTransaction transaction, + TransactionEndEventData eventData, + CancellationToken cancellationToken = default) + { + CommitCount++; + return Task.CompletedTask; + } +} diff --git a/tests/BuildFlow.Workflow.Tests/Performance/EnumerableFormSchemaExcelService.cs b/tests/BuildFlow.Workflow.Tests/Performance/EnumerableFormSchemaExcelService.cs new file mode 100644 index 0000000..b8e6ce4 --- /dev/null +++ b/tests/BuildFlow.Workflow.Tests/Performance/EnumerableFormSchemaExcelService.cs @@ -0,0 +1,17 @@ +using Workflow.Application.BuildFlow.Common.Services; + +namespace BuildFlow.Workflow.Tests.Performance; + +/// 从给定枚举返回导入行的 Excel 服务测试替身。 +public sealed class EnumerableFormSchemaExcelService(IEnumerable rows) + : IFormSchemaExcelService +{ + public byte[] BuildTemplate(string schemaJson, string formName) => []; + + public byte[] BuildExport( + string schemaJson, + string formName, + IEnumerable dataJsonRows) => []; + + public IEnumerable Parse(string schemaJson, byte[] xlsxBytes) => rows; +} diff --git a/tests/BuildFlow.Workflow.Tests/Performance/ImportJobBatchingTests.cs b/tests/BuildFlow.Workflow.Tests/Performance/ImportJobBatchingTests.cs new file mode 100644 index 0000000..1fed14e --- /dev/null +++ b/tests/BuildFlow.Workflow.Tests/Performance/ImportJobBatchingTests.cs @@ -0,0 +1,203 @@ +using BuildFlow.Workflow.Tests.Reliability; +using FluentAssertions; +using Microsoft.EntityFrameworkCore; +using Workflow.Application.BuildFlow.Common.Services; +using Workflow.Application.BuildFlow.Imports.Handlers; +using Workflow.Application.BuildFlow.Imports.Services; +using Workflow.Application.Form.Aggregates; +using Workflow.Application.Form.Derivation; +using Workflow.Domain.Entities; +using Workflow.Domain.Enums; +using Xunit; + +namespace BuildFlow.Workflow.Tests.Performance; + +/// +/// 导入任务有界批次事务和取消恢复测试。 +/// +[Collection(PostgresWorkflowCollection.Name)] +public class ImportJobBatchingTests(PostgresWorkflowFixture fixture) +{ + [Fact] + public async Task FourHundredOneRowsUseThreeBatchCommitsAndOneCompletionCommit() + { + await ResetAsync(); + var filePath = await CreateTempFileAsync(); + try + { + var seeded = await SeedImportAsync(filePath); + var rows = Enumerable.Range(0, 401) + .Select(index => ValidRow(index + 2, $"row-{index}")) + .ToArray(); + var transactions = new CountingTransactionInterceptor(); + await using var db = fixture.CreateDbContext(transactions); + var processor = CreateProcessor(db, rows); + + await processor.ProcessAsync( + seeded.JobId, + seeded.LeaseOwner, + TimeSpan.FromMinutes(5), + 200, + CancellationToken.None); + + transactions.CommitCount.Should().Be(4); + await using var verify = fixture.CreateDbContext(); + var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == seeded.JobId); + job.Status.Should().Be(BuildFlowImportJobStatus.Success); + job.TotalRows.Should().Be(401); + job.SuccessRows.Should().Be(401); + job.FailedRows.Should().Be(0); + job.LeaseOwner.Should().BeNull(); + (await verify.FormData.CountAsync(x => + x.FormDefinitionId == seeded.FormDefinitionId)).Should().Be(401); + } + finally + { + File.Delete(filePath); + } + } + + [Fact] + public async Task CancellationBeforeRowTwoHundredOneKeepsFirstBatchCommitted() + { + await ResetAsync(); + var filePath = await CreateTempFileAsync(); + using var cts = new CancellationTokenSource(); + try + { + var seeded = await SeedImportAsync(filePath); + var rows = new CancellingExcelRows( + Enumerable.Range(0, 401) + .Select(index => ValidRow(index + 2, $"row-{index}")) + .ToArray(), + cancelAtIndex: 200, + cts.Cancel); + await using var db = fixture.CreateDbContext(); + var processor = CreateProcessor(db, rows); + + var act = () => processor.ProcessAsync( + seeded.JobId, + seeded.LeaseOwner, + TimeSpan.FromMinutes(5), + 200, + cts.Token); + + await act.Should().ThrowAsync(); + await using var verify = fixture.CreateDbContext(); + var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == seeded.JobId); + job.Status.Should().Be(BuildFlowImportJobStatus.Pending); + job.TotalRows.Should().Be(200); + job.SuccessRows.Should().Be(200); + job.FailedRows.Should().Be(0); + job.LeaseOwner.Should().BeNull(); + (await verify.FormData.CountAsync(x => + x.FormDefinitionId == seeded.FormDefinitionId)).Should().Be(200); + } + finally + { + File.Delete(filePath); + } + } + + [Fact] + public async Task ExistingFormWithNullSchemaUsesEmptyObjectSchema() + { + await ResetAsync(); + var filePath = await CreateTempFileAsync(); + try + { + var seeded = await SeedImportAsync(filePath, schemaJson: null); + await using var db = fixture.CreateDbContext(); + var processor = CreateProcessor(db, [ValidRow(2, "row")]); + + await processor.ProcessAsync( + seeded.JobId, + seeded.LeaseOwner, + TimeSpan.FromMinutes(5), + 1, + CancellationToken.None); + + await using var verify = fixture.CreateDbContext(); + var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == seeded.JobId); + job.Status.Should().Be(BuildFlowImportJobStatus.Success); + job.SuccessRows.Should().Be(1); + } + finally + { + File.Delete(filePath); + } + } + + private static ImportJobProcessor CreateProcessor( + WorkflowDbContext db, + IEnumerable rows) + { + var derivation = new FormValueDerivationService([], [], db); + var aggregates = new FormAggregateRefreshService(db); + return new ImportJobProcessor( + db, + new EnumerableFormSchemaExcelService(rows), + new ImportBatchWriter(db, derivation, aggregates)); + } + + private async Task ResetAsync() + { + await using var db = fixture.CreateDbContext(); + await db.FormData.ExecuteDeleteAsync(); + await db.BuildFlowImportJobs.ExecuteDeleteAsync(); + await db.FormDefinitions.ExecuteDeleteAsync(); + } + + private async Task<(Guid JobId, Guid FormDefinitionId, string LeaseOwner)> SeedImportAsync( + string filePath, + string? schemaJson = Schema) + { + await using var db = fixture.CreateDbContext(); + var formDefinitionId = Guid.NewGuid(); + var jobId = Guid.NewGuid(); + const string leaseOwner = "batch-import-worker"; + db.FormDefinitions.Add(new FormDefinition + { + Id = formDefinitionId, + Name = "批量导入测试表单", + Code = $"batch-import-{formDefinitionId:N}", + Version = 1, + SchemaJson = schemaJson, + Status = FormStatus.Published, + }); + db.BuildFlowImportJobs.Add(new BuildFlowImportJob + { + Id = jobId, + FormDefinitionId = formDefinitionId, + Status = BuildFlowImportJobStatus.Running, + FileUrl = filePath, + OriginalFileName = "batch.xlsx", + LeaseOwner = leaseOwner, + LeaseExpiresAt = DateTime.UtcNow.AddMinutes(5), + AttemptCount = 1, + LastHeartbeatAt = DateTime.UtcNow, + StartedAt = DateTime.UtcNow, + }); + await db.SaveChangesAsync(); + return (jobId, formDefinitionId, leaseOwner); + } + + private static ExcelParsedRow ValidRow(int rowNumber, string name) + => new(rowNumber, $$"""{"name":"{{name}}"}""", []); + + private static async Task CreateTempFileAsync() + { + var path = Path.Combine(Path.GetTempPath(), $"buildflow-import-{Guid.NewGuid():N}.xlsx"); + await File.WriteAllBytesAsync(path, [1, 2, 3]); + return path; + } + + private const string Schema = """ + { + "type": "object", + "properties": { + "name": { "type": "string", "required": true, "x-component": "Input" } + } + } + """; +} diff --git a/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobClaimTests.cs b/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobClaimTests.cs index b36cf0a..de79a58 100644 --- a/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobClaimTests.cs +++ b/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobClaimTests.cs @@ -97,7 +97,8 @@ public class ImportJobClaimTests(PostgresWorkflowFixture fixture) owner, BatchSize: 10, LeaseDuration: TimeSpan.FromMinutes(5), - MaxAttempts: 5); + MaxAttempts: 5, + RowsPerBatch: 200); private static ProcessAvailableImportJobsCommandHandler CreateHandler( WorkflowDbContext db, @@ -144,9 +145,10 @@ public class ImportJobClaimTests(PostgresWorkflowFixture fixture) Guid jobId, string leaseOwner, TimeSpan leaseDuration, + int rowsPerBatch, CancellationToken ct) { - Calls.Enqueue(new ImportJobCall(jobId, leaseOwner, leaseDuration)); + Calls.Enqueue(new ImportJobCall(jobId, leaseOwner, leaseDuration, rowsPerBatch)); return Task.CompletedTask; } } @@ -154,5 +156,6 @@ public class ImportJobClaimTests(PostgresWorkflowFixture fixture) private sealed record ImportJobCall( Guid JobId, string LeaseOwner, - TimeSpan LeaseDuration); + TimeSpan LeaseDuration, + int RowsPerBatch); } diff --git a/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobRecoveryTests.cs b/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobRecoveryTests.cs index acdd788..4cb7950 100644 --- a/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobRecoveryTests.cs +++ b/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobRecoveryTests.cs @@ -2,6 +2,9 @@ using FluentAssertions; using Microsoft.EntityFrameworkCore; using Workflow.Application.BuildFlow.Common.Services; using Workflow.Application.BuildFlow.Imports.Handlers; +using Workflow.Application.BuildFlow.Imports.Services; +using Workflow.Application.Form.Aggregates; +using Workflow.Application.Form.Derivation; using Workflow.Domain.Entities; using Workflow.Domain.Enums; using Xunit; @@ -23,18 +26,18 @@ public class ImportJobRecoveryTests(PostgresWorkflowFixture fixture) { var seeded = await SeedImportAsync(filePath, successRows: 1); await using var db = fixture.CreateDbContext(); - var processor = new ImportJobProcessor( + var processor = CreateProcessor( db, - new StubExcelService( [ ValidRow(2, """{"name":"first"}"""), ValidRow(3, """{"name":"second"}"""), - ])); + ]); await processor.ProcessAsync( seeded.JobId, seeded.LeaseOwner, TimeSpan.FromMinutes(5), + 1, CancellationToken.None); await using var verify = fixture.CreateDbContext(); @@ -61,14 +64,15 @@ public class ImportJobRecoveryTests(PostgresWorkflowFixture fixture) { var seeded = await SeedImportAsync(filePath); await using var db = fixture.CreateDbContext(); - var processor = new ImportJobProcessor( + var processor = CreateProcessor( db, - new StubExcelService([ValidRow(2, """{"name":"only"}""")])); + [ValidRow(2, """{"name":"only"}""")]); await processor.ProcessAsync( seeded.JobId, seeded.LeaseOwner, TimeSpan.FromMinutes(5), + 1, CancellationToken.None); await using var verify = fixture.CreateDbContext(); @@ -102,12 +106,13 @@ public class ImportJobRecoveryTests(PostgresWorkflowFixture fixture) cancelAtIndex: 1, cts.Cancel); await using var db = fixture.CreateDbContext(); - var processor = new ImportJobProcessor(db, new StubExcelService(rows)); + var processor = CreateProcessor(db, rows); var act = () => processor.ProcessAsync( seeded.JobId, seeded.LeaseOwner, TimeSpan.FromMinutes(5), + 1, cts.Token); await act.Should().ThrowAsync(); @@ -184,6 +189,18 @@ public class ImportJobRecoveryTests(PostgresWorkflowFixture fixture) private static ExcelParsedRow ValidRow(int rowNumber, string json) => new(rowNumber, json, []); + private static ImportJobProcessor CreateProcessor( + WorkflowDbContext db, + IReadOnlyList rows) + { + var derivation = new FormValueDerivationService([], [], db); + var aggregates = new FormAggregateRefreshService(db); + return new ImportJobProcessor( + db, + new StubExcelService(rows), + new ImportBatchWriter(db, derivation, aggregates)); + } + private static async Task CreateTempFileAsync() { var path = Path.Combine(Path.GetTempPath(), $"buildflow-import-{Guid.NewGuid():N}.xlsx");