Add priority field to buckets

Author: simolus3

crates/core/src/bucket_priority.rs

@@ -0,0 +1,46 @@
+use sqlite_nostd::ResultCode;
+
+use crate::error::SQLiteError;
+
+#[repr(transparent)]
+#[derive(Clone, Copy, PartialEq, Eq)]
+pub struct BucketPriority(i32);
+
+impl BucketPriority {
+    pub fn may_publish_with_outstanding_uploads(self) -> bool {
+        self.0 == 0
+    }
+
+    pub const HIGHEST: BucketPriority = BucketPriority(0);
+    pub const LOWEST: BucketPriority = BucketPriority(3);
+}
+
+impl TryFrom<i32> for BucketPriority {
+    type Error = SQLiteError;
+
+    fn try_from(value: i32) -> Result<Self, Self::Error> {
+        if value < BucketPriority::LOWEST.0 || value > BucketPriority::HIGHEST.0 {
+            return Err(SQLiteError::from(ResultCode::MISUSE));
+        }
+
+        return Ok(BucketPriority(value));
+    }
+}
+
+impl Default for BucketPriority {
+    fn default() -> Self {
+        Self(1)
+    }
+}
+
+impl Into<i32> for BucketPriority {
+    fn into(self) -> i32 {
+        self.0
+    }
+}
+
+impl PartialOrd<BucketPriority> for BucketPriority {
+    fn partial_cmp(&self, other: &BucketPriority) -> Option<core::cmp::Ordering> {
+        Some(self.0.partial_cmp(&other.0)?.reverse())
+    }
+}

crates/core/src/lib.rs

@@ -12,6 +12,7 @@ use core::ffi::{c_char, c_int};
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 
+mod bucket_priority;
 mod checkpoint;
 mod crud_vtab;
 mod diff;

crates/core/src/migrations.rs

@@ -310,5 +310,21 @@ 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 NOT NULL DEFAULT 1;
+INSERT INTO ps_migration(id, down_migrations)
+VALUES(6,
+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(())
 }

crates/core/src/operations_vtab.rs

@@ -76,30 +76,29 @@ extern "C" fn update(
     } else if rowid.value_type() == sqlite::ColumnType::Null {
         // INSERT
         let op = args[2].text();
-        let data = args[3].text();
 
         let tab = unsafe { &mut *vtab.cast::<VirtualTable>() };
         let db = tab.db;
 
         if op == "save" {
-            let result = insert_operation(db, data);
+            let result = insert_operation(db, args[3].text());
             vtab_result(vtab, result)
         } else if op == "sync_local" {
-            let result = sync_local(db, data);
+            let result = sync_local(db, &args[3]);
             if let Ok(result_row) = result {
                 unsafe {
                     *p_row_id = result_row;
                 }
             }
             vtab_result(vtab, result)
         } else if op == "clear_remove_ops" {
-            let result = clear_remove_ops(db, data);
+            let result = clear_remove_ops(db, args[3].text());
             vtab_result(vtab, result)
         } else if op == "delete_pending_buckets" {
-            let result = delete_pending_buckets(db, data);
+            let result = delete_pending_buckets(db, args[3].text());
             vtab_result(vtab, result)
         } else if op == "delete_bucket" {
-            let result = delete_bucket(db, data);
+            let result = delete_bucket(db, args[3].text());
             vtab_result(vtab, result)
         } else {
             ResultCode::MISUSE as c_int

crates/core/src/sync_local.rs

@@ -2,21 +2,30 @@ use alloc::collections::BTreeSet;
 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::{self as sqlite, Value};
 use sqlite_nostd::{ColumnType, Connection, ResultCode};
 
 use crate::ext::SafeManagedStmt;
 use crate::util::{internal_table_name, quote_internal_name};
 
-pub fn can_update_local(db: *mut sqlite::sqlite3) -> Result<bool, SQLiteError> {
+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(
         "\
 SELECT group_concat(name)
 FROM ps_buckets
-WHERE target_op > last_op",
+WHERE (target_op > last_op) AND (priority <= ?)",
     )?;
+    statement.bind_int(1, priority.into())?;
 
     if statement.step()? != ResultCode::ROW {
         return Err(SQLiteError::from(ResultCode::ABORT));
@@ -26,22 +35,34 @@ WHERE target_op > last_op",
         return Ok(false);
     }
 
-    // This is specifically relevant for when data is added to crud before another batch is completed.
-
-    // language=SQLite
-    let statement = db.prepare_v2("SELECT 1 FROM ps_crud LIMIT 1")?;
-    if statement.step()? != ResultCode::DONE {
-        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);
+        }
     }
 
     Ok(true)
 }
 
-pub fn sync_local(db: *mut sqlite::sqlite3, _data: &str) -> Result<i64, SQLiteError> {
-    if !can_update_local(db)? {
+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),
+    }?;
+
+    if !can_apply_sync_changes(db, priority)? {
         return Ok(0);
     }
 
+    if priority >= BucketPriority::LOWEST {
+        todo!("Only consider changes from certain bucket priorities")
+    }
+
     // language=SQLite
     let statement = db
         .prepare_v2("SELECT name FROM sqlite_master WHERE type='table' AND name GLOB 'ps_data_*'")

crates/core/src/view_admin.rs

@@ -120,7 +120,7 @@ fn powersync_init_impl(
 
     setup_internal_views(local_db)?;
 
-    powersync_migrate(ctx, 6)?;
+    powersync_migrate(ctx, 7)?;
 
     Ok(String::from(""))
 }