feat: checkpoint requests

crates/core/src/crud_vtab.rs

@@ -247,7 +247,15 @@ 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})"))?;
+            // Also clear the seen/applied high-water marks: checkpoint request ids observed before
+            // this write can't acknowledge it, and stale values may predate a request counter
+            // restart. Keeping them around could open the apply gate for a newly allocated target
+            // id that compares below a stale seen value. The legacy `$local` bookkeeping had the
+            // same behavior by resetting the entire row on local writes.
+            db.exec_safe(formatcp!(
+                "INSERT OR REPLACE INTO ps_kv(key, value) VALUES('local_target_op', {MAX_OP_ID});
+DELETE FROM ps_kv WHERE key IN ('last_seen_checkpoint_request_id', 'last_applied_checkpoint_request_id')"
+            ))?;
             self.had_writes = true;
         }
 

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,148 @@ 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 legacy write checkpoint id or a sentinel such
+        // as i64::MAX while local writes are pending. Store it separately as `local_target_op`.
+        // Seeding `last_requested_checkpoint_request_id` from a concrete target would be possible,
+        // but should be redundant because SDKs reconcile the request counter with service state on
+        // connect before advancing it through `next_checkpoint_request_id`.
+        //
+        // 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.
+        //
+        // This migration can also run on a database that was previously on version 14 and then
+        // downgraded: the down migration rebuilds the `$local` row from ps_kv but keeps the ps_kv
+        // keys around, and an older SDK may have advanced `$local` since. Clear the keys first so
+        // the `$local` row is the source of truth and the inserts below can't conflict.
+        //
+        // After copying, the `$local` row is deleted: version 14 tracks this state exclusively in
+        // ps_kv, so ps_buckets only contains real sync buckets. The down migration recreates the
+        // row from ps_kv when needed.
+        //
+        // DROP COLUMN requires SQLite 3.35+; the extension already refuses to load below
+        // MIN_SQLITE_VERSION_NUMBER (3.44), so this is safe in the up path.
+        let up = "\
+DELETE FROM ps_kv
+ WHERE key IN (
+   'last_applied_checkpoint_request_id',
+   'last_seen_checkpoint_request_id',
+   'local_target_op'
+ );
+
+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 'local_target_op', target_op
+  FROM ps_buckets
+ WHERE name = '$local'
+   AND target_op > 0;
+
+DELETE FROM ps_buckets WHERE name = '$local';
+
+ALTER TABLE ps_buckets DROP COLUMN target_op;
+";
+        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] = &[
+            "ALTER TABLE ps_buckets RENAME TO ps_buckets_14",
+            "DROP INDEX ps_buckets_name",
+            "CREATE TABLE ps_buckets(
+  id INTEGER PRIMARY KEY,
+  name TEXT NOT NULL,
+  last_applied_op INTEGER NOT NULL DEFAULT 0,
+  last_op INTEGER NOT NULL DEFAULT 0,
+  target_op INTEGER NOT NULL DEFAULT 0,
+  add_checksum INTEGER NOT NULL DEFAULT 0,
+  op_checksum INTEGER NOT NULL DEFAULT 0,
+  pending_delete INTEGER NOT NULL DEFAULT 0
+) STRICT",
+            "CREATE UNIQUE INDEX ps_buckets_name ON ps_buckets (name)",
+            "ALTER TABLE ps_buckets ADD COLUMN count_at_last INTEGER NOT NULL DEFAULT 0",
+            "ALTER TABLE ps_buckets ADD COLUMN count_since_last INTEGER NOT NULL DEFAULT 0",
+            "ALTER TABLE ps_buckets ADD COLUMN downloaded_size INTEGER NOT NULL DEFAULT 0",
+            "INSERT INTO ps_buckets(
+  id,
+  name,
+  last_applied_op,
+  last_op,
+  add_checksum,
+  op_checksum,
+  pending_delete,
+  count_at_last,
+  count_since_last,
+  downloaded_size
+)
+SELECT
+  id,
+  name,
+  last_applied_op,
+  last_op,
+  add_checksum,
+  op_checksum,
+  pending_delete,
+  count_at_last,
+  count_since_last,
+  downloaded_size
+FROM ps_buckets_14",
+            "DROP TABLE ps_buckets_14",
+            "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

@@ -12,6 +12,7 @@ use crate::sync::diagnostics::{DiagnosticOptions, DiagnosticsEvent};
 use crate::sync::subscriptions::{StreamKey, apply_subscriptions};
 use alloc::borrow::Cow;
 use alloc::boxed::Box;
+use alloc::format;
 use alloc::rc::Rc;
 use alloc::{string::String, vec::Vec};
 use powersync_sqlite_nostd::bindings::SQLITE_RESULT_SUBTYPE;
@@ -76,6 +77,12 @@ pub enum SyncControlRequest<'a> {
     StartSyncStream(StartSyncStream),
     /// The client requests to stop the current sync iteration.
     StopSyncStream,
+    /// The client requests a new checkpoint request id.
+    NextCheckpointRequestId,
+    /// The client probes and optionally updates the local target op.
+    ///
+    /// This can run outside of a sync iteration and does not affect it.
+    ProbeLocalTargetOp { target_op: Option<i64> },
     /// The client is forwading a sync event to the core extension.
     SyncEvent(SyncEvent<'a>),
 }
@@ -89,6 +96,10 @@ pub enum SyncEvent<'a> {
     ///
     /// In response, we'll stop the current iteration to begin another one with the new token.
     DidRefreshToken,
+    /// Seeds the checkpoint request counter from service state.
+    SeedCheckpointRequestId {
+        request_id: Option<i64>,
+    },
     /// Notifies the sync client that the current CRUD upload (for which the client SDK is
     /// responsible) has finished.
     ///
@@ -128,13 +139,29 @@ pub enum Instruction {
     },
     /// Connect to the sync service using the [StreamingSyncRequest] created by the core extension,
     /// and then forward received lines via [SyncEvent::TextLine] and [SyncEvent::BinaryLine].
-    EstablishSyncStream { request: StreamingSyncRequest },
+    EstablishSyncStream {
+        request: StreamingSyncRequest,
+        /// The latest checkpoint request id known locally before opening this stream.
+        ///
+        /// This is simply the client's current counter state. SDKs use it on every connection
+        /// attempt to re-affirm checkpoint request state with the service, which may have deleted
+        /// its record. The re-affirmation works bidirectionally: it can restore the service-side
+        /// value from this hint or bump the local counter from the service's response, which is
+        /// reported back with `seed_checkpoint_request_id`.
+        last_checkpoint_request_id: Option<i64>,
+    },
     FetchCredentials {
         /// Whether the credentials currently used have expired.
         ///
         /// If false, this is a pre-fetch.
         did_expire: bool,
     },
+    /// Return a newly allocated checkpoint request id to the SDK.
+    CheckpointRequestId { request_id: i64 },
+    /// Notify the SDK that a checkpoint request id has been applied locally.
+    CheckpointRequestApplied { request_id: i64 },
+    /// Return the local target op value observed before an optional update.
+    LocalTargetOp { target_op: Option<i64> },
     // These are defined like this because deserializers in Kotlin can't support either an
     // object or a literal value
     /// Close the websocket / HTTP stream to the sync service.
@@ -232,6 +259,14 @@ pub fn register(db: *mut sqlite::sqlite3, state: Rc<DatabaseState>) -> Result<()
                     }
                 }),
                 "stop" => SyncControlRequest::StopSyncStream,
+                "next_checkpoint_request_id" => SyncControlRequest::NextCheckpointRequestId,
+                "local_target_op" => SyncControlRequest::ProbeLocalTargetOp {
+                    target_op: parse_optional_i64_payload(
+                        *payload,
+                        "local target op",
+                        "local target op must be an integer, integer string, or null",
+                    )?,
+                },
                 "line_text" => SyncControlRequest::SyncEvent(SyncEvent::TextLine {
                     data: if payload.value_type() == ColumnType::Text {
                         payload.text()
@@ -251,6 +286,15 @@ pub fn register(db: *mut sqlite::sqlite3, state: Rc<DatabaseState>) -> Result<()
                     },
                 }),
                 "refreshed_token" => SyncControlRequest::SyncEvent(SyncEvent::DidRefreshToken),
+                "seed_checkpoint_request_id" => {
+                    SyncControlRequest::SyncEvent(SyncEvent::SeedCheckpointRequestId {
+                        request_id: parse_optional_i64_payload(
+                            *payload,
+                            "checkpoint request id",
+                            "checkpoint request id must be an integer, integer string, or null",
+                        )?,
+                    })
+                }
                 "completed_upload" => SyncControlRequest::SyncEvent(SyncEvent::UploadFinished),
                 "update_subscriptions" => {
                     SyncControlRequest::SyncEvent(SyncEvent::DidUpdateSubscriptions {
@@ -345,3 +389,27 @@ create_sqlite_text_fn!(
     powersync_offline_sync_status_impl,
     "powersync_offline_sync_status"
 );
+
+fn parse_optional_i64_payload(
+    payload: *mut sqlite::value,
+    name: &'static str,
+    type_error: &'static str,
+) -> Result<Option<i64>, PowerSyncError> {
+    let value = match payload.value_type() {
+        ColumnType::Null => return Ok(None),
+        ColumnType::Integer => payload.int64(),
+        ColumnType::Text => payload
+            .text()
+            .parse::<i64>()
+            .map_err(|_| PowerSyncError::argument_error(type_error))?,
+        _ => return Err(PowerSyncError::argument_error(type_error)),
+    };
+
+    if value < 0 {
+        return Err(PowerSyncError::argument_error(format!(
+            "{name} must be a non-negative integer"
+        )));
+    }
+
+    Ok(Some(value))
+}