rename to target_checkpoint_request_id

Author: stevensJourney

crates/core/src/crud_vtab.rs

@@ -252,7 +252,7 @@ impl SimpleCrudTransactionMode {
             // 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('local_target_op', {MAX_OP_ID});
+                "INSERT OR REPLACE INTO ps_kv(key, value) VALUES('target_checkpoint_request_id', {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

@@ -469,7 +469,7 @@ DROP TABLE ps_sync_state_old;
         // 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`.
+        // 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`.
@@ -496,7 +496,7 @@ DELETE FROM ps_kv
  WHERE key IN (
    'last_applied_checkpoint_request_id',
    'last_seen_checkpoint_request_id',
-   'local_target_op'
+   'target_checkpoint_request_id'
  );
 
 INSERT INTO ps_kv(key, value)
@@ -512,7 +512,7 @@ SELECT 'last_seen_checkpoint_request_id', last_op
    AND last_op > 0;
 
 INSERT INTO ps_kv(key, value)
-SELECT 'local_target_op', target_op
+SELECT 'target_checkpoint_request_id', target_op
   FROM ps_buckets
  WHERE name = '$local'
    AND target_op > 0;
@@ -531,7 +531,7 @@ ALTER TABLE ps_buckets DROP COLUMN target_op;
         // `$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
+        // 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] = &[
@@ -582,10 +582,10 @@ SELECT '$local', 1, seen, applied, target
     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
+      (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 = 'local_target_op'
+    SELECT 1 FROM ps_kv WHERE key = 'target_checkpoint_request_id'
  )
 ON CONFLICT(name) DO UPDATE SET
   pending_delete = excluded.pending_delete,

crates/core/src/sync/interface.rs

@@ -273,17 +273,17 @@ pub fn register(db: *mut sqlite::sqlite3, state: Rc<DatabaseState>) -> Result<()
                     ctx.result_int64(request_id);
                     return Ok(());
                 }
-                "local_target_op" => {
-                    let target_op = parse_optional_i64_payload(
+                "target_checkpoint_request_id" => {
+                    let target = parse_optional_i64_payload(
                         *payload,
-                        "local target op",
-                        "local target op must be an integer, integer string, or null",
+                        "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_op = adapter.probe_local_target_op(target_op)?;
+                    let previous_target = adapter.probe_target_checkpoint_request_id(target)?;
 
-                    match previous_target_op {
-                        Some(target_op) => ctx.result_int64(target_op),
+                    match previous_target {
+                        Some(target) => ctx.result_int64(target),
                         None => ctx.result_null(),
                     }
 

crates/core/src/sync/storage_adapter.rs

@@ -30,10 +30,10 @@ const LAST_REQUESTED_CHECKPOINT_REQUEST_ID_KEY: &str = "last_requested_checkpoin
 const LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY: &str = "last_seen_checkpoint_request_id";
 const LAST_APPLIED_CHECKPOINT_REQUEST_ID_KEY: &str = "last_applied_checkpoint_request_id";
 
-// Tracks the local target used to block applying downloaded rows while local writes are
-// outstanding. When present, this is normally either the max-op sentinel for pending local writes or
-// a concrete checkpoint request id also stored in LAST_REQUESTED_CHECKPOINT_REQUEST_ID_KEY.
-const LOCAL_TARGET_OP_KEY: &str = "local_target_op";
+// Tracks the target used to block applying downloaded rows while local writes are outstanding.
+// When present, this is normally either the max-op sentinel for pending local writes or a concrete
+// checkpoint request id also stored in LAST_REQUESTED_CHECKPOINT_REQUEST_ID_KEY.
+const TARGET_CHECKPOINT_REQUEST_ID_KEY: &str = "target_checkpoint_request_id";
 
 /// An adapter for storing sync state.
 ///
@@ -509,51 +509,49 @@ WHERE bucket = ?1",
         Ok(())
     }
 
-    pub fn local_state(&self) -> Result<Option<LocalState>, PowerSyncError> {
-        Ok(self
-            .read_i64_kv(LOCAL_TARGET_OP_KEY)?
-            .map(|target_op| LocalState { target_op }))
+    pub fn target_checkpoint_request_id(&self) -> Result<Option<i64>, PowerSyncError> {
+        self.read_i64_kv(TARGET_CHECKPOINT_REQUEST_ID_KEY)
     }
 
-    /// Probes and optionally updates the local target op used to block applying downloaded rows
+    /// Probes and optionally updates the target checkpoint request id used to block applying downloaded rows
     /// while local writes are outstanding.
     ///
     /// In the write-checkpoint flow, callers allocate a checkpoint request id, post it to the
     /// service, and then update this from the max-op sentinel to the concrete checkpoint request id
     /// once the request succeeds. This is also used for older services where the SDK cannot create
     /// checkpoint requests explicitly.
     ///
-    /// The target op can also be used internally as a sentinel value such as max op id while local
+    /// The target can also be used internally as a sentinel value such as max op id while local
     /// writes are pending, so it must not always be interpreted as a checkpoint request id.
     ///
     /// This only updates the apply gate. It does not allocate, seed or overwrite
     /// `last_requested_checkpoint_request_id`, which is managed by `seed_checkpoint_request_id` and
     /// `next_checkpoint_request_id`.
     ///
-    /// Returns the target op value from before this call. When `target_op` is `None`, this only
-    /// reads the current value. A `target_op` of zero clears the stored target, removing the apply
+    /// Returns the target value from before this call. When `target` is `None`, this only
+    /// reads the current value. A `target` of zero clears the stored target, removing the apply
     /// gate entirely; any other value overwrites it.
     ///
     /// Negative values are rejected when parsing the `powersync_control` payload, before this is
     /// called.
-    pub fn probe_local_target_op(
+    pub fn probe_target_checkpoint_request_id(
         &self,
-        target_op: Option<i64>,
+        target: Option<i64>,
     ) -> Result<Option<i64>, PowerSyncError> {
-        let previous_target_op = self.local_state()?.map(|state| state.target_op);
+        let previous_target = self.target_checkpoint_request_id()?;
 
-        let Some(target_op) = target_op else {
-            return Ok(previous_target_op);
+        let Some(target) = target else {
+            return Ok(previous_target);
         };
 
-        if target_op == 0 {
-            self.delete_kv(LOCAL_TARGET_OP_KEY)?;
-            return Ok(previous_target_op);
+        if target == 0 {
+            self.delete_kv(TARGET_CHECKPOINT_REQUEST_ID_KEY)?;
+            return Ok(previous_target);
         }
 
-        self.write_i64_kv(LOCAL_TARGET_OP_KEY, target_op)?;
+        self.write_i64_kv(TARGET_CHECKPOINT_REQUEST_ID_KEY, target)?;
 
-        Ok(previous_target_op)
+        Ok(previous_target)
     }
 
     /// Persists the checkpoint request id observed in a complete sync checkpoint.
@@ -653,10 +651,6 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value",
     }
 }
 
-pub struct LocalState {
-    pub target_op: i64,
-}
-
 pub struct BucketInfo {
     pub id: i64,
     pub last_applied_op: i64,

crates/core/src/sync/streaming_sync.rs

@@ -651,7 +651,7 @@ impl StreamingSyncIteration {
             return Ok(());
         };
 
-        let target_write = self.adapter.local_state()?.map(|e| e.target_op);
+        let target_write = self.adapter.target_checkpoint_request_id()?;
         if checkpoint.write_checkpoint < target_write {
             // Note: None < Some(x). The pending checkpoint does not contain the write
             // checkpoint created during the upload, so we don't have to try applying it, it's

crates/core/src/sync/sync_local.rs

@@ -70,7 +70,7 @@ impl<'a> SyncOperation<'a> {
                 "SELECT 1
 FROM ps_kv AS target
 LEFT JOIN ps_kv AS seen ON seen.key = 'last_seen_checkpoint_request_id'
-WHERE target.key = 'local_target_op'
+WHERE target.key = 'target_checkpoint_request_id'
   AND CAST(target.value AS INTEGER) > COALESCE(CAST(seen.value AS INTEGER), 0)",
             )?;
 

dart/test/crud_test.dart

@@ -249,7 +249,7 @@ void main() {
           });
         });
 
-        test('updates local target op and updated rows', () {
+        test('updates target checkpoint request id and updated rows', () {
           // Stale high-water marks (e.g. from before a request counter restart) must be cleared
           // by a local write, so they can't open the apply gate for a smaller new target id.
           db.execute('''
@@ -274,10 +274,10 @@ INSERT INTO ps_kv(key, value) VALUES
               isEmpty);
           expect(
               db.select(
-                  "SELECT key, value FROM ps_kv WHERE key LIKE '%checkpoint_request_id' OR key = 'local_target_op'"),
+                  "SELECT key, value FROM ps_kv WHERE key LIKE '%checkpoint_request_id'"),
               [
                 {
-                  'key': 'local_target_op',
+                  'key': 'target_checkpoint_request_id',
                   'value': 9223372036854775807,
                 }
               ]);

dart/test/migration_test.dart

@@ -109,7 +109,7 @@ VALUES(1, '$local', 5, 6, 7, 0, 0, 1, 0, 0, 0);
       expect(db.select('SELECT key, value FROM ps_kv ORDER BY key'), [
         {'key': 'last_applied_checkpoint_request_id', 'value': 5},
         {'key': 'last_seen_checkpoint_request_id', 'value': 6},
-        {'key': 'local_target_op', 'value': 7},
+        {'key': 'target_checkpoint_request_id', 'value': 7},
       ]);
       expect(db.select(r"SELECT * FROM ps_buckets WHERE name = '$local'"),
           isEmpty);
@@ -142,7 +142,7 @@ VALUES(1, '$local', 5, 6, 9223372036854775807, 0, 0, 1, 0, 0, 0);
       expect(db.select('SELECT key, value FROM ps_kv ORDER BY key'), [
         {'key': 'last_applied_checkpoint_request_id', 'value': 5},
         {'key': 'last_seen_checkpoint_request_id', 'value': 6},
-        {'key': 'local_target_op', 'value': 9223372036854775807},
+        {'key': 'target_checkpoint_request_id', 'value': 9223372036854775807},
       ]);
     });
 
@@ -159,7 +159,7 @@ VALUES(1, '$local', 0, 0, 9223372036854775807, 0, 0, 1, 0, 0, 0);
       // The max-op sentinel is valid local target state, but target ops no longer seed
       // last_requested_checkpoint_request_id.
       expect(db.select('SELECT key, value FROM ps_kv ORDER BY key'), [
-        {'key': 'local_target_op', 'value': 9223372036854775807},
+        {'key': 'target_checkpoint_request_id', 'value': 9223372036854775807},
       ]);
     });
 
@@ -170,7 +170,7 @@ INSERT INTO ps_kv(key, value) VALUES
   ('last_requested_checkpoint_request_id', 7),
   ('last_seen_checkpoint_request_id', 6),
   ('last_applied_checkpoint_request_id', 5),
-  ('local_target_op', 7);
+  ('target_checkpoint_request_id', 7);
 ''');
 
       db.executeInTx('select powersync_test_migration(13)');
@@ -196,7 +196,7 @@ INSERT INTO ps_kv(key, value) VALUES
   ('last_requested_checkpoint_request_id', 7),
   ('last_seen_checkpoint_request_id', 6),
   ('last_applied_checkpoint_request_id', 5),
-  ('local_target_op', 7);
+  ('target_checkpoint_request_id', 7);
 ''');
 
       db.executeInTx('select powersync_test_migration(13)');
@@ -216,7 +216,7 @@ UPDATE ps_buckets
         {'key': 'last_applied_checkpoint_request_id', 'value': 8},
         {'key': 'last_requested_checkpoint_request_id', 'value': 7},
         {'key': 'last_seen_checkpoint_request_id', 'value': 8},
-        {'key': 'local_target_op', 'value': 9},
+        {'key': 'target_checkpoint_request_id', 'value': 9},
       ]);
       expect(db.select(r"SELECT * FROM ps_buckets WHERE name = '$local'"),
           isEmpty);