acta_maker_sdk/
paging.rs

1//! Shared continuation handling for HTTP and WebSocket lists.
2use crate::ws::types::Cursor;
3use std::collections::HashSet;
4
5/// Largest `limit` any paged request accepts; the server rejects more with `invalid_limit`.
6pub const MAX_PAGE_SIZE: u32 = 200;
7
8#[derive(Default)]
9pub(crate) struct Continuation {
10    seen: HashSet<String>,
11}
12impl Continuation {
13    pub(crate) fn advance(&mut self, next: Option<&str>) -> Result<Option<Cursor>, InvalidPage> {
14        match next {
15            None => Ok(None),
16            Some(id) if id.is_empty() || !self.seen.insert(id.to_owned()) => Err(InvalidPage),
17            Some(id) => Ok(Some(Cursor::After { id: id.to_owned() })),
18        }
19    }
20}
21#[derive(Debug, thiserror::Error)]
22#[error("page returned an empty or repeated continuation")]
23pub struct InvalidPage;
24
25/// Collect a list without exposing cursor loops to the caller.
26pub async fn fetch_all<T, E, F, Fut>(
27    mut load: F,
28    next: impl Fn(&T) -> Option<&str>,
29    append: impl Fn(&mut T, T),
30) -> Result<T, E>
31where
32    F: FnMut(Cursor) -> Fut,
33    Fut: std::future::Future<Output = Result<T, E>>,
34    E: From<InvalidPage>,
35{
36    let mut all = load(Cursor::Start {}).await?;
37    let mut continuation = Continuation::default();
38    while let Some(cursor) = continuation.advance(next(&all))? {
39        let page = load(cursor).await?;
40        append(&mut all, page);
41    }
42    Ok(all)
43}
44
45#[cfg(test)]
46mod tests {
47    use super::*;
48    #[derive(Debug)]
49    struct Page {
50        rows: Vec<u8>,
51        next: Option<String>,
52    }
53    fn append(all: &mut Page, page: Page) {
54        all.rows.extend(page.rows);
55        all.next = page.next;
56    }
57    #[tokio::test]
58    async fn follows_opaque_continuation_and_rejects_cycles() {
59        let mut requests = Vec::new();
60        let result: Result<Page, InvalidPage> = fetch_all(
61            |cursor| {
62                requests.push(cursor.clone());
63                std::future::ready(Ok(match cursor {
64                    Cursor::Start {} => Page {
65                        rows: vec![1],
66                        next: Some("opaque".into()),
67                    },
68                    Cursor::After { id } => {
69                        assert_eq!(id, "opaque");
70                        Page {
71                            rows: vec![2],
72                            next: None,
73                        }
74                    }
75                }))
76            },
77            |page| page.next.as_deref(),
78            append,
79        )
80        .await;
81        assert_eq!(result.unwrap().rows, [1, 2]);
82        assert_eq!(requests.len(), 2);
83        let result: Result<Page, InvalidPage> = fetch_all(
84            |_| {
85                std::future::ready(Ok(Page {
86                    rows: vec![],
87                    next: Some("cycle".into()),
88                }))
89            },
90            |page| page.next.as_deref(),
91            append,
92        )
93        .await;
94        assert!(result.is_err());
95    }
96    #[tokio::test]
97    async fn later_failure_never_returns_a_partial_list() {
98        let result: Result<Page, InvalidPage> = fetch_all(
99            |cursor| {
100                std::future::ready(match cursor {
101                    Cursor::Start {} => Ok(Page {
102                        rows: vec![1],
103                        next: Some("next".into()),
104                    }),
105                    Cursor::After { .. } => Err(InvalidPage),
106                })
107            },
108            |page| page.next.as_deref(),
109            append,
110        )
111        .await;
112        assert!(result.is_err());
113    }
114}
acta_maker_sdk/paging.rs

acta_maker_sdk/
paging.rs
