Merge pull request #82 from powersync-ja/serde-oplog-parsing

Use serde to parse data lines and checkpoints

Author: simolus3

crates/core/src/checkpoint.rs

@@ -5,17 +5,18 @@ use alloc::string::String;
 use alloc::vec::Vec;
 use core::ffi::c_int;
 
-use serde::{Deserialize, Serialize};
+use serde::Serialize;
 use serde_json as json;
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 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/operations.rs

@@ -1,8 +1,12 @@
 use alloc::format;
 use alloc::string::String;
+use alloc::vec::Vec;
 use num_traits::Zero;
+use serde::Deserialize;
 
 use crate::error::{PSResult, SQLiteError};
+use crate::sync::line::DataLine;
+use crate::sync::operations::insert_bucket_operations;
 use crate::sync::Checksum;
 use sqlite_nostd as sqlite;
 use sqlite_nostd::{Connection, ResultCode};
@@ -11,246 +15,15 @@ use crate::ext::SafeManagedStmt;
 
 // Run inside a transaction
 pub fn insert_operation(db: *mut sqlite::sqlite3, data: &str) -> Result<(), SQLiteError> {
-    // language=SQLite
-    let statement = db.prepare_v2(
-        "\
-SELECT
-    json_extract(e.value, '$.bucket') as bucket,
-    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
-FROM json_each(json_extract(?1, '$.buckets')) e",
-    )?;
-    statement.bind_text(1, data, sqlite::Destructor::STATIC)?;
-
-    while statement.step()? == ResultCode::ROW {
-        let bucket = statement.column_text(0)?;
-        let data = statement.column_text(1)?;
-        // let _has_more = statement.column_int(2)? != 0;
-        // let _after = statement.column_text(3)?;
-        // let _next_after = statement.column_text(4)?;
-
-        insert_bucket_operations(db, bucket, data)?;
-    }
-
-    Ok(())
-}
-
-pub fn insert_bucket_operations(
-    db: *mut sqlite::sqlite3,
-    bucket: &str,
-    data: &str,
-) -> Result<(), SQLiteError> {
-    // Statement to insert new operations (only for PUT and REMOVE).
-    // language=SQLite
-    let iterate_statement = db.prepare_v2(
-        "\
-SELECT
-    json_extract(e.value, '$.op_id') as op_id,
-    json_extract(e.value, '$.op') as op,
-    json_extract(e.value, '$.object_type') as object_type,
-    json_extract(e.value, '$.object_id') as object_id,
-    json_extract(e.value, '$.checksum') as checksum,
-    json_extract(e.value, '$.data') as data,
-    json_extract(e.value, '$.subkey') as subkey
-FROM json_each(?) e",
-    )?;
-    iterate_statement.bind_text(1, data, sqlite::Destructor::STATIC)?;
-
-    // We do an ON CONFLICT UPDATE simply so that the RETURNING bit works for existing rows.
-    // We can consider splitting this into separate SELECT and INSERT statements.
-    // language=SQLite
-    let bucket_statement = db.prepare_v2(
-        "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.step()?;
-
-    let bucket_id = bucket_statement.column_int64(0);
-
-    // This is an optimization for initial sync - we can avoid persisting individual REMOVE
-    // operations when last_applied_op = 0.
-    // We do still need to do the "supersede_statement" step for this case, since a REMOVE
-    // operation can supersede another PUT operation we're syncing at the same time.
-    let mut is_empty = bucket_statement.column_int64(1) == 0;
-
-    // Statement to supersede (replace) operations with the same key.
-    // language=SQLite
-    let supersede_statement = db.prepare_v2(
-        "\
-DELETE FROM ps_oplog
-    WHERE unlikely(ps_oplog.bucket = ?1)
-    AND ps_oplog.key = ?2
-RETURNING op_id, hash",
-    )?;
-    supersede_statement.bind_int64(1, bucket_id)?;
-
-    // language=SQLite
-    let insert_statement = db.prepare_v2("\
-INSERT INTO ps_oplog(bucket, op_id, key, row_type, row_id, data, hash) VALUES (?, ?, ?, ?, ?, ?, ?)")?;
-    insert_statement.bind_int64(1, bucket_id)?;
-
-    let updated_row_statement = db.prepare_v2(
-        "\
-INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
-    )?;
-
-    bucket_statement.reset()?;
-
-    let mut last_op: Option<i64> = None;
-    let mut add_checksum = Checksum::zero();
-    let mut op_checksum = Checksum::zero();
-    let mut added_ops: i32 = 0;
-
-    while iterate_statement.step()? == ResultCode::ROW {
-        let op_id = iterate_statement.column_int64(0);
-        let op = iterate_statement.column_text(1)?;
-        let object_type = iterate_statement.column_text(2);
-        let object_id = iterate_statement.column_text(3);
-        let checksum = Checksum::from_i32(iterate_statement.column_int(4));
-        let op_data = iterate_statement.column_text(5);
-
-        last_op = Some(op_id);
-        added_ops += 1;
-
-        if op == "PUT" || op == "REMOVE" {
-            let key: String;
-            if let (Ok(object_type), Ok(object_id)) = (object_type.as_ref(), object_id.as_ref()) {
-                let subkey = iterate_statement.column_text(6).unwrap_or("null");
-                key = format!("{}/{}/{}", &object_type, &object_id, subkey);
-            } else {
-                key = String::from("");
-            }
-
-            supersede_statement.bind_text(2, &key, sqlite::Destructor::STATIC)?;
-
-            let mut superseded = false;
-
-            while supersede_statement.step()? == ResultCode::ROW {
-                // Superseded (deleted) a previous operation, add the checksum
-                let supersede_checksum = Checksum::from_i32(supersede_statement.column_int(1));
-                add_checksum += supersede_checksum;
-                op_checksum -= supersede_checksum;
-
-                // Superseded an operation, only skip if the bucket was empty
-                // Previously this checked "superseded_op <= last_applied_op".
-                // However, that would not account for a case where a previous
-                // PUT operation superseded the original PUT operation in this
-                // same batch, in which case superseded_op is not accurate for this.
-                if !is_empty {
-                    superseded = true;
-                }
-            }
-            supersede_statement.reset()?;
-
-            if op == "REMOVE" {
-                let should_skip_remove = !superseded;
-
-                add_checksum += checksum;
-
-                if !should_skip_remove {
-                    if let (Ok(object_type), Ok(object_id)) = (object_type, object_id) {
-                        updated_row_statement.bind_text(
-                            1,
-                            object_type,
-                            sqlite::Destructor::STATIC,
-                        )?;
-                        updated_row_statement.bind_text(
-                            2,
-                            object_id,
-                            sqlite::Destructor::STATIC,
-                        )?;
-                        updated_row_statement.exec()?;
-                    }
-                }
-
-                continue;
-            }
-
-            insert_statement.bind_int64(2, op_id)?;
-            if key != "" {
-                insert_statement.bind_text(3, &key, sqlite::Destructor::STATIC)?;
-            } else {
-                insert_statement.bind_null(3)?;
-            }
-
-            if let (Ok(object_type), Ok(object_id)) = (object_type, object_id) {
-                insert_statement.bind_text(4, object_type, sqlite::Destructor::STATIC)?;
-                insert_statement.bind_text(5, object_id, sqlite::Destructor::STATIC)?;
-            } else {
-                insert_statement.bind_null(4)?;
-                insert_statement.bind_null(5)?;
-            }
-            if let Ok(data) = op_data {
-                insert_statement.bind_text(6, data, sqlite::Destructor::STATIC)?;
-            } else {
-                insert_statement.bind_null(6)?;
-            }
-
-            insert_statement.bind_int(7, checksum.bitcast_i32())?;
-            insert_statement.exec()?;
-
-            op_checksum += checksum;
-        } else if op == "MOVE" {
-            add_checksum += checksum;
-        } else if op == "CLEAR" {
-            // Any remaining PUT operations should get an implicit REMOVE
-            // language=SQLite
-            let clear_statement1 = db
-                .prepare_v2(
-                    "INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id)
-SELECT row_type, row_id
-FROM ps_oplog
-WHERE bucket = ?1",
-                )
-                .into_db_result(db)?;
-            clear_statement1.bind_int64(1, bucket_id)?;
-            clear_statement1.exec()?;
-
-            let clear_statement2 = db
-                .prepare_v2("DELETE FROM ps_oplog WHERE bucket = ?1")
-                .into_db_result(db)?;
-            clear_statement2.bind_int64(1, bucket_id)?;
-            clear_statement2.exec()?;
-
-            // And we need to re-apply all of those.
-            // We also replace the checksum with the checksum of the CLEAR op.
-            // language=SQLite
-            let clear_statement2 = db.prepare_v2(
-                "UPDATE ps_buckets SET last_applied_op = 0, add_checksum = ?1, op_checksum = 0 WHERE id = ?2",
-            )?;
-            clear_statement2.bind_int64(2, bucket_id)?;
-            clear_statement2.bind_int(1, checksum.bitcast_i32())?;
-            clear_statement2.exec()?;
-
-            add_checksum = Checksum::zero();
-            is_empty = true;
-            op_checksum = Checksum::zero();
-        }
+    #[derive(Deserialize)]
+    struct BucketBatch<'a> {
+        #[serde(borrow)]
+        buckets: Vec<DataLine<'a>>,
     }
 
-    if let Some(last_op) = &last_op {
-        // language=SQLite
-        let statement = db.prepare_v2(
-            "UPDATE ps_buckets
-                SET last_op = ?2,
-                    add_checksum = (add_checksum + ?3) & 0xffffffff,
-                    op_checksum = (op_checksum + ?4) & 0xffffffff,
-                    count_since_last = count_since_last + ?5
-            WHERE id = ?1",
-        )?;
-        statement.bind_int64(1, bucket_id)?;
-        statement.bind_int64(2, *last_op)?;
-        statement.bind_int(3, add_checksum.bitcast_i32())?;
-        statement.bind_int(4, op_checksum.bitcast_i32())?;
-        statement.bind_int(5, added_ops)?;
-
-        statement.exec()?;
+    let batch: BucketBatch = serde_json::from_str(data)?;
+    for line in &batch.buckets {
+        insert_bucket_operations(db, line)?;
     }
 
     Ok(())

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.