test(localdb): convergence polish — provable LWW race arbitration, collection serialization, dead-letter diagnostics

Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/BidirectionalConvergenceTests.cs

@@ -5,6 +5,7 @@ namespace ZB.MOM.WW.LocalDb.Tests.Convergence;
 /// concurrent writes, an update/delete race, offline accumulation, restart-mid-stream, and a pruned-horizon
 /// snapshot resync — each driven through the normal <c>ILocalDb</c> API while both sync sessions run.
 /// </summary>
+[Collection("Convergence")]
 public sealed class BidirectionalConvergenceTests
 {
     [Fact]
@@ -38,26 +39,52 @@ public sealed class BidirectionalConvergenceTests
         Assert.Equal(51, await ConvergenceFixture.CountOrdersAsync(fx.B));
     }
 
+    // Update/delete race, serialized so the winner is PROVABLY HLC-arbitrated (same technique as
+    // ConcurrentWrites' shared key): each contender writes only after observing the other's
+    // replicated stamp, so its HLC is strictly higher and the expected LWW winner is deterministic —
+    // not just "both sides agree" (which would also pass if both agreed for a wrong reason).
     [Fact]
-    public async Task UpdateDeleteRace_Converges()
+    public async Task UpdateDeleteRace_DeleteHasHigherHlc_DeleteWinsBothSides()
     {
         await using var fx = new ConvergenceFixture();
-
-        // Seed the contested key on both nodes and let it converge first.
         await ConvergenceFixture.UpsertAsync(fx.A, 7, "SEED", 1);
         await fx.StartAsync();
         await fx.AssertConvergedAsync();
 
-        // Race: A updates the same pk while B deletes it, inside one window. Whichever HLC wins, both
+        // A updates; converge (B's clock has observed A's stamp); then B deletes -> provably higher HLC.
-        // nodes must end identical (either the updated row on both, or absent on both).
         await ConvergenceFixture.UpsertAsync(fx.A, 7, "A_UPDATED", 99);
+        await fx.AssertConvergedAsync();
         await ConvergenceFixture.DeleteAsync(fx.B, 7);
 
         await fx.AssertConvergedAsync();
 
-        var a = await ConvergenceFixture.CountOrdersAsync(fx.A);
+        // The DELETE won on BOTH sides: row absent, row_version tombstoned.
-        var b = await ConvergenceFixture.CountOrdersAsync(fx.B);
+        Assert.Equal(0, await ConvergenceFixture.CountOrdersAsync(fx.A));
-        Assert.Equal(a, b); // identical outcome; AssertConverged already proved row-level identity
+        Assert.Equal(0, await ConvergenceFixture.CountOrdersAsync(fx.B));
+        Assert.True(await IsTombstoned(fx.A, 7), "A row_version for id=7 should be a tombstone");
+        Assert.True(await IsTombstoned(fx.B, 7), "B row_version for id=7 should be a tombstone");
+    }
+
+    [Fact]
+    public async Task UpdateDeleteRace_UpdateHasHigherHlc_UpdateWinsBothSides()
+    {
+        await using var fx = new ConvergenceFixture();
+        await ConvergenceFixture.UpsertAsync(fx.A, 7, "SEED", 1);
+        await fx.StartAsync();
+        await fx.AssertConvergedAsync();
+
+        // Mirrored: B deletes; converge (A's clock has observed the tombstone's stamp); then A
+        // re-writes the pk -> provably higher HLC -> the update resurrects the row on BOTH sides.
+        await ConvergenceFixture.DeleteAsync(fx.B, 7);
+        await fx.AssertConvergedAsync();
+        await ConvergenceFixture.UpsertAsync(fx.A, 7, "A_RESURRECTED", 99);
+
+        await fx.AssertConvergedAsync();
+
+        Assert.Equal((7L, "A_RESURRECTED", (long?)99), await ReadRow(fx.A, 7));
+        Assert.Equal((7L, "A_RESURRECTED", (long?)99), await ReadRow(fx.B, 7));
+        Assert.False(await IsTombstoned(fx.A, 7), "A row_version for id=7 should be live again");
+        Assert.False(await IsTombstoned(fx.B, 7), "B row_version for id=7 should be live again");
     }
 
     [Fact]
@@ -181,6 +208,14 @@ public sealed class BidirectionalConvergenceTests
         Assert.Equal((7L, "SECOND", (long?)2), await ReadRow(fx.B, 7));
     }
 
+    private static async Task<bool> IsTombstoned(ILocalDb db, long id)
+    {
+        var rows = await db.QueryAsync(
+            "SELECT is_tombstone FROM __localdb_row_version WHERE table_name = 'orders' AND pk_json = @pk",
+            static r => r.GetInt64(0), new { pk = $"{{\"id\":{id}}}" });
+        return Assert.Single(rows) != 0;
+    }
+
     private static async Task<(long Id, string? Sku, long? Qty)> ReadRow(ILocalDb db, long id)
     {
         var rows = await db.QueryAsync(

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/ConvergenceFixture.cs

@@ -15,6 +15,15 @@ using ZB.MOM.WW.LocalDb.Replication.Internal;
 
 namespace ZB.MOM.WW.LocalDb.Tests.Convergence;
 
+/// <summary>
+/// Serializes the convergence test classes against EACH OTHER (no shared class fixture — each test
+/// still builds its own <see cref="ConvergenceFixture"/>): both classes stand up real Kestrel
+/// listeners plus four SQLite files per test, and running the two heavy classes concurrently under
+/// CI contention is a flakiness risk. The rest of the assembly still parallelizes normally.
+/// </summary>
+[CollectionDefinition("Convergence")]
+public sealed class ConvergenceCollection;
+
 /// <summary>
 /// Two FULL replication stacks over a REAL loopback gRPC transport (Kestrel h2c on 127.0.0.1),
 /// proving end-to-end bidirectional convergence. Node A is the initiator (client:
@@ -183,9 +192,25 @@ public sealed class ConvergenceFixture : IAsyncDisposable
         sb.AppendLine("--- B.orders ---").AppendLine(await DumpOrdersAsync(_dbB));
         sb.AppendLine("--- A.row_version ---").AppendLine(await DumpRowVersionAsync(_dbA));
         sb.AppendLine("--- B.row_version ---").AppendLine(await DumpRowVersionAsync(_dbB));
+        sb.AppendLine("--- A.dead_letter ---").AppendLine(await DumpDeadLettersAsync(_dbA));
+        sb.AppendLine("--- B.dead_letter ---").AppendLine(await DumpDeadLettersAsync(_dbB));
         return sb.ToString();
     }
 
+    // Dead-lettered entries are a prime suspect when convergence stalls (a poison row silently
+    // diverts instead of applying), so the timeout dump surfaces the count + a small sample.
+    private static async Task<string> DumpDeadLettersAsync(ILocalDb db)
+    {
+        var count = (await db.QueryAsync(
+            "SELECT COUNT(*) FROM __localdb_dead_letter", static r => r.GetInt64(0)))[0];
+        if (count == 0)
+            return "count: 0";
+        var sample = await db.QueryAsync(
+            "SELECT table_name, pk_json, hlc, node_id, error FROM __localdb_dead_letter ORDER BY id LIMIT 5",
+            static r => $"{r.GetString(0)}|{r.GetString(1)}|{r.GetInt64(2)}|{r.GetString(3)}|{r.GetString(4)}");
+        return $"count: {count} (first {sample.Count})\n" + string.Join("\n", sample);
+    }
+
     // ---- test helpers -----------------------------------------------------------------------
 
     /// <summary>

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/RandomOpsConvergenceTests.cs

@@ -6,6 +6,7 @@ namespace ZB.MOM.WW.LocalDb.Tests.Convergence;
 /// must always converge to byte-identical state. Seeded Random only (no time-based seeds) so every
 /// failure is reproducible from the logged seed.
 /// </summary>
+[Collection("Convergence")]
 public sealed class RandomOpsConvergenceTests
 {
     private const int OpsPerNode = 500;