test(localdb): pin the S2 stale-rowid guard (in-txn + mixed multi-row PK-change regression tests)

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/CaptureTriggerTests.cs

@@ -154,6 +154,77 @@ public sealed class CaptureTriggerTests : IDisposable
         Assert.Equal(0, v.Tomb);
     }
 
+    // Load-bearing for the S2 stale-rowid guard: on a pinned transaction connection,
+    // last_insert_rowid() is a LIVE oplog seq when the non-PK update's tombstone insert does not
+    // fire; without the AND(pk-changed) guard, S2 would tombstone an unrelated row's row_version.
+    [Fact]
+    public async Task Update_NonPkUpdate_InSameTransaction_DoesNotCorruptOtherRowVersions()
+    {
+        using var db = await NewOrdersDb();
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'B', 1)");
+
+        await using (var tx = await db.BeginTransactionAsync())
+        {
+            await tx.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'A', 2)");
+            await tx.ExecuteAsync("UPDATE orders SET qty = 9 WHERE id = 1");
+            await tx.CommitAsync(default);
+        }
+
+        var oplog = await Oplog(db);
+        Assert.Equal(3, oplog.Count);
+        Assert.All(oplog, o => Assert.Equal(0, o.IsTombstone));
+        var aInsertHlc = oplog.Single(o => o.PkJson == "{\"id\":2}").Hlc;
+
+        var rv = await db.QueryAsync(
+            "SELECT pk_json, hlc, is_tombstone FROM __localdb_row_version WHERE table_name='orders'",
+            r => (PkJson: r.GetString(0), Hlc: r.GetInt64(1), Tomb: r.GetInt64(2)));
+        Assert.Equal(2, rv.Count);
+        Assert.All(rv, v => Assert.Equal(0, v.Tomb));
+
+        var a = rv.Single(v => v.PkJson == "{\"id\":2}");
+        Assert.Equal(aInsertHlc, a.Hlc);
+    }
+
+    // The spectator row (id=3) pins per-row guard re-evaluation: for the unchanged-PK row, the
+    // stale last_insert_rowid() is the spectator's live oplog seq — an unguarded S2 would
+    // tombstone the spectator's row_version and nothing later would repair it.
+    [Fact]
+    public async Task Update_MultiRow_MixedPkChange_PerRowConditionHolds()
+    {
+        using var db = await NewOrdersDb();
+
+        await using (var tx = await db.BeginTransactionAsync())
+        {
+            await tx.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A', 1)");
+            await tx.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'B', 2)");
+            await tx.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (3, 'C', 3)");
+            await tx.ExecuteAsync(
+                "UPDATE orders SET id = CASE WHEN id = 1 THEN 10 ELSE id END, qty = qty + 1 WHERE id IN (1, 2)");
+            await tx.CommitAsync(default);
+        }
+
+        var oplog = await Oplog(db);
+        var tomb = Assert.Single(oplog, o => o.IsTombstone == 1);
+        Assert.Equal("{\"id\":1}", tomb.PkJson);
+
+        var rv = await db.QueryAsync(
+            "SELECT pk_json, hlc, is_tombstone FROM __localdb_row_version WHERE table_name='orders'",
+            r => (PkJson: r.GetString(0), Hlc: r.GetInt64(1), Tomb: r.GetInt64(2)));
+        Assert.Equal(4, rv.Count);
+
+        var oldPk = rv.Single(v => v.PkJson == "{\"id\":1}");
+        Assert.Equal(1, oldPk.Tomb);
+        Assert.Equal(tomb.Hlc, oldPk.Hlc);
+
+        foreach (var pkJson in new[] { "{\"id\":2}", "{\"id\":3}", "{\"id\":10}" })
+        {
+            var v = rv.Single(x => x.PkJson == pkJson);
+            Assert.Equal(0, v.Tomb);
+            var newest = oplog.Where(o => o.PkJson == pkJson).Max(o => o.Hlc);
+            Assert.Equal(newest, v.Hlc);
+        }
+    }
+
     [Fact]
     public async Task MultiRowStatement_OneOplogEntryPerRow_DistinctHlcs()
     {