From 85a95f05f7ec3e7b81f124f8a0d8dd1f974bac26 Mon Sep 17 00:00:00 2001 From: xiangning Date: Fri, 24 Jul 2026 10:22:05 +0800 Subject: [PATCH] perf: repair file sizes in background --- src/BuildFlow.Host/BuildFlowHostModule.cs | 13 ++ .../FileSizeVerificationService.cs | 70 ++++++ src/BuildFlow.Host/appsettings.json | 8 + .../Common/FileSizeBackfillHelper.cs | 52 ----- .../Common/FileSizeVerificationOptions.cs | 20 ++ .../ProcessUnverifiedFileSizesCommand.cs | 8 + ...rocessUnverifiedFileSizesCommandHandler.cs | 14 ++ .../Services/FileSizeVerificationProcessor.cs | 106 +++++++++ .../Folders/Handlers/GetFolderQueryHandler.cs | 19 +- .../Handlers/GetFolderTreeQueryHandler.cs | 19 +- .../Handlers/GetRootFilesQueryHandler.cs | 9 +- .../file/Application/FileApplicationModule.cs | 2 + .../FileSizeVerificationProcessorTests.cs | 221 ++++++++++++++++++ .../FoldersBucketsSettingsTests.cs | 120 +++++++++- 14 files changed, 604 insertions(+), 77 deletions(-) create mode 100644 src/BuildFlow.Host/Services/BackgroundServices/FileSizeVerificationService.cs delete mode 100644 src/modules/file/Application/Common/FileSizeBackfillHelper.cs create mode 100644 src/modules/file/Application/Common/FileSizeVerificationOptions.cs create mode 100644 src/modules/file/Application/Features/Files/Commands/ProcessUnverifiedFileSizesCommand.cs create mode 100644 src/modules/file/Application/Features/Files/Handlers/ProcessUnverifiedFileSizesCommandHandler.cs create mode 100644 src/modules/file/Application/Features/Files/Services/FileSizeVerificationProcessor.cs create mode 100644 tests/BuildFlow.File.Tests/FileSizeVerificationProcessorTests.cs diff --git a/src/BuildFlow.Host/BuildFlowHostModule.cs b/src/BuildFlow.Host/BuildFlowHostModule.cs index 20f8697..a4edd53 100644 --- a/src/BuildFlow.Host/BuildFlowHostModule.cs +++ b/src/BuildFlow.Host/BuildFlowHostModule.cs @@ -12,6 +12,7 @@ using OpenTelemetry.Trace; using BuildFlow.Shared.Api; using BuildFlow.Shared.Infrastructure.Serialization; using File.Application; +using File.Application.Common; using File.Infrastructure; using IM.Application; using IM.Domain.Common; @@ -127,6 +128,18 @@ public class BuildFlowHostModule : AbpModule services.AddHostedService(); services.AddHostedService(); + // file 模块：在后台有限并发确认历史文件大小，GET 请求保持纯数据库读取。 + services.AddOptions() + .Bind(config.GetSection(FileSizeVerificationOptions.SectionName)) + .Validate( + options => options.PollIntervalSeconds > 0 + && options.StartupDelaySeconds > 0 + && options.BatchSize is > 0 and <= 1000 + && options.MaxConcurrency is > 0 and <= 32, + "File:SizeVerification 的间隔和启动延迟必须为正数，批量不得超过 1000，并发不得超过 32。") + .ValidateOnStart(); + services.AddHostedService(); + // monitoring 模块：监控同步调度器（定时轮询外部系统拉取数据） services.Configure(config.GetSection("Monitoring")); services.AddHostedService(); diff --git a/src/BuildFlow.Host/Services/BackgroundServices/FileSizeVerificationService.cs b/src/BuildFlow.Host/Services/BackgroundServices/FileSizeVerificationService.cs new file mode 100644 index 0000000..481fc6a --- /dev/null +++ b/src/BuildFlow.Host/Services/BackgroundServices/FileSizeVerificationService.cs @@ -0,0 +1,70 @@ +using File.Application.Common; +using File.Application.Features.Files.Commands; +using MediatR; +using Microsoft.Extensions.Options; + +namespace BuildFlow.Host.Services.BackgroundServices; + +/// 周期触发历史文件大小确认命令的 Host 后台服务。 +public sealed class FileSizeVerificationService( + IServiceScopeFactory scopeFactory, + IOptionsMonitor options, + ILogger logger) : BackgroundService +{ + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + var current = options.CurrentValue; + logger.LogInformation( + "FileSizeVerification 启动，轮询间隔 {Interval}s，批量 {BatchSize}，最大并发 {MaxConcurrency}", + current.PollIntervalSeconds, + current.BatchSize, + current.MaxConcurrency); + + try + { + await Task.Delay(TimeSpan.FromSeconds(current.StartupDelaySeconds), stoppingToken); + using var timer = new PeriodicTimer(TimeSpan.FromSeconds(current.PollIntervalSeconds)); + + do + { + try + { + await RunScanAsync(stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + logger.LogError(ex, "FileSizeVerification 单轮处理异常"); + } + } + while (await timer.WaitForNextTickAsync(stoppingToken)); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + // 正常停机。 + } + + logger.LogInformation("FileSizeVerification 停止"); + } + + /// 执行一轮历史文件大小确认。 + public async Task RunScanAsync(CancellationToken ct = default) + { + using var scope = scopeFactory.CreateScope(); + var mediator = scope.ServiceProvider.GetRequiredService(); + var current = options.CurrentValue; + var processed = await mediator.Send( + new ProcessUnverifiedFileSizesCommand(current.BatchSize, current.MaxConcurrency), + ct); + + if (processed > 0) + { + logger.LogInformation("FileSizeVerification 完成一轮处理，尝试 {Count} 个文件", processed); + } + + return processed; + } +} diff --git a/src/BuildFlow.Host/appsettings.json b/src/BuildFlow.Host/appsettings.json index 9c7b65e..6648a06 100644 --- a/src/BuildFlow.Host/appsettings.json +++ b/src/BuildFlow.Host/appsettings.json @@ -97,6 +97,14 @@ "MaxTotalBytes": 536870912 } }, + "File": { + "SizeVerification": { + "PollIntervalSeconds": 60, + "StartupDelaySeconds": 15, + "BatchSize": 100, + "MaxConcurrency": 4 + } + }, "Workflow": { "Jobs": { "Timer": { diff --git a/src/modules/file/Application/Common/FileSizeBackfillHelper.cs b/src/modules/file/Application/Common/FileSizeBackfillHelper.cs deleted file mode 100644 index d5447fe..0000000 --- a/src/modules/file/Application/Common/FileSizeBackfillHelper.cs +++ /dev/null @@ -1,52 +0,0 @@ -using File.Domain.Entities.FileMeta; -using File.Infrastructure.Persistence; -using File.Infrastructure.Storage; - -namespace File.Application.Common; - -/// -/// 文件大小回填助手：对 Size==0 的 FileMetaEntity，从 S3 HeadObject 取真实 ContentLength 写回。 -/// 用于读路径（列表加载）惰性修复早期未记录大小的旧数据——首次加载即回填并持久化， -/// 后续加载直接命中 DB 不再访问 S3。对象缺失或空文件保留 Size=0；单个失败不中断整体。 -/// -public static class FileSizeBackfillHelper -{ - public static async Task BackfillAsync( - FileDbContext db, - IS3StorageService s3, - IReadOnlyList files, - CancellationToken ct) - { - var stale = files.Where(f => f.Size == 0).ToList(); - if (stale.Count == 0) - { - return; - } - - var updated = 0; - foreach (var file in stale) - { - try - { - var size = await s3.GetObjectSizeAsync(file.S3Bucket, file.S3Key, ct); - if (size is null || size <= 0) - { - // 对象缺失（已被外部清理）或空文件：保留 Size=0 - continue; - } - - file.Size = size.Value; - updated++; - } - catch - { - // 单个对象查询失败不中断整体回填，保留 Size=0 下次可重试 - } - } - - if (updated > 0) - { - await db.SaveChangesAsync(ct); - } - } -} diff --git a/src/modules/file/Application/Common/FileSizeVerificationOptions.cs b/src/modules/file/Application/Common/FileSizeVerificationOptions.cs new file mode 100644 index 0000000..549546e --- /dev/null +++ b/src/modules/file/Application/Common/FileSizeVerificationOptions.cs @@ -0,0 +1,20 @@ +namespace File.Application.Common; + +/// 历史文件大小后台确认任务的轮询、批量和并发配置。 +public class FileSizeVerificationOptions +{ + /// 配置节名称。 + public const string SectionName = "File:SizeVerification"; + + /// 两轮扫描之间的间隔秒数。 + public int PollIntervalSeconds { get; set; } = 60; + + /// Host 启动后的首次扫描延迟秒数。 + public int StartupDelaySeconds { get; set; } = 15; + + /// 单轮最多处理的未确认文件数。 + public int BatchSize { get; set; } = 100; + + /// 单轮查询 S3 对象元数据的最大并发数。 + public int MaxConcurrency { get; set; } = 4; +} diff --git a/src/modules/file/Application/Features/Files/Commands/ProcessUnverifiedFileSizesCommand.cs b/src/modules/file/Application/Features/Files/Commands/ProcessUnverifiedFileSizesCommand.cs new file mode 100644 index 0000000..f74d049 --- /dev/null +++ b/src/modules/file/Application/Features/Files/Commands/ProcessUnverifiedFileSizesCommand.cs @@ -0,0 +1,8 @@ +using MediatR; + +namespace File.Application.Features.Files.Commands; + +/// 按有限批量和有限并发确认历史文件大小的后台命令。 +public sealed record ProcessUnverifiedFileSizesCommand( + int BatchSize, + int MaxConcurrency) : IRequest; diff --git a/src/modules/file/Application/Features/Files/Handlers/ProcessUnverifiedFileSizesCommandHandler.cs b/src/modules/file/Application/Features/Files/Handlers/ProcessUnverifiedFileSizesCommandHandler.cs new file mode 100644 index 0000000..dcb5903 --- /dev/null +++ b/src/modules/file/Application/Features/Files/Handlers/ProcessUnverifiedFileSizesCommandHandler.cs @@ -0,0 +1,14 @@ +using File.Application.Features.Files.Commands; +using File.Application.Features.Files.Services; +using MediatR; + +namespace File.Application.Features.Files.Handlers; + +/// 将后台文件大小确认命令委托给模块内处理器。 +public sealed class ProcessUnverifiedFileSizesCommandHandler( + FileSizeVerificationProcessor processor) + : IRequestHandler +{ + public Task Handle(ProcessUnverifiedFileSizesCommand request, CancellationToken ct) + => processor.ProcessAsync(request.BatchSize, request.MaxConcurrency, ct); +} diff --git a/src/modules/file/Application/Features/Files/Services/FileSizeVerificationProcessor.cs b/src/modules/file/Application/Features/Files/Services/FileSizeVerificationProcessor.cs new file mode 100644 index 0000000..857885d --- /dev/null +++ b/src/modules/file/Application/Features/Files/Services/FileSizeVerificationProcessor.cs @@ -0,0 +1,106 @@ +using System.Collections.Concurrent; +using File.Infrastructure.Persistence; +using File.Infrastructure.Storage; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; + +namespace File.Application.Features.Files.Services; + +/// +/// 分批确认历史文件的对象大小。S3 查询有限并发执行，EF 实体只在并发阶段结束后统一修改并保存。 +/// +public sealed class FileSizeVerificationProcessor( + FileDbContext db, + IS3StorageService s3, + ILogger logger) +{ + /// 处理一批最早创建且尚未确认大小的文件，返回本轮尝试数量。 + public async Task ProcessAsync(int batchSize, int maxConcurrency, CancellationToken ct) + { + if (batchSize is < 1 or > 1000) + { + throw new ArgumentOutOfRangeException( + nameof(batchSize), + batchSize, + "批量大小必须在 1 到 1000 之间"); + } + + if (maxConcurrency is < 1 or > 32) + { + throw new ArgumentOutOfRangeException( + nameof(maxConcurrency), + maxConcurrency, + "最大并发数必须在 1 到 32 之间"); + } + + var files = await db.Files + .Where(file => !file.SizeVerified) + .OrderBy(file => file.CreatedAt) + .ThenBy(file => file.Id) + .Take(batchSize) + .ToListAsync(ct); + if (files.Count == 0) + { + return 0; + } + + var workItems = files + .Select(file => (file.Id, file.S3Bucket, file.S3Key)) + .ToArray(); + var results = new ConcurrentQueue<(Guid Id, long? Size)>(); + await Parallel.ForEachAsync( + workItems, + new ParallelOptions + { + CancellationToken = ct, + MaxDegreeOfParallelism = maxConcurrency, + }, + async (file, token) => + { + try + { + var size = await s3.GetObjectSizeAsync(file.S3Bucket, file.S3Key, token); + results.Enqueue((file.Id, size)); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + logger.LogWarning( + ex, + "确认文件大小失败，稍后重试。FileId={FileId}, Bucket={Bucket}, Key={Key}", + file.Id, + file.S3Bucket, + file.S3Key); + } + }); + + var filesById = files.ToDictionary(file => file.Id); + var verifiedCount = 0; + foreach (var result in results) + { + var file = filesById[result.Id]; + file.Size = result.Size ?? 0; + file.SizeVerified = true; + verifiedCount++; + + if (result.Size is null) + { + logger.LogWarning( + "确认文件大小时发现 S3 对象不存在，停止重试。FileId={FileId}, Bucket={Bucket}, Key={Key}", + file.Id, + file.S3Bucket, + file.S3Key); + } + } + + if (verifiedCount > 0) + { + await db.SaveChangesAsync(ct); + } + + return files.Count; + } +} diff --git a/src/modules/file/Application/Features/Folders/Handlers/GetFolderQueryHandler.cs b/src/modules/file/Application/Features/Folders/Handlers/GetFolderQueryHandler.cs index 5329937..06a8507 100644 --- a/src/modules/file/Application/Features/Folders/Handlers/GetFolderQueryHandler.cs +++ b/src/modules/file/Application/Features/Folders/Handlers/GetFolderQueryHandler.cs @@ -4,35 +4,36 @@ using File.Application.Features.Folders.Commands; using File.Application.Features.Folders.DTOs; using File.Application.Features.Folders.Queries; using File.Infrastructure.Persistence; -using File.Infrastructure.Storage; using MediatR; namespace File.Application.Features.Folders.Handlers; /// /// 查询单文件夹处理器。校验归属后聚合子文件夹与文件清单。 -/// 加载时对 Size==0 的旧文件记录惰性回填（从 S3 HeadObject 取真实大小写回）， -/// 这样移动文件进文件夹后再加载也会自动补全大小。 +/// 请求路径只读取数据库；历史大小由后台修复任务异步确认。 /// public class GetFolderQueryHandler( FileDbContext db, - ICurrentUserContext currentUser, - IS3StorageService s3) : IRequestHandler + ICurrentUserContext currentUser) : IRequestHandler { public async Task Handle(GetFolderQuery req, CancellationToken ct) { var owner = currentUser.GetRequiredUserId(); - var folder = await db.Folders.FirstOrDefaultAsync(f => f.Id == req.Id && f.OwnerId == owner, ct); + var folder = await db.Folders + .AsNoTracking() + .FirstOrDefaultAsync(f => f.Id == req.Id && f.OwnerId == owner, ct); if (folder is null) return null; - var subFolders = await db.Folders.Where(f => f.ParentId == folder.Id).ToListAsync(ct); + var subFolders = await db.Folders + .AsNoTracking() + .Where(f => f.ParentId == folder.Id) + .ToListAsync(ct); var files = await db.Files + .AsNoTracking() .Where(f => f.FolderId == folder.Id) .OrderBy(f => f.Name) .ToListAsync(ct); - await FileSizeBackfillHelper.BackfillAsync(db, s3, files, ct); - return new FolderWithChildrenDto( FolderMapper.ToDto(folder), subFolders.Select(f => FolderMapper.ToDto(f)).ToList(), diff --git a/src/modules/file/Application/Features/Folders/Handlers/GetFolderTreeQueryHandler.cs b/src/modules/file/Application/Features/Folders/Handlers/GetFolderTreeQueryHandler.cs index fc30c6d..72f4286 100644 --- a/src/modules/file/Application/Features/Folders/Handlers/GetFolderTreeQueryHandler.cs +++ b/src/modules/file/Application/Features/Folders/Handlers/GetFolderTreeQueryHandler.cs @@ -4,7 +4,6 @@ using File.Application.Features.Folders.Commands; using File.Application.Features.Folders.DTOs; using File.Application.Features.Folders.Queries; using File.Infrastructure.Persistence; -using File.Infrastructure.Storage; using MediatR; namespace File.Application.Features.Folders.Handlers; @@ -12,22 +11,26 @@ namespace File.Application.Features.Folders.Handlers; /// /// 文件夹树查询处理器。返回当前用户范围内的全部文件夹。 /// 同时计算每个文件夹的递归总大小（自身文件 + 所有子孙文件夹文件），写入 FolderDto.Size。 -/// 计算前先回填历史 Size=0 的文件，保证汇总准确。 +/// 请求路径只汇总数据库中的已持久化大小；历史值由后台修复任务异步确认。 /// public class GetFolderTreeQueryHandler( FileDbContext db, - ICurrentUserContext currentUser, - IS3StorageService s3) : IRequestHandler> + ICurrentUserContext currentUser) : IRequestHandler> { public async Task> Handle(GetFolderTreeQuery req, CancellationToken ct) { var owner = currentUser.GetRequiredUserId(); - var folders = await db.Folders.Where(f => f.OwnerId == owner).ToListAsync(ct); + var folders = await db.Folders + .AsNoTracking() + .Where(f => f.OwnerId == owner) + .ToListAsync(ct); - // 加载该用户全部文件：先回填历史 Size=0 记录（保证文件夹大小准确），再按文件夹汇总 - var files = await db.Files.Where(f => f.OwnerId == owner).ToListAsync(ct); - await FileSizeBackfillHelper.BackfillAsync(db, s3, files, ct); + // 只使用持久化大小汇总，避免 GET 请求触发对象存储 I/O 和数据库写入。 + var files = await db.Files + .AsNoTracking() + .Where(f => f.OwnerId == owner) + .ToListAsync(ct); var filesSizeByFolder = files .Where(f => f.FolderId.HasValue) diff --git a/src/modules/file/Application/Features/Folders/Handlers/GetRootFilesQueryHandler.cs b/src/modules/file/Application/Features/Folders/Handlers/GetRootFilesQueryHandler.cs index b2eb863..cb5498f 100644 --- a/src/modules/file/Application/Features/Folders/Handlers/GetRootFilesQueryHandler.cs +++ b/src/modules/file/Application/Features/Folders/Handlers/GetRootFilesQueryHandler.cs @@ -3,7 +3,6 @@ using File.Application.Common; using File.Application.Features.Folders.DTOs; using File.Application.Features.Folders.Queries; using File.Infrastructure.Persistence; -using File.Infrastructure.Storage; using MediatR; namespace File.Application.Features.Folders.Handlers; @@ -11,23 +10,21 @@ namespace File.Application.Features.Folders.Handlers; /// /// 查询根级文件处理器。返回当前用户名下 FolderId=null 的所有 FileMetaEntity（按名称排序）。 /// 与 GetFolderQueryHandler 配合，使根目录与子文件夹共享同一数据源（DB），而非 S3 列表。 -/// 加载时对 Size==0 的旧记录惰性回填（从 S3 HeadObject 取真实大小写回）。 +/// 请求路径只读取数据库；历史大小由后台修复任务异步确认。 /// public class GetRootFilesQueryHandler( FileDbContext db, - ICurrentUserContext currentUser, - IS3StorageService s3) : IRequestHandler> + ICurrentUserContext currentUser) : IRequestHandler> { public async Task> Handle(GetRootFilesQuery req, CancellationToken ct) { var owner = currentUser.GetRequiredUserId(); var files = await db.Files + .AsNoTracking() .Where(f => f.FolderId == null && f.OwnerId == owner) .OrderBy(f => f.Name) .ToListAsync(ct); - await FileSizeBackfillHelper.BackfillAsync(db, s3, files, ct); - return files.Select(f => new FileMetaDto(f.Id, f.FolderId, f.Name, f.S3Key, f.S3Bucket, f.Size, f.ContentType, f.OwnerId, f.CreatedAt, f.UpdatedAt)).ToList(); } diff --git a/src/modules/file/Application/FileApplicationModule.cs b/src/modules/file/Application/FileApplicationModule.cs index b0720ee..aebcd1e 100644 --- a/src/modules/file/Application/FileApplicationModule.cs +++ b/src/modules/file/Application/FileApplicationModule.cs @@ -1,4 +1,5 @@ using File.Application.Common; +using File.Application.Features.Files.Services; using File.Infrastructure; using FluentValidation; using MediatR; @@ -27,5 +28,6 @@ public class FileApplicationModule : AbpModule // 默认存储桶解析器：所有文件操作统一从此获取 bucket（替代前端参数/硬编码） services.AddScoped(); + services.AddScoped(); } } diff --git a/tests/BuildFlow.File.Tests/FileSizeVerificationProcessorTests.cs b/tests/BuildFlow.File.Tests/FileSizeVerificationProcessorTests.cs new file mode 100644 index 0000000..9fe6341 --- /dev/null +++ b/tests/BuildFlow.File.Tests/FileSizeVerificationProcessorTests.cs @@ -0,0 +1,221 @@ +using File.Application.Features.Files.Services; +using File.Domain.Entities.FileMeta; +using File.Infrastructure.Persistence; +using File.Infrastructure.Storage; +using FluentAssertions; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; +using Moq; +using Xunit; + +namespace BuildFlow.File.Tests; + +/// 后台文件大小修复的结果语义、批量边界与 S3 并发上限测试。 +public class FileSizeVerificationProcessorTests +{ + private const string Bucket = "files"; + + [Fact] + public async Task PositiveObjectSizeUpdatesAndMarksVerified() + { + await using var db = CreateDbContext(); + db.Files.Add(CreateFile(0)); + await db.SaveChangesAsync(); + var s3 = new Mock(); + s3.Setup(x => x.GetObjectSizeAsync(Bucket, "key-0", It.IsAny())) + .ReturnsAsync(128L); + var processor = CreateProcessor(db, s3); + + var processed = await processor.ProcessAsync(100, 4, CancellationToken.None); + + processed.Should().Be(1); + var file = await db.Files.SingleAsync(); + file.Size.Should().Be(128); + file.SizeVerified.Should().BeTrue(); + } + + [Fact] + public async Task ZeroByteObjectMarksVerified() + { + await using var db = CreateDbContext(); + db.Files.Add(CreateFile(0)); + await db.SaveChangesAsync(); + var s3 = new Mock(); + s3.Setup(x => x.GetObjectSizeAsync(Bucket, "key-0", It.IsAny())) + .ReturnsAsync(0L); + var processor = CreateProcessor(db, s3); + + await processor.ProcessAsync(100, 4, CancellationToken.None); + + var file = await db.Files.SingleAsync(); + file.Size.Should().Be(0); + file.SizeVerified.Should().BeTrue(); + } + + [Fact] + public async Task MissingObjectLogsWarningAndMarksVerified() + { + await using var db = CreateDbContext(); + db.Files.Add(CreateFile(0)); + await db.SaveChangesAsync(); + var s3 = new Mock(); + s3.Setup(x => x.GetObjectSizeAsync(Bucket, "key-0", It.IsAny())) + .ReturnsAsync((long?)null); + var logger = new Mock>(); + var processor = new FileSizeVerificationProcessor(db, s3.Object, logger.Object); + + await processor.ProcessAsync(100, 4, CancellationToken.None); + + var file = await db.Files.SingleAsync(); + file.Size.Should().Be(0); + file.SizeVerified.Should().BeTrue(); + WarningCount(logger).Should().Be(1); + } + + [Fact] + public async Task TransientFailureLogsWarningAndRemainsUnverified() + { + await using var db = CreateDbContext(); + db.Files.Add(CreateFile(0)); + await db.SaveChangesAsync(); + var s3 = new Mock(); + s3.Setup(x => x.GetObjectSizeAsync(Bucket, "key-0", It.IsAny())) + .ThrowsAsync(new HttpRequestException("temporary")); + var logger = new Mock>(); + var processor = new FileSizeVerificationProcessor(db, s3.Object, logger.Object); + + await processor.ProcessAsync(100, 4, CancellationToken.None); + + var file = await db.Files.SingleAsync(); + file.SizeVerified.Should().BeFalse(); + WarningCount(logger).Should().Be(1); + } + + [Fact] + public async Task BatchSizeLimitsS3CallsToOldestFiles() + { + await using var db = CreateDbContext(); + db.Files.AddRange(Enumerable.Range(0, 5).Select(CreateFile)); + await db.SaveChangesAsync(); + var s3 = new Mock(); + s3.Setup(x => x.GetObjectSizeAsync( + It.IsAny(), + It.IsAny(), + It.IsAny())) + .ReturnsAsync(1L); + var processor = CreateProcessor(db, s3); + + var processed = await processor.ProcessAsync(2, 2, CancellationToken.None); + + processed.Should().Be(2); + s3.Verify(x => x.GetObjectSizeAsync( + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Exactly(2)); + var verifiedKeys = await db.Files + .Where(file => file.SizeVerified) + .OrderBy(file => file.CreatedAt) + .Select(file => file.S3Key) + .ToListAsync(); + verifiedKeys.Should().Equal("key-0", "key-1"); + } + + [Fact] + public async Task S3CallsRespectMaximumConcurrency() + { + await using var db = CreateDbContext(); + db.Files.AddRange(Enumerable.Range(0, 8).Select(CreateFile)); + await db.SaveChangesAsync(); + var active = 0; + var maximum = 0; + var s3 = new Mock(); + s3.Setup(x => x.GetObjectSizeAsync( + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns(async (string _, string _, CancellationToken ct) => + { + var current = Interlocked.Increment(ref active); + UpdateMaximum(ref maximum, current); + try + { + await Task.Delay(30, ct); + return 1L; + } + finally + { + Interlocked.Decrement(ref active); + } + }); + var processor = CreateProcessor(db, s3); + + await processor.ProcessAsync(8, 3, CancellationToken.None); + + maximum.Should().BeGreaterThan(1); + maximum.Should().BeLessThanOrEqualTo(3); + } + + [Theory] + [InlineData(0, 1)] + [InlineData(1001, 1)] + [InlineData(1, 0)] + [InlineData(1, 33)] + public async Task OutOfRangeLimitsAreRejected(int batchSize, int maxConcurrency) + { + await using var db = CreateDbContext(); + var processor = CreateProcessor(db, new Mock()); + + var act = () => processor.ProcessAsync(batchSize, maxConcurrency, CancellationToken.None); + + await act.Should().ThrowAsync(); + } + + private static FileSizeVerificationProcessor CreateProcessor( + FileDbContext db, + Mock s3) + => new(db, s3.Object, new Mock>().Object); + + private static FileDbContext CreateDbContext() + { + var options = new DbContextOptionsBuilder() + .UseInMemoryDatabase($"file-size-processor-{Guid.NewGuid():N}") + .Options; + return new FileDbContext(options); + } + + private static FileMetaEntity CreateFile(int index) + => new() + { + Id = Guid.NewGuid(), + Name = $"file-{index}.bin", + S3Key = $"key-{index}", + S3Bucket = Bucket, + ContentType = "application/octet-stream", + OwnerId = "user-001", + Size = 0, + SizeVerified = false, + CreatedBy = "test", + CreatedAt = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc).AddMinutes(index), + UpdatedBy = "test", + UpdatedAt = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc).AddMinutes(index), + OperatorIP = "127.0.0.1", + }; + + private static int WarningCount(Mock> logger) + => logger.Invocations.Count(invocation => + invocation.Method.Name == nameof(ILogger.Log) + && invocation.Arguments[0] is LogLevel.Warning); + + private static void UpdateMaximum(ref int maximum, int candidate) + { + while (true) + { + var observed = Volatile.Read(ref maximum); + if (candidate <= observed + || Interlocked.CompareExchange(ref maximum, candidate, observed) == observed) + { + return; + } + } + } +} diff --git a/tests/BuildFlow.File.Tests/FoldersBucketsSettingsTests.cs b/tests/BuildFlow.File.Tests/FoldersBucketsSettingsTests.cs index 39accac..15717b9 100644 --- a/tests/BuildFlow.File.Tests/FoldersBucketsSettingsTests.cs +++ b/tests/BuildFlow.File.Tests/FoldersBucketsSettingsTests.cs @@ -109,7 +109,7 @@ public class FoldersBucketsSettingsTests db.Files.Add(new FileMetaEntity { Name = "root1.txt", S3Key = "k1", S3Bucket = "b", ContentType = "text", OwnerId = OwnerId, Size = 100 }); db.Files.Add(new FileMetaEntity { Name = "other.txt", S3Key = "k2", S3Bucket = "b", ContentType = "text", OwnerId = "other-user", Size = 50 }); await db.SaveChangesAsync(); - var handler = new GetRootFilesQueryHandler(db, user.Object, s3.Object); + var handler = CreateHandler(db, user.Object, s3.Object); var result = await handler.Handle(new GetRootFilesQuery(), CancellationToken.None); @@ -124,13 +124,110 @@ public class FoldersBucketsSettingsTests var createHandler = new CreateFolderCommandHandler(db, user.Object, new Mock().Object); await createHandler.Handle(new CreateFolderCommand(null, "F1"), CancellationToken.None); await createHandler.Handle(new CreateFolderCommand(null, "F2"), CancellationToken.None); - var handler = new GetFolderTreeQueryHandler(db, user.Object, s3.Object); + var handler = CreateHandler(db, user.Object, s3.Object); var result = await handler.Handle(new GetFolderTreeQuery(), CancellationToken.None); result.Should().HaveCount(2); } + [Fact] + public async Task GetRootFiles_DoesNotRepairUnverifiedSizes() + { + var (db, s3, user, _) = CreateDeps("root-read-only"); + db.Files.Add(new FileMetaEntity + { + Name = "pending.txt", + S3Key = "pending-root", + S3Bucket = "files", + ContentType = "text/plain", + OwnerId = OwnerId, + Size = 0, + SizeVerified = false, + }); + await db.SaveChangesAsync(); + s3.Setup(x => x.GetObjectSizeAsync("files", "pending-root", It.IsAny())) + .ReturnsAsync(123L); + var handler = CreateHandler(db, user.Object, s3.Object); + + var result = await handler.Handle(new GetRootFilesQuery(), CancellationToken.None); + + result.Should().ContainSingle().Which.Size.Should().Be(0); + s3.Verify(x => x.GetObjectSizeAsync( + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Never); + db.ChangeTracker.Clear(); + (await db.Files.SingleAsync()).Size.Should().Be(0); + } + + [Fact] + public async Task GetFolder_DoesNotRepairUnverifiedSizes() + { + var (db, s3, user, _) = CreateDeps("folder-read-only"); + var folder = await new CreateFolderCommandHandler(db, user.Object, new Mock().Object) + .Handle(new CreateFolderCommand(null, "待修复"), CancellationToken.None); + db.Files.Add(new FileMetaEntity + { + FolderId = folder.Id, + Name = "pending.txt", + S3Key = "pending-folder", + S3Bucket = "files", + ContentType = "text/plain", + OwnerId = OwnerId, + Size = 0, + SizeVerified = false, + }); + await db.SaveChangesAsync(); + s3.Setup(x => x.GetObjectSizeAsync("files", "pending-folder", It.IsAny())) + .ReturnsAsync(123L); + var handler = CreateHandler(db, user.Object, s3.Object); + + var result = await handler.Handle(new GetFolderQuery(folder.Id), CancellationToken.None); + + result.Should().NotBeNull(); + result!.Files.Should().ContainSingle().Which.Size.Should().Be(0); + s3.Verify(x => x.GetObjectSizeAsync( + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Never); + db.ChangeTracker.Clear(); + (await db.Files.SingleAsync()).Size.Should().Be(0); + } + + [Fact] + public async Task GetFolderTree_DoesNotRepairUnverifiedSizes() + { + var (db, s3, user, _) = CreateDeps("tree-read-only"); + var folder = await new CreateFolderCommandHandler(db, user.Object, new Mock().Object) + .Handle(new CreateFolderCommand(null, "待修复"), CancellationToken.None); + db.Files.Add(new FileMetaEntity + { + FolderId = folder.Id, + Name = "pending.txt", + S3Key = "pending-tree", + S3Bucket = "files", + ContentType = "text/plain", + OwnerId = OwnerId, + Size = 0, + SizeVerified = false, + }); + await db.SaveChangesAsync(); + s3.Setup(x => x.GetObjectSizeAsync("files", "pending-tree", It.IsAny())) + .ReturnsAsync(123L); + var handler = CreateHandler(db, user.Object, s3.Object); + + var result = await handler.Handle(new GetFolderTreeQuery(), CancellationToken.None); + + result.Should().ContainSingle().Which.Size.Should().Be(0); + s3.Verify(x => x.GetObjectSizeAsync( + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Never); + db.ChangeTracker.Clear(); + (await db.Files.SingleAsync()).Size.Should().Be(0); + } + // ===== Buckets(纯 S3 代理,无 DB) ===== [Fact] @@ -288,4 +385,23 @@ public class FoldersBucketsSettingsTests result.Should().BeNull(); } + + private static THandler CreateHandler( + FileDbContext db, + ICurrentUserContext currentUser, + IS3StorageService s3) + where THandler : notnull + { + var dependencies = new Dictionary + { + [typeof(FileDbContext)] = db, + [typeof(ICurrentUserContext)] = currentUser, + [typeof(IS3StorageService)] = s3, + }; + var constructor = typeof(THandler).GetConstructors().Single(); + var arguments = constructor.GetParameters() + .Select(parameter => dependencies[parameter.ParameterType]) + .ToArray(); + return (THandler)constructor.Invoke(arguments); + } }