Disallow local writes during sync

Author: simolus3

crates/core/src/crud_vtab.rs

@@ -2,9 +2,10 @@ extern crate alloc;
 
 use alloc::boxed::Box;
 use alloc::string::String;
+use alloc::sync::Arc;
 use const_format::formatcp;
 use core::ffi::{c_char, c_int, c_void, CStr};
-use core::ptr::null_mut;
+use core::sync::atomic::Ordering;
 use serde::Serialize;
 use serde_json::value::RawValue;
 
@@ -15,6 +16,7 @@ use sqlite_nostd::{self as sqlite, ColumnType};
 use crate::error::SQLiteError;
 use crate::ext::SafeManagedStmt;
 use crate::schema::TableInfoFlags;
+use crate::state::DatabaseState;
 use crate::util::MAX_OP_ID;
 use crate::vtab_util::*;
 
@@ -41,6 +43,7 @@ struct VirtualTable {
     db: *mut sqlite::sqlite3,
     current_tx: Option<ActiveCrudTransaction>,
     is_simple: bool,
+    state: Arc<DatabaseState>,
 }
 
 struct ActiveCrudTransaction {
@@ -79,6 +82,15 @@ impl VirtualTable {
             .as_ref()
             .ok_or_else(|| SQLiteError(ResultCode::MISUSE, Some(String::from("No tx_id"))))?;
 
+        if self.state.is_in_sync_local.load(Ordering::Relaxed) {
+            // Don't collect CRUD writes while we're syncing the local database - writes made here
+            // aren't writes we should upload.
+            // This normally doesn't happen because we insert directly into the data tables, but
+            // users might have custom raw tables used for sycing with triggers on them to call
+            // this function. And those specifically should not trigger here.
+            return Ok(());
+        }
+
         match &current_tx.mode {
             CrudTransactionMode::Manual { stmt } => {
                 // Columns are (data TEXT, options INT HIDDEN)
@@ -213,7 +225,7 @@ SELECT * FROM insertion WHERE (NOT (?3 & {})) OR data->>'op' != 'PATCH' OR data-
 
 extern "C" fn connect(
     db: *mut sqlite::sqlite3,
-    _aux: *mut c_void,
+    aux: *mut c_void,
     argc: c_int,
     argv: *const *const c_char,
     vtab: *mut *mut sqlite::vtab,
@@ -244,6 +256,14 @@ extern "C" fn connect(
                 pModule: core::ptr::null(),
                 zErrMsg: core::ptr::null_mut(),
             },
+            state: {
+                // Increase refcount - we can't use from_raw alone because we don't own the aux
+                // data (connect could be called multiple times).
+                let state = Arc::from_raw(aux as *mut DatabaseState);
+                let clone = state.clone();
+                core::mem::forget(state);
+                clone
+            },
             db,
             current_tx: None,
             is_simple,
@@ -335,20 +355,20 @@ static MODULE: sqlite_nostd::module = sqlite_nostd::module {
     xIntegrity: None,
 };
 
-pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
+pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(), ResultCode> {
     sqlite::convert_rc(sqlite::create_module_v2(
         db,
         SIMPLE_NAME.as_ptr(),
         &MODULE,
-        null_mut(),
-        None,
+        Arc::into_raw(state.clone()) as *mut c_void,
+        Some(DatabaseState::destroy_arc),
     ))?;
     sqlite::convert_rc(sqlite::create_module_v2(
         db,
         MANUAL_NAME.as_ptr(),
         &MODULE,
-        null_mut(),
-        None,
+        Arc::into_raw(state) as *mut c_void,
+        Some(DatabaseState::destroy_arc),
     ))?;
 
     Ok(())

crates/core/src/lib.rs

@@ -73,7 +73,7 @@ fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
 
     crate::schema::register(db)?;
     crate::operations_vtab::register(db, state.clone())?;
-    crate::crud_vtab::register(db)?;
+    crate::crud_vtab::register(db, state)?;
 
     Ok(())
 }

crates/core/src/sync/storage_adapter.rs

@@ -1,6 +1,6 @@
 use core::{assert_matches::debug_assert_matches, fmt::Display};
 
-use alloc::{string::ToString, sync::Arc, vec::Vec};
+use alloc::{string::ToString, vec::Vec};
 use serde::Serialize;
 use sqlite_nostd::{self as sqlite, Connection, ManagedStmt, ResultCode};
 use streaming_iterator::StreamingIterator;

dart/test/sync_test.dart

@@ -780,6 +780,50 @@ END;
 
       expect(db.select('SELECT * FROM users'), isEmpty);
     });
+
+    test("can't use crud vtab during sync", () {
+      db.execute(
+          'CREATE TABLE users (id TEXT NOT NULL PRIMARY KEY, name TEXT NOT NULL) STRICT;');
+
+      invokeControl(
+        'start',
+        json.encode({
+          'schema': {
+            'raw_tables': [
+              {
+                'name': 'users',
+                'put': {
+                  // Inserting into powersync_crud_ during a sync operation is
+                  // forbidden, that vtab should only collect local writes.
+                  'sql': "INSERT INTO powersync_crud_(data) VALUES (?);",
+                  'params': [
+                    {'Column': 'name'}
+                  ],
+                },
+                'delete': {
+                  'sql': 'DELETE FROM users WHERE id = ?',
+                  'params': ['Id'],
+                },
+              }
+            ],
+            'tables': [],
+          },
+        }),
+      );
+
+      // Insert
+      pushCheckpoint(buckets: [bucketDescription('a')]);
+      pushSyncData(
+        'a',
+        '1',
+        'my_user',
+        'PUT',
+        {'name': 'First user'},
+        objectType: 'users',
+      );
+
+      expect(pushCheckpointComplete, throwsA(isA<SqliteException>()));
+    });
   });
 }