Support checkpoint request protocol

powersync/Cargo.toml

@@ -24,6 +24,7 @@ rusqlite = ["dep:rusqlite"]
 ffi = []
 
 [dependencies]
+async-broadcast = "0.7"
 async-channel = "2.5.0"
 async-lock = "3.4.1"
 async-executor = { version = "1.14.0", optional = true }

powersync/src/db/internal.rs

@@ -111,12 +111,53 @@ impl InnerPowerSyncState {
         writer.commit()
     }
 
+    pub async fn seed_checkpoint_request_id(&self, id: i64) -> Result<(), PowerSyncError> {
+        let mut writer = self.writer().await?;
+        let writer = TransactionGuard::new(writer.sqlite_connection_mut())?;
+
+        Self::checkpoint_request_control(&writer, CheckpointCounter::Seed, Some(id))?;
+        writer.commit()
+    }
+
+    pub async fn next_checkpoint_request_id(&self) -> Result<i64, PowerSyncError> {
+        let id = self
+            .read_checkpoint_request_id(CheckpointCounter::Next)
+            .await?;
+        Ok(id.expect("Core extension should return next checkpoint request id"))
+    }
+
+    pub async fn current_checkpoint_request_id(&self) -> Result<Option<i64>, PowerSyncError> {
+        self.read_checkpoint_request_id(CheckpointCounter::Current)
+            .await
+    }
+
     pub fn target_checkpoint_request_id(
         writer: &TransactionGuard,
         update: Option<i64>,
+    ) -> Result<Option<i64>, PowerSyncError> {
+        Self::checkpoint_request_control(writer, CheckpointCounter::Target, update)
+    }
+
+    async fn read_checkpoint_request_id(
+        &self,
+        counter: CheckpointCounter,
+    ) -> Result<Option<i64>, PowerSyncError> {
+        let mut writer = self.writer().await?;
+        let writer = TransactionGuard::new(writer.sqlite_connection_mut())?;
+
+        let id = Self::checkpoint_request_control(&writer, counter, None)?;
+        writer.commit()?;
+        Ok(id)
+    }
+
+    fn checkpoint_request_control(
+        writer: &TransactionGuard,
+        counter: CheckpointCounter,
+        update: Option<i64>,
     ) -> Result<Option<i64>, PowerSyncError> {
         let stmt = writer.inner.prepare("SELECT powersync_control(?, ?);")?;
-        stmt.bind_text(1, "target_checkpoint_request_id", Destructor::STATIC)?;
+
+        stmt.bind_text(1, counter.control_op(), Destructor::STATIC)?;
         if let Some(update) = update {
             stmt.bind_int64(2, update)?;
         } else {
@@ -205,3 +246,22 @@ impl InnerPowerSyncState {
         }
     }
 }
+
+#[derive(Clone, Copy)]
+pub enum CheckpointCounter {
+    Target,
+    Seed,
+    Current,
+    Next,
+}
+
+impl CheckpointCounter {
+    fn control_op(&self) -> &'static str {
+        match self {
+            CheckpointCounter::Target => "target_checkpoint_request_id",
+            CheckpointCounter::Seed => "seed_checkpoint_request_id",
+            CheckpointCounter::Current => "current_checkpoint_request_id",
+            CheckpointCounter::Next => "next_checkpoint_request_id",
+        }
+    }
+}

powersync/src/error.rs

@@ -5,6 +5,8 @@ use std::sync::Arc;
 use std::{borrow::Cow, fmt::Display};
 use thiserror::Error;
 
+use crate::sync::checkpoint::CheckpointError;
+
 pub type Result<T> = std::result::Result<T, PowerSyncError>;
 
 /// A [RawPowerSyncError], but boxed.
@@ -13,7 +15,7 @@ pub type Result<T> = std::result::Result<T, PowerSyncError>;
 /// [RawPowerSyncError] enum type).
 #[derive(Debug, Clone)]
 pub struct PowerSyncError {
-    inner: Arc<RawPowerSyncError>,
+    pub(crate) inner: Arc<RawPowerSyncError>,
 }
 
 impl PowerSyncError {
@@ -51,6 +53,12 @@ impl From<reqwest::Error> for PowerSyncError {
     }
 }
 
+impl From<CheckpointError> for PowerSyncError {
+    fn from(value: CheckpointError) -> Self {
+        RawPowerSyncError::Checkpoint { error: value }.into()
+    }
+}
+
 impl From<RawPowerSyncError> for PowerSyncError {
     fn from(value: RawPowerSyncError) -> Self {
         PowerSyncError {
@@ -122,6 +130,8 @@ pub(crate) enum RawPowerSyncError {
         #[source]
         source: Box<dyn Error + Send + Sync>,
     },
+    #[error("Checkpoint error: {error}")]
+    Checkpoint { error: CheckpointError },
 }
 
 impl From<ResultCode> for PowerSyncError {

powersync/src/lib.rs

@@ -12,7 +12,7 @@ pub use db::streams::StreamSubscription;
 pub use db::streams::StreamSubscriptionOptions;
 pub use db::streams::SyncStream;
 pub use sync::connector::{BackendConnector, PowerSyncCredentials};
-pub use sync::options::SyncOptions;
+pub use sync::options::{CheckpointMode, RequestsCheckpointMode, SyncOptions};
 pub use sync::status::SyncStatusData;
 pub use sync::stream_priority::StreamPriority;
 pub mod error;

powersync/src/sync/checkpoint.rs

@@ -0,0 +1,121 @@
+use std::sync::Arc;
+
+use log::{debug, warn};
+use thiserror::Error;
+
+use crate::{
+    BackendConnector, CheckpointMode, RequestsCheckpointMode, SyncOptions,
+    db::internal::InnerPowerSyncState,
+    error::{PowerSyncError, RawPowerSyncError},
+    sync::{
+        coordinator::SyncChannels, download::http::checkpoint_request,
+        instruction::CheckpointRequestPayload, upload::get_client_id,
+    },
+};
+
+#[derive(Error, Debug)]
+pub enum CheckpointError {
+    #[error(
+        "The PowerSync service does not support checkpoint requests. Update to PowerSync service version 1.24.0 or later to use this API."
+    )]
+    InstanceNotSupported,
+    #[error("Cannot request checkpoints, sync client is disconnected")]
+    Disconnected,
+    #[error("Connected with legacy checkpoint mode, cannot request checkpoints")]
+    Disabled,
+    #[error("Error on sync status before checkpoint was applied: {cause}")]
+    StatusError { cause: PowerSyncError },
+}
+
+pub async fn repost_unacknowledged_checkpoints(
+    db: Arc<InnerPowerSyncState>,
+    channels: SyncChannels,
+    options: SyncOptions,
+) {
+    let CheckpointMode::Requests(requests) = options.checkpoints else {
+        return;
+    };
+
+    loop {
+        // Make sure the system is seeded and ready.
+        let result = repost_unacknowledged_checkpoint_iteration(
+            &db,
+            &channels,
+            options.connector.as_ref(),
+            requests,
+        )
+        .await;
+
+        if let Err(err) = result {
+            if let RawPowerSyncError::Checkpoint {
+                error: CheckpointError::InstanceNotSupported,
+            } = err.inner.as_ref()
+            {
+                return;
+            }
+
+            warn!("Error retrying checkpoint request: {err}");
+            db.env.runtime.delay_once(requests.retry_delay).await;
+        }
+    }
+}
+
+async fn repost_unacknowledged_checkpoint_iteration(
+    db: &InnerPowerSyncState,
+    channels: &SyncChannels,
+    connector: &dyn BackendConnector,
+    mode: RequestsCheckpointMode,
+) -> Result<(), PowerSyncError> {
+    // Make sure the system is seeded and ready
+    channels
+        .checkpoints
+        .wait_for_checkpoint_requests_ready(false)
+        .await?;
+
+    // Get the current checkpoint_request_id
+    let Some(request_id) = db
+        .current_checkpoint_request_id()
+        .await?
+        .take_if(|id| *id > 0)
+    else {
+        // This should not be reached. For completeness sake - wait a bit.
+        db.env.runtime.delay_once(mode.retry_delay).await;
+        return Ok(());
+    };
+
+    // Give the request some time to sync
+    db.env.runtime.delay_once(mode.retry_delay).await;
+
+    if db.current_checkpoint_request_id().await? != Some(request_id) {
+        return Ok(());
+    }
+
+    // If the request was applied, we don't need to retry
+    if db
+        .status
+        .current_snapshot()
+        .is_checkpoint_request_applied(request_id)
+    {
+        return Ok(());
+    }
+
+    // Make sure we are online and ready before making the request
+    channels
+        .checkpoints
+        .wait_for_checkpoint_requests_ready(false)
+        .await?;
+
+    // It's safe if this request races with a new one. The service will reject it.
+    debug!("Retrying checkpoint request id {request_id}");
+    checkpoint_request(
+        &db,
+        connector,
+        &CheckpointRequestPayload {
+            client_id: get_client_id(db).await?,
+            checkpoint_request_id: request_id,
+        },
+    )
+    .await?;
+
+    Ok(())
+}

powersync/src/sync/connector.rs

@@ -1,3 +1,5 @@
+use std::pin::Pin;
+
 use async_trait::async_trait;
 use url::Url;
 
@@ -12,6 +14,26 @@ pub trait BackendConnector: Send + Sync {
 
     /// Inspects completed CRUD transactions on a database and uploads them.
     async fn upload_data(&self) -> Result<(), PowerSyncError>;
+
+    /// This is optional, and should only return a future for connectors capable of requesting
+    /// checkpoints.
+    ///
+    /// For upploads that are processed asynchronously by a backend (for example through a message
+    /// queue): The sync client as part of the PowerSync Rust SDK generates a checkpoint request id
+    /// and hands it to your backend via this function, which is responsible for creaeting a
+    /// matching checkpoint once the uploads preceeding the request have been processed.
+    ///
+    /// For more details, see [asynchronous backend uploads](https://docs.powersync.com/client-sdks/advanced/checkpoint-requests#asynchronous-upload-backends).
+    ///
+    /// To use this connector, using [crate::sync::options::CheckpointMode::Requests] is required.
+    /// Note that this requires PowerSync service version 1.24.0 or later.
+    fn post_checkpoint_request<'a>(
+        &'a self,
+        _client_id: &'a str,
+        _request_id: i64,
+    ) -> Option<Pin<Box<dyn Future<Output = Result<i64, PowerSyncError>> + Send + 'a>>> {
+        None
+    }
 }
 
 /// Credentials used to connect to a PowerSync service instance.

powersync/src/sync/coordinator.rs

@@ -9,7 +9,9 @@ use crate::{
     env::PowerSyncTask,
     error::PowerSyncError,
     sync::{
+        checkpoint::repost_unacknowledged_checkpoints,
         download::{DownloadEvent, download_loop},
+        state::CheckpointStateSignals,
         streams::ChangedSyncSubscriptions,
         upload::crud_upload_loop,
     },
@@ -39,15 +41,21 @@ impl SyncCoordinator {
         ));
         let uploads = db.env.spawn(crud_upload_loop(
             db.clone(),
-            options,
+            options.clone(),
             channels.clone(),
             uploads_receive,
         ));
+        let checkpoints = db.env.spawn(repost_unacknowledged_checkpoints(
+            db.clone(),
+            channels.clone(),
+            options,
+        ));
 
         *guard = Some(SyncTasks {
             channels,
             uploads: Some(uploads),
             downloads: Some(downloads),
+            retried_checkpoints: Some(checkpoints),
         });
     }
 
@@ -107,32 +115,40 @@ struct SyncTasks {
     channels: SyncChannels,
     downloads: Option<PowerSyncTask>,
     uploads: Option<PowerSyncTask>,
+    retried_checkpoints: Option<PowerSyncTask>,
 }
 
 impl SyncTasks {
+    fn tasks(&mut self) -> [&mut Option<PowerSyncTask>; 3] {
+        [
+            &mut self.downloads,
+            &mut self.uploads,
+            &mut self.retried_checkpoints,
+        ]
+    }
+
     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;
+        for maybe_task in self.tasks() {
+            if let Some(task) = maybe_task.take() {
+                task.cancel_and_join().await;
+            }
         }
     }
 }
 
 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();
+        for maybe_task in self.tasks() {
+            if let Some(task) = maybe_task.take() {
+                task.cancel();
+            }
         }
     }
 }
 
 #[derive(Clone)]
 pub struct SyncChannels {
+    pub checkpoints: Arc<CheckpointStateSignals>,
     local_download_events: Sender<DownloadEvent>,
     trigger_upload: Sender<()>,
 }
@@ -144,6 +160,7 @@ impl SyncChannels {
 
         (
             Self {
+                checkpoints: Default::default(),
                 local_download_events: download_send,
                 trigger_upload: uploads_send,
             },