using DodoSSH.Domain; using Microsoft.EntityFrameworkCore; using Npgsql; namespace DodoSSH.Infrastructure.Tests; /// /// Proves the per-vault advisory lock makes change-log sequence order equal commit order. /// /// /// /// This is the executable form of ADR 0003's central hazard. bigserial assigns sequence /// values when the INSERT runs, not when the transaction commits. So without serialising /// writers per vault, transaction A can take sequence 5 while B takes 6, and B can commit first. /// A reader that polls in between sees only 6, advances its cursor past 5, and never learns about /// it — silent, permanent data loss for that item. /// /// /// The first test demonstrates the hazard is real by reproducing it without the lock. The /// second shows the lock removes it. Without the first, the second proves nothing: it would pass /// just as happily if the interleaving never occurred. /// /// [Collection(PostgresCollection.Name)] public sealed class AdvisoryLockOrderingTests(PostgresFixture fixture) { private static readonly DateTimeOffset Now = new(2026, 7, 28, 12, 0, 0, TimeSpan.Zero); [Fact] public async Task WithoutTheLock_ACommittedGapIsObservable() { // Deliberately reproduces the bug. If this ever stops failing to produce a gap, either // PostgreSQL's sequence behaviour changed or the test stopped interleaving — and the // guarantee the next test claims would no longer be meaningful. var vaultId = await SeedVaultAsync(); await using var first = new NpgsqlConnection(fixture.ConnectionString); await using var second = new NpgsqlConnection(fixture.ConnectionString); await first.OpenAsync(); await second.OpenAsync(); await using var firstTransaction = await first.BeginTransactionAsync(); await using var secondTransaction = await second.BeginTransactionAsync(); // A takes the lower sequence... var lowSequence = await InsertChangeAsync(first, firstTransaction, vaultId); // ...B takes the higher one, and commits first. var highSequence = await InsertChangeAsync(second, secondTransaction, vaultId); await secondTransaction.CommitAsync(); highSequence.ShouldBeGreaterThan(lowSequence); // A reader now sees the high sequence but not the low one. Advancing a cursor to // highSequence would skip lowSequence forever once A commits. var visible = await VisibleSequencesAsync(vaultId); visible.ShouldContain(highSequence); visible.ShouldNotContain(lowSequence); await firstTransaction.CommitAsync(); // And here is the damage: the skipped row is now visible, but behind the cursor. var afterBothCommitted = await VisibleSequencesAsync(vaultId); afterBothCommitted.ShouldContain(lowSequence); afterBothCommitted.ShouldContain(highSequence); } [Fact] public async Task WithTheLock_SequenceOrderMatchesCommitOrder() { var vaultId = await SeedVaultAsync(); await using var first = new NpgsqlConnection(fixture.ConnectionString); await using var second = new NpgsqlConnection(fixture.ConnectionString); await first.OpenAsync(); await second.OpenAsync(); await using var firstTransaction = await first.BeginTransactionAsync(); await AcquireVaultLockAsync(first, firstTransaction, vaultId); var lowSequence = await InsertChangeAsync(first, firstTransaction, vaultId); // The second writer blocks on the lock rather than racing ahead to a higher sequence. await using var secondTransaction = await second.BeginTransactionAsync(); var blocked = Task.Run(async () => { await AcquireVaultLockAsync(second, secondTransaction, vaultId); var sequence = await InsertChangeAsync(second, secondTransaction, vaultId) ; await secondTransaction.CommitAsync(); return sequence; }); // Give it a moment to prove it really is waiting, not merely slow. var completedEarly = await Task.WhenAny(blocked, Task.Delay(TimeSpan.FromMilliseconds(750))) ; completedEarly.ShouldNotBe(blocked, "the second writer should be blocked on the advisory lock"); await firstTransaction.CommitAsync(); var highSequence = await blocked; // The writer that committed first holds the lower sequence, so a reader advancing its // cursor monotonically can never skip a committed row. highSequence.ShouldBeGreaterThan(lowSequence); var visible = await VisibleSequencesAsync(vaultId); visible.ShouldContain(lowSequence); visible.ShouldContain(highSequence); } [Fact] public async Task WithTheLock_ConcurrentWritersProduceNoGaps() { // The property that actually matters, under real contention: reading the log in sequence // order after every writer has committed must yield a contiguous, complete set. var vaultId = await SeedVaultAsync(); const int Writers = 12; var tasks = Enumerable.Range(0, Writers).Select(async _ => { await using var connection = new NpgsqlConnection(fixture.ConnectionString); await connection.OpenAsync(); await using var transaction = await connection.BeginTransactionAsync(); await AcquireVaultLockAsync(connection, transaction, vaultId); var sequence = await InsertChangeAsync(connection, transaction, vaultId); await transaction.CommitAsync(); return sequence; }); var sequences = await Task.WhenAll(tasks); sequences.Distinct().Count().ShouldBe(Writers); var visible = await VisibleSequencesAsync(vaultId); visible.Count.ShouldBe(Writers); visible.OrderBy(s => s).ShouldBe(sequences.OrderBy(s => s)); } private static Task AcquireVaultLockAsync( NpgsqlConnection connection, NpgsqlTransaction transaction, Guid vaultId) { var command = connection.CreateCommand(); command.Transaction = transaction; command.CommandText = "SELECT pg_advisory_xact_lock(hashtextextended(@vault::text, 0))"; command.Parameters.AddWithValue("vault", vaultId); return command.ExecuteNonQueryAsync(); } private static async Task InsertChangeAsync( NpgsqlConnection connection, NpgsqlTransaction transaction, Guid vaultId) { var command = connection.CreateCommand(); command.Transaction = transaction; command.CommandText = """ INSERT INTO dodo.sync_change (vault_id, entity_type, entity_id, operation, revision, actor_user_id, occurred_at_utc) VALUES (@vault, 1, @entity, 1, 1, @actor, now()) RETURNING sequence """; command.Parameters.AddWithValue("vault", vaultId); command.Parameters.AddWithValue("entity", Guid.CreateVersion7()); command.Parameters.AddWithValue("actor", Guid.CreateVersion7()); return (long)(await command.ExecuteScalarAsync())!; } /// Sequences a fresh reader can currently see, i.e. committed ones. private async Task> VisibleSequencesAsync(Guid vaultId) { await using var context = fixture.CreateContext(); return await context.VaultChanges .Where(c => c.VaultId == vaultId) .OrderBy(c => c.Sequence) .Select(c => c.Sequence) .ToListAsync() ; } private async Task SeedVaultAsync() { await using var context = fixture.CreateContext(); var user = new UserAccount { Id = Guid.CreateVersion7(), Issuer = "https://idp.example", Subject = Guid.CreateVersion7().ToString(), Status = UserStatus.Active, CreatedAtUtc = Now, UpdatedAtUtc = Now, }; var vault = new Vault { Id = Guid.CreateVersion7(), Name = "Personal", OwnerKind = VaultOwnerKind.Personal, OwnerUserId = user.Id, KeyGeneration = 1, CreatedAtUtc = Now, UpdatedAtUtc = Now, }; context.Users.Add(user); context.Vaults.Add(vault); await context.SaveChangesAsync(); return vault.Id; } }