Add function to determine whether local sync is active

Author: simolus3

crates/core/src/lib.rs

@@ -9,9 +9,12 @@ extern crate alloc;
 
 use core::ffi::{c_char, c_int};
 
+use alloc::sync::Arc;
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 
+use crate::state::DatabaseState;
+
 mod bson;
 mod checkpoint;
 mod crud_vtab;
@@ -26,6 +29,7 @@ mod migrations;
 mod operations;
 mod operations_vtab;
 mod schema;
+mod state;
 mod sync;
 mod sync_local;
 mod util;
@@ -53,6 +57,8 @@ pub extern "C" fn sqlite3_powersync_init(
 }
 
 fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
+    let state = Arc::new(DatabaseState::new());
+
     crate::version::register(db)?;
     crate::views::register(db)?;
     crate::uuid::register(db)?;
@@ -62,10 +68,11 @@ fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
     crate::view_admin::register(db)?;
     crate::checkpoint::register(db)?;
     crate::kv::register(db)?;
-    sync::register(db)?;
+    crate::state::register(db, state.clone())?;
+    sync::register(db, state.clone())?;
 
     crate::schema::register(db)?;
-    crate::operations_vtab::register(db)?;
+    crate::operations_vtab::register(db, state.clone())?;
     crate::crud_vtab::register(db)?;
 
     Ok(())

crates/core/src/operations_vtab.rs

@@ -1,6 +1,7 @@
 extern crate alloc;
 
 use alloc::boxed::Box;
+use alloc::sync::Arc;
 use core::ffi::{c_char, c_int, c_void};
 
 use sqlite::{Connection, ResultCode, Value};
@@ -9,21 +10,23 @@ use sqlite_nostd as sqlite;
 use crate::operations::{
     clear_remove_ops, delete_bucket, delete_pending_buckets, insert_operation,
 };
+use crate::state::DatabaseState;
 use crate::sync_local::sync_local;
 use crate::vtab_util::*;
 
 #[repr(C)]
 struct VirtualTable {
     base: sqlite::vtab,
     db: *mut sqlite::sqlite3,
+    state: Arc<DatabaseState>,
 
     target_applied: bool,
     target_validated: bool,
 }
 
 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,
@@ -43,6 +46,14 @@ extern "C" fn connect(
                 zErrMsg: core::ptr::null_mut(),
             },
             db,
+            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
+            },
             target_validated: false,
             target_applied: false,
         }));
@@ -83,7 +94,7 @@ extern "C" fn update(
             let result = insert_operation(db, args[3].text());
             vtab_result(vtab, result)
         } else if op == "sync_local" {
-            let result = sync_local(db, &args[3]);
+            let result = sync_local(&tab.state, db, &args[3]);
             if let Ok(result_row) = result {
                 unsafe {
                     *p_row_id = result_row;
@@ -139,8 +150,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_operations", &MODULE, None, None)?;
+pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(), ResultCode> {
+    db.create_module_v2(
+        "powersync_operations",
+        &MODULE,
+        Some(Arc::into_raw(state) as *mut c_void),
+        Some(DatabaseState::destroy_arc),
+    )?;
 
     Ok(())
 }

crates/core/src/state.rs

@@ -0,0 +1,74 @@
+use core::{
+    ffi::{c_int, c_void},
+    sync::atomic::{AtomicBool, Ordering},
+};
+
+use alloc::sync::Arc;
+use sqlite::{Connection, ResultCode};
+use sqlite_nostd::{self as sqlite, Context};
+
+/// State that is shared for a SQLite database connection after the core extension has been
+/// registered on it.
+///
+/// `init_extension` allocates an instance of this in an `Arc` that is shared as user-data for
+/// functions/vtabs that need access to it.
+pub struct DatabaseState {
+    pub is_in_sync_local: AtomicBool,
+}
+
+impl DatabaseState {
+    pub fn new() -> Self {
+        DatabaseState {
+            is_in_sync_local: AtomicBool::new(false),
+        }
+    }
+
+    pub fn sync_local_guard<'a>(&'a self) -> impl Drop + use<'a> {
+        self.is_in_sync_local
+            .compare_exchange(false, true, Ordering::Acquire, Ordering::Acquire)
+            .expect("should not be syncing already");
+
+        struct ClearOnDrop<'a>(&'a DatabaseState);
+
+        impl Drop for ClearOnDrop<'_> {
+            fn drop(&mut self) {
+                self.0.is_in_sync_local.store(false, Ordering::Release);
+            }
+        }
+
+        ClearOnDrop(self)
+    }
+
+    pub unsafe extern "C" fn destroy_arc(ptr: *mut c_void) {
+        drop(Arc::from_raw(ptr.cast::<DatabaseState>()));
+    }
+}
+
+pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(), ResultCode> {
+    unsafe extern "C" fn func(
+        ctx: *mut sqlite::context,
+        _argc: c_int,
+        _argv: *mut *mut sqlite::value,
+    ) {
+        let data = ctx.user_data().cast::<DatabaseState>();
+        let data = unsafe { data.as_ref() }.unwrap();
+
+        ctx.result_int(if data.is_in_sync_local.load(Ordering::Relaxed) {
+            1
+        } else {
+            0
+        });
+    }
+
+    db.create_function_v2(
+        "powersync_in_sync_operation",
+        0,
+        0,
+        Some(Arc::into_raw(state) as *mut c_void),
+        Some(func),
+        None,
+        None,
+        Some(DatabaseState::destroy_arc),
+    )?;
+    Ok(())
+}

crates/core/src/sync/interface.rs

@@ -5,6 +5,7 @@ use alloc::borrow::Cow;
 use alloc::boxed::Box;
 use alloc::rc::Rc;
 use alloc::string::ToString;
+use alloc::sync::Arc;
 use alloc::{string::String, vec::Vec};
 use serde::{Deserialize, Serialize};
 use sqlite::{ResultCode, Value};
@@ -13,6 +14,7 @@ use sqlite_nostd::{Connection, Context};
 
 use crate::error::SQLiteError;
 use crate::schema::Schema;
+use crate::state::DatabaseState;
 
 use super::streaming_sync::SyncClient;
 use super::sync_status::DownloadSyncStatus;
@@ -121,7 +123,7 @@ struct SqlController {
     client: SyncClient,
 }
 
-pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
+pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(), ResultCode> {
     extern "C" fn control(
         ctx: *mut sqlite::context,
         argc: c_int,
@@ -202,7 +204,7 @@ pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
     }
 
     let controller = Box::new(SqlController {
-        client: SyncClient::new(db),
+        client: SyncClient::new(db, state),
     });
 
     db.create_function_v2(

crates/core/src/sync/mod.rs

@@ -1,3 +1,4 @@
+use alloc::sync::Arc;
 use sqlite_nostd::{self as sqlite, ResultCode};
 
 mod bucket_priority;
@@ -13,6 +14,8 @@ mod sync_status;
 pub use bucket_priority::BucketPriority;
 pub use checksum::Checksum;
 
-pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
-    interface::register(db)
+use crate::state::DatabaseState;
+
+pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(), ResultCode> {
+    interface::register(db, state)
 }

crates/core/src/sync/storage_adapter.rs

@@ -1,6 +1,6 @@
 use core::{assert_matches::debug_assert_matches, fmt::Display};
 
-use alloc::{string::ToString, vec::Vec};
+use alloc::{string::ToString, sync::Arc, vec::Vec};
 use serde::Serialize;
 use sqlite_nostd::{self as sqlite, Connection, ManagedStmt, ResultCode};
 use streaming_iterator::StreamingIterator;
@@ -10,6 +10,7 @@ use crate::{
     ext::SafeManagedStmt,
     operations::delete_bucket,
     schema::Schema,
+    state::DatabaseState,
     sync::checkpoint::{validate_checkpoint, ChecksumMismatch},
     sync_local::{PartialSyncOperation, SyncOperation},
 };
@@ -144,6 +145,7 @@ impl StorageAdapter {
 
     pub fn sync_local(
         &self,
+        state: &DatabaseState,
         checkpoint: &OwnedCheckpoint,
         priority: Option<BucketPriority>,
         schema: &Schema,
@@ -185,7 +187,7 @@ impl StorageAdapter {
 
         let sync_result = match priority {
             None => {
-                let mut sync = SyncOperation::new(self.db, None);
+                let mut sync = SyncOperation::new(state, self.db, None);
                 sync.use_schema(schema);
                 sync.apply()
             }
@@ -208,6 +210,7 @@ impl StorageAdapter {
                 // TODO: Avoid this serialization, it's currently used to bind JSON SQL parameters.
                 let serialized_args = serde_json::to_string(&args)?;
                 let mut sync = SyncOperation::new(
+                    state,
                     self.db,
                     Some(PartialSyncOperation {
                         priority,

crates/core/src/sync/streaming_sync.rs

@@ -10,6 +10,7 @@ use alloc::{
     collections::{btree_map::BTreeMap, btree_set::BTreeSet},
     format,
     string::{String, ToString},
+    sync::Arc,
     vec::Vec,
 };
 use futures_lite::FutureExt;
@@ -18,6 +19,7 @@ use crate::{
     bson,
     error::SQLiteError,
     kv::client_id,
+    state::DatabaseState,
     sync::{checkpoint::OwnedBucketChecksum, interface::StartSyncStream},
 };
 use sqlite_nostd::{self as sqlite, ResultCode};
@@ -37,14 +39,16 @@ use super::{
 /// initialized.
 pub struct SyncClient {
     db: *mut sqlite::sqlite3,
+    db_state: Arc<DatabaseState>,
     /// The current [ClientState] (essentially an optional [StreamingSyncIteration]).
     state: ClientState,
 }
 
 impl SyncClient {
-    pub fn new(db: *mut sqlite::sqlite3) -> Self {
+    pub fn new(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Self {
         Self {
             db,
+            db_state: state,
             state: ClientState::Idle,
         }
     }
@@ -57,7 +61,7 @@ impl SyncClient {
             SyncControlRequest::StartSyncStream(options) => {
                 self.state.tear_down()?;
 
-                let mut handle = SyncIterationHandle::new(self.db, options)?;
+                let mut handle = SyncIterationHandle::new(self.db, options, self.db_state.clone())?;
                 let instructions = handle.initialize()?;
                 self.state = ClientState::IterationActive(handle);
 
@@ -125,10 +129,15 @@ struct SyncIterationHandle {
 impl SyncIterationHandle {
     /// Creates a new sync iteration in a pending state by preparing statements for
     /// [StorageAdapter] and setting up the initial downloading state for [StorageAdapter] .
-    fn new(db: *mut sqlite::sqlite3, options: StartSyncStream) -> Result<Self, ResultCode> {
+    fn new(
+        db: *mut sqlite::sqlite3,
+        options: StartSyncStream,
+        state: Arc<DatabaseState>,
+    ) -> Result<Self, ResultCode> {
         let runner = StreamingSyncIteration {
             db,
             options,
+            state,
             adapter: StorageAdapter::new(db)?,
             status: SyncStatusContainer::new(),
         };
@@ -192,6 +201,7 @@ impl<'a> ActiveEvent<'a> {
 
 struct StreamingSyncIteration {
     db: *mut sqlite::sqlite3,
+    state: Arc<DatabaseState>,
     adapter: StorageAdapter,
     options: StartSyncStream,
     status: SyncStatusContainer,
@@ -246,9 +256,12 @@ impl StreamingSyncIteration {
                 SyncEvent::BinaryLine { data } => bson::from_bytes(data)?,
                 SyncEvent::UploadFinished => {
                     if let Some(checkpoint) = validated_but_not_applied.take() {
-                        let result =
-                            self.adapter
-                                .sync_local(&checkpoint, None, &self.options.schema)?;
+                        let result = self.adapter.sync_local(
+                            &self.state,
+                            &checkpoint,
+                            None,
+                            &self.options.schema,
+                        )?;
 
                         match result {
                             SyncLocalResult::ChangesApplied => {
@@ -324,9 +337,9 @@ impl StreamingSyncIteration {
                             ),
                         ));
                     };
-                    let result = self
-                        .adapter
-                        .sync_local(target, None, &self.options.schema)?;
+                    let result =
+                        self.adapter
+                            .sync_local(&self.state, target, None, &self.options.schema)?;
 
                     match result {
                         SyncLocalResult::ChecksumFailure(checkpoint_result) => {
@@ -369,9 +382,12 @@ impl StreamingSyncIteration {
                             ),
                         ));
                     };
-                    let result =
-                        self.adapter
-                            .sync_local(target, Some(priority), &self.options.schema)?;
+                    let result = self.adapter.sync_local(
+                        &self.state,
+                        target,
+                        Some(priority),
+                        &self.options.schema,
+                    )?;
 
                     match result {
                         SyncLocalResult::ChecksumFailure(checkpoint_result) => {