fix(localdb): capture triggers survive consumer upserts (explicit ON CONFLICT upsert form, outer-statement override)

2026-07-18 00:57:27 -04:00
parent 801b042208
commit 154d584190
4 changed files with 88 additions and 31 deletions
@@ -18,6 +18,15 @@ internal static class TriggerSqlGenerator

    private const string TombstoneTail = "1, strftime('%Y-%m-%dT%H:%M:%fZ','now')";

+    // Explicit upsert, NOT `INSERT OR REPLACE`: SQLite replaces a trigger-body statement's
+    // OR-conflict algorithm with the OUTER statement's conflict handling, so under a consumer
+    // `INSERT ... ON CONFLICT DO UPDATE` an OR REPLACE here degraded to a plain INSERT and hit
+    // SQLITE_CONSTRAINT_PRIMARYKEY on the existing row_version row. An explicit ON CONFLICT
+    // clause is not overridden. (The SELECT's WHERE clause also disambiguates the upsert parse.)
+    private const string RowVersionConflictClause =
+        "ON CONFLICT(table_name, pk_json) DO UPDATE SET hlc = excluded.hlc, node_id = excluded.node_id, " +
+        "is_tombstone = excluded.is_tombstone, tombstone_utc = excluded.tombstone_utc";
+
    public static string TriggerName(string table, string suffix) => $"__localdb_{table}_{suffix}";

    /// <summary>DROP IF EXISTS + CREATE for all three triggers, ready to run in one transaction.</summary>
@@ -49,8 +58,9 @@ internal static class TriggerSqlGenerator
                      zb_hlc_next(),
                      (SELECT node_id FROM __localdb_meta WHERE id = 1),
                      1);
-              INSERT OR REPLACE INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc)
-              SELECT table_name, pk_json, hlc, node_id, {TombstoneTail} FROM __localdb_oplog WHERE seq = last_insert_rowid();
+              INSERT INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc)
+              SELECT table_name, pk_json, hlc, node_id, {TombstoneTail} FROM __localdb_oplog WHERE seq = last_insert_rowid()
+              {RowVersionConflictClause};

            """);

@@ -72,8 +82,9 @@ internal static class TriggerSqlGenerator
                  zb_hlc_next(),
                  (SELECT node_id FROM __localdb_meta WHERE id = 1),
                  0);
-          INSERT OR REPLACE INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc)
-          SELECT table_name, pk_json, hlc, node_id, 0, NULL FROM __localdb_oplog WHERE seq = last_insert_rowid();
+          INSERT INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc)
+          SELECT table_name, pk_json, hlc, node_id, 0, NULL FROM __localdb_oplog WHERE seq = last_insert_rowid()
+          {RowVersionConflictClause};

        """;

@@ -92,9 +103,10 @@ internal static class TriggerSqlGenerator
                     (SELECT node_id FROM __localdb_meta WHERE id = 1),
                     1
              WHERE {changed};
-              INSERT OR REPLACE INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc)
+              INSERT INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc)
              SELECT table_name, pk_json, hlc, node_id, {TombstoneTail} FROM __localdb_oplog
-              WHERE seq = last_insert_rowid() AND ({changed});
+              WHERE seq = last_insert_rowid() AND ({changed})
+              {RowVersionConflictClause};

            """;
    }
@@ -245,6 +245,36 @@ public sealed class CaptureTriggerTests : IDisposable
            Assert.Equal(o.Hlc, rv.Single(v => v.PkJson == o.PkJson).Hlc);
    }

+    // Pins the explicit ON CONFLICT upsert form in the row_version capture statements: SQLite
+    // replaces a trigger-body statement's OR REPLACE with the OUTER statement's conflict handling,
+    // so a consumer `INSERT ... ON CONFLICT DO UPDATE` used to turn the au trigger's OR REPLACE
+    // into a plain INSERT -> SQLITE_CONSTRAINT_PRIMARYKEY (1555) on the existing row_version row.
+    [Fact]
+    public async Task Upsert_OnConflictDoUpdate_BothBranches_Capture()
+    {
+        using var db = await NewOrdersDb();
+        const string upsert =
+            "INSERT INTO orders (id, sku, qty) VALUES (7, @sku, @qty) " +
+            "ON CONFLICT(id) DO UPDATE SET sku = excluded.sku, qty = excluded.qty";
+
+        await db.ExecuteAsync(upsert, new { sku = "FIRST", qty = 1 });  // insert branch -> ai
+        await db.ExecuteAsync(upsert, new { sku = "SECOND", qty = 2 }); // update branch -> au
+
+        var oplog = await Oplog(db);
+        Assert.Equal(2, oplog.Count);
+        Assert.Equal("{\"id\":7,\"sku\":\"FIRST\",\"qty\":1}", oplog[0].RowJson);
+        Assert.Equal("{\"id\":7,\"sku\":\"SECOND\",\"qty\":2}", oplog[1].RowJson);
+        Assert.All(oplog, o => Assert.Equal(0, o.IsTombstone));
+
+        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)));
+        var v = Assert.Single(rv);
+        Assert.Equal("{\"id\":7}", v.PkJson);
+        Assert.Equal(0, v.Tomb);
+        Assert.Equal(oplog[1].Hlc, v.Hlc);
+    }
+
    [Fact]
    public async Task Capture_CoexistsWithConsumerTrigger()
    {
@@ -141,28 +141,44 @@ public sealed class BidirectionalConvergenceTests
        Assert.Equal(21, await ConvergenceFixture.CountOrdersAsync(fx.B));
    }

-    // TODO(product bug): `INSERT ... ON CONFLICT(id) DO UPDATE` on a replicated table crashes the
-    // AFTER-UPDATE capture trigger `__localdb_<table>_au` with
-    //   SQLite Error 19 (1555 SQLITE_CONSTRAINT_PRIMARYKEY):
-    //   'UNIQUE constraint failed: __localdb_row_version.table_name, __localdb_row_version.pk_json'.
-    // Minimal repro: on a fresh replicated `orders` table, upsert the SAME pk twice via ON CONFLICT
-    // DO UPDATE (first = insert, second = the DO-UPDATE branch) — no replication, no concurrency.
-    // A plain `UPDATE orders SET ...` firing the same au trigger does NOT fail; the defect is
-    // specific to the upsert form perturbing last_insert_rowid() inside the au trigger's
-    // `INSERT OR REPLACE INTO __localdb_row_version ... WHERE seq = last_insert_rowid()` capture.
-    // The convergence tests intentionally avoid the upsert form (see ConvergenceFixture.UpsertAsync).
-    // Un-skip once the capture trigger supports UPSERT.
-    [Fact(Skip = "Product bug: ON CONFLICT DO UPDATE crashes the au capture trigger (UNIQUE on __localdb_row_version). See TODO above.")]
-    public async Task UpsertOnConflictDoUpdate_CrashesCaptureTrigger_ProductBug()
+    // Regression for the former product bug where a consumer `INSERT ... ON CONFLICT(id) DO UPDATE`
+    // crashed the au capture trigger: SQLite replaces a trigger-body statement's OR-conflict
+    // algorithm with the OUTER statement's conflict handling, so the trigger's row_version
+    // `INSERT OR REPLACE` degraded to a plain INSERT -> SQLITE_CONSTRAINT_PRIMARYKEY (1555).
+    // Fixed by an explicit ON CONFLICT DO UPDATE clause in the capture triggers (not overridden).
+    [Fact]
+    public async Task Upsert_OnConflictDoUpdate_CapturesCorrectly()
    {
        await using var fx = new ConvergenceFixture();
-        await fx.A.ExecuteAsync(
-            "INSERT INTO orders (id, sku, qty) VALUES (7, 'FIRST', 1) " +
-            "ON CONFLICT(id) DO UPDATE SET sku = excluded.sku, qty = excluded.qty");
-        // Second upsert takes the DO UPDATE branch and throws the UNIQUE-constraint SqliteException.
-        await fx.A.ExecuteAsync(
-            "INSERT INTO orders (id, sku, qty) VALUES (7, 'SECOND', 2) " +
-            "ON CONFLICT(id) DO UPDATE SET sku = excluded.sku, qty = excluded.qty");
+        const string upsert =
+            "INSERT INTO orders (id, sku, qty) VALUES (7, @sku, @qty) " +
+            "ON CONFLICT(id) DO UPDATE SET sku = excluded.sku, qty = excluded.qty";
+
+        await fx.A.ExecuteAsync(upsert, new { sku = "FIRST", qty = 1 });  // insert branch -> ai
+        await fx.A.ExecuteAsync(upsert, new { sku = "SECOND", qty = 2 }); // DO-UPDATE branch -> au
+
+        // Both branches captured, in order, with full-row payloads.
+        var oplog = await fx.A.QueryAsync(
+            "SELECT row_json, hlc, is_tombstone FROM __localdb_oplog WHERE table_name='orders' ORDER BY seq",
+            static r => (RowJson: r.GetString(0), Hlc: r.GetInt64(1), Tomb: r.GetInt64(2)));
+        Assert.Equal(2, oplog.Count);
+        Assert.Equal("{\"id\":7,\"sku\":\"FIRST\",\"qty\":1}", oplog[0].RowJson);
+        Assert.Equal("{\"id\":7,\"sku\":\"SECOND\",\"qty\":2}", oplog[1].RowJson);
+        Assert.All(oplog, o => Assert.Equal(0, o.Tomb));
+
+        // Single row_version entry for the pk, correlated to the LATEST (au-branch) oplog hlc.
+        var rv = await fx.A.QueryAsync(
+            "SELECT pk_json, hlc, is_tombstone FROM __localdb_row_version WHERE table_name='orders'",
+            static r => (PkJson: r.GetString(0), Hlc: r.GetInt64(1), Tomb: r.GetInt64(2)));
+        var v = Assert.Single(rv);
+        Assert.Equal("{\"id\":7}", v.PkJson);
+        Assert.Equal(0, v.Tomb);
+        Assert.Equal(oplog[1].Hlc, v.Hlc);
+
+        // And the captured upsert replicates end-to-end.
+        await fx.StartAsync();
+        await fx.AssertConvergedAsync();
+        Assert.Equal((7L, "SECOND", (long?)2), await ReadRow(fx.B, 7));
    }

    private static async Task<(long Id, string? Sku, long? Qty)> ReadRow(ILocalDb db, long id)
@@ -190,11 +190,10 @@ public sealed class ConvergenceFixture : IAsyncDisposable

    /// <summary>
    /// Inserts or updates one row through the normal consumer API (fires the ai/au capture triggers).
-    /// Uses a plain UPDATE-else-INSERT inside one transaction — NOT <c>INSERT ... ON CONFLICT DO
-    /// UPDATE</c>: that upsert form crashes the AFTER-UPDATE capture trigger with
-    /// "UNIQUE constraint failed: __localdb_row_version" (see the Skipped regression test in
-    /// BidirectionalConvergenceTests). The transaction makes the update-or-insert atomic against a
-    /// concurrent replicated apply of the same pk (no INSERT/UNIQUE race on orders).
+    /// Uses a plain UPDATE-else-INSERT inside one transaction; the transaction makes the
+    /// update-or-insert atomic against a concurrent replicated apply of the same pk (no
+    /// INSERT/UNIQUE race on orders). <c>INSERT ... ON CONFLICT DO UPDATE</c> is also supported by
+    /// the capture triggers — covered by <c>Upsert_OnConflictDoUpdate_CapturesCorrectly</c>.
    /// </summary>
    public static async Task UpsertAsync(ILocalDb db, long id, string sku, long qty)
    {