Persist pending db_version in transaction guard vtab xSync

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):

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::<crsql_ExtData>(),
-        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"
+    );
 }

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;
@@ -48,56 +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<i64, String> {
-    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<i64, String> {
-    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 });
     }
-
-    // update db_version in db if it changed
-    if ret != unsafe { (*ext_data).pendingDbVersion } {
-        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());
-            }
-            (*ext_data).pendingDbVersion = ret;
+    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(

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]