use std::collections::HashSet;
use std::future::Future;
pub const MAX_PAGES: usize = 100_000;
pub const MAX_BYTES: usize = 512 * 1024 * 1024;
pub const MAX_ITEMS: u64 = 1_000_000;
pub struct Page<T> {
pub rows: Vec<T>,
pub total: Option<u64>,
}
#[derive(Debug)]
pub struct Walked<T> {
pub rows: Vec<T>,
pub stopped_at_max: bool,
}
pub fn total_of(total: Option<f64>) -> Option<u64> {
let total = total?;
let in_range = (0.0..=9_007_199_254_740_991.0).contains(&total) && total.fract() == 0.0;
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
in_range.then_some(total as u64)
}
fn pagination_error() -> scm::Error {
scm::Error::new(scm::ErrorKind::Pagination)
}
pub async fn walk<T, F, Fut>(
limit: u32,
max: u64,
mut fetch: F,
size: impl Fn(&T) -> usize,
key: impl Fn(&T) -> Option<String>,
) -> Result<Walked<T>, scm::Error>
where
F: FnMut(u64, u32) -> Fut,
Fut: Future<Output = Result<Page<T>, scm::Error>>,
{
let user_cap = max != 0 && max <= MAX_ITEMS;
let cap = if user_cap { max } else { MAX_ITEMS };
let cap = usize::try_from(cap).unwrap_or(usize::MAX);
let page_size = usize::try_from(limit).unwrap_or(usize::MAX);
let mut rows: Vec<T> = Vec::new();
let mut bytes = 0usize;
let mut seen: HashSet<String> = HashSet::new();
let mut skip: u64 = 0;
for _ in 0..MAX_PAGES {
let page = fetch(skip, limit).await?;
let received = page.rows.len();
if received == 0 {
if page.total.is_some_and(|total| skip < total) {
return Err(pagination_error());
}
return Ok(Walked {
rows,
stopped_at_max: false,
});
}
let next = skip
.checked_add(u64::try_from(received).map_err(|_| pagination_error())?)
.ok_or_else(pagination_error)?;
let more = match page.total {
Some(total) => next < total,
None => received >= page_size,
};
for row in page.rows {
if let Some(key) = key(&row)
&& !seen.insert(key)
{
return Err(pagination_error());
}
if rows.len() == cap {
return stop_at_cap(rows, user_cap);
}
bytes = bytes.saturating_add(size(&row));
if bytes > MAX_BYTES {
return Err(crate::pagination::incomplete_walk());
}
rows.push(row);
}
if !more {
return Ok(Walked {
rows,
stopped_at_max: false,
});
}
if rows.len() == cap {
return stop_at_cap(rows, user_cap);
}
skip = next;
}
Err(crate::pagination::incomplete_walk())
}
fn stop_at_cap<T>(rows: Vec<T>, user_cap: bool) -> Result<Walked<T>, scm::Error> {
if user_cap {
Ok(Walked {
rows,
stopped_at_max: true,
})
} else {
Err(crate::pagination::incomplete_walk())
}
}
pub fn slice<T>(rows: Vec<T>, params: &super::ListParams) -> (Vec<T>, bool) {
if params.all {
let (rows, end) = crate::pagination::first(rows, params.max);
return (rows, end == crate::pagination::WalkEnd::StoppedAtMax);
}
let skip = usize::try_from(params.offset.unwrap_or(0)).unwrap_or(usize::MAX);
let rows = rows.into_iter().skip(skip);
let rows = match params.limit {
Some(limit) => rows
.take(usize::try_from(limit).unwrap_or(usize::MAX))
.collect(),
None => rows.collect(),
};
(rows, false)
}
#[cfg(test)]
mod tests {
use super::*;
use std::cell::RefCell;
fn service(
count: u64,
total: Option<u64>,
skips: &RefCell<Vec<(u64, u32)>>,
) -> impl FnMut(u64, u32) -> std::future::Ready<Result<Page<u64>, scm::Error>> + '_ {
move |skip, limit| {
skips.borrow_mut().push((skip, limit));
let rows = (skip..count.min(skip + u64::from(limit))).collect();
std::future::ready(Ok(Page { rows, total }))
}
}
fn run<T>(future: impl Future<Output = T>) -> T {
tokio::runtime::Builder::new_current_thread()
.build()
.unwrap()
.block_on(future)
}
#[test]
fn walks_until_the_total_a_short_page_or_an_empty_page() {
let skips = RefCell::new(Vec::new());
let walked = run(walk(
2,
10_000,
service(5, Some(5), &skips),
|_| 8,
|row| Some(row.to_string()),
))
.unwrap();
assert_eq!(walked.rows, [0, 1, 2, 3, 4]);
assert!(!walked.stopped_at_max);
assert_eq!(*skips.borrow(), [(0, 2), (2, 2), (4, 2)]);
let skips = RefCell::new(Vec::new());
let walked = run(walk(
2,
10_000,
service(3, None, &skips),
|_| 8,
|row| Some(row.to_string()),
))
.unwrap();
assert_eq!(walked.rows, [0, 1, 2]);
assert_eq!(skips.borrow().len(), 2);
let skips = RefCell::new(Vec::new());
let walked = run(walk(
2,
10_000,
service(4, None, &skips),
|_| 8,
|row| Some(row.to_string()),
))
.unwrap();
assert_eq!(walked.rows.len(), 4);
assert_eq!(skips.borrow().len(), 3);
}
#[test]
fn a_short_page_before_the_total_continues_the_walk() {
let skips = RefCell::new(Vec::new());
let capped = |skip: u64, limit: u32| {
skips.borrow_mut().push((skip, limit));
let rows = (skip..7.min(skip + 3)).collect();
std::future::ready(Ok(Page {
rows,
total: Some(7),
}))
};
let walked = run(walk(5, 10_000, capped, |_| 8, |row| Some(row.to_string()))).unwrap();
assert_eq!(walked.rows, [0, 1, 2, 3, 4, 5, 6]);
assert_eq!(*skips.borrow(), [(0, 5), (3, 5), (6, 5)]);
}
#[test]
fn an_empty_page_before_the_total_is_an_error() {
let skips = RefCell::new(Vec::new());
let error = run(walk(
2,
10_000,
service(3, Some(9), &skips),
|_| 8,
|row| Some(row.to_string()),
))
.unwrap_err();
assert_eq!(error.kind(), scm::ErrorKind::Pagination);
let skips = RefCell::new(Vec::new());
let error = run(walk(
2,
10_000,
service(0, Some(1), &skips),
|_| 8,
|row| Some(row.to_string()),
))
.unwrap_err();
assert_eq!(error.kind(), scm::ErrorKind::Pagination);
let skips = RefCell::new(Vec::new());
let walked = run(walk(
2,
10_000,
service(0, Some(0), &skips),
|_| 8,
|row| Some(row.to_string()),
))
.unwrap();
assert!(walked.rows.is_empty());
}
#[test]
fn a_repeated_row_is_a_pagination_error() {
let skips = RefCell::new(Vec::new());
let repeating = |skip: u64, limit: u32| {
skips.borrow_mut().push((skip, limit));
std::future::ready(Ok(Page {
rows: vec![1u64, 2],
total: Some(4),
}))
};
let error = run(walk(
2,
10_000,
repeating,
|_| 8,
|row| Some(row.to_string()),
))
.unwrap_err();
assert_eq!(error.kind(), scm::ErrorKind::Pagination);
assert_eq!(*skips.borrow(), [(0, 2), (2, 2)], "stops at the repeat");
let skips = RefCell::new(Vec::new());
let walked = run(walk(
2,
10_000,
service(4, Some(4), &skips),
|_| 8,
|_| None,
))
.unwrap();
assert_eq!(walked.rows, [0, 1, 2, 3]);
}
#[test]
fn max_stops_with_a_flag_and_safety_bounds_fail() {
let skips = RefCell::new(Vec::new());
let walked = run(walk(
2,
3,
service(9, Some(9), &skips),
|_| 8,
|row| Some(row.to_string()),
))
.unwrap();
assert_eq!(walked.rows, [0, 1, 2]);
assert!(walked.stopped_at_max);
let skips = RefCell::new(Vec::new());
let walked = run(walk(
2,
4,
service(4, Some(4), &skips),
|_| 8,
|row| Some(row.to_string()),
))
.unwrap();
assert_eq!(walked.rows.len(), 4);
assert!(!walked.stopped_at_max);
let skips = RefCell::new(Vec::new());
let error = run(walk(
2,
0,
service(4, Some(4), &skips),
|_| MAX_BYTES,
|row| Some(row.to_string()),
))
.unwrap_err();
assert_eq!(error.kind(), scm::ErrorKind::Limit);
}
#[test]
fn totals_and_slices() {
assert_eq!(total_of(Some(3.0)), Some(3));
assert_eq!(total_of(Some(-1.0)), None);
assert_eq!(total_of(Some(1.5)), None);
assert_eq!(total_of(None), None);
let params = |limit, offset, all, max| super::super::ListParams {
limit,
offset,
all,
max,
};
let rows: Vec<u32> = (0..5).collect();
assert_eq!(
slice(rows.clone(), ¶ms(Some(2), Some(1), false, 0)).0,
[1, 2]
);
assert_eq!(
slice(rows.clone(), ¶ms(None, Some(3), false, 0)).0,
[3, 4]
);
assert_eq!(
slice(rows.clone(), ¶ms(None, None, true, 2)),
(vec![0, 1], true)
);
assert_eq!(
slice(rows.clone(), ¶ms(None, None, true, 0)),
(rows.clone(), false)
);
assert_eq!(
slice(rows.clone(), ¶ms(None, None, true, 5)),
(rows, false)
);
}
}