1use std::collections::BTreeSet;
2use std::error::Error;
3use std::fmt::{self, Display, Formatter};
4
5use lodestone_wire::{
6 Cursor, ErrorCode, ExportAuditEventsRequest, ExportAuditEventsResult, ExportUsageRequest, ExportUsageResult,
7 GetMaintenanceStatusRequest, GetMaintenanceStatusResult, ListAliasesRequest, ListAliasesResult,
8 ListAuditEventsRequest, ListAuditEventsResult, ListComponentTemplatesRequest, ListComponentTemplatesResult,
9 ListDataStreamsRequest, ListDataStreamsResult, ListIndexTemplatesRequest, ListIndexTemplatesResult,
10 ListIndexesRequest, ListIndexesResult, ListIngestPipelinesRequest, ListIngestPipelinesResult,
11 ListLifecyclePoliciesRequest, ListLifecyclePoliciesResult, ListOperationsRequest, ListOperationsResult,
12 ListServiceCredentialsRequest, ListServiceCredentialsResult, SearchRequest, SearchResult,
13};
14
15use crate::config::parse_u64;
16use crate::{CallOptions, Client, ClientError};
17
18const MAX_PAGES: u32 = 10_000;
19const MAX_CATALOG_RESTARTS: u32 = 8;
20
21pub struct IndexPager {
22 client: Client,
23 request: ListIndexesRequest,
24 options: CallOptions,
25 seen_cursors: BTreeSet<String>,
26 pages: u32,
27 ended: bool,
28}
29
30pub struct SearchPager {
31 client: Client,
32 request: SearchRequest,
33 options: CallOptions,
34 seen_cursors: BTreeSet<String>,
35 pages: u32,
36 ended: bool,
37}
38
39pub struct UsagePager {
40 client: Client,
41 request: ExportUsageRequest,
42 options: CallOptions,
43 seen_cursors: BTreeSet<String>,
44 prior_cumulative_through: Option<u64>,
45 prior_sequence: Option<u64>,
46 prior_authority_revision: Option<u64>,
47 pages: u32,
48 ended: bool,
49}
50
51macro_rules! define_catalog_pager {
52 ($pager:ident, $request:ty, $result:ty, $constructor:ident, $method:ident) => {
53 pub struct $pager {
54 client: Client,
55 request: $request,
56 options: CallOptions,
57 seen_cursors: BTreeSet<String>,
58 pages: u32,
59 ended: bool,
60 }
61
62 impl Client {
63 pub fn $constructor(&self, request: $request, options: CallOptions) -> Result<$pager, PagerError> {
64 validate_page_size(request.page_size)?;
65 validate_restart_count(request.restart_count)?;
66 Ok($pager {
67 client: self.clone(),
68 seen_cursors: cursor_history(&request.cursor)?,
69 request,
70 options,
71 pages: 0,
72 ended: false,
73 })
74 }
75 }
76
77 impl $pager {
78 pub async fn next_page(&mut self) -> Result<Option<$result>, PagerError> {
79 if self.ended {
80 return Ok(None);
81 }
82 consume_budget(&mut self.pages)?;
83 let result = self
84 .client
85 .$method(self.request.clone(), self.options.clone())
86 .await
87 .map_err(map_client_error)?;
88 if result.restart_count != self.request.restart_count {
89 self.ended = true;
90 return Err(PagerError::StaleCursor);
91 }
92 advance_cursor(
93 &mut self.request.cursor,
94 &mut self.seen_cursors,
95 result.next_cursor.clone(),
96 &mut self.ended,
97 )?;
98 Ok(Some(result))
99 }
100 }
101 };
102}
103
104macro_rules! define_cursor_pager {
105 ($pager:ident, $request:ty, $result:ty, $constructor:ident, $method:ident) => {
106 pub struct $pager {
107 client: Client,
108 request: $request,
109 options: CallOptions,
110 seen_cursors: BTreeSet<String>,
111 pages: u32,
112 ended: bool,
113 }
114
115 impl Client {
116 pub fn $constructor(&self, request: $request, options: CallOptions) -> Result<$pager, PagerError> {
117 validate_page_size(request.page_size)?;
118 Ok($pager {
119 client: self.clone(),
120 seen_cursors: cursor_history(&request.cursor)?,
121 request,
122 options,
123 pages: 0,
124 ended: false,
125 })
126 }
127 }
128
129 impl $pager {
130 pub async fn next_page(&mut self) -> Result<Option<$result>, PagerError> {
131 if self.ended {
132 return Ok(None);
133 }
134 consume_budget(&mut self.pages)?;
135 let result = self
136 .client
137 .$method(self.request.clone(), self.options.clone())
138 .await
139 .map_err(map_client_error)?;
140 advance_cursor(
141 &mut self.request.cursor,
142 &mut self.seen_cursors,
143 result.next_cursor.clone(),
144 &mut self.ended,
145 )?;
146 Ok(Some(result))
147 }
148 }
149 };
150}
151
152define_catalog_pager!(
153 AliasPager,
154 ListAliasesRequest,
155 ListAliasesResult,
156 alias_pager,
157 list_aliases
158);
159define_catalog_pager!(
160 DataStreamPager,
161 ListDataStreamsRequest,
162 ListDataStreamsResult,
163 data_stream_pager,
164 list_data_streams
165);
166define_catalog_pager!(
167 ComponentTemplatePager,
168 ListComponentTemplatesRequest,
169 ListComponentTemplatesResult,
170 component_template_pager,
171 list_component_templates
172);
173define_catalog_pager!(
174 IndexTemplatePager,
175 ListIndexTemplatesRequest,
176 ListIndexTemplatesResult,
177 index_template_pager,
178 list_index_templates
179);
180define_catalog_pager!(
181 IngestPipelinePager,
182 ListIngestPipelinesRequest,
183 ListIngestPipelinesResult,
184 ingest_pipeline_pager,
185 list_ingest_pipelines
186);
187define_catalog_pager!(
188 LifecyclePolicyPager,
189 ListLifecyclePoliciesRequest,
190 ListLifecyclePoliciesResult,
191 lifecycle_policy_pager,
192 list_lifecycle_policies
193);
194
195define_cursor_pager!(
196 OperationsPager,
197 ListOperationsRequest,
198 ListOperationsResult,
199 operations_pager,
200 list_operations
201);
202define_cursor_pager!(
203 AuditEventsPager,
204 ListAuditEventsRequest,
205 ListAuditEventsResult,
206 audit_events_pager,
207 list_audit_events
208);
209define_cursor_pager!(
210 AuditExportPager,
211 ExportAuditEventsRequest,
212 ExportAuditEventsResult,
213 audit_export_pager,
214 export_audit_events
215);
216define_cursor_pager!(
217 ServiceCredentialsPager,
218 ListServiceCredentialsRequest,
219 ListServiceCredentialsResult,
220 service_credentials_pager,
221 list_service_credentials
222);
223define_cursor_pager!(
224 MaintenancePager,
225 GetMaintenanceStatusRequest,
226 GetMaintenanceStatusResult,
227 maintenance_pager,
228 get_maintenance_status
229);
230
231impl Client {
232 pub fn index_pager(&self, request: ListIndexesRequest, options: CallOptions) -> Result<IndexPager, PagerError> {
233 validate_page_size(request.page_size)?;
234 validate_restart_count(request.restart_count)?;
235 Ok(IndexPager {
236 client: self.clone(),
237 seen_cursors: cursor_history(&request.cursor)?,
238 request,
239 options,
240 pages: 0,
241 ended: false,
242 })
243 }
244
245 pub fn search_pager(&self, request: SearchRequest, options: CallOptions) -> Result<SearchPager, PagerError> {
246 validate_page_size(request.page_size)?;
247 let valid_binding = match (&request.indexes, &request.pit) {
248 (Some(indexes), None) => !indexes.is_empty(),
249 (None, Some(_)) => true,
250 _ => false,
251 };
252 if !valid_binding {
253 return Err(PagerError::InvalidBinding);
254 }
255 Ok(SearchPager {
256 client: self.clone(),
257 seen_cursors: cursor_history(&request.cursor)?,
258 request,
259 options,
260 pages: 0,
261 ended: false,
262 })
263 }
264
265 pub fn usage_pager(&self, request: ExportUsageRequest, options: CallOptions) -> Result<UsagePager, PagerError> {
266 validate_page_size(request.page_size)?;
267 let start = parse_u64(&request.window_start.value).ok_or(PagerError::InvalidBinding)?;
268 let end = parse_u64(&request.window_end.value).ok_or(PagerError::InvalidBinding)?;
269 if start >= end {
270 return Err(PagerError::InvalidBinding);
271 }
272 Ok(UsagePager {
273 client: self.clone(),
274 seen_cursors: cursor_history(&request.cursor)?,
275 request,
276 options,
277 prior_cumulative_through: None,
278 prior_sequence: None,
279 prior_authority_revision: None,
280 pages: 0,
281 ended: false,
282 })
283 }
284}
285
286impl IndexPager {
287 pub async fn next_page(&mut self) -> Result<Option<ListIndexesResult>, PagerError> {
288 if self.ended {
289 return Ok(None);
290 }
291 consume_budget(&mut self.pages)?;
292 let result = self
293 .client
294 .list_indexes(self.request.clone(), self.options.clone())
295 .await
296 .map_err(map_client_error)?;
297 if result.restart_count != self.request.restart_count {
298 self.ended = true;
299 return Err(PagerError::StaleCursor);
300 }
301 advance_cursor(
302 &mut self.request.cursor,
303 &mut self.seen_cursors,
304 result.next_cursor.clone(),
305 &mut self.ended,
306 )?;
307 Ok(Some(result))
308 }
309}
310
311impl SearchPager {
312 pub async fn next_page(&mut self) -> Result<Option<SearchResult>, PagerError> {
313 if self.ended {
314 return Ok(None);
315 }
316 consume_budget(&mut self.pages)?;
317 let result = self
318 .client
319 .search(self.request.clone(), self.options.clone())
320 .await
321 .map_err(map_client_error)?;
322 advance_cursor(
323 &mut self.request.cursor,
324 &mut self.seen_cursors,
325 result.next_cursor.clone(),
326 &mut self.ended,
327 )?;
328 Ok(Some(result))
329 }
330}
331
332impl UsagePager {
333 pub async fn next_page(&mut self) -> Result<Option<ExportUsageResult>, PagerError> {
334 if self.ended {
335 return Ok(None);
336 }
337 consume_budget(&mut self.pages)?;
338 let result = self
339 .client
340 .export_usage(self.request.clone(), self.options.clone())
341 .await
342 .map_err(map_client_error)?;
343 if let Err(error) = validate_usage_page(
344 &self.request,
345 &result,
346 &mut self.prior_cumulative_through,
347 &mut self.prior_sequence,
348 &mut self.prior_authority_revision,
349 ) {
350 self.ended = true;
351 return Err(error);
352 }
353 advance_cursor(
354 &mut self.request.cursor,
355 &mut self.seen_cursors,
356 result.next_cursor.clone(),
357 &mut self.ended,
358 )?;
359 Ok(Some(result))
360 }
361}
362
363fn validate_usage_page(
364 request: &ExportUsageRequest,
365 result: &ExportUsageResult,
366 prior_cumulative_through: &mut Option<u64>,
367 prior_sequence: &mut Option<u64>,
368 prior_authority_revision: &mut Option<u64>,
369) -> Result<(), PagerError> {
370 let request_start = parse_u64(&request.window_start.value).ok_or(PagerError::InvalidBinding)?;
371 let request_end = parse_u64(&request.window_end.value).ok_or(PagerError::InvalidBinding)?;
372 let cumulative = parse_u64(&result.cumulative_through.value).ok_or(PagerError::InvalidCorePage)?;
373 if prior_cumulative_through.is_some_and(|prior| cumulative < prior) {
374 return Err(PagerError::AuthorityRegression);
375 }
376
377 let mut rows = Vec::with_capacity(result.units.len().saturating_add(result.corrections.len()));
378 for row in &result.units {
379 rows.push(validate_usage_row(
380 &row.sequence.value,
381 &row.window_start.value,
382 &row.window_end.value,
383 &row.authority_revision.value,
384 &row.source_digest,
385 request_start,
386 request_end,
387 )?);
388 }
389 for row in &result.corrections {
390 if parse_u64(&row.correction_sequence.value).is_none_or(|value| value == 0) {
391 return Err(PagerError::InvalidCorePage);
392 }
393 rows.push(validate_usage_row(
394 &row.sequence.value,
395 &row.window_start.value,
396 &row.window_end.value,
397 &row.authority_revision.value,
398 &row.source_digest,
399 request_start,
400 request_end,
401 )?);
402 }
403 rows.sort_unstable_by_key(|row| row.0);
404 for (sequence, authority_revision) in rows {
405 if prior_sequence.is_some_and(|prior| sequence <= prior)
406 || prior_authority_revision.is_some_and(|prior| authority_revision != prior)
407 {
408 return Err(PagerError::AuthorityRegression);
409 }
410 *prior_sequence = Some(sequence);
411 *prior_authority_revision = Some(authority_revision);
412 }
413 if prior_sequence.is_some_and(|sequence| cumulative < sequence)
414 || result.next_cursor.is_some() && result.units.is_empty() && result.corrections.is_empty()
415 {
416 return Err(PagerError::InvalidCorePage);
417 }
418 *prior_cumulative_through = Some(cumulative);
419 Ok(())
420}
421
422fn validate_usage_row(
423 sequence: &str,
424 window_start: &str,
425 window_end: &str,
426 authority_revision: &str,
427 source_digest: &lodestone_wire::Digest,
428 request_start: u64,
429 request_end: u64,
430) -> Result<(u64, u64), PagerError> {
431 let sequence = parse_u64(sequence).ok_or(PagerError::InvalidCorePage)?;
432 let window_start = parse_u64(window_start).ok_or(PagerError::InvalidCorePage)?;
433 let window_end = parse_u64(window_end).ok_or(PagerError::InvalidCorePage)?;
434 let authority_revision = parse_u64(authority_revision).ok_or(PagerError::InvalidCorePage)?;
435 if sequence == 0
436 || authority_revision == 0
437 || window_start < request_start
438 || window_end > request_end
439 || window_start >= window_end
440 || source_digest.validate().is_err()
441 {
442 return Err(PagerError::InvalidCorePage);
443 }
444 Ok((sequence, authority_revision))
445}
446
447fn validate_page_size(page_size: u32) -> Result<(), PagerError> {
448 if page_size == 0 || page_size > 10_000 {
449 Err(PagerError::InvalidBinding)
450 } else {
451 Ok(())
452 }
453}
454
455fn validate_restart_count(restart_count: u32) -> Result<(), PagerError> {
456 if restart_count > MAX_CATALOG_RESTARTS {
457 Err(PagerError::RestartBudgetExhausted)
458 } else {
459 Ok(())
460 }
461}
462
463fn map_client_error(error: ClientError) -> PagerError {
464 match &error {
465 ClientError::Server(body) if body.code == ErrorCode::StaleCursor => PagerError::StaleCursor,
466 ClientError::Server(body) if body.code == ErrorCode::RestartBudgetExhausted => {
467 PagerError::RestartBudgetExhausted
468 }
469 _ => PagerError::Client(error),
470 }
471}
472
473fn consume_budget(pages: &mut u32) -> Result<(), PagerError> {
474 if *pages >= MAX_PAGES {
475 return Err(PagerError::PageBudgetExhausted);
476 }
477 *pages = pages.saturating_add(1);
478 Ok(())
479}
480
481fn advance_cursor(
482 request_cursor: &mut Option<Cursor>,
483 seen_cursors: &mut BTreeSet<String>,
484 next_cursor: Option<Cursor>,
485 ended: &mut bool,
486) -> Result<(), PagerError> {
487 if let Some(next) = &next_cursor {
488 if !valid_cursor(next) || !seen_cursors.insert(next.value.clone()) {
489 *ended = true;
490 return Err(PagerError::CursorCycle);
491 }
492 } else {
493 *ended = true;
494 }
495 *request_cursor = next_cursor;
496 Ok(())
497}
498
499fn cursor_history(cursor: &Option<Cursor>) -> Result<BTreeSet<String>, PagerError> {
500 let mut seen = BTreeSet::new();
501 if let Some(cursor) = cursor {
502 if !valid_cursor(cursor) {
503 return Err(PagerError::InvalidBinding);
504 }
505 let _ = seen.insert(cursor.value.clone());
506 }
507 Ok(seen)
508}
509
510fn valid_cursor(cursor: &Cursor) -> bool {
511 !cursor.value.is_empty()
512 && cursor.value.len() <= 16 * 1024
513 && !cursor.value.bytes().any(|byte| byte.is_ascii_control())
514}
515
516#[derive(Debug)]
517pub enum PagerError {
518 InvalidBinding,
519 Client(ClientError),
520 StaleCursor,
521 RestartBudgetExhausted,
522 CursorCycle,
523 AuthorityRegression,
524 InvalidCorePage,
525 PageBudgetExhausted,
526}
527
528impl Display for PagerError {
529 fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
530 formatter.write_str(match self {
531 Self::InvalidBinding => "pager filter or page binding is invalid",
532 Self::Client(_) => "Core page request failed",
533 Self::StaleCursor => "Core catalog cursor became stale",
534 Self::RestartBudgetExhausted => "Core catalog restart budget was exhausted",
535 Self::CursorCycle => "Core pagination cursor repeated or is invalid",
536 Self::AuthorityRegression => "Core paged authority watermark regressed",
537 Self::InvalidCorePage => "Core page violates its generated semantic contract",
538 Self::PageBudgetExhausted => "Core pager reached its bounded page budget",
539 })
540 }
541}
542
543impl Error for PagerError {}
544
545#[cfg(test)]
546mod tests {
547 use super::*;
548 use serde_json::Value;
549
550 fn cursor(value: &str) -> Cursor {
551 Cursor {
552 value: value.to_owned(),
553 }
554 }
555
556 #[test]
557 fn cursor_history_refuses_non_adjacent_cycles() -> Result<(), PagerError> {
558 let mut request = None;
559 let mut seen = BTreeSet::new();
560 let mut ended = false;
561 advance_cursor(&mut request, &mut seen, Some(cursor("a")), &mut ended)?;
562 advance_cursor(&mut request, &mut seen, Some(cursor("b")), &mut ended)?;
563 assert!(matches!(
564 advance_cursor(&mut request, &mut seen, Some(cursor("a")), &mut ended),
565 Err(PagerError::CursorCycle)
566 ));
567 assert!(ended);
568 Ok(())
569 }
570
571 #[test]
572 fn initial_cursor_is_validated_and_participates_in_cycle_detection() -> Result<(), PagerError> {
573 assert!(matches!(
574 cursor_history(&Some(cursor("bad\nvalue"))),
575 Err(PagerError::InvalidBinding)
576 ));
577 let mut request = Some(cursor("initial"));
578 let mut seen = cursor_history(&request)?;
579 let mut ended = false;
580 assert!(matches!(
581 advance_cursor(&mut request, &mut seen, Some(cursor("initial")), &mut ended),
582 Err(PagerError::CursorCycle)
583 ));
584 Ok(())
585 }
586
587 #[test]
588 fn catalog_restart_budget_fails_closed() {
589 assert!(matches!(
590 validate_restart_count(MAX_CATALOG_RESTARTS.saturating_add(1)),
591 Err(PagerError::RestartBudgetExhausted)
592 ));
593 }
594
595 #[test]
596 fn shared_sdk_workflow_fixture_preserves_fail_closed_outcomes() -> Result<(), Box<dyn Error>> {
597 let fixture: Value = serde_json::from_str(include_str!(
598 "../../../clients/native/fixtures/lodestone-sdk-workflow-parity-v1.json"
599 ))?;
600 assert_eq!(fixture["format_version"], 1);
601 assert_eq!(fixture["paging"][0]["outcome"], "cursor_cycle");
602 assert_eq!(fixture["paging"][1]["outcome"], "stale_cursor");
603 assert_eq!(fixture["mutation"][0]["resubmit"], false);
604 assert_eq!(fixture["mutation"][0]["recovery"], "resolve");
605 Ok(())
606 }
607
608 #[test]
609 fn usage_sequence_watermark_is_not_compared_to_time_window() -> Result<(), PagerError> {
610 let request = ExportUsageRequest {
611 window_start: lodestone_wire::U64 { value: "10".to_owned() },
612 window_end: lodestone_wire::U64 { value: "20".to_owned() },
613 page_size: 10,
614 cursor: None,
615 };
616 let result = ExportUsageResult {
617 units: Vec::new(),
618 corrections: Vec::new(),
619 next_cursor: None,
620 cumulative_through: lodestone_wire::U64 {
621 value: "1000000".to_owned(),
622 },
623 };
624 validate_usage_page(&request, &result, &mut None, &mut None, &mut None)?;
625 Ok(())
626 }
627}