Stop a finished push deleting the edit made while it was in flight, and pull a newly synced type from the start

- The outbox coalesces a new save into the row a push already read, under a new operation id, but
  Complete, Revise and Park found the row by sequence alone. The acknowledgement of the earlier state
  deleted the row holding the later one and the edit was gone without a word. They now take the
  PendingOperation and act only while the row still holds it; a later edit is rebased onto the version
  just accepted, so it pushes cleanly instead of conflicting with its own earlier save.
- A pull filtered by type moves the cursor past changes of types it did not ask for, so a build that
  added a type (SpelunkKey, ObjectStore) never received the items of that type made before it.
  sync_state.pulled_types records the list a cursor was read under, and a different list starts the
  pull over once. Migration AddSyncStatePulledTypes.

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();
+    }
+
+    /// <summary>
+    /// An answer to an operation that has since been replaced by a newer edit applies to nothing.
+    /// </summary>
+    /// <remarks>
+    /// 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.
+    /// </remarks>
+    [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]

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));
     }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <remarks>
+    /// 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.
+    /// </remarks>
+    [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");
+    }
+
+    /// <summary>
+    /// A cursor read under a shorter list of item types starts the pull over, once.
+    /// </summary>
+    /// <remarks>
+    /// 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.
+    /// </remarks>
+    [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();
+    }
 }