feat: checkpoint requests

crates/core/src/crud_vtab.rs

@@ -247,7 +247,9 @@ impl SimpleCrudTransactionMode {
 
     fn record_local_write(&mut self, db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
         if !self.had_writes {
-            db.exec_safe(formatcp!("INSERT OR REPLACE INTO ps_buckets(name, last_op, target_op) VALUES('$local', 0, {MAX_OP_ID})"))?;
+            db.exec_safe(formatcp!(
+                "INSERT OR REPLACE INTO ps_kv(key, value) VALUES('local_target_op', {MAX_OP_ID})"
+            ))?;
             self.had_writes = true;
         }
 

crates/core/src/macros.rs

@@ -43,3 +43,49 @@ macro_rules! create_sqlite_optional_text_fn {
         }
     };
 }
+
+#[macro_export]
+macro_rules! create_sqlite_int_fn {
+    ($fn_name:ident, $fn_impl_name:ident, $description:literal) => {
+        extern "C" fn $fn_name(
+            ctx: *mut sqlite::context,
+            argc: c_int,
+            argv: *mut *mut sqlite::value,
+        ) {
+            let args = sqlite::args!(argc, argv);
+
+            let result = $fn_impl_name(ctx, args);
+
+            if let Err(err) = result {
+                PowerSyncError::from(err).apply_to_ctx($description, ctx);
+            } else if let Ok(r) = result {
+                ctx.result_int64(r);
+            }
+        }
+    };
+}
+
+#[macro_export]
+macro_rules! create_sqlite_optional_int_fn {
+    ($fn_name:ident, $fn_impl_name:ident, $description:literal) => {
+        extern "C" fn $fn_name(
+            ctx: *mut sqlite::context,
+            argc: c_int,
+            argv: *mut *mut sqlite::value,
+        ) {
+            let args = sqlite::args!(argc, argv);
+
+            let result = $fn_impl_name(ctx, args);
+
+            if let Err(err) = result {
+                PowerSyncError::from(err).apply_to_ctx($description, ctx);
+            } else if let Ok(r) = result {
+                if let Some(i) = r {
+                    ctx.result_int64(i);
+                } else {
+                    ctx.result_null();
+                }
+            }
+        }
+    };
+}

crates/core/src/migrations.rs

@@ -16,7 +16,7 @@ use crate::fix_data::apply_v035_fix;
 use crate::schema::inspection::ExistingView;
 use crate::sync::BucketPriority;
 
-pub const LATEST_VERSION: i32 = 13;
+pub const LATEST_VERSION: i32 = 14;
 
 pub fn powersync_migrate(
     ctx: *mut sqlite::context,
@@ -460,6 +460,91 @@ DROP TABLE ps_sync_state_old;
         track_migration.exec()?;
     }
 
+    if current_version < 14 && target_version >= 14 {
+        // Move the legacy `$local` checkpoint bookkeeping into ps_kv.
+        //
+        // In older databases, `$local.last_applied_op` represented the latest legacy write
+        // checkpoint that was actually applied, so it becomes the last applied checkpoint request.
+        // `$local.last_op` represented the latest legacy write checkpoint seen in the sync stream,
+        // so it becomes the last seen checkpoint request.
+        //
+        // `$local.target_op` can either be a concrete checkpoint request id or a sentinel such as
+        // i64::MAX while local writes are pending. Store it separately as `local_target_op`, but
+        // only treat concrete values as requested checkpoint ids. We intentionally don't seed
+        // `last_requested_checkpoint_request_id` from `$local.last_applied_op` because that is an
+        // applied value, not necessarily the current requested target.
+        //
+        // An absent local target can safely start client-created checkpoint requests from 1. The
+        // ambiguous case is an existing max-op local target without a concrete requested id:
+        // pending local writes may already be associated with legacy service-created write
+        // checkpoints, so SDKs should bridge once through the legacy endpoint before starting
+        // client-created checkpoint requests.
+        let up = "\
+INSERT INTO ps_kv(key, value)
+SELECT 'last_applied_checkpoint_request_id', last_applied_op
+  FROM ps_buckets
+ WHERE name = '$local'
+   AND last_applied_op > 0;
+
+INSERT INTO ps_kv(key, value)
+SELECT 'last_seen_checkpoint_request_id', last_op
+  FROM ps_buckets
+ WHERE name = '$local'
+   AND last_op > 0;
+
+INSERT INTO ps_kv(key, value)
+SELECT 'last_requested_checkpoint_request_id', target_op
+  FROM ps_buckets
+ WHERE name = '$local'
+   AND target_op > 0
+   AND target_op != 9223372036854775807;
+
+INSERT INTO ps_kv(key, value)
+SELECT 'local_target_op', target_op
+  FROM ps_buckets
+ WHERE name = '$local'
+   AND target_op > 0;
+";
+        local_db.exec_safe(up).into_db_result(local_db)?;
+
+        // Downgrading needs to rebuild the old `$local` row from the new ps_kv state so older SDKs
+        // can keep using their target-op based blocking behavior. In that model, `$local.last_op`
+        // tracked the latest seen legacy write checkpoint and was compared with `$local.target_op`
+        // to decide whether downloaded changes could be applied. The `$local.last_applied_op`
+        // value represented the checkpoint that had actually been applied locally.
+        // `$local.pending_delete = 1` marked this as a synthetic local-only bucket instead of a
+        // normal service bucket. Restore each old progress column from its matching ps_kv key.
+        // The 0 defaults cover a local target that exists before any checkpoint has been seen or
+        // applied. If `local_target_op` is absent, don't create a `$local` row: the old
+        // implementation also didn't have a `$local` bucket unless there was local target state to
+        // track.
+        const DOWN_STATEMENTS: &[&str] = &[
+            "INSERT INTO ps_buckets(name, pending_delete, last_op, last_applied_op, target_op)
+SELECT '$local', 1, seen, applied, target
+  FROM (
+    SELECT
+      IFNULL((SELECT CAST(value AS INTEGER) FROM ps_kv WHERE key = 'last_seen_checkpoint_request_id'), 0) AS seen,
+      IFNULL((SELECT CAST(value AS INTEGER) FROM ps_kv WHERE key = 'last_applied_checkpoint_request_id'), 0) AS applied,
+      (SELECT CAST(value AS INTEGER) FROM ps_kv WHERE key = 'local_target_op') AS target
+  )
+ WHERE EXISTS (
+    SELECT 1 FROM ps_kv WHERE key = 'local_target_op'
+ )
+ON CONFLICT(name) DO UPDATE SET
+  pending_delete = excluded.pending_delete,
+  last_op = excluded.last_op,
+  last_applied_op = excluded.last_applied_op,
+  target_op = excluded.target_op",
+            "DELETE FROM ps_migration WHERE id >= 14",
+        ];
+        let down = serialize_down_statements(DOWN_STATEMENTS)?;
+        let track_migration =
+            local_db.prepare_v2("INSERT INTO ps_migration(id, down_migrations) VALUES (?, ?)")?;
+        track_migration.bind_int(1, 14)?;
+        track_migration.bind_text(2, &down, Destructor::STATIC)?;
+        track_migration.exec()?;
+    }
+
     Ok(())
 }
 

crates/core/src/sync/interface.rs

@@ -4,12 +4,12 @@ use core::ffi::{c_int, c_void};
 use super::streaming_sync::SyncClient;
 use super::sync_status::DownloadSyncStatus;
 use crate::constants::SUBTYPE_JSON;
-use crate::create_sqlite_text_fn;
 use crate::error::PowerSyncError;
 use crate::schema::Schema;
 use crate::state::DatabaseState;
 use crate::sync::diagnostics::{DiagnosticOptions, DiagnosticsEvent};
 use crate::sync::subscriptions::{StreamKey, apply_subscriptions};
+use crate::{create_sqlite_int_fn, create_sqlite_optional_int_fn, create_sqlite_text_fn};
 use alloc::borrow::Cow;
 use alloc::boxed::Box;
 use alloc::rc::Rc;
@@ -324,6 +324,28 @@ pub fn register(db: *mut sqlite::sqlite3, state: Rc<DatabaseState>) -> Result<()
         Some(DatabaseState::destroy_rc),
     )?;
 
+    db.create_function_v2(
+        "powersync_probe_local_target_op",
+        1,
+        sqlite::UTF8 | sqlite::DIRECTONLY,
+        Some(Rc::into_raw(state.clone()) as *mut c_void),
+        Some(powersync_probe_local_target_op),
+        None,
+        None,
+        Some(DatabaseState::destroy_rc),
+    )?;
+
+    db.create_function_v2(
+        "powersync_next_checkpoint_request_id",
+        0,
+        sqlite::UTF8 | sqlite::DIRECTONLY,
+        Some(Rc::into_raw(state) as *mut c_void),
+        Some(powersync_next_checkpoint_request_id),
+        None,
+        None,
+        Some(DatabaseState::destroy_rc),
+    )?;
+
     Ok(())
 }
 
@@ -340,8 +362,74 @@ fn powersync_offline_sync_status_impl(
     Ok(serialized)
 }
 
+fn powersync_probe_local_target_op_impl(
+    ctx: *mut sqlite::context,
+    args: &[*mut sqlite::value],
+) -> Result<Option<i64>, PowerSyncError> {
+    if args.len() != 1 {
+        return Err(PowerSyncError::argument_error(
+            "powersync_probe_local_target_op takes one argument",
+        ));
+    }
+
+    let arg = args[0];
+    let new_target_op =
+        match arg.value_type() {
+            ColumnType::Null => None,
+            ColumnType::Integer => Some(arg.int64()),
+            ColumnType::Text => Some(arg.text().parse::<i64>().map_err(|_| {
+                PowerSyncError::argument_error("target op must be an integer string")
+            })?),
+            _ => {
+                return Err(PowerSyncError::argument_error(
+                    "target op must be an integer, integer string, or null",
+                ));
+            }
+        };
+
+    let db = ctx.db_handle();
+
+    if new_target_op.is_some() {
+        verify_in_transaction(db)?;
+    }
+
+    let db_state = unsafe { DatabaseState::from_context(&ctx) };
+    let adapter = db_state.storage_adapter(db)?;
+    adapter.probe_local_target_op(new_target_op)
+}
+
+fn powersync_next_checkpoint_request_id_impl(
+    ctx: *mut sqlite::context,
+    args: &[*mut sqlite::value],
+) -> Result<i64, PowerSyncError> {
+    if !args.is_empty() {
+        return Err(PowerSyncError::argument_error(
+            "powersync_next_checkpoint_request_id does not take arguments",
+        ));
+    }
+
+    let db = ctx.db_handle();
+    verify_in_transaction(db)?;
+
+    let db_state = unsafe { DatabaseState::from_context(&ctx) };
+    let adapter = db_state.storage_adapter(db)?;
+    adapter.next_checkpoint_request_id()
+}
+
 create_sqlite_text_fn!(
     powersync_offline_sync_status,
     powersync_offline_sync_status_impl,
     "powersync_offline_sync_status"
 );
+
+create_sqlite_optional_int_fn!(
+    powersync_probe_local_target_op,
+    powersync_probe_local_target_op_impl,
+    "powersync_probe_local_target_op"
+);
+
+create_sqlite_int_fn!(
+    powersync_next_checkpoint_request_id,
+    powersync_next_checkpoint_request_id_impl,
+    "powersync_next_checkpoint_request_id"
+);