update next_db_version to return what will actually be the next db version after commit

This also allows us to stop  clearing the db version cache and should give us quicker TPS.

Author: tantaman

core/rs/core/src/c.rs

@@ -75,6 +75,7 @@ pub struct crsql_ExtData {
     pub pPragmaDataVersionStmt: *mut sqlite::stmt,
     pub pragmaDataVersion: ::core::ffi::c_int,
     pub dbVersion: sqlite::int64,
+    pub pendingDbVersion: sqlite::int64,
     pub pragmaSchemaVersion: ::core::ffi::c_int,
     pub pragmaSchemaVersionForTableInfos: ::core::ffi::c_int,
     pub siteId: *mut ::core::ffi::c_uchar,
@@ -448,7 +449,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     let ptr = UNINIT.as_ptr();
     assert_eq!(
         ::std::mem::size_of::<crsql_ExtData>(),
-        120usize,
+        128usize,
         concat!("Size of: ", stringify!(crsql_ExtData))
     );
     assert_eq!(
@@ -497,8 +498,18 @@ fn bindgen_test_layout_crsql_ExtData() {
         )
     );
     assert_eq!(
-        unsafe { ::std::ptr::addr_of!((*ptr).pragmaSchemaVersion) as usize - ptr as usize },
+        unsafe { ::std::ptr::addr_of!((*ptr).pendingDbVersion) as usize - ptr as usize },
         32usize,
+        concat!(
+            "Offset of field: ",
+            stringify!(crsql_ExtData),
+            "::",
+            stringify!(pendingDbVersion)
+        )
+    );
+    assert_eq!(
+        unsafe { ::std::ptr::addr_of!((*ptr).pragmaSchemaVersion) as usize - ptr as usize },
+        40usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -510,7 +521,7 @@ fn bindgen_test_layout_crsql_ExtData() {
         unsafe {
             ::std::ptr::addr_of!((*ptr).pragmaSchemaVersionForTableInfos) as usize - ptr as usize
         },
-        36usize,
+        44usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -520,7 +531,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     );
     assert_eq!(
         unsafe { ::std::ptr::addr_of!((*ptr).siteId) as usize - ptr as usize },
-        40usize,
+        48usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -530,7 +541,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     );
     assert_eq!(
         unsafe { ::std::ptr::addr_of!((*ptr).pDbVersionStmt) as usize - ptr as usize },
-        48usize,
+        56usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -540,7 +551,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     );
     assert_eq!(
         unsafe { ::std::ptr::addr_of!((*ptr).zpTableInfos) as usize - ptr as usize },
-        56usize,
+        64usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -550,7 +561,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     );
     assert_eq!(
         unsafe { ::std::ptr::addr_of!((*ptr).tableInfosLen) as usize - ptr as usize },
-        64usize,
+        72usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -560,7 +571,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     );
     assert_eq!(
         unsafe { ::std::ptr::addr_of!((*ptr).rowsImpacted) as usize - ptr as usize },
-        68usize,
+        76usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -570,7 +581,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     );
     assert_eq!(
         unsafe { ::std::ptr::addr_of!((*ptr).seq) as usize - ptr as usize },
-        72usize,
+        80usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -580,7 +591,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     );
     assert_eq!(
         unsafe { ::std::ptr::addr_of!((*ptr).pSetSyncBitStmt) as usize - ptr as usize },
-        80usize,
+        88usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -590,7 +601,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     );
     assert_eq!(
         unsafe { ::std::ptr::addr_of!((*ptr).pClearSyncBitStmt) as usize - ptr as usize },
-        88usize,
+        96usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -600,7 +611,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     );
     assert_eq!(
         unsafe { ::std::ptr::addr_of!((*ptr).pSetSiteIdOrdinalStmt) as usize - ptr as usize },
-        96usize,
+        104usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -610,7 +621,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     );
     assert_eq!(
         unsafe { ::std::ptr::addr_of!((*ptr).pSelectSiteIdOrdinalStmt) as usize - ptr as usize },
-        104usize,
+        112usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),
@@ -620,7 +631,7 @@ fn bindgen_test_layout_crsql_ExtData() {
     );
     assert_eq!(
         unsafe { ::std::ptr::addr_of!((*ptr).pStmtCache) as usize - ptr as usize },
-        112usize,
+        120usize,
         concat!(
             "Offset of field: ",
             stringify!(crsql_ExtData),

core/rs/core/src/changes_vtab_write.rs

@@ -237,7 +237,7 @@ fn set_winner_clock(
               {pk_bind_list},
               ?,
               ?,
-              MAX(crsql_next_db_version(), ?),
+              crsql_next_db_version(?),
               ?,
               ?
             ) RETURNING _rowid_",
@@ -373,7 +373,7 @@ fn zero_clocks_on_resurrect(
     let stmt_key = get_cache_key(CachedStmtType::ZeroClocksOnResurrect, table_name, None)?;
     let zero_stmt = get_cached_stmt_rt_wt(db, ext_data, stmt_key, || {
         Ok(format!(
-            "UPDATE \"{table_name}__crsql_clock\" SET __crsql_col_version = 0, __crsql_db_version = MAX(crsql_next_db_version(), ?) WHERE {pk_where_list} AND __crsql_col_name IS NOT '{sentinel}'",
+            "UPDATE \"{table_name}__crsql_clock\" SET __crsql_col_version = 0, __crsql_db_version = crsql_next_db_version(?) WHERE {pk_where_list} AND __crsql_col_name IS NOT '{sentinel}'",
             table_name = crate::util::escape_ident(table_name),
             pk_where_list = pk_where_list_from_tbl_info(tbl_info, None)?,
             sentinel = crate::c::INSERT_SENTINEL

core/src/crsqlite.c

@@ -72,14 +72,33 @@ static void nextDbVersionFunc(sqlite3_context *context, int argc,
   char *errmsg = 0;
   crsql_ExtData *pExtData = (crsql_ExtData *)sqlite3_user_data(context);
   sqlite3 *db = sqlite3_context_db_handle(context);
+  // "getDbVersion" is really just filling the cached db version value if
+  // invalid
   int rc = crsql_getDbVersion(db, pExtData, &errmsg);
   if (rc != SQLITE_OK) {
     sqlite3_result_error(context, errmsg, -1);
     sqlite3_free(errmsg);
     return;
   }
 
-  sqlite3_result_int64(context, pExtData->dbVersion + 1);
+  sqlite3_int64 providedVersion = 0;
+  if (argc == 1) {
+    providedVersion = sqlite3_value_int64(argv[0]);
+  }
+
+  // now return max of:
+  // dbVersion + 1, pendingDbVersion, arg (if there is one)
+  // and set pendingDbVersion to that max
+  sqlite3_int64 ret = pExtData->dbVersion + 1;
+  if (ret < pExtData->pendingDbVersion) {
+    ret = pExtData->pendingDbVersion;
+  }
+  if (ret < providedVersion) {
+    ret = providedVersion;
+  }
+  pExtData->pendingDbVersion = ret;
+
+  sqlite3_result_int64(context, ret);
 }
 
 static void incrementAndGetSeqFunc(sqlite3_context *context, int argc,
@@ -320,15 +339,16 @@ static void crsqlRowsImpacted(sqlite3_context *context, int argc,
 static int commitHook(void *pUserData) {
   crsql_ExtData *pExtData = (crsql_ExtData *)pUserData;
 
-  pExtData->dbVersion = -1;
+  pExtData->pendingDbVersion = -1;
   pExtData->seq = 0;
   return SQLITE_OK;
 }
 
 static void rollbackHook(void *pUserData) {
   crsql_ExtData *pExtData = (crsql_ExtData *)pUserData;
 
-  pExtData->dbVersion = -1;
+  pExtData->pendingDbVersion = -1;
+  pExtData->seq = 0;
 }
 
 int sqlite3_crsqlrustbundle_init(sqlite3 *db, char **pzErrMsg,
@@ -400,7 +420,7 @@ __declspec(dllexport)
                                     dbVersionFunc, 0, 0, freeConnectionExtData);
   }
   if (rc == SQLITE_OK) {
-    rc = sqlite3_create_function(db, "crsql_next_db_version", 0,
+    rc = sqlite3_create_function(db, "crsql_next_db_version", -1,
                                  // dbversion can change on each invocation.
                                  SQLITE_UTF8 | SQLITE_INNOCUOUS, pExtData,
                                  nextDbVersionFunc, 0, 0);

core/src/ext-data.c

@@ -36,6 +36,7 @@ crsql_ExtData *crsql_newExtData(sqlite3 *db, unsigned char *siteIdBuffer) {
       SQLITE_PREPARE_PERSISTENT, &(pExtData->pSelectSiteIdOrdinalStmt), 0);
 
   pExtData->dbVersion = -1;
+  pExtData->pendingDbVersion = -1;
   pExtData->seq = 0;
   pExtData->pragmaSchemaVersion = -1;
   pExtData->pragmaDataVersion = -1;
@@ -254,11 +255,12 @@ int crsql_fetchDbVersionFromStorage(sqlite3 *db, crsql_ExtData *pExtData,
 int crsql_getDbVersion(sqlite3 *db, crsql_ExtData *pExtData, char **errmsg) {
   int rc = SQLITE_OK;
 
-  // version is cached. We clear the cached version
-  // at the end of each transaction so it is safe to return this
-  // without checking the schema version.
-  // It is an error to use crsqlite in such a way that you modify
-  // a schema and fetch changes in the same transaction.
+  // Version is cached. We update the cached version at the end of every
+  // transaction to match what is written to disk. The cache is only invalid if
+  // another connection also made writes. We detect this via `pragmaDataVersion`
+  // and force a re-fetch of the db version when `pragmaDataVersion` changes. It
+  // is an error to use crsqlite in such a way that you modify a schema and
+  // fetch changes in the same transaction.
   rc = crsql_fetchPragmaDataVersion(db, pExtData);
   if (rc == -1) {
     *errmsg = sqlite3_mprintf("failed to fetch PRAGMA data_version");

core/src/ext-data.h

@@ -18,6 +18,9 @@ struct crsql_ExtData {
   // to crsql_next_db_version()
   // and re-set on transaction commit or rollback.
   sqlite3_int64 dbVersion;
+  // the version that the db will be set to at the end of the transaction
+  // if that transaction were to commit at the time this value is checked.
+  sqlite3_int64 pendingDbVersion;
   int pragmaSchemaVersion;
 
   // we need another schema version number that tracks when we checked it