Nostos.Backend.Tests/Portability/EfMigrationJobStoreTests.cs
Rama main · réplica de lectura del repositorio original.
using System.Data.Common;
using FluentAssertions;
using Microsoft.Data.Sqlite;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Diagnostics;
using Nostos.Backend.Data;
using Nostos.Backend.Data.Models;
using Nostos.Backend.Services.Portability;
using Nostos.Backend.Services.Portability.Migration;
using Npgsql;
using Xunit;
namespace Nostos.Backend.Tests.Portability;
// Slice 5 of issue #679: the EF-backed migration job store. Every test runs
// against a real SQLite database file and a manual TimeProvider, so lease
// expiry, restart and concurrency behaviour is deterministic and never sleeps.
public sealed class EfMigrationJobStoreTests : IDisposable
{
private static readonly DateTimeOffset T0 = new(2026, 10, 4, 12, 0, 0, TimeSpan.Zero);
private static readonly TimeSpan LeaseDuration =
TimeSpan.FromMinutes(MigrationContractLimits.WorkerLeaseDurationMinutes);
private readonly List<string> _databasePaths = new();
[Fact]
public async Task Create_with_same_key_and_payload_replays_the_original_job()
{
await using var harness = await NewHarnessAsync();
var first = await harness.Store.CreateAsync(
MigrationDirection.Import,
"create-key",
CancellationToken.None);
first.IsConflict.Should().BeFalse();
first.WasReplay.Should().BeFalse();
var job = first.Resource!;
job.State.Should().Be(MigrationJobState.Pending);
job.Direction.Should().Be(MigrationDirection.Import);
job.RecoveryStatus.Should().Be(MigrationRecoveryStatus.NotRequired);
job.LeaseToken.Should().BeNull();
job.LeaseExpiresAtUtc.Should().BeNull();
job.FailureCode.Should().BeNull();
job.FailureMessage.Should().BeNull();
job.CreatedAtUtc.Should().Be(T0);
job.UpdatedAtUtc.Should().Be(T0);
var replay = await harness.Store.CreateAsync(
MigrationDirection.Import,
"create-key",
CancellationToken.None);
replay.WasReplay.Should().BeTrue();
replay.IsConflict.Should().BeFalse();
replay.Resource!.Id.Should().Be(job.Id);
(await harness.CountJobsAsync()).Should().Be(1);
var record = await harness.GetRecordAsync(job.Id);
record.AttemptNumber.Should().Be(1);
record.Version.Should().Be(0);
record.ExpiresAtUtc.Should().Be(T0.UtcDateTime.Add(EfMigrationJobStore.JobLifetime));
record.CreationPayloadHash.Should().Be(
EfMigrationJobStore.ComputeCreationPayloadHash(MigrationDirection.Import));
}
[Fact]
public async Task Create_with_same_key_and_different_direction_returns_typed_conflict()
{
await using var harness = await NewHarnessAsync();
var created = await harness.Store.CreateAsync(
MigrationDirection.Import,
"reused-key",
CancellationToken.None);
created.Resource.Should().NotBeNull();
var conflict = await harness.Store.CreateAsync(
MigrationDirection.Export,
"reused-key",
CancellationToken.None);
conflict.IsConflict.Should().BeTrue();
conflict.WasReplay.Should().BeFalse();
conflict.Resource.Should().BeNull();
conflict.Conflict!.Kind.Should().Be(
MigrationIdempotencyConflictKind.KeyReusedWithDifferentPayload);
(await harness.CountJobsAsync()).Should().Be(1);
}
[Fact]
public async Task Parallel_creators_of_the_same_key_create_exactly_one_job()
{
await using var harness = await NewHarnessAsync();
await using var firstDb = new NostosDbContext(harness.Options);
await using var secondDb = new NostosDbContext(harness.Options);
// Unrelated tracked work must survive the losing creator's unique-key
// recovery; the store detaches only its own failed insert.
var firstUnrelated = new TopicModel { Topic = "tracked-unrelated-first" };
var secondUnrelated = new TopicModel { Topic = "tracked-unrelated-second" };
firstDb.Topics.Add(firstUnrelated);
secondDb.Topics.Add(secondUnrelated);
var firstStore = new EfMigrationJobStore(firstDb, harness.Clock);
var secondStore = new EfMigrationJobStore(secondDb, harness.Clock);
var results = await Task.WhenAll(
firstStore.CreateAsync(MigrationDirection.Import, "race-key", CancellationToken.None),
secondStore.CreateAsync(MigrationDirection.Import, "race-key", CancellationToken.None));
results.Count(result => result.IsConflict).Should().Be(0);
results.Count(result => !result.WasReplay).Should().Be(1);
results.Count(result => result.WasReplay).Should().Be(1);
results.Select(result => result.Resource!.Id).Distinct().Should().ContainSingle();
(await harness.CountJobsAsync()).Should().Be(1);
firstDb.Entry(firstUnrelated).State.Should().NotBe(
EntityState.Detached,
"unique-key recovery must not clear unrelated tracked work");
secondDb.Entry(secondUnrelated).State.Should().NotBe(
EntityState.Detached,
"unique-key recovery must not clear unrelated tracked work");
}
[Fact]
public async Task Parallel_creators_with_different_payloads_create_one_job_and_one_conflict()
{
await using var harness = await NewHarnessAsync();
await using var firstDb = new NostosDbContext(harness.Options);
await using var secondDb = new NostosDbContext(harness.Options);
var firstStore = new EfMigrationJobStore(firstDb, harness.Clock);
var secondStore = new EfMigrationJobStore(secondDb, harness.Clock);
var results = await Task.WhenAll(
firstStore.CreateAsync(MigrationDirection.Import, "race-key", CancellationToken.None),
secondStore.CreateAsync(MigrationDirection.Export, "race-key", CancellationToken.None));
results.Count(result => !result.IsConflict && !result.WasReplay).Should().Be(1);
var conflict = results.Single(result => result.IsConflict);
conflict.Conflict!.Kind.Should().Be(
MigrationIdempotencyConflictKind.KeyReusedWithDifferentPayload);
conflict.Resource.Should().BeNull();
(await harness.CountJobsAsync()).Should().Be(1);
}
[Fact]
public async Task Create_rethrows_a_non_unique_database_failure()
{
var path = NewDatabasePath();
var clock = new ManualTimeProvider(T0);
await using var harness = new Harness(
path,
clock,
builder => builder.AddInterceptors(new InsertFailingInterceptor()));
await harness.InitializeAsync();
var create = () => harness.Store.CreateAsync(
MigrationDirection.Import,
"insert-failure-key",
CancellationToken.None);
await create.Should().ThrowAsync<DbUpdateException>();
(await harness.CountJobsAsync()).Should().Be(0);
}
[Fact]
public void Unique_violation_detection_recognizes_both_providers()
{
EfMigrationJobStore.IsUniqueConstraintViolation(
new DbUpdateException(
"sqlite unique",
new SqliteException("UNIQUE constraint failed", 2067)))
.Should().BeTrue();
EfMigrationJobStore.IsUniqueConstraintViolation(
new DbUpdateException(
"postgres unique",
new PostgresException("duplicate key", "ERROR", "ERROR", "23505")))
.Should().BeTrue();
EfMigrationJobStore.IsUniqueConstraintViolation(
new DbUpdateException(
"sqlite busy",
new SqliteException("database is locked", 5)))
.Should().BeFalse("a busy timeout is not the idempotency-key race");
EfMigrationJobStore.IsUniqueConstraintViolation(
new DbUpdateException(
"postgres fk",
new PostgresException("foreign key violation", "ERROR", "ERROR", "23503")))
.Should().BeFalse("a foreign-key violation is not the idempotency-key race");
EfMigrationJobStore.IsUniqueConstraintViolation(new DbUpdateException("no inner"))
.Should().BeFalse("an unrecognized database failure must propagate");
}
[Fact]
public async Task Get_returns_null_for_an_unknown_job()
{
await using var harness = await NewHarnessAsync();
(await harness.Store.GetAsync(Guid.NewGuid(), CancellationToken.None)).Should().BeNull();
}
[Fact]
public async Task First_acquire_wins_and_second_acquire_returns_null()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var first = await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None);
first.Should().NotBeNull();
var second = await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None);
second.Should().BeNull("an unexpired lease must block a second acquire");
var record = await harness.GetRecordAsync(jobId);
record.MigrationLeaseToken.Should().Be(first);
record.LeaseExpiresAtUtc.Should().Be(T0.UtcDateTime.Add(LeaseDuration));
record.HeartbeatAtUtc.Should().Be(T0.UtcDateTime);
record.UpdatedAtUtc.Should().Be(T0.UtcDateTime);
record.Version.Should().Be(1);
}
[Fact]
public async Task Acquire_after_expiry_succeeds_and_the_old_token_is_rejected_everywhere()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var oldToken = await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None);
oldToken.Should().NotBeNull();
var unexpiredSteal = await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None);
unexpiredSteal.Should().BeNull("an unexpired lease can never be stolen");
harness.Clock.Advance(LeaseDuration + TimeSpan.FromSeconds(1));
var newToken = await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None);
newToken.Should().NotBeNull("an expired lease must be reclaimable");
newToken.Should().NotBe(oldToken);
(await harness.Store.RenewLeaseAsync(
jobId,
oldToken!,
LeaseDuration,
CancellationToken.None))
.Should().BeFalse("a superseded token must never renew");
var transition = () => harness.Store.TransitionAsync(
jobId,
MigrationJobState.Validating,
oldToken!,
CancellationToken.None);
var transitionException =
await transition.Should().ThrowAsync<MigrationJobStoreException>();
transitionException.Which.Code.Should().Be(MigrationJobStoreErrorCodes.LeaseConflict);
var progress = () => harness.Store.UpdateProgressAsync(
jobId,
new MigrationProgress(MigrationProgressPhase.Transferring, 1, 100),
oldToken!,
CancellationToken.None);
var progressException = await progress.Should().ThrowAsync<MigrationJobStoreException>();
progressException.Which.Code.Should().Be(MigrationJobStoreErrorCodes.LeaseConflict);
var afterRejection = await harness.GetRecordAsync(jobId);
afterRejection.State.Should().Be((int)MigrationJobState.Transferring);
afterRejection.ProgressBytesProcessed.Should().Be(0);
afterRejection.MigrationLeaseToken.Should().Be(newToken);
(await harness.Store.RenewLeaseAsync(
jobId,
newToken!,
LeaseDuration,
CancellationToken.None))
.Should().BeTrue("the new owner must be able to renew");
await harness.Store.UpdateProgressAsync(
jobId,
new MigrationProgress(MigrationProgressPhase.Transferring, 7, 100),
newToken!,
CancellationToken.None);
var transitioned = await harness.Store.TransitionAsync(
jobId,
MigrationJobState.Validating,
newToken!,
CancellationToken.None);
transitioned.State.Should().Be(MigrationJobState.Validating);
}
[Fact]
public async Task Renew_extends_only_an_unexpired_matching_lease()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var token = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
(await harness.Store.RenewLeaseAsync(
jobId,
"someone-elses-token",
LeaseDuration,
CancellationToken.None))
.Should().BeFalse("a mismatched token must never renew");
var initial = await harness.GetRecordAsync(jobId);
initial.LeaseExpiresAtUtc.Should().Be(T0.UtcDateTime.Add(LeaseDuration));
initial.HeartbeatAtUtc.Should().Be(T0.UtcDateTime);
harness.Clock.Advance(TimeSpan.FromMinutes(1));
(await harness.Store.RenewLeaseAsync(
jobId,
token,
LeaseDuration,
CancellationToken.None))
.Should().BeTrue();
var renewed = await harness.GetRecordAsync(jobId);
renewed.LeaseExpiresAtUtc.Should().Be(
T0.UtcDateTime.AddMinutes(1).Add(LeaseDuration));
renewed.HeartbeatAtUtc.Should().Be(T0.UtcDateTime.AddMinutes(1));
renewed.Version.Should().Be(2);
harness.Clock.Advance(LeaseDuration);
(await harness.Store.RenewLeaseAsync(
jobId,
token,
LeaseDuration,
CancellationToken.None))
.Should().BeFalse("renewal never revives a lease that already expired");
var afterExpiry = await harness.GetRecordAsync(jobId);
afterExpiry.LeaseExpiresAtUtc.Should().Be(
T0.UtcDateTime.AddMinutes(1).Add(LeaseDuration),
"a failed renewal must not extend the stored expiry");
}
[Fact]
public async Task Renew_fails_when_the_job_is_no_longer_processable()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Cancelled,
leaseToken: "still-present-token",
leaseExpiresAtUtc: T0.UtcDateTime.AddMinutes(5));
(await harness.Store.RenewLeaseAsync(
jobId,
"still-present-token",
LeaseDuration,
CancellationToken.None))
.Should().BeFalse(
"cancellation must stop renewal even when the token row was not yet cleared");
}
[Fact]
public async Task Release_with_a_stale_token_is_a_no_op()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var staleToken = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
harness.Clock.Advance(LeaseDuration + TimeSpan.FromSeconds(1));
var currentToken = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
await harness.Store.ReleaseLeaseAsync(jobId, staleToken, CancellationToken.None);
await harness.Store.ReleaseLeaseAsync(jobId, "never-issued", CancellationToken.None);
var record = await harness.GetRecordAsync(jobId);
record.MigrationLeaseToken.Should().Be(
currentToken,
"releasing with a stale token must not clear the current owner's lease");
record.LeaseExpiresAtUtc.Should().NotBeNull();
}
[Fact]
public async Task Release_with_the_current_token_clears_it_and_unknown_jobs_are_a_no_op()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var token = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
await harness.Store.ReleaseLeaseAsync(jobId, token, CancellationToken.None);
var record = await harness.GetRecordAsync(jobId);
record.MigrationLeaseToken.Should().BeNull();
record.LeaseExpiresAtUtc.Should().BeNull();
await harness.Store.ReleaseLeaseAsync(jobId, token, CancellationToken.None);
(await harness.GetRecordAsync(jobId)).Version.Should().Be(record.Version);
await harness.Store.ReleaseLeaseAsync(
Guid.NewGuid(),
token,
CancellationToken.None);
}
[Fact]
public async Task TryAcquire_rejects_non_positive_durations_and_unknown_jobs_are_not_found()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Pending);
var zero = () => harness.Store.TryAcquireLeaseAsync(
jobId,
TimeSpan.Zero,
CancellationToken.None);
await zero.Should().ThrowAsync<ArgumentOutOfRangeException>();
var negative = () => harness.Store.TryAcquireLeaseAsync(
jobId,
TimeSpan.FromSeconds(-1),
CancellationToken.None);
await negative.Should().ThrowAsync<ArgumentOutOfRangeException>();
var unknown = () => harness.Store.TryAcquireLeaseAsync(
Guid.NewGuid(),
LeaseDuration,
CancellationToken.None);
var exception = await unknown.Should().ThrowAsync<MigrationJobStoreException>();
exception.Which.Code.Should().Be(MigrationJobStoreErrorCodes.NotFound);
var terminalId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Completed);
(await harness.Store.TryAcquireLeaseAsync(
terminalId,
LeaseDuration,
CancellationToken.None))
.Should().BeNull("a terminal job cannot be leased");
}
[Fact]
public async Task Every_legal_transition_for_each_direction_succeeds_through_the_store()
{
await using var harness = await NewHarnessAsync();
foreach (var direction in new[] { MigrationDirection.Import, MigrationDirection.Export })
{
var transitions = MigrationJobTransitions.AllowedTransitions(direction);
foreach (var (from, targets) in transitions)
{
foreach (var target in targets)
{
var jobId = await harness.SeedJobAsync(direction, from);
var token = await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None);
token.Should().NotBeNull(
$"the seeded {direction} job in {from} is non-terminal and must be leasable");
var now = harness.Clock.GetUtcNow().UtcDateTime;
var job = await harness.Store.TransitionAsync(
jobId,
target,
token!,
CancellationToken.None);
job.Id.Should().Be(jobId);
job.State.Should().Be(
target,
$"{direction} {from} -> {target} is legal");
if (target == MigrationJobState.Cancelled)
{
job.LeaseToken.Should().BeNull(
"entering Cancelled always clears the worker lease");
job.LeaseExpiresAtUtc.Should().BeNull();
}
else
{
job.LeaseToken.Should().Be(token);
}
job.UpdatedAtUtc.Should().Be(harness.Clock.GetUtcNow());
var record = await harness.GetRecordAsync(jobId);
record.State.Should().Be((int)target);
record.UpdatedAtUtc.Should().Be(now);
if (target == MigrationJobState.Completed)
record.CompletedAtUtc.Should().Be(now);
if (target == MigrationJobState.Cancelled)
record.CancelledAtUtc.Should().Be(now);
}
}
}
}
[Fact]
public async Task Every_illegal_transition_for_each_direction_is_typed_invalid_state_and_leaves_state_unchanged()
{
await using var harness = await NewHarnessAsync();
foreach (var direction in new[] { MigrationDirection.Import, MigrationDirection.Export })
{
var allowed = MigrationJobTransitions.AllowedTransitions(direction);
foreach (var from in Enum.GetValues<MigrationJobState>())
{
var illegalTargets = Enum.GetValues<MigrationJobState>()
.Where(target => !allowed[from].Contains(target))
.ToList();
var jobId = await harness.SeedJobAsync(direction, from);
string token;
if (MigrationJobTransitions.IsTerminal(from))
{
token = "no-lease-token";
}
else
{
token = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
token.Should().NotBeNull();
}
foreach (var target in illegalTargets)
{
var transition = () => harness.Store.TransitionAsync(
jobId,
target,
token,
CancellationToken.None);
var exception =
await transition.Should().ThrowAsync<MigrationJobStoreException>();
exception.Which.Code.Should().Be(
MigrationJobStoreErrorCodes.InvalidState,
$"{direction} {from} -> {target} must be rejected");
var record = await harness.GetRecordAsync(jobId);
record.State.Should().Be(
(int)from,
"a rejected transition must not mutate the job");
}
}
}
}
[Fact]
public async Task Stale_version_with_a_valid_lease_token_is_a_typed_lease_conflict()
{
var path = NewDatabasePath();
var clock = new ManualTimeProvider(T0);
var jobId = Guid.NewGuid();
var versionBumper = new VersionBumpingInterceptor(path, jobId);
await using var harness = new Harness(
path,
clock,
builder => builder.AddInterceptors(versionBumper));
await harness.InitializeAsync();
await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring,
id: jobId);
var token = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
var before = await harness.GetRecordAsync(jobId);
versionBumper.Enabled = true;
var transition = () => harness.Store.TransitionAsync(
jobId,
MigrationJobState.Validating,
token,
CancellationToken.None);
var exception = await transition.Should().ThrowAsync<MigrationJobStoreException>();
exception.Which.Code.Should().Be(
MigrationJobStoreErrorCodes.LeaseConflict,
"a row version that keeps moving is an optimistic concurrency conflict");
versionBumper.Enabled = false;
var after = await harness.GetRecordAsync(jobId);
after.State.Should().Be((int)MigrationJobState.Transferring);
after.MigrationLeaseToken.Should().Be(token);
after.Version.Should().BeGreaterThan(before.Version);
}
[Fact]
public async Task Progress_is_rejected_when_the_job_advanced_to_another_state_before_the_guarded_update()
{
var path = NewDatabasePath();
var clock = new ManualTimeProvider(T0);
var jobId = Guid.NewGuid();
var advancing = new StateAdvancingInterceptor(
path,
jobId,
MigrationJobState.Validating);
await using var harness = new Harness(
path,
clock,
builder => builder.AddInterceptors(advancing));
await harness.InitializeAsync();
await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring,
id: jobId);
var token = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
advancing.Enabled = true;
var progress = () => harness.Store.UpdateProgressAsync(
jobId,
new MigrationProgress(MigrationProgressPhase.Transferring, 123, 500),
token,
CancellationToken.None);
var exception = await progress.Should().ThrowAsync<MigrationJobStoreException>();
exception.Which.Code.Should().Be(MigrationJobStoreErrorCodes.InvalidState);
advancing.Enabled = false;
var record = await harness.GetRecordAsync(jobId);
record.State.Should().Be((int)MigrationJobState.Validating);
record.ProgressPhase.Should().Be(
(int)MigrationProgressPhase.Pending,
"old-phase progress must never be written onto the new phase");
record.ProgressBytesProcessed.Should().Be(0);
record.ProgressTotalBytes.Should().BeNull();
record.ProgressMessage.Should().BeNull();
}
[Fact]
public async Task Progress_retries_across_benign_version_churn()
{
var path = NewDatabasePath();
var clock = new ManualTimeProvider(T0);
var jobId = Guid.NewGuid();
var churn = new CountingVersionBumpingInterceptor(path, jobId, bumps: 1);
await using var harness = new Harness(
path,
clock,
builder => builder.AddInterceptors(churn));
await harness.InitializeAsync();
await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring,
id: jobId);
var token = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
await harness.Store.UpdateProgressAsync(
jobId,
new MigrationProgress(MigrationProgressPhase.Transferring, 123, 500),
token,
CancellationToken.None);
var record = await harness.GetRecordAsync(jobId);
record.State.Should().Be((int)MigrationJobState.Transferring);
record.ProgressBytesProcessed.Should().Be(123);
record.ProgressTotalBytes.Should().Be(500);
record.Version.Should().Be(3, "one lease, one churn bump and one progress write");
}
[Fact]
public async Task Transition_retries_across_version_churn_and_applies_exactly_once()
{
var path = NewDatabasePath();
var clock = new ManualTimeProvider(T0);
var jobId = Guid.NewGuid();
var churn = new CountingVersionBumpingInterceptor(path, jobId, bumps: 1);
await using var harness = new Harness(
path,
clock,
builder => builder.AddInterceptors(churn));
await harness.InitializeAsync();
await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring,
id: jobId);
var token = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
var transitioned = await harness.Store.TransitionAsync(
jobId,
MigrationJobState.Validating,
token,
CancellationToken.None);
transitioned.State.Should().Be(MigrationJobState.Validating);
var record = await harness.GetRecordAsync(jobId);
record.State.Should().Be((int)MigrationJobState.Validating);
record.Version.Should().Be(
3,
"the transition is applied exactly once after one benign version bump");
}
[Fact]
public async Task Progress_updates_are_rejected_without_a_valid_unexpired_lease()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var token = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
var progress = new MigrationProgress(MigrationProgressPhase.Transferring, 10, 100);
var wrongToken = () => harness.Store.UpdateProgressAsync(
jobId,
progress,
"not-the-token",
CancellationToken.None);
var wrongTokenException =
await wrongToken.Should().ThrowAsync<MigrationJobStoreException>();
wrongTokenException.Which.Code.Should().Be(MigrationJobStoreErrorCodes.LeaseConflict);
harness.Clock.Advance(LeaseDuration + TimeSpan.FromSeconds(1));
var expired = () => harness.Store.UpdateProgressAsync(
jobId,
progress,
token,
CancellationToken.None);
var expiredException = await expired.Should().ThrowAsync<MigrationJobStoreException>();
expiredException.Which.Code.Should().Be(MigrationJobStoreErrorCodes.LeaseConflict);
var unknown = () => harness.Store.UpdateProgressAsync(
Guid.NewGuid(),
progress,
token,
CancellationToken.None);
var unknownException = await unknown.Should().ThrowAsync<MigrationJobStoreException>();
unknownException.Which.Code.Should().Be(MigrationJobStoreErrorCodes.NotFound);
var record = await harness.GetRecordAsync(jobId);
record.ProgressPhase.Should().Be((int)MigrationProgressPhase.Pending);
record.ProgressBytesProcessed.Should().Be(0);
record.ProgressTotalBytes.Should().BeNull();
record.ProgressMessage.Should().BeNull();
record.MigrationLeaseToken.Should().Be(token);
}
[Fact]
public async Task Progress_updates_under_the_valid_lease_are_durable_and_map_back()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var token = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
await harness.Store.UpdateProgressAsync(
jobId,
new MigrationProgress(
MigrationProgressPhase.Transferring,
1234,
5000,
CompletedChunks: 1,
TotalChunks: 2,
Message: "halfway"),
token,
CancellationToken.None);
var record = await harness.GetRecordAsync(jobId);
record.ProgressPhase.Should().Be((int)MigrationProgressPhase.Transferring);
record.ProgressBytesProcessed.Should().Be(1234);
record.ProgressTotalBytes.Should().Be(5000);
record.ProgressCompletedChunks.Should().Be(1);
record.ProgressTotalChunks.Should().Be(2);
record.ProgressMessage.Should().Be("halfway");
record.HeartbeatAtUtc.Should().Be(T0.UtcDateTime);
record.UpdatedAtUtc.Should().Be(T0.UtcDateTime);
record.Version.Should().Be(2);
var mapped = await harness.Store.GetAsync(jobId, CancellationToken.None);
mapped!.State.Should().Be(MigrationJobState.Transferring);
mapped.UpdatedAtUtc.Should().Be(T0);
}
[Fact]
public async Task Cancel_is_allowed_from_every_pre_activation_state_and_is_idempotent()
{
await using var harness = await NewHarnessAsync();
foreach (var state in new[]
{
MigrationJobState.Pending,
MigrationJobState.Preparing,
MigrationJobState.Transferring,
MigrationJobState.Validating,
MigrationJobState.ReadyToActivate,
})
{
var jobId = await harness.SeedJobAsync(MigrationDirection.Import, state);
if (state == MigrationJobState.Transferring)
{
var token = await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None);
token.Should().NotBeNull("cancel must work for a leased job without the token");
}
var now = harness.Clock.GetUtcNow().UtcDateTime;
await harness.Store.CancelAsync(
jobId,
new MigrationCancelRequest("user asked"),
CancellationToken.None);
var record = await harness.GetRecordAsync(jobId);
record.State.Should().Be((int)MigrationJobState.Cancelled, $"{state} is cancellable");
record.CancellationReason.Should().Be("user asked");
record.CancelledAtUtc.Should().Be(now);
record.UpdatedAtUtc.Should().Be(now);
record.MigrationLeaseToken.Should().BeNull();
record.LeaseExpiresAtUtc.Should().BeNull();
record.AttemptNumber.Should().Be(1);
var cancelledVersion = record.Version;
await harness.Store.CancelAsync(
jobId,
new MigrationCancelRequest("second click"),
CancellationToken.None);
var replayed = await harness.GetRecordAsync(jobId);
replayed.State.Should().Be((int)MigrationJobState.Cancelled);
replayed.CancellationReason.Should().Be(
"user asked",
"an idempotent replay must not rewrite cancellation history");
replayed.Version.Should().Be(cancelledVersion);
}
}
[Fact]
public async Task Cancel_marks_active_sessions_cancelled_and_preserves_chunk_receipts()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var sessionId = await harness.SeedSessionAsync(
jobId,
MigrationSessionState.Receiving,
withReceipt: true);
await harness.Store.CancelAsync(
jobId,
new MigrationCancelRequest("stop the transfer"),
CancellationToken.None);
var session = await harness.GetSessionAsync(sessionId);
session!.State.Should().Be((int)MigrationSessionState.Cancelled);
session.ReceivedBytes.Should().Be(16L * 1024 * 1024);
(await harness.CountReceiptsAsync(sessionId)).Should().Be(
1,
"verified chunk receipts survive cancellation for a possible retry");
}
[Fact]
public async Task Cancel_is_rejected_at_the_activation_boundary_and_for_terminal_states()
{
await using var harness = await NewHarnessAsync();
var activatingId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Activating);
var activating = () => harness.Store.CancelAsync(
activatingId,
new MigrationCancelRequest(),
CancellationToken.None);
var activatingException =
await activating.Should().ThrowAsync<MigrationJobStoreException>();
activatingException.Which.Code.Should().Be(MigrationJobStoreErrorCodes.CannotCancel);
foreach (var state in new[]
{
MigrationJobState.Completed,
MigrationJobState.Failed,
MigrationJobState.Expired,
})
{
var jobId = await harness.SeedJobAsync(MigrationDirection.Import, state);
var cancel = () => harness.Store.CancelAsync(
jobId,
new MigrationCancelRequest(),
CancellationToken.None);
var exception = await cancel.Should().ThrowAsync<MigrationJobStoreException>();
exception.Which.Code.Should().Be(MigrationJobStoreErrorCodes.InvalidState);
var record = await harness.GetRecordAsync(jobId);
record.State.Should().Be((int)state);
record.CancelledAtUtc.Should().BeNull();
}
var unknown = () => harness.Store.CancelAsync(
Guid.NewGuid(),
new MigrationCancelRequest(),
CancellationToken.None);
var unknownException = await unknown.Should().ThrowAsync<MigrationJobStoreException>();
unknownException.Which.Code.Should().Be(MigrationJobStoreErrorCodes.NotFound);
}
[Fact]
public async Task Transition_to_cancelled_applies_the_same_cancellation_semantics_as_cancel_async()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var sessionId = await harness.SeedSessionAsync(
jobId,
MigrationSessionState.Receiving,
withReceipt: true);
var token = (await harness.Store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None))!;
var now = harness.Clock.GetUtcNow().UtcDateTime;
var cancelled = await harness.Store.TransitionAsync(
jobId,
MigrationJobState.Cancelled,
token,
CancellationToken.None);
cancelled.State.Should().Be(MigrationJobState.Cancelled);
cancelled.LeaseToken.Should().BeNull();
cancelled.LeaseExpiresAtUtc.Should().BeNull();
cancelled.UpdatedAtUtc.Should().Be(harness.Clock.GetUtcNow());
var record = await harness.GetRecordAsync(jobId);
record.CancelledAtUtc.Should().Be(now);
record.MigrationLeaseToken.Should().BeNull();
record.LeaseExpiresAtUtc.Should().BeNull();
(await harness.GetSessionAsync(sessionId))!.State.Should().Be(
(int)MigrationSessionState.Cancelled,
"the transition route into Cancelled must cancel sessions too");
(await harness.CountReceiptsAsync(sessionId)).Should().Be(1);
}
[Fact]
public async Task Cancellation_is_atomic_and_rolls_back_the_job_when_the_session_update_fails()
{
var path = NewDatabasePath();
var clock = new ManualTimeProvider(T0);
var failingSessions = new SessionUpdateFailingInterceptor();
await using var harness = new Harness(
path,
clock,
builder => builder.AddInterceptors(failingSessions));
await harness.InitializeAsync();
var cancelJobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var cancelSessionId = await harness.SeedSessionAsync(
cancelJobId,
MigrationSessionState.Receiving,
withReceipt: true);
var cancelToken = (await harness.Store.TryAcquireLeaseAsync(
cancelJobId,
LeaseDuration,
CancellationToken.None))!;
failingSessions.Enabled = true;
var cancel = () => harness.Store.CancelAsync(
cancelJobId,
new MigrationCancelRequest("crash test"),
CancellationToken.None);
await cancel.Should().ThrowAsync<InvalidOperationException>();
var jobAfterRollback = await harness.GetRecordAsync(cancelJobId);
jobAfterRollback.State.Should().Be(
(int)MigrationJobState.Transferring,
"the job update must roll back with the failed session update");
jobAfterRollback.MigrationLeaseToken.Should().Be(cancelToken);
jobAfterRollback.CancelledAtUtc.Should().BeNull();
(await harness.GetSessionAsync(cancelSessionId))!.State.Should().Be(
(int)MigrationSessionState.Receiving);
// The TransitionAsync route into Cancelled shares the same atomic
// protocol, so the same injected failure must roll both writes back.
var transitionJobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var transitionSessionId = await harness.SeedSessionAsync(
transitionJobId,
MigrationSessionState.Receiving,
withReceipt: false);
var transitionToken = (await harness.Store.TryAcquireLeaseAsync(
transitionJobId,
LeaseDuration,
CancellationToken.None))!;
var transition = () => harness.Store.TransitionAsync(
transitionJobId,
MigrationJobState.Cancelled,
transitionToken,
CancellationToken.None);
await transition.Should().ThrowAsync<InvalidOperationException>();
var transitionAfterRollback = await harness.GetRecordAsync(transitionJobId);
transitionAfterRollback.State.Should().Be((int)MigrationJobState.Transferring);
transitionAfterRollback.MigrationLeaseToken.Should().Be(transitionToken);
(await harness.GetSessionAsync(transitionSessionId))!.State.Should().Be(
(int)MigrationSessionState.Receiving);
failingSessions.Enabled = false;
await harness.Store.CancelAsync(
cancelJobId,
new MigrationCancelRequest("crash test"),
CancellationToken.None);
var retriedTransition = await harness.Store.TransitionAsync(
transitionJobId,
MigrationJobState.Cancelled,
transitionToken,
CancellationToken.None);
retriedTransition.State.Should().Be(MigrationJobState.Cancelled);
(await harness.GetRecordAsync(cancelJobId)).State.Should().Be(
(int)MigrationJobState.Cancelled);
(await harness.GetSessionAsync(cancelSessionId))!.State.Should().Be(
(int)MigrationSessionState.Cancelled);
(await harness.GetSessionAsync(transitionSessionId))!.State.Should().Be(
(int)MigrationSessionState.Cancelled);
}
[Fact]
public async Task Retry_reactivates_each_retryable_state_and_clears_operational_fields()
{
await using var harness = await NewHarnessAsync();
foreach (var state in new[]
{
MigrationJobState.Failed,
MigrationJobState.Cancelled,
MigrationJobState.Expired,
})
{
var preparedStagingId = Guid.NewGuid();
var reservationId = Guid.NewGuid();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
state,
mutate: record =>
{
record.AttemptNumber = 2;
record.MigrationLeaseToken = "old-lease";
record.LeaseExpiresAtUtc = T0.UtcDateTime.AddMinutes(5);
record.HeartbeatAtUtc = T0.UtcDateTime;
record.FailureCode = "source_media_changed";
record.FailureMessage = "The archive changed.";
record.CancellationReason = "user asked";
record.CancelledAtUtc = T0.UtcDateTime;
record.CompletedAtUtc = T0.UtcDateTime;
record.ProgressPhase = (int)MigrationProgressPhase.Validating;
record.ProgressBytesProcessed = 42;
record.ProgressTotalBytes = 100;
record.ProgressCompletedChunks = 1;
record.ProgressTotalChunks = 2;
record.ProgressMessage = "stale";
record.DestinationRevision = "rev-9";
record.PreparedStagingId = preparedStagingId;
record.PreparedImportMetadataJson = "{\"version\":1}";
record.ReservedStorageBytes = 321;
record.ReservationId = reservationId;
});
harness.Clock.Advance(TimeSpan.FromHours(1));
var now = harness.Clock.GetUtcNow();
var retried = await harness.Store.RetryAsync(
jobId,
new MigrationRetryRequest(),
CancellationToken.None);
retried.State.Should().Be(MigrationJobState.Pending, $"{state} is retryable");
retried.LeaseToken.Should().BeNull();
retried.LeaseExpiresAtUtc.Should().BeNull();
retried.FailureCode.Should().BeNull();
retried.FailureMessage.Should().BeNull();
retried.UpdatedAtUtc.Should().Be(now);
var record = await harness.GetRecordAsync(jobId);
record.State.Should().Be((int)MigrationJobState.Pending);
record.AttemptNumber.Should().Be(3, "retry increments the attempt number once");
record.MigrationLeaseToken.Should().BeNull();
record.LeaseExpiresAtUtc.Should().BeNull();
record.HeartbeatAtUtc.Should().BeNull();
record.FailureCode.Should().BeNull();
record.FailureMessage.Should().BeNull();
record.CancellationReason.Should().BeNull();
record.CancelledAtUtc.Should().BeNull();
record.CompletedAtUtc.Should().BeNull();
record.ProgressPhase.Should().Be((int)MigrationProgressPhase.Pending);
record.ProgressBytesProcessed.Should().Be(0);
record.ProgressTotalBytes.Should().BeNull();
record.ProgressCompletedChunks.Should().BeNull();
record.ProgressTotalChunks.Should().BeNull();
record.ProgressMessage.Should().BeNull();
record.ExpiresAtUtc.Should().Be(now.UtcDateTime.Add(EfMigrationJobStore.JobLifetime));
record.DestinationRevision.Should().Be("rev-9");
record.PreparedStagingId.Should().Be(preparedStagingId);
record.PreparedImportMetadataJson.Should().Be("{\"version\":1}");
record.ReservedStorageBytes.Should().Be(321);
record.ReservationId.Should().Be(reservationId);
}
}
[Fact]
public async Task Retry_preserves_sessions_and_chunk_receipts_for_reuse()
{
await using var harness = await NewHarnessAsync();
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Failed);
var sessionId = await harness.SeedSessionAsync(
jobId,
MigrationSessionState.Expired,
withReceipt: true);
await harness.Store.RetryAsync(
jobId,
new MigrationRetryRequest("double-click"),
CancellationToken.None);
var session = await harness.GetSessionAsync(sessionId);
session.Should().NotBeNull("retry preserves the retained upload session");
session!.State.Should().Be(
(int)MigrationSessionState.Expired,
"session reactivation belongs to the transfer service, not the job store");
(await harness.CountReceiptsAsync(sessionId)).Should().Be(1);
(await harness.GetRecordAsync(jobId)).AttemptNumber.Should().Be(2);
}
[Fact]
public async Task Retry_is_rejected_from_every_non_retryable_state()
{
await using var harness = await NewHarnessAsync();
foreach (var state in new[]
{
MigrationJobState.Pending,
MigrationJobState.Preparing,
MigrationJobState.Transferring,
MigrationJobState.Validating,
MigrationJobState.ReadyToActivate,
MigrationJobState.Activating,
MigrationJobState.Completed,
})
{
var jobId = await harness.SeedJobAsync(MigrationDirection.Import, state);
var retry = () => harness.Store.RetryAsync(
jobId,
new MigrationRetryRequest(),
CancellationToken.None);
var exception = await retry.Should().ThrowAsync<MigrationJobStoreException>();
exception.Which.Code.Should().Be(MigrationJobStoreErrorCodes.NotRetryable);
var record = await harness.GetRecordAsync(jobId);
record.State.Should().Be((int)state);
record.AttemptNumber.Should().Be(1);
}
var unknown = () => harness.Store.RetryAsync(
Guid.NewGuid(),
new MigrationRetryRequest(),
CancellationToken.None);
var unknownException = await unknown.Should().ThrowAsync<MigrationJobStoreException>();
unknownException.Which.Code.Should().Be(MigrationJobStoreErrorCodes.NotFound);
}
[Fact]
public async Task Recovery_discovery_returns_exactly_jobs_with_absent_or_expired_leases_in_mutation_order()
{
await using var harness = await NewHarnessAsync();
var noLease = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Pending,
updatedAtUtc: T0.UtcDateTime.AddMinutes(-30));
var expiredLease = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring,
leaseToken: "expired-lease",
leaseExpiresAtUtc: T0.UtcDateTime.AddMinutes(-10),
updatedAtUtc: T0.UtcDateTime.AddMinutes(-20));
var nullExpiry = await harness.SeedJobAsync(
MigrationDirection.Export,
MigrationJobState.Preparing,
leaseToken: "orphaned-token",
leaseExpiresAtUtc: null,
updatedAtUtc: T0.UtcDateTime.AddMinutes(-10));
await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring,
leaseToken: "live-lease",
leaseExpiresAtUtc: T0.UtcDateTime.AddMinutes(10),
updatedAtUtc: T0.UtcDateTime.AddMinutes(-40));
await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Preparing,
leaseToken: "boundary-lease",
leaseExpiresAtUtc: T0.UtcDateTime,
updatedAtUtc: T0.UtcDateTime.AddMinutes(-50));
await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Completed,
updatedAtUtc: T0.UtcDateTime.AddMinutes(-60));
await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Failed,
leaseToken: "failed-lease",
leaseExpiresAtUtc: T0.UtcDateTime.AddMinutes(-10),
updatedAtUtc: T0.UtcDateTime.AddMinutes(-70));
var jobs = await harness.Store.GetJobsNeedingRecoveryAsync(
T0,
CancellationToken.None);
jobs.Select(job => job.Id).Should().Equal(
noLease,
expiredLease,
nullExpiry);
}
[Fact]
public async Task Parallel_workers_contending_for_one_job_yield_exactly_one_lease()
{
await using var harness = await NewHarnessAsync();
for (var iteration = 0; iteration < 2; iteration++)
{
var jobId = await harness.SeedJobAsync(
MigrationDirection.Import,
MigrationJobState.Transferring);
var attempts = Enumerable.Range(0, 8)
.Select(async _ =>
{
await using var db = new NostosDbContext(harness.Options);
var store = new EfMigrationJobStore(db, harness.Clock);
return await store.TryAcquireLeaseAsync(
jobId,
LeaseDuration,
CancellationToken.None);
})
.ToArray();
var tokens = await Task.WhenAll(attempts);
tokens.Count(token => token is not null).Should().Be(
1,
$"iteration {iteration}: exactly one worker may hold the lease");
var winner = tokens.Single(token => token is not null)!;
var record = await harness.GetRecordAsync(jobId);
record.MigrationLeaseToken.Should().Be(winner);
await harness.Store.ReleaseLeaseAsync(jobId, winner, CancellationToken.None);
}
}
[Fact]
public async Task A_store_recreated_over_the_same_database_file_sees_the_same_state()
{
await using var harness = await NewHarnessAsync();
var job = await harness.CreateJobAsync(MigrationDirection.Import, "restart-key");
var token = (await harness.Store.TryAcquireLeaseAsync(
job.Id,
LeaseDuration,
CancellationToken.None))!;
await harness.Store.UpdateProgressAsync(
job.Id,
new MigrationProgress(MigrationProgressPhase.Transferring, 512, 2048),
token,
CancellationToken.None);
await harness.RestartAsync();
var reloaded = await harness.Store.GetAsync(job.Id, CancellationToken.None);
reloaded.Should().NotBeNull();
reloaded!.State.Should().Be(MigrationJobState.Pending);
reloaded.LeaseToken.Should().Be(token, "the lease must survive a process restart");
(await harness.Store.RenewLeaseAsync(
job.Id,
token,
LeaseDuration,
CancellationToken.None))
.Should().BeTrue("the persisted lease is still renewable after restart");
var record = await harness.GetRecordAsync(job.Id);
record.ProgressBytesProcessed.Should().Be(512);
record.ProgressTotalBytes.Should().Be(2048);
var recovery = await harness.Store.GetJobsNeedingRecoveryAsync(
T0.AddMinutes(1),
CancellationToken.None);
recovery.Should().NotContain(
candidate => candidate.Id == job.Id,
"a job with an unexpired persisted lease is not an orphan");
}
private async Task<Harness> NewHarnessAsync(DateTimeOffset? now = null)
{
var harness = new Harness(
NewDatabasePath(),
new ManualTimeProvider(now ?? T0),
configure: null);
await harness.InitializeAsync();
return harness;
}
private string NewDatabasePath()
{
var path = Path.Combine(
Path.GetTempPath(),
$"nostos-679-job-store-{Guid.NewGuid():N}.db");
_databasePaths.Add(path);
return path;
}
public void Dispose()
{
foreach (var path in _databasePaths)
{
foreach (var suffix in new[] { "", "-shm", "-wal" })
{
try
{
File.Delete(path + suffix);
}
catch (IOException)
{
// Best-effort cleanup only.
}
}
}
}
// Forces a lost optimistic race even though the caller still owns an
// unexpired lease: before every UPDATE of the job row it bumps the row's
// Version on a separate connection, so the store's Version predicate can
// never match.
private sealed class VersionBumpingInterceptor(string databasePath, Guid jobId)
: DbCommandInterceptor
{
public bool Enabled { get; set; }
public override ValueTask<InterceptionResult<int>> NonQueryExecutingAsync(
DbCommand command,
CommandEventData eventData,
InterceptionResult<int> result,
CancellationToken cancellationToken = default)
{
if (Enabled
&& command.CommandText.Contains(
"UPDATE \"MigrationJobRecords\"",
StringComparison.Ordinal))
{
using var connection = new SqliteConnection($"Data Source={databasePath}");
connection.Open();
using var bump = connection.CreateCommand();
bump.CommandText =
"UPDATE \"MigrationJobRecords\" " +
"SET \"Version\" = \"Version\" + 1 WHERE \"Id\" = @id";
bump.Parameters.AddWithValue("@id", jobId);
bump.ExecuteNonQuery();
}
return new ValueTask<InterceptionResult<int>>(result);
}
}
private sealed class SessionUpdateFailingInterceptor : DbCommandInterceptor
{
public bool Enabled { get; set; }
public override ValueTask<InterceptionResult<int>> NonQueryExecutingAsync(
DbCommand command,
CommandEventData eventData,
InterceptionResult<int> result,
CancellationToken cancellationToken = default)
{
if (Enabled
&& command.CommandText.Contains(
"UPDATE \"MigrationSessionRecords\"",
StringComparison.Ordinal))
{
throw new InvalidOperationException("injected session update failure");
}
return new ValueTask<InterceptionResult<int>>(result);
}
}
private sealed class InsertFailingInterceptor : DbCommandInterceptor
{
public override InterceptionResult<DbDataReader> ReaderExecuting(
DbCommand command,
CommandEventData eventData,
InterceptionResult<DbDataReader> result)
{
ThrowIfInsert(command);
return result;
}
public override ValueTask<InterceptionResult<DbDataReader>> ReaderExecutingAsync(
DbCommand command,
CommandEventData eventData,
InterceptionResult<DbDataReader> result,
CancellationToken cancellationToken = default)
{
ThrowIfInsert(command);
return new ValueTask<InterceptionResult<DbDataReader>>(result);
}
public override ValueTask<InterceptionResult<int>> NonQueryExecutingAsync(
DbCommand command,
CommandEventData eventData,
InterceptionResult<int> result,
CancellationToken cancellationToken = default)
{
ThrowIfInsert(command);
return new ValueTask<InterceptionResult<int>>(result);
}
private static void ThrowIfInsert(DbCommand command)
{
if (command.CommandText.Contains(
"INSERT INTO \"MigrationJobRecords\"",
StringComparison.Ordinal))
{
throw new DbUpdateException("injected non-unique database failure");
}
}
}
// Simulates another connection committing a phase transition between this
// store's read and its guarded progress UPDATE.
private sealed class StateAdvancingInterceptor(
string databasePath,
Guid jobId,
MigrationJobState targetState) : DbCommandInterceptor
{
public bool Enabled { get; set; }
public override ValueTask<InterceptionResult<int>> NonQueryExecutingAsync(
DbCommand command,
CommandEventData eventData,
InterceptionResult<int> result,
CancellationToken cancellationToken = default)
{
if (Enabled
&& command.CommandText.Contains(
"UPDATE \"MigrationJobRecords\"",
StringComparison.Ordinal)
&& command.CommandText.Contains("\"ProgressPhase\"", StringComparison.Ordinal))
{
using var connection = new SqliteConnection($"Data Source={databasePath}");
connection.Open();
using var advance = connection.CreateCommand();
advance.CommandText =
"UPDATE \"MigrationJobRecords\" " +
"SET \"State\" = @state, \"Version\" = \"Version\" + 1 WHERE \"Id\" = @id";
advance.Parameters.AddWithValue("@state", (int)targetState);
advance.Parameters.AddWithValue("@id", jobId);
advance.ExecuteNonQuery();
}
return new ValueTask<InterceptionResult<int>>(result);
}
}
// Simulates benign same-owner version churn (a heartbeat renewal) for the
// first N guarded updates, leaving state/token/expiry unchanged.
private sealed class CountingVersionBumpingInterceptor(
string databasePath,
Guid jobId,
int bumps) : DbCommandInterceptor
{
private int _remainingBumps = bumps;
public override ValueTask<InterceptionResult<int>> NonQueryExecutingAsync(
DbCommand command,
CommandEventData eventData,
InterceptionResult<int> result,
CancellationToken cancellationToken = default)
{
if (_remainingBumps > 0
&& command.CommandText.Contains(
"UPDATE \"MigrationJobRecords\"",
StringComparison.Ordinal))
{
_remainingBumps--;
using var connection = new SqliteConnection($"Data Source={databasePath}");
connection.Open();
using var bump = connection.CreateCommand();
bump.CommandText =
"UPDATE \"MigrationJobRecords\" " +
"SET \"Version\" = \"Version\" + 1 WHERE \"Id\" = @id";
bump.Parameters.AddWithValue("@id", jobId);
bump.ExecuteNonQuery();
}
return new ValueTask<InterceptionResult<int>>(result);
}
}
private sealed class Harness : IAsyncDisposable
{
public Harness(
string databasePath,
ManualTimeProvider clock,
Action<DbContextOptionsBuilder<NostosDbContext>>? configure)
{
DatabasePath = databasePath;
Clock = clock;
var builder = new DbContextOptionsBuilder<NostosDbContext>()
.UseSqlite($"Data Source={databasePath}");
configure?.Invoke(builder);
Options = builder.Options;
Db = new NostosDbContext(Options);
Store = new EfMigrationJobStore(Db, clock);
}
public string DatabasePath { get; }
public ManualTimeProvider Clock { get; }
public DbContextOptions<NostosDbContext> Options { get; }
public NostosDbContext Db { get; private set; }
public EfMigrationJobStore Store { get; private set; }
public Task InitializeAsync() => Db.Database.EnsureCreatedAsync();
public async Task RestartAsync()
{
await Db.DisposeAsync();
Db = new NostosDbContext(Options);
Store = new EfMigrationJobStore(Db, Clock);
}
public async Task<MigrationJob> CreateJobAsync(
MigrationDirection direction,
string idempotencyKey)
{
var result = await Store.CreateAsync(
direction,
idempotencyKey,
CancellationToken.None);
result.IsConflict.Should().BeFalse();
return result.Resource!;
}
public async Task<Guid> SeedJobAsync(
MigrationDirection direction,
MigrationJobState state,
string? leaseToken = null,
DateTime? leaseExpiresAtUtc = null,
DateTime? updatedAtUtc = null,
Action<MigrationJobRecord>? mutate = null,
Guid? id = null)
{
var now = Clock.GetUtcNow().UtcDateTime;
var record = new MigrationJobRecord
{
Id = id ?? Guid.NewGuid(),
Direction = (int)direction,
State = (int)state,
RecoveryStatus = (int)MigrationRecoveryStatus.NotRequired,
ProgressPhase = (int)MigrationProgressPhase.Pending,
IdempotencyKey = $"seed-{Guid.NewGuid():N}",
CreationPayloadHash = new string('a', 64),
CreatedAtUtc = now,
UpdatedAtUtc = updatedAtUtc ?? now,
ExpiresAtUtc = now.AddDays(7),
AttemptNumber = 1,
MigrationLeaseToken = leaseToken,
LeaseExpiresAtUtc = leaseExpiresAtUtc,
};
mutate?.Invoke(record);
await using var db = new NostosDbContext(Options);
db.MigrationJobRecords.Add(record);
await db.SaveChangesAsync();
return record.Id;
}
public async Task<Guid> SeedSessionAsync(
Guid jobId,
MigrationSessionState state,
bool withReceipt)
{
var now = Clock.GetUtcNow().UtcDateTime;
var session = new MigrationSessionRecord
{
JobId = jobId,
Purpose = (int)MigrationSessionPurpose.Import,
State = (int)state,
TotalBytes = 32L * 1024 * 1024,
ChunkSize = MigrationContractLimits.DefaultChunkBytes,
TotalChunks = 2,
FileIdentitySizeBytes = 32L * 1024 * 1024,
FileIdentitySha256 = new string('b', 64),
IdempotencyKey = $"seed-session-{Guid.NewGuid():N}",
CreationPayloadHash = new string('c', 64),
ReceivedBytes = withReceipt ? 16L * 1024 * 1024 : 0,
CreatedAtUtc = now,
UpdatedAtUtc = now,
ExpiresAtUtc = now.AddHours(MigrationContractLimits.SessionExpiryHours),
StorageKey = $"uploads/session-{Guid.NewGuid():N}/archive.part",
};
await using var db = new NostosDbContext(Options);
db.MigrationSessionRecords.Add(session);
if (withReceipt)
{
db.MigrationChunkReceiptRecords.Add(new MigrationChunkReceiptRecord
{
SessionId = session.Id,
ChunkIndex = 0,
OffsetBytes = 0,
LengthBytes = 16 * 1024 * 1024,
Sha256 = new string('d', 64),
ReceivedAtUtc = now,
});
}
await db.SaveChangesAsync();
return session.Id;
}
public async Task<MigrationJobRecord> GetRecordAsync(Guid jobId)
{
await using var db = new NostosDbContext(Options);
return await db.MigrationJobRecords
.AsNoTracking()
.SingleAsync(record => record.Id == jobId);
}
public async Task<MigrationSessionRecord?> GetSessionAsync(Guid sessionId)
{
await using var db = new NostosDbContext(Options);
return await db.MigrationSessionRecords
.AsNoTracking()
.SingleOrDefaultAsync(session => session.Id == sessionId);
}
public async Task<int> CountJobsAsync()
{
await using var db = new NostosDbContext(Options);
return await db.MigrationJobRecords.CountAsync();
}
public async Task<int> CountReceiptsAsync(Guid sessionId)
{
await using var db = new NostosDbContext(Options);
return await db.MigrationChunkReceiptRecords
.CountAsync(receipt => receipt.SessionId == sessionId);
}
public async ValueTask DisposeAsync() => await Db.DisposeAsync();
}
}