diff --git a/CHANGELOG.md b/CHANGELOG.md index 37eaa86..9624fbe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,7 @@ +## 0.1.1 + +- Add `PriorityStatus`, exposing sync status grouped by priority groups of Sync Streams. + ## 0.1.0 - __Breaking__: `PowerSyncEnvironment::custom` now needs a way to spawn async diff --git a/Cargo.lock b/Cargo.lock index bc18c90..1554df4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3317,7 +3317,7 @@ checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" [[package]] name = "powersync" -version = "0.1.0" +version = "0.1.1" dependencies = [ "async-broadcast", "async-channel", diff --git a/powersync/Cargo.toml b/powersync/Cargo.toml index ebe049b..c14262d 100644 --- a/powersync/Cargo.toml +++ b/powersync/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "powersync" -version = "0.1.0" +version = "0.1.1" edition = "2024" license = "Apache-2.0" diff --git a/powersync/src/lib.rs b/powersync/src/lib.rs index fc16497..2843c2b 100644 --- a/powersync/src/lib.rs +++ b/powersync/src/lib.rs @@ -14,7 +14,7 @@ pub use db::{DisconnectAndClearFlags, PowerSyncDatabase}; pub use sync::checkpoint::{CheckpointError, CheckpointRequest}; pub use sync::connector::{BackendConnector, PowerSyncCredentials}; pub use sync::options::{CheckpointMode, RequestsCheckpointMode, SyncOptions}; -pub use sync::status::SyncStatusData; +pub use sync::status::{PriorityStatus, SyncStatusData}; pub use sync::stream_priority::StreamPriority; pub mod error; pub mod http; diff --git a/powersync/src/sync/instruction.rs b/powersync/src/sync/instruction.rs index 42ca2c1..a20edee 100644 --- a/powersync/src/sync/instruction.rs +++ b/powersync/src/sync/instruction.rs @@ -4,7 +4,10 @@ use serde::{Deserialize, Serialize, de::IgnoredAny}; use serde_json::value::RawValue; use serde_with::{DisplayFromStr, serde_as}; -use crate::{sync::progress::ProgressCounters, util::SerializedJsonObject}; +use crate::{ + sync::{progress::ProgressCounters, status::PriorityStatus}, + util::SerializedJsonObject, +}; /// An instruction sent by the core extension to the SDK. #[derive(Deserialize, Debug)] @@ -69,6 +72,7 @@ pub struct DownloadSyncStatus { /// Whether we've requested the client SDK to connect to the socket while not receiving sync /// lines yet. pub connecting: bool, + pub priority_status: Vec, pub streams: Vec, pub downloading: Option, diff --git a/powersync/src/sync/status.rs b/powersync/src/sync/status.rs index 7efa7d3..cc8064f 100644 --- a/powersync/src/sync/status.rs +++ b/powersync/src/sync/status.rs @@ -4,12 +4,14 @@ use std::{ Arc, Mutex, atomic::{AtomicBool, Ordering}, }, + time::SystemTime, }; use event_listener::{Event, EventListener}; use powersync_sqlite_nostd::ResultCode; +use serde::Deserialize; -use crate::db::connection::SqliteConnection; +use crate::{StreamPriority, db::connection::SqliteConnection, sync::instruction::Timestamp}; use crate::{ error::PowerSyncError, sync::{ @@ -140,6 +142,35 @@ impl SyncStatusData { &self.downloading.streams } + /// Information about when streams in priorities have last synced. + pub fn priority_status(&self) -> &[PriorityStatus] { + &self.downloading.priority_status + } + + pub fn priority_status_for(&self, priority: StreamPriority) -> Option<&PriorityStatus> { + // Lower-priority values always override higher-priority status entries. The internal + // priority is already sorted in ascending priority order. + for entry in self.priority_status() { + if entry.priority <= priority { + return Some(entry); + } + } + + None + } + + /// Whether all tracked Sync Streams have ever synced. + pub fn has_synced(&self) -> bool { + self.priority_status_for(StreamPriority::SENTINEL) + .is_some_and(|status| status.has_synced()) + } + + /// The last timestamp at which all tracked Sync Streams have synced. + pub fn last_synced_at(&self) -> Option { + self.priority_status_for(StreamPriority::SENTINEL)? + .last_synced_at() + } + /// All sync streams currently being tracked in the database. pub fn streams<'a>(&'a self) -> impl Iterator> + 'a { self.downloading @@ -229,3 +260,27 @@ pub struct SyncStreamStatus<'a> { pub progress: Option, pub subscription: StreamSubscriptionDescription<'a>, } + +/// Information about partial sync. +/// +/// When different Sync Streams have different priorities assigned to them, clients may be in a +/// state where only some of them have synced. This information is exposed on +/// [ActiveStreamSubscription] for affected streams, [DownloadSyncStatus::priority_status] reports +/// it aggregated by priority. +#[derive(Deserialize, Debug)] +pub struct PriorityStatus { + /// The priority for which information is reported. + pub priority: StreamPriority, + /// The last timestamp at which streams with this priority have synced. + last_synced_at: Option, +} + +impl PriorityStatus { + pub fn last_synced_at(&self) -> Option { + Some(self.last_synced_at?.into()) + } + + pub fn has_synced(&self) -> bool { + self.last_synced_at.is_some() + } +} diff --git a/powersync/tests/sync_test.rs b/powersync/tests/sync_test.rs index bee0f06..5d82ceb 100644 --- a/powersync/tests/sync_test.rs +++ b/powersync/tests/sync_test.rs @@ -22,7 +22,7 @@ use powersync_test_utils::{ DatabaseTest, execute, mock_sync_service::TestConnector, query_all, - sync_line::{Checkpoint, SyncLine}, + sync_line::{BucketChecksum, Checkpoint, SyncLine}, }; use rusqlite::params; use serde_json::json; @@ -340,7 +340,46 @@ fn progress_without_priorities() { sync.wait_for_progress("a", 2, 2).await; request.send_checkpoint_complete(oplog_id, None).await; - sync.wait_for_status(|s| !s.is_downloading()).await; + sync.wait_for_status(|s| s.has_synced()).await; + sync.db.status().last_synced_at().unwrap(); + }); +} + +#[test] +fn per_priority_status() { + let sync = SyncStreamTest::new(); + sync.connect(); + + sync.run(async { + let request = sync.test.http.receive_requests.recv().await.unwrap(); + sync.wait_for_status(|s| s.is_connected()).await; + + request + .send_checkpoint(Checkpoint { + last_op_id: 0, + write_checkpoint: None, + buckets: vec![ + BucketChecksum::without_stream("p0", 0, Some(0.try_into().unwrap())), + BucketChecksum::without_stream("p1", 0, Some(1.try_into().unwrap())), + BucketChecksum::without_stream("p2", 0, Some(2.try_into().unwrap())), + BucketChecksum::without_stream("p3", 0, Some(3.try_into().unwrap())), + ], + streams: vec![], + }) + .await; + sync.wait_for_status(|s| s.is_downloading()).await; + + for priority in 0..=3 { + let priority = StreamPriority::try_from(priority).unwrap(); + + assert!(!sync.db.status().priority_status_for(priority).is_some()); + request.send_checkpoint_complete(0, Some(priority)).await; + sync.wait_for_status(|s| s.priority_status_for(priority).is_some()) + .await; + } + + request.send_checkpoint_complete(0, None).await; + sync.wait_for_status(|s| s.has_synced()).await; }); } diff --git a/powersync_test_utils/src/sync_line.rs b/powersync_test_utils/src/sync_line.rs index 37b792e..8a574e7 100644 --- a/powersync_test_utils/src/sync_line.rs +++ b/powersync_test_utils/src/sync_line.rs @@ -97,15 +97,21 @@ impl<'a> BucketChecksum<'a> { is_default: true, errors: Default::default(), }; - let cs = BucketChecksum { + let mut cs = Self::without_stream(name, count, prio); + cs.subscriptions + .push(BucketSubscriptionReason::DerivedFromDefaultStream(0)); + + (stream, cs) + } + + pub fn without_stream(name: &'a str, count: i64, prio: Option) -> Self { + Self { bucket: name, checksum: 0, priority: prio, count: Some(count), - subscriptions: vec![BucketSubscriptionReason::DerivedFromDefaultStream(0)], - }; - - (stream, cs) + subscriptions: vec![], + } } }