Fix deserializing checkpoints

simolus3 · simolus3 · commit eb2e8c667089 · 2025-05-06T12:37:56.000+02:00
diff --git a/crates/core/src/bson/de.rs b/crates/core/src/bson/de.rs
@@ -196,29 +196,8 @@ impl<'de, 'a> de::Deserializer<'de> for &'a mut Deserializer<'de> {
         }
     }
 
-    fn deserialize_i32<V>(self, visitor: V) -> Result<V::Value, Self::Error>
-    where
-        V: Visitor<'de>,
-    {
-        // Since the sync service is written in JavaScript, we'll get numbers as doubles...
-        let element_type = self.prepare_to_read_value()?;
-        match element_type {
-            ElementType::Int32 => visitor.visit_i32(self.parser.read_int32()?),
-            ElementType::Double => {
-                let value = self.parser.read_double()?;
-                let converted: i32 = num_traits::cast(value).ok_or_else(|| {
-                    self.parser
-                        .error(ErrorKind::IllegalFloatToIntConversion(value))
-                })?;
-
-                visitor.visit_i32(converted)
-            }
-            _ => self.deserialize_any(visitor),
-        }
-    }
-
     forward_to_deserialize_any! {
-        bool i8 i16 i64 i128 u8 u16 u32 u64 u128 f32 f64 char str string
+        bool i8 i16 i32 i64 i128 u8 u16 u32 u64 u128 f32 f64 char str string
         bytes byte_buf unit unit_struct newtype_struct seq tuple
         tuple_struct map struct ignored_any identifier
     }
diff --git a/crates/core/src/bson/error.rs b/crates/core/src/bson/error.rs
@@ -36,7 +36,6 @@ pub enum ErrorKind {
     InvalidStateExpectedValue,
     ExpectedEnum { actual: ElementType },
     ExpectedString,
-    IllegalFloatToIntConversion(f64),
     UnexpectedEndOfDocumentForEnumVariant,
 }
 
diff --git a/crates/core/src/bson/mod.rs b/crates/core/src/bson/mod.rs
@@ -15,12 +15,9 @@ pub fn from_bytes<'de, T: Deserialize<'de>>(bytes: &'de [u8]) -> Result<T, BsonE
 
 #[cfg(test)]
 mod test {
-    use core::assert_matches::assert_matches;
-
-    use crate::sync::line::{Checkpoint, SyncLine};
+    use crate::sync::line::SyncLine;
 
     use super::*;
-    use serde::de::DeserializeOwned;
 
     #[test]
     fn test_hello_world() {
diff --git a/crates/core/src/sync/checksum.rs b/crates/core/src/sync/checksum.rs
@@ -0,0 +1,169 @@
+use core::{
+    num::Wrapping,
+    ops::{Add, AddAssign},
+};
+
+use num_traits::float::FloatCore;
+use num_traits::Zero;
+use serde::{de::Visitor, Deserialize, Serialize};
+
+/// A checksum as received from the sync service.
+///
+/// Conceptually, we use unsigned 32 bit integers to represent checksums, and adding checksums
+/// should be a wrapping add.
+#[repr(transparent)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+pub struct Checksum(Wrapping<u32>);
+
+impl Checksum {
+    pub const fn value(self) -> u32 {
+        self.0 .0
+    }
+
+    pub const fn from_value(value: u32) -> Self {
+        Self(Wrapping(value))
+    }
+
+    pub const fn from_i32(value: i32) -> Self {
+        Self::from_value(value as u32)
+    }
+
+    pub const fn bitcast_i32(self) -> i32 {
+        self.value() as i32
+    }
+}
+
+impl Zero for Checksum {
+    fn zero() -> Self {
+        const { Self::from_value(0) }
+    }
+
+    fn is_zero(&self) -> bool {
+        self.value() == 0
+    }
+}
+
+impl Add for Checksum {
+    type Output = Self;
+
+    #[inline]
+    fn add(self, rhs: Self) -> Self::Output {
+        Self(self.0 + rhs.0)
+    }
+}
+
+impl AddAssign for Checksum {
+    #[inline]
+    fn add_assign(&mut self, rhs: Self) {
+        self.0 += rhs.0
+    }
+}
+
+impl From<u32> for Checksum {
+    fn from(value: u32) -> Self {
+        Self::from_value(value)
+    }
+}
+
+impl<'de> Deserialize<'de> for Checksum {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct MyVisitor;
+
+        impl<'de> Visitor<'de> for MyVisitor {
+            type Value = Checksum;
+
+            fn expecting(&self, formatter: &mut core::fmt::Formatter) -> core::fmt::Result {
+                write!(formatter, "a number to interpret as a checksum")
+            }
+
+            fn visit_u32<E>(self, v: u32) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                Ok(v.into())
+            }
+
+            fn visit_u64<E>(self, v: u64) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                let as_u32: u32 = v.try_into().map_err(|_| {
+                    E::invalid_value(serde::de::Unexpected::Unsigned(v), &"a 32-bit int")
+                })?;
+                Ok(as_u32.into())
+            }
+
+            fn visit_i32<E>(self, v: i32) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                Ok(Checksum::from_i32(v))
+            }
+
+            fn visit_i64<E>(self, v: i64) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                // This is supposed to be an u32, but it could also be a i32 that we need to
+                // normalize.
+                let min: i64 = u32::MIN.into();
+                let max: i64 = u32::MAX.into();
+
+                if v >= min && v <= max {
+                    return Ok(Checksum::from(v as u32));
+                }
+
+                let as_i32: i32 = v.try_into().map_err(|_| {
+                    E::invalid_value(serde::de::Unexpected::Signed(v), &"a 32-bit int")
+                })?;
+                Ok(Checksum::from_i32(as_i32))
+            }
+
+            fn visit_f64<E>(self, v: f64) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                if !v.is_finite() || v.trunc() != v {
+                    return Err(E::invalid_value(
+                        serde::de::Unexpected::Float(v),
+                        &"a whole number",
+                    ));
+                }
+
+                self.visit_i64(v as i64)
+            }
+        }
+
+        deserializer.deserialize_u32(MyVisitor)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::Checksum;
+
+    #[test]
+    pub fn test_binary_representation() {
+        assert_eq!(Checksum::from_i32(-1).value(), u32::MAX);
+        assert_eq!(Checksum::from(u32::MAX).value(), u32::MAX);
+        assert_eq!(Checksum::from(u32::MAX).bitcast_i32(), -1);
+    }
+
+    fn deserialize(from: &str) -> Checksum {
+        serde_json::from_str(from).expect("should deserialize")
+    }
+
+    #[test]
+    pub fn test_deserialize() {
+        assert_eq!(deserialize("0").value(), 0);
+        assert_eq!(deserialize("-1").value(), u32::MAX);
+        assert_eq!(deserialize("-1.0").value(), u32::MAX);
+
+        assert_eq!(deserialize("3573495687").value(), 3573495687);
+        assert_eq!(deserialize("3573495687.0").value(), 3573495687);
+        assert_eq!(deserialize("-721471609.0").value(), 3573495687);
+    }
+}
diff --git a/crates/core/src/sync/line.rs b/crates/core/src/sync/line.rs
@@ -13,6 +13,7 @@ use crate::json_writer::JsonWriter;
 use crate::util::{deserialize_optional_string_to_i64, deserialize_string_to_i64};
 
 use super::bucket_priority::BucketPriority;
+use super::Checksum;
 
 #[derive(Deserialize, Debug)]
 
@@ -73,7 +74,7 @@ pub struct CheckpointPartiallyComplete {
 #[derive(Deserialize, Debug)]
 pub struct BucketChecksum<'a> {
     pub bucket: &'a str,
-    pub checksum: i32,
+    pub checksum: Checksum,
     pub priority: Option<BucketPriority>,
     pub count: Option<i64>,
     #[serde(default)]
@@ -95,7 +96,7 @@ pub struct DataLine<'a> {
 
 #[derive(Deserialize, Debug)]
 pub struct OplogEntry<'a> {
-    pub checksum: i32,
+    pub checksum: Checksum,
     #[serde(deserialize_with = "deserialize_string_to_i64")]
     pub op_id: i64,
     pub op: OpType,
@@ -324,7 +325,7 @@ mod tests {
         assert_eq!(checkpoint.buckets.len(), 1);
         let bucket = &checkpoint.buckets[0];
         assert_eq!(bucket.bucket, "a");
-        assert_eq!(bucket.checksum, 10);
+        assert_eq!(bucket.checksum, 10u32.into());
         assert_eq!(bucket.priority, None);
 
         let SyncLine::Checkpoint(checkpoint) = deserialize(
@@ -336,7 +337,7 @@ mod tests {
         assert_eq!(checkpoint.buckets.len(), 1);
         let bucket = &checkpoint.buckets[0];
         assert_eq!(bucket.bucket, "a");
-        assert_eq!(bucket.checksum, 10);
+        assert_eq!(bucket.checksum, 10u32.into());
         assert_eq!(bucket.priority, Some(BucketPriority { number: 1 }));
 
         assert_matches!(
@@ -400,10 +401,12 @@ mod tests {
         assert_eq!(data.next_after, None);
 
         assert_eq!(data.data.len(), 1);
+        let entry = &data.data[0];
+        assert_eq!(entry.checksum, 10u32.into());
         assert_matches!(
             &data.data[0],
             OplogEntry {
-                checksum: 10,
+                checksum: _,
                 op_id: 1,
                 object_id: Some(_),
                 object_type: Some(_),
diff --git a/crates/core/src/sync/mod.rs b/crates/core/src/sync/mod.rs
@@ -1,13 +1,16 @@
 use sqlite_nostd::{self as sqlite, ResultCode};
 
 pub mod bucket_priority;
+mod checksum;
 mod interface;
 pub mod line;
 pub mod operations;
 pub mod storage_adapter;
 mod streaming_sync;
 mod sync_status;
 
+pub use checksum::Checksum;
+
 pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
     interface::register(db)
 }
diff --git a/crates/core/src/sync/operations.rs b/crates/core/src/sync/operations.rs
@@ -1,4 +1,5 @@
 use alloc::{borrow::Cow, format};
+use num_traits::Zero;
 use sqlite_nostd::Connection;
 use sqlite_nostd::{self as sqlite, ResultCode};
 
@@ -8,6 +9,7 @@ use crate::{
 };
 
 use super::line::OplogData;
+use super::Checksum;
 use super::{
     line::{DataLine, OpType},
     storage_adapter::{BucketInfo, StorageAdapter},
@@ -51,8 +53,8 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
     )?;
 
     let mut last_op: Option<i64> = None;
-    let mut add_checksum: i32 = 0;
-    let mut op_checksum: i32 = 0;
+    let mut add_checksum = Checksum::zero();
+    let mut op_checksum = Checksum::zero();
     let mut added_ops: i32 = 0;
 
     for data in &line.data {
@@ -77,9 +79,9 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
 
                 while supersede_statement.step()? == ResultCode::ROW {
                     // Superseded (deleted) a previous operation, add the checksum
-                    let supersede_checksum = supersede_statement.column_int(1);
-                    add_checksum = add_checksum.wrapping_add(supersede_checksum);
-                    op_checksum = op_checksum.wrapping_sub(supersede_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".
@@ -95,7 +97,7 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
                 if data.op == OpType::REMOVE {
                     let should_skip_remove = !superseded;
 
-                    add_checksum = add_checksum.wrapping_add(checksum);
+                    add_checksum += checksum;
 
                     if !should_skip_remove {
                         if let (Some(object_type), Some(object_id)) =
@@ -144,13 +146,13 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
                     None => insert_statement.bind_null(6)?,
                 };
 
-                insert_statement.bind_int(7, checksum)?;
+                insert_statement.bind_int(7, checksum.bitcast_i32())?;
                 insert_statement.exec()?;
 
-                op_checksum = op_checksum.wrapping_add(checksum);
+                op_checksum += checksum;
             }
             OpType::MOVE => {
-                add_checksum = add_checksum.wrapping_add(checksum);
+                add_checksum += checksum;
             }
             OpType::CLEAR => {
                 // Any remaining PUT operations should get an implicit REMOVE
@@ -179,12 +181,12 @@ WHERE bucket = ?1",
                     "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)?;
+                clear_statement2.bind_int(1, checksum.bitcast_i32())?;
                 clear_statement2.exec()?;
 
-                add_checksum = 0;
+                add_checksum = Checksum::zero();
                 is_empty = true;
-                op_checksum = 0;
+                op_checksum = Checksum::zero();
             }
         }
     }
@@ -201,8 +203,8 @@ WHERE bucket = ?1",
         )?;
         statement.bind_int64(1, bucket_id)?;
         statement.bind_int64(2, last_op)?;
-        statement.bind_int(3, add_checksum)?;
-        statement.bind_int(4, op_checksum)?;
+        statement.bind_int(3, add_checksum.bitcast_i32())?;
+        statement.bind_int(4, op_checksum.bitcast_i32())?;
         statement.bind_int(5, added_ops)?;
 
         statement.exec()?;
diff --git a/crates/core/src/sync/storage_adapter.rs b/crates/core/src/sync/storage_adapter.rs
diff --git a/crates/core/src/sync/streaming_sync.rs b/crates/core/src/sync/streaming_sync.rs
diff --git a/dart/test/sync_test.dart b/dart/test/sync_test.dart

Original file line number	Diff line number	Diff line change
`@@ -36,7 +36,6 @@ pub enum ErrorKind {`
`36`	`36`	`InvalidStateExpectedValue,`
`37`	`37`	`ExpectedEnum { actual: ElementType },`
`38`	`38`	`ExpectedString,`
`39`		`- IllegalFloatToIntConversion(f64),`
`40`	`39`	`UnexpectedEndOfDocumentForEnumVariant,`
`41`	`40`	`}`
`42`	`41`