Use spawned async tasks for sync

.github/workflows/ci.yml

@@ -38,6 +38,9 @@ jobs:
       - name: Build without rusqlite
         run: cargo build --no-default-features
 
+      - name: Build with smol runtime (tokio disabled)
+        run: cargo build -p powersync --no-default-features --features smol,rusqlite --verbose
+
   windows:
     name: Check Windows build
     runs-on: windows-latest

CHANGELOG.md

@@ -1,6 +1,8 @@
 ## 0.0.8 (unreleased)
 
-- __Breaking__: Timers on `PowerSyncEnvironment::custom` are now passed by value.
+- __Breaking__: `PowerSyncEnvironment::custom` now needs a way to spawn async
+  tasks. Replace `tokio_timer()` or `async_io_timer()` timers with
+  `tokio()` or `async_io()`, respectively.
 - 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.

examples/egui_todolist/src/database.rs

@@ -12,7 +12,6 @@ use reqwest::StatusCode;
 use rusqlite::{Connection, params};
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
-use tokio::runtime::Runtime;
 
 pub struct TodoEntry {
     pub id: String,
@@ -83,20 +82,18 @@ pub struct TodoDatabase {
 }
 
 impl TodoDatabase {
-    pub fn new(rt: &Runtime) -> Self {
+    pub fn new() -> Self {
         let conn = Connection::open_in_memory().expect("should open connection");
         let env = PowerSyncEnvironment::custom(
             reqwest::Client::new(),
             ConnectionPool::single_connection(conn),
-            PowerSyncEnvironment::tokio_timer(),
+            PowerSyncEnvironment::tokio(),
         );
         let mut schema = Schema::default();
         schema.tables.push(TodoList::schema());
         schema.tables.push(TodoEntry::schema());
 
         let db = PowerSyncDatabase::new(env, schema);
-        db.async_tasks().spawn_with_tokio_runtime(rt);
-
         Self { db }
     }
 

examples/egui_todolist/src/ui.rs

@@ -38,7 +38,7 @@ impl Drop for SelectedTodoList {
 
 impl TodoListApp {
     pub fn new(rt: Runtime) -> Self {
-        let db = TodoDatabase::new(&rt);
+        let db = TodoDatabase::new();
 
         Self {
             rt,

powersync/Cargo.toml

@@ -18,15 +18,17 @@ crate-type = ["lib"]
 default = ["rusqlite"]
 
 tokio = ["dep:tokio"]
-smol = ["dep:async-io"]
+smol = ["dep:async-io", "dep:async-executor"]
 reqwest = ["dep:reqwest"]
 rusqlite = ["dep:rusqlite"]
 ffi = []
 
 [dependencies]
 async-channel = "2.5.0"
 async-lock = "3.4.1"
+async-executor = { version = "1.14.0", optional = true }
 async-io = { version = "2.6.0", optional = true }
+async-task = "4.7.1"
 async-trait = "0.1.89"
 async-oneshot = "0.5.9"
 atomic_enum = "0.3.0"
@@ -50,7 +52,6 @@ num-traits = "0.2.19"
 
 [dev-dependencies]
 async-executor = "1.13.3"
-async-task = "4.7.1"
 futures-lite = "2.6.1"
 futures-test = "0.3.31"
 powersync_test_utils = { path = "../powersync_test_utils" }

powersync/README.md

@@ -39,29 +39,8 @@ This crate provides asynchronous APIs to:
 3. Stream changes from the PowerSync service to your local database.
 4. Upload local writes to your backend.
 
-For maximum flexibility, the `powersync` crate is executor-agnostic and can run on async runtime. All tasks that run
-concurrently to the main application need to be spawned before starting PowerSync:
-
-```Rust
-#[tokio::main]
-async fn main() {
-    let db = PowerSyncDatabase::new(env, schema);
-    db.async_tasks().spawn_with(|future| {
-        tokio::spawn(future);
-    });
-}
-```
-
-The crate generally operates on a bring-your-own-runtime assumption, although optional features are available for
-popular runtimes. The above snippet can be simplified to:
-
-```Rust
-#[tokio::main]
-async fn main() {
-    let db = PowerSyncDatabase::new(env, schema);
-    db.async_tasks().spawn_with_tokio();
-}
-```
+For maximum flexibility, the `powersync` crate is executor-agnostic and can run on any async runtime,
+with builtin support for Tokio and smol-rs available through optional features.
 
 ### Running queries
 

powersync/src/db/internal.rs

@@ -6,13 +6,12 @@ use crate::{
     },
     env::PowerSyncEnvironment,
     error::PowerSyncError,
-    sync::{MAX_OP_ID, coordinator::SyncCoordinator, status::SyncStatus, status::SyncStatusData},
+    sync::{MAX_OP_ID, status::SyncStatus, status::SyncStatusData},
     util::SharedFuture,
 };
 use event_listener::EventListener;
 use futures_lite::{FutureExt, Stream, StreamExt, ready};
 use powersync_sqlite_nostd::{ColumnType, Destructor, ResultCode};
-use std::sync::Weak;
 use std::{
     pin::Pin,
     sync::Arc,
@@ -32,25 +31,16 @@ pub struct InnerPowerSyncState {
     pub status: SyncStatus,
     /// A collection of currently-referenced sync stream subscriptions.
     pub(crate) current_streams: SyncStreamTracker,
-    /// Clients have a strong reference to the sync coordinator, but since sync actors have a
-    /// 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>,
 }
 
 impl InnerPowerSyncState {
-    pub fn new(
-        env: PowerSyncEnvironment,
-        schema: SchemaOrCustom,
-        sync: &Arc<SyncCoordinator>,
-    ) -> Self {
+    pub fn new(env: PowerSyncEnvironment, schema: SchemaOrCustom) -> Self {
         Self {
             env,
             did_initialize: SharedFuture::new(),
             schema: Arc::new(schema),
             status: SyncStatus::new(),
             current_streams: SyncStreamTracker::default(),
-            sync: Arc::downgrade(sync),
         }
     }
 

powersync/src/db/mod.rs

@@ -3,7 +3,6 @@ use std::collections::HashSet;
 use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
-use crate::db::async_support::AsyncDatabaseTasks;
 use crate::db::watch::ListenerConfiguration;
 use crate::schema::SchemaOrCustom;
 use crate::sync::coordinator::SyncCoordinator;
@@ -15,11 +14,10 @@ use crate::{
     },
     env::PowerSyncEnvironment,
     error::PowerSyncError,
-    sync::{download::DownloadActor, status::SyncStatusData, upload::UploadActor},
+    sync::status::SyncStatusData,
 };
-use futures_lite::{FutureExt, Stream, StreamExt};
+use futures_lite::{Stream, StreamExt};
 
-mod async_support;
 pub(crate) mod connection;
 pub mod core_extension;
 pub mod crud;
@@ -58,38 +56,20 @@ impl PowerSyncDatabase {
         let coordinator = Arc::new(SyncCoordinator::default());
 
         Self {
-            inner: Arc::new(InnerPowerSyncState::new(env, schema.into(), &coordinator)),
+            inner: Arc::new(InnerPowerSyncState::new(env, schema.into())),
             sync: coordinator,
         }
     }
 
-    /// Returns a collection of [AsyncDatabaseTasks] that need to be started before connecting this
-    /// PowerSync database to a PowerSync service.
-    ///
-    /// To start the tasks, see the documentation on [AsyncDatabaseTasks].
-    ///
-    /// By exposing these async tasks instead of starting them automatically, the SDK stays
-    /// executor-agnostic and is easier to access from C.
-    #[must_use = "Returned tasks still need to be spawned on an async runtime"]
-    pub fn async_tasks(&self) -> AsyncDatabaseTasks {
-        let mut downloads = DownloadActor::new(self.inner.clone(), &self.sync);
-        let mut uploads = UploadActor::new(self.inner.clone(), &self.sync);
-
-        AsyncDatabaseTasks::new(
-            async move { downloads.run().await }.boxed(),
-            async move { uploads.run().await }.boxed(),
-        )
-    }
-
     /// 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).await
+        self.sync.clone().connect(self.inner.clone(), options).await
     }
 
     /// If the sync client is currently connected, requests it to disconnect.
     pub async fn disconnect(&self) {
-        self.sync.disconnect().await
+        self.sync.disconnect(&self.inner).await
     }
 
     /// Returns an asynchronous [Stream] emitting an empty event every time one of the specified

powersync/src/db/streams.rs

@@ -118,7 +118,10 @@ impl<'a> SyncStream<'a> {
             priority: options.priority,
         }))
         .await?;
-        self.db.sync.resolve_offline_sync_status().await;
+        self.db
+            .sync
+            .resolve_offline_sync_status(&self.db.inner)
+            .await?;
 
         let (stream, changed) = self
             .db
@@ -127,7 +130,11 @@ impl<'a> SyncStream<'a> {
             .reference_stream(&self.db.inner, &desc.into());
 
         if let Some(changed) = changed {
-            self.db.sync.handle_subscriptions_changed(changed).await;
+            self.db
+                .sync
+                .clone()
+                .handle_subscriptions_changed(changed)
+                .await;
         }
 
         Ok(StreamSubscription { group: stream })