Represent complete sync with low priority number

Author: simolus3

Cargo.toml

@@ -9,7 +9,6 @@ default-members = ["crates/shell", "crates/sqlite"]
 
 [profile.dev]
 panic = "abort"
-strip = true
 
 [profile.release]
 panic = "abort"

crates/core/src/bucket_priority.rs

@@ -5,40 +5,45 @@ use crate::error::SQLiteError;
 
 #[repr(transparent)]
 #[derive(Clone, Copy, PartialEq, Eq)]
-pub struct BucketPriority(i32);
+pub struct BucketPriority {
+    pub number: i32,
+}
 
 impl BucketPriority {
     pub fn may_publish_with_outstanding_uploads(self) -> bool {
         self == BucketPriority::HIGHEST
     }
 
-    pub const HIGHEST: BucketPriority = BucketPriority(0);
+    pub const HIGHEST: BucketPriority = BucketPriority { number: 0 };
+
+    /// A low priority used to represent fully-completed sync operations across all priorities.
+    pub const SENTINEL: BucketPriority = BucketPriority { number: i32::MAX };
 }
 
 impl TryFrom<i32> for BucketPriority {
     type Error = SQLiteError;
 
     fn try_from(value: i32) -> Result<Self, Self::Error> {
-        if value < BucketPriority::HIGHEST.0 {
+        if value < BucketPriority::HIGHEST.number || value == Self::SENTINEL.number {
             return Err(SQLiteError(
                 ResultCode::MISUSE,
                 Some("Invalid bucket priority".into()),
             ));
         }
 
-        return Ok(BucketPriority(value));
+        return Ok(BucketPriority { number: value });
     }
 }
 
 impl Into<i32> for BucketPriority {
     fn into(self) -> i32 {
-        self.0
+        self.number
     }
 }
 
 impl PartialOrd<BucketPriority> for BucketPriority {
     fn partial_cmp(&self, other: &BucketPriority) -> Option<core::cmp::Ordering> {
-        Some(self.0.partial_cmp(&other.0)?.reverse())
+        Some(self.number.partial_cmp(&other.number)?.reverse())
     }
 }
 

crates/core/src/kv.rs

@@ -9,6 +9,7 @@ use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 use sqlite_nostd::{Connection, Context};
 
+use crate::bucket_priority::BucketPriority;
 use crate::create_sqlite_optional_text_fn;
 use crate::create_sqlite_text_fn;
 use crate::error::SQLiteError;
@@ -46,8 +47,8 @@ fn powersync_last_synced_at_impl(
     let db = ctx.db_handle();
 
     // language=SQLite
-    let statement =
-        db.prepare_v2("select last_synced_at from ps_sync_state where priority = -1")?;
+    let statement = db.prepare_v2("select last_synced_at from ps_sync_state where priority = ?")?;
+    statement.bind_int(1, BucketPriority::SENTINEL.into())?;
 
     if statement.step()? == ResultCode::ROW {
         let client_id = statement.column_text(0)?;

crates/core/src/migrations.rs

@@ -8,6 +8,7 @@ use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 use sqlite_nostd::{Connection, Context};
 
+use crate::bucket_priority::BucketPriority;
 use crate::error::{PSResult, SQLiteError};
 use crate::fix035::apply_v035_fix;
 
@@ -311,26 +312,25 @@ json_array(
     }
 
     if current_version < 7 && target_version >= 7 {
-        local_db
-            .exec_safe(
-                "\
+        const SENTINEL_PRIORITY: i32 = BucketPriority::SENTINEL.number;
+        let stmt = format!("\
 CREATE TABLE ps_sync_state (
   priority INTEGER NOT NULL,
   last_synced_at TEXT NOT NULL
 ) STRICT;
 INSERT OR IGNORE INTO ps_sync_state (priority, last_synced_at)
-  SELECT -1, value from ps_kv where key = 'last_synced_at';
+  SELECT {}, value from ps_kv where key = 'last_synced_at';
 
 INSERT INTO ps_migration(id, down_migrations)
 VALUES(7,
 json_array(
-json_object('sql', 'INSERT OR REPLACE INTO ps_kv(key, value) SELECT ''last_synced_at'', last_synced_at FROM ps_sync_state WHERE priority = -1'),
+json_object('sql', 'INSERT OR REPLACE INTO ps_kv(key, value) SELECT ''last_synced_at'', last_synced_at FROM ps_sync_state WHERE priority = {}'),
 json_object('sql', 'DROP TABLE ps_sync_state'),
 json_object('sql', 'DELETE FROM ps_migration WHERE id >= 7')
 ));
-",
-            )
-            .into_db_result(local_db)?;
+", SENTINEL_PRIORITY, SENTINEL_PRIORITY);
+
+        local_db.exec_safe(&stmt).into_db_result(local_db)?;
     }
 
     Ok(())

crates/core/src/sync_local.rs

@@ -1,7 +1,4 @@
-use core::ffi::c_void;
-
 use alloc::collections::BTreeSet;
-use alloc::fmt::format;
 use alloc::format;
 use alloc::string::String;
 use alloc::vec::Vec;
@@ -291,24 +288,25 @@ GROUP BY b.row_type, b.row_id",
     }
 
     fn mark_completed(&self) -> Result<(), SQLiteError> {
-        let priority_code = match &self.partial {
+        let priority_code: i32 = match &self.partial {
             None => {
                 // language=SQLite
                 self.db
                     .exec_safe("DELETE FROM ps_updated_rows")
                     .into_db_result(self.db)?;
-                -1
+                BucketPriority::SENTINEL
             }
-            Some(partial) => partial.priority.into(),
-        };
+            Some(partial) => partial.priority,
+        }
+        .into();
 
         // Higher-priority buckets are always part of lower-priority sync operations too, so we can
         // delete information about higher-priority syncs (represented as lower priority numbers).
-        // A complete sync is represented as -1.
+        // A complete sync is represented by a number higher than the lowest priority we allow.
         // language=SQLite
         let stmt = self
             .db
-            .prepare_v2("DELETE FROM ps_sync_state WHERE (priority < ?1) OR (?1 = -1);")
+            .prepare_v2("DELETE FROM ps_sync_state WHERE priority < ?1;")
             .into_db_result(self.db)?;
         stmt.bind_int(1, priority_code)?;
         stmt.exec()?;

dart/test/utils/migration_fixtures.dart

@@ -215,7 +215,7 @@ const expectedState = <int, String>{
 ;INSERT INTO ps_migration(id, down_migrations) VALUES(4, '[{"sql":"DELETE FROM ps_migration WHERE id >= 4"},{"sql":"ALTER TABLE ps_buckets DROP COLUMN op_checksum"},{"sql":"ALTER TABLE ps_buckets DROP COLUMN remove_operations"}]')
 ;INSERT INTO ps_migration(id, down_migrations) VALUES(5, '[{"sql":"SELECT powersync_drop_view(view.name)\n  FROM sqlite_master view\n  WHERE view.type = ''view''\n    AND view.sql GLOB  ''*-- powersync-auto-generated''"},{"sql":"ALTER TABLE ps_buckets RENAME TO ps_buckets_5"},{"sql":"ALTER TABLE ps_oplog RENAME TO ps_oplog_5"},{"sql":"CREATE TABLE ps_buckets(\n  name TEXT PRIMARY KEY,\n  last_applied_op INTEGER NOT NULL DEFAULT 0,\n  last_op INTEGER NOT NULL DEFAULT 0,\n  target_op INTEGER NOT NULL DEFAULT 0,\n  add_checksum INTEGER NOT NULL DEFAULT 0,\n  pending_delete INTEGER NOT NULL DEFAULT 0\n, op_checksum INTEGER NOT NULL DEFAULT 0, remove_operations INTEGER NOT NULL DEFAULT 0)"},{"sql":"INSERT INTO ps_buckets(name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete)\n    SELECT name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete FROM ps_buckets_5"},{"sql":"CREATE TABLE ps_oplog(\n  bucket TEXT NOT NULL,\n  op_id INTEGER NOT NULL,\n  op INTEGER NOT NULL,\n  row_type TEXT,\n  row_id TEXT,\n  key TEXT,\n  data TEXT,\n  hash INTEGER NOT NULL,\n  superseded INTEGER NOT NULL)"},{"sql":"CREATE INDEX ps_oplog_by_row ON ps_oplog (row_type, row_id) WHERE superseded = 0"},{"sql":"CREATE INDEX ps_oplog_by_opid ON ps_oplog (bucket, op_id)"},{"sql":"CREATE INDEX ps_oplog_by_key ON ps_oplog (bucket, key) WHERE superseded = 0"},{"sql":"INSERT INTO ps_oplog(bucket, op_id, op, row_type, row_id, key, data, hash, superseded)\n    SELECT ps_buckets_5.name, oplog.op_id, 3, oplog.row_type, oplog.row_id, oplog.key, oplog.data, oplog.hash, 0\n    FROM ps_oplog_5 oplog\n    JOIN ps_buckets_5\n        ON ps_buckets_5.id = oplog.bucket"},{"sql":"DROP TABLE ps_oplog_5"},{"sql":"DROP TABLE ps_buckets_5"},{"sql":"INSERT INTO ps_oplog(bucket, op_id, op, row_type, row_id, hash, superseded)\n    SELECT ''$local'', 1, 4, r.row_type, r.row_id, 0, 0\n    FROM ps_updated_rows r"},{"sql":"INSERT OR REPLACE INTO ps_buckets(name, pending_delete, last_op, target_op) VALUES(''$local'', 1, 0, 9223372036854775807)"},{"sql":"DROP TABLE ps_updated_rows"},{"sql":"DELETE FROM ps_migration WHERE id >= 5"}]')
 ;INSERT INTO ps_migration(id, down_migrations) VALUES(6, '[{"sql":"DELETE FROM ps_migration WHERE id >= 6"}]')
-;INSERT INTO ps_migration(id, down_migrations) VALUES(7, '[{"sql":"INSERT OR REPLACE INTO ps_kv(key, value) SELECT ''last_synced_at'', last_synced_at FROM ps_sync_state WHERE priority = -1"},{"sql":"DROP TABLE ps_sync_state"},{"sql":"DELETE FROM ps_migration WHERE id >= 7"}]')
+;INSERT INTO ps_migration(id, down_migrations) VALUES(7, '[{"sql":"INSERT OR REPLACE INTO ps_kv(key, value) SELECT ''last_synced_at'', last_synced_at FROM ps_sync_state WHERE priority = 2147483647"},{"sql":"DROP TABLE ps_sync_state"},{"sql":"DELETE FROM ps_migration WHERE id >= 7"}]')
 ''',
 };