diff --git a/docs/adr/0003-sync-protocol.md b/docs/adr/0003-sync-protocol.md index 05443d8..5c7d7fd 100644 --- a/docs/adr/0003-sync-protocol.md +++ b/docs/adr/0003-sync-protocol.md @@ -68,7 +68,15 @@ the Npgsql connection string** — the default; do not enable multiplexing. visible conflicted copy for opaque ones. **It must never silently drop a key or a host.** - Deletes are revisioned tombstones, garbage-collected after 90 days. Sync must therefore be able to read tombstones, which is the one place that legitimately bypasses the soft-delete - query filter — guarded by an explicit permission check. + query filter — guarded by an explicit permission check. **The collection has since been built** + (`TombstoneCollector`), and it costs a machine away for longer than the retention its news of + those deletions. So a collection raises the vault's sync horizon, and a cursor from before it is + refused. The client then reads from the beginning, and when that read finishes it removes the + items it held that the read never mentioned. An item with an unsent local edit is kept: the + push finds the server without it, puts it back, and records a `RecreatedAfterCollection` + conflict. A cursor carries the horizon it was issued under. Without that, the read from the + beginning would be refused on its own early pages, which sit behind the horizon whenever an + older item survives. - **An `Infrastructure.Tests` case must prove cursor ordering under N concurrent pushes.** Without it this ADR's central bug is invisible until production. - Two concurrency mechanisms, deliberately: `version integer` is the client-visible monotonic diff --git a/src/DodoSSH.Client.Shell/ViewModels/VaultViewModel.cs b/src/DodoSSH.Client.Shell/ViewModels/VaultViewModel.cs index a029eda..d0b3b13 100644 --- a/src/DodoSSH.Client.Shell/ViewModels/VaultViewModel.cs +++ b/src/DodoSSH.Client.Shell/ViewModels/VaultViewModel.cs @@ -6451,6 +6451,8 @@ internal sealed partial class VaultViewModel( || pass.PushedItems > 0 || pass.Merged > 0 || pass.Resurrected > 0 + || pass.Recreated > 0 + || pass.Pruned > 0 || pass.DeletesAbandoned > 0 || pass.Parked > 0 || pass.ResyncedFromStart); @@ -13159,13 +13161,27 @@ internal sealed partial class VaultViewModel( + "the beginning. " : string.Empty; + // Not a matter for attention — the server had already deleted these, and the machine is only catching + // up — but said, because items going from the list unannounced looks like losing them. + var pruned = report.Pruned > 0 + ? $" {report.Pruned} item(s) deleted elsewhere while this machine was away were removed." + : string.Empty; + if (!report.NeedsAttention) { - return replayed + (report.Pulled == 0 && report.Pushed == 0 + return replayed + (report.Pulled == 0 && report.Pushed == 0 && report.Pruned == 0 ? "Already up to date." - : $"Synchronised: {report.Pulled} in, {report.Pushed} out."); + : $"Synchronised: {report.Pulled} in, {report.Pushed} out.") + pruned; } + var notes = AttentionNotes(report); + + return replayed + "Synchronised, but: " + string.Join("; ", notes) + "." + pruned; + } + + /// What a pass that needs attention says, one clause per thing. + private static List AttentionNotes(SyncReport report) + { var notes = new List(); // "item(s)", not "host(s)": a vault now holds keys as well, and a report that named the wrong kind @@ -13175,6 +13191,11 @@ internal sealed partial class VaultViewModel( notes.Add($"{report.Resurrected} item(s) deleted elsewhere were kept under a new name"); } + if (report.Recreated > 0) + { + notes.Add($"{report.Recreated} item(s) deleted elsewhere long ago were put back, with this machine's changes"); + } + if (report.DeletesAbandoned > 0) { notes.Add($"{report.DeletesAbandoned} deletion(s) were not applied because of a newer edit"); @@ -13200,7 +13221,7 @@ internal sealed partial class VaultViewModel( + "before you can see anything written since"); } - return replayed + "Synchronised, but: " + string.Join("; ", notes) + "."; + return notes; } private async Task RunAsync(string busyMessage, Func work) diff --git a/src/DodoSSH.Client.Storage/CacheRows.cs b/src/DodoSSH.Client.Storage/CacheRows.cs index 1ce2643..cfca77b 100644 --- a/src/DodoSSH.Client.Storage/CacheRows.cs +++ b/src/DodoSSH.Client.Storage/CacheRows.cs @@ -240,6 +240,15 @@ internal sealed class CachedItemRow public bool IsDeleted { get; set; } public DateTimeOffset UpdatedAtUtc { get; set; } + + /// The last full pull that saw this item on the server; null if none has since this was kept. + /// + /// The server collects tombstones after their retention (ADR 0003), so a machine away longer than that + /// is never told of some deletions: it is refused, starts over, and pulls everything that still exists. + /// What the pull from nothing did not mark is what the server no longer has. See + /// ItemStore.SweepAsync. + /// + public long? SweepMark { get; set; } } /// @@ -361,6 +370,20 @@ internal sealed class SyncStateRow /// beside the cursor so a change of list starts the pull over; see SyncEngine.PullAsync. /// public string? PulledTypes { get; set; } + + /// The mark of the pull from nothing in progress, or null when none is. + /// + /// Kept here rather than in memory because a pull from nothing can take several passes — the pages per + /// pass are capped — and what it has seen has to survive between them. See . + /// + public long? SweepMark { get; set; } + + /// The highest change sequence that pull has read so far. + /// + /// The sweep's upper bound: an item written here later than this — one this machine pushed while the + /// pull was under way — is newer than anything the pull has read, and not evidence of anything. + /// + public long SweepSeenUpTo { get; set; } } /// Something the merge had to override, or an item that could not be processed. diff --git a/src/DodoSSH.Client.Storage/ItemStore.cs b/src/DodoSSH.Client.Storage/ItemStore.cs index a08805e..e3cd271 100644 --- a/src/DodoSSH.Client.Storage/ItemStore.cs +++ b/src/DodoSSH.Client.Storage/ItemStore.cs @@ -177,6 +177,94 @@ public sealed class ItemStore( await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); } + /// Marks the items a pull from nothing has just read as ones the server still has. + /// One update per type in the page; see . + public async Task MarkSeenAsync( + Guid vaultId, + long mark, + IReadOnlyCollection<(SyncEntityType Type, Guid Id)> seen, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(seen); + + if (seen.Count == 0) + { + return; + } + + var context = contexts.CreateDbContext(); + await using var scope = context.ConfigureAwait(false); + + foreach (var group in seen.GroupBy(item => item.Type)) + { + var type = group.Key; + var ids = group.Select(item => item.Id).Distinct().ToList(); + + await context.Set() + .Where(row => row.VaultId == vaultId && row.EntityType == type && ids.Contains(row.EntityId)) + .ExecuteUpdateAsync(set => set.SetProperty(row => row.SweepMark, mark), cancellationToken) + .ConfigureAwait(false); + } + } + + /// The highest change sequence held for a vault, or zero for none. + /// Where a pull from nothing starts its sweep's upper bound; see . + public async Task HighestSequenceAsync(Guid vaultId, CancellationToken cancellationToken) + { + var context = contexts.CreateDbContext(); + await using var scope = context.ConfigureAwait(false); + + return await context.Set() + .Where(row => row.VaultId == vaultId) + .MaxAsync(row => (long?)row.ChangeSequence, cancellationToken) + .ConfigureAwait(false) ?? 0; + } + + /// + /// Removes the items a completed pull from nothing did not see, which are the ones the server no longer has. + /// + /// + /// + /// The other half of the server's tombstone collection. A machine that missed a deletion for longer than + /// the retention is refused its cursor and pulls everything again — and everything again does not include + /// the item, so its copy here would otherwise last for ever, and its next edit would push it back. + /// + /// + /// ◆ Two kinds of row are not evidence and are kept. One newer than was + /// written by this machine's own push during the pull: the bound is the higher of what the cache held + /// when the pull began and what it has read since, and the server's head is past both, so a push lands + /// above it. One with a change still waiting to go is the user's work: it is + /// left for the push, which finds the server without the item and says so — see + /// SyncEngine.RetryAsCreateAsync — rather than being thrown away here unseen. + /// + /// + /// How many items went. + public async Task SweepAsync( + Guid vaultId, + IReadOnlyCollection types, + long mark, + long seenUpTo, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(types); + + var context = contexts.CreateDbContext(); + await using var scope = context.ConfigureAwait(false); + + // Only the types the pull asked for: a row of any other type was never going to be marked. + return await context.Set() + .Where(row => row.VaultId == vaultId + && types.Contains(row.EntityType) + && (row.SweepMark == null || row.SweepMark != mark) + && row.ChangeSequence <= seenUpTo + && !context.Set().Any(pending => + pending.VaultId == row.VaultId + && pending.EntityType == row.EntityType + && pending.EntityId == row.EntityId)) + .ExecuteDeleteAsync(cancellationToken) + .ConfigureAwait(false); + } + /// Removes a tombstone whose change has been seen by everything that needed it. /// /// Only ever called for a row that is already a tombstone. Collecting a live item here would make diff --git a/src/DodoSSH.Client.Storage/Migrations/20261002152108_AddSweepMarks.Designer.cs b/src/DodoSSH.Client.Storage/Migrations/20261002152108_AddSweepMarks.Designer.cs new file mode 100644 index 0000000..4db64d8 --- /dev/null +++ b/src/DodoSSH.Client.Storage/Migrations/20261002152108_AddSweepMarks.Designer.cs @@ -0,0 +1,481 @@ +// +using System; +using DodoSSH.Client.Storage; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; + +#nullable disable + +namespace DodoSSH.Client.Storage.Migrations +{ + [DbContext(typeof(ClientCacheContext))] + [Migration("20261002152108_AddSweepMarks")] + partial class AddSweepMarks + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder.HasAnnotation("ProductVersion", "10.0.10"); + + modelBuilder.Entity("DodoSSH.Client.Storage.CachedItemRow", b => + { + b.Property("VaultId") + .HasColumnType("TEXT") + .HasColumnName("vault_id"); + + b.Property("EntityType") + .HasColumnType("INTEGER") + .HasColumnName("entity_type"); + + b.Property("EntityId") + .HasColumnType("TEXT") + .HasColumnName("entity_id"); + + b.Property("AadVersion") + .HasColumnType("INTEGER") + .HasColumnName("aad_version"); + + b.Property("ChangeSequence") + .HasColumnType("INTEGER") + .HasColumnName("change_sequence"); + + b.Property("DataKeyId") + .HasColumnType("TEXT") + .HasColumnName("data_key_id"); + + b.Property("IsDeleted") + .HasColumnType("INTEGER") + .HasColumnName("is_deleted"); + + b.Property("KeyGeneration") + .HasColumnType("INTEGER") + .HasColumnName("key_generation"); + + b.Property("Payload") + .HasColumnType("BLOB") + .HasColumnName("payload"); + + b.Property("ProtectedFields") + .HasColumnType("BLOB") + .HasColumnName("protected_fields"); + + b.Property("SweepMark") + .HasColumnType("INTEGER") + .HasColumnName("sweep_mark"); + + b.Property("UpdatedAtUtc") + .HasColumnType("INTEGER") + .HasColumnName("updated_at_utc"); + + b.Property("Version") + .HasColumnType("INTEGER") + .HasColumnName("version"); + + b.Property("WrappedDataKey") + .HasColumnType("BLOB") + .HasColumnName("wrapped_data_key"); + + b.HasKey("VaultId", "EntityType", "EntityId") + .HasName("pk_item"); + + b.HasIndex("VaultId", "ChangeSequence") + .HasDatabaseName("ix_item_vault_id_change_sequence"); + + b.HasIndex("VaultId", "EntityType") + .HasDatabaseName("ix_item_vault_id_entity_type"); + + b.ToTable("item", (string)null); + }); + + modelBuilder.Entity("DodoSSH.Client.Storage.CachedVaultKeyWrapRow", b => + { + b.Property("VaultId") + .HasColumnType("TEXT") + .HasColumnName("vault_id"); + + b.Property("KeyGeneration") + .HasColumnType("INTEGER") + .HasColumnName("key_generation"); + + b.Property("WrappedKey") + .IsRequired() + .HasColumnType("BLOB") + .HasColumnName("wrapped_key"); + + b.HasKey("VaultId", "KeyGeneration") + .HasName("pk_vault_key_wrap"); + + b.ToTable("vault_key_wrap", (string)null); + }); + + modelBuilder.Entity("DodoSSH.Client.Storage.CachedVaultRow", b => + { + b.Property("VaultId") + .HasColumnType("TEXT") + .HasColumnName("vault_id"); + + b.Property("Hidden") + .HasColumnType("INTEGER") + .HasColumnName("hidden"); + + b.Property("IsPersonal") + .HasColumnType("INTEGER") + .HasColumnName("is_personal"); + + b.Property("KeyGeneration") + .HasColumnType("INTEGER") + .HasColumnName("key_generation"); + + b.Property("Name") + .IsRequired() + .HasColumnType("TEXT") + .HasColumnName("name"); + + b.Property("Permissions") + .HasColumnType("INTEGER") + .HasColumnName("permissions"); + + b.Property("RekeyRequired") + .HasColumnType("INTEGER") + .HasColumnName("rekey_required"); + + b.Property("TeamId") + .HasColumnType("TEXT") + .HasColumnName("team_id"); + + b.Property("UpdatedAtUtc") + .HasColumnType("INTEGER") + .HasColumnName("updated_at_utc"); + + b.Property("WrappedVaultKey") + .HasColumnType("BLOB") + .HasColumnName("wrapped_vault_key"); + + b.HasKey("VaultId") + .HasName("pk_vault"); + + b.ToTable("vault", (string)null); + }); + + modelBuilder.Entity("DodoSSH.Client.Storage.ConflictRow", b => + { + b.Property("Id") + .HasColumnType("TEXT") + .HasColumnName("id"); + + b.Property("Acknowledged") + .HasColumnType("INTEGER") + .HasColumnName("acknowledged"); + + b.Property("Detail") + .IsRequired() + .HasColumnType("BLOB") + .HasColumnName("detail"); + + b.Property("DetectedAtUtc") + .HasColumnType("INTEGER") + .HasColumnName("detected_at_utc"); + + b.Property("EntityId") + .HasColumnType("TEXT") + .HasColumnName("entity_id"); + + b.Property("EntityType") + .HasColumnType("INTEGER") + .HasColumnName("entity_type"); + + b.Property("Kind") + .HasColumnType("INTEGER") + .HasColumnName("kind"); + + b.Property("VaultId") + .HasColumnType("TEXT") + .HasColumnName("vault_id"); + + b.HasKey("Id") + .HasName("pk_conflict"); + + b.HasIndex("VaultId", "Acknowledged") + .HasDatabaseName("ix_conflict_vault_id_acknowledged"); + + b.HasIndex("VaultId", "EntityType", "EntityId") + .HasDatabaseName("ix_conflict_vault_id_entity_type_entity_id"); + + b.ToTable("conflict", (string)null); + }); + + modelBuilder.Entity("DodoSSH.Client.Storage.OutboxRow", b => + { + b.Property("Sequence") + .ValueGeneratedOnAdd() + .HasColumnType("INTEGER") + .HasColumnName("sequence"); + + b.Property("AadVersion") + .HasColumnType("INTEGER") + .HasColumnName("aad_version"); + + b.Property("AncestorAadVersion") + .HasColumnType("INTEGER") + .HasColumnName("ancestor_aad_version"); + + b.Property("AncestorDataKeyId") + .HasColumnType("TEXT") + .HasColumnName("ancestor_data_key_id"); + + b.Property("AncestorKeyGeneration") + .HasColumnType("INTEGER") + .HasColumnName("ancestor_key_generation"); + + b.Property("AncestorPayload") + .HasColumnType("BLOB") + .HasColumnName("ancestor_payload"); + + b.Property("AncestorProtectedFields") + .HasColumnType("BLOB") + .HasColumnName("ancestor_protected_fields"); + + b.Property("AncestorVersion") + .HasColumnType("INTEGER") + .HasColumnName("ancestor_version"); + + b.Property("AncestorWrappedDataKey") + .HasColumnType("BLOB") + .HasColumnName("ancestor_wrapped_data_key"); + + b.Property("Attempts") + .HasColumnType("INTEGER") + .HasColumnName("attempts"); + + b.Property("DataKeyId") + .HasColumnType("TEXT") + .HasColumnName("data_key_id"); + + b.Property("EntityId") + .HasColumnType("TEXT") + .HasColumnName("entity_id"); + + b.Property("EntityType") + .HasColumnType("INTEGER") + .HasColumnName("entity_type"); + + b.Property("ExpectedVersion") + .HasColumnType("INTEGER") + .HasColumnName("expected_version"); + + b.Property("IsParked") + .HasColumnType("INTEGER") + .HasColumnName("is_parked"); + + b.Property("KeyGeneration") + .HasColumnType("INTEGER") + .HasColumnName("key_generation"); + + b.Property("LastError") + .HasColumnType("TEXT") + .HasColumnName("last_error"); + + b.Property("Operation") + .HasColumnType("INTEGER") + .HasColumnName("operation"); + + b.Property("OperationId") + .HasColumnType("TEXT") + .HasColumnName("operation_id"); + + b.Property("Payload") + .HasColumnType("BLOB") + .HasColumnName("payload"); + + b.Property("ProtectedFields") + .HasColumnType("BLOB") + .HasColumnName("protected_fields"); + + b.Property("QueuedAtUtc") + .HasColumnType("INTEGER") + .HasColumnName("queued_at_utc"); + + b.Property("VaultId") + .HasColumnType("TEXT") + .HasColumnName("vault_id"); + + b.Property("WrappedDataKey") + .HasColumnType("BLOB") + .HasColumnName("wrapped_data_key"); + + b.HasKey("Sequence") + .HasName("pk_outbox"); + + b.HasIndex("OperationId") + .IsUnique() + .HasDatabaseName("ix_outbox_operation_id"); + + b.HasIndex("VaultId", "EntityType", "EntityId") + .IsUnique() + .HasDatabaseName("ix_outbox_vault_id_entity_type_entity_id"); + + b.HasIndex("VaultId", "IsParked", "Sequence") + .HasDatabaseName("ix_outbox_vault_id_is_parked_sequence"); + + b.ToTable("outbox", (string)null); + }); + + modelBuilder.Entity("DodoSSH.Client.Storage.RememberedSignInRow", b => + { + b.Property("Id") + .HasColumnType("INTEGER") + .HasColumnName("id"); + + b.Property("SealedRefreshToken") + .IsRequired() + .HasColumnType("BLOB") + .HasColumnName("sealed_refresh_token"); + + b.Property("UpdatedAtUtc") + .HasColumnType("INTEGER") + .HasColumnName("updated_at_utc"); + + b.HasKey("Id") + .HasName("pk_remembered_sign_in"); + + b.ToTable("remembered_sign_in", null, t => + { + t.HasCheckConstraint("ck_remembered_sign_in_singleton", "id = 1"); + }); + }); + + modelBuilder.Entity("DodoSSH.Client.Storage.SyncStateRow", b => + { + b.Property("VaultId") + .HasColumnType("TEXT") + .HasColumnName("vault_id"); + + b.Property("Cursor") + .HasColumnType("TEXT") + .HasColumnName("cursor"); + + b.Property("KeyGeneration") + .HasColumnType("INTEGER") + .HasColumnName("key_generation"); + + b.Property("LastPulledAtUtc") + .HasColumnType("INTEGER") + .HasColumnName("last_pulled_at_utc"); + + b.Property("LastPushedAtUtc") + .HasColumnType("INTEGER") + .HasColumnName("last_pushed_at_utc"); + + b.Property("PulledTypes") + .HasColumnType("TEXT") + .HasColumnName("pulled_types"); + + b.Property("ServerTimeSkewMs") + .HasColumnType("INTEGER") + .HasColumnName("server_time_skew_ms"); + + b.Property("SweepMark") + .HasColumnType("INTEGER") + .HasColumnName("sweep_mark"); + + b.Property("SweepSeenUpTo") + .HasColumnType("INTEGER") + .HasColumnName("sweep_seen_up_to"); + + b.HasKey("VaultId") + .HasName("pk_sync_state"); + + b.ToTable("sync_state", (string)null); + }); + + modelBuilder.Entity("DodoSSH.Client.Storage.UnlockMaterialRow", b => + { + b.Property("Id") + .HasColumnType("INTEGER") + .HasColumnName("id"); + + b.Property("DeviceId") + .HasColumnType("TEXT") + .HasColumnName("device_id"); + + b.Property("DeviceWrappedPrivateKey") + .HasColumnType("BLOB") + .HasColumnName("device_wrapped_private_key"); + + b.Property("DisplayName") + .HasColumnType("TEXT") + .HasColumnName("display_name"); + + b.Property("Email") + .HasColumnType("TEXT") + .HasColumnName("email"); + + b.Property("Issuer") + .IsRequired() + .HasColumnType("TEXT") + .HasColumnName("issuer"); + + b.Property("KdfAlgorithm") + .IsRequired() + .HasColumnType("TEXT") + .HasColumnName("kdf_algorithm"); + + b.Property("KdfMemoryKibibytes") + .HasColumnType("INTEGER") + .HasColumnName("kdf_memory_kibibytes"); + + b.Property("KdfParallelism") + .HasColumnType("INTEGER") + .HasColumnName("kdf_parallelism"); + + b.Property("KdfPasses") + .HasColumnType("INTEGER") + .HasColumnName("kdf_passes"); + + b.Property("KdfSalt") + .IsRequired() + .HasColumnType("BLOB") + .HasColumnName("kdf_salt"); + + b.Property("KeyGeneration") + .HasColumnType("INTEGER") + .HasColumnName("key_generation"); + + b.Property("ServerUrl") + .IsRequired() + .HasColumnType("TEXT") + .HasColumnName("server_url"); + + b.Property("Subject") + .IsRequired() + .HasColumnType("TEXT") + .HasColumnName("subject"); + + b.Property("UpdatedAtUtc") + .HasColumnType("INTEGER") + .HasColumnName("updated_at_utc"); + + b.Property("UserId") + .HasColumnType("TEXT") + .HasColumnName("user_id"); + + b.Property("WrappedPrivateKey") + .IsRequired() + .HasColumnType("BLOB") + .HasColumnName("wrapped_private_key"); + + b.HasKey("Id") + .HasName("pk_unlock_material"); + + b.ToTable("unlock_material", null, t => + { + t.HasCheckConstraint("ck_unlock_material_singleton", "id = 1"); + }); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/DodoSSH.Client.Storage/Migrations/20261002152108_AddSweepMarks.cs b/src/DodoSSH.Client.Storage/Migrations/20261002152108_AddSweepMarks.cs new file mode 100644 index 0000000..5ab5967 --- /dev/null +++ b/src/DodoSSH.Client.Storage/Migrations/20261002152108_AddSweepMarks.cs @@ -0,0 +1,49 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace DodoSSH.Client.Storage.Migrations +{ + /// + public partial class AddSweepMarks : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.AddColumn( + name: "sweep_mark", + table: "sync_state", + type: "INTEGER", + nullable: true); + + migrationBuilder.AddColumn( + name: "sweep_seen_up_to", + table: "sync_state", + type: "INTEGER", + nullable: false, + defaultValue: 0L); + + migrationBuilder.AddColumn( + name: "sweep_mark", + table: "item", + type: "INTEGER", + nullable: true); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropColumn( + name: "sweep_mark", + table: "sync_state"); + + migrationBuilder.DropColumn( + name: "sweep_seen_up_to", + table: "sync_state"); + + migrationBuilder.DropColumn( + name: "sweep_mark", + table: "item"); + } + } +} diff --git a/src/DodoSSH.Client.Storage/Migrations/ClientCacheContextModelSnapshot.cs b/src/DodoSSH.Client.Storage/Migrations/ClientCacheContextModelSnapshot.cs index 60cb699..a0b2dd7 100644 --- a/src/DodoSSH.Client.Storage/Migrations/ClientCacheContextModelSnapshot.cs +++ b/src/DodoSSH.Client.Storage/Migrations/ClientCacheContextModelSnapshot.cs @@ -59,6 +59,10 @@ namespace DodoSSH.Client.Storage.Migrations .HasColumnType("BLOB") .HasColumnName("protected_fields"); + b.Property("SweepMark") + .HasColumnType("INTEGER") + .HasColumnName("sweep_mark"); + b.Property("UpdatedAtUtc") .HasColumnType("INTEGER") .HasColumnName("updated_at_utc"); @@ -370,6 +374,14 @@ namespace DodoSSH.Client.Storage.Migrations .HasColumnType("INTEGER") .HasColumnName("server_time_skew_ms"); + b.Property("SweepMark") + .HasColumnType("INTEGER") + .HasColumnName("sweep_mark"); + + b.Property("SweepSeenUpTo") + .HasColumnType("INTEGER") + .HasColumnName("sweep_seen_up_to"); + b.HasKey("VaultId") .HasName("pk_sync_state"); diff --git a/src/DodoSSH.Client.Storage/StoredTypes.cs b/src/DodoSSH.Client.Storage/StoredTypes.cs index 08b7a8f..2854102 100644 --- a/src/DodoSSH.Client.Storage/StoredTypes.cs +++ b/src/DodoSSH.Client.Storage/StoredTypes.cs @@ -35,6 +35,12 @@ public enum ConflictKind /// The server refused the operation outright. Retrying will not help. Rejected = 6, + + /// + /// Someone else deleted an item this machine had edited, so long ago that the server had already let go + /// of the deletion. The edit was sent as the item again, under its own name and id. + /// + RecreatedAfterCollection = 7, } /// What an offline unlock needs. @@ -191,7 +197,9 @@ public sealed record StoredSyncState( DateTimeOffset? LastPulledAt = null, DateTimeOffset? LastPushedAt = null, long ServerTimeSkewMs = 0, - string? PulledTypes = null); + string? PulledTypes = null, + long? SweepMark = null, + long SweepSeenUpTo = 0); /// Something the merge overrode, or an item that could not be processed. /// This record's own id, which its sealed detail is bound to. diff --git a/src/DodoSSH.Client.Storage/SyncStateStore.cs b/src/DodoSSH.Client.Storage/SyncStateStore.cs index 6e0af23..75aa219 100644 --- a/src/DodoSSH.Client.Storage/SyncStateStore.cs +++ b/src/DodoSSH.Client.Storage/SyncStateStore.cs @@ -47,7 +47,9 @@ public sealed class SyncStateStore(IDbContextFactory context row.LastPulledAtUtc, row.LastPushedAtUtc, row.ServerTimeSkewMs, - row.PulledTypes); + row.PulledTypes, + row.SweepMark, + row.SweepSeenUpTo); } /// Records a vault's position. @@ -74,6 +76,8 @@ public sealed class SyncStateStore(IDbContextFactory context row.LastPushedAtUtc = state.LastPushedAt; row.ServerTimeSkewMs = state.ServerTimeSkewMs; row.PulledTypes = state.PulledTypes; + row.SweepMark = state.SweepMark; + row.SweepSeenUpTo = state.SweepSeenUpTo; await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); } diff --git a/src/DodoSSH.Client.Sync/SyncEngine.cs b/src/DodoSSH.Client.Sync/SyncEngine.cs index 7283f4e..cbefd64 100644 --- a/src/DodoSSH.Client.Sync/SyncEngine.cs +++ b/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 } } - /// - /// Whether the server refused the cursor this client sent, rather than failing for some other reason. - /// - /// - /// 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 that 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. - /// + private static void CountPulled(IReadOnlyList 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++; + } + } + } + /// Whether a change is one the machine wrote about itself rather than one somebody made. /// /// 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; + /// + /// Whether the server refused the cursor this client sent, rather than failing for some other reason. + /// + /// + /// 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 that 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. + /// private static bool WasTheCursorRefused(DodoSshApiException exception, string? cursor) => !string.IsNullOrEmpty(cursor) && string.Equals(exception.Code, ProblemCodes.InvalidCursor, StringComparison.Ordinal); + /// Where the pull should carry on from, which is the start when the types it reads have changed. + /// + /// 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. + /// + private async Task 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; + } + /// /// Forgets this vault's position and reads the log again from the beginning. /// @@ -234,44 +267,22 @@ public sealed class SyncEngine /// cache itself is the thing in doubt. /// /// - /// That leaves one gap, and it is worth naming rather than leaving to be discovered: once the server - /// starts collecting tombstones — DodoOptions.TombstoneRetentionDays, 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 — DodoOptions.TombstoneRetentionDays + /// — 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 + /// . /// /// /// 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. /// /// - /// Where the pull should carry on from, which is the start when the types it reads have changed. - /// - /// 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. - /// - private async Task 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 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; } + /// The state with a sweep begun: a new mark, and the bound set from what the cache holds now. + /// + /// 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. + /// + private async Task SweepingAsync(StoredSyncState state, CancellationToken cancellationToken) => + state with + { + SweepMark = BitConverter.ToInt64(RandomNumberGenerator.GetBytes(sizeof(long))), + SweepSeenUpTo = await items.HighestSequenceAsync(state.VaultId, cancellationToken).ConfigureAwait(false), + }; + + /// + /// Marks what a pull from nothing has read, and at its end removes what it never did. + /// + /// + /// + /// 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. + /// + /// + /// 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 's to say. + /// + /// + private async Task 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; } diff --git a/src/DodoSSH.Client.Sync/SyncReport.cs b/src/DodoSSH.Client.Sync/SyncReport.cs index 644d36a..6bc3602 100644 --- a/src/DodoSSH.Client.Sync/SyncReport.cs +++ b/src/DodoSSH.Client.Sync/SyncReport.cs @@ -84,6 +84,17 @@ public sealed record SyncOptions /// changed anything, which is otherwise the sort of thing that looks like a fault. /// /// +/// How many of were log entries. +/// How many of were log entries. +/// +/// Items this machine had edited that somebody else had deleted so long ago the server no longer +/// remembered it. The edit was kept and the item sent again under its own name — and the user is told, +/// because the deletion was somebody's decision. +/// +/// +/// Items removed from this machine because a read from the beginning found the server no longer has them: +/// deleted elsewhere while this machine was away for longer than the server keeps deletions. +/// public sealed record SyncReport( Guid VaultId, int Pulled, @@ -99,11 +110,13 @@ public sealed record SyncReport( bool RoundsExhausted, bool ResyncedFromStart, int PushedLogEntries = 0, - int PulledLogEntries = 0) + int PulledLogEntries = 0, + int Recreated = 0, + int Pruned = 0) { /// Whether anything happened that a user should be told about. public bool NeedsAttention => - Resurrected > 0 || DeletesAbandoned > 0 || Parked > 0 || Unreadable > 0 || RekeyRequired; + Resurrected > 0 || Recreated > 0 || DeletesAbandoned > 0 || Parked > 0 || Unreadable > 0 || RekeyRequired; /// /// Operations the server accepted that were somebody's own work rather than a log entry. @@ -160,6 +173,10 @@ internal sealed class SyncReportBuilder(Guid vaultId) internal bool ResyncedFromStart { get; set; } + internal int Recreated { get; set; } + + internal int Pruned { get; set; } + internal SyncReport Build() => new( vaultId, @@ -176,7 +193,9 @@ internal sealed class SyncReportBuilder(Guid vaultId) RoundsExhausted, ResyncedFromStart, PushedLogEntries, - PulledLogEntries); + PulledLogEntries, + Recreated, + Pruned); } /// The record written to the conflict log when a merge had to override something. diff --git a/tests/DodoSSH.Client.Sync.Tests/FakeVaultServer.cs b/tests/DodoSSH.Client.Sync.Tests/FakeVaultServer.cs index d11897d..1b83361 100644 --- a/tests/DodoSSH.Client.Sync.Tests/FakeVaultServer.cs +++ b/tests/DodoSSH.Client.Sync.Tests/FakeVaultServer.cs @@ -102,6 +102,7 @@ internal sealed class FakeVaultServer : ISyncApi SyncPullRequest request, CancellationToken cancellationToken) { + var issuedUnder = horizon; var after = DecodeCursor(request.Cursor); var limit = Math.Clamp(request.Limit ?? MaxPullLimit, 1, MaxPullLimit); @@ -128,7 +129,7 @@ internal sealed class FakeVaultServer : ISyncApi return Task.FromResult(new SyncPullResponse( [.. page.Select(entry => Hydrate(entry))], - EncodeCursor(next), + EncodeCursor(next, issuedUnder), hasMore, Now, KeyGeneration)); @@ -206,7 +207,25 @@ internal sealed class FakeVaultServer : ISyncApi internal Row? Find(Guid entityId, SyncEntityType entityType = SyncEntityType.Host) => rows.TryGetValue((entityType, entityId), out var row) ? row : null; - private long Head => log.Count == 0 ? 0 : log[^1].Sequence; + /// + /// Forgets every deletion, as the real server does once one is past its retention: the tombstone and + /// its log entries go, and a cursor from before them is refused from then on. + /// + internal void CollectTombstones() + { + foreach (var (key, row) in rows.Where(entry => entry.Value.IsDeleted).ToList()) + { + rows.Remove(key); + log.RemoveAll(entry => entry.EntityType == key.Type && entry.EntityId == key.EntityId); + horizon = Math.Max(horizon, row.ChangeSequence); + } + } + + /// The newest sequence ever issued, which a collection must not take back. + private long Head => Math.Max(log.Count == 0 ? 0 : log[^1].Sequence, horizon); + + /// The last deletion forgotten; see . + private long horizon; // ---- The decision table ---- @@ -481,8 +500,12 @@ internal sealed class FakeVaultServer : ISyncApi /// rejects. The real server HMAC-tags them; the property that matters to the client is only that it /// must round-trip what it is given. /// - private static string EncodeCursor(long sequence) => - "fake-v1:" + sequence.ToString(CultureInfo.InvariantCulture); + /// + /// Carries the horizon it was read under, as the real one does, so a read from the beginning that started + /// after a collection is not refused on its own early pages. See DodoSSH.Domain.Sync.SyncCursor. + /// + private static string EncodeCursor(long sequence, long issuedUnder = 0) => + string.Create(CultureInfo.InvariantCulture, $"fake-v1:{sequence}@{issuedUnder}"); private long DecodeCursor(string? cursor) { @@ -502,12 +525,26 @@ internal sealed class FakeVaultServer : ISyncApi return 0; } + var at = cursor.IndexOf('@', StringComparison.Ordinal); + if (!cursor.StartsWith("fake-v1:", StringComparison.Ordinal) - || !long.TryParse(cursor.AsSpan(8), CultureInfo.InvariantCulture, out var sequence)) + || at < 0 + || !long.TryParse(cursor.AsSpan(8, at - 8), CultureInfo.InvariantCulture, out var sequence) + || !long.TryParse(cursor.AsSpan(at + 1), CultureInfo.InvariantCulture, out var issuedUnder)) { throw new InvalidOperationException($"A client sent a cursor it should not have: '{cursor}'."); } + if (sequence < horizon && issuedUnder < horizon) + { + // From before a deletion this server no longer remembers, so it could never be told of it. + throw new DodoSshApiException( + HttpStatusCode.BadRequest, + ProblemCodes.InvalidCursor, + "The server returned 400: The sync cursor is not valid for this vault. " + + "Resync from the beginning."); + } + return sequence; } diff --git a/tests/DodoSSH.Client.Sync.Tests/TombstoneCollectionTests.cs b/tests/DodoSSH.Client.Sync.Tests/TombstoneCollectionTests.cs new file mode 100644 index 0000000..f8fb1cd --- /dev/null +++ b/tests/DodoSSH.Client.Sync.Tests/TombstoneCollectionTests.cs @@ -0,0 +1,142 @@ +using DodoSSH.Client.Storage; +using static DodoSSH.Client.Sync.Tests.SyncHarness; + +namespace DodoSSH.Client.Sync.Tests; + +/// +/// A machine away for longer than the server keeps deletions (ADR 0003). +/// +/// +/// The server forgets a deletion once it is past its retention and refuses any cursor from before it. The +/// machine then reads everything again — which says nothing about what is gone — so what it held and did +/// not read again is what has to go. Except the user's own unsent work, which is kept and said out loud. +/// +public sealed class TombstoneCollectionTests +{ + [Fact] + public async Task AnItemDeletedAndForgottenWhileAway_IsRemovedByTheReadFromTheBeginning() + { + using var harness = await CreateAsync(); + + var gone = await harness.Second.CreateAsync(Host("old-db")); + var kept = await harness.Second.CreateAsync(Host("prod-db")); + await harness.Second.SyncAsync(); + await harness.First.SyncAsync(); + + await harness.Second.DeleteAsync(gone); + await harness.Second.SyncAsync(); + harness.Server.CollectTombstones(); + + var report = await harness.First.SyncAsync(); + + report.ResyncedFromStart.ShouldBeTrue(); + report.Pruned.ShouldBe(1); + (await harness.First.ListAsync()).Items.Select(item => item.EntityId).ShouldBe([kept]); + } + + [Fact] + public async Task TheOnlyItemInAVault_IsRemovedAsWell() + { + // The read from the beginning brings back nothing at all, so a sweep bounded by what it read would + // reach nothing either. It is bounded by what the cache held when it started. + using var harness = await CreateAsync(); + + var gone = await harness.Second.CreateAsync(Host("old-db")); + await harness.Second.SyncAsync(); + await harness.First.SyncAsync(); + + await harness.Second.DeleteAsync(gone); + await harness.Second.SyncAsync(); + harness.Server.CollectTombstones(); + + var report = await harness.First.SyncAsync(); + + report.Pruned.ShouldBe(1); + (await harness.First.ListAsync()).Items.ShouldBeEmpty(); + } + + [Fact] + public async Task AnItemWithAnUnsentEdit_IsKept_PutBack_AndSaidSo() + { + using var harness = await CreateAsync(); + + var entityId = await harness.Second.CreateAsync(Host("old-db")); + await harness.Second.SyncAsync(); + await harness.First.SyncAsync(); + + await harness.First.UpdateAsync(entityId, Host("old-db, still needed")); + + await harness.Second.DeleteAsync(entityId); + await harness.Second.SyncAsync(); + harness.Server.CollectTombstones(); + + var report = await harness.First.SyncAsync(); + + report.Pruned.ShouldBe(0); + report.Recreated.ShouldBe(1); + report.NeedsAttention.ShouldBeTrue(); + + (await harness.First.FindAsync(entityId)).Secret.Label.ShouldBe("old-db, still needed"); + harness.Server.Find(entityId).ShouldNotBeNull().IsDeleted.ShouldBeFalse(); + + (await harness.First.ConflictsAsync()) + .ShouldContain(conflict => conflict.EntityId == entityId + && conflict.Kind == ConflictKind.RecreatedAfterCollection); + + await harness.Second.SyncAsync(); + (await harness.Second.FindAsync(entityId)).Secret.Label.ShouldBe("old-db, still needed"); + } + + [Fact] + public async Task AReadSpreadOverSeveralPasses_SweepsAtTheEnd_AndKeepsWhatWasPushedMeanwhile() + { + // Pages per pass are capped, so the read from the beginning can take more than one pass, and what it + // has seen has to outlast each one. A host made here during it is pushed between the passes. + using var harness = await CreateAsync(new SyncOptions { PullPageSize = 1, MaxPullPages = 2 }); + + harness.Server.MaxPullLimit = 1; + + var gone = await harness.Second.CreateAsync(Host("old-db")); + var kept = new List(); + + for (var index = 0; index < 4; index++) + { + kept.Add(await harness.Second.CreateAsync(Host($"host-{index}"))); + } + + // Settled by hand: one pass reads at most two pages, and each needs several. + for (var pass = 0; pass < 6; pass++) + { + await harness.Second.SyncAsync(); + await harness.First.SyncAsync(); + } + + await harness.Second.DeleteAsync(gone); + await harness.Second.SyncAsync(); + harness.Server.CollectTombstones(); + + kept.Add(await harness.First.CreateAsync(Host("made-meanwhile"))); + + var pruned = 0; + + for (var pass = 0; pass < 10; pass++) + { + pruned += (await harness.First.SyncAsync()).Pruned; + + var state = await harness.First.SyncState.ReadAsync(VaultId, TestContext.Current.CancellationToken); + + if (state.SweepMark is null) + { + break; + } + + // Mid-sweep, nothing has gone yet. + pruned.ShouldBe(0); + (await harness.First.ListAsync()).Items.Count.ShouldBe(6); + } + + pruned.ShouldBe(1); + (await harness.First.ListAsync()).Items.Select(item => item.EntityId).ShouldBe(kept, ignoreOrder: true); + harness.Server.Find(kept[^1]).ShouldNotBeNull(); + } +}