341 lines
14 KiB
C#
341 lines
14 KiB
C#
using System.Diagnostics;
|
|
using Jiaowu.Api.Domain.Academic;
|
|
using Jiaowu.Api.Domain.System;
|
|
using Jiaowu.Api.Infrastructure.Persistence;
|
|
using Microsoft.EntityFrameworkCore;
|
|
|
|
namespace Jiaowu.Api.Infrastructure.BackgroundJobs;
|
|
|
|
public sealed class BackgroundJobOutboxPublisher(
|
|
IServiceScopeFactory scopeFactory,
|
|
IBackgroundJobTransport transport,
|
|
BackgroundJobOptions options,
|
|
BackgroundJobTelemetry telemetry,
|
|
ILogger<BackgroundJobOutboxPublisher> logger) : BackgroundService
|
|
{
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
await PrepareForStartupAsync(stoppingToken);
|
|
var nextMaintenanceAt = DateTime.MinValue;
|
|
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
try
|
|
{
|
|
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)
|
|
{
|
|
await Task.Delay(options.PollIntervalMilliseconds, stoppingToken);
|
|
continue;
|
|
}
|
|
|
|
await PublishClaimedAsync(claimed.Value.Message, claimed.Value.Token,
|
|
stoppingToken);
|
|
}
|
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
logger.LogError(exception, "Background job outbox publishing failed.");
|
|
await Task.Delay(TimeSpan.FromSeconds(2), stoppingToken);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task PrepareForStartupAsync(CancellationToken cancellationToken)
|
|
{
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
|
|
if (!transport.IsDurable)
|
|
{
|
|
await db.BackgroundJobOutboxMessages
|
|
.Where(x => x.State != BackgroundJobOutboxState.Completed)
|
|
.ExecuteUpdateAsync(
|
|
setters => setters
|
|
.SetProperty(x => x.State, BackgroundJobOutboxState.Pending)
|
|
.SetProperty(x => x.ProcessingToken, (Guid?)null)
|
|
.SetProperty(x => x.LeaseExpiresAt, (DateTime?)null),
|
|
cancellationToken);
|
|
}
|
|
|
|
await AddMissingOutboxMessagesAsync(db, cancellationToken);
|
|
}
|
|
|
|
private async Task AddMissingOutboxMessagesAsync(
|
|
AppDbContext db,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var automaticJobs = await db.AutomaticScheduleJobs.AsNoTracking()
|
|
.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) &&
|
|
!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) &&
|
|
!db.BackgroundJobOutboxMessages.Any(message =>
|
|
message.JobKind == BackgroundJobKind.MakeupExamAuto &&
|
|
message.JobId == x.Id))
|
|
.Select(x => x.Id)
|
|
.ToListAsync(cancellationToken);
|
|
var arrangementJobs = await db.ExamArrangementJobs.AsNoTracking()
|
|
.Where(x =>
|
|
(x.Status == ExamArrangementJobStatus.Queued ||
|
|
x.Status == ExamArrangementJobStatus.Running) &&
|
|
!db.BackgroundJobOutboxMessages.Any(message =>
|
|
message.JobKind == BackgroundJobKind.ExamArrangement &&
|
|
message.JobId == x.Id))
|
|
.Select(x => x.Id)
|
|
.ToListAsync(cancellationToken);
|
|
var exportJobs = await db.ExamSignInExportJobs.AsNoTracking()
|
|
.Where(x =>
|
|
(x.Status == ExamSignInExportJobStatus.Queued ||
|
|
x.Status == ExamSignInExportJobStatus.Running) &&
|
|
!db.BackgroundJobOutboxMessages.Any(message =>
|
|
message.JobKind == BackgroundJobKind.ExamSignInExport &&
|
|
message.JobId == x.Id))
|
|
.Select(x => x.Id)
|
|
.ToListAsync(cancellationToken);
|
|
var publishJobs2 = await db.ExamPublishJobs.AsNoTracking()
|
|
.Where(x =>
|
|
(x.Status == ExamPublishJobStatus.Queued ||
|
|
x.Status == ExamPublishJobStatus.Running) &&
|
|
!db.BackgroundJobOutboxMessages.Any(message =>
|
|
message.JobKind == BackgroundJobKind.ExamPublish &&
|
|
message.JobId == x.Id))
|
|
.Select(x => x.Id)
|
|
.ToListAsync(cancellationToken);
|
|
var gradeStatisticsJobs = await db.CourseGradeStatisticsRefreshJobs.AsNoTracking()
|
|
.Where(x =>
|
|
(x.Status == CourseGradeStatisticsRefreshJobStatus.Queued ||
|
|
x.Status == CourseGradeStatisticsRefreshJobStatus.Running) &&
|
|
!db.BackgroundJobOutboxMessages.Any(message =>
|
|
message.JobKind == BackgroundJobKind.CourseGradeStatisticsRefresh &&
|
|
message.JobId == x.Id))
|
|
.Select(x => x.Id)
|
|
.ToListAsync(cancellationToken);
|
|
|
|
var missingKeys = automaticJobs
|
|
.Select(id => (BackgroundJobKind.AutomaticSchedule, id))
|
|
.Concat(publishJobs.Select(id =>
|
|
(BackgroundJobKind.SchedulePublish, id)))
|
|
.Concat(makeupJobs.Select(id =>
|
|
(BackgroundJobKind.MakeupExamAuto, id)))
|
|
.Concat(arrangementJobs.Select(id =>
|
|
(BackgroundJobKind.ExamArrangement, id)))
|
|
.Concat(exportJobs.Select(id =>
|
|
(BackgroundJobKind.ExamSignInExport, id)))
|
|
.Concat(publishJobs2.Select(id =>
|
|
(BackgroundJobKind.ExamPublish, id)))
|
|
.Concat(gradeStatisticsJobs.Select(id =>
|
|
(BackgroundJobKind.CourseGradeStatisticsRefresh, id)))
|
|
.ToList();
|
|
foreach (var (kind, jobId) in missingKeys)
|
|
{
|
|
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<AppDbContext>();
|
|
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(
|
|
CancellationToken cancellationToken)
|
|
{
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
var now = DateTime.UtcNow;
|
|
var recoveredState = transport.IsDurable
|
|
? BackgroundJobOutboxState.Published
|
|
: BackgroundJobOutboxState.Pending;
|
|
|
|
await db.BackgroundJobOutboxMessages
|
|
.Where(x =>
|
|
(x.State == BackgroundJobOutboxState.Publishing ||
|
|
x.State == BackgroundJobOutboxState.Processing) &&
|
|
x.LeaseExpiresAt != null &&
|
|
x.LeaseExpiresAt < now)
|
|
.ExecuteUpdateAsync(
|
|
setters => setters
|
|
.SetProperty(x => x.State, recoveredState)
|
|
.SetProperty(x => x.ProcessingToken, (Guid?)null)
|
|
.SetProperty(x => x.LeaseExpiresAt, (DateTime?)null),
|
|
cancellationToken);
|
|
}
|
|
|
|
private async Task<(BackgroundJobEnvelope Message, Guid Token)?> ClaimNextAsync(
|
|
CancellationToken cancellationToken)
|
|
{
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
var candidateId = await db.BackgroundJobOutboxMessages.AsNoTracking()
|
|
.Where(x => x.State == BackgroundJobOutboxState.Pending)
|
|
.OrderBy(x => x.CreatedAt)
|
|
.Select(x => (Guid?)x.Id)
|
|
.FirstOrDefaultAsync(cancellationToken);
|
|
if (!candidateId.HasValue)
|
|
return null;
|
|
|
|
var token = Guid.NewGuid();
|
|
var leaseExpiresAt = DateTime.UtcNow.AddSeconds(options.LeaseSeconds);
|
|
var claimed = await db.BackgroundJobOutboxMessages
|
|
.Where(x =>
|
|
x.Id == candidateId.Value &&
|
|
x.State == BackgroundJobOutboxState.Pending)
|
|
.ExecuteUpdateAsync(
|
|
setters => setters
|
|
.SetProperty(x => x.State, BackgroundJobOutboxState.Publishing)
|
|
.SetProperty(x => x.ProcessingToken, token)
|
|
.SetProperty(x => x.LeaseExpiresAt, leaseExpiresAt)
|
|
.SetProperty(x => x.PublishAttempts, x => x.PublishAttempts + 1)
|
|
.SetProperty(x => x.LastError, (string?)null),
|
|
cancellationToken);
|
|
if (claimed == 0)
|
|
return null;
|
|
|
|
var message = await db.BackgroundJobOutboxMessages.AsNoTracking()
|
|
.Where(x => x.Id == candidateId.Value)
|
|
.Select(x => new BackgroundJobEnvelope(x.Id, x.JobKind, x.JobId))
|
|
.SingleAsync(cancellationToken);
|
|
return (message, token);
|
|
}
|
|
|
|
private async Task PublishClaimedAsync(
|
|
BackgroundJobEnvelope message,
|
|
Guid token,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var startedAt = Stopwatch.GetTimestamp();
|
|
try
|
|
{
|
|
await transport.PublishAsync(message, cancellationToken);
|
|
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
await db.BackgroundJobOutboxMessages
|
|
.Where(x =>
|
|
x.Id == message.OutboxMessageId &&
|
|
x.State == BackgroundJobOutboxState.Publishing &&
|
|
x.ProcessingToken == token)
|
|
.ExecuteUpdateAsync(
|
|
setters => setters
|
|
.SetProperty(x => x.State, BackgroundJobOutboxState.Published)
|
|
.SetProperty(x => x.PublishedAt, DateTime.UtcNow)
|
|
.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)
|
|
{
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
var messageText = exception.GetBaseException().Message;
|
|
if (messageText.Length > 2000)
|
|
messageText = messageText[..2000];
|
|
await db.BackgroundJobOutboxMessages
|
|
.Where(x =>
|
|
x.Id == message.OutboxMessageId &&
|
|
x.State == BackgroundJobOutboxState.Publishing &&
|
|
x.ProcessingToken == token)
|
|
.ExecuteUpdateAsync(
|
|
setters => setters
|
|
.SetProperty(x => x.State, BackgroundJobOutboxState.Pending)
|
|
.SetProperty(x => x.ProcessingToken, (Guid?)null)
|
|
.SetProperty(x => x.LeaseExpiresAt, (DateTime?)null)
|
|
.SetProperty(x => x.LastError, messageText),
|
|
cancellationToken);
|
|
telemetry.RecordPublish(
|
|
message.JobKind,
|
|
transport.IsDurable,
|
|
"failed",
|
|
Stopwatch.GetElapsedTime(startedAt));
|
|
throw;
|
|
}
|
|
}
|
|
}
|