Sync: after a read from the beginning, remove what the server no longer has, and flag an edit that brings a forgotten item back

The server now forgets a deletion past its retention and refuses cursors
from before it. The client then reads everything again, but everything
again says nothing about what went. So a machine away for longer than the
retention kept those items for good, and its next edit of one put it back
on the server for everyone.

A read from the beginning now marks each item it reads, keeping the mark in
the sync state because the read can span passes. When it finishes, it
removes the items of the synced types it never marked. Two kinds are kept:
items newer than the read's bound, which this machine pushed during it, and
items with a change still waiting, which are the user's work. For those,
the push finds the server without the item and sends it as a create, as
before, but now records a RecreatedAfterCollection conflict and counts it,
so the sync message says so instead of undoing someone's deletion silently.

The bound is the highest sequence the cache held when the read began, or
anything higher it has read since. Taking only what the read returned would
have left the last item of a vault in place for ever.

src/DodoSSH.Client.Sync/SyncEngine.cs

@@ -1,5 +1,6 @@
 using System.Globalization;
 using System.Runtime.InteropServices;
+using System.Security.Cryptography;
 using DodoSSH.Client.Api;
 using DodoSSH.Client.Storage;
 using DodoSSH.Contracts;
@@ -155,23 +156,12 @@ public sealed class SyncEngine
             }
 
             await ApplyPageAsync(vaultId, response.Changes, report, cancellationToken).ConfigureAwait(false);
-
-            foreach (var change in response.Changes)
-            {
-                report.Pulled++;
-
-                // Counted apart for the same reason the pushed ones are: a machine reads back the log
-                // entries it just wrote, so a pass that "pulled five changes" may have carried nothing
-                // anybody did. See SyncReport.PulledItems.
-                if (IsLog(change.EntityType))
-                {
-                    report.PulledLogEntries++;
-                }
-            }
+            CountPulled(response.Changes, report);
 
             var advanced = !string.Equals(state.Cursor, response.NextCursor, StringComparison.Ordinal);
 
             state = Record(vaultId, state, response, report);
+            state = await CarrySweepAsync(vaultId, state, response, report, cancellationToken).ConfigureAwait(false);
             await syncState.SaveAsync(state, cancellationToken).ConfigureAwait(false);
 
             if (!response.HasMore)
@@ -188,15 +178,22 @@ public sealed class SyncEngine
         }
     }
 
-    /// <summary>
-    /// Whether the server refused the cursor this client sent, rather than failing for some other reason.
-    /// </summary>
-    /// <remarks>
-    /// Read from the problem code and not from the prose, which is free to change, and never claimed for a
-    /// request that carried no cursor: "from the beginning" is the one position a client is allowed to ask
-    /// for, so a rejection of <em>that</em> is a server this code cannot reason about and has to surface.
-    /// It is also what keeps the retry from looping — the restarted request sends no cursor.
-    /// </remarks>
+    private static void CountPulled(IReadOnlyList<SyncChange> changes, SyncReportBuilder report)
+    {
+        foreach (var change in changes)
+        {
+            report.Pulled++;
+
+            // Counted apart for the same reason the pushed ones are: a machine reads back the log entries it
+            // just wrote, so a pass that "pulled five changes" may have carried nothing anybody did. See
+            // SyncReport.PulledItems.
+            if (IsLog(change.EntityType))
+            {
+                report.PulledLogEntries++;
+            }
+        }
+    }
+
     /// <summary>Whether a change is one the machine wrote about itself rather than one somebody made.</summary>
     /// <remarks>
     /// Used only for reporting. Log entries sync exactly like every other item — they are pulled, pushed,
@@ -206,10 +203,46 @@ public sealed class SyncEngine
     private static bool IsLog(SyncEntityType type) =>
         type is SyncEntityType.ConnectionLogEntry or SyncEntityType.ActivityLogEntry;
 
+    /// <summary>
+    /// Whether the server refused the cursor this client sent, rather than failing for some other reason.
+    /// </summary>
+    /// <remarks>
+    /// Read from the problem code and not from the prose, which is free to change, and never claimed for a
+    /// request that carried no cursor: "from the beginning" is the one position a client is allowed to ask
+    /// for, so a rejection of <em>that</em> is a server this code cannot reason about and has to surface.
+    /// It is also what keeps the retry from looping — the restarted request sends no cursor.
+    /// </remarks>
     private static bool WasTheCursorRefused(DodoSshApiException exception, string? cursor) =>
         !string.IsNullOrEmpty(cursor)
         && string.Equals(exception.Code, ProblemCodes.InvalidCursor, StringComparison.Ordinal);
 
+    /// <summary>Where the pull should carry on from, which is the start when the types it reads have changed.</summary>
+    /// <remarks>
+    /// A cursor read under a different list of types is past changes of the types it did not ask for, so an
+    /// upgrade that adds one — SpelunkKey, ObjectStore — would never receive the items of that type made
+    /// before it. Starting over once is the only way to see them; applying is idempotent, which is what makes
+    /// the restart for a refused cursor safe as well. A state from before the list was kept has none and
+    /// starts over once for the same reason: it may already have missed some.
+    /// </remarks>
+    private async Task<StoredSyncState> ReadStateAsync(
+        Guid vaultId,
+        SyncReportBuilder report,
+        CancellationToken cancellationToken)
+    {
+        var state = await syncState.ReadAsync(vaultId, cancellationToken).ConfigureAwait(false);
+
+        if (state.Cursor is not null && !string.Equals(state.PulledTypes, PulledTypes, StringComparison.Ordinal))
+        {
+            return await StartOverAsync(state, report, cancellationToken).ConfigureAwait(false);
+        }
+
+        // A pull from nothing that is not a restart — the vault's first, or a cache reset — sweeps as well.
+        // There is usually nothing to remove; the one time there is, it is the same case.
+        return state is { Cursor: null, SweepMark: null }
+            ? await SweepingAsync(state, cancellationToken).ConfigureAwait(false)
+            : state;
+    }
+
     /// <summary>
     /// Forgets this vault's position and reads the log again from the beginning.
     /// </summary>
@@ -234,44 +267,22 @@ public sealed class SyncEngine
     /// cache itself is the thing in doubt.
     /// </para>
     /// <para>
-    /// That leaves one gap, and it is worth naming rather than leaving to be discovered: once the server
-    /// starts collecting tombstones — <c>DodoOptions.TombstoneRetentionDays</c>, not implemented yet — a
-    /// replay no longer carries a deletion older than the retention window. A machine that missed such a
-    /// delete and then had its cursor refused would keep the row. Nothing here can tell that from a row
-    /// the server still has, so the answer when it matters will be to clear the mirror as well, not to
-    /// guess.
+    /// A replay does not carry a deletion the server has collected — <c>DodoOptions.TombstoneRetentionDays</c>
+    /// — and that is the commonest refusal of all: a machine away longer than the retention. The rows it no
+    /// longer hears about are the ones the replay does not mark, and are swept when it ends; see
+    /// <see cref="CarrySweepAsync"/>.
     /// </para>
     /// <para>
     /// Written down before the replay begins, so a process that dies mid-replay starts the next one from
     /// the beginning as well, rather than meeting the same refusal again.
     /// </para>
     /// </remarks>
-    /// <summary>Where the pull should carry on from, which is the start when the types it reads have changed.</summary>
-    /// <remarks>
-    /// A cursor read under a different list of types is past changes of the types it did not ask for, so an
-    /// upgrade that adds one — SpelunkKey, ObjectStore — would never receive the items of that type made
-    /// before it. Starting over once is the only way to see them; applying is idempotent, which is what makes
-    /// the restart for a refused cursor safe as well. A state from before the list was kept has none and
-    /// starts over once for the same reason: it may already have missed some.
-    /// </remarks>
-    private async Task<StoredSyncState> ReadStateAsync(
-        Guid vaultId,
-        SyncReportBuilder report,
-        CancellationToken cancellationToken)
-    {
-        var state = await syncState.ReadAsync(vaultId, cancellationToken).ConfigureAwait(false);
-
-        return state.Cursor is not null && !string.Equals(state.PulledTypes, PulledTypes, StringComparison.Ordinal)
-            ? await StartOverAsync(state, report, cancellationToken).ConfigureAwait(false)
-            : state;
-    }
-
     private async Task<StoredSyncState> StartOverAsync(
         StoredSyncState state,
         SyncReportBuilder report,
         CancellationToken cancellationToken)
     {
-        var restarted = state with { Cursor = null };
+        var restarted = await SweepingAsync(state with { Cursor = null }, cancellationToken).ConfigureAwait(false);
 
         await syncState.SaveAsync(restarted, cancellationToken).ConfigureAwait(false);
 
@@ -280,6 +291,68 @@ public sealed class SyncEngine
         return restarted;
     }
 
+    /// <summary>The state with a sweep begun: a new mark, and the bound set from what the cache holds now.</summary>
+    /// <remarks>
+    /// The mark is random rather than the time, because a sweep must never share one with the last: rows that
+    /// one marked would count as seen by this one. A clock can stand still, or go back.
+    /// </remarks>
+    private async Task<StoredSyncState> SweepingAsync(StoredSyncState state, CancellationToken cancellationToken) =>
+        state with
+        {
+            SweepMark = BitConverter.ToInt64(RandomNumberGenerator.GetBytes(sizeof(long))),
+            SweepSeenUpTo = await items.HighestSequenceAsync(state.VaultId, cancellationToken).ConfigureAwait(false),
+        };
+
+    /// <summary>
+    /// Marks what a pull from nothing has read, and at its end removes what it never did.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// The server forgets a deletion once it is past its retention. A machine that was away longer is
+    /// refused its cursor and reads everything again — and everything again is the items that still exist,
+    /// with nothing said about the ones that went. Without this their rows stayed here for good, and the
+    /// next edit of one put it back on the server for everybody.
+    /// </para>
+    /// <para>
+    /// A pull from nothing can take more than one pass, so the mark is in the stored state, not in memory.
+    /// Every pull from nothing sweeps — the first one ever as well, where there is nothing to remove. What
+    /// is kept, and why, is <see cref="ItemStore.SweepAsync"/>'s to say.
+    /// </para>
+    /// </remarks>
+    private async Task<StoredSyncState> CarrySweepAsync(
+        Guid vaultId,
+        StoredSyncState state,
+        SyncPullResponse response,
+        SyncReportBuilder report,
+        CancellationToken cancellationToken)
+    {
+        if (state.SweepMark is not { } mark)
+        {
+            return state;
+        }
+
+        var read = response.Changes.Where(change => reconcilers.ContainsKey(change.EntityType)).ToList();
+
+        await items.MarkSeenAsync(
+            vaultId,
+            mark,
+            read.Select(change => (change.EntityType, change.EntityId)).ToList(),
+            cancellationToken).ConfigureAwait(false);
+
+        var seenUpTo = read.Aggregate(state.SweepSeenUpTo, (highest, change) => Math.Max(highest, change.ChangeSequence));
+
+        if (response.HasMore)
+        {
+            return state with { SweepSeenUpTo = seenUpTo };
+        }
+
+        report.Pruned += await items
+            .SweepAsync(vaultId, ItemKinds.SyncedTypes, mark, seenUpTo, cancellationToken)
+            .ConfigureAwait(false);
+
+        return state with { SweepMark = null, SweepSeenUpTo = 0 };
+    }
+
     private StoredSyncState Record(
         Guid vaultId,
         StoredSyncState state,
@@ -794,6 +867,26 @@ public sealed class SyncEngine
 
         if (failure is null)
         {
+            // An edit of a version this machine once saw means the server had the item and has since let go
+            // of it: deleted elsewhere, and the deletion collected past its retention before this machine
+            // heard of it. The edit is the user's work and goes back, but not silently — somebody deleted
+            // this on purpose. A create has no expected version and is the ordinary case.
+            if (operation.ExpectedVersion is not null)
+            {
+                await conflicts.RecordAsync(
+                    vaultId,
+                    operation.EntityType,
+                    operation.EntityId,
+                    ConflictKind.RecreatedAfterCollection,
+                    ConflictDetails.Encode(
+                        "This was deleted elsewhere while this machine had unsaved changes to it, long enough ago "
+                        + "that the server no longer remembered the deletion. The changes were kept and the item "
+                        + "was put back."),
+                    cancellationToken).ConfigureAwait(false);
+
+                report.Recreated++;
+            }
+
             return true;
         }