ferric_fred/paginate.rs
1//! Auto-pagination: fetch every page of a paginated FRED endpoint in one call.
2//!
3//! FRED's list endpoints return one page at a time, alongside `count` (the total
4//! number of results across *all* pages), `offset`, and `limit`. Walking those
5//! pages by hand — bumping `offset` by `limit` until `offset >= count` — is
6//! mechanical and easy to get wrong, so [`Paginate::send_all`] does it for you
7//! and returns the full `Vec`.
8//!
9//! The two traits here are **sealed**: they are implemented for the crate's
10//! request builders and `*Results` types and cannot be implemented downstream,
11//! so the set of paginated endpoints stays closed and new methods can be added
12//! without a breaking change (the same forward-compatibility posture as
13//! [`Error`](crate::Error), ADR-0004). See ADR-0020 for the design.
14
15use std::future::Future;
16use std::time::Duration;
17
18use chrono::NaiveDate;
19use futures_core::Stream;
20
21use crate::{
22 Error, Release, ReleaseDate, ReleaseDatesResults, ReleasesResults, Result, Series,
23 SeriesSearchResults, Source, SourcesResults, Tag, TagsResults, VintageDates,
24};
25
26pub(crate) mod sealed {
27 /// Sealing marker: implemented only within this crate, so [`Page`](super::Page)
28 /// and [`Paginate`](super::Paginate) cannot be implemented downstream.
29 pub trait Sealed {}
30}
31
32/// One page of results from a paginated FRED endpoint.
33///
34/// Implemented for the crate's `*Results` types (and [`VintageDates`]). Sealed —
35/// see the [module docs](self).
36pub trait Page: sealed::Sealed {
37 /// The element type of this page (e.g. [`Series`], [`Tag`]).
38 type Item: Send;
39
40 /// FRED's `count`: the total number of results across *all* pages, not just
41 /// this one.
42 fn total(&self) -> u32;
43
44 /// The number of items on this page.
45 fn items_len(&self) -> usize;
46
47 /// Consume the page into its items.
48 fn into_items(self) -> Vec<Self::Item>;
49}
50
51/// A request builder for a paginated FRED endpoint.
52///
53/// Its headline method is [`send_all`](Paginate::send_all), which pages an
54/// endpoint to exhaustion. Sealed and implemented for the crate's paginated
55/// request builders (see the [module docs](self)); you don't implement it.
56pub trait Paginate: Clone + Send + Sized + sealed::Sealed {
57 /// The page type this request returns.
58 type Page: Page + Send;
59
60 /// FRED's maximum page size for this endpoint — the largest `limit` it
61 /// honors. 1000 for most lists; 10000 for release dates and vintage dates.
62 const MAX_PAGE: u32;
63
64 /// The caller's requested `limit`, if set. [`send_all`](Paginate::send_all)
65 /// treats it as a *ceiling* on the total number of items returned.
66 fn requested_limit(&self) -> Option<u32>;
67
68 /// The caller's requested `offset`, if set — the point
69 /// [`send_all`](Paginate::send_all) starts paging from.
70 fn requested_offset(&self) -> Option<u32>;
71
72 /// Return a copy of this request with `limit` and `offset` set.
73 #[must_use]
74 fn with_paging(self, limit: u32, offset: u32) -> Self;
75
76 /// Send a single page — equivalent to the builder's own `send`.
77 fn send_page(self) -> impl Future<Output = Result<Self::Page>> + Send;
78
79 /// Fetch **every** result across all pages, in order.
80 ///
81 /// Pages are requested in chunks of at most [`MAX_PAGE`](Paginate::MAX_PAGE).
82 /// A `limit` set on the builder caps the *total* number of items returned (a
83 /// ceiling, not a per-page size); an `offset` set on the builder is the
84 /// starting point. With neither set, the entire result set is returned.
85 ///
86 /// This issues up to ⌈`count` / `MAX_PAGE`⌉ HTTP requests. On a `429` it
87 /// retries a bounded number of times, waiting FRED's `Retry-After` when
88 /// present (see [`Error::RateLimited`]) and otherwise backing off. Mind
89 /// FRED's rate limits when collecting large result sets.
90 ///
91 /// ```no_run
92 /// # async fn run(client: &ferric_fred::Client) -> ferric_fred::Result<()> {
93 /// use ferric_fred::Paginate;
94 /// // Every source FRED knows about, across as many pages as it takes:
95 /// let sources = client.sources().send_all().await?;
96 /// // At most 250 matches for a search, however many pages that spans:
97 /// let series = client.search("gdp").limit(250).send_all().await?;
98 /// # Ok(())
99 /// # }
100 /// ```
101 ///
102 /// # Errors
103 ///
104 /// Returns the first error encountered on any page (transport, API, or
105 /// deserialize).
106 fn send_all(self) -> impl Future<Output = Result<Vec<<Self::Page as Page>::Item>>> + Send {
107 collect_all(self)
108 }
109
110 /// Stream every result across all pages, lazily — items are yielded as they
111 /// arrive, and the next page is fetched only once the current one is drained.
112 ///
113 /// Same paging semantics as [`send_all`](Paginate::send_all) (a builder
114 /// `limit` is a ceiling, `offset` is the start), but a `Stream` rather than a
115 /// materialized `Vec`: memory stays flat regardless of the total, a consumer
116 /// can stop early, and each item is a `Result` — a mid-stream error is
117 /// surfaced as an `Err` item and ends the stream, so results retrieved before
118 /// it aren't lost. On a `429` the same bounded retry as `send_all` applies.
119 ///
120 /// Drive it with [`StreamExt`](https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html)
121 /// from the `futures` crate. The returned stream is `!Unpin`, so to consume it
122 /// with `.next()` in a loop, pin it first (`pin_mut!`); the by-value
123 /// combinators like `try_collect` / `for_each` need no pinning.
124 ///
125 /// ```no_run
126 /// # async fn run(client: &ferric_fred::Client) -> ferric_fred::Result<()> {
127 /// use ferric_fred::Paginate;
128 /// use futures_util::{pin_mut, StreamExt};
129 ///
130 /// let series = client.search("gdp").stream();
131 /// pin_mut!(series);
132 /// while let Some(item) = series.next().await {
133 /// let series = item?;
134 /// println!("{}", series.id);
135 /// }
136 /// # Ok(())
137 /// # }
138 /// ```
139 fn stream(self) -> impl Stream<Item = Result<<Self::Page as Page>::Item>> + Send {
140 stream_pages(self)
141 }
142}
143
144/// Maximum number of retries for a single page when FRED returns `429`.
145const MAX_RETRIES: u32 = 3;
146
147/// Walk every page of `request`; see [`Paginate::send_all`].
148async fn collect_all<P: Paginate>(request: P) -> Result<Vec<<P::Page as Page>::Item>> {
149 let ceiling = request.requested_limit();
150 let mut offset = request.requested_offset().unwrap_or(0);
151 let mut collected: Vec<<P::Page as Page>::Item> = Vec::new();
152
153 loop {
154 // Under a ceiling, never fetch more than we still need.
155 let page_size = match ceiling {
156 Some(cap) => {
157 let remaining = cap.saturating_sub(collected.len() as u32);
158 if remaining == 0 {
159 break;
160 }
161 remaining.min(P::MAX_PAGE)
162 }
163 None => P::MAX_PAGE,
164 };
165
166 let page = send_page_with_retry(request.clone().with_paging(page_size, offset)).await?;
167 let total = page.total();
168 let got = page.items_len() as u32;
169 collected.extend(page.into_items());
170 offset = offset.saturating_add(got);
171
172 // Stop on an empty page (guards against a misreported `count`) or once
173 // we've walked past the total.
174 if got == 0 || offset >= total {
175 break;
176 }
177 }
178
179 // A ceiling that lands mid-page: trim the overshoot.
180 if let Some(cap) = ceiling {
181 collected.truncate(cap as usize);
182 }
183 Ok(collected)
184}
185
186/// Stream every page of `request`, one item at a time; see [`Paginate::stream`].
187///
188/// A page is fetched only when the previous one has been fully yielded, so at
189/// most one page is held in memory at a time. An error ends the stream after
190/// surfacing it as an `Err` item.
191fn stream_pages<P: Paginate>(
192 request: P,
193) -> impl Stream<Item = Result<<P::Page as Page>::Item>> + Send {
194 async_stream::try_stream! {
195 let ceiling = request.requested_limit();
196 let mut offset = request.requested_offset().unwrap_or(0);
197 let mut yielded: u32 = 0;
198
199 loop {
200 let page_size = match ceiling {
201 Some(cap) => {
202 let remaining = cap.saturating_sub(yielded);
203 if remaining == 0 {
204 break;
205 }
206 remaining.min(P::MAX_PAGE)
207 }
208 None => P::MAX_PAGE,
209 };
210
211 let page = send_page_with_retry(request.clone().with_paging(page_size, offset)).await?;
212 let total = page.total();
213 let got = page.items_len() as u32;
214 if got == 0 {
215 break;
216 }
217
218 for item in page.into_items() {
219 yield item;
220 yielded += 1;
221 if ceiling.is_some_and(|cap| yielded >= cap) {
222 break;
223 }
224 }
225
226 offset = offset.saturating_add(got);
227 if offset >= total || ceiling.is_some_and(|cap| yielded >= cap) {
228 break;
229 }
230 }
231 }
232}
233
234/// Send one page, retrying on `429` up to [`MAX_RETRIES`] times — waiting FRED's
235/// `Retry-After` when it provides one, otherwise backing off exponentially.
236async fn send_page_with_retry<P: Paginate>(request: P) -> Result<P::Page> {
237 let mut attempt = 0;
238 loop {
239 match request.clone().send_page().await {
240 Err(Error::RateLimited { retry_after }) if attempt < MAX_RETRIES => {
241 let delay = retry_after.unwrap_or_else(|| backoff(attempt));
242 tokio::time::sleep(delay).await;
243 attempt += 1;
244 }
245 result => return result,
246 }
247 }
248}
249
250/// Exponential backoff for a 0-based retry `attempt`: 1s, 2s, 4s.
251fn backoff(attempt: u32) -> Duration {
252 Duration::from_secs(1u64 << attempt)
253}
254
255impl sealed::Sealed for SeriesSearchResults {}
256impl Page for SeriesSearchResults {
257 type Item = Series;
258 fn total(&self) -> u32 {
259 self.count
260 }
261 fn items_len(&self) -> usize {
262 self.series.len()
263 }
264 fn into_items(self) -> Vec<Self::Item> {
265 self.series
266 }
267}
268
269impl sealed::Sealed for TagsResults {}
270impl Page for TagsResults {
271 type Item = Tag;
272 fn total(&self) -> u32 {
273 self.count
274 }
275 fn items_len(&self) -> usize {
276 self.tags.len()
277 }
278 fn into_items(self) -> Vec<Self::Item> {
279 self.tags
280 }
281}
282
283impl sealed::Sealed for ReleasesResults {}
284impl Page for ReleasesResults {
285 type Item = Release;
286 fn total(&self) -> u32 {
287 self.count
288 }
289 fn items_len(&self) -> usize {
290 self.releases.len()
291 }
292 fn into_items(self) -> Vec<Self::Item> {
293 self.releases
294 }
295}
296
297impl sealed::Sealed for ReleaseDatesResults {}
298impl Page for ReleaseDatesResults {
299 type Item = ReleaseDate;
300 fn total(&self) -> u32 {
301 self.count
302 }
303 fn items_len(&self) -> usize {
304 self.release_dates.len()
305 }
306 fn into_items(self) -> Vec<Self::Item> {
307 self.release_dates
308 }
309}
310
311impl sealed::Sealed for SourcesResults {}
312impl Page for SourcesResults {
313 type Item = Source;
314 fn total(&self) -> u32 {
315 self.count
316 }
317 fn items_len(&self) -> usize {
318 self.sources.len()
319 }
320 fn into_items(self) -> Vec<Self::Item> {
321 self.sources
322 }
323}
324
325impl sealed::Sealed for VintageDates {}
326impl Page for VintageDates {
327 type Item = NaiveDate;
328 fn total(&self) -> u32 {
329 self.count
330 }
331 fn items_len(&self) -> usize {
332 self.vintage_dates.len()
333 }
334 fn into_items(self) -> Vec<Self::Item> {
335 self.vintage_dates
336 }
337}