Refresh all connection-local state on PRAGMA data_version changes

core/rs/core/src/alter.rs

@@ -11,7 +11,7 @@ use num_traits::FromPrimitive;
 use sqlite_nostd::{sqlite3, Connection, ResultCode, StrRef};
 
 use crate::c::crsql_ExtData;
-use crate::db_version::fill_db_version_if_needed;
+use crate::ext_data::refresh_connection_state_if_needed;
 use crate::tableinfo::{crsql_ensure_table_infos_are_up_to_date, TableInfo};
 
 #[no_mangle]
@@ -33,7 +33,7 @@ unsafe fn compact_post_alter(
     errmsg: *mut *mut c_char,
 ) -> Result<ResultCode, ResultCode> {
     let tbl_name_str = CStr::from_ptr(tbl_name).to_str()?;
-    fill_db_version_if_needed(db, ext_data).or_else(|msg| {
+    refresh_connection_state_if_needed(db, ext_data).or_else(|msg| {
         errmsg.set(&msg);
         Err(ResultCode::ERROR)
     })?;

core/rs/core/src/changes_vtab_write.rs

@@ -431,6 +431,12 @@ unsafe fn merge_insert(
     let tab = vtab.cast::<crsql_Changes_vtab>();
     let db = (*tab).db;
 
+    if let Err(msg) = crate::ext_data::refresh_connection_state_if_needed(db, (*tab).pExtData) {
+        let err = CString::new(format!("failed to refresh connection state: {}", msg))?;
+        *errmsg = err.into_raw();
+        return Err(ResultCode::ERROR);
+    }
+
     let rc = crsql_ensure_table_infos_are_up_to_date(db, (*tab).pExtData, errmsg);
     if rc != ResultCode::OK as i32 {
         let err = CString::new("Failed to update CRR table information")?;

core/rs/core/src/config.rs

@@ -2,12 +2,33 @@ use alloc::format;
 
 use sqlite::{Connection, Context};
 use sqlite_nostd as sqlite;
-use sqlite_nostd::{ManagedStmt, ResultCode, Value};
+use sqlite_nostd::{ColumnType, ManagedStmt, ResultCode, Value};
 
 use crate::c::crsql_ExtData;
 
 pub const MERGE_EQUAL_VALUES: &str = "merge-equal-values";
 
+pub fn reload_connection_config(
+    db: *mut sqlite::sqlite3,
+    ext_data: *mut crsql_ExtData,
+) -> Result<(), alloc::string::String> {
+    let stmt = db
+        .prepare_v2("SELECT value FROM crsql_master WHERE key = 'config.merge-equal-values'")
+        .map_err(|rc| format!("failed to prepare config refresh: {rc}"))?;
+    let value = match stmt.step() {
+        Ok(ResultCode::ROW) => {
+            if stmt.column_type(0) != Ok(ColumnType::Integer) {
+                return Err("invalid merge-equal-values setting".into());
+            }
+            stmt.column_int(0)
+        }
+        Ok(ResultCode::DONE) => 0,
+        Ok(rc) | Err(rc) => return Err(format!("failed to read config: {rc}")),
+    };
+    unsafe { (*ext_data).mergeEqualValues = value };
+    Ok(())
+}
+
 pub extern "C" fn crsql_config_set(
     ctx: *mut sqlite::context,
     argc: i32,
@@ -73,10 +94,14 @@ pub extern "C" fn crsql_config_get(
     let args = sqlite::args!(argc, argv);
 
     let name = args[0].text();
+    let ext_data = ctx.user_data() as *mut crsql_ExtData;
+    if let Err(message) = crate::ext_data::refresh_connection_state_if_needed(ctx.db_handle(), ext_data) {
+        ctx.result_error(&message);
+        return;
+    }
 
     match name {
         MERGE_EQUAL_VALUES => {
-            let ext_data = ctx.user_data() as *mut crsql_ExtData;
             ctx.result_int(unsafe { (*ext_data).mergeEqualValues });
         }
         _ => {

core/rs/core/src/db_version.rs

@@ -1,7 +1,6 @@
 use core::ffi::c_void;
 use core::mem;
 
-use crate::alloc::string::ToString;
 use crate::alloc::{boxed::Box, vec::Vec};
 use alloc::collections::btree_map::Entry;
 use alloc::collections::BTreeMap;
@@ -14,26 +13,10 @@ use sqlite::{sqlite3, Stmt};
 use sqlite_nostd as sqlite;
 
 use crate::c::crsql_ExtData;
-use crate::c::crsql_fetchPragmaDataVersion;
 use crate::consts::MIN_POSSIBLE_DB_VERSION;
 use crate::consts::SITE_ID_LEN;
 use crate::stmt_cache::reset_cached_stmt;
 
-#[no_mangle]
-pub extern "C" fn crsql_fill_db_version_if_needed(
-    db: *mut sqlite3,
-    ext_data: *mut crsql_ExtData,
-    errmsg: *mut *mut c_char,
-) -> c_int {
-    match fill_db_version_if_needed(db, ext_data) {
-        Ok(rc) => rc as c_int,
-        Err(msg) => {
-            errmsg.set(&msg);
-            ResultCode::ERROR as c_int
-        }
-    }
-}
-
 #[no_mangle]
 pub extern "C" fn crsql_next_db_version(
     db: *mut sqlite3,
@@ -65,7 +48,7 @@ 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> {
-    fill_db_version_if_needed(db, ext_data)?;
+    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 } {
@@ -80,7 +63,7 @@ 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<i64, String> {
-    fill_db_version_if_needed(db, ext_data)?;
+    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 } {
@@ -117,23 +100,7 @@ pub fn next_db_version(db: *mut sqlite3, ext_data: *mut crsql_ExtData) -> Result
     Ok(ret)
 }
 
-pub fn fill_db_version_if_needed(
-    db: *mut sqlite3,
-    ext_data: *mut crsql_ExtData,
-) -> Result<ResultCode, String> {
-    unsafe {
-        let rc = crsql_fetchPragmaDataVersion(db, ext_data);
-        if rc == -1 {
-            return Err("failed to fetch PRAGMA data_version".to_string());
-        }
-        if (*ext_data).dbVersion != -1 && rc == 0 {
-            return Ok(ResultCode::OK);
-        }
-        fetch_db_version_from_storage(db, ext_data)
-    }
-}
-
-pub fn fetch_db_version_from_storage(
+pub fn reload_db_version(
     _db: *mut sqlite3,
     ext_data: *mut crsql_ExtData,
 ) -> Result<ResultCode, String> {

core/rs/core/src/ext_data.rs

@@ -4,10 +4,40 @@ use alloc::vec;
 use core::ffi::c_int;
 use core::ptr::null_mut;
 
+use alloc::string::String;
 use sqlite::{sqlite3, Connection, ResultCode, Stmt};
 use sqlite_nostd as sqlite;
 
-use crate::{c::crsql_ExtData, util::get_db_version_union_query};
+use crate::{
+    c::{crsql_ExtData, crsql_fetchPragmaDataVersion},
+    util::get_db_version_union_query,
+};
+
+/**
+ * Refresh connection-local state cached in `crsql_ExtData` after another
+ * connection committed changes to the database.
+ *
+ * `PRAGMA data_version` only changes for commits made by other connections
+ * and is pinned while this connection holds a transaction open, so this is a
+ * cheap no-op in the common case. When the data version did change, every
+ * cached value is reloaded through its own reload_* function so independent
+ * caches cannot observe the same change inconsistently.
+ */
+pub fn refresh_connection_state_if_needed(
+    db: *mut sqlite3,
+    ext_data: *mut crsql_ExtData,
+) -> Result<(), String> {
+    let rc = unsafe { crsql_fetchPragmaDataVersion(db, ext_data) };
+    if rc == -1 {
+        return Err("failed to fetch PRAGMA data_version".to_string());
+    }
+    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(())
+}
 
 #[no_mangle]
 pub extern "C" fn crsql_recreate_db_version_stmt(

core/rs/core/src/lib.rs

@@ -62,10 +62,7 @@ use c::{crsql_freeExtData, crsql_initSiteIdExt, crsql_newExtData};
 use config::{crsql_config_get, crsql_config_set};
 use core::ffi::{c_int, c_void, CStr};
 use create_crr::create_crr;
-use db_version::{
-    crsql_fill_db_version_if_needed, crsql_next_db_version, crsql_peek_next_db_version,
-    insert_db_version,
-};
+use db_version::{crsql_next_db_version, crsql_peek_next_db_version, insert_db_version};
 use is_crr::*;
 use local_writes::after_delete::x_crsql_after_delete;
 use local_writes::after_insert::x_crsql_after_insert;
@@ -972,6 +969,12 @@ unsafe extern "C" fn x_crsql_set_ts(
         }
     };
     let ext_data = ctx.user_data() as *mut c::crsql_ExtData;
+    if let Err(message) =
+        ext_data::refresh_connection_state_if_needed(ctx.db_handle(), ext_data)
+    {
+        ctx.result_error(&message);
+        return;
+    }
     (*ext_data).timestamp = ts_u64;
     ctx.result_text_static("OK");
 }
@@ -1091,11 +1094,8 @@ unsafe extern "C" fn x_crsql_db_version(
 ) {
     let ext_data = ctx.user_data() as *mut c::crsql_ExtData;
     let db = ctx.db_handle();
-    let mut err_msg = null_mut();
-    let rc = crsql_fill_db_version_if_needed(db, ext_data, &mut err_msg as *mut _);
-    if rc != ResultCode::OK as c_int {
-        // TODO: pass err_msg!
-        ctx.result_error("failed to fill db version");
+    if let Err(message) = ext_data::refresh_connection_state_if_needed(db, ext_data) {
+        ctx.result_error(&message);
         return;
     }
     sqlite::result_int64(ctx, (*ext_data).dbVersion);

core/rs/integration_check/src/t/test_db_version.rs

@@ -35,13 +35,13 @@ fn test_fetch_db_version_from_storage() -> Result<ResultCode, String> {
     let rc = unsafe { test_exports::c::crsql_initSiteIdExt(raw_db, ext_data, site_id) };
     assert_eq!(rc, 0);
 
-    test_exports::db_version::fetch_db_version_from_storage(raw_db, ext_data)?;
+    test_exports::db_version::reload_db_version(raw_db, ext_data)?;
     // no clock tables, no version.
     assert_eq!(0, unsafe { (*ext_data).dbVersion });
 
     // this was a bug where calling twice on a fresh db would fail the second
     // time.
-    test_exports::db_version::fetch_db_version_from_storage(raw_db, ext_data)?;
+    test_exports::db_version::reload_db_version(raw_db, ext_data)?;
     // should still return same data on a subsequent call with no schema
     assert_eq!(0, unsafe { (*ext_data).dbVersion });
 
@@ -50,14 +50,14 @@ fn test_fetch_db_version_from_storage() -> Result<ResultCode, String> {
         .expect("made foo");
     db.exec_safe("SELECT crsql_as_crr('foo');")
         .expect("made foo crr");
-    test_exports::db_version::fetch_db_version_from_storage(raw_db, ext_data)?;
+    test_exports::db_version::reload_db_version(raw_db, ext_data)?;
     // still v0 since no rows are inserted
     assert_eq!(0, unsafe { (*ext_data).dbVersion });
 
     // version is bumped due to insert
     db.exec_safe("INSERT INTO foo (a, b) VALUES (1, 2);")
         .expect("inserted");
-    test_exports::db_version::fetch_db_version_from_storage(raw_db, ext_data)?;
+    test_exports::db_version::reload_db_version(raw_db, ext_data)?;
     assert_eq!(1, unsafe { (*ext_data).dbVersion });
 
     db.exec_safe("CREATE TABLE bar (a primary key not null, b);")
@@ -67,10 +67,10 @@ fn test_fetch_db_version_from_storage() -> Result<ResultCode, String> {
     db.exec_safe("INSERT INTO bar VALUES (1, 2)")
         .expect("inserted into bar");
 
-    test_exports::db_version::fetch_db_version_from_storage(raw_db, ext_data)?;
+    test_exports::db_version::reload_db_version(raw_db, ext_data)?;
     assert_eq!(2, unsafe { (*ext_data).dbVersion });
 
-    test_exports::db_version::fetch_db_version_from_storage(raw_db, ext_data)?;
+    test_exports::db_version::reload_db_version(raw_db, ext_data)?;
     assert_eq!(2, unsafe { (*ext_data).dbVersion });
 
     unsafe {
@@ -447,7 +447,57 @@ fn test_insert_db_version_cache() -> Result<(), ResultCode> {
     Ok(())
 }
 
+fn scalar(db: &sqlite::ManagedConnection, sql: &str) -> Result<i64, String> {
+    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()
+            ))
+        }
+    }
+    Ok(stmt.column_int64(0))
+}
+
+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:?}"))?;
+    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);
+    Ok(())
+}
+
 pub fn run_suite() -> Result<(), String> {
+    test_connection_state_refresh().map_err(|e| format!("test_connection_state_refresh failed: {:?}", e))?;
     test_fetch_db_version_from_storage()?;
     test_next_db_version()?;
     test_get_or_set_site_ordinal()

core/src/rust.h

@@ -21,8 +21,6 @@ int crsql_create_crr(sqlite3 *db, const char *schemaName, const char *tblName,
 int crsql_ensure_table_infos_are_up_to_date(sqlite3 *db,
                                             crsql_ExtData *pExtData,
                                             char **err);
-int crsql_fill_db_version_if_needed(sqlite3 *db, crsql_ExtData *pExtData,
-                                    char **errmsg);
 sqlite_int64 crsql_next_db_version(sqlite3 *db, crsql_ExtData *pExtData,
                                    sqlite3_int64 mergingVersion, char **errmsg);
 sqlite_int64 crsql_peek_next_db_version(sqlite3 *db, crsql_ExtData *pExtData,