diff --git a/src/DodoSSH.Client.Storage/CacheRows.cs b/src/DodoSSH.Client.Storage/CacheRows.cs index dc94438..1ce2643 100644 --- a/src/DodoSSH.Client.Storage/CacheRows.cs +++ b/src/DodoSSH.Client.Storage/CacheRows.cs @@ -350,6 +350,17 @@ internal sealed class SyncStateRow /// decide which edit wins. /// public long ServerTimeSkewMs { get; set; } + + /// + /// The item types was read under, as this build names them; null before this was kept. + /// + /// + /// A pull asks for the types this build can apply, and the cursor it gets back is past every change it + /// read — including the changes of types it did not ask for. So a build that adds a type to that list + /// and carries on from the old cursor never sees the items of that type made before it arrived. Kept + /// beside the cursor so a change of list starts the pull over; see SyncEngine.PullAsync. + /// + public string? PulledTypes { get; set; } } /// Something the merge had to override, or an item that could not be processed. diff --git a/src/DodoSSH.Client.Storage/Migrations/20261002104920_AddSyncStatePulledTypes.Designer.cs b/src/DodoSSH.Client.Storage/Migrations/20261002104920_AddSyncStatePulledTypes.Designer.cs new file mode 100644 index 0000000..007c37e --- /dev/null +++ b/src/DodoSSH.Client.Storage/Migrations/20261002104920_AddSyncStatePulledTypes.Designer.cs @@ -0,0 +1,469 @@ +// +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("20261002104920_AddSyncStatePulledTypes")] + partial class AddSyncStatePulledTypes + { + /// + 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("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.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/20261002104920_AddSyncStatePulledTypes.cs b/src/DodoSSH.Client.Storage/Migrations/20261002104920_AddSyncStatePulledTypes.cs new file mode 100644 index 0000000..cbd4b4b --- /dev/null +++ b/src/DodoSSH.Client.Storage/Migrations/20261002104920_AddSyncStatePulledTypes.cs @@ -0,0 +1,28 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace DodoSSH.Client.Storage.Migrations +{ + /// + public partial class AddSyncStatePulledTypes : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.AddColumn( + name: "pulled_types", + table: "sync_state", + type: "TEXT", + nullable: true); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropColumn( + name: "pulled_types", + table: "sync_state"); + } + } +} diff --git a/src/DodoSSH.Client.Storage/Migrations/ClientCacheContextModelSnapshot.cs b/src/DodoSSH.Client.Storage/Migrations/ClientCacheContextModelSnapshot.cs index 3573005..60cb699 100644 --- a/src/DodoSSH.Client.Storage/Migrations/ClientCacheContextModelSnapshot.cs +++ b/src/DodoSSH.Client.Storage/Migrations/ClientCacheContextModelSnapshot.cs @@ -362,6 +362,10 @@ namespace DodoSSH.Client.Storage.Migrations .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"); diff --git a/src/DodoSSH.Client.Storage/OutboxStore.cs b/src/DodoSSH.Client.Storage/OutboxStore.cs index 65253fa..e8abc18 100644 --- a/src/DodoSSH.Client.Storage/OutboxStore.cs +++ b/src/DodoSSH.Client.Storage/OutboxStore.cs @@ -44,6 +44,14 @@ public sealed record QueuedChange( /// ancestor that is this client's own earlier edit. That merge finds no disagreement, so it converges /// on the newest state with nothing for the user to arbitrate. /// +/// +/// ◆ Which is only true if an answer is applied to the operation it answers. A push reads a row, +/// sends it, and acts on the reply some time later; a user who saves in between coalesces a new state into +/// that same row. So , and +/// take the operation that was sent and do nothing when the row has since become a different one. Keyed by +/// sequence alone, an acknowledgement of the earlier state deleted the row holding the later one, and the +/// edit was gone with nothing to say so. +/// /// public sealed class OutboxStore( IDbContextFactory contexts, @@ -116,7 +124,7 @@ public sealed class OutboxStore( /// /// public async Task ReviseAsync( - long sequence, + PendingOperation revising, SyncOperation operation, int? expectedVersion, EncryptedPayload? payload, @@ -127,12 +135,18 @@ public sealed class OutboxStore( var context = contexts.CreateDbContext(); await using var scope = context.ConfigureAwait(false); + ArgumentNullException.ThrowIfNull(revising); + var row = await context.Set() - .SingleOrDefaultAsync(r => r.Sequence == sequence, cancellationToken) + .SingleOrDefaultAsync( + r => r.Sequence == revising.Sequence && r.OperationId == revising.OperationId, + cancellationToken) .ConfigureAwait(false); if (row is null) { + // Completed, or replaced by a newer edit since it was read. Either way the merge was of a state + // that is no longer the one queued, and the next push will meet the server with the current one. return null; } @@ -233,16 +247,19 @@ public sealed class OutboxStore( /// Records that an operation has been sent, so a repeated failure can be noticed. public Task MarkDispatchedAsync(long sequence, CancellationToken cancellationToken) => - UpdateAsync(sequence, row => row.Attempts++, cancellationToken); + UpdateAsync(sequence, operationId: null, row => row.Attempts++, cancellationToken); - /// Removes an operation the server accepted. - public async Task CompleteAsync(long sequence, CancellationToken cancellationToken) + /// Removes an operation the server accepted, unless a newer edit has replaced it since. + /// Whether a row was removed. + public async Task CompleteAsync(PendingOperation operation, CancellationToken cancellationToken) { + ArgumentNullException.ThrowIfNull(operation); + var context = contexts.CreateDbContext(); await using var scope = context.ConfigureAwait(false); var removed = await context.Set() - .Where(r => r.Sequence == sequence) + .Where(r => r.Sequence == operation.Sequence && r.OperationId == operation.OperationId) .ExecuteDeleteAsync(cancellationToken) .ConfigureAwait(false); @@ -257,9 +274,10 @@ public sealed class OutboxStore( /// invalid, or the caller no longer having permission. Retrying either would spin forever and, far /// worse, would block every change queued behind it in a vault the user can still write to. /// - public Task ParkAsync(long sequence, string reason, CancellationToken cancellationToken) => + public Task ParkAsync(PendingOperation operation, string reason, CancellationToken cancellationToken) => UpdateAsync( - sequence, + operation.Sequence, + operation.OperationId, row => { row.IsParked = true; @@ -269,7 +287,7 @@ public sealed class OutboxStore( /// Records a transient failure without parking the operation. public Task RecordFailureAsync(long sequence, string reason, CancellationToken cancellationToken) => - UpdateAsync(sequence, row => row.LastError = reason, cancellationToken); + UpdateAsync(sequence, operationId: null, row => row.LastError = reason, cancellationToken); private static Task FindRowAsync( ClientCacheContext context, @@ -302,6 +320,7 @@ public sealed class OutboxStore( private async Task UpdateAsync( long sequence, + Guid? operationId, Action mutate, CancellationToken cancellationToken) { @@ -309,7 +328,9 @@ public sealed class OutboxStore( await using var scope = context.ConfigureAwait(false); var row = await context.Set() - .SingleOrDefaultAsync(r => r.Sequence == sequence, cancellationToken) + .SingleOrDefaultAsync( + r => r.Sequence == sequence && (operationId == null || r.OperationId == operationId), + cancellationToken) .ConfigureAwait(false); if (row is null) diff --git a/src/DodoSSH.Client.Storage/StoredTypes.cs b/src/DodoSSH.Client.Storage/StoredTypes.cs index 0777707..08b7a8f 100644 --- a/src/DodoSSH.Client.Storage/StoredTypes.cs +++ b/src/DodoSSH.Client.Storage/StoredTypes.cs @@ -183,13 +183,15 @@ public sealed record PendingOperation( /// When the last pull completed. /// When the last push completed. /// Observed clock difference, recorded and never acted on. +/// The item types the cursor was read under; see SyncStateRow.PulledTypes. public sealed record StoredSyncState( Guid VaultId, string? Cursor, uint KeyGeneration, DateTimeOffset? LastPulledAt = null, DateTimeOffset? LastPushedAt = null, - long ServerTimeSkewMs = 0); + long ServerTimeSkewMs = 0, + string? PulledTypes = null); /// 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 9238644..6e0af23 100644 --- a/src/DodoSSH.Client.Storage/SyncStateStore.cs +++ b/src/DodoSSH.Client.Storage/SyncStateStore.cs @@ -46,7 +46,8 @@ public sealed class SyncStateStore(IDbContextFactory context row.KeyGeneration, row.LastPulledAtUtc, row.LastPushedAtUtc, - row.ServerTimeSkewMs); + row.ServerTimeSkewMs, + row.PulledTypes); } /// Records a vault's position. @@ -72,6 +73,7 @@ public sealed class SyncStateStore(IDbContextFactory context row.LastPulledAtUtc = state.LastPulledAt; row.LastPushedAtUtc = state.LastPushedAt; row.ServerTimeSkewMs = state.ServerTimeSkewMs; + row.PulledTypes = state.PulledTypes; await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); } diff --git a/src/DodoSSH.Client.Sync/ItemReconciler.cs b/src/DodoSSH.Client.Sync/ItemReconciler.cs index b0e3f63..bddd5e9 100644 --- a/src/DodoSSH.Client.Sync/ItemReconciler.cs +++ b/src/DodoSSH.Client.Sync/ItemReconciler.cs @@ -120,7 +120,7 @@ internal sealed class ItemReconciler( { return pending.Operation == SyncOperation.Delete // Both sides deleted it. Nothing to arbitrate and nothing to tell the user. - ? outbox.CompleteAsync(pending.Sequence, cancellationToken) + ? outbox.CompleteAsync(pending, cancellationToken) : ResurrectAsync(vaultId, remote, pending, report, cancellationToken); } @@ -166,7 +166,7 @@ internal sealed class ItemReconciler( } await outbox.ReviseAsync( - pending.Sequence, + pending, SyncOperation.Upsert, expectedVersion: null, kind.Seal(local.Secret, vaultKey.Span, pending.EntityId, generation, itemVersion: 1), @@ -212,7 +212,7 @@ internal sealed class ItemReconciler( { // Field for field the same item: this is our own create coming back. Nothing to do but stop // trying to send it again. - await outbox.CompleteAsync(pending.Sequence, cancellationToken).ConfigureAwait(false); + await outbox.CompleteAsync(pending, cancellationToken).ConfigureAwait(false); return; } @@ -362,7 +362,7 @@ internal sealed class ItemReconciler( cancellationToken).ConfigureAwait(false); // Now the tombstone can stand. - await outbox.CompleteAsync(pending.Sequence, cancellationToken).ConfigureAwait(false); + await outbox.CompleteAsync(pending, cancellationToken).ConfigureAwait(false); await conflicts.RecordAsync( vaultId, @@ -385,7 +385,7 @@ internal sealed class ItemReconciler( SyncReportBuilder report, CancellationToken cancellationToken) { - await outbox.CompleteAsync(pending.Sequence, cancellationToken).ConfigureAwait(false); + await outbox.CompleteAsync(pending, cancellationToken).ConfigureAwait(false); await conflicts.RecordAsync( vaultId, @@ -413,7 +413,7 @@ internal sealed class ItemReconciler( var nextVersion = SyncVersions.NextVersion(remote.Version); await outbox.ReviseAsync( - pending.Sequence, + pending, SyncOperation.Upsert, expectedVersion: remote.Version, kind.Seal(secret, vaultKey.Span, remote.EntityId, generation, nextVersion), @@ -486,7 +486,7 @@ internal sealed class ItemReconciler( CancellationToken cancellationToken) { await outbox.ParkAsync( - pending.Sequence, + pending, $"Written by a newer version of DodoSSH; update before editing this {kind.Noun}.", cancellationToken).ConfigureAwait(false); @@ -512,7 +512,7 @@ internal sealed class ItemReconciler( CancellationToken cancellationToken) { await outbox.ParkAsync( - pending.Sequence, + pending, $"The local or the server copy of this {kind.Noun} could not be decrypted.", cancellationToken).ConfigureAwait(false); diff --git a/src/DodoSSH.Client.Sync/SyncEngine.cs b/src/DodoSSH.Client.Sync/SyncEngine.cs index 492c8b6..281d9c3 100644 --- a/src/DodoSSH.Client.Sync/SyncEngine.cs +++ b/src/DodoSSH.Client.Sync/SyncEngine.cs @@ -127,7 +127,7 @@ public sealed class SyncEngine SyncReportBuilder report, CancellationToken cancellationToken) { - var state = await syncState.ReadAsync(vaultId, cancellationToken).ConfigureAwait(false); + var state = await ReadStateAsync(vaultId, report, cancellationToken).ConfigureAwait(false); // At most one restart per pull. A server that rejects the cursor it has just issued is not // telling this client anything it can act on, and replaying the whole log against it would turn @@ -245,6 +245,26 @@ public sealed class SyncEngine /// 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, @@ -282,9 +302,14 @@ public sealed class SyncEngine KeyGeneration = response.CurrentKeyGeneration, LastPulledAt = now, ServerTimeSkewMs = skew, + PulledTypes = PulledTypes, }; } + /// The types this build pulls, in a form a stored state can be compared against. + private static string PulledTypes { get; } = + string.Join(',', ItemKinds.SyncedTypes.Select(type => (int)type).Order()); + private async Task ApplyAsync( Guid vaultId, SyncChange change, @@ -446,7 +471,7 @@ public sealed class SyncEngine } await outbox.ReviseAsync( - operation.Sequence, + operation, operation.Operation, operation.ExpectedVersion, resealed, @@ -551,7 +576,76 @@ public sealed class SyncEngine cancellationToken).ConfigureAwait(false); } - await outbox.CompleteAsync(operation.Sequence, cancellationToken).ConfigureAwait(false); + var completed = await outbox.CompleteAsync(operation, cancellationToken).ConfigureAwait(false); + + if (!completed && canMirror && !isDelete) + { + await RebaseOntoAcceptedAsync(vaultId, operation, version, cancellationToken).ConfigureAwait(false); + } + } + + /// + /// Moves an edit made while was in flight onto the version that push produced. + /// + /// + /// + /// The user saved again between the push being read and its answer arriving, so the row now holds a + /// later state branched from the same base as the one the server just took. Left alone it is still the + /// user's edit and still goes — but it goes expecting the old version, comes back Conflict, and is + /// merged against an ancestor that predates the save the server already has, which makes every field + /// changed twice look like a disagreement with somebody else. Its true parent is the state just accepted, + /// so that becomes its ancestor and expected version, and its payload is re-sealed at the version it + /// will now produce, because the AAD binds that. + /// + /// + /// Only for a row still branched from the same base. One that is not has been revised by something else + /// since, and the ordinary conflict path is the one that knows what to do with it; so is a row whose + /// payload this session cannot re-seal. + /// + /// + private async Task RebaseOntoAcceptedAsync( + Guid vaultId, + PendingOperation accepted, + int version, + CancellationToken cancellationToken) + { + var later = await outbox + .FindAsync(vaultId, accepted.EntityType, accepted.EntityId, cancellationToken) + .ConfigureAwait(false); + + if (later is null || later.ExpectedVersion != accepted.ExpectedVersion || accepted.Payload is null) + { + return; + } + + EncryptedPayload? payload = null; + + if (later.Payload is { } queued) + { + payload = PayloadReseal.TryReseal( + keyring, + vaultId, + later.EntityType, + later.EntityId, + queued, + openAtVersion: SyncVersions.NextVersion(later.ExpectedVersion), + sealAtVersion: SyncVersions.NextVersion(version)); + + if (payload is null) + { + return; + } + } + + await outbox.ReviseAsync( + later, + later.Operation, + version, + payload, + later.Fields, + new StoredAncestor(version, accepted.Payload, accepted.Fields), + cancellationToken) + .ConfigureAwait(false); } private async Task ResolveAsync( @@ -617,7 +711,7 @@ public sealed class SyncEngine if (operation.Operation == SyncOperation.Delete) { // Nothing there to delete, so the intent is already satisfied. - await outbox.CompleteAsync(operation.Sequence, cancellationToken).ConfigureAwait(false); + await outbox.CompleteAsync(operation, cancellationToken).ConfigureAwait(false); return false; } @@ -645,7 +739,7 @@ public sealed class SyncEngine SyncReportBuilder report, CancellationToken cancellationToken) { - await outbox.ParkAsync(operation.Sequence, reason, cancellationToken).ConfigureAwait(false); + await outbox.ParkAsync(operation, reason, cancellationToken).ConfigureAwait(false); await conflicts.RecordAsync( vaultId, diff --git a/src/DodoSSH.Client.Sync/VaultItemRepository.cs b/src/DodoSSH.Client.Sync/VaultItemRepository.cs index bc0936b..34e3392 100644 --- a/src/DodoSSH.Client.Sync/VaultItemRepository.cs +++ b/src/DodoSSH.Client.Sync/VaultItemRepository.cs @@ -327,7 +327,7 @@ internal sealed class VaultItemRepository( if (pending is not null && NeverReachedTheServer(pending)) { - await outbox.CompleteAsync(pending.Sequence, cancellationToken).ConfigureAwait(false); + await outbox.CompleteAsync(pending, cancellationToken).ConfigureAwait(false); // Recorded even though nothing goes to the server. Somebody created an item and then removed it, // which is two things they did — and a log that showed only the create would describe a keychain diff --git a/tests/DodoSSH.Client.Storage.Tests/OutboxStoreTests.cs b/tests/DodoSSH.Client.Storage.Tests/OutboxStoreTests.cs index 10c28b9..16b7764 100644 --- a/tests/DodoSSH.Client.Storage.Tests/OutboxStoreTests.cs +++ b/tests/DodoSSH.Client.Storage.Tests/OutboxStoreTests.cs @@ -173,7 +173,7 @@ public sealed class OutboxStoreTests : IAsyncLifetime var merged = new StoredAncestor(4, Payload(seed: 20), new SyncPlaintextFields()); var revised = await harness.Outbox.ReviseAsync( - queued.Sequence, + queued, SyncOperation.Upsert, expectedVersion: 4, Payload(seed: 30), @@ -199,7 +199,7 @@ public sealed class OutboxStoreTests : IAsyncLifetime var queued = await harness.Outbox.QueueAsync(Change(parked), Token); await harness.Outbox.QueueAsync(Change(healthy), Token); - await harness.Outbox.ParkAsync(queued.Sequence, "Entity type not supported.", Token); + await harness.Outbox.ParkAsync(queued, "Entity type not supported.", Token); var pending = await harness.Outbox.TakeAsync(VaultId, 10, Token); pending.ShouldHaveSingleItem().EntityId.ShouldBe(healthy); @@ -216,7 +216,7 @@ public sealed class OutboxStoreTests : IAsyncLifetime var entityId = Guid.CreateVersion7(); var queued = await harness.Outbox.QueueAsync(Change(entityId), Token); - await harness.Outbox.ParkAsync(queued.Sequence, "nope", Token); + await harness.Outbox.ParkAsync(queued, "nope", Token); var requeued = await harness.Outbox.QueueAsync(Change(entityId, seed: 5), Token); @@ -230,11 +230,42 @@ public sealed class OutboxStoreTests : IAsyncLifetime { var queued = await harness.Outbox.QueueAsync(Change(Guid.CreateVersion7()), Token); - (await harness.Outbox.CompleteAsync(queued.Sequence, Token)).ShouldBeTrue(); + (await harness.Outbox.CompleteAsync(queued, Token)).ShouldBeTrue(); (await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldBeEmpty(); // Idempotent: a drain that retries after a crash must not fail on an already-cleared row. - (await harness.Outbox.CompleteAsync(queued.Sequence, Token)).ShouldBeFalse(); + (await harness.Outbox.CompleteAsync(queued, Token)).ShouldBeFalse(); + } + + /// + /// An answer to an operation that has since been replaced by a newer edit applies to nothing. + /// + /// + /// The newer edit coalesced into the same row, so the sequence still matches; only the operation id says + /// the row is no longer the one that was sent. Completing, revising or parking it would act on the newer + /// edit on the strength of an answer about the older one. + /// + [Fact] + public async Task AnAnswerToAReplacedOperation_LeavesTheNewerEditAlone() + { + var entityId = Guid.CreateVersion7(); + + var sent = await harness.Outbox.QueueAsync(Change(entityId, seed: 1), Token); + var newer = await harness.Outbox.QueueAsync(Change(entityId, seed: 2), Token); + + newer.Sequence.ShouldBe(sent.Sequence, "the newer edit coalesces into the same row"); + + (await harness.Outbox.CompleteAsync(sent, Token)).ShouldBeFalse(); + await harness.Outbox.ParkAsync(sent, "refused", Token); + + (await harness.Outbox.ReviseAsync( + sent, sent.Operation, sent.ExpectedVersion, sent.Payload, sent.Fields, sent.Ancestor, Token)) + .ShouldBeNull(); + + var left = (await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldHaveSingleItem(); + + left.OperationId.ShouldBe(newer.OperationId); + left.IsParked.ShouldBeFalse(); } [Fact] diff --git a/tests/DodoSSH.Client.Sync.Tests/SyncEngineTests.cs b/tests/DodoSSH.Client.Sync.Tests/SyncEngineTests.cs index 52bba4a..05c6096 100644 --- a/tests/DodoSSH.Client.Sync.Tests/SyncEngineTests.cs +++ b/tests/DodoSSH.Client.Sync.Tests/SyncEngineTests.cs @@ -231,4 +231,68 @@ public sealed class SyncEngineTests async () => await harness.First.Hosts.ListAsync( unknown, TestContext.Current.CancellationToken)); } + + /// + /// A save made while the previous save of the same host is on its way to the server survives that push + /// being acknowledged, and reaches the server as the newer state rather than as a conflict. + /// + /// + /// The outbox coalesces the second save into the row the push read. Acknowledging by sequence alone + /// deleted that row — the newer state with it — and the server and both machines kept the older one. + /// + [Fact] + public async Task ASaveMadeWhileAPushIsInFlight_IsNotLostWhenThatPushIsAcknowledged() + { + using var harness = await CreateAsync(); + + var id = await harness.First.CreateAsync(Host("prod-db", hostname: "first.internal")); + await harness.First.SyncAsync(); + + await harness.First.UpdateAsync(id, Host("prod-db", hostname: "second.internal")); + + harness.Server.OnPush = () => harness.First + .UpdateAsync(id, Host("prod-db", hostname: "third.internal")) + .GetAwaiter() + .GetResult(); + + await harness.First.SyncAsync(); + await harness.First.SyncAsync(); + + (await harness.First.FindAsync(id)).Secret.Hostname.ShouldBe("third.internal"); + (await harness.First.ConflictsAsync()).ShouldBeEmpty(); + + await harness.Second.SyncAsync(); + + (await harness.Second.FindAsync(id)).Secret.Hostname.ShouldBe("third.internal"); + } + + /// + /// A cursor read under a shorter list of item types starts the pull over, once. + /// + /// + /// The cursor a pull returns is past every change it read, including the changes of types it did not ask + /// for. A build that adds a type to the list — SpelunkKey did, ObjectStore before it — and carried on from + /// that cursor would never receive the items of the new type made before it arrived. + /// + [Fact] + public async Task ACursorReadUnderAnotherListOfTypes_StartsThePullOverOnce() + { + using var harness = await CreateAsync(); + + await harness.First.CreateAsync(Host("prod-db")); + await harness.First.SyncAsync(); + + var state = await harness.First.SyncState.ReadAsync(VaultId, TestContext.Current.CancellationToken); + state.Cursor.ShouldNotBeNull(); + state.PulledTypes.ShouldNotBeNull(); + + // What an older build, which pulled fewer types, would have left behind. + await harness.First.SyncState.SaveAsync( + state with { PulledTypes = "1" }, TestContext.Current.CancellationToken); + + (await harness.First.SyncAsync()).ResyncedFromStart.ShouldBeTrue(); + (await harness.First.SyncAsync()).ResyncedFromStart.ShouldBeFalse("the list is recorded with the cursor"); + + (await harness.First.ListAsync()).Items.ShouldHaveSingleItem(); + } }