fix(sqlite): backfill _sqlite_rowid slot on pre-existing rel tables

sqlite/docs/cdc_design.md

@@ -38,9 +38,10 @@ The WAL frame index is the LSN:
   with nonzero commit size is a **commit boundary**. Only frames up to the
   last commit frame are visible; the uncommitted tail is never applied.
 - The replication cursor is `(walSalt1, walSalt2, appliedFrames)`,
-  persisted in the `__sqlite_cdc_cursor` node table. It is written **last**
-  (at-least-once semantics); replays are safe because applies are
-  idempotent and the cursor only moves forward.
+  persisted in the `__sqlite_cdc_cursor` node table. Each SQLite WAL commit
+  maps to one Ladybug transaction (batch statements + cursor advance commit
+  atomically), so a mid-tick failure leaves no partial rows and the next
+  tick replays whole batches idempotently; the cursor only moves forward.
 - Crash recovery = resume from the cursor. Uncommitted tail frames are
   re-read, never double-applied. Multi-commit batches apply atomically
   per commit for correctness, but converge identically when coalesced:
@@ -68,7 +69,8 @@ extension/sqlite/src/function/
   sync_engine     One tick: read WAL + DB bytes, load cursor, schemas, and
                   explicit mappings; tail/resolve via cdc/; execute through
                   a dedicated inner ClientContext (reusing the outer CALL's
-                  context self-deadlocks on its mutex); cursor last.
+                  context self-deadlocks on its mutex). Each WAL commit
+                  batch applies in one transaction (statements + cursor).
                   Resolver memory persists across CALLs (process-lifetime,
                   keyed by path; a zero cursor resets it).
   sync_functions  CALL sqlite_sync / sqlite_register_node /

sqlite/src/function/sync_engine.cpp

@@ -724,6 +724,19 @@ SyncResult SyncEngine::sync(main::ClientContext* context, const std::string& dbP
             if (!execOk(inner, cdc::buildRelDdl(entry.target), out.error)) {
                 return out;
             }
+            // Rowid slot for replace-on-reuse upserts (see EDGE_RID_PROP):
+            // tables the sync creates carry it from buildRelDdl, but
+            // pre-existing native tables don't. Backfill it idempotently so
+            // the slot-clear DELETE and slot-carrying CREATE in the edge
+            // upserts bind on every mapped table. Pre-sync edges read NULL
+            // (correct: they are not sync-owned; endpoint deletes still
+            // cover them).
+            auto addSlot = std::format("ALTER TABLE {} ADD IF NOT EXISTS {} INT64",
+                cdc::quoteCypherIdent(entry.target.mapping->label),
+                cdc::quoteCypherIdent(cdc::EDGE_RID_PROP));
+            if (!execOk(inner, addSlot, out.error)) {
+                return out;
+            }
         }
 
         // WAL overlay fetcher: newest frame wins, then the DB file. Owned
@@ -818,6 +831,41 @@ SyncResult SyncEngine::sync(main::ClientContext* context, const std::string& dbP
             return true;
         };
 
+        // One SQLite WAL commit maps to one Ladybug transaction: the batch's
+        // statements plus its cursor advance commit atomically, so a mid-tick
+        // failure (e.g. a binder error on one edge table) leaves no partial
+        // rows behind — the next tick replays the whole batch idempotently.
+        // Stats only count committed work: snapshotted and restored on
+        // rollback. DDL above stays outside the transaction: it is
+        // replay-safe by construction (IF NOT EXISTS / one-shot healing).
+        auto applyBatchTx = [&](const cdc::OpBatch& batch, uint32_t salt1, uint32_t salt2,
+                                size_t applied) -> bool {
+            int64_t savedUpserted = out.stats.upserted;
+            int64_t savedDeleted = out.stats.deleted;
+            if (!execOk(inner, "BEGIN TRANSACTION", out.error)) {
+                return false;
+            }
+            auto rollback = [&]() {
+                std::string ignored;
+                execOk(inner, "ROLLBACK", ignored);
+                out.stats.upserted = savedUpserted;
+                out.stats.deleted = savedDeleted;
+            };
+            if (!applyBatch(batch)) {
+                rollback();
+                return false;
+            }
+            if (!execOk(inner, cdc::buildCursorUpsert(syncName, salt1, salt2, applied),
+                    out.error)) {
+                rollback();
+                return false;
+            }
+            if (!execOk(inner, "COMMIT", out.error)) {
+                return false;
+            }
+            return true;
+        };
+
         Overlay overlay{nullptr, 0, &dbBytes, pageSize};
         cdc::PageFetcher fetcher = [&](uint32_t pageNo) { return overlay.fetch(pageNo); };
         bool haveWal = !walBytes.empty();
@@ -863,7 +911,7 @@ SyncResult SyncEngine::sync(main::ClientContext* context, const std::string& dbP
                     out.error = r.error;
                     return out;
                 }
-                if (!applyBatch(r.batch)) {
+                if (!applyBatchTx(r.batch, clock.salt1, clock.salt2, clock.appliedFrames)) {
                     return out;
                 }
                 out.stats.status = "Bootstrapped from database file (no WAL)";
@@ -878,12 +926,12 @@ SyncResult SyncEngine::sync(main::ClientContext* context, const std::string& dbP
                 out.error = r.error;
                 return out;
             }
-            if (!applyBatch(r.batch)) {
-                return out;
-            }
             clock.salt1 = parsed.log.header.salt1;
             clock.salt2 = parsed.log.header.salt2;
             clock.appliedFrames = parsed.log.frames.size();
+            if (!applyBatchTx(r.batch, clock.salt1, clock.salt2, clock.appliedFrames)) {
+                return out;
+            }
             out.stats.status = "Reconciled after WAL restart";
         } else {
             // Page-1 dirt in the new window means DDL: reconcile instead.
@@ -902,10 +950,10 @@ SyncResult SyncEngine::sync(main::ClientContext* context, const std::string& dbP
                     out.error = r.error;
                     return out;
                 }
-                if (!applyBatch(r.batch)) {
+                clock.appliedFrames = parsed.log.frames.size();
+                if (!applyBatchTx(r.batch, clock.salt1, clock.salt2, clock.appliedFrames)) {
                     return out;
                 }
-                clock.appliedFrames = parsed.log.frames.size();
                 out.stats.status = "Reconciled around schema change";
             } else {
                 for (auto& commit : parsed.log.commits) {
@@ -922,23 +970,20 @@ SyncResult SyncEngine::sync(main::ClientContext* context, const std::string& dbP
                         out.error = r.error;
                         return out;
                     }
-                    if (!applyBatch(r.batch)) {
+                    clock.appliedFrames = commit.lastFrame + 1;
+                    if (!applyBatchTx(r.batch, parsed.log.header.salt1, parsed.log.header.salt2,
+                            clock.appliedFrames)) {
                         return out;
                     }
-                    clock.appliedFrames = commit.lastFrame + 1;
                 }
                 out.stats.status = "Applied incrementally";
             }
             clock.salt1 = parsed.log.header.salt1;
             clock.salt2 = parsed.log.header.salt2;
         }
         out.stats.appliedFrames = static_cast<int64_t>(clock.appliedFrames);
-        // Cursor last: crash-before-cursor replays idempotently.
-        auto cursor =
-            cdc::buildCursorUpsert(syncName, clock.salt1, clock.salt2, clock.appliedFrames);
-        if (!execOk(inner, cursor, out.error)) {
-            return out;
-        }
+        // The cursor advances inside each batch transaction above, so a
+        // crash mid-tick resumes from the last committed SQLite commit.
         out.ok = true;
         return out;
     } catch (const common::Exception& e) {