diff --git a/sqlite/docs/cdc_design.md b/sqlite/docs/cdc_design.md index 66868959..6ee2b711 100644 --- a/sqlite/docs/cdc_design.md +++ b/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 / diff --git a/sqlite/src/function/sync_engine.cpp b/sqlite/src/function/sync_engine.cpp index f43c3a71..af68e9ee 100644 --- a/sqlite/src/function/sync_engine.cpp +++ b/sqlite/src/function/sync_engine.cpp @@ -818,6 +818,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 +898,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 +913,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 +937,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,10 +957,11 @@ 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"; } @@ -933,12 +969,8 @@ SyncResult SyncEngine::sync(main::ClientContext* context, const std::string& dbP clock.salt2 = parsed.log.header.salt2; } out.stats.appliedFrames = static_cast(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) {