Handle sync lines with escapes

Author: simolus3

crates/core/src/sync/line.rs

@@ -15,6 +15,12 @@ use crate::util::{deserialize_optional_string_to_i64, deserialize_string_to_i64}
 use super::bucket_priority::BucketPriority;
 use super::Checksum;
 
+/// 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 enum SyncLine<'a> {
@@ -53,7 +59,7 @@ pub struct CheckpointDiff<'a> {
     #[serde(borrow)]
     pub updated_buckets: Vec<BucketChecksum<'a>>,
     #[serde(borrow)]
-    pub removed_buckets: Vec<&'a str>,
+    pub removed_buckets: Vec<SyncLineStr<'a>>,
     #[serde(default)]
     #[serde(deserialize_with = "deserialize_optional_string_to_i64")]
     pub write_checkpoint: Option<i64>,
@@ -74,7 +80,8 @@ pub struct CheckpointPartiallyComplete {
 
 #[derive(Deserialize, Debug)]
 pub struct BucketChecksum<'a> {
-    pub bucket: &'a str,
+    #[serde(borrow)]
+    pub bucket: SyncLineStr<'a>,
     pub checksum: Checksum,
     #[serde(default)]
     pub priority: Option<BucketPriority>,
@@ -87,14 +94,15 @@ pub struct BucketChecksum<'a> {
 
 #[derive(Deserialize, Debug)]
 pub struct DataLine<'a> {
-    pub bucket: &'a str,
+    #[serde(borrow)]
+    pub bucket: SyncLineStr<'a>,
     pub data: Vec<OplogEntry<'a>>,
     //    #[serde(default)]
     //    pub has_more: bool,
     //    #[serde(default, borrow)]
-    //    pub after: Option<&'a str>,
+    //    pub after: Option<SyncLineStr<'a>>,
     //    #[serde(default, borrow)]
-    //    pub next_after: Option<&'a str>,
+    //    pub next_after: Option<SyncLineStr<'a>>,
 }
 
 #[derive(Deserialize, Debug)]
@@ -104,11 +112,11 @@ pub struct OplogEntry<'a> {
     pub op_id: i64,
     pub op: OpType,
     #[serde(default, borrow)]
-    pub object_id: Option<&'a str>,
+    pub object_id: Option<SyncLineStr<'a>>,
     #[serde(default, borrow)]
-    pub object_type: Option<&'a str>,
+    pub object_type: Option<SyncLineStr<'a>>,
     #[serde(default, borrow)]
-    pub subkey: Option<&'a str>,
+    pub subkey: Option<SyncLineStr<'a>>,
     #[serde(default, borrow)]
     pub data: Option<OplogData<'a>>,
 }
@@ -367,6 +375,17 @@ mod tests {
         assert_eq!(diff.removed_buckets.len(), 0);
     }
 
+    #[test]
+    fn parse_checkpoint_diff_escape() {
+        let SyncLine::CheckpointDiff(diff) = deserialize(
+            r#"{"checkpoint_diff": {"last_op_id": "10", "buckets": [], "updated_buckets": [], "removed_buckets": ["foo\""], "write_checkpoint": null}}"#,
+        ) else {
+            panic!("Expected checkpoint diff")
+        };
+
+        assert_eq!(diff.removed_buckets[0], "foo\"");
+    }
+
     #[test]
     fn parse_checkpoint_diff_no_write_checkpoint() {
         let SyncLine::CheckpointDiff(diff) = deserialize(

crates/core/src/sync/operations.rs

@@ -23,7 +23,7 @@ pub fn insert_bucket_operations(
     let BucketInfo {
         id: bucket_id,
         last_applied_op,
-    } = adapter.lookup_bucket(line.bucket)?;
+    } = adapter.lookup_bucket(&*line.bucket)?;
 
     // This is an optimization for initial sync - we can avoid persisting individual REMOVE
     // operations when last_applied_op = 0.
@@ -65,10 +65,10 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
         match data.op {
             OpType::PUT | OpType::REMOVE => {
                 let key: Cow<'static, str> = if let (Some(object_type), Some(object_id)) =
-                    (data.object_type, data.object_id)
+                    (&data.object_type, &data.object_id)
                 {
-                    let subkey = data.subkey.unwrap_or("null");
-                    Cow::Owned(format!("{}/{}/{}", &object_type, &object_id, subkey))
+                    let subkey = data.subkey.as_ref().map(|i| &**i).unwrap_or("null");
+                    Cow::Owned(format!("{}/{}/{}", &*object_type, &*object_id, subkey))
                 } else {
                     Cow::Borrowed("")
                 };
@@ -101,16 +101,16 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
 
                     if !should_skip_remove {
                         if let (Some(object_type), Some(object_id)) =
-                            (data.object_type, data.object_id)
+                            (&data.object_type, &data.object_id)
                         {
                             updated_row_statement.bind_text(
                                 1,
-                                object_type,
+                                &*object_type,
                                 sqlite::Destructor::STATIC,
                             )?;
                             updated_row_statement.bind_text(
                                 2,
-                                object_id,
+                                &*object_id,
                                 sqlite::Destructor::STATIC,
                             )?;
                             updated_row_statement.exec()?;
@@ -127,9 +127,9 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
                     insert_statement.bind_null(3)?;
                 }
 
-                if let (Some(object_type), Some(object_id)) = (data.object_type, data.object_id) {
-                    insert_statement.bind_text(4, object_type, sqlite::Destructor::STATIC)?;
-                    insert_statement.bind_text(5, object_id, sqlite::Destructor::STATIC)?;
+                if let (Some(object_type), Some(object_id)) = (&data.object_type, &data.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)?;

crates/core/src/sync/streaming_sync.rs

@@ -290,7 +290,7 @@ impl StreamingSyncIteration {
                     target.apply_diff(&diff);
                     validated_but_not_applied = None;
                     self.adapter
-                        .delete_buckets(diff.removed_buckets.iter().copied())?;
+                        .delete_buckets(diff.removed_buckets.iter().map(|i| &**i))?;
 
                     let progress = self.load_progress(target)?;
                     self.status.update(
@@ -493,7 +493,7 @@ impl SyncTarget {
         let mut buckets = BTreeMap::<String, OwnedBucketChecksum>::new();
         for bucket in &checkpoint.buckets {
             buckets.insert(bucket.bucket.to_string(), OwnedBucketChecksum::from(bucket));
-            to_delete.remove(bucket.bucket);
+            to_delete.remove(&*bucket.bucket);
         }
 
         *self = SyncTarget::Tracking(OwnedCheckpoint::from_checkpoint(checkpoint, buckets));
@@ -522,7 +522,7 @@ impl OwnedCheckpoint {
 
     fn apply_diff<'a>(&mut self, diff: &CheckpointDiff<'a>) {
         for removed in &diff.removed_buckets {
-            self.buckets.remove(*removed);
+            self.buckets.remove(&**removed);
         }
 
         for updated in &diff.updated_buckets {

crates/core/src/sync/sync_status.rs

@@ -206,7 +206,7 @@ impl SyncDownloadProgress {
     }
 
     pub fn increment_download_count(&mut self, line: &DataLine) {
-        if let Some(info) = self.buckets.get_mut(line.bucket) {
+        if let Some(info) = self.buckets.get_mut(&*line.bucket) {
             info.since_last += line.data.len() as i64
         }
     }