From 034f1eacb8dc2f11d6654d729ab24658be7af087 Mon Sep 17 00:00:00 2001 From: biss Date: Sun, 26 Jul 2026 21:05:18 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AC=AC=E4=BA=8C=E9=98=B6=E6=AE=B5=E4=BC=98?= =?UTF-8?q?=E5=8C=96=E5=B7=B2=E5=AE=8C=E6=88=90=EF=BC=8C=E9=87=8D=E7=82=B9?= =?UTF-8?q?=E6=98=AF=E6=8F=90=E9=AB=98=E5=90=8E=E5=8F=B0=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E5=90=9E=E5=90=90=E5=B9=B6=E9=99=8D=E4=BD=8E=20Outbox=20?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=BA=93=E5=BC=80=E9=94=80=E3=80=82=20Rabbit?= =?UTF-8?q?MQ=20=E9=BB=98=E8=AE=A4=E4=BB=8E=E5=8D=95=E4=B8=80=E5=9B=9E?= =?UTF-8?q?=E8=B0=83=E9=80=9A=E9=81=93=E6=8F=90=E5=8D=87=E4=B8=BA=E4=B8=89?= =?UTF-8?q?=E7=B1=BB=E4=BB=BB=E5=8A=A1=E7=8B=AC=E7=AB=8B=E5=B9=B6=E8=A1=8C?= =?UTF-8?q?=EF=BC=8C=E8=B7=A8=E7=B1=BB=E5=9E=8B=E5=B9=B6=E5=8F=91=E8=83=BD?= =?UTF-8?q?=E5=8A=9B=E7=94=B1=201=20=E6=8F=90=E5=8D=87=E5=88=B0=203?= =?UTF-8?q?=EF=BC=9B=E6=AF=8F=E7=B1=BB=E8=BF=98=E8=83=BD=E7=8B=AC=E7=AB=8B?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=201=E2=80=9316=20=E4=B8=AA=E6=B6=88=E8=B4=B9?= =?UTF-8?q?=E8=80=85=E3=80=82[RabbitMqBackgroundJobs.cs=20(line=20143)](E:?= =?UTF-8?q?/jiaowu/src/Jiaowu.Api/Infrastructure/BackgroundJobs/RabbitMqBa?= =?UTF-8?q?ckgroundJobs.cs:143)=20InMemory=20=E5=BC=80=E5=8F=91=E6=A8=A1?= =?UTF-8?q?=E5=BC=8F=E5=90=8C=E6=AD=A5=E6=94=B9=E4=B8=BA=E6=8C=89=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E7=B1=BB=E5=9E=8B=E9=9A=94=E7=A6=BB=E9=98=9F=E5=88=97?= =?UTF-8?q?=EF=BC=8C=E9=81=BF=E5=85=8D=E6=9F=90=E7=B1=BB=E9=95=BF=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E5=A0=B5=E5=A1=9E=E5=85=B6=E4=BB=96=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E3=80=82=20=E7=A7=9F=E7=BA=A6=E6=81=A2=E5=A4=8D=E6=A3=80?= =?UTF-8?q?=E6=9F=A5=E4=BB=8E=E2=80=9C=E6=AF=8F=E5=8F=91=E5=B8=83=E4=B8=80?= =?UTF-8?q?=E6=9D=A1=E6=89=A7=E8=A1=8C=E4=B8=80=E6=AC=A1=E2=80=9D=E6=94=B9?= =?UTF-8?q?=E4=B8=BA=E9=BB=98=E8=AE=A4=E6=AF=8F=2060=20=E7=A7=92=E7=BB=B4?= =?UTF-8?q?=E6=8A=A4=E4=B8=80=E6=AC=A1=E3=80=82=20=E5=90=AF=E5=8A=A8?= =?UTF-8?q?=E6=81=A2=E5=A4=8D=E6=94=B9=E4=B8=BA=E6=95=B0=E6=8D=AE=E5=BA=93?= =?UTF-8?q?=20NOT=20EXISTS=20=E6=9F=A5=E8=AF=A2=EF=BC=8C=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E6=8A=8A=E5=85=A8=E9=83=A8=E5=8E=86=E5=8F=B2=20Outbox=20?= =?UTF-8?q?=E5=8A=A0=E8=BD=BD=E8=BF=9B=E5=86=85=E5=AD=98=E3=80=82[Backgrou?= =?UTF-8?q?ndJobOutboxPublisher.cs=20(line=209)](E:/jiaowu/src/Jiaowu.Api/?= =?UTF-8?q?Infrastructure/BackgroundJobs/BackgroundJobOutboxPublisher.cs:9?= =?UTF-8?q?)=20=E5=AE=8C=E6=88=90=E6=B6=88=E6=81=AF=E9=BB=98=E8=AE=A4?= =?UTF-8?q?=E4=BF=9D=E7=95=99=2014=20=E5=A4=A9=EF=BC=8C=E4=B9=8B=E5=90=8E?= =?UTF-8?q?=E6=8C=89=E6=AF=8F=E6=89=B9=20500=20=E6=9D=A1=E6=B8=85=E7=90=86?= =?UTF-8?q?=EF=BC=8C=E5=B9=B6=E5=A2=9E=E5=8A=A0=E5=AF=B9=E5=BA=94=E7=BB=84?= =?UTF-8?q?=E5=90=88=E7=B4=A2=E5=BC=95=E3=80=82=20/health/messaging=20?= =?UTF-8?q?=E7=8E=B0=E5=9C=A8=E8=BF=94=E5=9B=9E=E5=90=84=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E7=A7=AF=E5=8E=8B=E9=87=8F=E3=80=81=E8=BF=87=E6=9C=9F=E7=A7=9F?= =?UTF-8?q?=E7=BA=A6=E5=92=8C=E6=9C=80=E8=80=81=E4=BB=BB=E5=8A=A1=E7=AD=89?= =?UTF-8?q?=E5=BE=85=E6=97=B6=E9=97=B4=E3=80=82[BackgroundJobMonitoringSer?= =?UTF-8?q?vice.cs=20(line=2016)](E:/jiaowu/src/Jiaowu.Api/Infrastructure/?= =?UTF-8?q?BackgroundJobs/BackgroundJobMonitoringService.cs:16)=20?= =?UTF-8?q?=E6=96=B0=E5=A2=9E=20Jiaowu.BackgroundJobs=20=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E6=97=B6=E6=8C=87=E6=A0=87=EF=BC=8C=E8=A6=86=E7=9B=96=E5=8F=91?= =?UTF-8?q?=E5=B8=83=E9=87=8F=E3=80=81=E5=A4=84=E7=90=86=E9=87=8F=E3=80=81?= =?UTF-8?q?=E5=8F=91=E5=B8=83=E8=80=97=E6=97=B6=E3=80=81=E5=A4=84=E7=90=86?= =?UTF-8?q?=E8=80=97=E6=97=B6=E5=92=8C=E6=B8=85=E7=90=86=E9=87=8F=E3=80=82?= =?UTF-8?q?[BackgroundJobTelemetry.cs=20(line=209)](E:/jiaowu/src/Jiaowu.A?= =?UTF-8?q?pi/Infrastructure/BackgroundJobs/BackgroundJobTelemetry.cs:9)?= =?UTF-8?q?=20=E9=85=8D=E7=BD=AE=E3=80=81Compose=20=E5=92=8C=E8=B0=83?= =?UTF-8?q?=E4=BC=98=E5=BB=BA=E8=AE=AE=E5=B7=B2=E6=9B=B4=E6=96=B0=E3=80=82?= =?UTF-8?q?[README.md=20(line=20257)](E:/jiaowu/README.md:257)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.docker.example | 3 + .env.example | 3 + README.md | 21 ++- compose.example.yml | 3 + .../BackgroundJobMonitoringService.cs | 56 +++++++ .../BackgroundJobs/BackgroundJobOptions.cs | 16 ++ .../BackgroundJobOutboxPublisher.cs | 139 ++++++++++++---- .../BackgroundJobs/BackgroundJobRunner.cs | 15 ++ .../BackgroundJobs/BackgroundJobTelemetry.cs | 74 +++++++++ .../BackgroundJobs/BackgroundJobTransport.cs | 25 +-- .../InMemoryBackgroundJobWorker.cs | 37 +++-- .../BackgroundJobs/RabbitMqBackgroundJobs.cs | 153 ++++++++++-------- .../Persistence/AppDbContext.cs | 1 + .../Persistence/DevelopmentSqliteMigrator.cs | 4 + ...0726220000_BackgroundJobOutbox.Designer.cs | 2 + .../20260726220000_BackgroundJobOutbox.cs | 5 + .../MySql/AppDbContextModelSnapshot.cs | 2 + src/Jiaowu.Api/Program.cs | 42 ++++- src/Jiaowu.Api/appsettings.json | 8 +- .../AutomaticScheduleGeneratorTests.cs | 14 +- .../BackgroundJobOutboxTests.cs | 91 ++++++++++- 21 files changed, 579 insertions(+), 135 deletions(-) create mode 100644 src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobMonitoringService.cs create mode 100644 src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobTelemetry.cs diff --git a/.env.docker.example b/.env.docker.example index 060499b..a93f1e6 100644 --- a/.env.docker.example +++ b/.env.docker.example @@ -10,6 +10,9 @@ MYSQL_ROOT_PASSWORD= RABBITMQ_USER=jiaowu RABBITMQ_PASSWORD= +BACKGROUND_JOB_AUTOMATIC_SCHEDULE_CONCURRENCY=1 +BACKGROUND_JOB_SCHEDULE_PUBLISH_CONCURRENCY=1 +BACKGROUND_JOB_MAKEUP_EXAM_AUTO_CONCURRENCY=1 JWT_KEY= ALLOWED_HOSTS=localhost diff --git a/.env.example b/.env.example index b118a3e..dfc5197 100644 --- a/.env.example +++ b/.env.example @@ -11,6 +11,9 @@ ConnectionStrings__MySql="Server=db.example.edu.cn;Port=3306;Database=jiaowu;Use # 单机可继续使用 InMemory；多实例生产部署建议启用 RabbitMQ。 BackgroundJobs__Transport=InMemory +BackgroundJobs__AutomaticScheduleConcurrency=1 +BackgroundJobs__SchedulePublishConcurrency=1 +BackgroundJobs__MakeupExamAutoConcurrency=1 # RabbitMq__HostName=rabbitmq.example.edu.cn # RabbitMq__Port=5671 # RabbitMq__UserName=jiaowu diff --git a/README.md b/README.md index bed2381..05831d2 100644 --- a/README.md +++ b/README.md @@ -224,7 +224,8 @@ SQLite 只用于本地开发：新库通过 `EnsureCreated` 建立，已有开 - `/health/cache`：检查可选 Redis；未配置 Redis 时返回 `disabled`，Redis 故障不会影响数据库就绪探针。 - `/health/messaging`：检查后台任务传输；单机内存队列返回 `memory`，启用 - RabbitMQ 时实际检查代理连接。 + RabbitMQ 时实际检查代理连接，同时返回各 Outbox 状态的积压数、过期租约数和 + 最老未完成任务的等待时间。 ### 查询缓存与 Redis @@ -270,6 +271,9 @@ BackgroundJobs__Transport=InMemory ```text BackgroundJobs__Transport=RabbitMq +BackgroundJobs__AutomaticScheduleConcurrency=1 +BackgroundJobs__SchedulePublishConcurrency=1 +BackgroundJobs__MakeupExamAutoConcurrency=1 RabbitMq__HostName=rabbitmq.example.edu.cn RabbitMq__Port=5671 RabbitMq__UserName=jiaowu @@ -280,9 +284,18 @@ RabbitMq__TlsServerName=rabbitmq.example.edu.cn ``` RabbitMQ 传输使用持久消息、发布确认、手动消费确认、每种任务独立队列和死信队列。 -默认创建 Quorum Queue，重任务的消费者预取数为 1。MySQL 或 RabbitMQ 暂时不可用时， -未完成消息会根据 Outbox 状态和租约继续补投，消费者必须保持业务处理幂等。迁移服务 -应先应用 `BackgroundJobOutbox` 数据库迁移，再启动应用实例。 +默认创建 Quorum Queue，重任务的消费者预取数为 1。三类任务各自至少有一个消费者， +因此不同类型的任务不会再互相阻塞；单类任务的并发度可独立设置为 1-16。自动排课 +通常最消耗 CPU，建议先保持为 1，再根据 CPU、数据库连接池和任务等待时间逐级调到 +2 或 3；不要只提高预取数。 + +Outbox 租约恢复改为按维护周期执行，避免积压发布时每条消息都额外扫描数据库；已完成 +消息默认保留 14 天并按每批 500 条清理，可使用 `CompletedRetentionDays`、 +`MaintenanceIntervalSeconds` 和 `CleanupBatchSize` 调整。应用暴露 +`Jiaowu.BackgroundJobs` Meter，其中包含发布量、处理量、发布耗时、处理耗时和清理量， +可接入现有 OpenTelemetry/运行时指标采集器。MySQL 或 RabbitMQ 暂时不可用时，未完成 +消息会根据 Outbox 状态和租约继续补投。迁移服务应先应用 +`BackgroundJobOutbox` 数据库迁移，再启动应用实例。 ## 跨平台发布与 Docker diff --git a/compose.example.yml b/compose.example.yml index 17996b0..fce0a6e 100644 --- a/compose.example.yml +++ b/compose.example.yml @@ -17,6 +17,9 @@ x-jiaowu-environment: &jiaowu-environment Cache__Enabled: "true" Cache__KeyPrefix: "jiaowu:v1" BackgroundJobs__Transport: RabbitMq + BackgroundJobs__AutomaticScheduleConcurrency: "${BACKGROUND_JOB_AUTOMATIC_SCHEDULE_CONCURRENCY:-1}" + BackgroundJobs__SchedulePublishConcurrency: "${BACKGROUND_JOB_SCHEDULE_PUBLISH_CONCURRENCY:-1}" + BackgroundJobs__MakeupExamAutoConcurrency: "${BACKGROUND_JOB_MAKEUP_EXAM_AUTO_CONCURRENCY:-1}" RabbitMq__HostName: rabbitmq RabbitMq__Port: "5672" RabbitMq__UserName: "${RABBITMQ_USER:-jiaowu}" diff --git a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobMonitoringService.cs b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobMonitoringService.cs new file mode 100644 index 0000000..c75b2a3 --- /dev/null +++ b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobMonitoringService.cs @@ -0,0 +1,56 @@ +using Jiaowu.Api.Domain.System; +using Jiaowu.Api.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; + +namespace Jiaowu.Api.Infrastructure.BackgroundJobs; + +public sealed record BackgroundJobBacklogSnapshot( + int Pending, + int Publishing, + int Published, + int Processing, + int ExpiredLeases, + DateTime? OldestUnfinishedAt, + double? OldestUnfinishedAgeSeconds); + +public sealed class BackgroundJobMonitoringService(AppDbContext db) +{ + public async Task GetSnapshotAsync( + CancellationToken cancellationToken) + { + var counts = await db.BackgroundJobOutboxMessages.AsNoTracking() + .Where(x => x.State != BackgroundJobOutboxState.Completed) + .GroupBy(x => x.State) + .Select(group => new { State = group.Key, Count = group.Count() }) + .ToListAsync(cancellationToken); + var countByState = counts.ToDictionary(x => x.State, x => x.Count); + var now = DateTime.UtcNow; + var expiredLeases = await db.BackgroundJobOutboxMessages.AsNoTracking() + .CountAsync( + x => (x.State == BackgroundJobOutboxState.Publishing || + x.State == BackgroundJobOutboxState.Processing) && + x.LeaseExpiresAt != null && + x.LeaseExpiresAt < now, + cancellationToken); + var oldestUnfinishedAt = await db.BackgroundJobOutboxMessages + .AsNoTracking() + .Where(x => x.State != BackgroundJobOutboxState.Completed) + .Select(x => (DateTime?)x.CreatedAt) + .MinAsync(cancellationToken); + double? ageSeconds = oldestUnfinishedAt.HasValue + ? Math.Max(0, (now - oldestUnfinishedAt.Value).TotalSeconds) + : null; + + return new BackgroundJobBacklogSnapshot( + GetCount(BackgroundJobOutboxState.Pending), + GetCount(BackgroundJobOutboxState.Publishing), + GetCount(BackgroundJobOutboxState.Published), + GetCount(BackgroundJobOutboxState.Processing), + expiredLeases, + oldestUnfinishedAt, + ageSeconds); + + int GetCount(BackgroundJobOutboxState state) => + countByState.GetValueOrDefault(state); + } +} diff --git a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobOptions.cs b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobOptions.cs index cc23ef5..ba2babf 100644 --- a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobOptions.cs +++ b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobOptions.cs @@ -1,3 +1,5 @@ +using Jiaowu.Api.Domain.System; + namespace Jiaowu.Api.Infrastructure.BackgroundJobs; public sealed class BackgroundJobOptions @@ -8,13 +10,27 @@ public sealed class BackgroundJobOptions public int PollIntervalMilliseconds { get; set; } = 500; public int LeaseSeconds { get; set; } = 120; public ushort PrefetchCount { get; set; } = 1; + public int AutomaticScheduleConcurrency { get; set; } = 1; + public int SchedulePublishConcurrency { get; set; } = 1; + public int MakeupExamAutoConcurrency { get; set; } = 1; public string Exchange { get; set; } = "jiaowu.background-jobs"; public string QueuePrefix { get; set; } = "jiaowu.background-jobs"; public bool UseQuorumQueues { get; set; } = true; public int ProcessingAttemptLimit { get; set; } = 5; + public int MaintenanceIntervalSeconds { get; set; } = 60; + public int CompletedRetentionDays { get; set; } = 14; + public int CleanupBatchSize { get; set; } = 500; public bool UsesRabbitMq => Transport.Equals("RabbitMq", StringComparison.OrdinalIgnoreCase); + + public int ConsumerConcurrency(BackgroundJobKind kind) => kind switch + { + BackgroundJobKind.AutomaticSchedule => AutomaticScheduleConcurrency, + BackgroundJobKind.SchedulePublish => SchedulePublishConcurrency, + BackgroundJobKind.MakeupExamAuto => MakeupExamAutoConcurrency, + _ => throw new ArgumentOutOfRangeException(nameof(kind), kind, null) + }; } public sealed class RabbitMqOptions diff --git a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobOutboxPublisher.cs b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobOutboxPublisher.cs index beeb81f..b97a808 100644 --- a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobOutboxPublisher.cs +++ b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobOutboxPublisher.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using Jiaowu.Api.Domain.Academic; using Jiaowu.Api.Domain.System; using Jiaowu.Api.Infrastructure.Persistence; @@ -9,17 +10,26 @@ public sealed class BackgroundJobOutboxPublisher( IServiceScopeFactory scopeFactory, IBackgroundJobTransport transport, BackgroundJobOptions options, + BackgroundJobTelemetry telemetry, ILogger logger) : BackgroundService { protected override async Task ExecuteAsync(CancellationToken stoppingToken) { await PrepareForStartupAsync(stoppingToken); + var nextMaintenanceAt = DateTime.MinValue; while (!stoppingToken.IsCancellationRequested) { try { - await RecoverExpiredProcessingAsync(stoppingToken); + if (DateTime.UtcNow >= nextMaintenanceAt) + { + await RecoverExpiredProcessingAsync(stoppingToken); + await CleanupCompletedAsync(stoppingToken); + nextMaintenanceAt = DateTime.UtcNow.AddSeconds( + options.MaintenanceIntervalSeconds); + } + var claimed = await ClaimNextAsync(stoppingToken); if (claimed is null) { @@ -62,51 +72,107 @@ public sealed class BackgroundJobOutboxPublisher( await AddMissingOutboxMessagesAsync(db, cancellationToken); } - private static async Task AddMissingOutboxMessagesAsync( + private async Task AddMissingOutboxMessagesAsync( AppDbContext db, CancellationToken cancellationToken) { - var existing = await db.BackgroundJobOutboxMessages.AsNoTracking() - .Select(x => new { x.JobKind, x.JobId }) - .ToListAsync(cancellationToken); - var existingKeys = existing - .Select(x => (x.JobKind, x.JobId)) - .ToHashSet(); - var automaticJobs = await db.AutomaticScheduleJobs.AsNoTracking() - .Where(x => x.Status == AutomaticScheduleJobStatus.Queued || - x.Status == AutomaticScheduleJobStatus.Running) + .Where(x => + (x.Status == AutomaticScheduleJobStatus.Queued || + x.Status == AutomaticScheduleJobStatus.Running) && + !db.BackgroundJobOutboxMessages.Any(message => + message.JobKind == BackgroundJobKind.AutomaticSchedule && + message.JobId == x.Id)) .Select(x => x.Id) .ToListAsync(cancellationToken); var publishJobs = await db.SchedulePublishJobs.AsNoTracking() - .Where(x => x.Status == SchedulePublishJobStatus.Queued || - x.Status == SchedulePublishJobStatus.Running) + .Where(x => + (x.Status == SchedulePublishJobStatus.Queued || + x.Status == SchedulePublishJobStatus.Running) && + !db.BackgroundJobOutboxMessages.Any(message => + message.JobKind == BackgroundJobKind.SchedulePublish && + message.JobId == x.Id)) .Select(x => x.Id) .ToListAsync(cancellationToken); var makeupJobs = await db.MakeupExamAutoJobs.AsNoTracking() - .Where(x => x.Status == MakeupExamAutoJobStatus.Queued || - x.Status == MakeupExamAutoJobStatus.Running) + .Where(x => + (x.Status == MakeupExamAutoJobStatus.Queued || + x.Status == MakeupExamAutoJobStatus.Running) && + !db.BackgroundJobOutboxMessages.Any(message => + message.JobKind == BackgroundJobKind.MakeupExamAuto && + message.JobId == x.Id)) .Select(x => x.Id) .ToListAsync(cancellationToken); - AddMissing(BackgroundJobKind.AutomaticSchedule, automaticJobs); - AddMissing(BackgroundJobKind.SchedulePublish, publishJobs); - AddMissing(BackgroundJobKind.MakeupExamAuto, makeupJobs); - - if (db.ChangeTracker.HasChanges()) - await db.SaveChangesAsync(cancellationToken); - - void AddMissing(BackgroundJobKind kind, IEnumerable jobIds) + var missingKeys = automaticJobs + .Select(id => (BackgroundJobKind.AutomaticSchedule, id)) + .Concat(publishJobs.Select(id => + (BackgroundJobKind.SchedulePublish, id))) + .Concat(makeupJobs.Select(id => + (BackgroundJobKind.MakeupExamAuto, id))) + .ToList(); + foreach (var (kind, jobId) in missingKeys) { - foreach (var jobId in jobIds) - { - if (existingKeys.Add((kind, jobId))) - { - db.BackgroundJobOutboxMessages.Add( - BackgroundJobOutboxMessage.Create(kind, jobId)); - } - } + db.BackgroundJobOutboxMessages.Add( + BackgroundJobOutboxMessage.Create(kind, jobId)); } + + if (!db.ChangeTracker.HasChanges()) + return; + + try + { + await db.SaveChangesAsync(cancellationToken); + } + catch (DbUpdateException) + { + db.ChangeTracker.Clear(); + foreach (var (kind, jobId) in missingKeys) + { + var recovered = await db.BackgroundJobOutboxMessages + .AsNoTracking() + .AnyAsync( + message => message.JobKind == kind && + message.JobId == jobId, + cancellationToken); + if (!recovered) + throw; + } + + logger.LogInformation( + "Another application instance recovered {Count} missing outbox messages first.", + missingKeys.Count); + } + } + + private async Task CleanupCompletedAsync(CancellationToken cancellationToken) + { + await using var scope = scopeFactory.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var cutoff = DateTime.UtcNow.AddDays(-options.CompletedRetentionDays); + var completedIds = await db.BackgroundJobOutboxMessages.AsNoTracking() + .Where(x => + x.State == BackgroundJobOutboxState.Completed && + x.CompletedAt != null && + x.CompletedAt < cutoff) + .OrderBy(x => x.CompletedAt) + .Select(x => x.Id) + .Take(options.CleanupBatchSize) + .ToListAsync(cancellationToken); + if (completedIds.Count == 0) + return; + + foreach (var id in completedIds) + { + db.BackgroundJobOutboxMessages.Remove( + new BackgroundJobOutboxMessage { Id = id }); + } + await db.SaveChangesAsync(cancellationToken); + telemetry.RecordCleaned(completedIds.Count); + logger.LogInformation( + "Removed {Count} completed background job outbox messages older than {Cutoff}.", + completedIds.Count, + cutoff); } private async Task RecoverExpiredProcessingAsync( @@ -175,6 +241,7 @@ public sealed class BackgroundJobOutboxPublisher( Guid token, CancellationToken cancellationToken) { + var startedAt = Stopwatch.GetTimestamp(); try { await transport.PublishAsync(message, cancellationToken); @@ -193,6 +260,11 @@ public sealed class BackgroundJobOutboxPublisher( .SetProperty(x => x.ProcessingToken, (Guid?)null) .SetProperty(x => x.LeaseExpiresAt, (DateTime?)null), cancellationToken); + telemetry.RecordPublish( + message.JobKind, + transport.IsDurable, + "published", + Stopwatch.GetElapsedTime(startedAt)); } catch (Exception exception) { @@ -213,6 +285,11 @@ public sealed class BackgroundJobOutboxPublisher( .SetProperty(x => x.LeaseExpiresAt, (DateTime?)null) .SetProperty(x => x.LastError, messageText), cancellationToken); + telemetry.RecordPublish( + message.JobKind, + transport.IsDurable, + "failed", + Stopwatch.GetElapsedTime(startedAt)); throw; } } diff --git a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobRunner.cs b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobRunner.cs index 79b2bb1..f0a57e9 100644 --- a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobRunner.cs +++ b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobRunner.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using Jiaowu.Api.Domain.Academic; using Jiaowu.Api.Domain.System; using Jiaowu.Api.Infrastructure.Exams; @@ -28,12 +29,14 @@ public sealed class BackgroundJobRunner( IServiceScopeFactory scopeFactory, IBackgroundJobTransport transport, BackgroundJobOptions options, + BackgroundJobTelemetry telemetry, ILogger logger) { public async Task RunAsync( BackgroundJobEnvelope message, CancellationToken cancellationToken) { + var startedAt = Stopwatch.GetTimestamp(); var token = Guid.NewGuid(); var claim = await TryClaimAsync(message, token, cancellationToken); if (!claim.Claimed) @@ -48,6 +51,10 @@ public sealed class BackgroundJobRunner( await MarkJobRetryLimitExceededAsync(message, cancellationToken); await MarkCompletedAsync(message.OutboxMessageId, token, cancellationToken); + telemetry.RecordProcessing( + message.JobKind, + "retry-limit", + Stopwatch.GetElapsedTime(startedAt)); return BackgroundJobRunResult.Completed; } @@ -85,6 +92,10 @@ public sealed class BackgroundJobRunner( await MarkCompletedAsync(message.OutboxMessageId, token, cancellationToken); + telemetry.RecordProcessing( + message.JobKind, + "completed", + Stopwatch.GetElapsedTime(startedAt)); return BackgroundJobRunResult.Completed; } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) @@ -103,6 +114,10 @@ public sealed class BackgroundJobRunner( token, exception, CancellationToken.None); + telemetry.RecordProcessing( + message.JobKind, + "retry", + Stopwatch.GetElapsedTime(startedAt)); return BackgroundJobRunResult.Retry(TimeSpan.FromSeconds(2)); } finally diff --git a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobTelemetry.cs b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobTelemetry.cs new file mode 100644 index 0000000..9402ebe --- /dev/null +++ b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobTelemetry.cs @@ -0,0 +1,74 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; +using Jiaowu.Api.Domain.System; + +namespace Jiaowu.Api.Infrastructure.BackgroundJobs; + +public sealed class BackgroundJobTelemetry : IDisposable +{ + public const string MeterName = "Jiaowu.BackgroundJobs"; + + private readonly Meter _meter = new(MeterName, "1.0.0"); + private readonly Counter _published; + private readonly Counter _processed; + private readonly Histogram _publishDuration; + private readonly Histogram _processingDuration; + private readonly Counter _cleaned; + + public BackgroundJobTelemetry() + { + _published = _meter.CreateCounter( + "jiaowu.background_jobs.published", + unit: "{message}"); + _processed = _meter.CreateCounter( + "jiaowu.background_jobs.processed", + unit: "{job}"); + _publishDuration = _meter.CreateHistogram( + "jiaowu.background_jobs.publish.duration", + unit: "ms"); + _processingDuration = _meter.CreateHistogram( + "jiaowu.background_jobs.processing.duration", + unit: "ms"); + _cleaned = _meter.CreateCounter( + "jiaowu.background_jobs.outbox.cleaned", + unit: "{message}"); + } + + public void RecordPublish( + BackgroundJobKind kind, + bool durable, + string outcome, + TimeSpan duration) + { + var tags = new TagList + { + { "job.kind", kind.ToString() }, + { "messaging.backend", durable ? "rabbitmq" : "memory" }, + { "job.outcome", outcome } + }; + _published.Add(1, tags); + _publishDuration.Record(duration.TotalMilliseconds, tags); + } + + public void RecordProcessing( + BackgroundJobKind kind, + string outcome, + TimeSpan duration) + { + var tags = new TagList + { + { "job.kind", kind.ToString() }, + { "job.outcome", outcome } + }; + _processed.Add(1, tags); + _processingDuration.Record(duration.TotalMilliseconds, tags); + } + + public void RecordCleaned(int count) + { + if (count > 0) + _cleaned.Add(count); + } + + public void Dispose() => _meter.Dispose(); +} diff --git a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobTransport.cs b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobTransport.cs index ff1c381..862daf3 100644 --- a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobTransport.cs +++ b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobTransport.cs @@ -21,26 +21,31 @@ public interface IBackgroundJobTransport public sealed class InMemoryBackgroundJobTransport : IBackgroundJobTransport { - private readonly Channel _channel = - Channel.CreateBounded( - new BoundedChannelOptions(256) - { - FullMode = BoundedChannelFullMode.Wait, - SingleReader = true, - SingleWriter = false - }); + private readonly IReadOnlyDictionary< + BackgroundJobKind, + Channel> _channels = + Enum.GetValues().ToDictionary( + kind => kind, + _ => Channel.CreateBounded( + new BoundedChannelOptions(256) + { + FullMode = BoundedChannelFullMode.Wait, + SingleReader = false, + SingleWriter = false + })); public bool IsDurable => false; public ValueTask PublishAsync( BackgroundJobEnvelope message, CancellationToken cancellationToken) => - _channel.Writer.WriteAsync(message, cancellationToken); + _channels[message.JobKind].Writer.WriteAsync(message, cancellationToken); public Task CheckHealthAsync(CancellationToken cancellationToken) => Task.FromResult(true); public IAsyncEnumerable ReadAllAsync( + BackgroundJobKind kind, CancellationToken cancellationToken) => - _channel.Reader.ReadAllAsync(cancellationToken); + _channels[kind].Reader.ReadAllAsync(cancellationToken); } diff --git a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/InMemoryBackgroundJobWorker.cs b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/InMemoryBackgroundJobWorker.cs index 67d4a24..c5094cd 100644 --- a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/InMemoryBackgroundJobWorker.cs +++ b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/InMemoryBackgroundJobWorker.cs @@ -1,27 +1,46 @@ +using Jiaowu.Api.Domain.System; + namespace Jiaowu.Api.Infrastructure.BackgroundJobs; public sealed class InMemoryBackgroundJobWorker( InMemoryBackgroundJobTransport transport, BackgroundJobRunner runner, + BackgroundJobOptions options, ILogger logger) : BackgroundService { protected override async Task ExecuteAsync(CancellationToken stoppingToken) { + var workers = Enum.GetValues() + .SelectMany(kind => Enumerable.Range( + 0, + options.ConsumerConcurrency(kind)) + .Select(_ => ConsumeAsync(kind, stoppingToken))) + .ToArray(); + try { - await foreach (var message in transport.ReadAllAsync(stoppingToken)) - { - var result = await runner.RunAsync(message, stoppingToken); - if (result.Outcome != BackgroundJobRunOutcome.Retry) - continue; - - await Task.Delay(result.RetryAfter, stoppingToken); - await transport.PublishAsync(message, stoppingToken); - } + await Task.WhenAll(workers); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { logger.LogInformation("In-memory background job worker is stopping."); } } + + private async Task ConsumeAsync( + BackgroundJobKind kind, + CancellationToken cancellationToken) + { + await foreach (var message in transport.ReadAllAsync( + kind, + cancellationToken)) + { + var result = await runner.RunAsync(message, cancellationToken); + if (result.Outcome != BackgroundJobRunOutcome.Retry) + continue; + + await Task.Delay(result.RetryAfter, cancellationToken); + await transport.PublishAsync(message, cancellationToken); + } + } } diff --git a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/RabbitMqBackgroundJobs.cs b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/RabbitMqBackgroundJobs.cs index 06809a1..865bbb2 100644 --- a/src/Jiaowu.Api/Infrastructure/BackgroundJobs/RabbitMqBackgroundJobs.cs +++ b/src/Jiaowu.Api/Infrastructure/BackgroundJobs/RabbitMqBackgroundJobs.cs @@ -82,6 +82,7 @@ public sealed class RabbitMqBackgroundJobTransport( _connection = await RabbitMqBackgroundJobTopology.CreateConnectionAsync( rabbitOptions, "jiaowu-background-job-publisher", + consumerDispatchConcurrency: 1, cancellationToken); _channel = await _connection.CreateChannelAsync( new CreateChannelOptions( @@ -153,101 +154,114 @@ public sealed class RabbitMqBackgroundJobWorker( var channels = new List(); try { + var consumerCount = RabbitMqBackgroundJobTopology.JobKinds.Sum( + jobOptions.ConsumerConcurrency); connection = await RabbitMqBackgroundJobTopology.CreateConnectionAsync( rabbitOptions, "jiaowu-background-job-worker", + (ushort)consumerCount, stoppingToken); foreach (var kind in RabbitMqBackgroundJobTopology.JobKinds) { - var channel = await connection.CreateChannelAsync( - new CreateChannelOptions( - publisherConfirmationsEnabled: false, - publisherConfirmationTrackingEnabled: false), - stoppingToken); - channels.Add(channel); - await RabbitMqBackgroundJobTopology.DeclareAsync( - channel, - jobOptions, - stoppingToken); - await channel.BasicQosAsync( - 0, - jobOptions.PrefetchCount, - global: false, - stoppingToken); - - var consumer = new AsyncEventingBasicConsumer(channel); - consumer.ReceivedAsync += async (_, eventArgs) => + for (var workerIndex = 0; + workerIndex < jobOptions.ConsumerConcurrency(kind); + workerIndex++) { - using var deliveryCancellation = - CancellationTokenSource.CreateLinkedTokenSource( - eventArgs.CancellationToken, - stoppingToken); - var deliveryToken = deliveryCancellation.Token; - try + var channel = await connection.CreateChannelAsync( + new CreateChannelOptions( + publisherConfirmationsEnabled: false, + publisherConfirmationTrackingEnabled: false), + stoppingToken); + channels.Add(channel); + await RabbitMqBackgroundJobTopology.DeclareAsync( + channel, + jobOptions, + stoppingToken); + await channel.BasicQosAsync( + 0, + jobOptions.PrefetchCount, + global: false, + stoppingToken); + + var consumer = new AsyncEventingBasicConsumer(channel); + consumer.ReceivedAsync += async (_, eventArgs) => { - var message = JsonSerializer.Deserialize( - eventArgs.Body.Span); - if (message is null || message.JobKind != kind) + using var deliveryCancellation = + CancellationTokenSource.CreateLinkedTokenSource( + eventArgs.CancellationToken, + stoppingToken); + var deliveryToken = deliveryCancellation.Token; + try { + var message = JsonSerializer + .Deserialize( + eventArgs.Body.Span); + if (message is null || message.JobKind != kind) + { + await channel.BasicNackAsync( + eventArgs.DeliveryTag, + multiple: false, + requeue: false, + deliveryToken); + return; + } + + var result = await runner.RunAsync( + message, + deliveryToken); + if (result.Outcome == + BackgroundJobRunOutcome.Completed) + { + await channel.BasicAckAsync( + eventArgs.DeliveryTag, + multiple: false, + deliveryToken); + return; + } + + await Task.Delay( + result.RetryAfter, + deliveryToken); await channel.BasicNackAsync( eventArgs.DeliveryTag, multiple: false, - requeue: false, + requeue: true, deliveryToken); - return; } - - var result = await runner.RunAsync( - message, - deliveryToken); - if (result.Outcome == BackgroundJobRunOutcome.Completed) + catch (OperationCanceledException) { - await channel.BasicAckAsync( - eventArgs.DeliveryTag, - multiple: false, - deliveryToken); - return; + // Closing the channel requeues unacknowledged deliveries. } - - await Task.Delay( - result.RetryAfter, - deliveryToken); - await channel.BasicNackAsync( - eventArgs.DeliveryTag, - multiple: false, - requeue: true, - deliveryToken); - } - catch (OperationCanceledException) - { - // Closing the channel requeues unacknowledged deliveries. - } - catch (Exception exception) - { - logger.LogError( - exception, - "RabbitMQ delivery for {JobKind} failed and will be requeued.", - kind); - if (channel.IsOpen) + catch (Exception exception) { + logger.LogError( + exception, + "RabbitMQ delivery for {JobKind} failed and will be requeued.", + kind); + if (!channel.IsOpen) + return; + await channel.BasicNackAsync( eventArgs.DeliveryTag, multiple: false, requeue: true, CancellationToken.None); } - } - }; - await channel.BasicConsumeAsync( - RabbitMqBackgroundJobTopology.QueueName(jobOptions, kind), - autoAck: false, - consumer, - stoppingToken); + }; + await channel.BasicConsumeAsync( + RabbitMqBackgroundJobTopology.QueueName( + jobOptions, + kind), + autoAck: false, + consumer, + stoppingToken); + } } logger.LogInformation( - "RabbitMQ background job consumers are connected to {HostName}:{Port}.", + "{ConsumerCount} RabbitMQ background job consumers are connected to {HostName}:{Port}.", + consumerCount, rabbitOptions.HostName, rabbitOptions.Port); while (connection.IsOpen && !stoppingToken.IsCancellationRequested) @@ -308,6 +322,7 @@ internal static class RabbitMqBackgroundJobTopology public static async Task CreateConnectionAsync( RabbitMqOptions options, string clientName, + ushort consumerDispatchConcurrency, CancellationToken cancellationToken) { var factory = new ConnectionFactory @@ -321,7 +336,7 @@ internal static class RabbitMqBackgroundJobTopology AutomaticRecoveryEnabled = false, TopologyRecoveryEnabled = false, RequestedHeartbeat = TimeSpan.FromSeconds(30), - ConsumerDispatchConcurrency = 1 + ConsumerDispatchConcurrency = consumerDispatchConcurrency }; if (options.UseTls) { diff --git a/src/Jiaowu.Api/Infrastructure/Persistence/AppDbContext.cs b/src/Jiaowu.Api/Infrastructure/Persistence/AppDbContext.cs index c84962e..ea77a4e 100644 --- a/src/Jiaowu.Api/Infrastructure/Persistence/AppDbContext.cs +++ b/src/Jiaowu.Api/Infrastructure/Persistence/AppDbContext.cs @@ -1000,6 +1000,7 @@ public sealed class AppDbContext(DbContextOptions options) entity.Property(x => x.LastError).HasMaxLength(2000); entity.HasIndex(x => new { x.JobKind, x.JobId }).IsUnique(); entity.HasIndex(x => new { x.State, x.CreatedAt }); + entity.HasIndex(x => new { x.State, x.CompletedAt }); entity.HasIndex(x => x.LeaseExpiresAt); }); diff --git a/src/Jiaowu.Api/Infrastructure/Persistence/DevelopmentSqliteMigrator.cs b/src/Jiaowu.Api/Infrastructure/Persistence/DevelopmentSqliteMigrator.cs index c4abaa6..8d7c616 100644 --- a/src/Jiaowu.Api/Infrastructure/Persistence/DevelopmentSqliteMigrator.cs +++ b/src/Jiaowu.Api/Infrastructure/Persistence/DevelopmentSqliteMigrator.cs @@ -2050,6 +2050,10 @@ public sealed class DevelopmentSqliteMigrator( """ CREATE INDEX "IX_BackgroundJobOutboxMessages_State_CreatedAt" ON "BackgroundJobOutboxMessages" ("State", "CreatedAt"); + """, + """ + CREATE INDEX "IX_BackgroundJobOutboxMessages_State_CompletedAt" + ON "BackgroundJobOutboxMessages" ("State", "CompletedAt"); """ ]; } diff --git a/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/20260726220000_BackgroundJobOutbox.Designer.cs b/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/20260726220000_BackgroundJobOutbox.Designer.cs index 95e6b5e..b557267 100644 --- a/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/20260726220000_BackgroundJobOutbox.Designer.cs +++ b/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/20260726220000_BackgroundJobOutbox.Designer.cs @@ -3478,6 +3478,8 @@ namespace Jiaowu.Api.Infrastructure.Persistence.Migrations.MySql b.HasIndex("State", "CreatedAt"); + b.HasIndex("State", "CompletedAt"); + b.ToTable("BackgroundJobOutboxMessages"); }); diff --git a/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/20260726220000_BackgroundJobOutbox.cs b/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/20260726220000_BackgroundJobOutbox.cs index 364017e..ca1b689 100644 --- a/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/20260726220000_BackgroundJobOutbox.cs +++ b/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/20260726220000_BackgroundJobOutbox.cs @@ -50,6 +50,11 @@ namespace Jiaowu.Api.Infrastructure.Persistence.Migrations.MySql name: "IX_BackgroundJobOutboxMessages_State_CreatedAt", table: "BackgroundJobOutboxMessages", columns: new[] { "State", "CreatedAt" }); + + migrationBuilder.CreateIndex( + name: "IX_BackgroundJobOutboxMessages_State_CompletedAt", + table: "BackgroundJobOutboxMessages", + columns: new[] { "State", "CompletedAt" }); } /// diff --git a/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/AppDbContextModelSnapshot.cs b/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/AppDbContextModelSnapshot.cs index a99c945..7d058ad 100644 --- a/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/AppDbContextModelSnapshot.cs +++ b/src/Jiaowu.Api/Infrastructure/Persistence/Migrations/MySql/AppDbContextModelSnapshot.cs @@ -3475,6 +3475,8 @@ namespace Jiaowu.Api.Infrastructure.Persistence.Migrations.MySql b.HasIndex("State", "CreatedAt"); + b.HasIndex("State", "CompletedAt"); + b.ToTable("BackgroundJobOutboxMessages"); }); diff --git a/src/Jiaowu.Api/Program.cs b/src/Jiaowu.Api/Program.cs index a1eae50..b21e8a1 100644 --- a/src/Jiaowu.Api/Program.cs +++ b/src/Jiaowu.Api/Program.cs @@ -137,7 +137,13 @@ if (!backgroundJobOptions.Transport.Equals( if (backgroundJobOptions.PollIntervalMilliseconds is < 100 or > 30000 || backgroundJobOptions.LeaseSeconds is < 30 or > 3600 || backgroundJobOptions.PrefetchCount is < 1 or > 100 || + backgroundJobOptions.AutomaticScheduleConcurrency is < 1 or > 16 || + backgroundJobOptions.SchedulePublishConcurrency is < 1 or > 16 || + backgroundJobOptions.MakeupExamAutoConcurrency is < 1 or > 16 || backgroundJobOptions.ProcessingAttemptLimit is < 1 or > 100 || + backgroundJobOptions.MaintenanceIntervalSeconds is < 10 or > 3600 || + backgroundJobOptions.CompletedRetentionDays is < 1 or > 3650 || + backgroundJobOptions.CleanupBatchSize is < 10 or > 5000 || string.IsNullOrWhiteSpace(backgroundJobOptions.Exchange) || string.IsNullOrWhiteSpace(backgroundJobOptions.QueuePrefix)) { @@ -269,6 +275,8 @@ builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); +builder.Services.AddSingleton(); +builder.Services.AddScoped(); builder.Services.AddSingleton(); if (backgroundJobOptions.UsesRabbitMq) { @@ -534,15 +542,41 @@ static async Task CheckCacheHealthAsync( static async Task CheckMessagingHealthAsync( IBackgroundJobTransport transport, + BackgroundJobMonitoringService monitoring, CancellationToken cancellationToken) { var healthy = await transport.CheckHealthAsync(cancellationToken); var backend = transport.IsDurable ? "rabbitmq" : "memory"; - return healthy - ? Results.Ok(new { Status = "healthy", Backend = backend }) - : Results.Json( - new { Status = "unhealthy", Backend = backend }, + try + { + var backlog = await monitoring.GetSnapshotAsync(cancellationToken); + return healthy + ? Results.Ok(new + { + Status = "healthy", + Backend = backend, + Backlog = backlog + }) + : Results.Json( + new + { + Status = "unhealthy", + Backend = backend, + Backlog = backlog + }, + statusCode: StatusCodes.Status503ServiceUnavailable); + } + catch + { + return Results.Json( + new + { + Status = "unhealthy", + Backend = backend, + Backlog = (object?)null + }, statusCode: StatusCodes.Status503ServiceUnavailable); + } } public partial class Program; diff --git a/src/Jiaowu.Api/appsettings.json b/src/Jiaowu.Api/appsettings.json index 8ed5bed..a07516c 100644 --- a/src/Jiaowu.Api/appsettings.json +++ b/src/Jiaowu.Api/appsettings.json @@ -24,10 +24,16 @@ "PollIntervalMilliseconds": 500, "LeaseSeconds": 120, "PrefetchCount": 1, + "AutomaticScheduleConcurrency": 1, + "SchedulePublishConcurrency": 1, + "MakeupExamAutoConcurrency": 1, "Exchange": "jiaowu.background-jobs", "QueuePrefix": "jiaowu.background-jobs", "UseQuorumQueues": true, - "ProcessingAttemptLimit": 5 + "ProcessingAttemptLimit": 5, + "MaintenanceIntervalSeconds": 60, + "CompletedRetentionDays": 14, + "CleanupBatchSize": 500 }, "RabbitMq": { "HostName": "localhost", diff --git a/tests/Jiaowu.Api.Tests/AutomaticScheduleGeneratorTests.cs b/tests/Jiaowu.Api.Tests/AutomaticScheduleGeneratorTests.cs index c463b06..91fb3d9 100644 --- a/tests/Jiaowu.Api.Tests/AutomaticScheduleGeneratorTests.cs +++ b/tests/Jiaowu.Api.Tests/AutomaticScheduleGeneratorTests.cs @@ -30,14 +30,22 @@ public sealed class AutomaticScheduleGeneratorTests Assert.True(await db.ScheduleTimeSlots.CountAsync() == 0); Assert.True(await db.AutomaticScheduleJobs.CountAsync() == 0); Assert.True(await db.BackgroundJobOutboxMessages.CountAsync() == 0); - Assert.True(await db.Database - .SqlQueryRaw( + Assert.True(await db.Database + .SqlQueryRaw( + """ + SELECT COUNT(*) AS "Value" + FROM pragma_index_list('BackgroundJobOutboxMessages') + WHERE name = 'IX_BackgroundJobOutboxMessages_State_CompletedAt' + """) + .AnyAsync(value => value > 0)); + Assert.True(await db.Database + .SqlQueryRaw( """ SELECT COUNT(*) AS "Value" FROM pragma_table_info('ScheduleEntries') WHERE name = 'ClassroomId' AND "notnull" = 0 """) - .AnyAsync(value => value > 0)); + .AnyAsync(value => value > 0)); Assert.True(await db.Database .SqlQueryRaw( """ diff --git a/tests/Jiaowu.Api.Tests/BackgroundJobOutboxTests.cs b/tests/Jiaowu.Api.Tests/BackgroundJobOutboxTests.cs index ae82c51..77615ad 100644 --- a/tests/Jiaowu.Api.Tests/BackgroundJobOutboxTests.cs +++ b/tests/Jiaowu.Api.Tests/BackgroundJobOutboxTests.cs @@ -10,7 +10,36 @@ namespace Jiaowu.Api.Tests; public sealed class BackgroundJobOutboxTests { [Fact] - public async Task Publisher_moves_pending_message_to_in_memory_transport() + public async Task In_memory_transport_keeps_job_types_on_independent_channels() + { + var transport = new InMemoryBackgroundJobTransport(); + var automatic = new BackgroundJobEnvelope( + Guid.NewGuid(), + BackgroundJobKind.AutomaticSchedule, + Guid.NewGuid()); + var publish = new BackgroundJobEnvelope( + Guid.NewGuid(), + BackgroundJobKind.SchedulePublish, + Guid.NewGuid()); + await transport.PublishAsync(automatic, CancellationToken.None); + await transport.PublishAsync(publish, CancellationToken.None); + + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1)); + await using var automaticReader = transport + .ReadAllAsync(BackgroundJobKind.AutomaticSchedule, timeout.Token) + .GetAsyncEnumerator(timeout.Token); + await using var publishReader = transport + .ReadAllAsync(BackgroundJobKind.SchedulePublish, timeout.Token) + .GetAsyncEnumerator(timeout.Token); + + Assert.True(await automaticReader.MoveNextAsync()); + Assert.Equal(automatic, automaticReader.Current); + Assert.True(await publishReader.MoveNextAsync()); + Assert.Equal(publish, publishReader.Current); + } + + [Fact] + public async Task Publisher_recovers_pending_message_and_cleans_old_completion() { var databasePath = Path.Combine( Path.GetTempPath(), @@ -21,18 +50,22 @@ public sealed class BackgroundJobOutboxTests options.UseSqlite($"Data Source={databasePath};Pooling=False")); var options = new BackgroundJobOptions { - PollIntervalMilliseconds = 100 + PollIntervalMilliseconds = 100, + CompletedRetentionDays = 1, + CleanupBatchSize = 10 }; services.AddSingleton(options); services.AddSingleton(); services.AddSingleton(provider => provider.GetRequiredService()); + services.AddSingleton(); services.AddSingleton(); var provider = services.BuildServiceProvider(); try { Guid jobId; + Guid oldCompletedOutboxId; await using (var scope = provider.CreateAsyncScope()) { var db = scope.ServiceProvider.GetRequiredService(); @@ -57,8 +90,14 @@ public sealed class BackgroundJobOutboxTests SchedulePlan = plan, ActiveSchedulePlanId = plan.Id }; + var oldCompletedOutbox = BackgroundJobOutboxMessage.Create( + BackgroundJobKind.SchedulePublish, + Guid.NewGuid()); + oldCompletedOutbox.State = BackgroundJobOutboxState.Completed; + oldCompletedOutbox.CompletedAt = DateTime.UtcNow.AddDays(-2); + oldCompletedOutboxId = oldCompletedOutbox.Id; jobId = job.Id; - db.AddRange(term, plan, job); + db.AddRange(term, plan, job, oldCompletedOutbox); await db.SaveChangesAsync(); } @@ -76,7 +115,9 @@ public sealed class BackgroundJobOutboxTests .AnyAsync(x => x.JobId == jobId && x.JobKind == BackgroundJobKind.AutomaticSchedule && - x.State == BackgroundJobOutboxState.Published); + x.State == BackgroundJobOutboxState.Published) && + !await db.BackgroundJobOutboxMessages.AsNoTracking() + .AnyAsync(x => x.Id == oldCompletedOutboxId); } Assert.True(published); @@ -93,6 +134,47 @@ public sealed class BackgroundJobOutboxTests } } + [Fact] + public async Task Monitoring_snapshot_reports_backlog_and_expired_leases() + { + var databasePath = Path.Combine( + Path.GetTempPath(), + $"jiaowu-monitoring-{Guid.NewGuid():N}.sqlite"); + var options = new DbContextOptionsBuilder() + .UseSqlite($"Data Source={databasePath};Pooling=False") + .Options; + + try + { + await using var db = new AppDbContext(options); + await db.Database.EnsureCreatedAsync(); + var pending = BackgroundJobOutboxMessage.Create( + BackgroundJobKind.AutomaticSchedule, + Guid.NewGuid()); + pending.CreatedAt = DateTime.UtcNow.AddMinutes(-2); + var processing = BackgroundJobOutboxMessage.Create( + BackgroundJobKind.MakeupExamAuto, + Guid.NewGuid()); + processing.State = BackgroundJobOutboxState.Processing; + processing.LeaseExpiresAt = DateTime.UtcNow.AddMinutes(-1); + db.AddRange(pending, processing); + await db.SaveChangesAsync(); + + var snapshot = await new BackgroundJobMonitoringService(db) + .GetSnapshotAsync(CancellationToken.None); + + Assert.Equal(1, snapshot.Pending); + Assert.Equal(1, snapshot.Processing); + Assert.Equal(1, snapshot.ExpiredLeases); + Assert.NotNull(snapshot.OldestUnfinishedAt); + Assert.True(snapshot.OldestUnfinishedAgeSeconds >= 100); + } + finally + { + File.Delete(databasePath); + } + } + [Fact] public async Task Runner_stops_after_retry_limit_and_ignores_duplicate_delivery() { @@ -109,6 +191,7 @@ public sealed class BackgroundJobOutboxTests services.AddSingleton(); services.AddSingleton(provider => provider.GetRequiredService()); + services.AddSingleton(); services.AddSingleton(); var provider = services.BuildServiceProvider();