diff --git a/syncserver/src/web/extractors/bso_query_params.rs b/syncserver/src/web/extractors/bso_query_params.rs index e48bddf9be..b497c29251 100644 --- a/syncserver/src/web/extractors/bso_query_params.rs +++ b/syncserver/src/web/extractors/bso_query_params.rs @@ -32,13 +32,6 @@ impl From for params::Offset { impl FromStr for Offset { type Err = ParseIntError; fn from_str(s: &str) -> Result { - // 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::()?, - }; - /* 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,6 +278,50 @@ 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 { @@ -295,12 +329,9 @@ mod tests { 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 index 9a6d124738..a2180da3f3 100644 --- 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 { - // 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::()?, - }; - /* 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 {}, diff --git a/syncstorage-db-common/src/util.rs b/syncstorage-db-common/src/util.rs index 2c1db8417a..1bb7776ed2 100644 --- 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 { ))) } +/// 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, + 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"); + } + } } diff --git a/syncstorage-mysql/src/db/db_impl.rs b/syncstorage-mysql/src/db/db_impl.rs index 77163ef196..82e2e66e10 100644 --- 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,15 @@ 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::Oldest => query = query.filter(bso::modified.ge(ts.as_i64())), + Sorting::Newest | Sorting::None => { + query = query.filter(bso::modified.le(ts.as_i64())) + } + Sorting::Index => {} + } + } if let Some(older) = params.older { query = query.filter(bso::modified.lt(older.as_i64())); } @@ -340,18 +351,9 @@ 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::Newest | Sorting::None => query.order((bso::modified.desc(), bso::id.desc())), Sorting::Oldest => query.order((bso::modified.asc(), bso::id.asc())), - _ => query, }; let limit = params @@ -363,6 +365,11 @@ 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 +386,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 = 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 +433,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 index cf6329b97c..4b037f91ec 100644 --- a/syncstorage-postgres/src/db/db_impl.rs +++ b/syncstorage-postgres/src/db/db_impl.rs @@ -9,7 +9,10 @@ use diesel::{ use diesel_async::{AsyncConnection, RunQueryDsl, TransactionManager}; use futures::TryStreamExt; use syncstorage_db_common::{ - DEFAULT_BSO_TTL, Db, Sorting, error::DbErrorIntrospect, params, results, util::SyncTimestamp, + DEFAULT_BSO_TTL, Db, Sorting, + error::DbErrorIntrospect, + params, results, + util::{SyncTimestamp, encode_next_offset}, }; use super::{PgDb, TOMBSTONE}; @@ -338,16 +341,38 @@ impl Db for PgDb { } async fn get_bsos(&mut self, params: params::GetBsos) -> DbResult { - let (bsos, offset) = bsos_query!(self, params, GetBso::as_select()); - let items = bsos + let (bsos, did_overflow, _limit, numeric_offset) = + bsos_query!(self, params, GetBso::as_select()); + let items: Vec = bsos .into_iter() .map(TryInto::try_into) .collect::>()?; + let offset = if did_overflow { + let modified_timestamps: Vec = items.iter().map(|b| b.modified.as_i64()).collect(); + let prev_ts = params + .offset + .as_ref() + .and_then(|o| o.timestamp) + .map(|t| t.as_i64()); + Some(encode_next_offset( + params.sort, + numeric_offset as u64, + prev_ts, + &modified_timestamps, + )) + } else { + None + }; Ok(results::GetBsos { items, offset }) } async fn get_bso_ids(&mut self, params: params::GetBsoIds) -> DbResult { - let (items, offset) = bsos_query!(self, params, bsos::bso_id); + let (items, did_overflow, limit, numeric_offset) = bsos_query!(self, params, bsos::bso_id); + let offset = if did_overflow { + Some((limit + numeric_offset).to_string()) + } else { + None + }; Ok(results::GetBsoIds { items, offset }) } diff --git a/syncstorage-postgres/src/db/mod.rs b/syncstorage-postgres/src/db/mod.rs index a296f39cef..71a35f7235 100644 --- a/syncstorage-postgres/src/db/mod.rs +++ b/syncstorage-postgres/src/db/mod.rs @@ -284,6 +284,13 @@ macro_rules! bsos_query { .filter(bsos::expiry.gt(now)) .into_boxed(); + if let Some(ts) = $params.offset.as_ref().and_then(|o| o.timestamp) { + match $params.sort { + Sorting::Oldest => query = query.filter(bsos::modified.ge(ts.as_datetime()?)), + Sorting::Newest | Sorting::None => query = query.filter(bsos::modified.le(ts.as_datetime()?)), + Sorting::Index => {} + } + } if let Some(older) = $params.older { query = query.filter(bsos::modified.lt(older.as_datetime()?)); } @@ -297,9 +304,8 @@ macro_rules! bsos_query { query = match $params.sort { Sorting::Index => query.order((bsos::sortindex.desc(), bsos::bso_id.desc())), - Sorting::Newest => query.order((bsos::modified.desc(), bsos::bso_id.desc())), + Sorting::Newest | Sorting::None => query.order((bsos::modified.desc(), bsos::bso_id.desc())), Sorting::Oldest => query.order((bsos::modified.asc(), bsos::bso_id.asc())), - _ => query, }; // fetch an extra row to detect if there are more rows that @@ -307,7 +313,7 @@ macro_rules! bsos_query { if let Some(limit) = limit { query = query.limit(limit + 1); } - let numeric_offset = $params.offset.map_or(0, |offset| offset.offset as i64); + let numeric_offset = $params.offset.as_ref().map_or(0, |offset| offset.offset as i64); if numeric_offset != 0 { // XXX: copy over this optimization: // https://github.com/mozilla-services/server-syncstorage/blob/a0f8117/syncstorage/storage/sql/__init__.py#L404 @@ -320,13 +326,11 @@ macro_rules! bsos_query { // returned in those cases let limit = limit.unwrap_or(-1); - let next_offset = if limit >= 0 && items.len() > limit as usize { + let did_overflow = limit >= 0 && items.len() > limit as usize; + if did_overflow { items.pop(); - Some((limit + numeric_offset).to_string()) - } else { - None - }; - (items, next_offset) + } + (items, did_overflow, limit, numeric_offset) } } }