diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7326612..53704ab 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 945ad69..2a949fd 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/examples/egui_todolist/src/database.rs b/examples/egui_todolist/src/database.rs index c0b8f5c..0343878 100644 --- a/examples/egui_todolist/src/database.rs +++ b/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 } } diff --git a/examples/egui_todolist/src/ui.rs b/examples/egui_todolist/src/ui.rs index 61b6c3c..e2fca46 100644 --- a/examples/egui_todolist/src/ui.rs +++ b/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, diff --git a/powersync/Cargo.toml b/powersync/Cargo.toml index 33ddfa7..060180b 100644 --- a/powersync/Cargo.toml +++ b/powersync/Cargo.toml @@ -18,7 +18,7 @@ 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 = [] @@ -26,7 +26,9 @@ 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" } diff --git a/powersync/README.md b/powersync/README.md index c7364dd..25182aa 100644 --- a/powersync/README.md +++ b/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 diff --git a/powersync/src/db/async_support.rs b/powersync/src/db/async_support.rs deleted file mode 100644 index 6e50456..0000000 --- a/powersync/src/db/async_support.rs +++ /dev/null @@ -1,38 +0,0 @@ -use futures_lite::future::Boxed; -#[cfg(feature = "tokio")] -use tokio::{runtime::Runtime, spawn, task::JoinHandle}; - -/// A collection of tasks, represented as pending futures, that should be started concurrently for -/// PowerSync to work. -pub struct AsyncDatabaseTasks { - download: Boxed<()>, - upload: Boxed<()>, -} - -impl AsyncDatabaseTasks { - pub(crate) fn new(download: Boxed<()>, upload: Boxed<()>) -> Self { - Self { download, upload } - } - - /// Spawns pending futures. - /// - /// This invokes the `spawner` function with each future that the PowerSync SDK would like to - /// run asynchronously. The function is responsible for continuously polling the futures. - /// The futures will complete once all [PowerSyncDatabase]s have been closed, so spawned tasks - /// don't have to be cancelled or dropped manually. - pub fn spawn_with(self, mut spawner: impl FnMut(Boxed<()>) -> T) -> Vec { - vec![spawner(self.download), spawner(self.upload)] - } - - /// Spawns pending futures as tokio tasks on the given [Runtime]. - #[cfg(feature = "tokio")] - pub fn spawn_with_tokio_runtime(self, runtime: &Runtime) -> Vec> { - self.spawn_with(|f| runtime.spawn(f)) - } - - /// Spawns pending futures as tokio tasks on the current [Runtime]. - #[cfg(feature = "tokio")] - pub fn spawn_with_tokio(self) -> Vec> { - self.spawn_with(|f| spawn(f)) - } -} diff --git a/powersync/src/db/internal.rs b/powersync/src/db/internal.rs index 6edd23f..4bf5369 100644 --- a/powersync/src/db/internal.rs +++ b/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, } impl InnerPowerSyncState { - pub fn new( - env: PowerSyncEnvironment, - schema: SchemaOrCustom, - sync: &Arc, - ) -> 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), } } diff --git a/powersync/src/db/mod.rs b/powersync/src/db/mod.rs index e423dd7..3fe5ad5 100644 --- a/powersync/src/db/mod.rs +++ b/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 diff --git a/powersync/src/db/streams.rs b/powersync/src/db/streams.rs index c31dc78..c358716 100644 --- a/powersync/src/db/streams.rs +++ b/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 }) diff --git a/powersync/src/env.rs b/powersync/src/env.rs index 06b2b37..b820bff 100644 --- a/powersync/src/env.rs +++ b/powersync/src/env.rs @@ -1,7 +1,13 @@ use super::db::pool::ConnectionPool; use crate::error::{PowerSyncError, RawPowerSyncError}; use crate::http::HttpClient; +#[cfg(feature = "smol")] +use async_executor::Executor; +use async_task::Task; +use futures_lite::FutureExt; +use futures_lite::future::Boxed; use num_traits::FromPrimitive; +use pin_project_lite::pin_project; use powersync_core::powersync_init_static; use powersync_sqlite_nostd::ResultCode; use std::sync::Arc; @@ -18,18 +24,26 @@ 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: Arc, + pub(crate) runtime: Arc, } impl PowerSyncEnvironment { - pub fn custom(client: C, pool: ConnectionPool, timer: T) -> Self { + pub fn custom( + client: C, + pool: ConnectionPool, + runtime: T, + ) -> Self { Self { client: Box::new(client), pool, - timer: Arc::new(timer), + runtime: Arc::new(runtime), } } + pub(crate) fn spawn(&self, f: impl Future + Send + 'static) -> PowerSyncTask { + self.runtime.spawn(f.boxed()) + } + /// Calls `sqlite3_auto_extension` with the statically-linked core extension. /// /// This needs to be invoked before using the PowerSync SDK. It can safely be called multiple @@ -45,13 +59,16 @@ impl PowerSyncEnvironment { } } - /// A [Timer] implementation based on [async_io::Timer]. + /// An [AsyncRuntime] implementation based on `async_task` and [async_io::Timer]. #[cfg(feature = "smol")] - pub fn async_io_timer() -> impl Timer { + pub fn async_io(executor: Arc>) -> impl AsyncRuntime { use async_io::Timer as PlatformTimer; - struct AsyncIoTimer; - impl Timer for AsyncIoTimer { + struct AsyncIoRuntime { + executor: Arc>, + } + + impl AsyncRuntime for AsyncIoRuntime { fn delay_once(&self, duration: Duration) -> Pin + Send>> { use futures_lite::FutureExt; @@ -60,35 +77,134 @@ impl PowerSyncEnvironment { } .boxed() } + + fn spawn(&self, task: Boxed<()>) -> PowerSyncTask<()> { + self.executor.spawn(task).into() + } } - AsyncIoTimer + AsyncIoRuntime { executor } } - /// A [Timer] implementation based on [tokio::time::sleep]. + /// An [AsyncRuntime] implementation based on tokio. #[cfg(feature = "tokio")] - pub fn tokio_timer() -> impl Timer { + pub fn tokio() -> impl AsyncRuntime { use tokio::time::sleep; - struct TokioTimer; - impl Timer for TokioTimer { + struct TokioRuntime; + + impl AsyncRuntime for TokioRuntime { fn delay_once(&self, duration: Duration) -> Pin + Send>> { use futures_lite::FutureExt; sleep(duration).boxed() } + + fn spawn(&self, task: Boxed<()>) -> PowerSyncTask<()> { + tokio::spawn(task).into() + } } - TokioTimer + TokioRuntime } } -/// An implementation of a timer as part of an event loop or async runtime hosting the PowerSync -/// SDK. +/// An implementation of an asynchronous executor and timer for the PowerSync SDK. /// -/// 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: Send + Sync + 'static { +/// Because the native PowerSync SDK is executor-agnostic, it can't use a builtin spawn function to +/// start background sync task or to schedule a delay to recover from errors. +/// +/// This trait, as part of the [PowerSyncEnvironment], is thus used to schedule the delay. +pub trait AsyncRuntime: 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 + Send>>; + + fn spawn(&self, task: Boxed<()>) -> PowerSyncTask<()>; +} + +pub struct PowerSyncTask { + raw: RawPowerSyncTask, +} + +impl PowerSyncTask { + pub fn cancel(self) { + match self.raw { + #[cfg(feature = "tokio")] + RawPowerSyncTask::Tokio { task } => { + task.abort(); + } + RawPowerSyncTask::AsyncTask { task } => { + // async_task cancels tasks when their handle is dropped. + drop(task) + } + } + } + + pub async fn cancel_and_join(self) -> Option { + match self.raw { + #[cfg(feature = "tokio")] + RawPowerSyncTask::Tokio { task } => { + task.abort(); + + match task.await { + Ok(e) => Some(e), + Err(e) => { + if e.is_cancelled() { + None + } else { + std::panic::resume_unwind(e.into_panic()) + } + } + } + } + RawPowerSyncTask::AsyncTask { task } => task.cancel().await, + } + } + + pub async fn join(self) -> T { + match self.raw { + #[cfg(feature = "tokio")] + RawPowerSyncTask::Tokio { task } => task.await.expect("Task should complete"), + RawPowerSyncTask::AsyncTask { task } => task.await, + } + } +} + +// We can't use cfg macros in pin_project +#[cfg(feature = "tokio")] +pin_project! { + #[project = RawPowerSyncTaskProj] + enum RawPowerSyncTask { + Tokio { + #[pin] task: tokio::task::JoinHandle, + }, + AsyncTask { + #[pin] task: Task + }, + } +} + +#[cfg(not(feature = "tokio"))] +pin_project! { + enum RawPowerSyncTask { + AsyncTask { + #[pin] task: Task + }, + } +} + +impl From> for PowerSyncTask { + fn from(value: Task) -> Self { + Self { + raw: RawPowerSyncTask::AsyncTask { task: value }, + } + } +} + +#[cfg(feature = "tokio")] +impl From> for PowerSyncTask { + fn from(value: tokio::task::JoinHandle) -> Self { + Self { + raw: RawPowerSyncTask::Tokio { task: value }, + } + } } diff --git a/powersync/src/sync/coordinator.rs b/powersync/src/sync/coordinator.rs index 441802d..cd460b2 100644 --- a/powersync/src/sync/coordinator.rs +++ b/powersync/src/sync/coordinator.rs @@ -1,143 +1,168 @@ -use std::sync::RwLock; +use async_lock::Mutex as AsyncMutex; +use std::sync::Arc; use async_channel::{Receiver, Sender}; -use async_oneshot::oneshot; use crate::{ SyncOptions, + db::internal::InnerPowerSyncState, + env::PowerSyncTask, + error::PowerSyncError, sync::{ - download::DownloadActorCommand, streams::ChangedSyncSubscriptions, - upload::UploadActorCommand, + download::{DownloadEvent, download_loop}, + streams::ChangedSyncSubscriptions, + upload::crud_upload_loop, }, }; -pub struct AsyncRequest { - pub command: T, - pub response: async_oneshot::Sender<()>, -} - -impl AsyncRequest { - pub fn new(command: T) -> (Self, async_oneshot::Receiver<()>) { - let (tx, rx) = oneshot(); - ( - Self { - command, - response: tx, - }, - rx, - ) - } -} - -/// Implements `connect()` and `disconnect()` by dispatching messages to the upload and download -/// actors. +/// Implements `connect()` and `disconnect()` by starting asynchronous tasks driving those loops. /// -/// Since actors only have access to the receiving end of their channels, dropping the coordinator -/// will also terminate all actors (albeit asynchronously). +/// Dropping the coordinator will also terminate sync tasks (albeit asynchronously). #[derive(Default)] pub struct SyncCoordinator { - control_downloads: RwLock>>>, - control_uploads: RwLock>>>, + task: AsyncMutex>, } impl SyncCoordinator { - pub async fn connect(&self, options: SyncOptions) { - self.download_actor_request(DownloadActorCommand::Connect(options.clone())) - .await; - self.upload_actor_request(UploadActorCommand::Connect(options)) - .await; + pub async fn connect(self: Arc, db: Arc, options: SyncOptions) { + self.disconnect(&db).await; + + let mut guard = self.task.lock().await; + + let (channels, download_receive, uploads_receive) = SyncChannels::create(); + + let downloads = db.env.spawn(download_loop( + db.clone(), + channels.clone(), + options.clone(), + download_receive, + )); + let uploads = db.env.spawn(crud_upload_loop( + db.clone(), + options, + channels.clone(), + uploads_receive, + )); + + *guard = Some(SyncTasks { + channels, + uploads: Some(uploads), + downloads: Some(downloads), + }); } - pub async fn disconnect(&self) { - self.download_actor_request(DownloadActorCommand::Disconnect) - .await; - self.upload_actor_request(UploadActorCommand::Disconnect) - .await; - } + pub async fn disconnect(&self, db: &InnerPowerSyncState) { + let mut guard = self.task.lock().await; - /// Requests a round of CRUD uploads. - pub async fn trigger_crud_uploads(&self) { - self.upload_actor_request(UploadActorCommand::TriggerCrudUpload) - .await; - } - - /// Marks CRUD uploads as complete, allowing the download client to retry if a previous - /// checkpoint was blocked by pending uploads. - pub async fn mark_crud_uploads_completed(&self) { - self.download_actor_request(DownloadActorCommand::CrudUploadComplete) - .await; + if let Some(task) = guard.take() { + task.cancel().await; + let _ = Self::fetch_offline_sync_status(db).await; + } } - /// Causes the download actor to call `powersync_offline_sync_status()` and emit those results. + /// If we're offline, update the offline sync status and emit it into the database. /// /// This is used after adding a new subscription to include it in the sync status even if we're /// disconnected. /// This is a no-op while connected. - pub async fn resolve_offline_sync_status(&self) { - self.download_actor_request(DownloadActorCommand::ResolveOfflineSyncStatusIfNotConnected) - .await; + pub async fn resolve_offline_sync_status( + &self, + db: &InnerPowerSyncState, + ) -> Result<(), PowerSyncError> { + let guard = self.task.lock().await; + if guard.is_some() { + return Ok(()); + } + + Self::fetch_offline_sync_status(db).await + } + + async fn fetch_offline_sync_status(db: &InnerPowerSyncState) -> Result<(), PowerSyncError> { + let writer = db.writer().await?; + db.status + .update(|s| s.resolve_offline_state(writer.sqlite_connection())) } /// Handle the set of active sync stream subscriptions changing. /// /// This is a no-op if not connected. pub async fn handle_subscriptions_changed(&self, update: ChangedSyncSubscriptions) { - self.download_actor_request(DownloadActorCommand::SubscriptionsChanged(update)) - .await; - } - - fn install_actor_channel( - slot: &RwLock>>>, - ) -> Receiver> { - let mut slot = slot.write().unwrap(); - if slot.is_some() { - drop(slot); - panic!("Actor already installed") - } + let Some(channel) = ({ + let guard = self.task.lock().await; + + guard + .as_ref() + .map(|tasks| tasks.channels.local_download_events.clone()) + }) else { + return; + }; - let (send, receive) = async_channel::bounded(1); - *slot = Some(send); - receive + let _ = channel + .send(DownloadEvent::UpdateSubscriptions { keys: update.0 }) + .await; } +} - fn obtain_channel( - slot: &RwLock>>>, - ) -> Sender> { - let slot = slot.read().unwrap(); - let Some(slot) = &*slot else { - panic!("Actor has not been registered"); - }; +struct SyncTasks { + channels: SyncChannels, + downloads: Option, + uploads: Option, +} - slot.clone() +impl SyncTasks { + pub async fn cancel(mut self) { + if let Some(task) = self.downloads.take() { + task.cancel_and_join().await; + } + if let Some(task) = self.uploads.take() { + task.cancel_and_join().await; + } } +} - pub fn receive_download_commands(&self) -> Receiver> { - Self::install_actor_channel(&self.control_downloads) +impl Drop for SyncTasks { + fn drop(&mut self) { + if let Some(task) = self.downloads.take() { + task.cancel(); + } + if let Some(task) = self.uploads.take() { + task.cancel(); + } } +} - pub fn receive_upload_commands(&self) -> Receiver> { - Self::install_actor_channel(&self.control_uploads) - } +#[derive(Clone)] +pub struct SyncChannels { + local_download_events: Sender, + trigger_upload: Sender<()>, +} - async fn download_actor_request(&self, cmd: DownloadActorCommand) { - let downloads = Self::obtain_channel(&self.control_downloads); +impl SyncChannels { + pub fn create() -> (Self, Receiver, Receiver<()>) { + let (download_send, download_receive) = async_channel::unbounded(); + let (uploads_send, uploads_receive) = async_channel::bounded(1); - let (request, response) = AsyncRequest::new(cmd); - downloads - .send(request) - .await - .expect("Download actor not running, start it with download_actor()"); - let _ = response.await; + ( + Self { + local_download_events: download_send, + trigger_upload: uploads_send, + }, + download_receive, + uploads_receive, + ) } - async fn upload_actor_request(&self, cmd: UploadActorCommand) { - let uploads = Self::obtain_channel(&self.control_uploads); + pub fn trigger_crud_upload(&self) { + // If an existing crud request is already buffered in the channel, we can replace it. + let _ = self.trigger_upload.force_send(()); + } - let (request, response) = AsyncRequest::new(cmd); - uploads - .send(request) - .await - .expect("Upload actor not running, start it with upload_actor()"); - let _ = response.await; + /// Marks CRUD uploads as complete, allowing the download client to retry if a previous + /// checkpoint was blocked by pending uploads. + pub async fn mark_crud_uploads_completed(&self) { + let _ = self + .local_download_events + .send(DownloadEvent::CompletedUpload) + .await; } } diff --git a/powersync/src/sync/download/actor.rs b/powersync/src/sync/download/actor.rs deleted file mode 100644 index 9e5cc22..0000000 --- a/powersync/src/sync/download/actor.rs +++ /dev/null @@ -1,275 +0,0 @@ -use std::sync::Arc; - -use futures_lite::{ - FutureExt, - future::{self, Boxed}, -}; -use log::warn; -use serde_json::Map; - -use crate::sync::coordinator::SyncCoordinator; -use crate::{ - SyncOptions, - db::internal::InnerPowerSyncState, - error::PowerSyncError, - sync::{ - coordinator::AsyncRequest, - download::sync_iteration::{DownloadClient, DownloadEvent, StartDownloadIteration}, - instruction::CloseSyncStream, - streams::ChangedSyncSubscriptions, - }, -}; - -/// A command sent from a database to the download actor. -pub enum DownloadActorCommand { - Connect(SyncOptions), - Disconnect, - ResolveOfflineSyncStatusIfNotConnected, - SubscriptionsChanged(ChangedSyncSubscriptions), - CrudUploadComplete, -} - -pub struct DownloadActor { - state: DownloadActorState, - commands: async_channel::Receiver>, - db: Arc, - options: Option, -} - -impl DownloadActor { - pub fn new(db: Arc, sync: &SyncCoordinator) -> Self { - let commands = sync.receive_download_commands(); - - Self { - state: DownloadActorState::Idle, - commands, - db, - options: None, - } - } - - pub async fn run(&mut self) { - while !self.state.is_stopped() { - self.handle_event().await; - } - } - - fn start_iteration(&mut self, options: SyncOptions) { - let (send_events, receive_event) = async_channel::bounded(1); - let start = StartDownloadIteration { - parameters: serde_json::Value::Object(Map::new()), - schema: self.db.schema.clone(), - include_defaults: options.include_default_streams, - active_streams: self.db.current_streams.collect_active_streams(), - }; - let future = DownloadClient::new(self.db.clone(), receive_event) - .run(options) - .boxed(); - send_events - .try_send(DownloadEvent::Start(start)) - .expect("should send start message"); - - self.state = DownloadActorState::Running { - iteration: future, - send_events, - }; - } - - 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 => { - // When we're idle, the only thing that can trigger a connection is a connect() - // call. - let Ok(mut command) = self.commands.recv().await else { - self.state = DownloadActorState::Stopped; - return; - }; - - match command.command { - DownloadActorCommand::Connect(options) => { - self.options = Some(options.clone()); - self.start_iteration(options); - let _ = command.response.send(()); - } - DownloadActorCommand::ResolveOfflineSyncStatusIfNotConnected => { - let res = async { - let writer = self.db.writer().await?; - self.db - .status - .update(|s| s.resolve_offline_state(writer.sqlite_connection()))?; - - Ok::<(), PowerSyncError>(()) - } - .await; - if let Err(e) = res { - warn!("Could not resolve offline sync state: {e}") - } - } - DownloadActorCommand::Disconnect - | DownloadActorCommand::SubscriptionsChanged(_) - | DownloadActorCommand::CrudUploadComplete => { - // Not connected, nothing to do. - } - } - } - DownloadActorState::Running { - send_events, - iteration, - } => { - // The only thing that triggers a state transition is for the current iteration to - // end. That can happen due to network errors, but also if disconnect() is called. - // So we have to listen for both. - enum Event { - ForwardedMessage, - SyncIterationComplete(CloseSyncStream), - SyncIterationError(PowerSyncError), - } - - let forwarding_request = async { - match self.commands.recv().await { - Ok(command) => match command.command { - DownloadActorCommand::Connect(_) => { - // We're already connected, do nothing. - // TODO: Compare options and potentially reconnect - } - DownloadActorCommand::ResolveOfflineSyncStatusIfNotConnected => { - // We're connected, so nothing we'd have to do. - } - DownloadActorCommand::SubscriptionsChanged(changed) => { - let _ = send_events - .send(DownloadEvent::UpdateSubscriptions { keys: changed.0 }) - .await; - } - DownloadActorCommand::CrudUploadComplete => { - let _ = send_events.send(DownloadEvent::CompletedUpload).await; - } - DownloadActorCommand::Disconnect => { - let _ = send_events.send(DownloadEvent::Stop).await; - } - }, - Err(_) => { - // There are no remaining instances of the PowerSync database left, - // close the stream. - let _ = send_events.send(DownloadEvent::Stop).await; - } - } - - Event::ForwardedMessage - }; - - let iteration_done = async { - match iteration.await { - Ok(e) => Event::SyncIterationComplete(e), - Err(e) => { - warn!("Sync iteration failed, {e}"); - Event::SyncIterationError(e) - } - } - }; - - let event = future::race(forwarding_request, iteration_done).await; - match event { - Event::ForwardedMessage => { - // Message was handled, we can go on immediately. - } - Event::SyncIterationComplete(close) => { - let timeout = if close.hide_disconnect { - async {}.boxed() - } else { - self.retry_delay() - }; - - self.state = DownloadActorState::WaitingForReconnect { timeout } - } - Event::SyncIterationError(e) => { - self.db.status.update(|status| status.set_download_error(e)); - self.state = DownloadActorState::WaitingForReconnect { - timeout: self.retry_delay(), - } - } - } - } - DownloadActorState::WaitingForReconnect { timeout } => { - // Either the timeout expires, in which case we reconnect, or a disconnect is - // requested. - enum Event { - DisconnectRequested, - TimeoutExpired, - } - - let commands = self.commands.clone(); - let disconnect_requested = async move { - Self::wait_for_disconnect_request(&commands).await; - Event::DisconnectRequested - }; - - let timeout_expired = async { - timeout.await; - Event::TimeoutExpired - }; - - match future::race(disconnect_requested, timeout_expired).await { - Event::DisconnectRequested => { - self.state = DownloadActorState::Idle; - } - Event::TimeoutExpired => { - self.start_iteration(self.options.as_ref().unwrap().clone()); - } - } - } - DownloadActorState::Stopped => panic!("No further state transitions after stopped"), - }; - } - - /// Polls on the given channel until we receive a command indicating that the actor should - /// disconnect. - async fn wait_for_disconnect_request( - commands: &async_channel::Receiver>, - ) { - loop { - match commands.recv().await { - Ok(command) => match command.command { - DownloadActorCommand::Connect(_) - | DownloadActorCommand::SubscriptionsChanged(_) - | DownloadActorCommand::ResolveOfflineSyncStatusIfNotConnected - | DownloadActorCommand::CrudUploadComplete => { - continue; - } - DownloadActorCommand::Disconnect => { - return; - } - }, - Err(_) => { - // No clients left, treat that as a disconnect request and clean up resources. - return; - } - } - } - } -} - -enum DownloadActorState { - Idle, - Running { - send_events: async_channel::Sender, - iteration: Boxed>, - }, - WaitingForReconnect { - timeout: Boxed<()>, - }, - Stopped, -} - -impl DownloadActorState { - fn is_stopped(&self) -> bool { - matches!(self, Self::Stopped) - } -} diff --git a/powersync/src/sync/download/http.rs b/powersync/src/sync/download/http.rs index 3a307f7..c3de724 100644 --- a/powersync/src/sync/download/http.rs +++ b/powersync/src/sync/download/http.rs @@ -146,10 +146,9 @@ mod tests { use super::*; use crate::{ db::{internal::InnerPowerSyncState, pool::ConnectionPool}, - env::{PowerSyncEnvironment, Timer}, + env::{AsyncRuntime, PowerSyncEnvironment}, http::{HttpClient, Request, ResponseBody}, schema::Schema, - sync::coordinator::SyncCoordinator, }; struct FailingClient; @@ -179,21 +178,23 @@ mod tests { struct UnusedTimer; - impl Timer for UnusedTimer { + impl AsyncRuntime for UnusedTimer { fn delay_once(&self, _duration: Duration) -> Pin + Send>> { Box::pin(future::pending()) } + + fn spawn(&self, _task: future::Boxed<()>) -> crate::env::PowerSyncTask<()> { + panic!("Unsupported in tests") + } } fn first_event(client: impl HttpClient) -> Result, PowerSyncError> { PowerSyncEnvironment::powersync_auto_extension().unwrap(); let pool = ConnectionPool::single_connection(Connection::open_in_memory().unwrap()); let environment = PowerSyncEnvironment::custom(client, pool, UnusedTimer); - let coordinator = Arc::new(SyncCoordinator::default()); let db = Arc::new(InnerPowerSyncState::new( environment, Schema::default().into(), - &coordinator, )); let credentials = PowerSyncCredentials { endpoint: "https://rust.unit.test.powersync.com/".to_string(), diff --git a/powersync/src/sync/download/mod.rs b/powersync/src/sync/download/mod.rs index cd95fb1..97983f9 100644 --- a/powersync/src/sync/download/mod.rs +++ b/powersync/src/sync/download/mod.rs @@ -1,5 +1,33 @@ -mod actor; pub mod http; mod sync_iteration; -pub use actor::{DownloadActor, DownloadActorCommand}; +use std::sync::Arc; + +use log::warn; +pub use sync_iteration::{DownloadClient, DownloadEvent}; + +use crate::{SyncOptions, db::internal::InnerPowerSyncState, sync::coordinator::SyncChannels}; + +pub async fn download_loop( + db: Arc, + channels: SyncChannels, + options: SyncOptions, + events: async_channel::Receiver, +) { + loop { + let download = DownloadClient::new(db.clone(), &channels, &events, &options); + let delay_retry = match download.run().await { + Ok(end) => !end.hide_disconnect, + Err(e) => { + warn!("Sync iteration failed, {e}"); + db.status.update(|data| data.set_download_error(e)); + + true + } + }; + + if delay_retry { + options.retry_delay(&db.env).await + } + } +} diff --git a/powersync/src/sync/download/sync_iteration.rs b/powersync/src/sync/download/sync_iteration.rs index 8d703cb..688e423 100644 --- a/powersync/src/sync/download/sync_iteration.rs +++ b/powersync/src/sync/download/sync_iteration.rs @@ -4,10 +4,12 @@ use futures_lite::{StreamExt, future, stream::Boxed as BoxedStream}; use log::{debug, info, trace, warn}; use powersync_sqlite_nostd::{Destructor, ManagedStmt, ResultCode}; use serde::Serialize; +use serde_json::Map; use serde_json::value::RawValue; use crate::db::connection::{SqliteConnection, TransactionGuard}; use crate::schema::SchemaOrCustom; +use crate::sync::coordinator::SyncChannels; use crate::{ SyncOptions, db::internal::InnerPowerSyncState, @@ -19,26 +21,42 @@ use crate::{ }, }; -pub struct DownloadClient { +pub struct DownloadClient<'a> { db: Arc, + channels: &'a SyncChannels, + receive_commands: &'a async_channel::Receiver, + options: &'a SyncOptions, stream: Option>>, - receive_commands: async_channel::Receiver, } -impl DownloadClient { +impl<'a> DownloadClient<'a> { pub fn new( db: Arc, - events: async_channel::Receiver, + channels: &'a SyncChannels, + events: &'a async_channel::Receiver, + options: &'a SyncOptions, ) -> Self { Self { db, - stream: None, + channels, receive_commands: events, + options, + stream: None, } } - pub async fn run(mut self, options: SyncOptions) -> Result { - 'event: loop { + pub async fn run(mut self) -> Result { + let start = StartDownloadIteration { + parameters: serde_json::Value::Object(Map::new()), + schema: self.db.schema.clone(), + include_defaults: self.options.include_default_streams, + active_streams: self.db.current_streams.collect_active_streams(), + }; + if let Some(end) = self.handle_event(DownloadEvent::Start(start)).await? { + return Ok(end); + } + + loop { let event = match &mut self.stream { Some(stream) => { future::or( @@ -50,57 +68,66 @@ impl DownloadClient { None => Self::receive_command(&self.receive_commands).await, }?; - trace!("Handling event {event:?}"); - let instructions = { - let mut conn = self.db.writer().await?; - event.invoke_control(conn.sqlite_connection_mut())? - }; - - for instr in instructions { - trace!("Handling instruction {instr:?}"); - - match instr { - Instruction::LogLine { severity, line } => match severity { - LogSeverity::Debug => debug!("{}", line), - LogSeverity::Info => info!("{}", line), - LogSeverity::Warning => warn!("{}", line), - }, - Instruction::UpdateSyncStatus { status } => { - self.db.status.update(|s| s.update_from_core(status)) - } - Instruction::EstablishSyncStream { request } => { - trace!("Establishing sync stream with {request}"); - Self::establish_sync_stream( - Arc::clone(&self.db), - &mut self.stream, - request, - &options, - ) - .await?; - - // Trigger a crud upload after establishing a sync stream. - if let Some(sync) = self.db.sync.upgrade() { - sync.trigger_crud_uploads().await; - } - } - Instruction::FetchCredentials { .. } => { - // TODO: Pre-fetching credentials - // If did_expire is true, the core extension will also emit a stop - // instruction. So we don't have to handle that separately. - } - Instruction::CloseSyncStream(close) => { - break 'event Ok(close); - } - Instruction::FlushFileSystem {} => { - // Not applicable outside of Dart web. - } - Instruction::DidCompleteSync {} => self - .db - .status - .update(|status| status.clear_download_errors()), + if let Some(end) = self.handle_event(event).await? { + return Ok(end); + } + } + } + + async fn handle_event( + &mut self, + event: DownloadEvent, + ) -> Result, PowerSyncError> { + trace!("Handling event {event:?}"); + let instructions = { + let mut conn = self.db.writer().await?; + event.invoke_control(conn.sqlite_connection_mut())? + }; + + for instr in instructions { + trace!("Handling instruction {instr:?}"); + + match instr { + Instruction::LogLine { severity, line } => match severity { + LogSeverity::Debug => debug!("{}", line), + LogSeverity::Info => info!("{}", line), + LogSeverity::Warning => warn!("{}", line), + }, + Instruction::UpdateSyncStatus { status } => { + self.db.status.update(|s| s.update_from_core(status)) + } + Instruction::EstablishSyncStream { request } => { + trace!("Establishing sync stream with {request}"); + Self::establish_sync_stream( + Arc::clone(&self.db), + &mut self.stream, + request, + self.options, + ) + .await?; + + // Trigger a crud upload after establishing a sync stream. + self.channels.trigger_crud_upload(); + } + Instruction::FetchCredentials { .. } => { + // TODO: Pre-fetching credentials + // If did_expire is true, the core extension will also emit a stop + // instruction. So we don't have to handle that separately. + } + Instruction::CloseSyncStream(close) => { + return Ok(Some(close)); } + Instruction::FlushFileSystem {} => { + // Not applicable outside of Dart web. + } + Instruction::DidCompleteSync {} => self + .db + .status + .update(|status| status.clear_download_errors()), } } + + Ok(None) } async fn establish_sync_stream( diff --git a/powersync/src/sync/options.rs b/powersync/src/sync/options.rs index 65c0788..5620156 100644 --- a/powersync/src/sync/options.rs +++ b/powersync/src/sync/options.rs @@ -42,11 +42,15 @@ impl SyncOptions { env: &PowerSyncEnvironment, ) -> impl Future + 'static { let delay = self.retry_delay; - let timer = env.timer.clone(); + let future = if delay > Duration::ZERO { + Some(env.runtime.delay_once(delay)) + } else { + None + }; async move { - if delay > Duration::ZERO { - timer.delay_once(delay).await + if let Some(future) = future { + future.await; } else { yield_now().await } diff --git a/powersync/src/sync/upload.rs b/powersync/src/sync/upload.rs index a398446..5294fee 100644 --- a/powersync/src/sync/upload.rs +++ b/powersync/src/sync/upload.rs @@ -1,14 +1,13 @@ use std::{collections::HashSet, ops::ControlFlow, sync::Arc}; +use async_channel::Receiver; use futures_lite::{ - FutureExt, StreamExt, - future::{self, Boxed}, + StreamExt, + future::{self}, }; use log::{debug, info, warn}; use powersync_sqlite_nostd::{Destructor, ResultCode}; -use crate::db::watch::ListenerConfiguration; -use crate::sync::coordinator::SyncCoordinator; use crate::{ SyncOptions, db::connection::{SqliteConnection, TransactionGuard}, @@ -16,212 +15,59 @@ use crate::{ use crate::{ db::internal::InnerPowerSyncState, error::PowerSyncError, - sync::{ - MAX_OP_ID, coordinator::AsyncRequest, download::http::write_checkpoint, - status::UploadStatus, - }, + sync::{MAX_OP_ID, download::http::write_checkpoint, status::UploadStatus}, }; +use crate::{db::watch::ListenerConfiguration, sync::coordinator::SyncChannels}; -pub enum UploadActorCommand { - Connect(SyncOptions), - TriggerCrudUpload, - Disconnect, -} - -pub struct UploadActor { - state: UploadActorState, - commands: async_channel::Receiver>, +pub async fn crud_upload_loop( db: Arc, -} - -impl UploadActor { - pub fn new(db: Arc, sync: &SyncCoordinator) -> Self { - let commands = sync.receive_upload_commands(); - - Self { - state: UploadActorState::Idle, - commands, - db, - } - } - - pub async fn run(&mut self) { - while !self.state.is_stopped() { - self.handle_event().await - } - } - - fn connected_state( - db: &Arc, - options: SyncOptions, - ) -> ConnectedUploadActor { - let mut tables = HashSet::new(); - tables.insert("ps_crud".to_string()); - - let stream = db - .env - .pool - .update_notifiers() - .listen(ListenerConfiguration::if_matches(tables, false)); - ConnectedUploadActor { - options, - crud_stream: stream.map(|_| ()).boxed(), - } - } - - async fn state_transition_from_command_while_uploading( - commands: &async_channel::Receiver>, - db: &Arc, - ) -> Option { - match commands.recv().await { - Ok(command) => match command.command { - UploadActorCommand::TriggerCrudUpload => { - // Already in progress, don't start another. - None - } - UploadActorCommand::Connect(options) => { - // TODO: Only abort if options have changed? - Some(UploadActorState::Connected(Self::connected_state( - db, options, - ))) - } - UploadActorCommand::Disconnect => Some(UploadActorState::Idle), + options: SyncOptions, + channels: SyncChannels, + trigger_uploads: Receiver<()>, +) { + let mut tables = HashSet::new(); + tables.insert("ps_crud".to_string()); + + let mut stream = db + .env + .pool + .update_notifiers() + .listen(ListenerConfiguration::if_matches(tables, false)); + + loop { + let next_trigger = future::or( + async { + stream.next().await?; + Some(()) }, - Err(_) => { - // There are no remaining instances of the PowerSync database left. - Some(UploadActorState::Stopped) - } - } - } - - async fn handle_event(&mut self) { - let mut old_state = std::mem::replace(&mut self.state, UploadActorState::Idle); - - self.state = match old_state { - UploadActorState::Idle => { - // Wait for a connect() call - let Ok(mut command) = self.commands.recv().await else { - self.state = UploadActorState::Stopped; - return; - }; - - match command.command { - UploadActorCommand::Connect(options) => { - let _ = command.response.send(()); - UploadActorState::Connected(Self::connected_state(&self.db, options)) - } - UploadActorCommand::TriggerCrudUpload => { - // We can't upload because we're not connected - old_state - } - UploadActorCommand::Disconnect => { - // Not connected, nothing to do. - old_state - } - } - } - UploadActorState::Connected(mut state) => { - enum Transition { - StartUpload, - Abort(UploadActorState), - } - - let trigger_by_crud_change = async { - state.crud_stream.next().await; - Transition::StartUpload - }; - - let trigger_by_command = async { - let Ok(mut command) = self.commands.recv().await else { - self.state = UploadActorState::Stopped; - return Transition::StartUpload; - }; - - let _ = command.response.send(()); - - match command.command { - UploadActorCommand::Connect(options) => Transition::Abort( - UploadActorState::Connected(Self::connected_state(&self.db, options)), - ), - UploadActorCommand::TriggerCrudUpload => Transition::StartUpload, - UploadActorCommand::Disconnect => Transition::Abort(UploadActorState::Idle), - } - }; - - match future::race(trigger_by_crud_change, trigger_by_command).await { - Transition::StartUpload => self.start_upload(state), - Transition::Abort(state) => state, - } - } - UploadActorState::RunningUpload { ref mut result } => { - // A state transition can happen when the current upload is finished or when we - // receive a disconnect call. - - let request = - Self::state_transition_from_command_while_uploading(&self.commands, &self.db); - - let upload_done = async { - let state = result.await; - self.db - .status - .update(|s| s.set_upload_state(UploadStatus::Idle)); - - // The upload is done and we transition back into the ready connected state to start the next iteration when needed. - Some(UploadActorState::Connected(state)) - }; + async { + trigger_uploads.recv().await.ok()?; + Some(()) + }, + ); - future::race(request, upload_done) - .await - .unwrap_or(old_state) - } - UploadActorState::Stopped => panic!("No further state transitions after stopped"), + let mut upload = CrudUpload { + options: &options, + db: &db, + channels: &channels, }; - } - - fn start_upload(&self, state: ConnectedUploadActor) -> UploadActorState { - let db = self.db.clone(); - UploadActorState::RunningUpload { - result: async move { - let mut upload = CrudUpload { - options: &state.options, - db, - }; - upload.run().await; - - state - } - .boxed(), - } + upload.run().await; + next_trigger.await; } } -enum UploadActorState { - Idle, - Connected(ConnectedUploadActor), - RunningUpload { result: Boxed }, - Stopped, -} - -impl UploadActorState { - fn is_stopped(&self) -> bool { - matches!(self, Self::Stopped) - } -} - -struct ConnectedUploadActor { - options: SyncOptions, - /// A stream emitting changes when the `ps_crud` table is updated locally. - crud_stream: futures_lite::stream::Boxed<()>, -} - struct CrudUpload<'a> { options: &'a SyncOptions, - db: Arc, + channels: &'a SyncChannels, + db: &'a InnerPowerSyncState, } impl<'a> CrudUpload<'a> { pub async fn run(&mut self) { let mut last_item_id = None::; + scopeguard::defer! { + self.db.status.update(|s| s.set_upload_state(UploadStatus::Idle)); + } // Invoke upload method on connector until there are no remaining CRUD items to upload. loop { @@ -242,7 +88,7 @@ impl<'a> CrudUpload<'a> { } async fn upload_step( - &mut self, + &self, last_item_id: &mut Option, ) -> Result, PowerSyncError> { let Some(item) = self.oldest_crud_item_id().await? else { @@ -254,9 +100,7 @@ impl<'a> CrudUpload<'a> { // It's possible that pending CRUD uploads were preventing data from syncing. So now // that that's completed, notify the download client in case it needs to retry. - if let Some(sync) = self.db.sync.upgrade() { - sync.mark_crud_uploads_completed().await; - } + self.channels.mark_crud_uploads_completed().await; return Ok(ControlFlow::Break(())); }; diff --git a/powersync/tests/sync_test.rs b/powersync/tests/sync_test.rs index e0895ef..4ee0fb9 100644 --- a/powersync/tests/sync_test.rs +++ b/powersync/tests/sync_test.rs @@ -6,7 +6,6 @@ use std::{ time::{Duration, SystemTime}, }; -use async_task::Task; use async_trait::async_trait; use event_listener::Event; use futures_lite::{StreamExt, future}; @@ -26,7 +25,6 @@ use thiserror::Error; struct SyncStreamTest { test: DatabaseTest, db: PowerSyncDatabase, - tasks: Vec>, } impl SyncStreamTest { @@ -34,8 +32,7 @@ impl SyncStreamTest { let test = DatabaseTest::new(); let db = test.in_memory_database(); - let tasks = db.async_tasks().spawn_with(|f| test.ex.spawn(f)); - Self { db, test, tasks } + Self { db, test } } fn connect(&self) { @@ -80,18 +77,6 @@ impl SyncStreamTest { } } -#[test] -fn dropping_database_completes_actors() { - let sync = SyncStreamTest::new(); - drop(sync.db); - - future::block_on(sync.test.ex.run(async move { - for task in sync.tasks { - task.await; - } - })); -} - #[test] fn can_disable_default_stream() { let sync = SyncStreamTest::new(); diff --git a/powersync_test_utils/src/lib.rs b/powersync_test_utils/src/lib.rs index 3fa31ca..769397c 100644 --- a/powersync_test_utils/src/lib.rs +++ b/powersync_test_utils/src/lib.rs @@ -10,7 +10,7 @@ use async_executor::Executor; use futures_lite::FutureExt; use log::LevelFilter; use powersync::{ - env::{PowerSyncEnvironment, Timer}, + env::{AsyncRuntime, PowerSyncEnvironment}, schema::{Column, Schema, Table}, *, }; @@ -27,7 +27,7 @@ pub struct DatabaseTest { pub dir: TempDir, pub http: Arc, timer: Arc>, - pub ex: Executor<'static>, + pub ex: Arc>, } impl Default for DatabaseTest { @@ -41,7 +41,7 @@ impl Default for DatabaseTest { dir: TempDir::new("powersync_rust").expect("should create test directory"), http: Arc::new(MockSyncService::new()), timer: Default::default(), - ex: Executor::new(), + ex: Arc::new(Executor::new()), } } } @@ -96,9 +96,11 @@ impl DatabaseTest { PowerSyncEnvironment::powersync_auto_extension().expect("should load core extension"); let timer = self.timer.clone(); + let executor = self.ex.clone(); struct TestTimer { state: Arc>, + ex: Arc>, } struct TestDelay { @@ -107,7 +109,7 @@ impl DatabaseTest { did_register: bool, } - impl Timer for TestTimer { + impl AsyncRuntime for TestTimer { fn delay_once( &self, duration: Duration, @@ -122,6 +124,10 @@ impl DatabaseTest { } .boxed() } + + fn spawn(&self, task: futures_lite::future::Boxed<()>) -> env::PowerSyncTask<()> { + self.ex.spawn(task).into() + } } impl Future for TestDelay { @@ -155,7 +161,14 @@ impl DatabaseTest { } } - PowerSyncEnvironment::custom(self.http.clone().client(), pool, TestTimer { state: timer }) + PowerSyncEnvironment::custom( + self.http.clone().client(), + pool, + TestTimer { + state: timer, + ex: executor, + }, + ) } pub fn default_schema() -> Schema {