Add per-priority sync status

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

powersync/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "powersync"
-version = "0.1.0"
+version = "0.1.1"
 
 edition = "2024"
 license = "Apache-2.0"

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;

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<PriorityStatus>,
     pub streams: Vec<ActiveStreamSubscription>,
     pub downloading: Option<IgnoredAny>,
 

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<SystemTime> {
+        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<Item = SyncStreamStatus<'a>> + 'a {
         self.downloading
@@ -229,3 +260,27 @@ pub struct SyncStreamStatus<'a> {
     pub progress: Option<ProgressCounters>,
     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<Timestamp>,
+}
+
+impl PriorityStatus {
+    pub fn last_synced_at(&self) -> Option<SystemTime> {
+        Some(self.last_synced_at?.into())
+    }
+
+    pub fn has_synced(&self) -> bool {
+        self.last_synced_at.is_some()
+    }
+}

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;
     });
 }
 

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<StreamPriority>) -> Self {
+        Self {
             bucket: name,
             checksum: 0,
             priority: prio,
             count: Some(count),
-            subscriptions: vec![BucketSubscriptionReason::DerivedFromDefaultStream(0)],
-        };
-
-        (stream, cs)
+            subscriptions: vec![],
+        }
     }
 }