Don't store priorities

simolus3 · simolus3 · commit 7a3df3ee76b6 · 2025-02-12T09:34:34.000+01:00
diff --git a/Cargo.toml b/Cargo.toml
@@ -9,7 +9,6 @@ default-members = ["crates/shell", "crates/sqlite"]
 
 [profile.dev]
 panic = "abort"
-strip = true
 
 [profile.release]
 panic = "abort"
diff --git a/crates/core/src/bucket_priority.rs b/crates/core/src/bucket_priority.rs
@@ -1,5 +1,6 @@
 use core::ops::RangeInclusive;
 
+use serde::{de::Visitor, Deserialize};
 use sqlite_nostd::ResultCode;
 
 use crate::error::SQLiteError;
@@ -34,12 +35,6 @@ impl TryFrom<i32> for BucketPriority {
     }
 }
 
-impl Default for BucketPriority {
-    fn default() -> Self {
-        Self(1)
-    }
-}
-
 impl Into<i32> for BucketPriority {
     fn into(self) -> i32 {
         self.0
@@ -51,3 +46,44 @@ impl PartialOrd<BucketPriority> for BucketPriority {
         Some(self.0.partial_cmp(&other.0)?.reverse())
     }
 }
+
+impl<'de> Deserialize<'de> for BucketPriority {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct PriorityVisitor;
+        impl<'de> Visitor<'de> for PriorityVisitor {
+            type Value = BucketPriority;
+
+            fn expecting(&self, formatter: &mut core::fmt::Formatter) -> core::fmt::Result {
+                formatter.write_str("a priority as an integer between 0 and 3 (inclusive)")
+            }
+
+            fn visit_i32<E>(self, v: i32) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                BucketPriority::try_from(v).map_err(|e| E::custom(e.1.unwrap_or_default()))
+            }
+
+            fn visit_i64<E>(self, v: i64) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                let i: i32 = v.try_into().map_err(|_| E::custom("int too large"))?;
+                Self::visit_i32(self, i)
+            }
+
+            fn visit_u64<E>(self, v: u64) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                let i: i32 = v.try_into().map_err(|_| E::custom("int too large"))?;
+                Self::visit_i32(self, i)
+            }
+        }
+
+        deserializer.deserialize_i32(PriorityVisitor)
+    }
+}
diff --git a/crates/core/src/migrations.rs b/crates/core/src/migrations.rs
@@ -310,21 +310,5 @@ json_array(
             .into_db_result(local_db)?;
     }
 
-    if current_version < 7 && target_version >= 7 {
-        local_db
-            .exec_safe(
-                "\
-ALTER TABLE ps_buckets ADD COLUMN priority INTEGER NOT NULL DEFAULT 1;
-INSERT INTO ps_migration(id, down_migrations)
-VALUES(7,
-json_array(
-  json_object('sql', 'ALTER TABLE ps_buckets DROP COLUMN priority'),
-  json_object('sql', 'DELETE FROM ps_migration WHERE id >= 7')
-));
-",
-            )
-            .into_db_result(local_db)?;
-    }
-
     Ok(())
 }
diff --git a/crates/core/src/operations.rs b/crates/core/src/operations.rs
@@ -1,7 +1,6 @@
 use alloc::format;
 use alloc::string::String;
 
-use crate::bucket_priority::BucketPriority;
 use crate::error::{PSResult, SQLiteError};
 use sqlite_nostd as sqlite;
 use sqlite_nostd::{Connection, ResultCode};
@@ -18,8 +17,7 @@ SELECT
     json_extract(e.value, '$.data') as data,
     json_extract(e.value, '$.has_more') as has_more,
     json_extract(e.value, '$.after') as after,
-    json_extract(e.value, '$.next_after') as next_after,
-    json_extract(d.value, '$.priority') as priority
+    json_extract(e.value, '$.next_after') as next_after
 FROM json_each(json_extract(?1, '$.buckets')) e
     LEFT OUTER JOIN json_each(json_extract(?1, '$.descriptions')) d
         ON json_extract(e.value, '$.bucket') == d.key",
@@ -32,15 +30,8 @@ FROM json_each(json_extract(?1, '$.buckets')) e
         // let _has_more = statement.column_int(2)? != 0;
         // let _after = statement.column_text(3)?;
         // let _next_after = statement.column_text(4)?;
-        let priority = match statement.column_type(5)? {
-            sqlite_nostd::ColumnType::Integer => {
-                BucketPriority::try_from(statement.column_int(5)?).ok()
-            }
-            _ => None,
-        }
-        .unwrap_or_default();
 
-        insert_bucket_operations(db, bucket, data, priority)?;
+        insert_bucket_operations(db, bucket, data)?;
     }
 
     Ok(())
@@ -50,7 +41,6 @@ pub fn insert_bucket_operations(
     db: *mut sqlite::sqlite3,
     bucket: &str,
     data: &str,
-    priority: BucketPriority,
 ) -> Result<(), SQLiteError> {
     // Statement to insert new operations (only for PUT and REMOVE).
     // language=SQLite
@@ -72,14 +62,13 @@ FROM json_each(?) e",
     // We can consider splitting this into separate SELECT and INSERT statements.
     // language=SQLite
     let bucket_statement = db.prepare_v2(
-        "INSERT INTO ps_buckets(name, priority)
-                            VALUES(?, ?)
+        "INSERT INTO ps_buckets(name)
+                            VALUES(?)
                         ON CONFLICT DO UPDATE
                             SET last_applied_op = last_applied_op
                         RETURNING id, last_applied_op",
     )?;
     bucket_statement.bind_text(1, bucket, sqlite::Destructor::STATIC)?;
-    bucket_statement.bind_int(2, priority.into())?;
     bucket_statement.step()?;
 
     let bucket_id = bucket_statement.column_int64(0)?;
diff --git a/crates/core/src/operations_vtab.rs b/crates/core/src/operations_vtab.rs
@@ -84,7 +84,7 @@ extern "C" fn update(
             let result = insert_operation(db, args[3].text());
             vtab_result(vtab, result)
         } else if op == "sync_local" {
-            let result = sync_local(db, &args[3]);
+            let result = sync_local(db, args[3]);
             if let Ok(result_row) = result {
                 unsafe {
                     *p_row_id = result_row;
diff --git a/crates/core/src/sync_local.rs b/crates/core/src/sync_local.rs
@@ -1,6 +1,8 @@
 use alloc::collections::BTreeSet;
 use alloc::format;
 use alloc::string::String;
+use alloc::vec::Vec;
+use serde::Deserialize;
 
 use crate::bucket_priority::BucketPriority;
 use crate::error::{PSResult, SQLiteError};
@@ -14,30 +16,25 @@ fn can_apply_sync_changes(
     db: *mut sqlite::sqlite3,
     priority: BucketPriority,
 ) -> Result<bool, SQLiteError> {
-    // We can only make sync changes visible if data is consistent, meaning that we've seen the
-    // target operation sent in the original checkpoint message. We allow weakening consistency when
-    // buckets from different priorities are involved (buckets with higher priorities or a lower
-    // priority number can be published before we've reached the checkpoint for other buckets).
-    // language=SQLite
-    let statement = db.prepare_v2(
-        "\
+    // Don't publish downloaded data until the upload queue is empty (except for downloaded data in
+    // priority 0, which is published earlier).
+    if !priority.may_publish_with_outstanding_uploads() {
+        // language=SQLite
+        let statement = db.prepare_v2(
+            "\
 SELECT group_concat(name)
 FROM ps_buckets
-WHERE (target_op > last_op) AND (priority <= ?)",
-    )?;
-    statement.bind_int(1, priority.into())?;
+WHERE target_op > last_op AND name = '$local'",
+        )?;
 
-    if statement.step()? != ResultCode::ROW {
-        return Err(SQLiteError::from(ResultCode::ABORT));
-    }
+        if statement.step()? != ResultCode::ROW {
+            return Err(SQLiteError::from(ResultCode::ABORT));
+        }
 
-    if statement.column_type(0)? == ColumnType::Text {
-        return Ok(false);
-    }
+        if statement.column_type(0)? == ColumnType::Text {
+            return Ok(false);
+        }
 
-    // Don't publish downloaded data until the upload queue is empty (except for downloaded data in
-    // priority 0, which is published earlier).
-    if !priority.may_publish_with_outstanding_uploads() {
         let statement = db.prepare_v2("SELECT 1 FROM ps_crud LIMIT 1")?;
         if statement.step()? != ResultCode::DONE {
             return Ok(false);
@@ -47,13 +44,27 @@ WHERE (target_op > last_op) AND (priority <= ?)",
     Ok(true)
 }
 
-pub fn sync_local<V: Value>(db: *mut sqlite::sqlite3, data: &V) -> Result<i64, SQLiteError> {
-    let priority = match data.value_type() {
-        ColumnType::Integer => BucketPriority::try_from(data.int()),
-        ColumnType::Float => BucketPriority::try_from(data.double() as i32),
-        // Older clients without bucket priority support typically send an empty string here.
-        _ => Ok(BucketPriority::LOWEST),
-    }?;
+pub fn sync_local(db: *mut sqlite::sqlite3, data: *mut sqlite::value) -> Result<i64, SQLiteError> {
+    #[derive(Deserialize)]
+    struct SyncLocalArguments {
+        #[serde(rename = "buckets")]
+        _buckets: Vec<String>,
+        priority: Option<BucketPriority>,
+    }
+
+    const FALLBACK_PRIORITY: BucketPriority = BucketPriority::LOWEST;
+    let (has_args, priority) = match data.value_type() {
+        ColumnType::Text => {
+            let text = data.text();
+            if text.len() > 0 {
+                let args: SyncLocalArguments = serde_json::from_str(text)?;
+                (true, args.priority.unwrap_or(FALLBACK_PRIORITY))
+            } else {
+                (false, FALLBACK_PRIORITY)
+            }
+        }
+        _ => (false, FALLBACK_PRIORITY),
+    };
 
     if !can_apply_sync_changes(db, priority)? {
         return Ok(0);
@@ -78,12 +89,17 @@ pub fn sync_local<V: Value>(db: *mut sqlite::sqlite3, data: &V) -> Result<i64, S
             "\
 -- 1. Filter oplog by the ops added but not applied yet (oplog b).
 --    SELECT DISTINCT / UNION is important for cases with many duplicate ids.
-WITH updated_rows AS (
-  SELECT DISTINCT FALSE as local, b.row_type, b.row_id FROM ps_buckets AS buckets
-    CROSS JOIN ps_oplog AS b ON b.bucket = buckets.id AND (b.op_id > buckets.last_applied_op)
-    WHERE buckets.priority <= ?1
-  UNION SELECT TRUE, row_type, row_id FROM ps_updated_rows
-)
+WITH 
+  involved_buckets (id) AS (
+    SELECT id FROM ps_buckets WHERE ?1 IS NULL
+      OR name IN (SELECT value FROM json_each(json_extract(?1, '$.buckets')))
+  ),
+  updated_rows AS (
+    SELECT DISTINCT FALSE as local, b.row_type, b.row_id FROM ps_buckets AS buckets
+      CROSS JOIN ps_oplog AS b ON b.bucket = buckets.id AND (b.op_id > buckets.last_applied_op)
+      WHERE buckets.id IN (SELECT id FROM involved_buckets)
+    UNION SELECT TRUE, row_type, row_id FROM ps_updated_rows
+  )
 
 -- 3. Group the objects from different buckets together into a single one (ops).
 SELECT b.row_type as type,
@@ -98,15 +114,19 @@ FROM updated_rows b
     LEFT OUTER JOIN ps_oplog AS r
                ON r.row_type = b.row_type
                  AND r.row_id = b.row_id
-                 AND (SELECT priority FROM ps_buckets WHERE id = r.bucket) <= ?1
+                 AND r.bucket IN (SELECT id FROM involved_buckets)
 -- Group for (3)
 GROUP BY b.row_type, b.row_id",
         )
         .into_db_result(db)?;
 
-    // TODO: cache statements
+    if has_args {
+        statement.bind_value(1, data)?;
+    } else {
+        statement.bind_null(1)?;
+    }
 
-    statement.bind_int(1, priority.into())?;
+    // TODO: cache statements
     while statement.step().into_db_result(db)? == ResultCode::ROW {
         let type_name = statement.column_text(0)?;
         let id = statement.column_text(1)?;
@@ -170,10 +190,15 @@ GROUP BY b.row_type, b.row_id",
         .prepare_v2(
             "UPDATE ps_buckets
                  SET last_applied_op = last_op
-                 WHERE last_applied_op != last_op AND priority <= ?",
+                 WHERE last_applied_op != last_op AND
+                    (?1 IS NULL OR name IN (SELECT value FROM json_each(json_extract(?1, '$.buckets'))))",
         )
         .into_db_result(db)?;
-    updated.bind_int(1, priority.into())?;
+    if has_args {
+        updated.bind_value(1, data)?;
+    } else {
+        updated.bind_null(1)?;
+    }
     updated.exec()?;
 
     if priority == BucketPriority::LOWEST {
diff --git a/dart/test/sync_test.dart b/dart/test/sync_test.dart
@@ -75,8 +75,16 @@ void main() {
         ],
       );
 
-      db.execute('INSERT INTO powersync_operations(op, data) VALUES (?, ?)',
-          ['sync_local', priority]);
+      db.execute('INSERT INTO powersync_operations(op, data) VALUES (?, ?)', [
+        'sync_local',
+        jsonEncode({
+          'priority': priority,
+          'buckets': [
+            for (final cs in checksums.cast<Map<String, dynamic>>())
+              if (priority == null || cs['priority'] <= priority) cs['bucket']
+          ],
+        })
+      ]);
       return db.lastInsertRowId == 1;
     }
 
diff --git a/dart/test/utils/fix_035_fixtures.dart b/dart/test/utils/fix_035_fixtures.dart
@@ -18,9 +18,9 @@ const dataBroken = '''
 
 /// Data after applying the migration fix, but before sync_local
 const dataMigrated = '''
-;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete, priority) VALUES
-  (1, 'b1', 0, 0, 0, 0, 120, 0, 1),
-  (2, 'b2', 0, 0, 0, 0, 3, 0, 1)
+;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete) VALUES
+  (1, 'b1', 0, 0, 0, 0, 120, 0),
+  (2, 'b2', 0, 0, 0, 0, 3, 0)
 ;INSERT INTO ps_oplog(bucket, op_id, row_type, row_id, key, data, hash) VALUES
   (1, 1, 'todos', 't1', '', '{}', 100),
   (1, 2, 'todos', 't2', '', '{}', 20),
@@ -39,9 +39,9 @@ const dataMigrated = '''
 
 /// Data after applying the migration fix and sync_local
 const dataFixed = '''
-;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete, priority) VALUES
-  (1, 'b1', 0, 0, 0, 0, 120, 0, 1),
-  (2, 'b2', 0, 0, 0, 0, 3, 0, 1)
+;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete) VALUES
+  (1, 'b1', 0, 0, 0, 0, 120, 0),
+  (2, 'b2', 0, 0, 0, 0, 3, 0)
 ;INSERT INTO ps_oplog(bucket, op_id, row_type, row_id, key, data, hash) VALUES
   (1, 1, 'todos', 't1', '', '{}', 100),
   (1, 2, 'todos', 't2', '', '{}', 20),
diff --git a/dart/test/utils/migration_fixtures.dart b/dart/test/utils/migration_fixtures.dart
