Public Access
Add sync engine: cursors, push/pull, and the advisory-lock ordering proof (M1)
The vault write path. Push is the only way items change — no per-entity POST/PUT/DELETE — so one place enforces revisions, the change log and access control. The concurrency hazard, now proven rather than asserted: bigserial assigns sequence values when the INSERT runs, not at commit, so transaction A can take sequence 5 while B takes 6 and commits first. A reader polling in between sees only 6, advances past 5, and never learns about it. AdvisoryLockOrderingTests reproduces that gap WITHOUT the lock first — otherwise the with-lock test proves nothing, since it would pass just as happily if the interleaving never occurred — then shows pg_advisory_xact_lock removes it, and that 12 concurrent writers produce no gaps. Cursors are opaque and HMAC-tagged, and carry their vault id. 29 unit tests cover the rejections, which are the point: an accepted-but-wrong cursor is silent data loss, strictly worse than an error a client can resync from. Rejected: tampered tag, tampered payload, foreign signing key, a legitimately-issued cursor from another vault, truncation, and hostile input (never throws — cursors come from clients). Push semantics: - 200 even on partial failure, with per-operation status, so one stale item cannot block everything a client queued while offline. - Conflict returns the server's current row for client-side three-way merge. The server cannot merge ciphertext, so never last-writer-wins. - opId receipts make retries exactly-once per operation, not per batch — a client retrying a partially-overlapping batch after a timeout would otherwise double-apply what landed. - A tombstone beats a late upsert, and delete clears hostname/port: leaving the address would keep the server able to resolve a host the user believes they deleted. - Relay field validation mirrors the DB CHECK so a bad request is a clear Invalid rather than a constraint violation surfacing as a 500. Authorization goes through IVaultAccessService, which returns the same answer for "absent" and "forbidden" — distinguishing them is an existence oracle for other tenants' vault ids. Team vaults are explicitly denied until M3 rather than falling through to a permissive default. JIT provisioning keys on (issuer, subject), never email, and handles the concurrent-first-request race via the unique index. Renamed two domain types: Host -> SshHost, because Host collides with Microsoft.Extensions.Hosting.Host in every file of a web project, and SyncChange -> VaultChange to stop it colliding with the Contracts DTO of the same name. Aliasing at every use site would have been permanent friction. Worth noting: `ef migrations has-pending-model-changes` reported clean after those renames even though the snapshot still said "DodoSSH.Domain.Host" — it diffs tables, not CLR type names. The snapshot was regenerated and the emitted DDL diffed against the previous artifacts/schema/v0.1.sql to confirm the rename produced no schema change. Also removed ConfigureAwait(false) from test methods: xUnit1030 flags it as bypassing parallelization limits, which is why MA0004 is suppressed in test projects. Verified: 0 warnings on a clean rebuild, 146 tests pass (up from 122), format clean. Endpoint-level tests are the immediate next step: they need a WireMock OIDC/JWKS stub and real JWT minting, so the "wrong user is denied" matrix does not exist yet for these two routes. The service-layer authorization and the concurrency property are covered.
This commit is contained in:
@@ -0,0 +1,217 @@
|
||||
using DodoSSH.Domain;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Npgsql;
|
||||
|
||||
namespace DodoSSH.Infrastructure.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Proves the per-vault advisory lock makes change-log sequence order equal commit order.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// This is the executable form of ADR 0003's central hazard. <c>bigserial</c> assigns sequence
|
||||
/// values when the <c>INSERT</c> 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The first test demonstrates the hazard is real by reproducing it <em>without</em> 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
[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<int> 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<long> 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())!;
|
||||
}
|
||||
|
||||
/// <summary>Sequences a fresh reader can currently see, i.e. committed ones.</summary>
|
||||
private async Task<List<long>> 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<Guid> 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;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user