feat: checkpoint requests

.github/actions/xcframework/action.yml

@@ -24,8 +24,7 @@ runs:
     - name: Set up XCode
       uses: maxim-lobanov/setup-xcode@ed7a3b1fda3918c0306d1b724322adc0b8cc0a90 # v1.7.0
       with:
-        # TODO: Update to latest-stable once GH installs iOS 26 simulators
-        xcode-version: '^16.4.0'
+        xcode-version: latest-stable
 
     - name: List simulators
       shell: bash

crates/core/src/crud_vtab.rs

@@ -15,6 +15,10 @@ use crate::error::PowerSyncError;
 use crate::ext::SafeManagedStmt;
 use crate::schema::TableInfoFlags;
 use crate::state::DatabaseState;
+use crate::sync::storage_adapter::{
+    LAST_APPLIED_CHECKPOINT_REQUEST_ID_KEY, LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY,
+    TARGET_CHECKPOINT_REQUEST_ID_KEY,
+};
 use crate::utils::MAX_OP_ID;
 use crate::vtab_util::*;
 
@@ -29,7 +33,7 @@ const SIMPLE_NAME: &CStr = c"powersync_crud";
 // ps_crud(tx_id, data).
 // The second form (without the trailing underscore) takes the data to insert as individual
 // components and constructs the data to insert into `ps_crud` internally. It will also update
-// `ps_updated_rows` and the `$local` bucket.
+// `ps_updated_rows` and the local write apply gate.
 //
 // Using a virtual table like this allows us to hook into xBegin, xCommit and xRollback to automatically
 // increment transaction ids. These are only called when powersync_crud_ is used as part of a transaction,
@@ -166,7 +170,7 @@ impl VirtualTable {
                 stmt.bind_text(2, &serialized, sqlite::Destructor::STATIC)?;
                 stmt.exec()?;
 
-                // However, we also set ps_updated_rows and update the $local bucket
+                // However, we also set ps_updated_rows and mark the local write state.
                 let set_updated_rows = simple.set_updated_rows_statement(db)?;
                 set_updated_rows.bind_text(1, row_type, sqlite::Destructor::STATIC)?;
                 set_updated_rows.bind_text(2, id, sqlite::Destructor::STATIC)?;
@@ -247,7 +251,14 @@ 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.
+            db.exec_safe(formatcp!(
+                "INSERT OR REPLACE INTO ps_kv(key, value) VALUES('{TARGET_CHECKPOINT_REQUEST_ID_KEY}', {MAX_OP_ID});
+DELETE FROM ps_kv WHERE key IN ('{LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY}', '{LAST_APPLIED_CHECKPOINT_REQUEST_ID_KEY}')"
+            ))?;
             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 `target_checkpoint_request_id`.
+        // 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',
+   'target_checkpoint_request_id'
+ );
+
+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 'target_checkpoint_request_id', 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 `target_checkpoint_request_id` 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 = 'target_checkpoint_request_id') AS target
+  )
+ WHERE EXISTS (
+    SELECT 1 FROM ps_kv WHERE key = 'target_checkpoint_request_id'
+ )
+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/schema/raw_table.rs

@@ -267,8 +267,7 @@ pub fn generate_raw_table_trigger(
             // Prevent illegal writes to a table marked as insert-only by raising errors here.
             buffer.push_str("SELECT RAISE(FAIL, 'Unexpected update on insert-only table');\n");
         } else {
-            // Write directly to powersync_crud_ to skip writing the $local bucket for insert-only
-            // tables.
+            // Insert-only tables use manual CRUD writes so they don't block incoming data.
             let fragment = table_columns_to_json_object("NEW", &as_schema_table)?;
             buffer.powersync_crud_manual_put(&table.name, &fragment);
         }

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;
@@ -128,21 +129,33 @@ 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,
     },
-    // 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.
     CloseSyncStream(CloseSyncStream),
     /// Flush the file-system if it's non-durable (only applicable to the Dart SDK).
     FlushFileSystem {},
     /// Notify that a sync has been completed, prompting client SDKs to clear earlier errors.
-    DidCompleteSync {},
+    DidCompleteSync {
+        /// The checkpoint request id applied by this completed sync, if the checkpoint had one.
+        #[serde(skip_serializing_if = "Option::is_none")]
+        applied_checkpoint_request_id: Option<i64>,
+    },
 
     /// Handle a diagnostic event.
     ///
@@ -232,6 +245,77 @@ pub fn register(db: *mut sqlite::sqlite3, state: Rc<DatabaseState>) -> Result<()
                     }
                 }),
                 "stop" => SyncControlRequest::StopSyncStream,
+                "seed_checkpoint_request_id" => {
+                    let has_sync_iteration = {
+                        let client = state.sync_client.borrow();
+                        client
+                            .as_ref()
+                            .map(|client| client.has_sync_iteration())
+                            .unwrap_or(false)
+                    };
+
+                    if !has_sync_iteration {
+                        return Err(PowerSyncError::state_error("No iteration is active"));
+                    }
+
+                    let request_id = parse_positive_i64_payload(
+                        *payload,
+                        "checkpoint request id",
+                        "checkpoint request id must be an integer or integer string",
+                    )?;
+                    let adapter = state.storage_adapter(db)?;
+                    adapter.seed_checkpoint_request_id(request_id)?;
+                    ctx.result_int64(request_id);
+                    return Ok(());
+                }
+                "next_checkpoint_request_id" => {
+                    let has_sync_iteration = {
+                        let client = state.sync_client.borrow();
+                        client
+                            .as_ref()
+                            .map(|client| client.has_sync_iteration())
+                            .unwrap_or(false)
+                    };
+
+                    if !has_sync_iteration {
+                        return Err(PowerSyncError::state_error("No iteration is active"));
+                    }
+
+                    let adapter = state.storage_adapter(db)?;
+                    if !adapter.has_checkpoint_request_id()? {
+                        return Err(PowerSyncError::state_error(
+                            "Checkpoint request state has not been seeded",
+                        ));
+                    }
+
+                    let request_id = adapter.next_checkpoint_request_id()?;
+                    ctx.result_int64(request_id);
+                    return Ok(());
+                }
+                "target_checkpoint_request_id" => {
+                    let target = parse_optional_i64_payload(
+                        *payload,
+                        "target checkpoint request id",
+                        "target checkpoint request id must be an integer, integer string, or null",
+                    )?;
+                    let adapter = state.storage_adapter(db)?;
+                    let previous_target = adapter.probe_target_checkpoint_request_id(target)?;
+
+                    match previous_target {
+                        Some(target) => ctx.result_int64(target),
+                        None => ctx.result_null(),
+                    }
+
+                    return Ok(());
+                }
+                "current_checkpoint_request_id" => {
+                    let adapter = state.storage_adapter(db)?;
+                    match adapter.last_checkpoint_request_id()? {
+                        Some(request_id) => ctx.result_int64(request_id),
+                        None => ctx.result_null(),
+                    }
+                    return Ok(());
+                }
                 "line_text" => SyncControlRequest::SyncEvent(SyncEvent::TextLine {
                     data: if payload.value_type() == ColumnType::Text {
                         payload.text()
@@ -345,3 +429,47 @@ 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(),
+        // Allow decimal strings as a fallback for JavaScript SQLite drivers that can't bind
+        // 64-bit integers as BigInt without losing precision through Number.
+        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))
+}
+
+fn parse_positive_i64_payload(
+    payload: *mut sqlite::value,
+    name: &'static str,
+    type_error: &'static str,
+) -> Result<i64, PowerSyncError> {
+    let Some(value) = parse_optional_i64_payload(payload, name, type_error)? else {
+        return Err(PowerSyncError::argument_error(type_error));
+    };
+
+    if value == 0 {
+        return Err(PowerSyncError::argument_error(format!(
+            "{name} must be a positive integer"
+        )));
+    }
+
+    Ok(value)
+}