diff --git a/powersync/src/sync/upload.rs b/powersync/src/sync/upload.rs index 2b0000e..955ea39 100644 --- a/powersync/src/sync/upload.rs +++ b/powersync/src/sync/upload.rs @@ -60,7 +60,7 @@ impl UploadActor { .env .pool .update_notifiers() - .listen(ListenerConfiguration::if_matches(tables, false)); + .listen(ListenerConfiguration::if_matches(tables, true)); ConnectedUploadActor { connector, crud_stream: stream.map(|_| ()).boxed(), diff --git a/powersync/tests/sync_test.rs b/powersync/tests/sync_test.rs index 57b9112..d096224 100644 --- a/powersync/tests/sync_test.rs +++ b/powersync/tests/sync_test.rs @@ -92,6 +92,61 @@ fn dropping_database_completes_actors() { })); } +#[test] +fn reconnect_uploads_pending_writes_after_an_early_trigger() { + struct Connector { + db: PowerSyncDatabase, + uploads: Arc, + } + + #[async_trait] + impl BackendConnector for Connector { + async fn fetch_credentials(&self) -> Result { + TestConnector.fetch_credentials().await + } + + async fn upload_data(&self) -> Result<(), PowerSyncError> { + while let Some(tx) = self.db.next_crud_transaction().await? { + self.uploads.fetch_add(1, Ordering::SeqCst); + tx.complete().await?; + } + Ok(()) + } + } + + future::block_on(async { + let test = DatabaseTest::new(); + let db = test.in_memory_database(); + powersync_test_utils::execute( + &db, + "INSERT INTO users (id, name) VALUES ('1', 'offline')", + [], + ) + .await; + let uploads = Arc::new(AtomicUsize::new(0)); + let mut actors = db.async_tasks().spawn_with(|task| task); + let mut upload = actors.pop().unwrap(); + let mut download = actors.pop().unwrap(); + let mut connect = Box::pin(db.connect(SyncOptions::new(Connector { + db: db.clone(), + uploads: uploads.clone(), + }))); + + // Let the download actor's upload trigger arrive before the upload + // actor receives Connect. No new writes should be needed afterward. + assert!(future::poll_once(connect.as_mut()).await.is_none()); + assert!(future::poll_once(&mut download).await.is_none()); + assert!(future::poll_once(&mut upload).await.is_none()); + assert!(future::poll_once(&mut download).await.is_none()); + assert!(future::poll_once(connect.as_mut()).await.is_none()); + assert!(future::poll_once(&mut upload).await.is_none()); + assert!(future::poll_once(connect.as_mut()).await.is_some()); + + assert_eq!(uploads.load(Ordering::SeqCst), 1); + assert!(db.next_crud_transaction().await.unwrap().is_none()); + }); +} + #[test] fn can_disable_default_stream() { let sync = SyncStreamTest::new();