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

chenba · chenba · commit a70f7a8e7bba · 2026-03-18T15:22:00.000-05:00
#53 enabled #559 disabled And this patch enables it for MySQL and Postgres but not Spanner.
diff --git a/syncserver/src/web/extractors/bso_query_params.rs b/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,8 +193,7 @@ 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)
                 {
@@ -230,7 +221,7 @@ impl FromRequest for BsoQueryParams {
                     }
                 }
             }
-            */
+
             Ok(params)
         })
     }
@@ -288,19 +279,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,);
     }
 }
diff --git a/syncstorage-db-common/src/params.rs b/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(676767)),
+            offset: 2,
+        };
+        assert_eq!(offset.to_string(), "676767: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(71138383838)),
+            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 {},
diff --git a/syncstorage-db-common/src/util.rs b/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,51 @@ 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, &[5555, 4242, 3838]);
+            assert_eq!(result, "3838:1");
+        }
+
+        #[test]
+        fn skip_counts_identical_tail_timestamps() {
+            let result = encode_next_offset(Sorting::Newest, 0, None, &[5000, 3838, 3838]);
+            assert_eq!(result, "3838:2");
+        }
+
+        #[test]
+        fn identical_timestamps_no_prev_bound_match() {
+            let result = encode_next_offset(Sorting::Newest, 0, Some(2048), &[3838, 3838, 3838]);
+            assert_eq!(result, "3838:3");
+        }
+
+        #[test]
+        fn identical_timestamps_with_matching_prev_bound_sums() {
+            let result = encode_next_offset(Sorting::Newest, 2, Some(9001), &[9001, 9001, 9001]);
+            assert_eq!(result, "9001:5");
+        }
+
+        #[test]
+        fn oldest_sort_works() {
+            let result = encode_next_offset(Sorting::Oldest, 0, None, &[8999, 9000, 9001]);
+            assert_eq!(result, "9001:1");
+        }
+    }
 }
diff --git a/syncstorage-mysql/src/db/db_impl.rs b/syncstorage-mysql/src/db/db_impl.rs
@@ -10,8 +10,10 @@ use diesel::{
 };
 use diesel_async::{AsyncConnection, RunQueryDsl, TransactionManager};
 use syncstorage_db_common::{
-    DEFAULT_BSO_TTL, Db, Sorting, UserIdentifier, error::DbErrorIntrospect, params, results,
-    util::SyncTimestamp,
+    DEFAULT_BSO_TTL, Db, Sorting, UserIdentifier,
+    error::DbErrorIntrospect,
+    params, results,
+    util::{SyncTimestamp, encode_next_offset},
 };
 use syncstorage_settings::DEFAULT_MAX_TOTAL_RECORDS;
 
@@ -324,6 +326,13 @@ impl Db for MysqlDb {
             .filter(bso::expiry.gt(now))
             .into_boxed();
 
+        if let Some(ts) = params.offset.as_ref().and_then(|o| o.timestamp) {
+            match params.sort {
+                Sorting::Newest => query = query.filter(bso::modified.le(ts.as_i64())),
+                Sorting::Oldest => query = query.filter(bso::modified.ge(ts.as_i64())),
+                _ => {}
+            }
+        }
         if let Some(older) = params.older {
             query = query.filter(bso::modified.lt(older.as_i64()));
         }
@@ -340,14 +349,6 @@ impl Db for MysqlDb {
         // an error. We "fudge" a bit here by taking the id order as a secondary, since
         // that is guaranteed to be unique by the client.
         query = match params.sort {
-            // issue559: Revert to previous sorting
-            /*
-            Sorting::Index => query.order(bso::id.desc()).order(bso::sortindex.desc()),
-            Sorting::Newest | Sorting::None => {
-                query.order(bso::id.desc()).order(bso::modified.desc())
-            }
-            Sorting::Oldest => query.order(bso::id.asc()).order(bso::modified.asc()),
-            */
             Sorting::Index => query.order(bso::sortindex.desc()),
             Sorting::Newest => query.order((bso::modified.desc(), bso::id.desc())),
             Sorting::Oldest => query.order((bso::modified.asc(), bso::id.asc())),
@@ -363,6 +364,7 @@ impl Db for MysqlDb {
         // match the query conditions
         query = query.limit(if limit > 0 { limit + 1 } else { limit });
 
+        let prev_ts = params.offset.as_ref().and_then(|o| o.timestamp).map(|t| t.as_i64());
         let numeric_offset = params.offset.map_or(0, |offset| offset.offset as i64);
 
         if numeric_offset > 0 {
@@ -379,7 +381,13 @@ impl Db for MysqlDb {
 
         let next_offset = if limit >= 0 && bsos.len() > limit as usize {
             bsos.pop();
-            Some((limit + numeric_offset).to_string())
+            let modified_timestamps: Vec<i64> = bsos.iter().map(|b| b.modified.as_i64()).collect();
+            Some(encode_next_offset(
+                params.sort,
+                numeric_offset as u64,
+                prev_ts,
+                &modified_timestamps,
+            ))
         } else {
             // if an explicit "limit=0" is sent, return the offset of "0"
             // Otherwise, this would break at least the db::tests::db::get_bsos_limit_offset
@@ -420,8 +428,8 @@ impl Db for MysqlDb {
 
         query = match params.sort {
             Sorting::Index => query.order(bso::sortindex.desc()),
-            Sorting::Newest => query.order(bso::modified.desc()),
-            Sorting::Oldest => query.order(bso::modified.asc()),
+            Sorting::Newest => query.order((bso::modified.desc(), bso::id.desc())),
+            Sorting::Oldest => query.order((bso::modified.asc(), bso::id.asc())),
             _ => query,
         };
 
diff --git a/syncstorage-postgres/src/db/db_impl.rs b/syncstorage-postgres/src/db/db_impl.rs
diff --git a/syncstorage-postgres/src/db/mod.rs b/syncstorage-postgres/src/db/mod.rs
