Server: collect tombstones past their retention, compact superseded change entries, and refuse cursors from before a collection

src/DodoSSH.Api/Features/Sync/TombstoneCollector.cs

@@ -0,0 +1,198 @@
+using DodoSSH.Api.Setup;
+using DodoSSH.Domain;
+using DodoSSH.Infrastructure;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Options;
+
+namespace DodoSSH.Api.Features.Sync;
+
+/// <summary>
+/// Removes tombstones past their retention, and change-log entries a later one has superseded.
+/// </summary>
+/// <remarks>
+/// <para>
+/// ADR 0003 promised this and nothing did it: every deletion stayed a row for ever, and every edit stayed a
+/// change-log entry for ever, so the log grew without bound and a client starting from nothing walked every
+/// edit ever made to read the items that exist.
+/// </para>
+/// <para>
+/// <b>Superseded entries go at any time.</b> A pull returns entries after a cursor in order, so a client
+/// before both an item's old entry and its new one reads the new one, and a client between them already has
+/// the old one. Nothing reads history: a conflict resolves against the newest entry alone.
+/// </para>
+/// <para>
+/// <b>Tombstones go after the retention</b>, and taking them is the one change here a client can notice: a
+/// machine that last synced before the deletion would never hear of it. So each vault records the highest
+/// sequence collected from it — <see cref="Vault.SyncHorizon"/> — and a pull from a cursor below that is
+/// refused. The client answers by starting over, and a pull from nothing removes what the server no longer
+/// has; see the client's <c>SyncEngine</c>.
+/// </para>
+/// <para>
+/// Per vault under the same advisory lock a push takes, so a collection never interleaves with a write to
+/// the vault — and the items are read again under it, so one restored since it was found is left alone.
+/// </para>
+/// </remarks>
+internal sealed class TombstoneCollector(
+    DodoDbContext database,
+    IOptions<SyncOptions> syncOptions,
+    TimeProvider clock,
+    ILogger<TombstoneCollector> logger)
+{
+    /// <summary>Collects what is due, once.</summary>
+    /// <returns>How many tombstones and superseded entries went.</returns>
+    internal async Task<int> RunOnceAsync(CancellationToken cancellationToken)
+    {
+        var before = clock.GetUtcNow() - TimeSpan.FromDays(syncOptions.Value.TombstoneRetentionDays);
+
+        var collected = 0;
+
+        collected += await CollectAsync(db => db.Hosts, ChangeEntityType.SshHost, before, cancellationToken).ConfigureAwait(false);
+        collected += await CollectAsync(db => db.SshKeys, ChangeEntityType.SshKey, before, cancellationToken).ConfigureAwait(false);
+        collected += await CollectAsync(db => db.Credentials, ChangeEntityType.Credential, before, cancellationToken).ConfigureAwait(false);
+        collected += await CollectAsync(db => db.KnownHostKeys, ChangeEntityType.KnownHostKey, before, cancellationToken).ConfigureAwait(false);
+        collected += await CollectAsync(db => db.HostGroups, ChangeEntityType.HostGroup, before, cancellationToken).ConfigureAwait(false);
+        collected += await CollectAsync(db => db.Tags, ChangeEntityType.Tag, before, cancellationToken).ConfigureAwait(false);
+        collected += await CollectAsync(db => db.Snippets, ChangeEntityType.Snippet, before, cancellationToken).ConfigureAwait(false);
+        collected += await CollectAsync(db => db.ConnectionLog, ChangeEntityType.ConnectionLogEntry, before, cancellationToken).ConfigureAwait(false);
+        collected += await CollectAsync(db => db.ActivityLog, ChangeEntityType.ActivityLogEntry, before, cancellationToken).ConfigureAwait(false);
+        collected += await CollectAsync(db => db.ObjectStores, ChangeEntityType.ObjectStore, before, cancellationToken).ConfigureAwait(false);
+        collected += await CollectAsync(db => db.SpelunkKeys, ChangeEntityType.SpelunkKey, before, cancellationToken).ConfigureAwait(false);
+
+        collected += await CompactAsync(cancellationToken).ConfigureAwait(false);
+
+        SyncLog.Collected(logger, collected);
+
+        return collected;
+    }
+
+    private async Task<int> CollectAsync<T>(
+        Func<DodoDbContext, DbSet<T>> table,
+        ChangeEntityType type,
+        DateTimeOffset before,
+        CancellationToken cancellationToken)
+        where T : class, IVaultItem
+    {
+        var due = await table(database)
+            .Where(item => item.DeletedAtUtc != null && item.DeletedAtUtc < before)
+            .Select(item => new { item.VaultId, item.Id })
+            .ToListAsync(cancellationToken)
+            .ConfigureAwait(false);
+
+        var collected = 0;
+
+        foreach (var vault in due.GroupBy(item => item.VaultId))
+        {
+            var ids = vault.Select(item => item.Id).ToArray();
+
+            collected += await database.ExecuteFreshAsync(
+                [],
+                () => CollectFromVaultAsync(table, type, vault.Key, ids, before, cancellationToken))
+                .ConfigureAwait(false);
+        }
+
+        return collected;
+    }
+
+    private async Task<int> CollectFromVaultAsync<T>(
+        Func<DodoDbContext, DbSet<T>> table,
+        ChangeEntityType type,
+        Guid vaultId,
+        Guid[] ids,
+        DateTimeOffset before,
+        CancellationToken cancellationToken)
+        where T : class, IVaultItem
+    {
+        var transaction = await database.Database.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);
+        await using var _ = transaction.ConfigureAwait(false);
+
+        await database.Database
+            .ExecuteSqlAsync($"SELECT pg_advisory_xact_lock(hashtextextended({vaultId.ToString()}, 0))", cancellationToken)
+            .ConfigureAwait(false);
+
+        // Read again under the lock: an item restored since it was found is no longer due.
+        var still = await table(database)
+            .Where(item => item.VaultId == vaultId
+                && ids.Contains(item.Id)
+                && item.DeletedAtUtc != null
+                && item.DeletedAtUtc < before)
+            .Select(item => new { item.Id, item.ChangeSequence })
+            .ToListAsync(cancellationToken)
+            .ConfigureAwait(false);
+
+        if (still.Count == 0)
+        {
+            return 0;
+        }
+
+        var going = still.Select(item => item.Id).ToArray();
+        var horizon = still.Max(item => item.ChangeSequence);
+
+        await table(database)
+            .Where(item => item.VaultId == vaultId && going.Contains(item.Id))
+            .ExecuteDeleteAsync(cancellationToken)
+            .ConfigureAwait(false);
+
+        await database.VaultChanges
+            .Where(change => change.VaultId == vaultId && change.EntityType == type && going.Contains(change.EntityId))
+            .ExecuteDeleteAsync(cancellationToken)
+            .ConfigureAwait(false);
+
+        await database.Vaults
+            .Where(v => v.Id == vaultId && v.SyncHorizon < horizon)
+            .ExecuteUpdateAsync(set => set.SetProperty(v => v.SyncHorizon, horizon), cancellationToken)
+            .ConfigureAwait(false);
+
+        await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
+
+        return still.Count;
+    }
+
+    /// <summary>Removes every change-log entry a later entry for the same item has superseded.</summary>
+    private Task<int> CompactAsync(CancellationToken cancellationToken) =>
+        database.Database.ExecuteSqlRawAsync(
+            "DELETE FROM " + DodoDbContext.SchemaName + ".sync_change AS older "
+            + "USING " + DodoDbContext.SchemaName + ".sync_change AS newer "
+            + "WHERE newer.vault_id = older.vault_id "
+            + "AND newer.entity_type = older.entity_type "
+            + "AND newer.entity_id = older.entity_id "
+            + "AND newer.sequence > older.sequence",
+            cancellationToken);
+}
+
+/// <summary>Runs the <see cref="TombstoneCollector"/> on a schedule, in a scope of its own each time.</summary>
+internal sealed class TombstoneCollectionService(
+    IServiceScopeFactory scopes,
+    IOptions<SyncOptions> syncOptions,
+    TimeProvider clock,
+    ILogger<TombstoneCollectionService> logger) : BackgroundService
+{
+    /// <inheritdoc />
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        var interval = TimeSpan.FromMinutes(syncOptions.Value.TombstoneCollectionIntervalMinutes);
+
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            try
+            {
+                await Task.Delay(interval, clock, stoppingToken).ConfigureAwait(false);
+
+                var scope = scopes.CreateAsyncScope();
+                await using var _ = scope.ConfigureAwait(false);
+
+                await scope.ServiceProvider.GetRequiredService<TombstoneCollector>()
+                    .RunOnceAsync(stoppingToken)
+                    .ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                return;
+            }
+            catch (Exception exception) when (exception is not OutOfMemoryException)
+            {
+                // The next run tries again; nothing collected late is harmed by waiting.
+                SyncLog.CollectionFailed(logger, exception);
+            }
+        }
+    }
+}