diff --git a/src/DodoSSH.Api/Features/Sync/SyncService.cs b/src/DodoSSH.Api/Features/Sync/SyncService.cs index 5b6e001..e50e479 100644 --- a/src/DodoSSH.Api/Features/Sync/SyncService.cs +++ b/src/DodoSSH.Api/Features/Sync/SyncService.cs @@ -26,12 +26,59 @@ internal sealed class SyncService( CancellationToken cancellationToken) { var afterSequence = 0L; + var issuedUnder = 0L; if (!string.IsNullOrEmpty(request.Cursor) - && !SyncCursor.TryDecode(cursorKeys.Key, request.Cursor, vault.Id, out afterSequence)) + && !SyncCursor.TryDecode(cursorKeys.Key, request.Cursor, vault.Id, out afterSequence, out issuedUnder)) { throw new InvalidCursorException(); } + // Read before the changes, for the cursor this read hands out; see the check after it. + var horizonBefore = await HorizonAsync(vault.Id, cancellationToken).ConfigureAwait(false); + + var (changes, hasMore) = await ReadPageAsync(vault.Id, afterSequence, request, cancellationToken) + .ConfigureAwait(false); + + // ◆ After the read, not before it. The collector deletes a tombstone's entries and raises the horizon + // in one transaction, so a read that already missed an entry is followed by a horizon that says so; + // checked first, the two could straddle a collection and the deletion would go unnoticed for good. + // + // A cursor behind the horizon is refused only if it was issued before it. One issued since belongs to + // a read from the beginning that started after the collection, and its early pages are behind the + // horizon because old, untouched items are; refusing those would send that read back to the start + // for ever. See SyncCursor. + var horizon = await HorizonAsync(vault.Id, cancellationToken).ConfigureAwait(false); + + if (afterSequence > 0 && afterSequence < horizon && issuedUnder < horizon) + { + throw new InvalidCursorException(); + } + + var hydrated = await HydrateAsync(vault, changes, cancellationToken).ConfigureAwait(false); + + // When nothing came back the cursor must not move, or a concurrent write landing between + // this read and the next would be skipped forever. + var nextSequence = changes.Count > 0 ? changes[^1].Sequence : afterSequence; + + return new SyncPullResponse( + Changes: hydrated, + // The horizon from before the read, never the one after: a collection landing between the two + // could have taken an entry this page would otherwise have carried, and a cursor claiming the + // later horizon would vouch for a client that never heard of it. The earlier one costs at most + // a restart. + NextCursor: SyncCursor.Encode(cursorKeys.Key, vault.Id, nextSequence, horizonBefore), + HasMore: hasMore, + ServerTime: clock.GetUtcNow(), + CurrentKeyGeneration: (uint)vault.KeyGeneration); + } + + /// The changes after a position, one page of them, and whether there are more. + private async Task<(List Changes, bool HasMore)> ReadPageAsync( + Guid vaultId, + long afterSequence, + SyncPullRequest request, + CancellationToken cancellationToken) + { var limit = Math.Clamp( request.Limit ?? options.DefaultPullLimit, 1, @@ -43,7 +90,7 @@ internal sealed class SyncService( // One extra row, so "is there more" needs no second query. var query = database.VaultChanges - .Where(c => c.VaultId == vault.Id && c.Sequence > afterSequence); + .Where(c => c.VaultId == vaultId && c.Sequence > afterSequence); if (types is not null) { @@ -62,26 +109,7 @@ internal sealed class SyncService( changes.RemoveAt(changes.Count - 1); } - // ◆ After the read, not before it. The collector deletes a tombstone's entries and raises the horizon - // in one transaction, so a read that already missed an entry is followed by a horizon that says so; - // checked first, the two could straddle a collection and the deletion would go unnoticed for good. - if (afterSequence > 0 && afterSequence < await HorizonAsync(vault.Id, cancellationToken).ConfigureAwait(false)) - { - throw new InvalidCursorException(); - } - - var hydrated = await HydrateAsync(vault, changes, cancellationToken).ConfigureAwait(false); - - // When nothing came back the cursor must not move, or a concurrent write landing between - // this read and the next would be skipped forever. - var nextSequence = changes.Count > 0 ? changes[^1].Sequence : afterSequence; - - return new SyncPullResponse( - Changes: hydrated, - NextCursor: SyncCursor.Encode(cursorKeys.Key, vault.Id, nextSequence), - HasMore: hasMore, - ServerTime: clock.GetUtcNow(), - CurrentKeyGeneration: (uint)vault.KeyGeneration); + return (changes, hasMore); } /// diff --git a/src/DodoSSH.Domain/Sync/SyncCursor.cs b/src/DodoSSH.Domain/Sync/SyncCursor.cs index 154e92a..17bae90 100644 --- a/src/DodoSSH.Domain/Sync/SyncCursor.cs +++ b/src/DodoSSH.Domain/Sync/SyncCursor.cs @@ -19,10 +19,21 @@ namespace DodoSSH.Domain.Sync; /// The cursor also carries its vault id, so a legitimately-issued cursor for one vault cannot be /// replayed against another. See ADR 0003. /// +/// +/// And the vault's sync horizon when it was issued — the last deletion the server had forgotten. A cursor +/// behind the horizon is refused when it is older than the collection, because its holder may still have the +/// deleted item; it is fine when it is newer, because then it belongs to a read from the beginning that +/// started after the collection, whose early pages are behind the horizon by nature. Position alone cannot +/// tell those two apart. A cursor issued under no horizon keeps the first format, so a server from before +/// this still reads it. +/// /// public static class SyncCursor { private const string Version = "v1"; + + /// The format that carries a horizon as well. + private const string HorizonVersion = "v2"; private const char Separator = '|'; /// Length of the truncated HMAC tag. 128 bits is ample for a non-secret position. @@ -37,7 +48,17 @@ public static class SyncCursor /// Deployment cursor signing key, at least 32 bytes. /// Vault the cursor belongs to. /// Last change sequence the client has consumed. - public static string Encode(ReadOnlySpan signingKey, Guid vaultId, long sequence) + public static string Encode(ReadOnlySpan signingKey, Guid vaultId, long sequence) => + Encode(signingKey, vaultId, sequence, horizon: 0); + + /// + /// Encodes a position, with the sync horizon it was read under. + /// + /// Deployment cursor signing key, at least 32 bytes. + /// Vault the cursor belongs to. + /// Last change sequence the client has consumed. + /// The vault's sync horizon before the read that produced this position. + public static string Encode(ReadOnlySpan signingKey, Guid vaultId, long sequence, long horizon) { RequireKey(signingKey); @@ -46,7 +67,12 @@ public static class SyncCursor throw new ArgumentOutOfRangeException(nameof(sequence), sequence, "Sequence must not be negative."); } - var payload = Encoding.UTF8.GetBytes(FormatPayload(vaultId, sequence)); + if (horizon < 0) + { + throw new ArgumentOutOfRangeException(nameof(horizon), horizon, "Horizon must not be negative."); + } + + var payload = Encoding.UTF8.GetBytes(FormatPayload(vaultId, sequence, horizon)); var buffer = new byte[payload.Length + TagLength]; payload.CopyTo(buffer, 0); @@ -70,10 +96,27 @@ public static class SyncCursor ReadOnlySpan signingKey, string? cursor, Guid expectedVaultId, - out long sequence) + out long sequence) => + TryDecode(signingKey, cursor, expectedVaultId, out sequence, out _); + + /// + /// Decodes and verifies a cursor, with the horizon it was issued under — zero for one issued under none. + /// + /// Deployment cursor signing key. + /// The cursor to verify. + /// Vault the request is scoped to. + /// The decoded sequence, when verification succeeds. + /// The decoded horizon, when verification succeeds. + public static bool TryDecode( + ReadOnlySpan signingKey, + string? cursor, + Guid expectedVaultId, + out long sequence, + out long horizon) { RequireKey(signingKey); sequence = 0; + horizon = 0; if (string.IsNullOrEmpty(cursor)) { @@ -97,17 +140,26 @@ public static class SyncCursor return false; } - return TryParsePayload(payload, expectedVaultId, out sequence); + return TryParsePayload(payload, expectedVaultId, out sequence, out horizon); } - private static string FormatPayload(Guid vaultId, long sequence) => - string.Create( - CultureInfo.InvariantCulture, - $"{Version}{Separator}{vaultId:D}{Separator}{sequence}"); + private static string FormatPayload(Guid vaultId, long sequence, long horizon) => + horizon == 0 + ? string.Create( + CultureInfo.InvariantCulture, + $"{Version}{Separator}{vaultId:D}{Separator}{sequence}") + : string.Create( + CultureInfo.InvariantCulture, + $"{HorizonVersion}{Separator}{vaultId:D}{Separator}{sequence}{Separator}{horizon}"); - private static bool TryParsePayload(ReadOnlySpan payload, Guid expectedVaultId, out long sequence) + private static bool TryParsePayload( + ReadOnlySpan payload, + Guid expectedVaultId, + out long sequence, + out long horizon) { sequence = 0; + horizon = 0; string text; try @@ -122,12 +174,14 @@ public static class SyncCursor // Tagged, so the shape is ours — but parse defensively anyway, because a key rotation // could make an old format verify against a new expectation. var parts = text.Split(Separator); - if (parts.Length != 3) + var expectedParts = parts[0] switch { - return false; - } + Version => 3, + HorizonVersion => 4, + _ => 0, + }; - if (!string.Equals(parts[0], Version, StringComparison.Ordinal)) + if (parts.Length != expectedParts) { return false; } @@ -137,8 +191,13 @@ public static class SyncCursor return false; } - return long.TryParse(parts[2], NumberStyles.None, CultureInfo.InvariantCulture, out sequence) - && sequence >= 0; + if (!long.TryParse(parts[2], NumberStyles.None, CultureInfo.InvariantCulture, out sequence)) + { + return false; + } + + return expectedParts == 3 + || long.TryParse(parts[3], NumberStyles.None, CultureInfo.InvariantCulture, out horizon); } private static void ComputeTag( diff --git a/tests/DodoSSH.Api.Tests/SyncEndpointTests.cs b/tests/DodoSSH.Api.Tests/SyncEndpointTests.cs index 8342067..b144bcc 100644 --- a/tests/DodoSSH.Api.Tests/SyncEndpointTests.cs +++ b/tests/DodoSSH.Api.Tests/SyncEndpointTests.cs @@ -658,6 +658,65 @@ public sealed class SyncEndpointTests(ApiFixture fixture) (await PullAsync(client, vaultId, null)).Changes.ShouldNotContain(change => change.EntityId == entityId); } + /// + /// + /// The client's way back after a refusal is to read from the beginning — and that read's own cursors are + /// behind the horizon whenever an item older than the collected deletion is still there, which is nearly + /// always. Refused on position alone, the second page sent the read back to the start, every time, for + /// good. The cursor carries the horizon it was issued under, and one issued since the collection passes. + /// + /// + /// One item a page, so the first page's cursor is the old item's position, well behind the horizon. + /// + /// + [Fact] + public async Task AReadFromTheBeginning_AfterACollection_IsNotRefusedOnItsOwnPages() + { + var (subject, vaultId) = await SeedUserWithVaultAsync(); + var client = fixture.CreateClientFor(subject); + + var old = NewCreateBatch(); + await client.PostContractAsync(PushUrl(vaultId), old); + await client.PostContractAsync(PushUrl(vaultId), NewCreateBatch()); + + var doomed = NewCreateBatch(); + var entityId = doomed.Operations[0].EntityId; + await client.PostContractAsync(PushUrl(vaultId), doomed); + + await client.PostContractAsync(PushUrl(vaultId), new SyncPushRequest( + [ + new SyncPushOperation( + Guid.CreateVersion7(), SyncEntityType.Host, entityId, SyncOperation.Delete, + ExpectedVersion: 1, Payload: null, PlaintextFields: null), + ])); + + await AgeTheDeletionAsync(entityId); + await CollectAsync(); + + var seen = new List(); + string? cursor = null; + + for (var page = 0; page < 5; page++) + { + var response = await client.PostContractAsync(PullUrl(vaultId), new SyncPullRequest(cursor, 1, null)); + + response.StatusCode.ShouldBe(HttpStatusCode.OK, $"page {page} was refused"); + + var body = (await response.Content.ReadContractAsync()).ShouldNotBeNull(); + seen.AddRange(body.Changes.Select(change => change.EntityId)); + cursor = body.NextCursor; + + if (!body.HasMore) + { + break; + } + } + + seen.Count.ShouldBe(2); + seen.ShouldContain(old.Operations[0].EntityId); + seen.ShouldNotContain(entityId); + } + /// /// The other half of the retention: a deletion younger than it stays, entries and all, so a machine that /// has been away a week still hears of it. diff --git a/tests/DodoSSH.Domain.Tests/Sync/SyncCursorTests.cs b/tests/DodoSSH.Domain.Tests/Sync/SyncCursorTests.cs index 2c14c0d..7ca2835 100644 --- a/tests/DodoSSH.Domain.Tests/Sync/SyncCursorTests.cs +++ b/tests/DodoSSH.Domain.Tests/Sync/SyncCursorTests.cs @@ -152,6 +152,36 @@ public sealed class SyncCursorTests Should.Throw(() => SyncCursor.Encode(Key, VaultId, -1)); } + [Fact] + public void CarriesTheHorizonItWasIssuedUnder() + { + var cursor = SyncCursor.Encode(Key, VaultId, 12, horizon: 40); + + SyncCursor.TryDecode(Key, cursor, VaultId, out var sequence, out var horizon).ShouldBeTrue(); + sequence.ShouldBe(12); + horizon.ShouldBe(40); + } + + [Fact] + public void UnderNoHorizon_KeepsTheFirstFormat() + { + // So a server from before horizons, still running beside a newer one mid-deploy, reads what the newer + // one issues for every vault nothing has been collected from. + SyncCursor.Encode(Key, VaultId, 12, horizon: 0).ShouldBe(SyncCursor.Encode(Key, VaultId, 12)); + + SyncCursor.TryDecode(Key, SyncCursor.Encode(Key, VaultId, 12), VaultId, out _, out var horizon) + .ShouldBeTrue(); + horizon.ShouldBe(0); + } + + [Fact] + public void ACursorWithAHorizon_IsStillBoundToItsVault() + { + var cursor = SyncCursor.Encode(Key, OtherVaultId, 12, horizon: 40); + + SyncCursor.TryDecode(Key, cursor, VaultId, out _, out _).ShouldBeFalse(); + } + [Fact] public void DifferentSequencesProduceDifferentCursors() {