feat(processing): persist safe atomic operation history
This commit is contained in:
cesnimda
@@ -9,6 +9,171 @@ namespace JobTrackerApi.Tests;
|
||||
|
||||
public sealed class UserOperationStoreTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Retry_uses_current_durable_state_and_emits_one_event_without_a_tracked_stale_operation()
|
||||
{
|
||||
await using var fixture = await Fixture.CreateAsync();
|
||||
await using var owner = fixture.Context("user-1");
|
||||
var store = fixture.Store(owner);
|
||||
var id = (await store.CreateAsync(Request("retry-current"), default)).Operation.Id;
|
||||
await using var worker = fixture.Context(null);
|
||||
var lease = Assert.IsType<UserOperationLease>(await fixture.Store(worker).ClaimNextAsync(TimeSpan.FromSeconds(20), default));
|
||||
Assert.True(await store.FailAsync(id, lease.LeaseToken, false, "timeout", "Synthetic timeout", TimeSpan.Zero, default));
|
||||
Assert.True(await store.RetryAsync(id, default));
|
||||
Assert.False(await store.RetryAsync(id, default));
|
||||
await using var readback = fixture.Context("user-1");
|
||||
Assert.Equal(OperationStatuses.Queued, (await readback.UserOperations.SingleAsync()).Status);
|
||||
Assert.Single(await readback.UserOperationEvents.Where(item => item.Kind == "Retried").ToListAsync());
|
||||
Assert.Empty(await readback.UserNotifications.ToListAsync());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Stage_history_is_deduplicated_owner_scoped_and_redacts_unknown_values()
|
||||
{
|
||||
await using var fixture = await Fixture.CreateAsync();
|
||||
await using var owner = fixture.Context("user-1");
|
||||
var store = fixture.Store(owner);
|
||||
var id = (await store.CreateAsync(Request("stage-history"), default)).Operation.Id;
|
||||
await using var worker = fixture.Context(null);
|
||||
var lease = Assert.IsType<UserOperationLease>(await fixture.Store(worker).ClaimNextAsync(TimeSpan.FromSeconds(20), default));
|
||||
Assert.Equal(0, await store.ReportStageAsync(id, "wrong", "cv_extracting", default));
|
||||
Assert.Equal(1, await store.ReportStageAsync(id, lease.LeaseToken, "cv_extracting", default));
|
||||
Assert.Equal(1, await store.ReportStageAsync(id, lease.LeaseToken, "cv_extracting", default));
|
||||
Assert.Equal(1, await store.ReportStageAsync(id, lease.LeaseToken, "SECRET document and token", default));
|
||||
Assert.True(await store.FailAsync(id, lease.LeaseToken, false, "SECRET error", "SECRET document content", TimeSpan.Zero, default));
|
||||
var events = await owner.UserOperationEvents.OrderBy(item => item.Id).ToListAsync();
|
||||
Assert.Equal(new[] { "Created", "Claimed", "StageChanged", "StageChanged", "Failed" }, events.Select(item => item.Kind));
|
||||
Assert.Equal("cv_extracting", events[2].Stage);
|
||||
Assert.Null(events[3].Stage);
|
||||
Assert.Equal("processing_failed", events[4].FailureCode);
|
||||
Assert.DoesNotContain("SECRET", System.Text.Json.JsonSerializer.Serialize(events));
|
||||
Assert.Empty(await worker.UserOperationEvents.ToListAsync());
|
||||
await using var other = fixture.Context("user-2");
|
||||
Assert.Empty(await other.UserOperationEvents.ToListAsync());
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("Claimed")]
|
||||
[InlineData("StageChanged")]
|
||||
[InlineData("Completed")]
|
||||
[InlineData("Retried")]
|
||||
public async Task Failed_event_insert_rolls_back_claim_stage_or_completion(string kind)
|
||||
{
|
||||
await using var fixture = await Fixture.CreateAsync();
|
||||
await using var owner = fixture.Context("user-1");
|
||||
var store = fixture.Store(owner);
|
||||
var id = (await store.CreateAsync(Request("rollback-history"), default)).Operation.Id;
|
||||
await using var worker = fixture.Context(null);
|
||||
UserOperationLease? lease = null;
|
||||
if (kind != "Claimed") lease = await fixture.Store(worker).ClaimNextAsync(TimeSpan.FromSeconds(20), default);
|
||||
if (kind == "Retried") Assert.True(await store.FailAsync(id, lease!.LeaseToken, false, "timeout", "Synthetic timeout", TimeSpan.Zero, default));
|
||||
await owner.Database.ExecuteSqlRawAsync("CREATE TRIGGER fail_history BEFORE INSERT ON UserOperationEvents BEGIN SELECT RAISE(ABORT, 'synthetic event failure'); END;");
|
||||
await Assert.ThrowsAnyAsync<Exception>(async () =>
|
||||
{
|
||||
if (kind == "Claimed") await fixture.Store(worker).ClaimNextAsync(TimeSpan.FromSeconds(20), default);
|
||||
else if (kind == "StageChanged") await store.ReportStageAsync(id, lease!.LeaseToken, "cv_extracting", default);
|
||||
else if (kind == "Retried") await store.RetryAsync(id, default);
|
||||
else await store.CompleteAsync(id, lease!.LeaseToken, null, default);
|
||||
});
|
||||
await using var readback = fixture.Context("user-1");
|
||||
var operation = await readback.UserOperations.SingleAsync();
|
||||
Assert.Equal(kind == "Claimed" ? OperationStatuses.Queued : kind == "Retried" ? OperationStatuses.Failed : OperationStatuses.Running, operation.Status);
|
||||
Assert.Equal(kind == "Claimed" ? 0 : 1, operation.AttemptCount);
|
||||
Assert.Null(operation.ProgressStage);
|
||||
Assert.Equal(kind == "Retried" ? 1 : 0, await readback.UserNotifications.CountAsync());
|
||||
Assert.Equal(kind == "Claimed" ? 1 : kind == "Retried" ? 3 : 2, await readback.UserOperationEvents.CountAsync());
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(false)]
|
||||
[InlineData(true)]
|
||||
public async Task Expired_or_cancel_requested_lease_cannot_publish_a_stage(bool cancel)
|
||||
{
|
||||
await using var fixture = await Fixture.CreateAsync();
|
||||
await using var owner = fixture.Context("user-1");
|
||||
var store = fixture.Store(owner);
|
||||
var id = (await store.CreateAsync(Request("inactive-stage"), default)).Operation.Id;
|
||||
await using var worker = fixture.Context(null);
|
||||
var lease = Assert.IsType<UserOperationLease>(await fixture.Store(worker).ClaimNextAsync(TimeSpan.FromSeconds(20), default));
|
||||
if (cancel) Assert.True(await store.RequestCancellationAsync(id, default));
|
||||
else fixture.Time.Advance(TimeSpan.FromSeconds(21));
|
||||
Assert.Equal(0, await store.ReportStageAsync(id, lease.LeaseToken, "cv_extracting", default));
|
||||
Assert.Empty(await owner.UserOperationEvents.Where(item => item.Kind == "StageChanged").ToListAsync());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Worker_recovery_history_matches_actual_reclaims_and_final_expiry()
|
||||
{
|
||||
await using var fixture = await Fixture.CreateAsync();
|
||||
await using var owner = fixture.Context("user-1");
|
||||
await fixture.Store(owner).CreateAsync(Request("recover-history", maxAttempts: 2), default);
|
||||
await using var worker = fixture.Context(null);
|
||||
var store = fixture.Store(worker);
|
||||
Assert.NotNull(await store.ClaimNextAsync(TimeSpan.FromSeconds(5), default));
|
||||
fixture.Time.Advance(TimeSpan.FromSeconds(6));
|
||||
Assert.NotNull(await store.ClaimNextAsync(TimeSpan.FromSeconds(5), default));
|
||||
fixture.Time.Advance(TimeSpan.FromSeconds(6));
|
||||
Assert.Null(await store.ClaimNextAsync(TimeSpan.FromSeconds(5), default));
|
||||
var events = await owner.UserOperationEvents.OrderBy(item => item.Id).ToListAsync();
|
||||
Assert.Equal(new[] { "Created", "Claimed", "LeaseRecovered", "Claimed", "Expired" }, events.Select(item => item.Kind));
|
||||
Assert.Equal(new[] { 0, 1, 1, 2, 2 }, events.Select(item => item.Attempt));
|
||||
Assert.Equal(OperationStatuses.Failed, events[^1].Status);
|
||||
Assert.Equal("lease_expired", events[^1].FailureCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Stage_event_does_not_flush_unrelated_tracked_user_changes()
|
||||
{
|
||||
await using var fixture = await Fixture.CreateAsync();
|
||||
await using var owner = fixture.Context("user-1");
|
||||
owner.Users.Add(new() { Id = "user-1", UserName = "synthetic", DisplayName = "Original" });
|
||||
var store = fixture.Store(owner);
|
||||
var id = (await store.CreateAsync(Request("no-flush"), default)).Operation.Id;
|
||||
await using var worker = fixture.Context(null);
|
||||
var lease = Assert.IsType<UserOperationLease>(await fixture.Store(worker).ClaimNextAsync(TimeSpan.FromSeconds(20), default));
|
||||
(await owner.Users.SingleAsync()).DisplayName = "Uncommitted";
|
||||
Assert.Equal(1, await store.ReportStageAsync(id, lease.LeaseToken, "cv_extracting", default));
|
||||
await using var readback = fixture.Context("user-1");
|
||||
Assert.Equal("Original", (await readback.Users.SingleAsync()).DisplayName);
|
||||
Assert.Equal(3, await readback.UserOperationEvents.CountAsync());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task History_retention_and_operation_deletion_preserve_other_owners()
|
||||
{
|
||||
await using var fixture = await Fixture.CreateAsync();
|
||||
await using var owner = fixture.Context("user-1");
|
||||
await fixture.Store(owner).CreateAsync(Request("old-history"), default);
|
||||
fixture.Time.Advance(TimeSpan.FromDays(31));
|
||||
await using var other = fixture.Context("user-2");
|
||||
await fixture.Store(other).CreateAsync(Request("fresh-history"), default);
|
||||
await using var neutral = fixture.Context(null);
|
||||
Assert.Equal(1, await OperationEventLog.PruneAsync(neutral, fixture.Time.GetUtcNow().UtcDateTime, default));
|
||||
Assert.Empty(await owner.UserOperationEvents.ToListAsync());
|
||||
Assert.Single(await other.UserOperationEvents.ToListAsync());
|
||||
Assert.Single(await owner.UserOperations.AsNoTracking().ToListAsync());
|
||||
await other.UserOperations.ExecuteDeleteAsync();
|
||||
Assert.Empty(await neutral.UserOperationEvents.IgnoreQueryFilters().ToListAsync());
|
||||
Assert.Single(await owner.UserOperations.AsNoTracking().ToListAsync());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Durable_history_records_only_successful_transitions_not_duplicate_creation_or_heartbeat()
|
||||
{
|
||||
await using var fixture = await Fixture.CreateAsync();
|
||||
await using var owner = fixture.Context("user-1");
|
||||
var store = fixture.Store(owner);
|
||||
var created = await store.CreateAsync(Request("history"), default);
|
||||
await store.CreateAsync(Request("history"), default);
|
||||
Assert.Equal(1, await owner.Database.SqlQueryRaw<int>("SELECT COUNT(*) AS Value FROM UserOperationEvents").SingleAsync());
|
||||
await using var worker = fixture.Context(null);
|
||||
var lease = Assert.IsType<UserOperationLease>(await fixture.Store(worker).ClaimNextAsync(TimeSpan.FromSeconds(20), default));
|
||||
Assert.Equal(1, await store.HeartbeatAsync(created.Operation.Id, lease.LeaseToken, TimeSpan.FromSeconds(20), null, null, default));
|
||||
Assert.Equal(1, await store.CompleteAsync(created.Operation.Id, lease.LeaseToken, null, default));
|
||||
Assert.Equal(0, await store.CompleteAsync(created.Operation.Id, lease.LeaseToken, null, default));
|
||||
Assert.Equal(3, await owner.Database.SqlQueryRaw<int>("SELECT COUNT(*) AS Value FROM UserOperationEvents").SingleAsync());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Create_is_idempotent_within_owner_and_isolated_between_owners()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user