257 lines
12 KiB
C#
257 lines
12 KiB
C#
using ItWorx.KnowledgeHub.Application.Identity;
|
|
using ItWorx.KnowledgeHub.Application.Jobs;
|
|
using ItWorx.KnowledgeHub.Application.Jobs.UseCases;
|
|
using ItWorx.KnowledgeHub.Domain.Jobs;
|
|
using ItWorx.KnowledgeHub.Infrastructure.Persistence;
|
|
using ItWorx.KnowledgeHub.IntegrationTests.Infrastructure;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
|
|
namespace ItWorx.KnowledgeHub.IntegrationTests.Jobs;
|
|
|
|
[Collection(PostgresTestGroup.Name)]
|
|
public sealed class BackgroundJobConsistencyTests(PostgresContainerFixture postgres)
|
|
: OwnerSetupTestBase(postgres, new Dictionary<string, string?> { ["KnowledgeHub:Jobs:Enabled"] = "false" })
|
|
{
|
|
[Fact]
|
|
public async Task Queue_and_cancel_audit_failures_roll_back_job_mutations_and_retries_are_idempotent()
|
|
{
|
|
var (workspaceId, actorId) = await GetOwnerContextAsync();
|
|
const string key = "audit-atomic-background-job";
|
|
|
|
using (var scope = Factory.Services.CreateScope())
|
|
{
|
|
var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
|
|
await InstallAuditFailureTriggerAsync(db, "job.queued");
|
|
}
|
|
|
|
try
|
|
{
|
|
await Assert.ThrowsAsync<DbUpdateException>(async () =>
|
|
{
|
|
using var scope = Factory.Services.CreateScope();
|
|
await scope.ServiceProvider.GetRequiredService<QueueBackgroundJobUseCase>()
|
|
.QueueBackupAsync(new QueueJobCommand(workspaceId, actorId, key, "audit-atomic"), CancellationToken.None);
|
|
});
|
|
}
|
|
finally
|
|
{
|
|
await RemoveAuditFailureTriggerAsync();
|
|
}
|
|
|
|
using (var scope = Factory.Services.CreateScope())
|
|
{
|
|
var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
|
|
Assert.Empty(await db.BackgroundJobs.AsNoTracking().ToListAsync());
|
|
Assert.False(await db.AuditEvents.AsNoTracking().AnyAsync(item => item.Action == "job.queued"));
|
|
}
|
|
|
|
EnqueueJobResult created;
|
|
using (var scope = Factory.Services.CreateScope())
|
|
{
|
|
created = await scope.ServiceProvider.GetRequiredService<QueueBackgroundJobUseCase>()
|
|
.QueueBackupAsync(new QueueJobCommand(workspaceId, actorId, key, "audit-atomic"), CancellationToken.None);
|
|
}
|
|
Assert.Equal(EnqueueJobStatus.Created, created.Status);
|
|
Assert.NotNull(created.Job);
|
|
|
|
using (var scope = Factory.Services.CreateScope())
|
|
{
|
|
var replay = await scope.ServiceProvider.GetRequiredService<QueueBackgroundJobUseCase>()
|
|
.QueueBackupAsync(new QueueJobCommand(workspaceId, actorId, key, "audit-atomic"), CancellationToken.None);
|
|
Assert.Equal(EnqueueJobStatus.Existing, replay.Status);
|
|
Assert.Equal(created.Job.Id, replay.Job?.Id);
|
|
}
|
|
|
|
var queuedVersion = created.Job.Version;
|
|
using (var scope = Factory.Services.CreateScope())
|
|
{
|
|
var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
|
|
await InstallAuditFailureTriggerAsync(db, "job.cancel_requested");
|
|
}
|
|
|
|
try
|
|
{
|
|
await Assert.ThrowsAsync<DbUpdateException>(async () =>
|
|
{
|
|
using var scope = Factory.Services.CreateScope();
|
|
await scope.ServiceProvider.GetRequiredService<CancelBackgroundJobUseCase>()
|
|
.ExecuteAsync(workspaceId, created.Job.Id, actorId, "audit-atomic", CancellationToken.None);
|
|
});
|
|
}
|
|
finally
|
|
{
|
|
await RemoveAuditFailureTriggerAsync();
|
|
}
|
|
|
|
using (var scope = Factory.Services.CreateScope())
|
|
{
|
|
var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
|
|
var persisted = await db.BackgroundJobs.AsNoTracking().SingleAsync(job => job.Id == created.Job.Id);
|
|
Assert.Equal(BackgroundJobStatus.Queued, persisted.Status);
|
|
Assert.Null(persisted.CancelRequestedAt);
|
|
Assert.Equal(queuedVersion, persisted.Version);
|
|
Assert.False(await db.AuditEvents.AsNoTracking().AnyAsync(item => item.Action == "job.cancel_requested"));
|
|
}
|
|
|
|
var lease = Guid.NewGuid();
|
|
var now = DateTimeOffset.UtcNow;
|
|
using (var scope = Factory.Services.CreateScope())
|
|
{
|
|
var repository = scope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
|
|
Assert.Equal(created.Job.Id, await repository.TryClaimNextAsync(
|
|
lease, now, now.AddMinutes(2), CancellationToken.None));
|
|
}
|
|
|
|
using (var scope = Factory.Services.CreateScope())
|
|
{
|
|
var cancellation = scope.ServiceProvider.GetRequiredService<CancelBackgroundJobUseCase>();
|
|
var first = await cancellation.ExecuteAsync(
|
|
workspaceId, created.Job.Id, actorId, "audit-atomic", CancellationToken.None);
|
|
var second = await cancellation.ExecuteAsync(
|
|
workspaceId, created.Job.Id, actorId, "audit-atomic", CancellationToken.None);
|
|
|
|
Assert.Equal(CancelBackgroundJobStatus.Requested, first.Status);
|
|
Assert.Equal(CancelBackgroundJobStatus.Requested, second.Status);
|
|
Assert.NotNull(first.Job);
|
|
Assert.NotNull(second.Job);
|
|
Assert.Equal(first.Job.Version, second.Job.Version);
|
|
Assert.Equal(first.Job.Version, queuedVersion + 2);
|
|
}
|
|
|
|
using (var scope = Factory.Services.CreateScope())
|
|
{
|
|
var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
|
|
Assert.Equal(1, await db.AuditEvents.AsNoTracking().CountAsync(item =>
|
|
item.Action == "job.queued" && item.TargetId == created.Job.Id.ToString("D")));
|
|
Assert.Equal(1, await db.AuditEvents.AsNoTracking().CountAsync(item =>
|
|
item.Action == "job.cancel_requested" && item.TargetId == created.Job.Id.ToString("D")));
|
|
}
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Worker_rebases_its_version_when_a_concurrent_cancellation_wins_the_row_lock()
|
|
{
|
|
var (workspaceId, actorId) = await GetOwnerContextAsync();
|
|
EnqueueJobResult queued;
|
|
using (var scope = Factory.Services.CreateScope())
|
|
{
|
|
queued = await scope.ServiceProvider.GetRequiredService<QueueBackgroundJobUseCase>()
|
|
.QueueBackupAsync(new QueueJobCommand(workspaceId, actorId, "version-race", "version-race"), CancellationToken.None);
|
|
}
|
|
Assert.Equal(EnqueueJobStatus.Created, queued.Status);
|
|
Assert.NotNull(queued.Job);
|
|
|
|
var leaseToken = Guid.NewGuid();
|
|
var now = DateTimeOffset.UtcNow;
|
|
using (var scope = Factory.Services.CreateScope())
|
|
{
|
|
var repository = scope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
|
|
Assert.Equal(queued.Job.Id, await repository.TryClaimNextAsync(
|
|
leaseToken, now, now.AddMinutes(2), CancellationToken.None));
|
|
}
|
|
|
|
using var lockScope = Factory.Services.CreateScope();
|
|
var lockDb = lockScope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
|
|
await lockDb.Database.OpenConnectionAsync();
|
|
await using var transaction = await lockDb.Database.BeginTransactionAsync();
|
|
var cancellationAt = now.AddSeconds(1);
|
|
Assert.Equal(1, await lockDb.Database.ExecuteSqlInterpolatedAsync($"""
|
|
UPDATE background_jobs
|
|
SET cancel_requested_at = {cancellationAt}, version = version + 1
|
|
WHERE id = {queued.Job.Id} AND status = 'Running' AND lease_token = {leaseToken}
|
|
""", CancellationToken.None));
|
|
var cancellationVersion = await lockDb.BackgroundJobs.AsNoTracking()
|
|
.Where(job => job.Id == queued.Job.Id)
|
|
.Select(job => job.Version)
|
|
.SingleAsync();
|
|
|
|
using var workerScope = Factory.Services.CreateScope();
|
|
var workerRepository = workerScope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
|
|
var workerWrite = workerRepository.RecordJobOutcomeAsync(
|
|
queued.Job.Id, leaseToken, BackgroundJobItemState.Succeeded, null, null, CancellationToken.None);
|
|
|
|
var updateBlockedByCancellation = false;
|
|
var waitUntil = DateTimeOffset.UtcNow.AddSeconds(5);
|
|
while (!updateBlockedByCancellation && DateTimeOffset.UtcNow < waitUntil)
|
|
{
|
|
// PostgreSQL caches statistics snapshots inside a transaction unless
|
|
// explicitly cleared, so refresh before each activity check.
|
|
await lockDb.Database.SqlQueryRaw<bool>("SELECT pg_stat_clear_snapshot() IS NULL AS \"Value\"")
|
|
.SingleAsync();
|
|
updateBlockedByCancellation = await lockDb.Database.SqlQueryRaw<bool>("""
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM pg_stat_activity
|
|
WHERE datname = current_database()
|
|
AND pid <> pg_backend_pid()
|
|
AND wait_event_type = 'Lock'
|
|
AND query ILIKE '%background_jobs%'
|
|
) AS "Value"
|
|
""").SingleAsync();
|
|
if (!updateBlockedByCancellation)
|
|
await Task.Delay(25);
|
|
}
|
|
|
|
await transaction.CommitAsync();
|
|
await workerWrite.WaitAsync(TimeSpan.FromSeconds(5));
|
|
Assert.True(updateBlockedByCancellation, "The worker update should wait behind the concurrent cancellation row lock.");
|
|
|
|
using var verifyScope = Factory.Services.CreateScope();
|
|
var current = await verifyScope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>()
|
|
.FindAsync(workspaceId, queued.Job.Id, CancellationToken.None);
|
|
Assert.NotNull(current);
|
|
Assert.Equal(BackgroundJobStatus.Running, current.Status);
|
|
Assert.Equal(leaseToken, current.LeaseToken);
|
|
Assert.NotNull(current.CancelRequestedAt);
|
|
Assert.InRange(Math.Abs((current.CancelRequestedAt.Value - cancellationAt).Ticks), 0, 10);
|
|
Assert.Equal(1, current.Processed);
|
|
Assert.Equal(1, current.Succeeded);
|
|
Assert.True(current.Version > cancellationVersion);
|
|
}
|
|
|
|
private async Task<(Guid WorkspaceId, Guid ActorId)> GetOwnerContextAsync()
|
|
{
|
|
using var scope = Factory.Services.CreateScope();
|
|
var provider = scope.ServiceProvider;
|
|
var workspaceId = await provider.GetRequiredService<IWorkspaceRepository>()
|
|
.GetDefaultWorkspaceIdAsync(CancellationToken.None);
|
|
var actorId = await provider.GetRequiredService<KnowledgeHubDbContext>().Users.AsNoTracking()
|
|
.Where(user => user.Email == OwnerEmail)
|
|
.Select(user => user.Id)
|
|
.SingleAsync();
|
|
return (workspaceId ?? throw new InvalidOperationException("The owner workspace is missing."), actorId);
|
|
}
|
|
|
|
private static async Task InstallAuditFailureTriggerAsync(KnowledgeHubDbContext db, string action)
|
|
{
|
|
await db.Database.ExecuteSqlRawAsync("""
|
|
CREATE TABLE test_job_audit_failure_control (action text PRIMARY KEY);
|
|
CREATE OR REPLACE FUNCTION test_reject_job_audit() RETURNS trigger
|
|
LANGUAGE plpgsql AS $$
|
|
BEGIN
|
|
IF EXISTS (SELECT 1 FROM test_job_audit_failure_control WHERE action = NEW.action) THEN
|
|
RAISE EXCEPTION 'intentional job audit failure';
|
|
END IF;
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
CREATE TRIGGER test_reject_job_audit
|
|
BEFORE INSERT ON audit_events
|
|
FOR EACH ROW EXECUTE FUNCTION test_reject_job_audit();
|
|
""");
|
|
await db.Database.ExecuteSqlInterpolatedAsync($"""
|
|
INSERT INTO test_job_audit_failure_control(action) VALUES ({action})
|
|
""");
|
|
}
|
|
|
|
private async Task RemoveAuditFailureTriggerAsync()
|
|
{
|
|
using var scope = Factory.Services.CreateScope();
|
|
await scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>().Database.ExecuteSqlRawAsync("""
|
|
DROP TRIGGER IF EXISTS test_reject_job_audit ON audit_events;
|
|
DROP FUNCTION IF EXISTS test_reject_job_audit();
|
|
DROP TABLE IF EXISTS test_job_audit_failure_control;
|
|
""");
|
|
}
|
|
}
|