Allow mocking timers in tests

CHANGELOG.md

@@ -1,3 +1,10 @@
+## 0.0.8 (unreleased)
+
+- __Breaking__: Timers on `PowerSyncEnvironment::custom` are now passed by value.
+- Don't mark sync status as connected when connection fails.
+- Fix sync client blocking writer for longer than necessary.
+- Set cache size and busy timeout on all connections instead of just the writer.
+
 ## 0.0.7
 
 - Update PowerSync core extension to version 0.5.2.

powersync/src/db/internal.rs

@@ -10,11 +10,9 @@ use crate::{
     util::SharedFuture,
 };
 use event_listener::EventListener;
-use futures_lite::future::yield_now;
 use futures_lite::{FutureExt, Stream, StreamExt, ready};
 use powersync_sqlite_nostd::{ColumnType, Destructor, ResultCode};
-use std::sync::{Mutex, Weak};
-use std::time::Duration;
+use std::sync::Weak;
 use std::{
     pin::Pin,
     sync::Arc,
@@ -38,7 +36,6 @@ pub struct InnerPowerSyncState {
     /// reference to [InnerPowerSyncState], we only keep a weak reference here to ensure we can drop
     /// actors through the channels owned by [SyncCoordinator].
     pub(crate) sync: Weak<SyncCoordinator>,
-    pub(crate) retry_delay: Mutex<Option<Duration>>,
 }
 
 impl InnerPowerSyncState {
@@ -53,7 +50,6 @@ impl InnerPowerSyncState {
             schema: Arc::new(schema),
             status: SyncStatus::new(),
             current_streams: SyncStreamTracker::default(),
-            retry_delay: Default::default(),
             sync: Arc::downgrade(sync),
         }
     }
@@ -156,21 +152,6 @@ impl InnerPowerSyncState {
         Ok(self.env.pool.writer().await)
     }
 
-    pub async fn sync_iteration_delay(&self) {
-        let delay = {
-            let guard = self.retry_delay.lock().unwrap();
-            *guard
-        };
-
-        if let Some(delay) = delay
-            && delay > Duration::ZERO
-        {
-            self.env.timer.delay_once(delay).await
-        } else {
-            yield_now().await
-        }
-    }
-
     pub fn watch_status<'a>(&'a self) -> impl Stream<Item = Arc<SyncStatusData>> + 'a {
         struct StreamImpl<'a> {
             db: &'a InnerPowerSyncState,

powersync/src/db/mod.rs

@@ -84,7 +84,7 @@ impl PowerSyncDatabase {
     /// Requests the download actor, started with [Self::download_actor], to start establishing a
     /// connection to the PowerSync service.
     pub async fn connect(&self, options: SyncOptions) {
-        self.sync.connect(options, &self.inner).await
+        self.sync.connect(options).await
     }
 
     /// If the sync client is currently connected, requests it to disconnect.

powersync/src/db/pool.rs

@@ -35,18 +35,23 @@ impl ConnectionPool {
             SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE,
         )?);
 
+        fn configure_common(connection: &SqliteConnection) -> Result<(), PowerSyncError> {
+            connection.exec(c"PRAGMA busy_timeout = 30000")?;
+            connection.exec(c"PRAGMA cache_size = -51200")?; // -(50 * 1024)
+            Ok(())
+        }
+
         writer.exec(c"PRAGMA journal_mode = WAL")?;
         writer.exec(c"PRAGMA journal_size_limit = 6291456")?; // 6 * 1024 * 1024
-        writer.exec(c"PRAGMA busy_timeout = 30000")?;
-        writer.exec(c"PRAGMA cache_size = -51200")?; // -(50 * 1024)
+        configure_common(&writer)?;
 
         let mut readers = vec![];
         for _ in 0..5 {
             let reader = SqliteConnection::from(RawSqliteConnection::open_path(
                 &path,
                 SQLITE_OPEN_READONLY,
             )?);
-            reader.exec(c"PRAGMA query_only = 1")?;
+            configure_common(&reader)?;
             readers.push(reader);
         }
 

powersync/src/env.rs

@@ -4,6 +4,7 @@ use crate::http::HttpClient;
 use num_traits::FromPrimitive;
 use powersync_core::powersync_init_static;
 use powersync_sqlite_nostd::ResultCode;
+use std::sync::Arc;
 use std::{pin::Pin, time::Duration};
 
 /// All external dependencies required for the PowerSync SDK.
@@ -17,19 +18,15 @@ pub struct PowerSyncEnvironment {
     /// The [ConnectionPool] used to obtain connections for queries asynchronously.
     pub(crate) pool: ConnectionPool,
     /// The [Timer] implementation used to delay sync iterations after errors.
-    pub(crate) timer: &'static (dyn Timer + Send + Sync),
+    pub(crate) timer: Arc<dyn Timer>,
 }
 
 impl PowerSyncEnvironment {
-    pub fn custom<C: HttpClient>(
-        client: C,
-        pool: ConnectionPool,
-        timer: &'static (dyn Timer + Send + Sync),
-    ) -> Self {
+    pub fn custom<C: HttpClient, T: Timer>(client: C, pool: ConnectionPool, timer: T) -> Self {
         Self {
             client: Box::new(client),
             pool,
-            timer,
+            timer: Arc::new(timer),
         }
     }
 
@@ -50,7 +47,7 @@ impl PowerSyncEnvironment {
 
     /// A [Timer] implementation based on [async_io::Timer].
     #[cfg(feature = "smol")]
-    pub fn async_io_timer() -> &'static (dyn Timer + Send + Sync) {
+    pub fn async_io_timer() -> impl Timer {
         use async_io::Timer as PlatformTimer;
 
         struct AsyncIoTimer;
@@ -64,12 +61,12 @@ impl PowerSyncEnvironment {
                 .boxed()
             }
         }
-        &AsyncIoTimer
+        AsyncIoTimer
     }
 
     /// A [Timer] implementation based on [tokio::time::sleep].
     #[cfg(feature = "tokio")]
-    pub fn tokio_timer() -> &'static (dyn Timer + Send + Sync) {
+    pub fn tokio_timer() -> impl Timer {
         use tokio::time::sleep;
 
         struct TokioTimer;
@@ -80,7 +77,7 @@ impl PowerSyncEnvironment {
                 sleep(duration).boxed()
             }
         }
-        &TokioTimer
+        TokioTimer
     }
 }
 
@@ -90,7 +87,7 @@ impl PowerSyncEnvironment {
 /// Because the native PowerSync SDK is executor-agnostic, it can't use a builtin function to retry
 /// sync after a delay to recover from errors. This trait, as part of the [PowerSyncEnvironment],
 /// is thus used to schedule the delay.
-pub trait Timer {
+pub trait Timer: Send + Sync + 'static {
     /// Returns a future that returns [Poll::Pending] when being polled the first time and schedules
     /// the context's waker to be woken after the specified `duration`.
     fn delay_once(&self, duration: Duration) -> Pin<Box<dyn Future<Output = ()> + Send>>;

powersync/src/sync/coordinator.rs

@@ -5,7 +5,6 @@ use async_oneshot::oneshot;
 
 use crate::{
     SyncOptions,
-    db::internal::InnerPowerSyncState,
     sync::{
         download::DownloadActorCommand, streams::ChangedSyncSubscriptions,
         upload::UploadActorCommand,
@@ -42,16 +41,10 @@ pub struct SyncCoordinator {
 }
 
 impl SyncCoordinator {
-    pub async fn connect(&self, options: SyncOptions, db: &InnerPowerSyncState) {
-        {
-            let mut lock = db.retry_delay.lock().unwrap();
-            *lock = Some(options.retry_delay);
-        }
-
-        let connector = options.connector.clone();
-        self.download_actor_request(DownloadActorCommand::Connect(options))
+    pub async fn connect(&self, options: SyncOptions) {
+        self.download_actor_request(DownloadActorCommand::Connect(options.clone()))
             .await;
-        self.upload_actor_request(UploadActorCommand::Connect(connector))
+        self.upload_actor_request(UploadActorCommand::Connect(options))
             .await;
     }
 

powersync/src/sync/download/actor.rs

@@ -75,6 +75,14 @@ impl DownloadActor {
         };
     }
 
+    fn retry_delay(&self) -> Boxed<()> {
+        if let Some(ref options) = self.options {
+            options.retry_delay(&self.db.env).boxed()
+        } else {
+            async {}.boxed()
+        }
+    }
+
     async fn handle_event(&mut self) {
         match &mut self.state {
             DownloadActorState::Idle => {
@@ -176,18 +184,15 @@ impl DownloadActor {
                         let timeout = if close.hide_disconnect {
                             async {}.boxed()
                         } else {
-                            let db = self.db.clone();
-
-                            async move { db.sync_iteration_delay().await }.boxed()
+                            self.retry_delay()
                         };
 
                         self.state = DownloadActorState::WaitingForReconnect { timeout }
                     }
                     Event::SyncIterationError(e) => {
                         self.db.status.update(|status| status.set_download_error(e));
-                        let db = self.db.clone();
                         self.state = DownloadActorState::WaitingForReconnect {
-                            timeout: async move { db.sync_iteration_delay().await }.boxed(),
+                            timeout: self.retry_delay(),
                         }
                     }
                 }

powersync/src/sync/download/http.rs

@@ -188,7 +188,7 @@ mod tests {
     fn first_event(client: impl HttpClient) -> Result<Option<DownloadEvent>, PowerSyncError> {
         PowerSyncEnvironment::powersync_auto_extension().unwrap();
         let pool = ConnectionPool::single_connection(Connection::open_in_memory().unwrap());
-        let environment = PowerSyncEnvironment::custom(client, pool, &UnusedTimer);
+        let environment = PowerSyncEnvironment::custom(client, pool, UnusedTimer);
         let coordinator = Arc::new(SyncCoordinator::default());
         let db = Arc::new(InnerPowerSyncState::new(
             environment,

powersync/src/sync/options.rs

@@ -1,6 +1,8 @@
 use std::{sync::Arc, time::Duration};
 
-use crate::sync::connector::BackendConnector;
+use futures_lite::future::yield_now;
+
+use crate::{env::PowerSyncEnvironment, sync::connector::BackendConnector};
 
 /// Options controlling how PowerSync connects to a sync service.
 #[derive(Clone)]
@@ -34,4 +36,20 @@ impl SyncOptions {
     pub fn with_retry_delay(&mut self, delay: Duration) {
         self.retry_delay = delay;
     }
+
+    pub(crate) fn retry_delay(
+        &self,
+        env: &PowerSyncEnvironment,
+    ) -> impl Future<Output = ()> + 'static {
+        let delay = self.retry_delay;
+        let timer = env.timer.clone();
+
+        async move {
+            if delay > Duration::ZERO {
+                timer.delay_once(delay).await
+            } else {
+                yield_now().await
+            }
+        }
+    }
 }