perf: repair file sizes in background
This commit is contained in:
parent
43a5911a5a
commit
85a95f05f7
@ -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<WebhookDispatcherService>();
|
||||
services.AddHostedService<ImportJobProcessorService>();
|
||||
|
||||
// file 模块:在后台有限并发确认历史文件大小,GET 请求保持纯数据库读取。
|
||||
services.AddOptions<FileSizeVerificationOptions>()
|
||||
.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<FileSizeVerificationService>();
|
||||
|
||||
// monitoring 模块:监控同步调度器(定时轮询外部系统拉取数据)
|
||||
services.Configure<MonitoringOptions>(config.GetSection("Monitoring"));
|
||||
services.AddHostedService<MonitoringSyncSchedulerService>();
|
||||
|
||||
@ -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;
|
||||
|
||||
/// <summary>周期触发历史文件大小确认命令的 Host 后台服务。</summary>
|
||||
public sealed class FileSizeVerificationService(
|
||||
IServiceScopeFactory scopeFactory,
|
||||
IOptionsMonitor<FileSizeVerificationOptions> options,
|
||||
ILogger<FileSizeVerificationService> 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 停止");
|
||||
}
|
||||
|
||||
/// <summary>执行一轮历史文件大小确认。</summary>
|
||||
public async Task<int> RunScanAsync(CancellationToken ct = default)
|
||||
{
|
||||
using var scope = scopeFactory.CreateScope();
|
||||
var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
|
||||
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;
|
||||
}
|
||||
}
|
||||
@ -97,6 +97,14 @@
|
||||
"MaxTotalBytes": 536870912
|
||||
}
|
||||
},
|
||||
"File": {
|
||||
"SizeVerification": {
|
||||
"PollIntervalSeconds": 60,
|
||||
"StartupDelaySeconds": 15,
|
||||
"BatchSize": 100,
|
||||
"MaxConcurrency": 4
|
||||
}
|
||||
},
|
||||
"Workflow": {
|
||||
"Jobs": {
|
||||
"Timer": {
|
||||
|
||||
@ -1,52 +0,0 @@
|
||||
using File.Domain.Entities.FileMeta;
|
||||
using File.Infrastructure.Persistence;
|
||||
using File.Infrastructure.Storage;
|
||||
|
||||
namespace File.Application.Common;
|
||||
|
||||
/// <summary>
|
||||
/// 文件大小回填助手:对 Size==0 的 FileMetaEntity,从 S3 HeadObject 取真实 ContentLength 写回。
|
||||
/// 用于读路径(列表加载)惰性修复早期未记录大小的旧数据——首次加载即回填并持久化,
|
||||
/// 后续加载直接命中 DB 不再访问 S3。对象缺失或空文件保留 Size=0;单个失败不中断整体。
|
||||
/// </summary>
|
||||
public static class FileSizeBackfillHelper
|
||||
{
|
||||
public static async Task BackfillAsync(
|
||||
FileDbContext db,
|
||||
IS3StorageService s3,
|
||||
IReadOnlyList<FileMetaEntity> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,20 @@
|
||||
namespace File.Application.Common;
|
||||
|
||||
/// <summary>历史文件大小后台确认任务的轮询、批量和并发配置。</summary>
|
||||
public class FileSizeVerificationOptions
|
||||
{
|
||||
/// <summary>配置节名称。</summary>
|
||||
public const string SectionName = "File:SizeVerification";
|
||||
|
||||
/// <summary>两轮扫描之间的间隔秒数。</summary>
|
||||
public int PollIntervalSeconds { get; set; } = 60;
|
||||
|
||||
/// <summary>Host 启动后的首次扫描延迟秒数。</summary>
|
||||
public int StartupDelaySeconds { get; set; } = 15;
|
||||
|
||||
/// <summary>单轮最多处理的未确认文件数。</summary>
|
||||
public int BatchSize { get; set; } = 100;
|
||||
|
||||
/// <summary>单轮查询 S3 对象元数据的最大并发数。</summary>
|
||||
public int MaxConcurrency { get; set; } = 4;
|
||||
}
|
||||
@ -0,0 +1,8 @@
|
||||
using MediatR;
|
||||
|
||||
namespace File.Application.Features.Files.Commands;
|
||||
|
||||
/// <summary>按有限批量和有限并发确认历史文件大小的后台命令。</summary>
|
||||
public sealed record ProcessUnverifiedFileSizesCommand(
|
||||
int BatchSize,
|
||||
int MaxConcurrency) : IRequest<int>;
|
||||
@ -0,0 +1,14 @@
|
||||
using File.Application.Features.Files.Commands;
|
||||
using File.Application.Features.Files.Services;
|
||||
using MediatR;
|
||||
|
||||
namespace File.Application.Features.Files.Handlers;
|
||||
|
||||
/// <summary>将后台文件大小确认命令委托给模块内处理器。</summary>
|
||||
public sealed class ProcessUnverifiedFileSizesCommandHandler(
|
||||
FileSizeVerificationProcessor processor)
|
||||
: IRequestHandler<ProcessUnverifiedFileSizesCommand, int>
|
||||
{
|
||||
public Task<int> Handle(ProcessUnverifiedFileSizesCommand request, CancellationToken ct)
|
||||
=> processor.ProcessAsync(request.BatchSize, request.MaxConcurrency, ct);
|
||||
}
|
||||
@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 分批确认历史文件的对象大小。S3 查询有限并发执行,EF 实体只在并发阶段结束后统一修改并保存。
|
||||
/// </summary>
|
||||
public sealed class FileSizeVerificationProcessor(
|
||||
FileDbContext db,
|
||||
IS3StorageService s3,
|
||||
ILogger<FileSizeVerificationProcessor> logger)
|
||||
{
|
||||
/// <summary>处理一批最早创建且尚未确认大小的文件,返回本轮尝试数量。</summary>
|
||||
public async Task<int> 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;
|
||||
}
|
||||
}
|
||||
@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 查询单文件夹处理器。校验归属后聚合子文件夹与文件清单。
|
||||
/// 加载时对 Size==0 的旧文件记录惰性回填(从 S3 HeadObject 取真实大小写回),
|
||||
/// 这样移动文件进文件夹后再加载也会自动补全大小。
|
||||
/// 请求路径只读取数据库;历史大小由后台修复任务异步确认。
|
||||
/// </summary>
|
||||
public class GetFolderQueryHandler(
|
||||
FileDbContext db,
|
||||
ICurrentUserContext currentUser,
|
||||
IS3StorageService s3) : IRequestHandler<GetFolderQuery, FolderWithChildrenDto?>
|
||||
ICurrentUserContext currentUser) : IRequestHandler<GetFolderQuery, FolderWithChildrenDto?>
|
||||
{
|
||||
public async Task<FolderWithChildrenDto?> 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(),
|
||||
|
||||
@ -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;
|
||||
/// <summary>
|
||||
/// 文件夹树查询处理器。返回当前用户范围内的全部文件夹。
|
||||
/// 同时计算每个文件夹的递归总大小(自身文件 + 所有子孙文件夹文件),写入 FolderDto.Size。
|
||||
/// 计算前先回填历史 Size=0 的文件,保证汇总准确。
|
||||
/// 请求路径只汇总数据库中的已持久化大小;历史值由后台修复任务异步确认。
|
||||
/// </summary>
|
||||
public class GetFolderTreeQueryHandler(
|
||||
FileDbContext db,
|
||||
ICurrentUserContext currentUser,
|
||||
IS3StorageService s3) : IRequestHandler<GetFolderTreeQuery, List<FolderDto>>
|
||||
ICurrentUserContext currentUser) : IRequestHandler<GetFolderTreeQuery, List<FolderDto>>
|
||||
{
|
||||
public async Task<List<FolderDto>> 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)
|
||||
|
||||
@ -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;
|
||||
/// <summary>
|
||||
/// 查询根级文件处理器。返回当前用户名下 FolderId=null 的所有 FileMetaEntity(按名称排序)。
|
||||
/// 与 GetFolderQueryHandler 配合,使根目录与子文件夹共享同一数据源(DB),而非 S3 列表。
|
||||
/// 加载时对 Size==0 的旧记录惰性回填(从 S3 HeadObject 取真实大小写回)。
|
||||
/// 请求路径只读取数据库;历史大小由后台修复任务异步确认。
|
||||
/// </summary>
|
||||
public class GetRootFilesQueryHandler(
|
||||
FileDbContext db,
|
||||
ICurrentUserContext currentUser,
|
||||
IS3StorageService s3) : IRequestHandler<GetRootFilesQuery, List<FileMetaDto>>
|
||||
ICurrentUserContext currentUser) : IRequestHandler<GetRootFilesQuery, List<FileMetaDto>>
|
||||
{
|
||||
public async Task<List<FileMetaDto>> 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();
|
||||
}
|
||||
|
||||
@ -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<IBucketResolver, BucketResolver>();
|
||||
services.AddScoped<FileSizeVerificationProcessor>();
|
||||
}
|
||||
}
|
||||
|
||||
221
tests/BuildFlow.File.Tests/FileSizeVerificationProcessorTests.cs
Normal file
221
tests/BuildFlow.File.Tests/FileSizeVerificationProcessorTests.cs
Normal file
@ -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;
|
||||
|
||||
/// <summary>后台文件大小修复的结果语义、批量边界与 S3 并发上限测试。</summary>
|
||||
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<IS3StorageService>();
|
||||
s3.Setup(x => x.GetObjectSizeAsync(Bucket, "key-0", It.IsAny<CancellationToken>()))
|
||||
.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<IS3StorageService>();
|
||||
s3.Setup(x => x.GetObjectSizeAsync(Bucket, "key-0", It.IsAny<CancellationToken>()))
|
||||
.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<IS3StorageService>();
|
||||
s3.Setup(x => x.GetObjectSizeAsync(Bucket, "key-0", It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync((long?)null);
|
||||
var logger = new Mock<ILogger<FileSizeVerificationProcessor>>();
|
||||
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<IS3StorageService>();
|
||||
s3.Setup(x => x.GetObjectSizeAsync(Bucket, "key-0", It.IsAny<CancellationToken>()))
|
||||
.ThrowsAsync(new HttpRequestException("temporary"));
|
||||
var logger = new Mock<ILogger<FileSizeVerificationProcessor>>();
|
||||
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<IS3StorageService>();
|
||||
s3.Setup(x => x.GetObjectSizeAsync(
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<CancellationToken>()))
|
||||
.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<string>(),
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<CancellationToken>()), 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<IS3StorageService>();
|
||||
s3.Setup(x => x.GetObjectSizeAsync(
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<CancellationToken>()))
|
||||
.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<IS3StorageService>());
|
||||
|
||||
var act = () => processor.ProcessAsync(batchSize, maxConcurrency, CancellationToken.None);
|
||||
|
||||
await act.Should().ThrowAsync<ArgumentOutOfRangeException>();
|
||||
}
|
||||
|
||||
private static FileSizeVerificationProcessor CreateProcessor(
|
||||
FileDbContext db,
|
||||
Mock<IS3StorageService> s3)
|
||||
=> new(db, s3.Object, new Mock<ILogger<FileSizeVerificationProcessor>>().Object);
|
||||
|
||||
private static FileDbContext CreateDbContext()
|
||||
{
|
||||
var options = new DbContextOptionsBuilder<FileDbContext>()
|
||||
.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<ILogger<FileSizeVerificationProcessor>> 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -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<GetRootFilesQueryHandler>(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<IMediator>().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<GetFolderTreeQueryHandler>(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<CancellationToken>()))
|
||||
.ReturnsAsync(123L);
|
||||
var handler = CreateHandler<GetRootFilesQueryHandler>(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<string>(),
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<CancellationToken>()), 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<IMediator>().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<CancellationToken>()))
|
||||
.ReturnsAsync(123L);
|
||||
var handler = CreateHandler<GetFolderQueryHandler>(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<string>(),
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<CancellationToken>()), 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<IMediator>().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<CancellationToken>()))
|
||||
.ReturnsAsync(123L);
|
||||
var handler = CreateHandler<GetFolderTreeQueryHandler>(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<string>(),
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<CancellationToken>()), 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<THandler>(
|
||||
FileDbContext db,
|
||||
ICurrentUserContext currentUser,
|
||||
IS3StorageService s3)
|
||||
where THandler : notnull
|
||||
{
|
||||
var dependencies = new Dictionary<Type, object>
|
||||
{
|
||||
[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);
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user