remove dead macros. Update migrations to fix potential bugs. update doc entries

Author: stevensJourney

crates/core/src/crud_vtab.rs

@@ -247,8 +247,15 @@ impl SimpleCrudTransactionMode {
 
     fn record_local_write(&mut self, db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
         if !self.had_writes {
+            // Also clear the seen/applied high-water marks: checkpoint request ids observed before
+            // this write can't acknowledge it, and they may come from an incompatible id namespace
+            // (legacy write checkpoints migrated by v14, or state from before a 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})"
+                "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/macros.rs

@@ -43,49 +43,3 @@ 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

@@ -468,18 +468,34 @@ DROP TABLE ps_sync_state_old;
         // `$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.
+        // `$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.
         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
@@ -492,19 +508,14 @@ SELECT 'last_seen_checkpoint_request_id', last_op
  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;
 
+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)?;

crates/core/src/sync/storage_adapter.rs

@@ -523,13 +523,15 @@ WHERE bucket = ?1",
     /// The target op 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.
     ///
-    /// When the target op is a positive, non-sentinel checkpoint request id, it also updates
-    /// `last_requested_checkpoint_request_id` so clients can migrate from the legacy target-op
-    /// flow to client-created checkpoint requests. `0` clears the local target, and sentinel
-    /// values such as max op id must not update the last requested 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.
+    ///
+    /// Negative values are rejected when parsing the `powersync_control` payload, before this is
+    /// called.
     pub fn probe_local_target_op(
         &self,
         target_op: Option<i64>,
@@ -540,25 +542,13 @@ WHERE bucket = ?1",
             return Ok(previous_target_op);
         };
 
-        if target_op < 0 {
-            return Err(PowerSyncError::argument_error(
-                "target op must be a non-negative integer",
-            ));
-        }
-
         if target_op == 0 {
             self.delete_kv(LOCAL_TARGET_OP_KEY)?;
             return Ok(previous_target_op);
         }
 
         self.write_i64_kv(LOCAL_TARGET_OP_KEY, target_op)?;
 
-        // Concrete target ops also seed the request counter for clients migrating from legacy
-        // service-created write checkpoints to client-created checkpoint requests.
-        if target_op != i64::MAX {
-            self.write_i64_kv(LAST_REQUESTED_CHECKPOINT_REQUEST_ID_KEY, target_op)?;
-        }
-
         Ok(previous_target_op)
     }
 

dart/test/crud_test.dart

@@ -250,6 +250,14 @@ void main() {
         });
 
         test('updates local target op and updated rows', () {
+          // Stale high-water marks (e.g. migrated legacy write checkpoints) must be cleared by a
+          // local write, so they can't open the apply gate for a smaller new target id.
+          db.execute('''
+INSERT INTO ps_kv(key, value) VALUES
+  ('last_seen_checkpoint_request_id', 6),
+  ('last_applied_checkpoint_request_id', 5);
+''');
+
           db.execute(
               'INSERT INTO powersync_crud (op, id, type, data) VALUES (?, ?, ?, ?)',
               [
@@ -266,7 +274,7 @@ void main() {
               isEmpty);
           expect(
               db.select(
-                  "SELECT key, value FROM ps_kv WHERE key = 'local_target_op'"),
+                  "SELECT key, value FROM ps_kv WHERE key LIKE '%checkpoint_request_id' OR key = 'local_target_op'"),
               [
                 {
                   'key': 'local_target_op',

dart/test/migration_test.dart

@@ -106,16 +106,15 @@ VALUES(1, '$local', 5, 6, 7, 0, 0, 1, 0, 0, 0);
 
       db.executeInTx('select powersync_init()');
 
-      expect(db.select('SELECT key, value FROM ps_kv ORDER BY key'), containsAll([
-        {'key': 'last_seen_checkpoint_request_id', 'value': 6},
-        {'key': 'last_requested_checkpoint_request_id', 'value': 7},
+      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},
-      ]));
+      ]);
+      expect(db.select(r"SELECT * FROM ps_buckets WHERE name = '$local'"),
+          isEmpty);
       expect(
-        db
-            .select("PRAGMA table_info('ps_buckets')")
-            .map((row) => row['name']),
+        db.select("PRAGMA table_info('ps_buckets')").map((row) => row['name']),
         isNot(contains('target_op')),
       );
       expect(
@@ -138,8 +137,8 @@ VALUES(1, '$local', 5, 6, 9223372036854775807, 0, 0, 1, 0, 0, 0);
       db.executeInTx('select powersync_init()');
 
       // last_applied_op becomes the applied checkpoint id, but it must not seed the requested
-      // checkpoint counter. The sentinel target is preserved for blocking, but is not concrete
-      // enough to become last_requested_checkpoint_request_id.
+      // checkpoint counter. The sentinel target is preserved for blocking, but target ops no longer
+      // seed last_requested_checkpoint_request_id.
       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},
@@ -157,8 +156,8 @@ VALUES(1, '$local', 0, 0, 9223372036854775807, 0, 0, 1, 0, 0, 0);
 
       db.executeInTx('select powersync_init()');
 
-      // The max-op sentinel is valid local target state, but it is not a concrete checkpoint
-      // request id and must not seed last_requested_checkpoint_request_id.
+      // 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},
       ]);
@@ -190,6 +189,42 @@ INSERT INTO ps_kv(key, value) VALUES
       );
     });
 
+    test('re-upgrades after downgrade with checkpoint state', () async {
+      db.execute(fixtures.finalState);
+      db.execute(r'''
+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);
+''');
+
+      db.executeInTx('select powersync_test_migration(13)');
+
+      // Simulate an older SDK advancing the restored $local row while downgraded.
+      db.execute(r'''
+UPDATE ps_buckets
+   SET last_op = 8, last_applied_op = 8, target_op = 9
+ WHERE name = '$local'
+''');
+
+      db.executeInTx('select powersync_init()');
+
+      // The $local row is the source of truth on re-upgrade; the request counter is unrelated to
+      // $local and survives the downgrade unchanged.
+      expect(db.select('SELECT key, value FROM ps_kv ORDER BY key'), [
+        {'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},
+      ]);
+      expect(db.select(r"SELECT * FROM ps_buckets WHERE name = '$local'"),
+          isEmpty);
+
+      final schema = '${getSchema(db)}\n${getMigrations(db)}';
+      expect(schema, equals(fixtures.finalState.trim()));
+    });
+
     test('does not restore local bucket without local target on downgrade',
         () async {
       db.execute(fixtures.finalState);

dart/test/sync_test.dart

@@ -474,26 +474,52 @@ void _syncTests<T>({
   syncTest('probes and updates local target op without sync iteration', (_) {
     expect(probeLocalTargetOp(), isNull);
     expect(probeLocalTargetOp(1), isNull);
-    expect(lastRequestedCheckpointRequestId(), 1);
+    expect(lastRequestedCheckpointRequestId(), isNull);
     expect(probeLocalTargetOp(), 1);
 
     expect(probeLocalTargetOp(2), 1);
-    expect(lastRequestedCheckpointRequestId(), 2);
+    expect(lastRequestedCheckpointRequestId(), isNull);
     expect(probeLocalTargetOp(), 2);
   });
 
   syncTest('accepts text checkpoint request ids for local target op', (_) {
     expect(probeLocalTargetOp('1'), isNull);
-    expect(lastRequestedCheckpointRequestId(), 1);
+    expect(lastRequestedCheckpointRequestId(), isNull);
     expect(probeLocalTargetOp(), 1);
   });
 
-  syncTest('does not store non-request target ops as checkpoint request id',
-      (_) {
+  syncTest('rejects negative local target ops', (_) {
+    expect(
+      () => invokeControlRaw('local_target_op', -1),
+      throwsA(isSqliteException(
+        3091,
+        contains('local target op must be a non-negative integer'),
+      )),
+    );
+  });
+
+  syncTest('local target op does not update checkpoint request id', (_) {
+    invokeControlRaw('start', null);
+    invokeControlRaw('seed_checkpoint_request_id', 10);
+
+    expect(lastRequestedCheckpointRequestId(), 10);
+    expect(probeLocalTargetOp(7), isNull);
+    expect(probeLocalTargetOp(), 7);
+    expect(lastRequestedCheckpointRequestId(), 10);
+  });
+
+  syncTest('does not store target ops as checkpoint request id', (_) {
     expect(probeLocalTargetOp(0), isNull);
     expect(lastRequestedCheckpointRequestId(), isNull);
     expect(probeLocalTargetOp(), isNull);
 
+    expect(probeLocalTargetOp(1), isNull);
+    expect(lastRequestedCheckpointRequestId(), isNull);
+    expect(probeLocalTargetOp(), 1);
+
+    expect(probeLocalTargetOp(0), 1);
+    expect(probeLocalTargetOp(), isNull);
+
     expect(probeLocalTargetOp(9223372036854775807), isNull);
     expect(lastRequestedCheckpointRequestId(), isNull);
     expect(probeLocalTargetOp(), 9223372036854775807);
@@ -564,6 +590,27 @@ void _syncTests<T>({
         db.select(r"SELECT * FROM ps_buckets WHERE name = '$local'"), isEmpty);
   });
 
+  syncTest('local writes clear checkpoint request high-water marks', (_) {
+    invokeControl('start', null);
+
+    pushCheckpoint(buckets: priorityBuckets, writeCheckpoint: '5');
+    pushCheckpointComplete();
+    expect(lastAppliedCheckpointRequestId(), 5);
+
+    // A local write can only be acknowledged by a checkpoint request id observed after it. Stale
+    // seen/applied values (which may come from another id namespace, like migrated legacy write
+    // checkpoints) must not remain to open the apply gate for a smaller new target id.
+    db.execute("insert into items (id, col) values ('local', 'data');");
+
+    expect(lastAppliedCheckpointRequestId(), isNull);
+    expect(
+      db.select(
+          "SELECT 1 FROM ps_kv WHERE key = 'last_seen_checkpoint_request_id'"),
+      isEmpty,
+    );
+    expect(probeLocalTargetOp(), 9223372036854775807);
+  });
+
   test('clearing database clears sync status', () {
     invokeControl('start', null);
     pushCheckpoint(buckets: priorityBuckets);
@@ -1030,7 +1077,7 @@ void _syncTests<T>({
       db.execute('DELETE FROM ps_crud');
       probeLocalTargetOp(1);
       expect(invokeControl('completed_upload', null), isEmpty);
-      expect(lastRequestedCheckpointRequestId(), 1);
+      expect(lastRequestedCheckpointRequestId(), 0);
 
       // Sync afterwards containing data and write checkpoint.
       pushCheckpoint(buckets: priorityBuckets, writeCheckpoint: '1');

docs/schema.md

@@ -31,9 +31,10 @@ a checkpoint and that we have validated its checksum).
 8. `count_since_last`: The amount of operations downloaded since the last verified checkpoint.
 
 Schema version 14 removes the legacy `target_op` column after migrating `$local.target_op` to
-`ps_kv.local_target_op`. This makes older SDKs fail with a hard SQLite error if they try to keep
-using the migrated database without downgrading. The down migration restores `target_op` for older
-schema versions.
+`ps_kv.local_target_op`, and deletes the `$local` row so `ps_buckets` only contains real sync
+buckets. This makes older SDKs fail with a hard SQLite error if they try to keep using the migrated
+database without downgrading. The down migration restores `target_op` and recreates the `$local`
+row from `ps_kv` for older schema versions.
 
 ## `ps_crud`