feat: (re-)enable timestamp+offset based pagination optimization

#53 enabled
#559 disabled

And this patch enables it for MySQL and Postgres but not Spanner.

Author: chenba

syncserver/src/web/extractors/bso_query_params.rs

@@ -32,13 +32,6 @@ impl From<Offset> for params::Offset {
 impl FromStr for Offset {
     type Err = ParseIntError;
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        // issue559: Disable ':' support for now: simply parse as i64 as
-        // previously (it was u64 previously but i64's close enough)
-        let result = Offset {
-            timestamp: None,
-            offset: s.parse::<u64>()?,
-        };
-        /*
         let result = match s.chars().position(|c| c == ':') {
             None => Offset {
                 timestamp: None,
@@ -55,7 +48,6 @@ impl FromStr for Offset {
                 }
             }
         };
-        */
         Ok(result)
     }
 }
@@ -201,36 +193,34 @@ impl FromRequest for BsoQueryParams {
                     None,
                 )
             })?;
-            // issue559: Dead code (timestamp always None)
-            /*
-            if params.sort != Sorting::Index {
-                if let Some(timestamp) = params.offset.as_ref().and_then(|offset| offset.timestamp)
-                {
-                    let bound = timestamp.as_i64();
-                    if let Some(newer) = params.newer {
-                        if bound < newer.as_i64() {
-                            return Err(ValidationErrorKind::FromDetails(
-                                format!("Invalid Offset {} {}", bound, newer.as_i64()),
-                                RequestErrorLocation::QueryString,
-                                Some("newer".to_owned()),
-                                None,
-                            )
-                            .into());
-                        }
-                    } else if let Some(older) = params.older {
-                        if bound > older.as_i64() {
-                            return Err(ValidationErrorKind::FromDetails(
-                                "Invalid Offset".to_owned(),
-                                RequestErrorLocation::QueryString,
-                                Some("older".to_owned()),
-                                None,
-                            )
-                            .into());
-                        }
+
+            if params.sort != Sorting::Index
+                && let Some(timestamp) = params.offset.as_ref().and_then(|offset| offset.timestamp)
+            {
+                let bound = timestamp.as_i64();
+                if let Some(newer) = params.newer {
+                    if bound < newer.as_i64() {
+                        return Err(ValidationErrorKind::FromDetails(
+                            format!("Invalid Offset {} {}", bound, newer.as_i64()),
+                            RequestErrorLocation::QueryString,
+                            Some("newer".to_owned()),
+                            None,
+                        )
+                        .into());
                     }
+                } else if let Some(older) = params.older
+                    && bound > older.as_i64()
+                {
+                    return Err(ValidationErrorKind::FromDetails(
+                        "Invalid Offset".to_owned(),
+                        RequestErrorLocation::QueryString,
+                        Some("older".to_owned()),
+                        None,
+                    )
+                    .into());
                 }
             }
-            */
+
             Ok(params)
         })
     }
@@ -288,19 +278,60 @@ mod tests {
         assert!(result.full);
     }
 
+    #[test]
+    fn test_offset_bound_below_newer() {
+        let state = make_state();
+        let req = TestRequest::with_uri("/?sort=newest&newer=2.22&offset=1111:1")
+            .data(state)
+            .to_http_request();
+        let result = block_on(BsoQueryParams::extract(&req));
+        assert!(result.is_err());
+        let resp: HttpResponse = result.err().unwrap().into();
+        assert_eq!(resp.status(), 400);
+    }
+
+    #[test]
+    fn test_offset_bound_above_older() {
+        let state = make_state();
+        let req = TestRequest::with_uri("/?sort=newest&older=2.22&offset=5858:1")
+            .data(state)
+            .to_http_request();
+        let result = block_on(BsoQueryParams::extract(&req));
+        assert!(result.is_err());
+        let resp: HttpResponse = result.err().unwrap().into();
+        assert_eq!(resp.status(), 400);
+    }
+
+    #[test]
+    fn test_offset_bound_within_range() {
+        let state = make_state();
+        let req = TestRequest::with_uri("/?sort=newest&newer=1.23&older=5.43&offset=3838:1")
+            .data(state)
+            .to_http_request();
+        let result = block_on(BsoQueryParams::extract(&req));
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_bound_validation_skipped_for_index_sort() {
+        let state = make_state();
+        let req = TestRequest::with_uri("/?sort=index&newer=2.22&offset=1111:1")
+            .data(state)
+            .to_http_request();
+        let result = block_on(BsoQueryParams::extract(&req));
+        assert!(result.is_ok());
+    }
+
     #[actix_rt::test]
     async fn test_offset() {
         let sample_offset = params::Offset {
             timestamp: Some(SyncTimestamp::default()),
             offset: 1234,
         };
 
-        let test_offset = Offset {
-            timestamp: None,
-            offset: sample_offset.offset,
-        };
-
         let offset_str = sample_offset.to_string();
-        assert!(test_offset == Offset::from_str(&offset_str).unwrap())
+        let parsed = Offset::from_str(&offset_str).unwrap();
+        assert_eq!(parsed.offset, sample_offset.offset);
+        assert_eq!(parsed.timestamp, sample_offset.timestamp,);
     }
 }

syncstorage-db-common/src/params.rs

@@ -68,27 +68,16 @@ pub struct Offset {
 
 impl Display for Offset {
     fn fmt(&self, fmt: &mut Formatter) -> Result<(), fmt::Error> {
-        // issue559: Disable ':' support for now.
-        write!(fmt, "{}", self.offset)
-        /*
         match self.timestamp {
-            None => self.offset.to_string(),
-            Some(ts) => format!("{}:{}", ts.as_i64(), self.offset),
+            None => write!(fmt, "{}", self.offset),
+            Some(ts) => write!(fmt, "{}:{}", ts.as_i64(), self.offset),
         }
-        */
     }
 }
 
 impl FromStr for Offset {
     type Err = ParseIntError;
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        // issue559: Disable ':' support for now: simply parse as i64 as
-        // previously (it was u64 previously but i64's close enough)
-        let result = Offset {
-            timestamp: None,
-            offset: s.parse::<u64>()?,
-        };
-        /*
         let result = match s.chars().position(|c| c == ':') {
             None => Offset {
                 timestamp: None,
@@ -105,11 +94,64 @@ impl FromStr for Offset {
                 }
             }
         };
-        */
         Ok(result)
     }
 }
 
+#[cfg(test)]
+mod tests {
+    use std::str::FromStr;
+
+    use super::Offset;
+    use crate::util::SyncTimestamp;
+
+    #[test]
+    fn offset_display_without_timestamp() {
+        let offset = Offset {
+            timestamp: None,
+            offset: 50,
+        };
+        assert_eq!(offset.to_string(), "50");
+    }
+
+    #[test]
+    fn offset_display_with_timestamp() {
+        let offset = Offset {
+            timestamp: Some(SyncTimestamp::from_milliseconds(676760)),
+            offset: 2,
+        };
+        assert_eq!(offset.to_string(), "676760:2");
+    }
+
+    #[test]
+    fn offset_without_timestamp_parsed() {
+        let original = Offset {
+            timestamp: None,
+            offset: 99,
+        };
+        let parsed = Offset::from_str(&original.to_string()).unwrap();
+        assert_eq!(parsed.offset, original.offset);
+        assert!(parsed.timestamp.is_none());
+    }
+
+    #[test]
+    fn offset_with_timestamp_parsed() {
+        let original = Offset {
+            timestamp: Some(SyncTimestamp::from_milliseconds(71138383830)),
+            offset: 3,
+        };
+        let parsed = Offset::from_str(&original.to_string()).unwrap();
+        assert_eq!(parsed.offset, original.offset);
+        assert_eq!(parsed.timestamp, original.timestamp);
+    }
+
+    #[test]
+    fn offset_fromstr_malformed_returns_error() {
+        assert!(Offset::from_str("quux").is_err());
+        assert!(Offset::from_str("wibble:buzz").is_err());
+    }
+}
+
 collection_data! {
     LockCollection {},
     DeleteCollection {},

syncstorage-db-common/src/util.rs

@@ -15,6 +15,7 @@ use diesel::{
 use serde::{Deserialize, Deserializer, Serialize, Serializer, ser};
 
 use super::error::SyncstorageDbError;
+use crate::{Sorting, params};
 
 /// Get the time since the UNIX epoch in milliseconds
 fn ms_since_epoch() -> i64 {
@@ -215,6 +216,39 @@ pub fn to_rfc3339(val: i64) -> Result<String, SyncstorageDbError> {
     )))
 }
 
+/// Encode a timestamp and the number of rows to skip for BSO query pagination.
+pub fn encode_next_offset(
+    sort: Sorting,
+    prev_offset: u64,
+    prev_timestamp: Option<i64>,
+    modified_timestamps: &[i64],
+) -> String {
+    if let Sorting::Index = sort {
+        return (prev_offset + modified_timestamps.len() as u64).to_string();
+    }
+    if modified_timestamps.is_empty() {
+        return prev_offset.to_string();
+    }
+
+    let bound = *modified_timestamps.last().unwrap();
+    let mut skip = 1usize;
+
+    skip += modified_timestamps[..modified_timestamps.len() - 1]
+        .iter()
+        .rev()
+        .take_while(|&&m| m == bound)
+        .count();
+    if skip == modified_timestamps.len() && prev_timestamp == Some(bound) {
+        skip += prev_offset as usize;
+    }
+
+    params::Offset {
+        timestamp: Some(SyncTimestamp::from_milliseconds(bound as u64)),
+        offset: skip as u64,
+    }
+    .to_string()
+}
+
 #[cfg(test)]
 mod tests {
     use std::error::Error;
@@ -262,4 +296,58 @@ mod tests {
         assert_eq!(zero, SyncTimestamp::from_i64(0).unwrap());
         assert_eq!(zero, SyncTimestamp::from_seconds(0.00));
     }
+
+    mod encode_next_offset_tests {
+        use crate::Sorting;
+        use crate::util::encode_next_offset;
+
+        #[test]
+        fn index_sort_returns_numeric_offset() {
+            let result = encode_next_offset(Sorting::Index, 50, None, &[19, 83, 747]);
+            assert_eq!(result, "53");
+        }
+
+        #[test]
+        fn empty_modified_timestamps_returns_prev_offset() {
+            let result = encode_next_offset(Sorting::Newest, 42, None, &[]);
+            assert_eq!(result, "42");
+        }
+
+        #[test]
+        fn unique_last_timestamp_skip_is_one() {
+            let result = encode_next_offset(Sorting::Newest, 0, None, &[5500, 4200, 3800]);
+            assert_eq!(result, "3800:1");
+        }
+
+        #[test]
+        fn skip_counts_identical_tail_timestamps() {
+            let result = encode_next_offset(Sorting::Newest, 0, None, &[5000, 3800, 3800]);
+            assert_eq!(result, "3800:2");
+        }
+
+        #[test]
+        fn identical_timestamps_no_prev_bound_match() {
+            let result = encode_next_offset(Sorting::Newest, 0, Some(2048), &[3800, 3800, 3800]);
+            assert_eq!(result, "3800:3");
+        }
+
+        #[test]
+        fn identical_timestamps_with_matching_prev_bound_sums() {
+            let result = encode_next_offset(Sorting::Newest, 2, Some(9000), &[9000, 9000, 9000]);
+            assert_eq!(result, "9000:5");
+        }
+
+        #[test]
+        fn oldest_sort_works() {
+            let result = encode_next_offset(Sorting::Oldest, 0, None, &[8900, 9000, 9100]);
+            assert_eq!(result, "9100:1");
+        }
+
+        #[test]
+        fn none_sort_produces_timestamp_token() {
+            // Sorting::None behaves like Newest
+            let result = encode_next_offset(Sorting::None, 0, None, &[5500, 4200, 3800]);
+            assert_eq!(result, "3800:1");
+        }
+    }
 }