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

perf: stream bounded excel imports

Browse Source
This commit is contained in:
xiangning 2026-07-24 10:42:18 +08:00
parent 85a95f05f7
commit de4241e5b5
20 changed files with 691 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/BuildFlow.Host/AggregateExceptionHandler.cs
View File

@ -33,8 +33,8 @@ public class AggregateExceptionHandler<TRequest, TResponse> : IPostProcessor<TRe
var ex = context.ExceptionDispatchInfo!.SourceException;
// 3 个通用异常（Unauthorized/NotFound/Business）现统一来自内核，两套模块共享同一类型；
// file-system 特有异常（Conflict/Forbidden/Share*/TooLarge）仍在 BuildFlow.File.Domain.Exceptions。
// 4 个通用异常（Unauthorized/NotFound/Business/TooLarge）现统一来自内核，各模块共享同一类型；
// file-system 旧有的 Conflict/Forbidden/Share*/TooLarge 异常继续兼容。
// code 取自异常的 Code 字段（语义错误码字符串），file 特有异常按其语义给常量码。
var (statusCode, code, message) = ex switch
{
@ -42,13 +42,15 @@ public class AggregateExceptionHandler<TRequest, TResponse> : IPostProcessor<TRe
RagExceptions.UnauthorizedException u => (HttpStatusCode.Unauthorized, u.Code, ex.Message),
RagExceptions.NotFoundException n => (HttpStatusCode.NotFound, n.Code, ex.Message),
RagExceptions.BusinessException b => (HttpStatusCode.BadRequest, b.Code, ex.Message),
RagExceptions.TooLargeException t => (HttpStatusCode.RequestEntityTooLarge, t.Code, ex.Message),
// ===== file-system 特有领域异常 =====
SharePasswordException => (HttpStatusCode.Unauthorized, ErrorCode.Unauthorized, ex.Message),
ForbiddenException => (HttpStatusCode.Forbidden, ErrorCode.Forbidden, ex.Message),
ConflictException => (HttpStatusCode.Conflict, ErrorCode.Conflict, ex.Message),
ShareExpiredException => (HttpStatusCode.Gone, ErrorCode.Expired, ex.Message),
ShareExhaustedException => (HttpStatusCode.Gone, ErrorCode.QuotaExceeded, ex.Message),
TooLargeException => (HttpStatusCode.RequestEntityTooLarge, ErrorCode.TooLarge, ex.Message),
global::File.Domain.Exceptions.TooLargeException =>
(HttpStatusCode.RequestEntityTooLarge, ErrorCode.TooLarge, ex.Message),
// ===== 通用 =====
FluentValidation.ValidationException vex => (HttpStatusCode.BadRequest, ErrorCode.InvalidParam,
string.Join("; ", vex.Errors.Select(e => e.ErrorMessage))),

8
src/BuildFlow.Host/BuildFlowHostModule.cs
View File

@ -119,8 +119,12 @@ public class BuildFlowHostModule : AbpModule
&& options.Import.BatchSize > 0
&& options.Import.LeaseSeconds > 0
&& options.Import.MaxAttempts > 0
&& options.Import.RowsPerBatch is > 0 and <= 1000,
"Workflow:Jobs 的间隔、批量、租约、重试延迟和最大尝试次数必须为正数，导入事务行数不得超过 1000。")
&& options.Import.RowsPerBatch is > 0 and <= 1000
&& options.Import.MaxFileBytes > 0
&& options.Import.MaxRows > 0
&& options.Import.MaxCellTextLength > 0
&& options.Import.ExportPageSize is > 0 and <= 5000,
"Workflow:Jobs 的间隔、批量、租约、重试延迟和最大尝试次数必须为正数，导入事务行数不得超过 1000，导出页大小不得超过 5000。")
.ValidateOnStart();
services.AddNotifications(config);
services.AddHostedService<TimeoutSchedulerService>();

10
src/BuildFlow.Host/Endpoints/Workflow/BuildFlow/Imports/CreateImportJobEndpoint.cs
View File

@ -20,12 +20,12 @@ public class CreateImportJobEndpoint(IMediator mediator) : Endpoint<CreateImport
public override async Task HandleAsync(CreateImportJobRequest req, CancellationToken ct)
{
using var stream = req.File.OpenReadStream();
using var ms = new MemoryStream();
await stream.CopyToAsync(ms, ct);
var bytes = ms.ToArray();
var result = await mediator.Send(
new CreateImportJobCommand(req.FormDefinitionId, bytes, req.File.FileName),
new CreateImportJobCommand(
req.FormDefinitionId,
stream,
req.File.Length,
req.File.FileName),
ct);
await Send.ResponseAsync(result, 201, ct);
}

6
src/BuildFlow.Host/appsettings.json
View File

@ -121,7 +121,11 @@
"BatchSize": 5,
"LeaseSeconds": 300,
"MaxAttempts": 5,
"RowsPerBatch": 200
"RowsPerBatch": 200,
"MaxFileBytes": 52428800,
"MaxRows": 100000,
"MaxCellTextLength": 32767,
"ExportPageSize": 500
}
}
},

157
src/modules/workflow/Application/BuildFlow/Common/Services/FormSchemaExcelService.cs
View File

@ -1,11 +1,14 @@
using System.Globalization;
using ClosedXML.Excel;
using DocumentFormat.OpenXml;
using DocumentFormat.OpenXml.Packaging;
using DocumentFormat.OpenXml.Spreadsheet;
using Workflow.Application.Form.Schema;
namespace Workflow.Application.BuildFlow.Common.Services;
/// <summary>
/// 基于 ClosedXML + SchemaValidator 实现的表单 Schema 与 Excel 互转服务。
/// 基于 ClosedXML、OpenXML 与 SchemaValidator 实现的表单 Schema 与 Excel 互转服务。
/// 字段顺序 = SchemaValidator.Validate 提取的 FieldSummary 顺序（深度优先）。
/// void 节点不计入字段。导出/模版仅取非 void、有 JsonType 的字段。
/// </summary>
@ -53,24 +56,69 @@ public class FormSchemaExcelService : IFormSchemaExcelService
}
/// <inheritdoc />
public IEnumerable<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes)
public IEnumerable<ExcelParsedRow> Parse(
string schemaJson,
Stream xlsxStream,
int maxRows,
int maxCellTextLength)
{
var fields = ExtractFields(schemaJson);
using var ms = new MemoryStream(xlsxBytes);
using var wb = new XLWorkbook(ms);
var ws = wb.Worksheets.First();
var lastRow = ws.LastRowUsed();
if (maxRows <= 0)
{
throw new ArgumentOutOfRangeException(nameof(maxRows));
}
if (maxCellTextLength <= 0)
{
throw new ArgumentOutOfRangeException(nameof(maxCellTextLength));
}
var headerMap = BuildHeaderMap(ws, fields);
if (lastRow is null)
var fields = ExtractFields(schemaJson);
using var document = SpreadsheetDocument.Open(xlsxStream, false);
var workbookPart = document.WorkbookPart
?? throw new InvalidOperationException("Excel 工作簿结构无效");
var firstSheet = workbookPart.Workbook.Sheets?
.Elements<Sheet>()
.FirstOrDefault();
if (firstSheet?.Id?.Value is null)
{
yield break;
}
for (var r = 2; r <= lastRow.RowNumber(); r++)
var worksheetPart = (WorksheetPart)workbookPart.GetPartById(firstSheet.Id.Value);
var sharedStrings = workbookPart.SharedStringTablePart?.SharedStringTable?
.Elements<SharedStringItem>()
.Select(item => item.InnerText)
.ToList()
?? [];
Dictionary<string, int>? headerMap = null;
var dataRowCount = 0;
using var reader = OpenXmlReader.Create(worksheetPart);
while (reader.Read())
{
var (json, errors) = BuildRowJson(ws, r, headerMap, fields);
yield return new ExcelParsedRow(r, json, errors);
if (!reader.IsStartElement || reader.ElementType != typeof(Row))
{
continue;
}
var row = reader.LoadCurrentElement() as Row
?? throw new InvalidOperationException("Excel 行结构无效");
var cells = ReadCells(row, sharedStrings, maxCellTextLength);
if (headerMap is null)
{
headerMap = BuildHeaderMap(cells, fields);
continue;
}
dataRowCount++;
if (dataRowCount > maxRows)
{
throw new BusinessException(
ErrorCode.TooLarge,
$"Excel 数据行数不能超过 {maxRows}");
}
var rowNumber = checked((int)(row.RowIndex?.Value ?? (uint)(dataRowCount + 1)));
var (json, errors) = BuildRowJson(rowNumber, cells, headerMap, fields);
yield return new ExcelParsedRow(rowNumber, json, errors);
}
}
@ -129,14 +177,14 @@ public class FormSchemaExcelService : IFormSchemaExcelService
}
}
private static Dictionary<string, int> BuildHeaderMap(IXLWorksheet ws, IReadOnlyList<FieldSummary> fields)
private static Dictionary<string, int> BuildHeaderMap(
IReadOnlyDictionary<int, string> cells,
IReadOnlyList<FieldSummary> fields)
{
var map = new Dictionary<string, int>(StringComparer.Ordinal);
var lastCol = ws.LastColumnUsed();
var colCount = lastCol?.ColumnNumber() ?? 0;
for (var c = 1; c <= colCount; c++)
foreach (var (column, value) in cells)
{
var headerText = ws.Cell(1, c).GetString().Trim();
var headerText = value.Trim();
if (string.IsNullOrEmpty(headerText))
{
continue;
@ -147,15 +195,15 @@ public class FormSchemaExcelService : IFormSchemaExcelService
string.Equals(f.Path, headerText, StringComparison.Ordinal));
if (matched is not null)
{
map[matched.Path] = c;
map[matched.Path] = column;
}
}
return map;
}
private static (string Json, IReadOnlyList<ExcelCellError> Errors) BuildRowJson(
IXLWorksheet ws,
int rowNumber,
IReadOnlyDictionary<int, string> cells,
Dictionary<string, int> headerMap,
IReadOnlyList<FieldSummary> fields)
{
@ -168,8 +216,7 @@ public class FormSchemaExcelService : IFormSchemaExcelService
continue;
}
var cell = ws.Cell(rowNumber, col);
var text = cell.GetString().Trim();
var text = cells.GetValueOrDefault(col, string.Empty).Trim();
if (string.IsNullOrEmpty(text))
{
if (field.Required)
@ -193,6 +240,74 @@ public class FormSchemaExcelService : IFormSchemaExcelService
return (json, errors);
}
private static Dictionary<int, string> ReadCells(
Row row,
IReadOnlyList<string> sharedStrings,
int maxCellTextLength)
{
var cells = new Dictionary<int, string>();
var nextColumn = 1;
foreach (var cell in row.Elements<Cell>())
{
var column = GetColumnIndex(cell.CellReference?.Value) ?? nextColumn;
nextColumn = column + 1;
var text = ReadCellText(cell, sharedStrings);
if (text.Length > maxCellTextLength)
{
throw new BusinessException(
ErrorCode.TooLarge,
$"Excel 单元格 {cell.CellReference?.Value ?? column.ToString(CultureInfo.InvariantCulture)} 文本长度不能超过 {maxCellTextLength}");
}
cells[column] = text;
}
return cells;
}
private static string ReadCellText(Cell cell, IReadOnlyList<string> sharedStrings)
{
var raw = cell.CellValue?.Text ?? string.Empty;
var dataType = cell.DataType?.Value;
if (dataType == CellValues.SharedString
&& int.TryParse(
raw,
NumberStyles.None,
CultureInfo.InvariantCulture,
out var index)
&& index >= 0
&& index < sharedStrings.Count)
{
return sharedStrings[index];
}
if (dataType == CellValues.InlineString)
{
return cell.InlineString?.InnerText ?? string.Empty;
}
if (dataType == CellValues.Boolean)
{
return raw == "1" ? "true" : raw == "0" ? "false" : raw;
}
return raw;
}
private static int? GetColumnIndex(string? cellReference)
{
if (string.IsNullOrEmpty(cellReference))
{
return null;
}
var column = 0;
foreach (var character in cellReference)
{
if (!char.IsLetter(character))
{
break;
}
column = checked(column * 26 + char.ToUpperInvariant(character) - 'A' + 1);
}
return column == 0 ? null : column;
}
private static object ConvertValue(FieldSummary field, string text)
{
return field.JsonType switch

8
src/modules/workflow/Application/BuildFlow/Common/Services/IFormSchemaExcelService.cs
View File

@ -12,6 +12,10 @@ public interface IFormSchemaExcelService
/// <summary>根据 SchemaJson 字段顺序导出已有 FormData 列表为 xlsx 字节流。</summary>
byte[] BuildExport(string schemaJson, string formName, IEnumerable<string> dataJsonRows);
/// <summary>解析上传 xlsx 字节流，按字段顺序惰性返回每行的 JSON 对象字符串。</summary>
IEnumerable<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes);
/// <summary>从 xlsx 流逐行解析数据，并在枚举阶段强制执行行数与单元格长度上限。</summary>
IEnumerable<ExcelParsedRow> Parse(
string schemaJson,
Stream xlsxStream,
int maxRows,
int maxCellTextLength);
}

7
src/modules/workflow/Application/BuildFlow/Imports/Commands/CreateImportJobCommand.cs
View File

@ -4,10 +4,11 @@ using Workflow.Application.BuildFlow.Imports.DTOs;
namespace Workflow.Application.BuildFlow.Imports.Commands;
/// <summary>
/// 创建导入任务命令。上传文件字节流由端点解析后传入，
/// 处理器仅负责持久化任务为 Pending 状态，实际执行交给后台服务。
/// 创建导入任务命令。上传流由处理器直接复制到临时文件，
/// 随后持久化 Pending 任务，实际解析与入库交给后台服务。
/// </summary>
public sealed record CreateImportJobCommand(
Guid FormDefinitionId,
byte[] FileBytes,
Stream FileStream,
long FileLength,
string OriginalFileName) : IRequest<ImportJobDto>;

126
src/modules/workflow/Application/BuildFlow/Imports/Handlers/CreateImportJobCommandHandler.cs
View File

@ -1,9 +1,13 @@
using System.Buffers;
using Microsoft.Extensions.Options;
using MediatR;
using Workflow.Application.BuildFlow.Imports.Commands;
using Workflow.Application.BuildFlow.Imports.DTOs;
using Workflow.Application.Reliability;
using Workflow.Domain.Entities;
using Workflow.Domain.Enums;
using Workflow.Infrastructure.Persistence;
using TooLargeException = BuildFlow.Shared.Domain.Exceptions.TooLargeException;
namespace Workflow.Application.BuildFlow.Imports.Handlers;
@ -12,46 +16,120 @@ namespace Workflow.Application.BuildFlow.Imports.Handlers;
/// 将上传文件保存到本地临时目录（{TempPath}/buildflow-imports/{jobId}.xlsx），
/// 并在数据库创建 Pending 任务，等待 ImportJobProcessorService 拉取执行。
/// </summary>
public class CreateImportJobCommandHandler(WorkflowDbContext db) : IRequestHandler<CreateImportJobCommand, ImportJobDto>
public class CreateImportJobCommandHandler(
WorkflowDbContext db,
IOptions<WorkflowJobOptions> options) : IRequestHandler<CreateImportJobCommand, ImportJobDto>
{
private const int CopyBufferSize = 81_920;
private static readonly string ImportDirectory =
Path.Combine(Path.GetTempPath(), "buildflow-imports");
public async Task<ImportJobDto> Handle(CreateImportJobCommand req, CancellationToken ct)
{
var formExists = await db.FormDefinitions
.AnyAsync(f => f.Id == req.FormDefinitionId, ct);
if (!formExists)
{
throw new BusinessException($"表单定义 {req.FormDefinitionId} 不存在");
}
if (req.FileBytes is { Length: 0 })
if (req.FileLength <= 0)
{
throw new BusinessException(ErrorCode.Business, "上传文件为空");
}
var maxFileBytes = options.Value.Import.MaxFileBytes;
if (req.FileLength > maxFileBytes)
{
throw new TooLargeException(
ErrorCode.TooLarge,
$"上传文件不能超过 {maxFileBytes} 字节");
}
var formExists = await db.FormDefinitions
.AnyAsync(f => f.Id == req.FormDefinitionId, ct);
if (!formExists)
{
throw new NotFoundException(
ErrorCode.NotFound,
$"表单定义 {req.FormDefinitionId} 不存在");
}
Directory.CreateDirectory(ImportDirectory);
var jobId = Guid.NewGuid();
var extension = Path.GetExtension(req.OriginalFileName);
if (string.IsNullOrEmpty(extension))
var fileUrl = Path.Combine(ImportDirectory, $"{jobId}.xlsx");
try
{
extension = ".xlsx";
await using (var destination = new FileStream(
fileUrl,
FileMode.CreateNew,
FileAccess.Write,
FileShare.None,
CopyBufferSize,
FileOptions.Asynchronous | FileOptions.SequentialScan))
{
var actualLength = await CopyToTemporaryFileAsync(
req.FileStream,
destination,
maxFileBytes,
ct);
await destination.FlushAsync(ct);
if (actualLength != req.FileLength)
{
throw new BusinessException(
ErrorCode.Business,
$"上传文件长度不一致，声明 {req.FileLength} 字节，实际 {actualLength} 字节");
}
}
var job = new BuildFlowImportJob
{
Id = jobId,
FormDefinitionId = req.FormDefinitionId,
Status = BuildFlowImportJobStatus.Pending,
FileUrl = fileUrl,
OriginalFileName = req.OriginalFileName,
};
db.BuildFlowImportJobs.Add(job);
await db.SaveChangesAsync(ct);
return ToDto(job);
}
var fileUrl = Path.Combine(ImportDirectory, $"{jobId}{extension}");
await File.WriteAllBytesAsync(fileUrl, req.FileBytes, ct);
var job = new BuildFlowImportJob
catch
{
Id = jobId,
FormDefinitionId = req.FormDefinitionId,
Status = BuildFlowImportJobStatus.Pending,
FileUrl = fileUrl,
OriginalFileName = req.OriginalFileName,
};
db.BuildFlowImportJobs.Add(job);
await db.SaveChangesAsync(ct);
File.Delete(fileUrl);
throw;
}
}
return ToDto(job);
private static async Task<long> CopyToTemporaryFileAsync(
Stream source,
Stream destination,
long maxFileBytes,
CancellationToken ct)
{
var buffer = ArrayPool<byte>.Shared.Rent(CopyBufferSize);
try
{
long totalBytes = 0;
while (true)
{
var bytesRead = await source.ReadAsync(
buffer.AsMemory(0, CopyBufferSize),
ct);
if (bytesRead == 0)
{
return totalBytes;
}
totalBytes += bytesRead;
if (totalBytes > maxFileBytes)
{
throw new TooLargeException(
ErrorCode.TooLarge,
$"上传文件不能超过 {maxFileBytes} 字节");
}
await destination.WriteAsync(buffer.AsMemory(0, bytesRead), ct);
}
}
finally
{
ArrayPool<byte>.Shared.Return(buffer);
}
}
internal static ImportJobDto ToDto(BuildFlowImportJob entity) => new(

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

@ -1,7 +1,9 @@
using System.Text.Json;
using Microsoft.Extensions.Options;
using Workflow.Application.BuildFlow.Common.Services;
using Workflow.Application.BuildFlow.Imports.Services;
using Workflow.Application.Form.Schema;
using Workflow.Application.Reliability;
using Workflow.Domain.Entities;
using Workflow.Domain.Enums;
using Workflow.Infrastructure.Persistence;
@ -14,7 +16,8 @@ namespace Workflow.Application.BuildFlow.Imports.Handlers;
public class ImportJobProcessor(
WorkflowDbContext db,
IFormSchemaExcelService excelService,
ImportBatchWriter batchWriter) : IImportJobProcessor
ImportBatchWriter batchWriter,
IOptions<WorkflowJobOptions>? options = null) : IImportJobProcessor
{
private const int MaxRowsPerBatch = 1000;
@ -74,12 +77,23 @@ public class ImportJobProcessor(
}
var preparedSchema = PreparedFormSchema.Create(schemaJson);
var fileBytes = await File.ReadAllBytesAsync(job.FileUrl, ct);
var importOptions = options?.Value.Import ?? new WorkflowJobOptions.ImportJobSection();
await using var fileStream = new FileStream(
job.FileUrl,
FileMode.Open,
FileAccess.Read,
FileShare.Read,
81_920,
FileOptions.Asynchronous | FileOptions.SequentialScan);
var committedCursor = job.SuccessRows + job.FailedRows;
var batch = new List<ImportBatchItem>(rowsPerBatch);
var rowIndex = 0;
foreach (var row in excelService.Parse(schemaJson, fileBytes))
foreach (var row in excelService.Parse(
schemaJson,
fileStream,
importOptions.MaxRows,
importOptions.MaxCellTextLength))
{
ct.ThrowIfCancellationRequested();
if (rowIndex >= committedCursor)

12
src/modules/workflow/Application/Reliability/WorkflowJobOptions.cs
View File

@ -56,5 +56,17 @@ public class WorkflowJobOptions
/// <summary>单个导入事务最多处理的 Excel 行数。</summary>
public int RowsPerBatch { get; set; } = 200;
/// <summary>单个上传文件的最大字节数。</summary>
public long MaxFileBytes { get; set; } = 50 * 1024 * 1024;
/// <summary>单个工作簿允许的数据行数上限，不含表头。</summary>
public int MaxRows { get; set; } = 100_000;
/// <summary>单个单元格允许的最大文本长度。</summary>
public int MaxCellTextLength { get; set; } = 32_767;
/// <summary>Excel 导出查询的单页行数。</summary>
public int ExportPageSize { get; set; } = 500;
}
}

1
src/modules/workflow/Application/Workflow.Application.csproj
View File

@ -13,6 +13,7 @@
<PackageReference Include="Microsoft.CodeAnalysis.CSharp.Scripting" Version="5.0.0" />
<PackageReference Include="Microsoft.CodeAnalysis.Scripting" Version="5.0.0" />
<PackageReference Include="ClosedXML" Version="0.104.2" />
<PackageReference Include="DocumentFormat.OpenXml" Version="3.1.1" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Infrastructure\Workflow.Infrastructure.csproj" />

1
src/shared/BuildFlow.Shared/Api/GlobalExceptionHandler.cs
View File

@ -54,6 +54,7 @@ public sealed class GlobalExceptionHandler<TRequest, TResponse> : IPostProcessor
UnauthorizedException u => (HttpStatusCode.Unauthorized, u.Code, ex.Message),
NotFoundException n => (HttpStatusCode.NotFound, n.Code, ex.Message),
BusinessException b => (HttpStatusCode.BadRequest, b.Code, ex.Message),
TooLargeException t => (HttpStatusCode.RequestEntityTooLarge, t.Code, ex.Message),
// 多条校验错误用分号拼接，便于前端一次性展示
ValidationException vex => (HttpStatusCode.BadRequest, ErrorCode.InvalidParam,
string.Join("; ", vex.Errors.Select(e => e.ErrorMessage))),

1
src/shared/BuildFlow.Shared/Api/GlobalExceptionMiddleware.cs
View File

@ -55,6 +55,7 @@ public sealed class GlobalExceptionMiddleware(RequestDelegate next)
UnauthorizedException u => (HttpStatusCode.Unauthorized, u.Code, ex.Message),
NotFoundException n => (HttpStatusCode.NotFound, n.Code, ex.Message),
BusinessException b => (HttpStatusCode.BadRequest, b.Code, ex.Message),
TooLargeException t => (HttpStatusCode.RequestEntityTooLarge, t.Code, ex.Message),
// 多条校验错误用分号拼接，便于前端一次性展示
ValidationException vex => (HttpStatusCode.BadRequest, ErrorCode.InvalidParam,
string.Join("; ", vex.Errors.Select(e => e.ErrorMessage))),

14
src/shared/BuildFlow.Shared/Domain/Exceptions/TooLargeException.cs Normal file
View File

@ -0,0 +1,14 @@
namespace BuildFlow.Shared.Domain.Exceptions;
/// <summary>请求内容超过业务允许上限的异常，统一映射为 HTTP 413。</summary>
public class TooLargeException : Exception
{
/// <summary>语义错误码。</summary>
public string Code { get; }
/// <summary>创建携带明确语义错误码的请求内容超限异常。</summary>
public TooLargeException(string code, string message) : base(message)
{
Code = code;
}
}

6
tests/BuildFlow.Workflow.Tests/BuildFlow/FormSchemaExcelServiceTests.cs
View File

@ -48,8 +48,9 @@ public class FormSchemaExcelServiceTests
ws.Cell(2, 3).Value = "true";
using var outMs = new MemoryStream();
wb.SaveAs(outMs);
outMs.Position = 0;
var rows = _service.Parse(SampleSchema, outMs.ToArray()).ToList();
var rows = _service.Parse(SampleSchema, outMs, 100, 32_767).ToList();
rows.Should().HaveCount(1);
var row = rows[0];
@ -72,8 +73,9 @@ public class FormSchemaExcelServiceTests
ws.Cell(2, 2).Value = 50;
using var outMs = new MemoryStream();
wb.SaveAs(outMs);
outMs.Position = 0;
var rows = _service.Parse(SampleSchema, outMs.ToArray()).ToList();
var rows = _service.Parse(SampleSchema, outMs, 100, 32_767).ToList();
rows.Should().HaveCount(1);
rows[0].Errors.Should().ContainSingle(e => e.Message.Contains("必填"));

6
tests/BuildFlow.Workflow.Tests/Performance/EnumerableFormSchemaExcelService.cs
View File

@ -13,5 +13,9 @@ public sealed class EnumerableFormSchemaExcelService(IEnumerable<ExcelParsedRow>
string formName,
IEnumerable<string> dataJsonRows) => [];
public IEnumerable<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes) => rows;
public IEnumerable<ExcelParsedRow> Parse(
string schemaJson,
Stream xlsxStream,
int maxRows,
int maxCellTextLength) => rows;
}

3
tests/BuildFlow.Workflow.Tests/Performance/LazyExcelRowTests.cs
View File

@ -23,8 +23,9 @@ public class LazyExcelRowTests
worksheet.Cell(4, 1).Value = "third";
using var output = new MemoryStream();
workbook.SaveAs(output);
output.Position = 0;
var rows = service.Parse(Schema, output.ToArray());
var rows = service.Parse(Schema, output, 100, 32_767);
rows.Should().NotBeAssignableTo<IReadOnlyList<ExcelParsedRow>>();
rows.Take(1).Should().ContainSingle(x =>

276
tests/BuildFlow.Workflow.Tests/Performance/OpenXmlImportBoundaryTests.cs Normal file
View File

@ -0,0 +1,276 @@
using System.Text.Json;
using BuildFlow.Shared.Domain.Errors;
using DocumentFormat.OpenXml;
using DocumentFormat.OpenXml.Packaging;
using DocumentFormat.OpenXml.Spreadsheet;
using FluentAssertions;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Options;
using Workflow.Application.BuildFlow.Common.Services;
using Workflow.Application.BuildFlow.Imports.Commands;
using Workflow.Application.BuildFlow.Imports.Handlers;
using Workflow.Application.Reliability;
using Workflow.Domain.Entities;
using Workflow.Domain.Enums;
using Xunit;
namespace BuildFlow.Workflow.Tests.Performance;
/// <summary>Excel 导入上传与 OpenXML 流式解析的资源边界测试。</summary>
public class OpenXmlImportBoundaryTests
{
[Fact]
public async Task OversizedUploadIsRejectedBeforeReadingStream()
{
await using var db = await CreateDbContextWithFormAsync();
var formId = await db.FormDefinitions.Select(form => form.Id).SingleAsync();
using var stream = new TrackingReadStream(new MemoryStream([1, 2, 3]));
var handler = CreateUploadHandler(db, maxFileBytes: 10);
var act = () => handler.Handle(
new CreateImportJobCommand(formId, stream, 11, "oversized.xlsx"),
CancellationToken.None);
var exception = await act.Should()
.ThrowAsync<global::BuildFlow.Shared.Domain.Exceptions.TooLargeException>();
exception.Which.Code.Should().Be(ErrorCode.TooLarge);
stream.BytesRead.Should().Be(0);
stream.ReadCalls.Should().Be(0);
db.BuildFlowImportJobs.Should().BeEmpty();
}
[Fact]
public async Task UploadCopiesStreamToTemporaryFileWithBoundedBuffer()
{
await using var db = await CreateDbContextWithFormAsync();
var formId = await db.FormDefinitions.Select(form => form.Id).SingleAsync();
var payload = Enumerable.Range(0, 200_000)
.Select(index => (byte)(index % 251))
.ToArray();
using var stream = new TrackingReadStream(new MemoryStream(payload));
var handler = CreateUploadHandler(db, maxFileBytes: 500_000);
string? filePath = null;
try
{
await handler.Handle(
new CreateImportJobCommand(formId, stream, payload.Length, "data.xlsx"),
CancellationToken.None);
var job = await db.BuildFlowImportJobs.SingleAsync();
filePath = job.FileUrl;
(await File.ReadAllBytesAsync(filePath)).Should().Equal(payload);
stream.BytesRead.Should().Be(payload.Length);
stream.ReadCalls.Should().BeGreaterThan(1);
stream.MaxRequestedReadSize.Should().BeLessThanOrEqualTo(81_920);
}
finally
{
if (filePath is not null)
{
File.Delete(filePath);
}
}
}
[Fact]
public async Task MissingFormIsRejectedWithoutReadingStream()
{
await using var db = TestDbFactory.Create();
using var stream = new TrackingReadStream(new MemoryStream([1]));
var handler = CreateUploadHandler(db, maxFileBytes: 10);
var act = () => handler.Handle(
new CreateImportJobCommand(Guid.NewGuid(), stream, 1, "data.xlsx"),
CancellationToken.None);
var exception = await act.Should().ThrowAsync<NotFoundException>();
exception.Which.Code.Should().Be(ErrorCode.NotFound);
stream.BytesRead.Should().Be(0);
}
[Fact]
public void ParserRejectsRowBeyondConfiguredLimit()
{
using var stream = CreateWorkbook(
[
HeaderRow("名称"),
InlineRow(2, "first"),
InlineRow(3, "second"),
InlineRow(4, "third"),
]);
var service = new FormSchemaExcelService();
var act = () => service.Parse(SingleTextFieldSchema, stream, 2, 100).ToList();
var exception = act.Should().Throw<BusinessException>();
exception.Which.Code.Should().Be(ErrorCode.TooLarge);
}
[Fact]
public void ParserRejectsCellTextBeyondConfiguredLimit()
{
using var stream = CreateWorkbook(
[HeaderRow("名称"), InlineRow(2, "123456")]);
var service = new FormSchemaExcelService();
var act = () => service.Parse(SingleTextFieldSchema, stream, 10, 5).ToList();
var exception = act.Should().Throw<BusinessException>();
exception.Which.Code.Should().Be(ErrorCode.TooLarge);
}
[Fact]
public void ParserReadsSharedInlineNumberBooleanAndBlankCells()
{
using var stream = CreateTypedWorkbook();
var service = new FormSchemaExcelService();
var row = service.Parse(TypedSchema, stream, 10, 100).Single();
using var json = JsonDocument.Parse(row.JsonObject);
var root = json.RootElement;
root.GetProperty("shared").GetString().Should().Be("shared-value");
root.GetProperty("inline").GetString().Should().Be("inline-value");
root.GetProperty("amount").GetDecimal().Should().Be(42.5m);
root.GetProperty("approved").GetBoolean().Should().BeTrue();
root.TryGetProperty("blank", out _).Should().BeFalse();
}
private static CreateImportJobCommandHandler CreateUploadHandler(
WorkflowDbContext db,
long maxFileBytes)
=> new(
db,
Options.Create(new WorkflowJobOptions
{
Import = new WorkflowJobOptions.ImportJobSection
{
MaxFileBytes = maxFileBytes,
},
}));
private static async Task<WorkflowDbContext> CreateDbContextWithFormAsync()
{
var db = TestDbFactory.Create();
db.FormDefinitions.Add(new FormDefinition
{
Id = Guid.NewGuid(),
Name = "导入边界测试",
Code = $"import-boundary-{Guid.NewGuid():N}",
Version = 1,
SchemaJson = SingleTextFieldSchema,
Status = FormStatus.Published,
});
await db.SaveChangesAsync();
return db;
}
private static MemoryStream CreateTypedWorkbook()
{
var stream = new MemoryStream();
using (var document = SpreadsheetDocument.Create(stream, SpreadsheetDocumentType.Workbook, true))
{
var workbookPart = document.AddWorkbookPart();
workbookPart.Workbook = new Workbook();
var sharedStringsPart = workbookPart.AddNewPart<SharedStringTablePart>();
sharedStringsPart.SharedStringTable = new SharedStringTable(
new SharedStringItem(new Text("shared-value")));
var worksheetPart = workbookPart.AddNewPart<WorksheetPart>();
worksheetPart.Worksheet = new Worksheet(new SheetData(
HeaderRow("Shared", "Inline", "Number", "Boolean", "Blank"),
new Row(
new Cell { CellReference = "A2", DataType = CellValues.SharedString, CellValue = new CellValue("0") },
InlineCell("B2", "inline-value"),
new Cell { CellReference = "C2", DataType = CellValues.Number, CellValue = new CellValue("42.5") },
new Cell { CellReference = "D2", DataType = CellValues.Boolean, CellValue = new CellValue("1") },
new Cell { CellReference = "E2" })
{
RowIndex = 2,
}));
AppendSheet(workbookPart, worksheetPart);
}
stream.Position = 0;
return stream;
}
private static MemoryStream CreateWorkbook(IEnumerable<Row> rows)
{
var stream = new MemoryStream();
using (var document = SpreadsheetDocument.Create(stream, SpreadsheetDocumentType.Workbook, true))
{
var workbookPart = document.AddWorkbookPart();
workbookPart.Workbook = new Workbook();
var worksheetPart = workbookPart.AddNewPart<WorksheetPart>();
worksheetPart.Worksheet = new Worksheet(new SheetData(rows));
AppendSheet(workbookPart, worksheetPart);
}
stream.Position = 0;
return stream;
}
private static void AppendSheet(WorkbookPart workbookPart, WorksheetPart worksheetPart)
{
var sheets = workbookPart.Workbook.AppendChild(new Sheets());
sheets.Append(new Sheet
{
Id = workbookPart.GetIdOfPart(worksheetPart),
SheetId = 1,
Name = "Sheet1",
});
workbookPart.Workbook.Save();
}
private static Row HeaderRow(params string[] headers)
=> new(headers.Select((header, index) => InlineCell($"{ColumnName(index + 1)}1", header)))
{
RowIndex = 1,
};
private static Row InlineRow(uint rowIndex, string value)
=> new(InlineCell($"A{rowIndex}", value))
{
RowIndex = rowIndex,
};
private static Cell InlineCell(string reference, string value)
=> new(new InlineString(new Text(value)))
{
CellReference = reference,
DataType = CellValues.InlineString,
};
private static string ColumnName(int index)
{
var name = string.Empty;
while (index > 0)
{
index--;
name = (char)('A' + index % 26) + name;
index /= 26;
}
return name;
}
private const string SingleTextFieldSchema = """
{
"type": "object",
"properties": {
"name": { "type": "string", "title": "名称", "required": true, "x-component": "Input" }
}
}
""";
private const string TypedSchema = """
{
"type": "object",
"properties": {
"shared": { "type": "string", "title": "Shared", "x-component": "Input" },
"inline": { "type": "string", "title": "Inline", "x-component": "Input" },
"amount": { "type": "number", "title": "Number", "x-component": "InputNumber" },
"approved": { "type": "boolean", "title": "Boolean", "x-component": "Switch" },
"blank": { "type": "string", "title": "Blank", "x-component": "Input" }
}
}
""";
}

84
tests/BuildFlow.Workflow.Tests/Performance/TrackingReadStream.cs Normal file
View File

@ -0,0 +1,84 @@
namespace BuildFlow.Workflow.Tests.Performance;
/// <summary>记录读取次数、总字节数和单次最大读取请求的只读测试流。</summary>
public sealed class TrackingReadStream(Stream inner) : Stream
{
/// <summary>累计实际读取字节数。</summary>
public long BytesRead { get; private set; }
/// <summary>累计读取调用次数。</summary>
public int ReadCalls { get; private set; }
/// <summary>单次读取请求的最大缓冲区长度。</summary>
public int MaxRequestedReadSize { get; private set; }
public override bool CanRead => inner.CanRead;
public override bool CanSeek => inner.CanSeek;
public override bool CanWrite => false;
public override long Length => inner.Length;
public override long Position
{
get => inner.Position;
set => inner.Position = value;
}
public override void Flush() => inner.Flush();
public override int Read(byte[] buffer, int offset, int count)
{
RecordRequest(count);
var read = inner.Read(buffer, offset, count);
BytesRead += read;
return read;
}
public override int Read(Span<byte> buffer)
{
RecordRequest(buffer.Length);
var read = inner.Read(buffer);
BytesRead += read;
return read;
}
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default)
{
RecordRequest(buffer.Length);
var read = await inner.ReadAsync(buffer, cancellationToken);
BytesRead += read;
return read;
}
public override async Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken)
{
RecordRequest(count);
var read = await inner.ReadAsync(buffer.AsMemory(offset, count), cancellationToken);
BytesRead += read;
return read;
}
public override long Seek(long offset, SeekOrigin origin) => inner.Seek(offset, origin);
public override void SetLength(long value) => throw new NotSupportedException();
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
protected override void Dispose(bool disposing)
{
if (disposing)
{
inner.Dispose();
}
base.Dispose(disposing);
}
private void RecordRequest(int count)
{
ReadCalls++;
MaxRequestedReadSize = Math.Max(MaxRequestedReadSize, count);
}
}

6
tests/BuildFlow.Workflow.Tests/Reliability/ImportJobRecoveryTests.cs
View File

@ -218,7 +218,11 @@ public class ImportJobRecoveryTests(PostgresWorkflowFixture fixture)
string formName,
IEnumerable<string> dataJsonRows) => [];
public IEnumerable<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes) => rows;
public IEnumerable<ExcelParsedRow> Parse(
string schemaJson,
Stream xlsxStream,
int maxRows,
int maxCellTextLength) => rows;
}
private sealed class CancellingRows(