Move checksum validation to new types

Author: simolus3

crates/core/src/checkpoint.rs

@@ -13,9 +13,10 @@ use sqlite_nostd::{Connection, Context, Value};
 
 use crate::create_sqlite_text_fn;
 use crate::error::SQLiteError;
-use crate::sync_types::Checkpoint;
+use crate::sync::checkpoint::{validate_checkpoint, OwnedBucketChecksum};
+use crate::sync::line::Checkpoint;
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize)]
 struct CheckpointResult {
     valid: bool,
     failed_buckets: Vec<String>,
@@ -26,53 +27,23 @@ fn powersync_validate_checkpoint_impl(
     args: &[*mut sqlite::value],
 ) -> Result<String, SQLiteError> {
     let data = args[0].text();
-
-    let _checkpoint: Checkpoint = serde_json::from_str(data)?;
-
+    let checkpoint: Checkpoint = serde_json::from_str(data)?;
     let db = ctx.db_handle();
-
-    // language=SQLite
-    let statement = db.prepare_v2(
-        "WITH
-bucket_list(bucket, checksum) AS (
-  SELECT
-        json_extract(json_each.value, '$.bucket') as bucket,
-        json_extract(json_each.value, '$.checksum') as checksum
-  FROM json_each(json_extract(?1, '$.buckets'))
-)
-SELECT
-  bucket_list.bucket as bucket,
-  IFNULL(buckets.add_checksum, 0) as add_checksum,
-  IFNULL(buckets.op_checksum, 0) as oplog_checksum,
-  bucket_list.checksum as expected_checksum
-FROM bucket_list
-  LEFT OUTER JOIN ps_buckets AS buckets ON
-     buckets.name = bucket_list.bucket
-GROUP BY bucket_list.bucket",
-    )?;
-
-    statement.bind_text(1, data, sqlite::Destructor::STATIC)?;
-
-    let mut failures: Vec<String> = alloc::vec![];
-
-    while statement.step()? == ResultCode::ROW {
-        let name = statement.column_text(0)?;
-        // checksums with column_int are wrapped to i32 by SQLite
-        let add_checksum = statement.column_int(1);
-        let oplog_checksum = statement.column_int(2);
-        let expected_checksum = statement.column_int(3);
-
-        // wrapping add is like +, but safely overflows
-        let checksum = oplog_checksum.wrapping_add(add_checksum);
-
-        if checksum != expected_checksum {
-            failures.push(String::from(name));
-        }
+    let buckets: Vec<OwnedBucketChecksum> = checkpoint
+        .buckets
+        .iter()
+        .map(OwnedBucketChecksum::from)
+        .collect();
+
+    let failures = validate_checkpoint(buckets.iter(), None, db)?;
+    let mut failed_buckets = Vec::<String>::with_capacity(failures.len());
+    for failure in failures {
+        failed_buckets.push(failure.bucket_name);
     }
 
     let result = CheckpointResult {
-        valid: failures.is_empty(),
-        failed_buckets: failures,
+        valid: failed_buckets.is_empty(),
+        failed_buckets: failed_buckets,
     };
 
     Ok(json::to_string(&result)?)

crates/core/src/lib.rs

@@ -27,7 +27,6 @@ mod operations_vtab;
 mod schema;
 mod sync;
 mod sync_local;
-mod sync_types;
 mod util;
 mod uuid;
 mod version;

crates/core/src/sync/bucket_priority.rs

@@ -4,7 +4,7 @@ use sqlite_nostd::ResultCode;
 use crate::error::SQLiteError;
 
 #[repr(transparent)]
-#[derive(Clone, Copy, PartialEq, Eq)]
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
 pub struct BucketPriority {
     pub number: i32,
 }
@@ -14,6 +14,8 @@ impl BucketPriority {
         self == BucketPriority::HIGHEST
     }
 
+    /// The priority to use when the sync service doesn't attach priorities in checkpoints.
+    pub const FALLBACK: BucketPriority = BucketPriority { number: 3 };
     pub const HIGHEST: BucketPriority = BucketPriority { number: 0 };
 
     /// A low priority used to represent fully-completed sync operations across all priorities.

crates/core/src/sync/checkpoint.rs

@@ -0,0 +1,93 @@
+use alloc::{string::String, vec::Vec};
+use num_traits::Zero;
+use serde::Deserialize;
+
+use crate::{
+    error::SQLiteError,
+    sync::{
+        line::{BucketChecksum, Checkpoint},
+        BucketPriority, Checksum,
+    },
+};
+use sqlite_nostd::{self as sqlite, Connection, ResultCode};
+
+#[derive(Debug, Clone)]
+pub struct OwnedBucketChecksum {
+    pub bucket: String,
+    pub checksum: Checksum,
+    pub priority: BucketPriority,
+    pub count: Option<i64>,
+}
+
+impl OwnedBucketChecksum {
+    pub fn is_in_priority(&self, prio: Option<BucketPriority>) -> bool {
+        match prio {
+            None => true,
+            Some(prio) => self.priority >= prio,
+        }
+    }
+}
+
+impl From<&'_ BucketChecksum<'_>> for OwnedBucketChecksum {
+    fn from(value: &'_ BucketChecksum<'_>) -> Self {
+        Self {
+            bucket: value.bucket.clone().into_owned(),
+            checksum: value.checksum,
+            priority: value.priority.unwrap_or(BucketPriority::FALLBACK),
+            count: value.count,
+        }
+    }
+}
+
+pub struct ChecksumMismatch {
+    pub bucket_name: String,
+    pub expected_checksum: Checksum,
+    pub actual_op_checksum: Checksum,
+    pub actual_add_checksum: Checksum,
+}
+
+pub fn validate_checkpoint<'a>(
+    buckets: impl Iterator<Item = &'a OwnedBucketChecksum>,
+    priority: Option<BucketPriority>,
+    db: *mut sqlite::sqlite3,
+) -> Result<Vec<ChecksumMismatch>, SQLiteError> {
+    // language=SQLite
+    let statement = db.prepare_v2(
+        "
+SELECT
+    ps_buckets.add_checksum as add_checksum,
+    ps_buckets.op_checksum as oplog_checksum
+FROM ps_buckets WHERE name = ?;",
+    )?;
+
+    let mut failures: Vec<ChecksumMismatch> = Vec::new();
+    for bucket in buckets {
+        if bucket.is_in_priority(priority) {
+            statement.bind_text(1, &bucket.bucket, sqlite_nostd::Destructor::STATIC)?;
+
+            let (add_checksum, oplog_checksum) = match statement.step()? {
+                ResultCode::ROW => {
+                    let add_checksum = Checksum::from_i32(statement.column_int(0));
+                    let oplog_checksum = Checksum::from_i32(statement.column_int(1));
+                    (add_checksum, oplog_checksum)
+                }
+                _ => (Checksum::zero(), Checksum::zero()),
+            };
+
+            let actual = add_checksum + oplog_checksum;
+
+            if actual != bucket.checksum {
+                failures.push(ChecksumMismatch {
+                    bucket_name: bucket.bucket.clone(),
+                    expected_checksum: bucket.checksum,
+                    actual_add_checksum: add_checksum,
+                    actual_op_checksum: oplog_checksum,
+                });
+            }
+
+            statement.reset()?;
+        }
+    }
+
+    Ok(failures)
+}

crates/core/src/sync/line.rs

@@ -0,0 +1,99 @@
+use alloc::borrow::Cow;
+use alloc::string::String;
+use alloc::string::ToString;
+use alloc::vec::Vec;
+use serde::Deserialize;
+
+use super::BucketPriority;
+use super::Checksum;
+
+use crate::util::{deserialize_optional_string_to_i64, deserialize_string_to_i64};
+
+/// While we would like to always borrow strings for efficiency, that's not consistently possible.
+/// With the JSON decoder, borrowing from input data is only possible when the string contains no
+/// escape sequences (otherwise, the string is not a direct view of input data and we need an
+/// internal copy).
+type SyncLineStr<'a> = Cow<'a, str>;
+
+#[derive(Deserialize, Debug)]
+pub struct Checkpoint<'a> {
+    #[serde(deserialize_with = "deserialize_string_to_i64")]
+    pub last_op_id: i64,
+    #[serde(default)]
+    #[serde(deserialize_with = "deserialize_optional_string_to_i64")]
+    pub write_checkpoint: Option<i64>,
+    #[serde(borrow)]
+    pub buckets: Vec<BucketChecksum<'a>>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct BucketChecksum<'a> {
+    #[serde(borrow)]
+    pub bucket: SyncLineStr<'a>,
+    pub checksum: Checksum,
+    #[serde(default)]
+    pub priority: Option<BucketPriority>,
+    #[serde(default)]
+    pub count: Option<i64>,
+    //    #[serde(default)]
+    //    #[serde(deserialize_with = "deserialize_optional_string_to_i64")]
+    //    pub last_op_id: Option<i64>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct DataLine<'a> {
+    #[serde(borrow)]
+    pub bucket: SyncLineStr<'a>,
+    pub data: Vec<OplogEntry<'a>>,
+    //    #[serde(default)]
+    //    pub has_more: bool,
+    //    #[serde(default, borrow)]
+    //    pub after: Option<SyncLineStr<'a>>,
+    //    #[serde(default, borrow)]
+    //    pub next_after: Option<SyncLineStr<'a>>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct OplogEntry<'a> {
+    pub checksum: Checksum,
+    #[serde(deserialize_with = "deserialize_string_to_i64")]
+    pub op_id: i64,
+    pub op: OpType,
+    #[serde(default, borrow)]
+    pub object_id: Option<SyncLineStr<'a>>,
+    #[serde(default, borrow)]
+    pub object_type: Option<SyncLineStr<'a>>,
+    #[serde(default, borrow)]
+    pub subkey: Option<SyncLineStr<'a>>,
+    #[serde(default, borrow)]
+    pub data: Option<OplogData<'a>>,
+}
+
+#[derive(Debug)]
+pub enum OplogData<'a> {
+    /// A string encoding a well-formed JSON object representing values of the row.
+    Json { data: Cow<'a, str> },
+    //    BsonDocument { data: Cow<'a, [u8]> },
+}
+
+#[derive(Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OpType {
+    CLEAR,
+    MOVE,
+    PUT,
+    REMOVE,
+}
+
+impl<'a, 'de: 'a> Deserialize<'de> for OplogData<'a> {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        // For now, we will always get oplog data as a string. In the future, there may be the
+        // option of the sync service sending BSON-encoded data lines too, but that's not relevant
+        // for now.
+        return Ok(OplogData::Json {
+            data: Deserialize::deserialize(deserializer)?,
+        });
+    }
+}

crates/core/src/sync/mod.rs

@@ -1,5 +1,7 @@
 mod bucket_priority;
+pub mod checkpoint;
 mod checksum;
+pub mod line;
 
 pub use bucket_priority::BucketPriority;
 pub use checksum::Checksum;