From cdbafbb777d06a82cb3fb9765c9611a24b1ed2a1 Mon Sep 17 00:00:00 2001 From: Grzegorz Uriasz Date: Sun, 11 Oct 2026 18:16:40 +0200 Subject: [PATCH 1/2] Persist pending db_version via transaction guard vtab xSync The first local write of a transaction enlists the crsql_version_guard virtual table; its xSync upserts the pending db_version into crsql_db_versions atomically before commit. A statement or savepoint rollback can no longer leave clocks committed under a db_version whose version marker was rolled back and later reused. A committed transaction may spend a version even when every write was rolled back to a savepoint; such gaps are harmless while reuse of a committed db_version is not. (cherry picked from commit f5db9a80, ported to V1 clock tables) Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- core/Makefile | 4 +- core/rs/core/src/c.rs | 18 +- core/rs/core/src/db_version.rs | 39 +-- .../src/t/test_db_version.rs | 285 +++++++++++++++--- core/src/crsqlite.c | 11 +- core/src/ext-data.c | 6 + core/src/ext-data.h | 2 + core/src/version-guard.c | 162 ++++++++++ core/src/version-guard.h | 14 + 9 files changed, 478 insertions(+), 63 deletions(-) create mode 100644 core/src/version-guard.c create mode 100644 core/src/version-guard.h diff --git a/core/Makefile b/core/Makefile index d97f2b983..2885ccf20 100644 --- a/core/Makefile +++ b/core/Makefile @@ -88,10 +88,12 @@ TARGET_TEST_ASAN=$(prefix)/test-asan # js/browser/wa-sqlite/Makefile, deps/sqlite/GNUMakefile, core/binding.gyp, core/Makefile ext_files=src/crsqlite.c \ src/changes-vtab.c \ - src/ext-data.c + src/ext-data.c \ + src/version-guard.c ext_headers=src/crsqlite.h \ src/util.h \ src/changes-vtab.h \ + src/version-guard.h \ src/ext-data.h $(prefix): diff --git a/core/rs/core/src/c.rs b/core/rs/core/src/c.rs index b9d59e440..f24d31f7b 100644 --- a/core/rs/core/src/c.rs +++ b/core/rs/core/src/c.rs @@ -74,6 +74,8 @@ pub struct crsql_ExtData { pub mergeEqualValues: ::core::ffi::c_int, pub timestamp: ::core::ffi::c_ulonglong, pub ordinalMap: *mut ::core::ffi::c_void, + pub pVersionGuardStmt: *mut sqlite::stmt, + pub versionGuardEnlisted: ::core::ffi::c_int, } #[repr(C)] @@ -117,6 +119,10 @@ extern "C" { ) -> c_int; pub fn crsql_freeExtData(pExtData: *mut crsql_ExtData); pub fn crsql_finalize(pExtData: *mut crsql_ExtData); + pub fn crsql_enlist_db_version( + db: *mut sqlite::sqlite3, + pExtData: *mut crsql_ExtData, + ) -> c_int; } #[test] @@ -271,7 +277,7 @@ fn bindgen_test_layout_crsql_ExtData() { let ptr = UNINIT.as_ptr(); assert_eq!( ::core::mem::size_of::(), - 168usize, + 184usize, concat!("Size of: ", stringify!(crsql_ExtData)) ); assert_eq!( @@ -511,4 +517,14 @@ fn bindgen_test_layout_crsql_ExtData() { stringify!(ordinalMap) ) ); + assert_eq!( + unsafe { ::core::ptr::addr_of!((*ptr).pVersionGuardStmt) as usize - ptr as usize }, + 168usize, + "Offset of pVersionGuardStmt" + ); + assert_eq!( + unsafe { ::core::ptr::addr_of!((*ptr).versionGuardEnlisted) as usize - ptr as usize }, + 176usize, + "Offset of versionGuardEnlisted" + ); } diff --git a/core/rs/core/src/db_version.rs b/core/rs/core/src/db_version.rs index 295773b30..0d2508d1f 100644 --- a/core/rs/core/src/db_version.rs +++ b/core/rs/core/src/db_version.rs @@ -13,6 +13,7 @@ use sqlite::{sqlite3, Stmt}; use sqlite_nostd as sqlite; use crate::c::crsql_ExtData; +use crate::c::crsql_enlist_db_version; use crate::consts::MIN_POSSIBLE_DB_VERSION; use crate::consts::SITE_ID_LEN; use crate::stmt_cache::reset_cached_stmt; @@ -63,37 +64,29 @@ pub fn peek_next_db_version(db: *mut sqlite3, ext_data: *mut crsql_ExtData) -> R * Need to save some bit that states we checked the pragma already and reset on tx commit or rollback. */ pub fn next_db_version(db: *mut sqlite3, ext_data: *mut crsql_ExtData) -> Result { - crate::ext_data::refresh_connection_state_if_needed(db, ext_data)?; + if unsafe { (*ext_data).pendingDbVersion } < 0 { + reload_db_version(db, ext_data)?; + } else { + crate::ext_data::refresh_connection_state_if_needed(db, ext_data)?; + } let mut ret = unsafe { (*ext_data).dbVersion + 1 }; if ret < unsafe { (*ext_data).pendingDbVersion } { ret = unsafe { (*ext_data).pendingDbVersion }; } - // update db_version in db if it changed - if ret != unsafe { (*ext_data).pendingDbVersion } { + if ret != unsafe { (*ext_data).pendingDbVersion } + || unsafe { (*ext_data).versionGuardEnlisted == 0 } + { unsafe { - let site_id_slice = - core::slice::from_raw_parts((*ext_data).siteId, SITE_ID_LEN as usize); - - let bind_result = (*ext_data) - .pSetDbVersionStmt - .bind_blob(1, site_id_slice, sqlite_nostd::Destructor::STATIC) - .and_then(|_| (*ext_data).pSetDbVersionStmt.bind_int64(2, ret)); - - if bind_result.is_err() { - return Err("failed binding to pSetDbVersionStmt".into()); - } - - let res = (*ext_data).pSetDbVersionStmt.step(); - - reset_cached_stmt((*ext_data).pSetDbVersionStmt) - .map_err(|_| "failed to reset cached pSetDbVersionStmt")?; - - if res.is_err() { - return Err("failed to insert db_version for current site ID".into()); - } + let previous = (*ext_data).pendingDbVersion; (*ext_data).pendingDbVersion = ret; + if crsql_enlist_db_version(db, ext_data) != ResultCode::OK as c_int { + if (*ext_data).pendingDbVersion == ret { + (*ext_data).pendingDbVersion = previous; + } + return Err("failed to enlist db_version guard".into()); + } } } diff --git a/core/rs/integration_check/src/t/test_db_version.rs b/core/rs/integration_check/src/t/test_db_version.rs index 78c16e3e9..3063b67c7 100644 --- a/core/rs/integration_check/src/t/test_db_version.rs +++ b/core/rs/integration_check/src/t/test_db_version.rs @@ -88,6 +88,7 @@ fn test_next_db_version() -> Result<(), String> { let rc = unsafe { test_exports::c::crsql_initSiteIdExt(raw_db, ext_data, make_site()) }; assert_eq!(rc, 0); + db.exec_safe("BEGIN").expect("started transaction"); // is current + 1 // doesn't bump forward on successive calls assert_eq!( @@ -110,6 +111,7 @@ fn test_next_db_version() -> Result<(), String> { // existing db version not touched assert_eq!(0, unsafe { (*ext_data).dbVersion }); + db.exec_safe("COMMIT").expect("committed transaction"); unsafe { test_exports::c::crsql_freeExtData(ext_data); @@ -447,52 +449,251 @@ fn test_insert_db_version_cache() -> Result<(), ResultCode> { Ok(()) } -fn scalar(db: &sqlite::ManagedConnection, sql: &str) -> Result { - let stmt = db.prepare_v2(sql).map_err(|e| format!("prepare {sql}: {e:?}"))?; - match stmt.step() { - Ok(ResultCode::ROW) => {} - rc => { - return Err(format!( - "step {sql}: {rc:?} errmsg={:?}", - db.errmsg() - )) +fn rollback_test_path() -> String { + let mut bytes = [0u8; 8]; + sqlite::randomness(&mut bytes); + format!("test_db_version_rollback_{:016x}.db", u64::from_ne_bytes(bytes)) +} + +fn remove_rollback_test_db(path: &str) { + let path = CString::new(path).unwrap(); + #[cfg(not(target_os = "windows"))] + unsafe { + extern "C" { + fn unlink(path: *const c_char) -> core::ffi::c_int; } + assert_eq!(unlink(path.as_ptr()), 0); } - Ok(stmt.column_int64(0)) + #[cfg(target_os = "windows")] + unsafe { + extern "C" { + fn _unlink(path: *const c_char) -> core::ffi::c_int; + } + assert_eq!(_unlink(path.as_ptr()), 0); + } +} + +fn open_rollback_test_db(path: &str) -> Result { + let db = crate::opendb_file(path)?; + db.db.exec_safe("CREATE TABLE t (id INTEGER PRIMARY KEY NOT NULL, v TEXT)")?; + db.db.exec_safe("SELECT crsql_as_crr('t')")?; + Ok(db) } -fn test_connection_state_refresh() -> Result<(), String> { - let a = crate::opendb_file("test_connection_state_refresh.db") - .map_err(|e| format!("open a: {e:?}"))?; - let b = crate::opendb_file("test_connection_state_refresh.db") - .map_err(|e| format!("open b: {e:?}"))?; +fn test_connection_state_refresh() -> Result<(), ResultCode> { + let a = crate::opendb_file("test_connection_state_refresh.db")?; + let b = crate::opendb_file("test_connection_state_refresh.db")?; a.db.exec_safe( "DROP TABLE IF EXISTS foo; DROP TABLE IF EXISTS foo__crsql_clock; DELETE FROM crsql_master WHERE key = 'config.merge-equal-values'; DELETE FROM crsql_db_versions; VACUUM;", - ) - .map_err(|e| format!("reset db: {e:?}"))?; - a.db.exec_safe("CREATE TABLE foo (id PRIMARY KEY NOT NULL, v)") - .map_err(|e| format!("create foo: {e:?}"))?; - a.db.exec_safe("SELECT crsql_as_crr('foo')") - .map_err(|e| format!("as_crr: {e:?}"))?; - assert_eq!(scalar(&a.db, "SELECT crsql_db_version()")?, 0); - - b.db.exec_safe("SELECT crsql_config_set('merge-equal-values', 1)") - .map_err(|e| format!("b set config 1: {e:?}"))?; - b.db.exec_safe("INSERT INTO foo VALUES (1, 'first')") - .map_err(|e| format!("b insert 1: {e:?}"))?; - assert_eq!(scalar(&a.db, "SELECT crsql_config_get('merge-equal-values')")?, 1); - assert_eq!(scalar(&a.db, "SELECT crsql_peek_next_db_version()")?, 2); - - b.db.exec_safe("SELECT crsql_config_set('merge-equal-values', 0)") - .map_err(|e| format!("b set config 0: {e:?}"))?; - b.db.exec_safe("INSERT INTO foo VALUES (2, 'second')") - .map_err(|e| format!("b insert 2: {e:?}"))?; - assert_eq!(scalar(&a.db, "SELECT crsql_peek_next_db_version()")?, 3); - assert_eq!(scalar(&a.db, "SELECT crsql_config_get('merge-equal-values')")?, 0); + )?; + a.db.exec_safe("CREATE TABLE foo (id PRIMARY KEY NOT NULL, v)")?; + a.db.exec_safe("SELECT crsql_as_crr('foo')")?; + assert_eq!(scalar(a.db.db, "SELECT crsql_db_version()")?, 0); + + b.db.exec_safe("SELECT crsql_config_set('merge-equal-values', 1)")?; + b.db.exec_safe("INSERT INTO foo VALUES (1, 'first')")?; + assert_eq!(scalar(a.db.db, "SELECT crsql_config_get('merge-equal-values')")?, 1); + assert_eq!(scalar(a.db.db, "SELECT crsql_peek_next_db_version()")?, 2); + + b.db.exec_safe("SELECT crsql_config_set('merge-equal-values', 0)")?; + b.db.exec_safe("INSERT INTO foo VALUES (2, 'second')")?; + assert_eq!(scalar(a.db.db, "SELECT crsql_peek_next_db_version()")?, 3); + assert_eq!(scalar(a.db.db, "SELECT crsql_config_get('merge-equal-values')")?, 0); + Ok(()) +} + +fn scalar(db: *mut sqlite::sqlite3, sql: &str) -> Result { + let stmt = db.prepare_v2(sql)?; + assert_eq!(stmt.step()?, ResultCode::ROW, "{sql}"); + Ok(stmt.column_int64(0)) +} + +fn test_savepoint_rollback_after_first_write() -> Result<(), ResultCode> { + let path = rollback_test_path(); + { + let db = open_rollback_test_db(&path)?; + db.db.exec_safe("BEGIN")?; + db.db.exec_safe("SELECT crsql_set_ts('1700000001')")?; + db.db.exec_safe("INSERT INTO t VALUES (1, 'before')")?; + db.db.exec_safe("SAVEPOINT s")?; + db.db.exec_safe("INSERT INTO t VALUES (2, 'rolled back')")?; + db.db.exec_safe("ROLLBACK TO s")?; + db.db.exec_safe("RELEASE s")?; + db.db.exec_safe("INSERT INTO t VALUES (3, 'after')")?; + db.db.exec_safe("COMMIT")?; + assert_eq!(scalar(db.db.db, "SELECT count(*) FROM t")?, 2); + assert_eq!( + scalar(db.db.db, "SELECT count(*) FROM t__crsql_clock WHERE db_version = 1")?, + 2 + ); + } + { + let db = crate::opendb_file(&path)?; + assert_eq!(scalar(db.db.db, "SELECT crsql_peek_next_db_version()")?, 2); + db.db.exec_safe("BEGIN")?; + db.db.exec_safe("SELECT crsql_set_ts('1700000002')")?; + db.db.exec_safe("INSERT INTO t VALUES (4, 'next')")?; + db.db.exec_safe("COMMIT")?; + assert_eq!(scalar(db.db.db, "SELECT MAX(db_version) FROM t__crsql_clock")?, 2); + } + remove_rollback_test_db(&path); + Ok(()) +} + +fn test_savepoint_rollback_without_surviving_writes() -> Result<(), ResultCode> { + let path = rollback_test_path(); + { + let db = open_rollback_test_db(&path)?; + db.db.exec_safe("BEGIN")?; + db.db.exec_safe("SELECT crsql_set_ts('1700000001')")?; + db.db.exec_safe("SAVEPOINT s")?; + db.db.exec_safe("INSERT INTO t VALUES (1, 'rolled back')")?; + db.db.exec_safe("ROLLBACK TO s")?; + db.db.exec_safe("RELEASE s")?; + db.db.exec_safe("COMMIT")?; + assert_eq!(scalar(db.db.db, "SELECT count(*) FROM t__crsql_clock")?, 0); + assert_eq!(scalar(db.db.db, "SELECT MAX(db_version) FROM crsql_db_versions")?, 1); + } + { + let db = crate::opendb_file(&path)?; + assert_eq!(scalar(db.db.db, "SELECT crsql_peek_next_db_version()")?, 2); + db.db.exec_safe("BEGIN")?; + db.db.exec_safe("SELECT crsql_set_ts('1700000002')")?; + db.db.exec_safe("INSERT INTO t VALUES (2, 'first committed')")?; + db.db.exec_safe("COMMIT")?; + assert_eq!(scalar(db.db.db, "SELECT MAX(db_version) FROM t__crsql_clock")?, 2); + } + remove_rollback_test_db(&path); + Ok(()) +} + +fn test_transaction_rollback_does_not_persist_version() -> Result<(), ResultCode> { + let path = rollback_test_path(); + { + let db = open_rollback_test_db(&path)?; + db.db.exec_safe("BEGIN")?; + db.db.exec_safe("SELECT crsql_set_ts('1700000001')")?; + db.db.exec_safe("INSERT INTO t VALUES (1, 'rolled back')")?; + db.db.exec_safe("ROLLBACK")?; + assert_eq!(scalar(db.db.db, "SELECT count(*) FROM crsql_db_versions")?, 0); + } + { + let db = crate::opendb_file(&path)?; + assert_eq!(scalar(db.db.db, "SELECT crsql_peek_next_db_version()")?, 1); + db.db.exec_safe("BEGIN")?; + db.db.exec_safe("SELECT crsql_set_ts('1700000002')")?; + db.db.exec_safe("INSERT INTO t VALUES (2, 'kept')")?; + db.db.exec_safe("COMMIT")?; + assert_eq!(scalar(db.db.db, "SELECT MAX(db_version) FROM crsql_db_versions")?, 1); + } + remove_rollback_test_db(&path); + Ok(()) +} + +fn test_stale_version_aborts_commit() -> Result<(), ResultCode> { + let path = rollback_test_path(); + { + let db = open_rollback_test_db(&path)?; + db.db.exec_safe("BEGIN")?; + db.db.exec_safe("SELECT crsql_set_ts('1700000001')")?; + db.db.exec_safe("INSERT INTO t VALUES (1, 'kept')")?; + db.db.exec_safe( + "INSERT INTO crsql_db_versions (site_id, db_version) VALUES (crsql_site_id(), 9)", + )?; + assert_eq!(db.db.exec_safe("COMMIT"), Err(ResultCode::CONSTRAINT)); + assert_eq!(scalar(db.db.db, "SELECT count(*) FROM t")?, 0); + assert_eq!(scalar(db.db.db, "SELECT count(*) FROM crsql_db_versions")?, 0); + } + remove_rollback_test_db(&path); + Ok(()) +} + +fn test_savepoint_rollback_first_write_then_commit() -> Result<(), ResultCode> { + let path = rollback_test_path(); + let persisted_version = { + let db = open_rollback_test_db(&path)?; + db.db.exec_safe("BEGIN")?; + db.db.exec_safe("SELECT crsql_set_ts('1700000001')")?; + db.db.exec_safe("SAVEPOINT s")?; + db.db.exec_safe("INSERT INTO t VALUES (1, 'rolled back')")?; + db.db.exec_safe("ROLLBACK TO s")?; + db.db.exec_safe("RELEASE s")?; + db.db.exec_safe("INSERT INTO t VALUES (2, 'kept')")?; + db.db.exec_safe("COMMIT")?; + assert_eq!(scalar(db.db.db, "SELECT MAX(db_version) FROM t__crsql_clock")?, 1); + scalar( + db.db.db, + "SELECT COALESCE((SELECT db_version FROM crsql_db_versions WHERE site_id = crsql_site_id()), 0)", + )? + }; + let (next_version, written_version, timestamps) = { + let db = crate::opendb_file(&path)?; + let next_version = scalar(db.db.db, "SELECT crsql_peek_next_db_version()")?; + db.db.exec_safe("BEGIN")?; + db.db.exec_safe("SELECT crsql_set_ts('1700000002')")?; + db.db.exec_safe("INSERT INTO t VALUES (3, 'next')")?; + db.db.exec_safe("COMMIT")?; + ( + next_version, + scalar(db.db.db, "SELECT MAX(db_version) FROM t__crsql_clock")?, + scalar( + db.db.db, + "SELECT COUNT(DISTINCT ts) FROM t__crsql_clock WHERE db_version = 1", + )?, + ) + }; + remove_rollback_test_db(&path); + assert_eq!(persisted_version, 1, "committed clock version must be persisted"); + assert_eq!(next_version, 2, "committed version 1 must not be reused after reopening"); + assert_eq!(written_version, 2); + assert_eq!(timestamps, 1, "separate transactions must not share db_version 1"); + Ok(()) +} + +fn test_failed_statement_then_commit_does_not_reuse_version() -> Result<(), ResultCode> { + let path = rollback_test_path(); + let persisted_version = { + let db = open_rollback_test_db(&path)?; + db.db.exec_safe("BEGIN")?; + db.db.exec_safe("SELECT crsql_set_ts('1700000001')")?; + assert_eq!( + db.db.exec_safe("INSERT INTO t VALUES (1, 'first'), (1, 'duplicate')"), + Err(ResultCode::CONSTRAINT) + ); + assert_eq!(scalar(db.db.db, "SELECT count(*) FROM t")?, 0); + db.db.exec_safe("INSERT INTO t VALUES (2, 'kept')")?; + db.db.exec_safe("COMMIT")?; + assert_eq!(scalar(db.db.db, "SELECT count(*) FROM t__crsql_clock")?, 1); + scalar( + db.db.db, + "SELECT COALESCE((SELECT db_version FROM crsql_db_versions WHERE site_id = crsql_site_id()), 0)", + )? + }; + let (next_version, written_version, timestamps) = { + let db = crate::opendb_file(&path)?; + let next_version = scalar(db.db.db, "SELECT crsql_peek_next_db_version()")?; + db.db.exec_safe("BEGIN")?; + db.db.exec_safe("SELECT crsql_set_ts('1700000002')")?; + db.db.exec_safe("INSERT INTO t VALUES (3, 'next')")?; + db.db.exec_safe("COMMIT")?; + ( + next_version, + scalar(db.db.db, "SELECT MAX(db_version) FROM t__crsql_clock")?, + scalar( + db.db.db, + "SELECT COUNT(DISTINCT ts) FROM t__crsql_clock WHERE db_version = 1", + )?, + ) + }; + remove_rollback_test_db(&path); + assert_eq!(persisted_version, 1, "the surviving write must persist version 1"); + assert_eq!(next_version, 2, "a later transaction must not reuse version 1"); + assert_eq!(written_version, 2); + assert_eq!(timestamps, 1, "separate transactions must not share db_version 1"); Ok(()) } @@ -505,6 +706,18 @@ pub fn run_suite() -> Result<(), String> { test_get_or_set_pk_cl().map_err(|e| format!("test_get_or_set_pk_cl failed: {:?}", e))?; test_insert_db_version_cache() .map_err(|e| format!("test_insert_db_version_cache failed: {:?}", e))?; + test_savepoint_rollback_after_first_write() + .map_err(|e| format!("test_savepoint_rollback_after_first_write failed: {:?}", e))?; + test_savepoint_rollback_without_surviving_writes() + .map_err(|e| format!("test_savepoint_rollback_without_surviving_writes failed: {:?}", e))?; + test_transaction_rollback_does_not_persist_version() + .map_err(|e| format!("test_transaction_rollback_does_not_persist_version failed: {:?}", e))?; + test_stale_version_aborts_commit() + .map_err(|e| format!("test_stale_version_aborts_commit failed: {:?}", e))?; + test_failed_statement_then_commit_does_not_reuse_version() + .map_err(|e| format!("test_failed_statement_then_commit_does_not_reuse_version failed: {:?}", e))?; + test_savepoint_rollback_first_write_then_commit() + .map_err(|e| format!("test_savepoint_rollback_first_write_then_commit failed: {:?}", e))?; Ok(()) } diff --git a/core/src/crsqlite.c b/core/src/crsqlite.c index ce55caaf3..9ca353e9d 100644 --- a/core/src/crsqlite.c +++ b/core/src/crsqlite.c @@ -11,9 +11,8 @@ LIBSQL_EXTENSION_INIT1 #include #include "changes-vtab.h" -#include "consts.h" #include "ext-data.h" -#include "rust.h" +#include "version-guard.h" // see // https://github.com/chromium/chromium/commit/579b3dd0ea41a40da8a61ab87a8b0bc39e158998 @@ -69,6 +68,14 @@ __declspec(dllexport) rc = sqlite3_create_module_v2(db, "crsql_changes", &crsql_changesModule, pExtData, 0); } + if (rc == SQLITE_OK) { + rc = sqlite3_create_module_v2(db, "crsql_version_guard", + &crsql_versionGuardModule, pExtData, 0); + } + if (rc != SQLITE_OK) { + crsql_finalize(pExtData); + return rc; + } if (rc == SQLITE_OK) { #ifdef LIBSQL diff --git a/core/src/ext-data.c b/core/src/ext-data.c index 5ae04582d..b897aeaaa 100644 --- a/core/src/ext-data.c +++ b/core/src/ext-data.c @@ -70,6 +70,8 @@ crsql_ExtData *crsql_newExtData(sqlite3 *db) { pExtData->tableInfos = 0; pExtData->lastDbVersions = 0; pExtData->ordinalMap = 0; + pExtData->pVersionGuardStmt = 0; + pExtData->versionGuardEnlisted = 0; pExtData->rowsImpacted = 0; pExtData->updatedTableInfosThisTx = 0; crsql_init_table_info_vec(pExtData); @@ -143,6 +145,7 @@ void crsql_freeExtData(crsql_ExtData *pExtData) { } sqlite3_finalize(pExtData->pDbVersionStmt); sqlite3_finalize(pExtData->pSetDbVersionStmt); + sqlite3_finalize(pExtData->pVersionGuardStmt); sqlite3_finalize(pExtData->pPragmaSchemaVersionStmt); sqlite3_finalize(pExtData->pPragmaDataVersionStmt); sqlite3_finalize(pExtData->pSetSyncBitStmt); @@ -156,6 +159,7 @@ void crsql_freeExtData(crsql_ExtData *pExtData) { } pExtData->pDbVersionStmt = 0; pExtData->pSetDbVersionStmt = 0; + pExtData->pVersionGuardStmt = 0; pExtData->pPragmaSchemaVersionStmt = 0; pExtData->pPragmaDataVersionStmt = 0; pExtData->pSetSyncBitStmt = 0; @@ -179,6 +183,7 @@ void crsql_finalize(crsql_ExtData *pExtData) { // printf("crsql_finalize\n"); sqlite3_finalize(pExtData->pDbVersionStmt); sqlite3_finalize(pExtData->pSetDbVersionStmt); + sqlite3_finalize(pExtData->pVersionGuardStmt); sqlite3_finalize(pExtData->pPragmaSchemaVersionStmt); sqlite3_finalize(pExtData->pPragmaDataVersionStmt); sqlite3_finalize(pExtData->pSetSyncBitStmt); @@ -193,6 +198,7 @@ void crsql_finalize(crsql_ExtData *pExtData) { crsql_clear_stmt_cache(pExtData); pExtData->pDbVersionStmt = 0; pExtData->pSetDbVersionStmt = 0; + pExtData->pVersionGuardStmt = 0; pExtData->pPragmaSchemaVersionStmt = 0; pExtData->pPragmaDataVersionStmt = 0; pExtData->pSetSyncBitStmt = 0; diff --git a/core/src/ext-data.h b/core/src/ext-data.h index 2d8dd9b41..0986b66bf 100644 --- a/core/src/ext-data.h +++ b/core/src/ext-data.h @@ -49,6 +49,8 @@ struct crsql_ExtData { int mergeEqualValues; unsigned long long timestamp; void *ordinalMap; + sqlite3_stmt *pVersionGuardStmt; + int versionGuardEnlisted; }; crsql_ExtData *crsql_newExtData(sqlite3 *db); diff --git a/core/src/version-guard.c b/core/src/version-guard.c new file mode 100644 index 000000000..959d1322e --- /dev/null +++ b/core/src/version-guard.c @@ -0,0 +1,162 @@ +/** + * The version guard virtual table exists to defer the db_version write to + * just before commit. The first local write of a transaction enlists it via + * crsql_enlist_db_version; its xSync then upserts the pending db_version + * into crsql_db_versions atomically before the transaction commits. SQLite + * guarantees xSync runs before either xCommit or xRollback, so the version + * row is never persisted for a transaction that rolls back. The guard only + * ever manages this node's own version, so xSync reads the site id and + * pending db_version from the connection's ExtData rather than taking + * parameters. + * + * Persisting the version marker in xSync means a statement or savepoint + * rollback can no longer leave clocks committed under a db_version whose + * crsql_db_versions row was rolled back. A committed transaction may spend + * a version even when all of its writes were rolled back to a savepoint; + * such gaps are harmless while version reuse is not. + */ +#include "version-guard.h" + +#include + +#include "consts.h" + +typedef struct { + sqlite3_vtab base; + crsql_ExtData *extData; +} crsql_VersionGuard; + +static int versionGuardConnect(sqlite3 *db, void *aux, int argc, + const char *const *argv, sqlite3_vtab **vtab, + char **err) { + int rc = sqlite3_declare_vtab(db, "CREATE TABLE x(dummy INTEGER)"); + if (rc != SQLITE_OK) return rc; + crsql_VersionGuard *guard = sqlite3_malloc(sizeof(*guard)); + if (!guard) return SQLITE_NOMEM; + memset(guard, 0, sizeof(*guard)); + guard->extData = aux; + *vtab = &guard->base; + return SQLITE_OK; +} + +static int versionGuardDisconnect(sqlite3_vtab *vtab) { + sqlite3_free(vtab); + return SQLITE_OK; +} + +static int versionGuardBestIndex(sqlite3_vtab *vtab, sqlite3_index_info *info) { + return SQLITE_OK; +} + +static int versionGuardOpen(sqlite3_vtab *vtab, sqlite3_vtab_cursor **cursor) { + sqlite3_vtab_cursor *c = sqlite3_malloc(sizeof(*c)); + if (!c) return SQLITE_NOMEM; + memset(c, 0, sizeof(*c)); + c->pVtab = vtab; + *cursor = c; + return SQLITE_OK; +} + +static int versionGuardClose(sqlite3_vtab_cursor *cursor) { + sqlite3_free(cursor); + return SQLITE_OK; +} + +static int versionGuardFilter(sqlite3_vtab_cursor *cursor, int idxNum, + const char *idxStr, int argc, + sqlite3_value **argv) { + return SQLITE_OK; +} + +static int versionGuardNext(sqlite3_vtab_cursor *cursor) { return SQLITE_OK; } +static int versionGuardEof(sqlite3_vtab_cursor *cursor) { return 1; } +static int versionGuardColumn(sqlite3_vtab_cursor *cursor, sqlite3_context *ctx, + int column) { + sqlite3_result_null(ctx); + return SQLITE_OK; +} +static int versionGuardRowid(sqlite3_vtab_cursor *cursor, sqlite3_int64 *rowid) { + *rowid = 0; + return SQLITE_OK; +} + +static int versionGuardBegin(sqlite3_vtab *vtab) { + crsql_VersionGuard *guard = (crsql_VersionGuard *)vtab; + guard->extData->versionGuardEnlisted = 1; + return SQLITE_OK; +} + +static int versionGuardUpdate(sqlite3_vtab *vtab, int argc, + sqlite3_value **argv, sqlite3_int64 *rowid) { + // The insert exists only to enlist this vtab in the transaction so that + // its xSync runs before commit; the values themselves are unused. + return SQLITE_OK; +} + +static int versionGuardSync(sqlite3_vtab *vtab) { + crsql_VersionGuard *guard = (crsql_VersionGuard *)vtab; + crsql_ExtData *extData = guard->extData; + sqlite3_int64 version = extData->pendingDbVersion; + sqlite3_stmt *stmt = extData->pSetDbVersionStmt; + int rc = sqlite3_bind_blob(stmt, 1, extData->siteId, SITE_ID_LEN, + SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(stmt, 2, version); + if (rc == SQLITE_OK) rc = sqlite3_step(stmt); + int result = SQLITE_OK; + if (rc == SQLITE_ROW) { + if (sqlite3_column_int64(stmt, 0) != version) + result = SQLITE_CONSTRAINT; + } else if (rc == SQLITE_DONE) { + stmt = extData->pDbVersionStmt; + rc = sqlite3_bind_blob(stmt, 1, extData->siteId, SITE_ID_LEN, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_step(stmt); + if (rc != SQLITE_ROW || sqlite3_column_int64(stmt, 0) != version) + result = rc == SQLITE_ROW || rc == SQLITE_DONE ? SQLITE_CONSTRAINT : rc; + rc = sqlite3_reset(stmt); + if (result == SQLITE_OK) result = rc; + stmt = extData->pSetDbVersionStmt; + } else { + result = rc; + } + rc = sqlite3_reset(stmt); + return result == SQLITE_OK ? rc : result; +} + +static int versionGuardEnd(sqlite3_vtab *vtab) { + crsql_VersionGuard *guard = (crsql_VersionGuard *)vtab; + guard->extData->versionGuardEnlisted = 0; + return SQLITE_OK; +} + +sqlite3_module crsql_versionGuardModule = { + .iVersion = 2, + .xConnect = versionGuardConnect, + .xBestIndex = versionGuardBestIndex, + .xDisconnect = versionGuardDisconnect, + .xOpen = versionGuardOpen, + .xClose = versionGuardClose, + .xFilter = versionGuardFilter, + .xNext = versionGuardNext, + .xEof = versionGuardEof, + .xColumn = versionGuardColumn, + .xRowid = versionGuardRowid, + .xUpdate = versionGuardUpdate, + .xBegin = versionGuardBegin, + .xSync = versionGuardSync, + .xCommit = versionGuardEnd, + .xRollback = versionGuardEnd, +}; + +int crsql_enlist_db_version(sqlite3 *db, crsql_ExtData *extData) { + if (!extData->pVersionGuardStmt) { + int rc = sqlite3_prepare_v3( + db, "INSERT INTO crsql_version_guard VALUES (NULL)", -1, + SQLITE_PREPARE_PERSISTENT, &extData->pVersionGuardStmt, 0); + if (rc != SQLITE_OK) return rc; + } + sqlite3_stmt *stmt = extData->pVersionGuardStmt; + int rc = sqlite3_step(stmt); + int result = rc == SQLITE_DONE ? SQLITE_OK : rc; + rc = sqlite3_reset(stmt); + return result == SQLITE_OK ? rc : result; +} diff --git a/core/src/version-guard.h b/core/src/version-guard.h new file mode 100644 index 000000000..aa404fba4 --- /dev/null +++ b/core/src/version-guard.h @@ -0,0 +1,14 @@ +#ifndef CRSQLITE_VERSION_GUARD_H +#define CRSQLITE_VERSION_GUARD_H + +#if !defined(SQLITEINT_H) +#include "sqlite3ext.h" +#endif +SQLITE_EXTENSION_INIT3 + +#include "ext-data.h" + +extern sqlite3_module crsql_versionGuardModule; +int crsql_enlist_db_version(sqlite3 *db, crsql_ExtData *pExtData); + +#endif From 1b7123f22569f2dfcce1342cfe714a15c35cf4ef Mon Sep 17 00:00:00 2001 From: Grzegorz Uriasz Date: Mon, 12 Oct 2026 12:29:01 +0200 Subject: [PATCH 2/2] Return pendingDbVersion directly once the guard is enlisted Once crsql_version_guard is enlisted, the transaction's db_version is fixed; next_db_version and peek_next_db_version now return pendingDbVersion directly instead of re-running the max() allocation math. The un-enlisted path keeps a single fill_db_version_if_needed refresh before allocating dbVersion + 1 and enlisting. Also document that the dbVersion != -1 check in refresh_connection_state_if_needed is the first-load guard for the lazily-initialized version cache, not a data_version condition. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- core/rs/core/src/db_version.rs | 53 +++++++------------ core/rs/core/src/ext_data.rs | 23 ++++++-- .../src/t/test_db_version.rs | 30 +++++++++++ core/src/version-guard.c | 12 ++--- 4 files changed, 76 insertions(+), 42 deletions(-) diff --git a/core/rs/core/src/db_version.rs b/core/rs/core/src/db_version.rs index 0d2508d1f..fa3b2d964 100644 --- a/core/rs/core/src/db_version.rs +++ b/core/rs/core/src/db_version.rs @@ -49,48 +49,35 @@ pub extern "C" fn crsql_peek_next_db_version( } pub fn peek_next_db_version(db: *mut sqlite3, ext_data: *mut crsql_ExtData) -> Result { - crate::ext_data::refresh_connection_state_if_needed(db, ext_data)?; - - let mut ret = unsafe { (*ext_data).dbVersion + 1 }; - if ret < unsafe { (*ext_data).pendingDbVersion } { - ret = unsafe { (*ext_data).pendingDbVersion }; + if unsafe { (*ext_data).versionGuardEnlisted } != 0 { + return Ok(unsafe { (*ext_data).pendingDbVersion }); } - Ok(ret) + crate::ext_data::refresh_connection_state_if_needed(db, ext_data)?; + Ok(unsafe { (*ext_data).dbVersion + 1 }) } /** - * Given this needs to do a pragma check, invoke it as little as possible. - * TODO: We could optimize to only do a pragma check once per transaction. - * Need to save some bit that states we checked the pragma already and reset on tx commit or rollback. + * Returns the db_version local writes of the current transaction must use. + * + * Once the version guard is enlisted the transaction's version is fixed: + * every subsequent call just returns pendingDbVersion. Otherwise we allocate + * one and enlist the guard so xSync persists it atomically at commit. */ pub fn next_db_version(db: *mut sqlite3, ext_data: *mut crsql_ExtData) -> Result { - if unsafe { (*ext_data).pendingDbVersion } < 0 { - reload_db_version(db, ext_data)?; - } else { - crate::ext_data::refresh_connection_state_if_needed(db, ext_data)?; + if unsafe { (*ext_data).versionGuardEnlisted } != 0 { + return Ok(unsafe { (*ext_data).pendingDbVersion }); } - - let mut ret = unsafe { (*ext_data).dbVersion + 1 }; - if ret < unsafe { (*ext_data).pendingDbVersion } { - ret = unsafe { (*ext_data).pendingDbVersion }; - } - - if ret != unsafe { (*ext_data).pendingDbVersion } - || unsafe { (*ext_data).versionGuardEnlisted == 0 } - { - unsafe { - let previous = (*ext_data).pendingDbVersion; - (*ext_data).pendingDbVersion = ret; - if crsql_enlist_db_version(db, ext_data) != ResultCode::OK as c_int { - if (*ext_data).pendingDbVersion == ret { - (*ext_data).pendingDbVersion = previous; - } - return Err("failed to enlist db_version guard".into()); - } + crate::ext_data::refresh_connection_state_if_needed(db, ext_data)?; + unsafe { + let ret = (*ext_data).dbVersion + 1; + (*ext_data).pendingDbVersion = ret; + if crsql_enlist_db_version(db, ext_data) != ResultCode::OK as c_int { + // xSync must not persist a version when enlistment failed. + (*ext_data).versionGuardEnlisted = 0; + return Err("failed to enlist db_version guard".into()); } + Ok(ret) } - - Ok(ret) } pub fn reload_db_version( diff --git a/core/rs/core/src/ext_data.rs b/core/rs/core/src/ext_data.rs index 3d0616a12..f51ab6341 100644 --- a/core/rs/core/src/ext_data.rs +++ b/core/rs/core/src/ext_data.rs @@ -27,16 +27,33 @@ pub fn refresh_connection_state_if_needed( db: *mut sqlite3, ext_data: *mut crsql_ExtData, ) -> Result<(), String> { + let (previous_data_version, previous_db_version) = unsafe { + ((*ext_data).pragmaDataVersion, (*ext_data).dbVersion) + }; let rc = unsafe { crsql_fetchPragmaDataVersion(db, ext_data) }; if rc == -1 { + unsafe { + (*ext_data).pragmaDataVersion = previous_data_version; + (*ext_data).dbVersion = previous_db_version; + } return Err("failed to fetch PRAGMA data_version".to_string()); } + // `data_version` cannot distinguish "unchanged" from "never loaded": + // dbVersion is the one lazily-initialized cache, so its -1 sentinel + // doubles as the first-load check for all of them. if rc == 0 && unsafe { (*ext_data).dbVersion } != -1 { return Ok(()); } - crate::db_version::reload_db_version(db, ext_data)?; - crate::config::reload_connection_config(db, ext_data)?; - Ok(()) + let result = crate::db_version::reload_db_version(db, ext_data) + .map(|_| ()) + .and_then(|_| crate::config::reload_connection_config(db, ext_data)); + if result.is_err() { + unsafe { + (*ext_data).pragmaDataVersion = previous_data_version; + (*ext_data).dbVersion = previous_db_version; + } + } + result } #[no_mangle] diff --git a/core/rs/integration_check/src/t/test_db_version.rs b/core/rs/integration_check/src/t/test_db_version.rs index 3063b67c7..d69888ad5 100644 --- a/core/rs/integration_check/src/t/test_db_version.rs +++ b/core/rs/integration_check/src/t/test_db_version.rs @@ -506,6 +506,34 @@ fn test_connection_state_refresh() -> Result<(), ResultCode> { Ok(()) } +fn test_refresh_retries_after_reload_failure() -> Result<(), ResultCode> { + let path = rollback_test_path(); + { + let a = open_rollback_test_db(&path)?; + let b = crate::opendb_file(&path)?; + assert_eq!(scalar(b.db.db, "SELECT crsql_db_version()")?, 0); + + b.db.exec_safe("BEGIN")?; + b.db.exec_safe("SELECT crsql_config_set('merge-equal-values', 1)")?; + b.db.exec_safe( + "UPDATE crsql_master SET value = 'invalid' WHERE key = 'config.merge-equal-values'", + )?; + b.db.exec_safe("INSERT INTO t VALUES (1, 'external')")?; + b.db.exec_safe("COMMIT")?; + + assert!(scalar(a.db.db, "SELECT crsql_config_get('merge-equal-values')").is_err()); + a.db.exec_safe( + "UPDATE crsql_master SET value = 1 WHERE key = 'config.merge-equal-values'", + )?; + assert_eq!( + scalar(a.db.db, "SELECT crsql_config_get('merge-equal-values')")?, + 1 + ); + } + remove_rollback_test_db(&path); + Ok(()) +} + fn scalar(db: *mut sqlite::sqlite3, sql: &str) -> Result { let stmt = db.prepare_v2(sql)?; assert_eq!(stmt.step()?, ResultCode::ROW, "{sql}"); @@ -698,6 +726,8 @@ fn test_failed_statement_then_commit_does_not_reuse_version() -> Result<(), Resu } pub fn run_suite() -> Result<(), String> { + test_refresh_retries_after_reload_failure() + .map_err(|e| format!("test_refresh_retries_after_reload_failure failed: {e:?}"))?; test_connection_state_refresh().map_err(|e| format!("test_connection_state_refresh failed: {:?}", e))?; test_fetch_db_version_from_storage()?; test_next_db_version()?; diff --git a/core/src/version-guard.c b/core/src/version-guard.c index 959d1322e..6cbb6eca9 100644 --- a/core/src/version-guard.c +++ b/core/src/version-guard.c @@ -80,11 +80,7 @@ static int versionGuardRowid(sqlite3_vtab_cursor *cursor, sqlite3_int64 *rowid) return SQLITE_OK; } -static int versionGuardBegin(sqlite3_vtab *vtab) { - crsql_VersionGuard *guard = (crsql_VersionGuard *)vtab; - guard->extData->versionGuardEnlisted = 1; - return SQLITE_OK; -} +static int versionGuardBegin(sqlite3_vtab *vtab) { return SQLITE_OK; } static int versionGuardUpdate(sqlite3_vtab *vtab, int argc, sqlite3_value **argv, sqlite3_int64 *rowid) { @@ -96,6 +92,7 @@ static int versionGuardUpdate(sqlite3_vtab *vtab, int argc, static int versionGuardSync(sqlite3_vtab *vtab) { crsql_VersionGuard *guard = (crsql_VersionGuard *)vtab; crsql_ExtData *extData = guard->extData; + if (!extData->versionGuardEnlisted) return SQLITE_OK; sqlite3_int64 version = extData->pendingDbVersion; sqlite3_stmt *stmt = extData->pSetDbVersionStmt; int rc = sqlite3_bind_blob(stmt, 1, extData->siteId, SITE_ID_LEN, @@ -158,5 +155,8 @@ int crsql_enlist_db_version(sqlite3 *db, crsql_ExtData *extData) { int rc = sqlite3_step(stmt); int result = rc == SQLITE_DONE ? SQLITE_OK : rc; rc = sqlite3_reset(stmt); - return result == SQLITE_OK ? rc : result; + if (result != SQLITE_OK) return result; + if (rc != SQLITE_OK) return rc; + extData->versionGuardEnlisted = 1; + return SQLITE_OK; }