xiongyuxing/tiku-backend.net
1
0
Fork 1
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

perf: optimize authorization scoreline and workers

Browse Source
This commit is contained in:
xiong
2026-07-30 09:12:27 +08:00
parent 4bea745b79
commit bedc77bffd
52 changed files with 21911 additions and 347 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
Tiku.Infrastructure/Jobs/BackgroundJobService.cs
View File

@@ -97,20 +97,38 @@ internal sealed class BackgroundJobService(
CancellationToken cancellationToken = default)
{
var now = DateTimeOffset.UtcNow;
var jobs = await dbContext.BackgroundJobs
.Where(job =>
job.Status == BackgroundJobStatus.Pending &&
(includeImmediateJobs || job.RunAfter != null) &&
(job.RunAfter == null || job.RunAfter <= now))
.OrderBy(job => job.CreatedAt)
.Take(Math.Clamp(batchSize, 1, 100))
var leaseExpiresAt = now.Add(LeaseDuration);
var claimedIds = await dbContext.Database.SqlQuery<Guid>($"""
UPDATE background_jobs AS job
SET status = 'processing',
locked_by = {workerId},
lock_expires_at = {leaseExpiresAt},
started_at = COALESCE(started_at, {now}),
updated_at = {now}
WHERE job.id IN (
SELECT candidate.id
FROM background_jobs AS candidate
WHERE (
(candidate.status = 'pending' AND ({includeImmediateJobs} OR candidate.run_after IS NOT NULL) AND
(candidate.run_after IS NULL OR candidate.run_after <= {now})) OR
(candidate.status = 'processing' AND candidate.lock_expires_at <= {now})
)
ORDER BY candidate.created_at, candidate.id
FOR UPDATE SKIP LOCKED
LIMIT {Math.Clamp(batchSize, 1, 100)}
)
RETURNING job.id AS "Value"
""")
.ToArrayAsync(cancellationToken);
var processed = 0;
foreach (var job in jobs)
dbContext.ChangeTracker.Clear();
foreach (var jobId in claimedIds)
{
cancellationToken.ThrowIfCancellationRequested();
if (await ProcessJobAsync(job, workerId, cancellationToken)) processed++;
var job = await dbContext.BackgroundJobs.SingleAsync(value => value.Id == jobId, cancellationToken);
if (await ProcessJobAsync(job, workerId, alreadyClaimed: true, cancellationToken)) processed++;
dbContext.ChangeTracker.Clear();
}
return processed;
@@ -124,6 +142,21 @@ internal sealed class BackgroundJobService(
CancellationToken cancellationToken = default)
{
var normalizedJobType = NormalizeJobType(jobType);
var claimed = await dbContext.BackgroundJobs
.Where(item => item.Id == jobId && item.TenantId == tenantId &&
item.JobType == normalizedJobType && item.Status == BackgroundJobStatus.Pending &&
item.RunAfter == null)
.ExecuteUpdateAsync(setters => setters
.SetProperty(item => item.Status, BackgroundJobStatus.Processing)
.SetProperty(item => item.LockedBy, workerId)
.SetProperty(item => item.LockExpiresAt, DateTimeOffset.UtcNow.Add(LeaseDuration))
.SetProperty(item => item.StartedAt, DateTimeOffset.UtcNow), cancellationToken);
if (claimed == 0)
{
return false;
}
dbContext.ChangeTracker.Clear();
var job = await dbContext.BackgroundJobs.SingleOrDefaultAsync(
item => item.Id == jobId && item.TenantId == tenantId,
cancellationToken);
@@ -135,20 +168,17 @@ internal sealed class BackgroundJobService(
{
throw new InvalidOperationException("The requested background job type does not match the persisted job.");
}
if (job.Status != BackgroundJobStatus.Pending || job.RunAfter is not null)
{
return false;
}
return await ProcessJobAsync(job, workerId, cancellationToken);
return await ProcessJobAsync(job, workerId, alreadyClaimed: true, cancellationToken);
}
private async Task<bool> ProcessJobAsync(
BackgroundJob job,
string workerId,
bool alreadyClaimed,
CancellationToken cancellationToken)
{
if (job.Status != BackgroundJobStatus.Pending)
if ((!alreadyClaimed && job.Status != BackgroundJobStatus.Pending) ||
(alreadyClaimed && (job.Status != BackgroundJobStatus.Processing || job.LockedBy != workerId)))
{
return false;
}
@@ -158,19 +188,25 @@ internal sealed class BackgroundJobService(
FeatureAccessOperation.Write,
cancellationToken)).Allowed)
{
job.Status = BackgroundJobStatus.Failed;
job.CompletedAt = DateTimeOffset.UtcNow;
job.LastError = "Tenant feature entitlement was revoked before job execution.";
await dbContext.SaveChangesAsync(cancellationToken);
await CompleteAsync(
job,
workerId,
BackgroundJobStatus.Failed,
JsonDefaults.Object(),
"Tenant feature entitlement was revoked before job execution.",
cancellationToken);
return true;
}
var now = DateTimeOffset.UtcNow;
job.Status = BackgroundJobStatus.Processing;
job.LockedBy = workerId;
job.LockExpiresAt = now.Add(LeaseDuration);
job.StartedAt = now;
await dbContext.SaveChangesAsync(cancellationToken);
if (!alreadyClaimed)
{
var now = DateTimeOffset.UtcNow;
job.Status = BackgroundJobStatus.Processing;
job.LockedBy = workerId;
job.LockExpiresAt = now.Add(LeaseDuration);
job.StartedAt = now;
await dbContext.SaveChangesAsync(cancellationToken);
}
try
{
@@ -198,14 +234,35 @@ internal sealed class BackgroundJobService(
}
finally
{
job.LockedBy = null;
job.LockExpiresAt = null;
await dbContext.SaveChangesAsync(cancellationToken);
await CompleteAsync(job, workerId, job.Status, job.Result, job.LastError, cancellationToken);
}
return true;
}
private async Task CompleteAsync(
BackgroundJob job,
string workerId,
BackgroundJobStatus status,
JsonElement result,
string? lastError,
CancellationToken cancellationToken)
{
await dbContext.BackgroundJobs
.Where(value => value.Id == job.Id && value.LockedBy == workerId)
.ExecuteUpdateAsync(setters => setters
.SetProperty(value => value.Status, status)
.SetProperty(value => value.RetryCount, job.RetryCount)
.SetProperty(value => value.RunAfter, job.RunAfter)
.SetProperty(value => value.CompletedAt, job.CompletedAt)
.SetProperty(value => value.LastError, lastError)
.SetProperty(value => value.OutputAssetId, job.OutputAssetId)
.SetProperty(value => value.Result, result)
.SetProperty(value => value.LockedBy, (string?)null)
.SetProperty(value => value.LockExpiresAt, (DateTimeOffset?)null),
cancellationToken);
}
public async Task<IReadOnlyCollection<BackgroundJobItem>> ListAsync(
Guid tenantId,
string? jobType = null,