Disallow local writes during sync

Author: simolus3

crates/core/src/crud_vtab.rs

@@ -2,8 +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};
+use core::sync::atomic::Ordering;
 
 use sqlite::{Connection, ResultCode, Value};
 use sqlite_nostd as sqlite;
@@ -13,6 +15,7 @@ use sqlite_nostd::ResultCode::NULL;
 use crate::error::SQLiteError;
 use crate::ext::SafeManagedStmt;
 use crate::schema::TableInfoFlags;
+use crate::state::DatabaseState;
 use crate::vtab_util::*;
 
 // Structure:
@@ -31,11 +34,12 @@ struct VirtualTable {
     db: *mut sqlite::sqlite3,
     current_tx: Option<i64>,
     insert_statement: Option<ManagedStmt>,
+    state: Arc<DatabaseState>,
 }
 
 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,
@@ -58,6 +62,14 @@ extern "C" fn connect(
             db,
             current_tx: None,
             insert_statement: None,
+            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
+            },
         }));
         *vtab = tab.cast::<sqlite::vtab>();
         let _ = sqlite::vtab_config(db, 0);
@@ -127,13 +139,20 @@ fn insert_operation(
     flags: TableInfoFlags,
 ) -> Result<(), SQLiteError> {
     let tab = unsafe { &mut *(vtab.cast::<VirtualTable>()) };
-    if tab.current_tx.is_none() {
+    if tab.state.is_in_sync_local.load(Ordering::Relaxed) {
         return Err(SQLiteError(
             ResultCode::MISUSE,
-            Some(String::from("No tx_id")),
+            Some(String::from("Using ps_crud during sync operation")),
         ));
     }
-    let current_tx = tab.current_tx.unwrap();
+
+    let Some(current_tx) = tab.current_tx else {
+        return Err(SQLiteError(
+            ResultCode::MISUSE,
+            Some(String::from("No tx_id")),
+        ));
+    };
+
     // language=SQLite
     let statement = tab
         .insert_statement
@@ -206,8 +225,13 @@ static MODULE: sqlite_nostd::module = sqlite_nostd::module {
     xIntegrity: None,
 };
 
-pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
-    db.create_module_v2("powersync_crud_", &MODULE, None, None)?;
+pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(), ResultCode> {
+    db.create_module_v2(
+        "powersync_crud_",
+        &MODULE,
+        Some(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>()));
+    });
   });
 }