Server: a cursor carries the sync horizon it was issued under, so a read from the beginning after a collection is not refused on its own pages

A cursor behind the horizon was refused on position alone. But the read a
refused client starts from the beginning is itself behind the horizon on
every page until it passes the collected deletion, since older untouched
items come first. So its second page was refused, it started over, and it
never finished, on any vault larger than one page.

The cursor now records the horizon read before the page that produced it,
and only a cursor issued under an older horizon is refused. A cursor issued
under no horizon keeps the v1 format, so a server from before this still
reads it.

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);
+    }
+
+    /// <summary>The changes after a position, one page of them, and whether there are more.</summary>
+    private async Task<(List<VaultChange> 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
+        return (changes, hasMore);
-        // 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);
     }
 
     /// <summary>

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.
 /// </para>
+/// <para>
+/// 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.
+/// </para>
 /// </remarks>
 public static class SyncCursor
 {
     private const string Version = "v1";
+
+    /// <summary>The format that carries a horizon as well.</summary>
+    private const string HorizonVersion = "v2";
     private const char Separator = '|';
 
     /// <summary>Length of the truncated HMAC tag. 128 bits is ample for a non-secret position.</summary>
@@ -37,7 +48,17 @@ public static class SyncCursor
     /// <param name="signingKey">Deployment cursor signing key, at least 32 bytes.</param>
     /// <param name="vaultId">Vault the cursor belongs to.</param>
     /// <param name="sequence">Last change sequence the client has consumed.</param>
-    public static string Encode(ReadOnlySpan<byte> signingKey, Guid vaultId, long sequence)
+    public static string Encode(ReadOnlySpan<byte> signingKey, Guid vaultId, long sequence) =>
+        Encode(signingKey, vaultId, sequence, horizon: 0);
+
+    /// <summary>
+    /// Encodes a position, with the sync horizon it was read under.
+    /// </summary>
+    /// <param name="signingKey">Deployment cursor signing key, at least 32 bytes.</param>
+    /// <param name="vaultId">Vault the cursor belongs to.</param>
+    /// <param name="sequence">Last change sequence the client has consumed.</param>
+    /// <param name="horizon">The vault's sync horizon before the read that produced this position.</param>
+    public static string Encode(ReadOnlySpan<byte> 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<byte> signingKey,
         string? cursor,
         Guid expectedVaultId,
-        out long sequence)
+        out long sequence) =>
+        TryDecode(signingKey, cursor, expectedVaultId, out sequence, out _);
+
+    /// <summary>
+    /// Decodes and verifies a cursor, with the horizon it was issued under — zero for one issued under none.
+    /// </summary>
+    /// <param name="signingKey">Deployment cursor signing key.</param>
+    /// <param name="cursor">The cursor to verify.</param>
+    /// <param name="expectedVaultId">Vault the request is scoped to.</param>
+    /// <param name="sequence">The decoded sequence, when verification succeeds.</param>
+    /// <param name="horizon">The decoded horizon, when verification succeeds.</param>
+    public static bool TryDecode(
+        ReadOnlySpan<byte> 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) =>
+    private static string FormatPayload(Guid vaultId, long sequence, long horizon) =>
-        string.Create(
+        horizon == 0
+            ? string.Create(
                 CultureInfo.InvariantCulture,
-            $"{Version}{Separator}{vaultId:D}{Separator}{sequence}");
+                $"{Version}{Separator}{vaultId:D}{Separator}{sequence}")
+            : string.Create(
+                CultureInfo.InvariantCulture,
+                $"{HorizonVersion}{Separator}{vaultId:D}{Separator}{sequence}{Separator}{horizon}");
 
-    private static bool TryParsePayload(ReadOnlySpan<byte> payload, Guid expectedVaultId, out long sequence)
+    private static bool TryParsePayload(
+        ReadOnlySpan<byte> 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)
+        if (!long.TryParse(parts[2], NumberStyles.None, CultureInfo.InvariantCulture, out sequence))
-            && sequence >= 0;
+        {
+            return false;
+        }
+
+        return expectedParts == 3
+            || long.TryParse(parts[3], NumberStyles.None, CultureInfo.InvariantCulture, out horizon);
     }
 
     private static void ComputeTag(

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);
     }
 
+    /// <remarks>
+    /// <para>
+    /// 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.
+    /// </para>
+    /// <para>
+    /// One item a page, so the first page's cursor is the old item's position, well behind the horizon.
+    /// </para>
+    /// </remarks>
+    [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<Guid>();
+        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<SyncPullResponse>()).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);
+    }
+
     /// <remarks>
     /// 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.

tests/DodoSSH.Domain.Tests/Sync/SyncCursorTests.cs

@@ -152,6 +152,36 @@ public sealed class SyncCursorTests
         Should.Throw<ArgumentOutOfRangeException>(() => 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()
     {