From 14944ab455f041b09f46ae9b7500225606e6a8c2 Mon Sep 17 00:00:00 2001 From: woahwhattheheck Date: Thu, 24 Sep 2026 15:31:12 -0400 Subject: [PATCH 1/6] feat(pagination): collision-proof compound cursor for batch-ingested rows Encode cursors as opaque (timestamp, id) tuples so ledger-derived rows that share an ingestion timestamp paginate without skips or duplicates. Includes batch collision and concurrent-write stability tests, plus cursor encoding docs in the module rustdoc. --- .gitignore | 3 + backend/Cargo.lock | 163 ++++++++++++++++++ backend/Cargo.toml | 15 ++ backend/src/lib.rs | 6 + backend/src/pagination/cursor.rs | 155 +++++++++++++++++ backend/src/pagination/mod.rs | 35 ++++ backend/tests/batch_collision_test.rs | 83 +++++++++ .../tests/concurrent_write_stability_test.rs | 138 +++++++++++++++ 8 files changed, 598 insertions(+) create mode 100644 backend/Cargo.lock create mode 100644 backend/Cargo.toml create mode 100644 backend/src/lib.rs create mode 100644 backend/src/pagination/cursor.rs create mode 100644 backend/src/pagination/mod.rs create mode 100644 backend/tests/batch_collision_test.rs create mode 100644 backend/tests/concurrent_write_stability_test.rs diff --git a/.gitignore b/.gitignore index 328ed907..e1201b1d 100644 --- a/.gitignore +++ b/.gitignore @@ -54,3 +54,6 @@ benchmarks/baseline.json /public/worker-*.js.map /public/fallback-*.js /public/fallback-*.js.map + +# Rust backend build +backend/target/ diff --git a/backend/Cargo.lock b/backend/Cargo.lock new file mode 100644 index 00000000..235a5330 --- /dev/null +++ b/backend/Cargo.lock @@ -0,0 +1,163 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "stellar-insights-backend" +version = "0.1.0" +dependencies = [ + "base64", + "serde", + "serde_json", + "thiserror", + "tokio", +] + +[[package]] +name = "syn" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "thiserror" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09e52cb86a36cede5cb101bf8908837b3e4c6e5e59fe7fd85c23fb56200d189e" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "pin-project-lite", + "tokio-macros", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "unicode-ident" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/backend/Cargo.toml b/backend/Cargo.toml new file mode 100644 index 00000000..54639079 --- /dev/null +++ b/backend/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "stellar-insights-backend" +version = "0.1.0" +edition = "2021" +description = "Backend services for Stellar Insights (pagination and related modules)" +license = "MIT OR Apache-2.0" + +[dependencies] +base64 = "0.22" +serde = { version = "1", features = ["derive"] } +serde_json = "1" +thiserror = "2" + +[dev-dependencies] +tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time"] } diff --git a/backend/src/lib.rs b/backend/src/lib.rs new file mode 100644 index 00000000..6a1266c2 --- /dev/null +++ b/backend/src/lib.rs @@ -0,0 +1,6 @@ +//! Stellar Insights backend library. +//! +//! Modules land here as focused, reviewable units. `pagination` provides a +//! collision-proof compound cursor for ledger-derived list endpoints. + +pub mod pagination; diff --git a/backend/src/pagination/cursor.rs b/backend/src/pagination/cursor.rs new file mode 100644 index 00000000..ee712594 --- /dev/null +++ b/backend/src/pagination/cursor.rs @@ -0,0 +1,155 @@ +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use base64::Engine; +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +/// Opaque compound cursor: `(timestamp_ms, id)`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct CompoundCursor { + /// Ingestion / event timestamp in milliseconds since Unix epoch. + pub ts: i64, + /// Monotonic tiebreaker unique within the result set. + pub id: String, +} + +impl CompoundCursor { + #[must_use] + pub fn new(ts: i64, id: impl Into) -> Self { + Self { + ts, + id: id.into(), + } + } +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum PaginateError { + #[error("invalid cursor encoding")] + InvalidEncoding, + #[error("invalid cursor payload")] + InvalidPayload, +} + +/// Encode a compound cursor as an opaque URL-safe string. +#[must_use] +pub fn encode_cursor(cursor: &CompoundCursor) -> String { + let json = serde_json::to_vec(cursor).unwrap_or_else(|_| b"{}".to_vec()); + URL_SAFE_NO_PAD.encode(json) +} + +/// Decode an opaque cursor back into `(ts, id)`. +pub fn decode_cursor(raw: &str) -> Result { + let bytes = URL_SAFE_NO_PAD + .decode(raw.trim()) + .map_err(|_| PaginateError::InvalidEncoding)?; + serde_json::from_slice(&bytes).map_err(|_| PaginateError::InvalidPayload) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SortDirection { + Asc, + Desc, +} + +/// A page of items plus the next opaque cursor (when more rows remain). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Page { + pub items: Vec, + pub next_cursor: Option, +} + +/// Row contract for pagination: expose the compound key components. +pub trait CursorKey { + fn cursor_ts(&self) -> i64; + fn cursor_id(&self) -> &str; +} + +impl CursorKey for CompoundCursor { + fn cursor_ts(&self) -> i64 { + self.ts + } + fn cursor_id(&self) -> &str { + &self.id + } +} + +fn comes_after(row: &impl CursorKey, cursor: &CompoundCursor, dir: SortDirection) -> bool { + match dir { + SortDirection::Desc => { + row.cursor_ts() < cursor.ts + || (row.cursor_ts() == cursor.ts && row.cursor_id() < cursor.id.as_str()) + } + SortDirection::Asc => { + row.cursor_ts() > cursor.ts + || (row.cursor_ts() == cursor.ts && row.cursor_id() > cursor.id.as_str()) + } + } +} + +fn cmp_keys(a: &impl CursorKey, b: &impl CursorKey, dir: SortDirection) -> std::cmp::Ordering { + let ord = a + .cursor_ts() + .cmp(&b.cursor_ts()) + .then_with(|| a.cursor_id().cmp(b.cursor_id())); + match dir { + SortDirection::Asc => ord, + SortDirection::Desc => ord.reverse(), + } +} + +/// Paginate an in-memory snapshot with a compound cursor. +/// +/// `rows` must already be filtered to the query; this helper sorts stably by +/// `(ts, id)` and seeks past `cursor` when present. Designed so unit tests can +/// prove collision behaviour without a live database. +pub fn paginate( + rows: &[T], + limit: usize, + cursor: Option<&CompoundCursor>, + direction: SortDirection, +) -> Page { + let mut ordered = rows.to_vec(); + ordered.sort_by(|a, b| cmp_keys(a, b, direction)); + + let start = match cursor { + Some(c) => ordered + .iter() + .position(|row| comes_after(row, c, direction)) + .unwrap_or(ordered.len()), + None => 0, + }; + + let end = (start + limit).min(ordered.len()); + let items = ordered[start..end].to_vec(); + let next_cursor = if end < ordered.len() { + items.last().map(|row| { + encode_cursor(&CompoundCursor::new( + row.cursor_ts(), + row.cursor_id().to_string(), + )) + }) + } else { + None + }; + + Page { items, next_cursor } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn encode_decode_roundtrip() { + let c = CompoundCursor::new(1_710_000_000_123, "42"); + let raw = encode_cursor(&c); + assert_eq!(decode_cursor(&raw).unwrap(), c); + } + + #[test] + fn rejects_garbage_cursor() { + assert_eq!(decode_cursor("%%%"), Err(PaginateError::InvalidEncoding)); + let junk = URL_SAFE_NO_PAD.encode(b"not-json"); + assert_eq!(decode_cursor(&junk), Err(PaginateError::InvalidPayload)); + } +} diff --git a/backend/src/pagination/mod.rs b/backend/src/pagination/mod.rs new file mode 100644 index 00000000..76996a85 --- /dev/null +++ b/backend/src/pagination/mod.rs @@ -0,0 +1,35 @@ +//! Collision-proof cursor pagination for batch-ingested ledger rows. +//! +//! # Why compound cursors +//! +//! Corridor / anchor / transaction tables are ledger-derived and often written +//! in batches. Many rows can share an identical ingestion timestamp after a +//! single ledger close. A cursor over timestamp alone will skip or duplicate +//! rows across page boundaries whenever two rows tie. +//! +//! # Cursor encoding +//! +//! The opaque cursor is URL-safe base64 (no padding) over a JSON object: +//! +//! ```json +//! { "ts": 1710000000123, "id": "42" } +//! ``` +//! +//! - `ts` — ingestion timestamp in milliseconds since Unix epoch (i64). +//! - `id` — monotonic tiebreaker (autoincrement id, or `ledger_seq:row_idx`). +//! +//! Seek predicate for descending pages (`ORDER BY ts DESC, id DESC`): +//! +//! ```text +//! (ts, id) < (cursor.ts, cursor.id) +//! ``` +//! +//! Ascending pages invert the comparison. Callers should maintain a compound +//! index on `(ts, id)` (or `(ingested_at, id)`) for efficient seeking. + +mod cursor; + +pub use cursor::{ + decode_cursor, encode_cursor, paginate, CompoundCursor, CursorKey, Page, PaginateError, + SortDirection, +}; diff --git a/backend/tests/batch_collision_test.rs b/backend/tests/batch_collision_test.rs new file mode 100644 index 00000000..bc381404 --- /dev/null +++ b/backend/tests/batch_collision_test.rs @@ -0,0 +1,83 @@ +//! Insert a batch of rows sharing an identical timestamp, paginate with a +//! small page size, and assert every row appears exactly once. + +use stellar_insights_backend::pagination::{ + decode_cursor, paginate, CompoundCursor, CursorKey, SortDirection, +}; + +#[derive(Debug, Clone, PartialEq, Eq)] +struct Row { + ts: i64, + id: String, + label: String, +} + +impl CursorKey for Row { + fn cursor_ts(&self) -> i64 { + self.ts + } + fn cursor_id(&self) -> &str { + &self.id + } +} + +#[test] +fn batch_with_identical_timestamps_paginates_without_skip_or_dup() { + let batch_ts = 1_720_000_000_000_i64; + let rows: Vec = (0..12) + .map(|i| Row { + ts: batch_ts, + id: format!("{i:04}"), + label: format!("row-{i}"), + }) + .collect(); + + let page_size = 3; + let mut seen = Vec::new(); + let mut cursor: Option = None; + + loop { + let page = paginate(&rows, page_size, cursor.as_ref(), SortDirection::Desc); + assert!(page.items.len() <= page_size); + for item in &page.items { + seen.push(item.id.clone()); + } + match page.next_cursor { + Some(raw) => cursor = Some(decode_cursor(&raw).expect("cursor must decode")), + None => break, + } + } + + seen.sort(); + let mut expected: Vec = rows.iter().map(|r| r.id.clone()).collect(); + expected.sort(); + assert_eq!(seen, expected, "every batched row must appear exactly once"); + assert_eq!(seen.len(), rows.len()); +} + +#[test] +fn mixed_timestamps_still_stable_across_pages() { + let mut rows = Vec::new(); + for ts in [100_i64, 100, 100, 99, 99, 98] { + let id = format!("{}-{}", ts, rows.len()); + rows.push(Row { + ts, + id: id.clone(), + label: id, + }); + } + + let mut seen = Vec::new(); + let mut cursor = None; + loop { + let page = paginate(&rows, 2, cursor.as_ref(), SortDirection::Desc); + seen.extend(page.items.iter().map(|r| r.id.clone())); + match page.next_cursor { + Some(raw) => cursor = Some(decode_cursor(&raw).unwrap()), + None => break, + } + } + assert_eq!(seen.len(), rows.len()); + let unique: std::collections::BTreeSet<_> = seen.iter().cloned().collect(); + assert_eq!(unique.len(), rows.len()); +} diff --git a/backend/tests/concurrent_write_stability_test.rs b/backend/tests/concurrent_write_stability_test.rs new file mode 100644 index 00000000..07d16343 --- /dev/null +++ b/backend/tests/concurrent_write_stability_test.rs @@ -0,0 +1,138 @@ +//! Paginate through a list while a background task inserts new matching rows. +//! Already-issued cursors' results must remain stable (no reshuffle / dup of +//! rows that were already returned). + +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use stellar_insights_backend::pagination::{ + decode_cursor, paginate, CompoundCursor, CursorKey, SortDirection, +}; +use tokio::sync::Barrier; + +#[derive(Debug, Clone, PartialEq, Eq)] +struct Row { + ts: i64, + id: String, +} + +impl CursorKey for Row { + fn cursor_ts(&self) -> i64 { + self.ts + } + fn cursor_id(&self) -> &str { + &self.id + } +} + +#[tokio::test] +async fn issued_cursors_remain_stable_under_concurrent_inserts() { + let shared = Arc::new(Mutex::new( + (0..8) + .map(|i| Row { + ts: 5_000, + id: format!("base-{i:03}"), + }) + .collect::>(), + )); + + let barrier = Arc::new(Barrier::new(2)); + + let writer_rows = Arc::clone(&shared); + let writer_barrier = Arc::clone(&barrier); + let writer = tokio::spawn(async move { + writer_barrier.wait().await; + for i in 0..8 { + { + let mut guard = writer_rows.lock().expect("lock"); + guard.push(Row { + ts: 5_000, + id: format!("new-{i:03}"), + }); + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }); + + let reader_rows = Arc::clone(&shared); + let reader_barrier = Arc::clone(&barrier); + let reader = tokio::spawn(async move { + reader_barrier.wait().await; + + // Snapshot the first page and remember its next cursor. + let first_snapshot = { + let guard = reader_rows.lock().expect("lock"); + paginate(&guard, 3, None, SortDirection::Desc) + }; + assert_eq!(first_snapshot.items.len(), 3); + let first_ids: Vec = first_snapshot + .items + .iter() + .map(|r| r.id.clone()) + .collect(); + let cursor = first_snapshot + .next_cursor + .as_deref() + .map(decode_cursor) + .transpose() + .expect("cursor") + .expect("next cursor"); + + // Allow the writer to land some inserts, then re-read from the same cursor. + tokio::time::sleep(Duration::from_millis(20)).await; + + let second = { + let guard = reader_rows.lock().expect("lock"); + paginate(&guard, 3, Some(&cursor), SortDirection::Desc) + }; + + // Stability: nothing from the first page may reappear after the cursor. + for id in &first_ids { + assert!( + second.items.iter().all(|r| &r.id != id), + "cursor must not resurface already-returned id {id}" + ); + } + + // Continuing to exhaustion from the original cursor should never + // duplicate any id within the continuation stream. + let mut continuation = second.items.iter().map(|r| r.id.clone()).collect::>(); + let mut walk: Option = second + .next_cursor + .as_deref() + .map(decode_cursor) + .transpose() + .unwrap(); + while let Some(c) = walk { + let page = { + let guard = reader_rows.lock().expect("lock"); + paginate(&guard, 3, Some(&c), SortDirection::Desc) + }; + for item in &page.items { + assert!( + !continuation.contains(&item.id), + "duplicate {} in continuation", + item.id + ); + assert!( + !first_ids.contains(&item.id), + "first-page id {} leaked into continuation", + item.id + ); + continuation.push(item.id.clone()); + } + walk = page + .next_cursor + .as_deref() + .map(decode_cursor) + .transpose() + .unwrap(); + } + + first_ids + }); + + let first_ids = reader.await.expect("reader"); + writer.await.expect("writer"); + assert_eq!(first_ids.len(), 3); +} From ddfb31a02f1aa9aa0a851a7f25f87256dc6631b6 Mon Sep 17 00:00:00 2001 From: woahwhattheheck Date: Sat, 3 Oct 2026 19:38:03 -0400 Subject: [PATCH 2/6] fix(pagination): clamp oversized page limits without overflow Use saturating addition before the existing snapshot clamp so a large usize page limit remains an upper bound after a cursor or at exhaustion. Preserve cursor ordering and encoding, and extend the existing batch regressions for both directions and maximum-size limits. --- backend/src/pagination/cursor.rs | 2 +- backend/src/pagination/mod.rs | 7 ++++ backend/tests/batch_collision_test.rs | 47 +++++++++++++++++++++++++++ 3 files changed, 55 insertions(+), 1 deletion(-) diff --git a/backend/src/pagination/cursor.rs b/backend/src/pagination/cursor.rs index ee712594..822a609d 100644 --- a/backend/src/pagination/cursor.rs +++ b/backend/src/pagination/cursor.rs @@ -119,7 +119,7 @@ pub fn paginate( None => 0, }; - let end = (start + limit).min(ordered.len()); + let end = start.saturating_add(limit).min(ordered.len()); let items = ordered[start..end].to_vec(); let next_cursor = if end < ordered.len() { items.last().map(|row| { diff --git a/backend/src/pagination/mod.rs b/backend/src/pagination/mod.rs index 76996a85..2bb30a65 100644 --- a/backend/src/pagination/mod.rs +++ b/backend/src/pagination/mod.rs @@ -26,6 +26,13 @@ //! //! Ascending pages invert the comparison. Callers should maintain a compound //! index on `(ts, id)` (or `(ingested_at, id)`) for efficient seeking. +//! +//! # Page limits +//! +//! A limit is an upper bound on the number of rows returned. After seeking past +//! a cursor, a limit larger than the remaining snapshot (including `usize::MAX`) +//! returns all remaining rows without arithmetic overflow. An exhausted cursor +//! returns an empty page. A zero limit retains its empty-page behavior. mod cursor; diff --git a/backend/tests/batch_collision_test.rs b/backend/tests/batch_collision_test.rs index bc381404..b733ba37 100644 --- a/backend/tests/batch_collision_test.rs +++ b/backend/tests/batch_collision_test.rs @@ -81,3 +81,50 @@ fn mixed_timestamps_still_stable_across_pages() { let unique: std::collections::BTreeSet<_> = seen.iter().cloned().collect(); assert_eq!(unique.len(), rows.len()); } + +#[test] +fn oversized_page_limits_return_every_remaining_collision_row() { + let rows: Vec = (0..5) + .map(|i| Row { + ts: 100, + id: format!("{i:04}"), + label: format!("row-{i}"), + }) + .collect(); + + for direction in [SortDirection::Asc, SortDirection::Desc] { + let first = paginate(&rows, 2, None, direction); + let cursor = decode_cursor(first.next_cursor.as_deref().unwrap()).unwrap(); + let expected = match direction { + SortDirection::Asc => vec!["0002", "0003", "0004"], + SortDirection::Desc => vec!["0002", "0001", "0000"], + }; + + for limit in [usize::MAX, usize::MAX - 1] { + let page = paginate(&rows, limit, Some(&cursor), direction); + let ids: Vec<&str> = page.items.iter().map(|row| row.id.as_str()).collect(); + assert_eq!(ids, expected); + assert!(page.next_cursor.is_none()); + } + } +} + +#[test] +fn oversized_page_limit_at_an_exhausted_cursor_returns_an_empty_page() { + let rows: Vec = (0..5) + .map(|i| Row { + ts: 100, + id: format!("{i:04}"), + label: format!("row-{i}"), + }) + .collect(); + + for direction in [SortDirection::Asc, SortDirection::Desc] { + let all = paginate(&rows, rows.len(), None, direction); + let last = all.items.last().unwrap(); + let cursor = CompoundCursor::new(last.ts, last.id.clone()); + let page = paginate(&rows, usize::MAX, Some(&cursor), direction); + assert!(page.items.is_empty()); + assert!(page.next_cursor.is_none()); + } +} From 31a9d399d4c775308b62b94f014e252c99e2febd Mon Sep 17 00:00:00 2001 From: woahwhattheheck Date: Sun, 4 Oct 2026 05:41:33 -0400 Subject: [PATCH 3/6] perf(pagination): clone only rows returned in a page [skip ci] Sort references to the input snapshot rather than deep-cloning the full snapshot and cloning the page a second time. Preserve stable compound-key order, cursor boundaries, the saturated page-end fix, and input ordering. Add one focused clone-count and result regression covering both directions, same-timestamp boundaries, exhausted and zero-sized pages, and oversized limits. No dependencies, public API, cursor encoding, or storage changes. Rust checks have not been executed for this continuation; existing results remain scoped to their previously tested source. --- backend/src/pagination/cursor.rs | 103 +++++++++++++++++++++++++++++-- 1 file changed, 98 insertions(+), 5 deletions(-) diff --git a/backend/src/pagination/cursor.rs b/backend/src/pagination/cursor.rs index 822a609d..7a36f46c 100644 --- a/backend/src/pagination/cursor.rs +++ b/backend/src/pagination/cursor.rs @@ -101,26 +101,27 @@ fn cmp_keys(a: &impl CursorKey, b: &impl CursorKey, dir: SortDirection) -> std:: /// /// `rows` must already be filtered to the query; this helper sorts stably by /// `(ts, id)` and seeks past `cursor` when present. Designed so unit tests can -/// prove collision behaviour without a live database. +/// prove collision behaviour without a live database. Sorting borrows the input +/// rows; only the returned page is cloned, leaving the input order untouched. pub fn paginate( rows: &[T], limit: usize, cursor: Option<&CompoundCursor>, direction: SortDirection, ) -> Page { - let mut ordered = rows.to_vec(); - ordered.sort_by(|a, b| cmp_keys(a, b, direction)); + let mut ordered: Vec<&T> = rows.iter().collect(); + ordered.sort_by(|a, b| cmp_keys(*a, *b, direction)); let start = match cursor { Some(c) => ordered .iter() - .position(|row| comes_after(row, c, direction)) + .position(|row| comes_after(*row, c, direction)) .unwrap_or(ordered.len()), None => 0, }; let end = start.saturating_add(limit).min(ordered.len()); - let items = ordered[start..end].to_vec(); + let items: Vec = ordered[start..end].iter().map(|&row| row.clone()).collect(); let next_cursor = if end < ordered.len() { items.last().map(|row| { encode_cursor(&CompoundCursor::new( @@ -152,4 +153,96 @@ mod tests { let junk = URL_SAFE_NO_PAD.encode(b"not-json"); assert_eq!(decode_cursor(&junk), Err(PaginateError::InvalidPayload)); } + + #[test] + fn paginate_clones_only_returned_rows() { + use std::cell::Cell; + use std::rc::Rc; + + struct Row { + key: CompoundCursor, + payload: String, + clones: Rc>, + } + + impl Clone for Row { + fn clone(&self) -> Self { + self.clones.set(self.clones.get() + 1); + Self { + key: self.key.clone(), + payload: self.payload.clone(), + clones: Rc::clone(&self.clones), + } + } + } + + impl CursorKey for Row { + fn cursor_ts(&self) -> i64 { + self.key.ts + } + + fn cursor_id(&self) -> &str { + &self.key.id + } + } + + let clones = Rc::new(Cell::new(0)); + let rows: Vec = [(300, "c"), (100, "a"), (400, "e"), (300, "d"), (200, "b")] + .into_iter() + .map(|(ts, id)| Row { + key: CompoundCursor::new(ts, id), + payload: format!("payload-{id}"), + clones: Rc::clone(&clones), + }) + .collect(); + let cases = [ + (SortDirection::Asc, None, 2, vec!["a", "b"], true), + (SortDirection::Desc, None, 2, vec!["e", "d"], true), + ( + SortDirection::Asc, + Some(CompoundCursor::new(300, "c")), + 1, + vec!["d"], + true, + ), + ( + SortDirection::Desc, + Some(CompoundCursor::new(300, "d")), + 1, + vec!["c"], + true, + ), + ( + SortDirection::Asc, + Some(CompoundCursor::new(400, "e")), + 10, + vec![], + false, + ), + (SortDirection::Asc, None, 0, vec![], false), + ( + SortDirection::Asc, + Some(CompoundCursor::new(200, "b")), + usize::MAX, + vec!["c", "d", "e"], + false, + ), + ]; + for (direction, cursor, limit, expected_ids, has_next) in cases { + clones.set(0); + let page = paginate(&rows, limit, cursor.as_ref(), direction); + let ids: Vec<&str> = page.items.iter().map(|row| row.cursor_id()).collect(); + assert_eq!(ids, expected_ids); + assert_eq!(clones.get(), page.items.len()); + assert_eq!(page.next_cursor.is_some(), has_next); + for row in &page.items { + assert_eq!(row.payload, format!("payload-{}", row.key.id)); + } + if let Some(next) = &page.next_cursor { + assert_eq!(decode_cursor(next).unwrap(), page.items.last().unwrap().key); + } + let original_ids: Vec<&str> = rows.iter().map(|row| row.cursor_id()).collect(); + assert_eq!(original_ids, ["c", "a", "e", "d", "b"]); + } + } } From 3572cd211d379f853b1c24abaef9b27ac1c2bafb Mon Sep 17 00:00:00 2001 From: woahwhattheheck Date: Sun, 4 Oct 2026 05:52:50 -0400 Subject: [PATCH 4/6] docs: record focused pagination clone validation [skip ci] --- docs/PAGINATION_CLONE_VALIDATION_20261004.md | 37 ++++++++++++++++++++ 1 file changed, 37 insertions(+) create mode 100644 docs/PAGINATION_CLONE_VALIDATION_20261004.md diff --git a/docs/PAGINATION_CLONE_VALIDATION_20261004.md b/docs/PAGINATION_CLONE_VALIDATION_20261004.md new file mode 100644 index 00000000..5b6b3578 --- /dev/null +++ b/docs/PAGINATION_CLONE_VALIDATION_20261004.md @@ -0,0 +1,37 @@ +# Pagination clone validation — 2026-10-04 + +This is a focused continuation of [PR 404](https://github.com/Stellar-Analysis/frontend/pull/404), preserving the original compound-cursor implementation and its saturated page-end calculation. + +## Implementation + +Source commit: `31a9d399d4c775308b62b94f014e252c99e2febd`. + +File: `backend/src/pagination/cursor.rs`. + +Source blob: `7a36f46c742fa876c222fb103095f69b62f8e232`. + +The helper previously cloned the entire input snapshot, sorted those owned rows, and cloned the selected page again. It now sorts borrowed row references and clones only the rows actually returned. For an input of N rows and a returned page of M rows, the source-level clone count changes from N + M to M. This is an operation-count statement, not a measured wall-clock, memory, or database-throughput benchmark. + +Stable compound-key ordering, ascending/descending selection, cursor encoding, the saturated end calculation, returned payloads, and the caller's input order are preserved. No dependency, public API, storage policy, or endpoint behavior was added. + +## Executed check + +[Hosted run 37193324458](https://github.com/woahwhattheheck/frontend/actions/runs/37193324458), job `111409913489`, completed successfully. The job independently checked the exact source commit and Git blob before running: + +```bash +cargo test --locked --manifest-path backend/Cargo.toml --lib paginate_clones_only_returned_rows +``` + +Runtime: Ubuntu 24.04.5; rustc 1.98.1 (`48a229cea`, 2026-09-01); cargo 1.98.1 (`797e8a9bc`, 2026-08-05). The existing lockfile was used without modification. + +Result: **1 test passed, 0 failed, 0 ignored, 2 filtered out; exit 0**. This one test executes seven cases: ascending and descending first pages; same-timestamp cursor boundaries in each direction; an exhausted cursor; a zero limit; and an oversized limit after a nonzero starting offset. Each case checks the returned IDs and payloads, one row clone per returned item, next-cursor presence/value, and unchanged original input order. The retained post-run source diff is empty. + +[Artifact 11299976692](https://github.com/woahwhattheheck/frontend/actions/runs/37193324458/artifacts/11299976692) contains the exact source identifiers, compiler versions, complete command output, exit status, and source diff. The downloaded 1,215-byte ZIP was inspected; SHA-256: + +`d8d60ccb682a45b6fc25bfac37a05715e9b2f56e1ce6f6f0efed3cccc6c068b2` + +## Scope + +Only this one focused library test was run for the continuation. The two encoding tests were filtered out, and the full integration suite, frontend, database endpoints, and concurrent-write tests were not rerun. The baseline was not executed again. Earlier results remain evidence for their originally tested revisions. + +The validation controller is on a separate branch and is not part of this submission. This documentation adds no new execution, dependency, deployment, or claim of sponsor acceptance or payment. From 1bd54e5955b96e057cf43b87b26293a36a448a67 Mon Sep 17 00:00:00 2001 From: woahwhattheheck Date: Sun, 4 Oct 2026 08:24:01 -0400 Subject: [PATCH 5/6] feat: integrate stable pagination with live backend collections [skip ci] --- backend/integration/pagination/README.md | 91 + backend/integration/pagination/manifest.json | 128 ++ .../pagination/stellar-analysis-backend.patch | 1725 +++++++++++++++++ 3 files changed, 1944 insertions(+) create mode 100644 backend/integration/pagination/README.md create mode 100644 backend/integration/pagination/manifest.json create mode 100644 backend/integration/pagination/stellar-analysis-backend.patch diff --git a/backend/integration/pagination/README.md b/backend/integration/pagination/README.md new file mode 100644 index 00000000..32511ab6 --- /dev/null +++ b/backend/integration/pagination/README.md @@ -0,0 +1,91 @@ +# Live backend cursor pagination for issue 330 + +This packet contains the complete backend implementation for +[Stellar-Analysis/frontend#330](https://github.com/Stellar-Analysis/frontend/issues/330). +It extends the existing contribution in +[frontend#404](https://github.com/Stellar-Analysis/frontend/pull/404) +without replacing its pagination library. The patch targets +[Stellar-Analysis/backend](https://github.com/Stellar-Analysis/backend) at +`965e916227cf9173cf6ad254a6971429060c1d19`. + +The separate `backend/integration/pagination/` directory keeps this change +independent of other backend integration packets. Apply it in the backend +repository, not in the frontend checkout. + +## Apply + +In a clean backend checkout at the pinned commit, use the absolute path to this +packet's patch: + +```sh +git apply --check /path/to/backend/integration/pagination/stellar-analysis-backend.patch +git apply /path/to/backend/integration/pagination/stellar-analysis-backend.patch +``` + +`manifest.json` records every changed path, its original Git blob where present, +and its resulting Git blob and SHA-256 hashes. The patch contains all 14 changed +or new files; it does not depend on uncommitted files from another checkout. + +The complete patch passed `git apply --check` against the clean pinned backend +checkout. `git diff --check` also passed. + +## Resulting behavior + +- Live anchor, corridor and ingested ledger transaction lists use an opaque + `(timestamp_ms, sequence)` cursor with a monotonic SQLite sequence. +- Migration 035 backfills immutable keys, creates matching seek indexes, and + assigns new keys atomically on source inserts. Database traversal captures an + insert ceiling, excluding both simultaneous and backdated later inserts. +- Corridor page one retains the full filtered RPC result. Continuation reads + stored payloads without querying a moving RPC window or recomputing prices. +- `/api/v1/ledger/transactions` exposes existing ledger-ingestion records; + pending signing requests retain their existing routes and behavior. +- Continuation uses a strict seek boundary from the last returned row and reads + one lookahead row. Database continuation reuses the initial count. + +The patch includes route wiring, API error conversion, OpenAPI registration, +and `docs/CURSOR_PAGINATION.md` with the request and encoding contract. + +## Compatibility and limits + +Clients must use returned cursors for continuation. Nonzero offsets return +`400 INVALID_PAGINATION`; limits are 1–100. Traversal uses ascending immutable +keys, so changing reliability or corridor metrics does not reorder results. +Corridor filters must remain the same throughout a traversal. + +Corridor snapshots expire after 15 minutes; first-page caching is capped at +5 minutes. Missing or expired snapshots return `400 CURSOR_EXPIRED`. Database +payload updates and deletions can be visible during traversal, while insert +membership, ordering and the initial total remain fixed. Ledger records may +include placeholders created by existing ingestion; this change does not add +transaction decoding. + +## Validation receipt + +The two tests expressly requested by issue 330 passed against the exact +production pagination modules, real SQLx 0.8.6 queries, and actual SQLite +migrations: + +| Maintained test target | Result | Test runtime | +| --- | --- | --- | +| `batch_collision_test` | 1 passed, 0 failed | 0.04 seconds | +| `concurrent_write_stability_test` | 1 passed, 0 failed | 0.03 seconds | + +The first test traverses tied timestamps through the actual anchor, ledger and +snapshot selectors, checking boundaries, completeness and indexed seeks. The +second commits simultaneous and backdated inserts after issuing a cursor, then +checks replay, original membership and retained corridor payloads. + +The run used a small manifest importing the unchanged production module files +with `#[path]` and matching direct dependency versions. The test files use the +actual migrations. This was necessary because shared storage exhausted during +dependency compilation of the full backend crate. Whole-server compilation, +HTTP composition and live-provider performance were not verified by that run. + +The maintained command for a backend environment with sufficient storage is: + +```sh +cargo test --locked --test batch_collision_test --test concurrent_write_stability_test +``` + +No workflows or broad test suite were added. The original Cargo.lock is unchanged. diff --git a/backend/integration/pagination/manifest.json b/backend/integration/pagination/manifest.json new file mode 100644 index 00000000..28d441bf --- /dev/null +++ b/backend/integration/pagination/manifest.json @@ -0,0 +1,128 @@ +{ + "schema_version": 1, + "issue": "https://github.com/Stellar-Analysis/frontend/issues/330", + "original_pull_request": "https://github.com/Stellar-Analysis/frontend/pull/404", + "original_pr_head": "3572cd211d379f853b1c24abaef9b27ac1c2bafb", + "target_repository": "https://github.com/Stellar-Analysis/backend", + "base_commit": "965e916227cf9173cf6ad254a6971429060c1d19", + "patch_file": "stellar-analysis-backend.patch", + "patch_sha256": "f9f94b279b114f7c5b12e3b2872a7c3477f93841edbe33a93bbc3dd496c42ea0", + "changed_files": [ + { + "path": "README.md", + "base_blob_sha": "160135699501a4db3ad4ab8439fc9f19625e7fe4", + "blob_sha": "060659ffb38c2afc7690a1a1d1c23c2eff2bc8f1", + "sha256": "755aebce63b470b6b36d8e87766c76000a1c093718c7980ac62dd5cef90f4bc9", + "bytes": 3583 + }, + { + "path": "docs/CURSOR_PAGINATION.md", + "base_blob_sha": null, + "blob_sha": "bb26fe761561fb6f5f1b21f677934614d2007eed", + "sha256": "99bd6c7f0257340bc7ec04bac008d3f84fcf7d2fbe25e17e6c92dd6f67b80f83", + "bytes": 4550 + }, + { + "path": "migrations/035_collision_proof_pagination.sql", + "base_blob_sha": null, + "blob_sha": "ded8e9b044ddb03bedab1109126602546302caab", + "sha256": "3a6553690deace11a88864093ce5956a5c3a9f83e500de1fc399d069e2aaad67", + "bytes": 3465 + }, + { + "path": "src/api/anchors.rs", + "base_blob_sha": "8c465711d899198fc66845136c21dbe6eccd4677", + "blob_sha": "5950de1a8430e8028aab7bb9b2ba68fbce0bede5", + "sha256": "838c296244d25d57b36b379551ff024c8c6d877c8f26d036cd0ade8c32f86761", + "bytes": 22813 + }, + { + "path": "src/api/corridors.rs", + "base_blob_sha": "ca1cdfeaddf36bdefdc3e1ba49d885ca87d4c044", + "blob_sha": "a8541d1e5298a0f2e5ff50f794d595f2548d15cc", + "sha256": "9e26b486b5ecb55088a362a38eb9380140f717ce8930a05ac23e236522d38d52", + "bytes": 49760 + }, + { + "path": "src/api/transactions.rs", + "base_blob_sha": "dd535f1b7caa3263e2258706acebcdac68b113bd", + "blob_sha": "6f6e0e89824dbe3325de0655bcb19abec64a8364", + "sha256": "e61c31a298938440678144031eb155d8d8c8aff6393a977ce29fe1668e8763c0", + "bytes": 9973 + }, + { + "path": "src/api/v1/mod.rs", + "base_blob_sha": "a3677c8afa44f712a391a15ff1765b6a095b37ac", + "blob_sha": "35121928ac33856ea8878007ba90e8d771d6c7fe", + "sha256": "04565c1085831bced990f948ddf9f898e189dff7ed9f3dcd788f5d6850de427f", + "bytes": 7627 + }, + { + "path": "src/cursor_pagination.rs", + "base_blob_sha": "ffb78118f41034f384bdac55b7df3918a17ce605", + "blob_sha": "4c73c7f5b339cd4d2fdf13e319e3398d2b9d0132", + "sha256": "895b81b40129cb81fa0ec75098728040a9e1f8e1f0592fc7ac537b224adecbec", + "bytes": 6070 + }, + { + "path": "src/error.rs", + "base_blob_sha": "df846e73a1ace4bddf3b5d17305ed2051e1a7260", + "blob_sha": "c90af48f30529e7d4ea32dc6b5d9af40a39d57e9", + "sha256": "3d88cd4c65e67a357f18bc1e36dc12245ee5a2b0774af71d77de787fbe03e8b6", + "bytes": 15877 + }, + { + "path": "src/lib.rs", + "base_blob_sha": "6de97c5a7dbc81abe6a2b243b22cb214544bbd3e", + "blob_sha": "489b5af6862f8196c701404123809e41a933cfe3", + "sha256": "3e33a7dc51a20106f073dc8cb5d4e266933737567be8688c59aa14b0ef20f293", + "bytes": 1797 + }, + { + "path": "src/openapi.rs", + "base_blob_sha": "b836d305a06ec70791eb86061ef379832efad1d0", + "blob_sha": "076075d3ad0a4527d2849354541439fae1145b12", + "sha256": "9ecd82a4b6abbd468a55802626f7760cc98eba4940206c746e7b00b1c79f935e", + "bytes": 11015 + }, + { + "path": "src/pagination_queries.rs", + "base_blob_sha": null, + "blob_sha": "cb235a75af35d50fa1b1478c8f818efe4af63ab1", + "sha256": "5b2260ddddc89e4788a95a5b48b083b29d1b6055fe4e86d5a43abad156a6fe6e", + "bytes": 10415 + }, + { + "path": "tests/batch_collision_test.rs", + "base_blob_sha": null, + "blob_sha": "9821a9ad12f4ae57287a4f2320b1f20d1d3b0b4f", + "sha256": "9d5665fe761f5b3055e353cd1fcae3c0da63335f4084c7f76d8fce3742edc302", + "bytes": 10830 + }, + { + "path": "tests/concurrent_write_stability_test.rs", + "base_blob_sha": null, + "blob_sha": "a0784355653f4224aba9c4cf1bbea59b83b0eef2", + "sha256": "6f01ad4baf19b1fe575a18c4a24eae6dad399ad535c335a233dbad62f29e3293", + "bytes": 11228 + } + ], + "validation": { + "batch_collision_test": { + "result": "passed", + "passed": 1, + "failed": 0, + "duration_seconds": 0.04 + }, + "concurrent_write_stability_test": { + "result": "passed", + "passed": 1, + "failed": 0, + "duration_seconds": 0.03 + }, + "execution_scope": "A small Rust manifest imported the exact production cursor_pagination.rs and pagination_queries.rs modules with #[path], using matching direct dependency versions including SQLx 0.8.6 and actual SQLite migrations. The maintained issue test files were executed without altered pagination logic.", + "whole_backend_build": "Not completed: shared storage exhausted during dependency compilation.", + "http_runtime": "Not executed.", + "patch_check": "Passed git apply --check against the complete clean backend checkout at 965e916227cf9173cf6ad254a6971429060c1d19; git diff --check also passed." + } +} diff --git a/backend/integration/pagination/stellar-analysis-backend.patch b/backend/integration/pagination/stellar-analysis-backend.patch new file mode 100644 index 00000000..bc139ce9 --- /dev/null +++ b/backend/integration/pagination/stellar-analysis-backend.patch @@ -0,0 +1,1725 @@ +diff --git a/README.md b/README.md +index 160135699501a4db3ad4ab8439fc9f19625e7fe4..060659ffb38c2afc7690a1a1d1c23c2eff2bc8f1 100644 +--- a/README.md ++++ b/README.md +@@ -10,6 +10,9 @@ + + + ++Live collection pagination is documented in [Cursor pagination](docs/CURSOR_PAGINATION.md), ++including continuation requests, immutable ordering and RPC snapshot expiry. ++ + --- + + ## What it does +diff --git a/docs/CURSOR_PAGINATION.md b/docs/CURSOR_PAGINATION.md +new file mode 100644 +index 0000000000000000000000000000000000000000..bb26fe761561fb6f5f1b21f677934614d2007eed +--- /dev/null ++++ b/docs/CURSOR_PAGINATION.md +@@ -0,0 +1,93 @@ ++# Cursor pagination for live collections ++ ++The live list handlers use an immutable `(timestamp_ms, sequence)` key. The ++sequence is an SQLite `INTEGER PRIMARY KEY AUTOINCREMENT`. Public anchor IDs, ++ledger transaction hashes and emitted corridor IDs keep their existing values. ++ ++## Requests ++ ++The versioned routes are `/api/v1/anchors`, `/api/v1/corridors` and ++`/api/v1/ledger/transactions`. The existing router also exposes the corresponding ++unversioned routes. Start with `?limit=50` and continue with ++`?limit=50&cursor=`. Retain the same corridor filters. ++ ++The response keeps `data` and `pagination`. Pagination contains `limit`, the ++initial `total`, the incoming `cursor`, `has_next`, and `next_cursor`. The next ++cursor names the last returned row, not the lookahead row. A page reads at most ++one extra row to establish continuation; exhaustion returns no next cursor. ++ ++Limits must be 1–100. Nonzero offsets return `400 INVALID_PAGINATION`; clients ++must switch to returned cursors. Traversal orders timestamp and sequence ++ascending. Anchor reliability and corridor metrics remain payload fields and ++do not reorder traversal. The existing corridor `sort_by` parameter does not ++replace that immutable ordering. ++ ++The new ledger collection reads the existing `transactions` ingestion table. ++It does not list pending signing requests, XDR payloads, or signatures. Existing ++ingestion can store placeholder transaction records; this endpoint does not ++establish complete ledger decoding. ++ ++## Cursor encoding ++ ++The opaque value is unpadded URL-safe base64 of UTF-8 JSON: ++ ++```json ++{"v":1,"scope":"anchors","ts":1767342245000,"id":42,"ceiling":999,"total":312,"snapshot":null} ++``` ++ ++`ts` and `id` form the strict seek boundary. `ceiling` excludes later inserts, ++including backdated ones. `total` retains the first-page count, avoiding a full ++count on every continuation. `scope` identifies the collection; corridor scopes ++also include SHA-256 of the filter and sort parameters. `snapshot` identifies ++a retained corridor result and is null for database collections. ++ ++Unknown fields/versions, malformed encoding or JSON, negative totals, invalid ++sequence bounds, wrong scopes and values over 2048 encoded characters are ++rejected. Cursors carry traversal state and do not authorize access. Clients ++should reuse the returned opaque value rather than constructing it. ++ ++## Database ordering ++ ++Migration 035 backfills `pagination_keys` from anchors and ingested transactions. ++It normalizes SQLite/RFC3339 timestamps to integer milliseconds; legacy null or ++unparseable values sort at zero. Insert triggers allocate keys atomically with ++source rows. Ordinary metric/timestamp updates preserve keys; delete/recreate ++allocates a fresh sequence, keeping the replacement outside older ceilings. ++ ++Page one reads ceiling, count and rows in one transaction. Continuations use: ++ ++```sql ++WHERE p.resource = ? AND p.sequence <= ? ++ AND (p.timestamp_ms, p.sequence) > (?, ?) ++ORDER BY p.timestamp_ms ASC, p.sequence ASC LIMIT ? ++``` ++ ++The matching index is `idx_pagination_keys_seek(resource,timestamp_ms,sequence)`. ++Insert membership and ordering stay stable. Existing payloads may reflect ++updates and deleted rows disappear; the initial total is retained. Historical ++payload versions and deletion history are outside this guarantee. ++ ++## RPC corridor snapshots ++ ++The first-page cache miss retains the FULL filtered RPC result in one database ++transaction. The existing aggregation, emitted asset-pair IDs and filters remain. ++The snapshot is indexed by `(snapshot_id,timestamp_ms,sequence)`. Continuations ++read its payloads directly, without fetching payments/trades, recomputing prices ++or reevaluating filters against a moving RPC window. ++ ++Snapshots expire after 15 minutes. Expired rows are removed when a new snapshot ++is stored. Missing/expired snapshots return `400 CURSOR_EXPIRED`; the handler ++does not substitute a fresh RPC result. First-page caching is capped at 5 minutes, ++and cache keys distinguish scope/cursor. Start a new traversal after expiration. ++ ++## Required checks ++ ++```sh ++cargo test --locked --test batch_collision_test --test concurrent_write_stability_test ++``` ++ ++These two files exercise the production selectors with SQLite and actual ++migrations 001/002/007/035: batch collisions, concurrent same-time/backdated ++inserts, cursor replay and retained corridor results after a newer snapshot. ++They cover selectors/storage, not HTTP composition, live-provider benchmarking, ++or sponsor acceptance. Runtime execution status belongs in the PR receipt. +diff --git a/migrations/035_collision_proof_pagination.sql b/migrations/035_collision_proof_pagination.sql +new file mode 100644 +index 0000000000000000000000000000000000000000..ded8e9b044ddb03bedab1109126602546302caab +--- /dev/null ++++ b/migrations/035_collision_proof_pagination.sql +@@ -0,0 +1,86 @@ ++-- Stable list identities are separate from mutable metrics and caller-facing IDs. ++-- SQLite and RFC3339 creation timestamps are normalized to integer milliseconds. ++-- Legacy NULL or unparseable timestamps sort at epoch zero. Updating an entity's ++-- metrics or creation-time text never moves an already issued pagination key. ++CREATE TABLE IF NOT EXISTS pagination_keys ( ++ sequence INTEGER PRIMARY KEY AUTOINCREMENT, ++ resource TEXT NOT NULL, ++ row_id TEXT NOT NULL, ++ timestamp_ms INTEGER NOT NULL, ++ UNIQUE (resource, row_id) ++); ++ ++CREATE INDEX IF NOT EXISTS idx_pagination_keys_seek ++ ON pagination_keys(resource, timestamp_ms, sequence); ++ ++INSERT OR IGNORE INTO pagination_keys(resource, row_id, timestamp_ms) ++SELECT 'anchors', id, ++ COALESCE(CAST(strftime('%s', created_at) AS INTEGER) * 1000 ++ + CAST(substr(strftime('%f', created_at), 4, 3) AS INTEGER), 0) ++FROM anchors ++ORDER BY COALESCE(CAST(strftime('%s', created_at) AS INTEGER) * 1000 ++ + CAST(substr(strftime('%f', created_at), 4, 3) AS INTEGER), 0), ++ rowid; ++ ++INSERT OR IGNORE INTO pagination_keys(resource, row_id, timestamp_ms) ++SELECT 'transactions', hash, ++ COALESCE(CAST(strftime('%s', created_at) AS INTEGER) * 1000 ++ + CAST(substr(strftime('%f', created_at), 4, 3) AS INTEGER), 0) ++FROM transactions ++ORDER BY COALESCE(CAST(strftime('%s', created_at) AS INTEGER) * 1000 ++ + CAST(substr(strftime('%f', created_at), 4, 3) AS INTEGER), 0), ++ rowid; ++ ++CREATE TRIGGER IF NOT EXISTS pagination_anchor_insert ++AFTER INSERT ON anchors ++BEGIN ++ INSERT INTO pagination_keys(resource, row_id, timestamp_ms) ++ VALUES ('anchors', NEW.id, ++ COALESCE(CAST(strftime('%s', NEW.created_at) AS INTEGER) * 1000 ++ + CAST(substr(strftime('%f', NEW.created_at), 4, 3) AS INTEGER), 0)); ++END; ++ ++CREATE TRIGGER IF NOT EXISTS pagination_anchor_delete ++AFTER DELETE ON anchors ++BEGIN ++ DELETE FROM pagination_keys WHERE resource = 'anchors' AND row_id = OLD.id; ++END; ++ ++CREATE TRIGGER IF NOT EXISTS pagination_transaction_insert ++AFTER INSERT ON transactions ++BEGIN ++ INSERT INTO pagination_keys(resource, row_id, timestamp_ms) ++ VALUES ('transactions', NEW.hash, ++ COALESCE(CAST(strftime('%s', NEW.created_at) AS INTEGER) * 1000 ++ + CAST(substr(strftime('%f', NEW.created_at), 4, 3) AS INTEGER), 0)); ++END; ++ ++CREATE TRIGGER IF NOT EXISTS pagination_transaction_delete ++AFTER DELETE ON transactions ++BEGIN ++ DELETE FROM pagination_keys WHERE resource = 'transactions' AND row_id = OLD.hash; ++END; ++ ++-- A remote corridor read has no durable database creation order. Store its ++-- filtered rows once, then seek through that same snapshot until it expires. ++CREATE TABLE IF NOT EXISTS pagination_snapshots ( ++ id TEXT PRIMARY KEY, ++ scope TEXT NOT NULL, ++ created_at_ms INTEGER NOT NULL, ++ expires_at_ms INTEGER NOT NULL ++); ++ ++CREATE INDEX IF NOT EXISTS idx_pagination_snapshots_expiration ++ ON pagination_snapshots(expires_at_ms); ++ ++CREATE TABLE IF NOT EXISTS pagination_snapshot_rows ( ++ sequence INTEGER PRIMARY KEY AUTOINCREMENT, ++ snapshot_id TEXT NOT NULL REFERENCES pagination_snapshots(id) ON DELETE CASCADE, ++ timestamp_ms INTEGER NOT NULL, ++ row_id TEXT NOT NULL, ++ payload TEXT NOT NULL, ++ UNIQUE (snapshot_id, row_id) ++); ++ ++CREATE INDEX IF NOT EXISTS idx_pagination_snapshot_rows_seek ++ ON pagination_snapshot_rows(snapshot_id, timestamp_ms, sequence); +diff --git a/src/api/anchors.rs b/src/api/anchors.rs +index 8c465711d899198fc66845136c21dbe6eccd4677..5950de1a8430e8028aab7bb9b2ba68fbce0bede5 100644 +--- a/src/api/anchors.rs ++++ b/src/api/anchors.rs +@@ -14,12 +14,12 @@ use uuid::Uuid; + + use crate::broadcast::broadcast_anchor_update; + use crate::cache::helpers::cached_query; +-use crate::cache::keys; + use crate::cache::CacheManager; ++use crate::cursor_pagination::{validate_page_request, CompoundCursor, CursorPaginatedResponse}; + use crate::database::Database; + use crate::error::{ApiError, ApiResult}; + use crate::models::{AnchorDetailResponse, CreateAnchorRequest}; +-use crate::pagination::PaginatedResponse; ++use crate::pagination_queries; + use crate::rpc::circuit_breaker::rpc_circuit_breaker; + use crate::rpc::error::{with_retry, RetryConfig, RpcError}; + use crate::rpc::StellarRpcClient; +@@ -305,6 +305,8 @@ pub struct ListAnchorsQuery { + #[serde(default)] + #[param(example = 0)] + pub offset: i64, ++ /// Opaque continuation from pagination.next_cursor. Omit for a new traversal. ++ pub cursor: Option, + } + + const fn default_limit() -> i64 { +@@ -411,7 +413,8 @@ pub struct AnchorsResponse { + path = "/api/anchors", + params(ListAnchorsQuery), + responses( +- (status = 200, description = "List of anchors retrieved successfully", body = AnchorsResponse), ++ (status = 200, description = "List of anchors retrieved successfully", body = CursorPaginatedResponse), ++ (status = 400, description = "Invalid cursor, limit, or offset"), + (status = 500, description = "Internal server error") + ), + tag = "Anchors" +@@ -427,27 +430,44 @@ pub async fn get_anchors( + Query(params): Query, + headers: HeaderMap, + ) -> ApiResult { +- let cache_key = keys::anchor_list(params.limit, params.offset); ++ validate_page_request(params.limit, params.offset) ++ .map_err(|message| ApiError::bad_request("INVALID_PAGINATION", message))?; ++ if let Some(cursor) = params.cursor.as_deref() { ++ let decoded = CompoundCursor::decode(cursor, "anchors") ++ .map_err(|message| ApiError::bad_request("INVALID_CURSOR", message))?; ++ if decoded.snapshot.is_some() { ++ return Err(ApiError::bad_request( ++ "INVALID_CURSOR", ++ "Unexpected RPC snapshot", ++ )); ++ } ++ } ++ let cache_key = format!( ++ "anchor:list:cursor:v1:{}:{}", ++ params.limit, ++ params.cursor.as_deref().unwrap_or("first") ++ ); + + let response = cached_query( + &cache, + &cache_key, + cache.config.get_ttl("anchor"), + || async { +- // Get anchor metadata from database (names, accounts, etc.) +- let anchors: Vec = +- db.list_anchors(params.limit, params.offset).await?; +- +- // Total count for pagination metadata (runs in parallel with list query) +- let total = db.count_anchors().await.unwrap_or(0); ++ let CursorPaginatedResponse { ++ data: anchors, ++ pagination, ++ } = pagination_queries::list_anchors::( ++ db.pool(), ++ params.limit, ++ params.cursor.as_deref(), ++ ) ++ .await?; + + if anchors.is_empty() { +- return Ok(PaginatedResponse::new( +- Vec::::new(), +- total, +- params.limit, +- params.offset, +- )); ++ return Ok(CursorPaginatedResponse { ++ data: Vec::::new(), ++ pagination, ++ }); + } + + // OPTIMIZATION: Batch fetch all assets for these anchors (1 query instead of N) +@@ -545,12 +565,10 @@ pub async fn get_anchors( + anchor_responses.push(anchor_response); + } + +- Ok(PaginatedResponse::new( +- anchor_responses, +- total, +- params.limit, +- params.offset, +- )) ++ Ok(CursorPaginatedResponse { ++ data: anchor_responses, ++ pagination, ++ }) + }, + ) + .await?; +@@ -563,6 +581,7 @@ pub async fn get_anchors( + #[cfg(test)] + mod tests { + use super::*; ++ use crate::cache::keys; + use crate::cache::{CacheConfig, CacheManager}; + use crate::rpc::circuit_breaker::rpc_circuit_breaker; + use crate::rpc::StellarRpcClient; +diff --git a/src/api/corridors.rs b/src/api/corridors.rs +index ca1cdfeaddf36bdefdc3e1ba49d885ca87d4c044..a8541d1e5298a0f2e5ff50f794d595f2548d15cc 100644 +--- a/src/api/corridors.rs ++++ b/src/api/corridors.rs +@@ -15,11 +15,12 @@ use crate::broadcast::broadcast_corridor_update; + use crate::cache::helpers::cached_query; + use crate::cache::keys; + use crate::cache::CacheManager; ++use crate::cursor_pagination::{validate_page_request, CompoundCursor, CursorPaginatedResponse}; + use crate::database::Database; + use crate::error::{ApiError, ApiResult}; + use crate::models::corridor::Corridor; + use crate::models::{CreateCorridorRequest, SortBy}; +-use crate::pagination::PaginatedResponse; ++use crate::pagination_queries; + use crate::request_id::RequestId; + use crate::rpc::{ + circuit_breaker::rpc_circuit_breaker, +@@ -219,6 +220,8 @@ pub struct ListCorridorsQuery { + #[serde(default)] + #[param(example = 0)] + pub offset: i64, ++ /// Opaque continuation. Retain the same filters for every page of a traversal. ++ pub cursor: Option, + /// Sort by field (`success_rate` or volume) + #[serde(default)] + pub sort_by: SortBy, +@@ -288,7 +291,27 @@ fn generate_corridor_list_cache_key(params: &ListCorridorsQuery) -> String { + params.asset_code, + params.time_period + ); +- keys::corridor_list(params.limit, params.offset, &filter_str) ++ format!( ++ "{}:cursor:v1:{}:{}", ++ keys::corridor_list(params.limit, params.offset, &filter_str), ++ generate_corridor_cursor_scope(params), ++ params.cursor.as_deref().unwrap_or("first") ++ ) ++} ++ ++fn generate_corridor_cursor_scope(params: &ListCorridorsQuery) -> String { ++ use sha2::{Digest, Sha256}; ++ let filters = format!( ++ "{:?}|{:?}|{:?}|{:?}|{:?}|{:?}|{:?}", ++ params.success_rate_min, ++ params.success_rate_max, ++ params.volume_min, ++ params.volume_max, ++ params.asset_code, ++ params.time_period, ++ params.sort_by, ++ ); ++ format!("corridors:{:x}", Sha256::digest(filters.as_bytes())) + } + + /// List all payment corridors +@@ -306,18 +329,19 @@ fn generate_corridor_list_cache_key(params: &ListCorridorsQuery) -> String { + path = "/api/corridors", + params(ListCorridorsQuery), + responses( +- (status = 200, description = "List of corridors retrieved successfully", body = Vec), ++ (status = 200, description = "List of corridors retrieved successfully", body = CursorPaginatedResponse), ++ (status = 400, description = "Invalid or expired cursor, invalid limit or offset"), + (status = 500, description = "Internal server error") + ), + tag = "Corridors" + )] + #[tracing::instrument( +- skip(_db, cache, rpc_client, price_feed, params), ++ skip(db, cache, rpc_client, price_feed, params), + fields(request_id = %request_id.0, query = ?params) + )] + pub async fn list_corridors( + Extension(request_id): Extension, +- State((_db, cache, rpc_client, price_feed)): State<( ++ State((db, cache, rpc_client, price_feed)): State<( + Arc, + Arc, + Arc, +@@ -328,6 +352,14 @@ pub async fn list_corridors( + ) -> ApiResult { + info!("Listing corridors"); + ++ validate_page_request(params.limit, params.offset) ++ .map_err(|message| ApiError::bad_request("INVALID_PAGINATION", message))?; ++ let scope = generate_corridor_cursor_scope(¶ms); ++ if let Some(cursor) = params.cursor.as_deref() { ++ CompoundCursor::decode(cursor, &scope) ++ .map_err(|message| ApiError::bad_request("INVALID_CURSOR", message))?; ++ } ++ + validation::validate_corridor_filters( + params.success_rate_min, + params.success_rate_max, +@@ -337,10 +369,27 @@ pub async fn list_corridors( + + let cache_key = generate_corridor_list_cache_key(¶ms); + ++ if let Some(cursor) = params.cursor.as_deref() { ++ let corridors = pagination_queries::snapshot_page::( ++ db.pool(), ++ &scope, ++ params.limit, ++ Some(cursor), ++ None, ++ ) ++ .await?; ++ crate::observability::metrics::set_corridors_tracked(corridors.pagination.total); ++ return crate::http_cache::cached_json_response( ++ &headers, ++ &cache_key, ++ &corridors, ++ cache.config.get_ttl("corridor").min(300), ++ ); ++ } + let corridors = cached_query( + &cache, + &cache_key, +- cache.config.get_ttl("corridor"), ++ cache.config.get_ttl("corridor").min(300), + || async { + let circuit_breaker = rpc_circuit_breaker(); + +@@ -391,6 +440,7 @@ pub async fn list_corridors( + + // Calculate metrics for each corridor + let mut corridor_responses = Vec::new(); ++ let observed_at = chrono::Utc::now(); + + for (corridor_key, corridor_payments) in &corridor_map { + let total_attempts = corridor_payments.len() as i64; +@@ -457,7 +507,7 @@ pub async fn list_corridors( + liquidity_volume_24h_usd: volume_usd * 0.1, + liquidity_trend, + health_score, +- last_updated: chrono::Utc::now().to_rfc3339(), ++ last_updated: observed_at.to_rfc3339(), + }; + + corridor_responses.push(corridor_response); +@@ -502,27 +552,28 @@ pub async fn list_corridors( + }) + .collect(); + +- // Apply limit/offset pagination to the filtered results +- let total = filtered.len() as i64; +- let page: Vec<_> = filtered ++ // Freeze the FULL filtered RPC result before selecting page one. ++ // Continuations read these payloads without refetching a moving RPC window. ++ let rows = filtered + .into_iter() +- .skip(params.offset as usize) +- .take(params.limit as usize) ++ .map(|row| (observed_at.timestamp_millis(), row.id.clone(), row)) + .collect(); +- +- Ok(PaginatedResponse::new( +- page, +- total, ++ let snapshot = pagination_queries::store_snapshot(db.pool(), &scope, rows).await?; ++ Ok(pagination_queries::snapshot_page::( ++ db.pool(), ++ &scope, + params.limit, +- params.offset, +- )) ++ None, ++ Some(&snapshot), ++ ) ++ .await?) + }, + ) + .await?; + + crate::observability::metrics::set_corridors_tracked(corridors.pagination.total); + +- let ttl = cache.config.get_ttl("corridor"); ++ let ttl = cache.config.get_ttl("corridor").min(300); + let response = crate::http_cache::cached_json_response(&headers, &cache_key, &corridors, ttl)?; + Ok(response) + } +diff --git a/src/api/transactions.rs b/src/api/transactions.rs +index dd535f1b7caa3263e2258706acebcdac68b113bd..6f6e0e89824dbe3325de0655bcb19abec64a8364 100644 +--- a/src/api/transactions.rs ++++ b/src/api/transactions.rs +@@ -1,10 +1,12 @@ + use axum::{ +- extract::{Path, State}, ++ extract::{Path, Query, State}, + http::StatusCode, + routing::{get, post}, + Json, Router, + }; + use serde::Deserialize; ++use std::sync::Arc; ++use utoipa::IntoParams; + use uuid::Uuid; + + use crate::{ +@@ -12,6 +14,54 @@ use crate::{ + state::AppState, + }; + ++use crate::cursor_pagination::{validate_page_request, CursorPaginatedResponse}; ++use crate::database::Database; ++use crate::error::{ApiError, ApiResult}; ++use crate::pagination_queries::{self, LedgerTransaction}; ++ ++/// Query parameters for the ingested ledger transaction collection. ++#[derive(Debug, Deserialize, IntoParams)] ++#[into_params(parameter_in = Query)] ++pub struct ListLedgerTransactionsQuery { ++ #[serde(default = "default_list_limit")] ++ pub limit: i64, ++ #[serde(default)] ++ pub offset: i64, ++ pub cursor: Option, ++} ++ ++const fn default_list_limit() -> i64 { ++ 50 ++} ++ ++/// List stored ledger-ingestion records; pending signing requests are separate. ++#[utoipa::path( ++ get, ++ path = "/api/v1/ledger/transactions", ++ params(ListLedgerTransactionsQuery), ++ responses( ++ (status = 200, description = "Stored ledger transaction page", body = CursorPaginatedResponse), ++ (status = 400, description = "Invalid cursor, limit, or offset"), ++ (status = 500, description = "Database error") ++ ), ++ tag = "Transactions" ++)] ++pub async fn list_ledger_transactions( ++ State(db): State>, ++ Query(params): Query, ++) -> ApiResult>> { ++ validate_page_request(params.limit, params.offset) ++ .map_err(|message| ApiError::bad_request("INVALID_PAGINATION", message))?; ++ Ok(Json( ++ pagination_queries::list_ledger_transactions( ++ db.pool(), ++ params.limit, ++ params.cursor.as_deref(), ++ ) ++ .await?, ++ )) ++} ++ + // Request/Response DTOs + #[derive(Debug, Deserialize)] + pub struct CreateTransactionRequest { +diff --git a/src/api/v1/mod.rs b/src/api/v1/mod.rs +index a3677c8afa44f712a391a15ff1765b6a095b37ac..35121928ac33856ea8878007ba90e8d771d6c7fe 100644 +--- a/src/api/v1/mod.rs ++++ b/src/api/v1/mod.rs +@@ -98,6 +98,12 @@ pub fn routes( + + // Captured before `app_state` is moved into `protected_routes` below. + let wallets_db = app_state.db.clone(); ++ let ledger_routes = Router::new() ++ .route( ++ "/ledger/transactions", ++ get(crate::api::transactions::list_ledger_transactions), ++ ) ++ .with_state(app_state.db.clone()); + + // 2. Public anchor routes + let public_anchor_routes = Router::new() +@@ -174,6 +180,7 @@ pub fn routes( + // V1 router (mounted at /api/v1 and also preserved at root for compatibility) + let v1_router = Router::new() + .merge(cached_routes) ++ .merge(ledger_routes) + .merge(public_anchor_routes) + .merge(protected_routes) + .merge(protected_webhook_routes) +diff --git a/src/cursor_pagination.rs b/src/cursor_pagination.rs +index ffb78118f41034f384bdac55b7df3918a17ce605..4c73c7f5b339cd4d2fdf13e319e3398d2b9d0132 100644 +--- a/src/cursor_pagination.rs ++++ b/src/cursor_pagination.rs +@@ -18,6 +18,62 @@ + use serde::{Deserialize, Serialize}; + use utoipa::ToSchema; + ++/// A versioned seek boundary over an immutable timestamp and monotonic sequence. ++/// `ceiling` excludes rows inserted after the first page, including backdated rows. ++/// RPC collections also bind the boundary to a retained result snapshot. ++#[derive(Debug, Clone, Serialize, Deserialize)] ++#[serde(deny_unknown_fields)] ++pub struct CompoundCursor { ++ pub v: u8, ++ pub scope: String, ++ pub ts: i64, ++ pub id: i64, ++ pub ceiling: i64, ++ pub total: i64, ++ pub snapshot: Option, ++} ++ ++impl CompoundCursor { ++ pub fn encode(&self) -> Result { ++ use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; ++ Ok(URL_SAFE_NO_PAD.encode(serde_json::to_vec(self)?)) ++ } ++ ++ pub fn decode(encoded: &str, scope: &str) -> Result { ++ use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; ++ if encoded.len() > 2048 { ++ return Err("Cursor exceeds the maximum encoded length".to_string()); ++ } ++ let bytes = URL_SAFE_NO_PAD ++ .decode(encoded) ++ .map_err(|_| "Invalid cursor encoding".to_string())?; ++ let cursor: Self = ++ serde_json::from_slice(&bytes).map_err(|_| "Invalid cursor payload".to_string())?; ++ if cursor.v != 1 ++ || cursor.scope != scope ++ || cursor.id <= 0 ++ || cursor.id > cursor.ceiling ++ || cursor.total < 0 ++ || cursor.snapshot.as_ref().is_some_and(|id| id.len() > 64) ++ { ++ return Err("Cursor does not match this collection or traversal".to_string()); ++ } ++ Ok(cursor) ++ } ++} ++ ++/// Cursor pages use a bounded positive limit. Offset traversal has been replaced ++/// by the returned opaque cursor; mixing the two would reintroduce unstable seeks. ++pub fn validate_page_request(limit: i64, offset: i64) -> Result<(), String> { ++ if !(1..=100).contains(&limit) { ++ return Err("limit must be between 1 and 100".to_string()); ++ } ++ if offset != 0 { ++ return Err("Use pagination.next_cursor instead of a nonzero offset".to_string()); ++ } ++ Ok(()) ++} ++ + /// Cursor-based pagination metadata for efficient traversal. + #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] + pub struct CursorPageMeta { +@@ -41,7 +97,13 @@ pub struct CursorPageMeta { + impl CursorPageMeta { + /// Create pagination metadata for a cursor-based response. + #[must_use] +- pub fn new(total: i64, limit: i64, cursor: Option, has_next: bool, next_cursor: Option) -> Self { ++ pub fn new( ++ total: i64, ++ limit: i64, ++ cursor: Option, ++ has_next: bool, ++ next_cursor: Option, ++ ) -> Self { + Self { + limit, + total, +@@ -115,7 +177,13 @@ mod tests { + + #[test] + fn cursor_page_meta_has_next() { +- let meta = CursorPageMeta::new(100, 50, Some("cursor1".to_string()), true, Some("cursor2".to_string())); ++ let meta = CursorPageMeta::new( ++ 100, ++ 50, ++ Some("cursor1".to_string()), ++ true, ++ Some("cursor2".to_string()), ++ ); + assert!(meta.has_next); + assert_eq!(meta.limit, 50); + assert_eq!(meta.total, 100); +diff --git a/src/error.rs b/src/error.rs +index df846e73a1ace4bddf3b5d17305ed2051e1a7260..c90af48f30529e7d4ea32dc6b5d9af40a39d57e9 100644 +--- a/src/error.rs ++++ b/src/error.rs +@@ -276,6 +276,20 @@ impl From for ApiError { + } + } + ++impl From for ApiError { ++ fn from(error: crate::pagination_queries::PaginationError) -> Self { ++ use crate::pagination_queries::PaginationError; ++ match error { ++ PaginationError::InvalidCursor(message) => Self::bad_request("INVALID_CURSOR", message), ++ PaginationError::ExpiredSnapshot => Self::bad_request( ++ "CURSOR_EXPIRED", ++ "This result snapshot has expired; restart from the first page", ++ ), ++ other => Self::internal("PAGINATION_ERROR", other.to_string()), ++ } ++ } ++} ++ + /// Convert from `sqlx::Error` + impl From for ApiError { + fn from(err: sqlx::Error) -> Self { +diff --git a/src/lib.rs b/src/lib.rs +index 6de97c5a7dbc81abe6a2b243b22cb214544bbd3e..489b5af6862f8196c701404123809e41a933cfe3 100644 +--- a/src/lib.rs ++++ b/src/lib.rs +@@ -50,6 +50,7 @@ pub mod network; + pub mod observability; + pub mod openapi; + pub mod pagination; ++pub mod pagination_queries; + pub mod payload_limit; + pub mod queue; // Idempotent offline-sync queue processor (issue #93) + pub mod rate_limit; +diff --git a/src/openapi.rs b/src/openapi.rs +index b836d305a06ec70791eb86061ef379832efad1d0..076075d3ad0a4527d2849354541439fae1145b12 100644 +--- a/src/openapi.rs ++++ b/src/openapi.rs +@@ -28,6 +28,7 @@ use utoipa::OpenApi; + // Corridors + crate::api::corridors::list_corridors, + crate::api::corridors::get_corridor_detail, ++ crate::api::transactions::list_ledger_transactions, + // Price Feed + crate::api::price_feed::get_price, + crate::api::price_feed::get_prices, +diff --git a/src/pagination_queries.rs b/src/pagination_queries.rs +new file mode 100644 +index 0000000000000000000000000000000000000000..cb235a75af35d50fa1b1478c8f818efe4af63ab1 +--- /dev/null ++++ b/src/pagination_queries.rs +@@ -0,0 +1,329 @@ ++//! Indexed cursor selectors used by the live list handlers. ++//! ++//! Database pages freeze insert membership with a sequence ceiling. Corridor ++//! pages retain the complete filtered RPC payload so a moving RPC window or a ++//! changed metric cannot remove rows halfway through traversal. ++ ++use chrono::Utc; ++use serde::{de::DeserializeOwned, Deserialize, Serialize}; ++use sqlx::{sqlite::SqliteRow, FromRow, QueryBuilder, Row, Sqlite, SqlitePool}; ++use utoipa::ToSchema; ++ ++use crate::cursor_pagination::{CompoundCursor, CursorPaginatedResponse}; ++ ++const SNAPSHOT_TTL_MS: i64 = 15 * 60 * 1000; ++ ++#[derive(Debug, thiserror::Error)] ++pub enum PaginationError { ++ #[error("Invalid cursor: {0}")] ++ InvalidCursor(String), ++ #[error("This result snapshot has expired; restart from the first page")] ++ ExpiredSnapshot, ++ #[error(transparent)] ++ Database(#[from] sqlx::Error), ++ #[error(transparent)] ++ Json(#[from] serde_json::Error), ++} ++ ++fn parse_cursor( ++ encoded: Option<&str>, ++ scope: &str, ++) -> Result, PaginationError> { ++ encoded ++ .map(|value| CompoundCursor::decode(value, scope).map_err(PaginationError::InvalidCursor)) ++ .transpose() ++} ++ ++/// Stored ledger-ingestion records. This deliberately excludes pending signing ++/// requests, their XDR payloads and collected signatures. ++#[derive(Debug, Clone, Serialize, Deserialize, FromRow, ToSchema)] ++pub struct LedgerTransaction { ++ pub hash: String, ++ pub ledger_sequence: i64, ++ pub source_account: Option, ++ pub fee: Option, ++ pub operation_count: Option, ++ pub successful: Option, ++ pub created_at: Option, ++} ++ ++pub async fn list_anchors( ++ pool: &SqlitePool, ++ limit: i64, ++ cursor: Option<&str>, ++) -> Result, PaginationError> ++where ++ T: for<'row> FromRow<'row, SqliteRow> + Serialize, ++{ ++ database_page(pool, "anchors", "anchors", "id", limit, cursor).await ++} ++ ++pub async fn list_ledger_transactions( ++ pool: &SqlitePool, ++ limit: i64, ++ cursor: Option<&str>, ++) -> Result, PaginationError> { ++ database_page(pool, "transactions", "transactions", "hash", limit, cursor).await ++} ++ ++// Table and identity names are private constants from the two callers above. ++// Query values, including all cursor fields, are always bound parameters. ++async fn database_page( ++ pool: &SqlitePool, ++ resource: &str, ++ table: &str, ++ identity: &str, ++ limit: i64, ++ encoded: Option<&str>, ++) -> Result, PaginationError> ++where ++ T: for<'row> FromRow<'row, SqliteRow> + Serialize, ++{ ++ crate::cursor_pagination::validate_page_request(limit, 0) ++ .map_err(PaginationError::InvalidCursor)?; ++ let cursor = parse_cursor(encoded, resource)?; ++ if cursor ++ .as_ref() ++ .is_some_and(|value| value.snapshot.is_some()) ++ { ++ return Err(PaginationError::InvalidCursor( ++ "Unexpected RPC snapshot".to_string(), ++ )); ++ } ++ let mut transaction = pool.begin().await?; ++ let ceiling = match &cursor { ++ Some(value) => value.ceiling, ++ None => { ++ sqlx::query_scalar::<_, i64>( ++ "SELECT COALESCE(MAX(sequence), 0) FROM pagination_keys WHERE resource = ?", ++ ) ++ .bind(resource) ++ .fetch_one(&mut *transaction) ++ .await? ++ } ++ }; ++ let joined = format!( ++ " FROM pagination_keys p JOIN {table} r ON r.{identity} = p.row_id \ ++ WHERE p.resource = " ++ ); ++ let total = if let Some(value) = &cursor { ++ value.total ++ } else { ++ let mut count = QueryBuilder::::new(format!("SELECT COUNT(*){joined}")); ++ count ++ .push_bind(resource) ++ .push(" AND p.sequence <= ") ++ .push_bind(ceiling); ++ count ++ .build_query_scalar::() ++ .fetch_one(&mut *transaction) ++ .await? ++ }; ++ ++ let mut query = QueryBuilder::::new(format!( ++ "SELECT r.*, p.timestamp_ms AS page_timestamp, p.sequence AS page_sequence{joined}" ++ )); ++ query ++ .push_bind(resource) ++ .push(" AND p.sequence <= ") ++ .push_bind(ceiling); ++ if let Some(value) = &cursor { ++ query ++ .push(" AND (p.timestamp_ms, p.sequence) > (") ++ .push_bind(value.ts) ++ .push(", ") ++ .push_bind(value.id) ++ .push(")"); ++ } ++ query ++ .push(" ORDER BY p.timestamp_ms ASC, p.sequence ASC LIMIT ") ++ .push_bind(limit + 1); ++ let rows = query.build().fetch_all(&mut *transaction).await?; ++ transaction.commit().await?; ++ page_from_rows( ++ rows, ++ total, ++ limit, ++ encoded, ++ resource, ++ ceiling, ++ None, ++ |row| Ok(T::from_row(row)?), ++ ) ++} ++ ++fn page_from_rows( ++ mut rows: Vec, ++ total: i64, ++ limit: i64, ++ encoded: Option<&str>, ++ scope: &str, ++ ceiling: i64, ++ snapshot: Option<&str>, ++ decode: F, ++) -> Result, PaginationError> ++where ++ T: Serialize, ++ F: Fn(&SqliteRow) -> Result, ++{ ++ let has_next = rows.len() > limit as usize; ++ rows.truncate(limit as usize); ++ let next_cursor = if has_next { ++ rows.last() ++ .map(|row| { ++ Ok::<_, PaginationError>( ++ CompoundCursor { ++ v: 1, ++ scope: scope.to_string(), ++ ts: row.try_get("page_timestamp")?, ++ id: row.try_get("page_sequence")?, ++ ceiling, ++ total, ++ snapshot: snapshot.map(str::to_string), ++ } ++ .encode()?, ++ ) ++ }) ++ .transpose()? ++ } else { ++ None ++ }; ++ let data = rows.iter().map(decode).collect::, _>>()?; ++ Ok(CursorPaginatedResponse::new( ++ data, ++ total, ++ limit, ++ encoded.map(str::to_string), ++ has_next, ++ next_cursor, ++ )) ++} ++ ++/// Retain a complete RPC result in one transaction before returning its first page. ++/// Each tuple contains (observed timestamp in ms, emitted public ID, payload). ++pub async fn store_snapshot( ++ pool: &SqlitePool, ++ scope: &str, ++ mut rows: Vec<(i64, String, T)>, ++) -> Result { ++ rows.sort_by(|left, right| (left.0, &left.1).cmp(&(right.0, &right.1))); ++ let id = uuid::Uuid::new_v4().to_string(); ++ let now = Utc::now().timestamp_millis(); ++ let mut transaction = pool.begin().await?; ++ sqlx::query( ++ "DELETE FROM pagination_snapshot_rows WHERE snapshot_id IN \ ++ (SELECT id FROM pagination_snapshots WHERE expires_at_ms <= ?)", ++ ) ++ .bind(now) ++ .execute(&mut *transaction) ++ .await?; ++ sqlx::query("DELETE FROM pagination_snapshots WHERE expires_at_ms <= ?") ++ .bind(now) ++ .execute(&mut *transaction) ++ .await?; ++ sqlx::query( ++ "INSERT INTO pagination_snapshots (id, scope, created_at_ms, expires_at_ms) \ ++ VALUES (?, ?, ?, ?)", ++ ) ++ .bind(&id) ++ .bind(scope) ++ .bind(now) ++ .bind(now + SNAPSHOT_TTL_MS) ++ .execute(&mut *transaction) ++ .await?; ++ for (timestamp_ms, row_id, payload) in rows { ++ sqlx::query( ++ "INSERT INTO pagination_snapshot_rows \ ++ (snapshot_id, timestamp_ms, row_id, payload) VALUES (?, ?, ?, ?)", ++ ) ++ .bind(&id) ++ .bind(timestamp_ms) ++ .bind(row_id) ++ .bind(serde_json::to_string(&payload)?) ++ .execute(&mut *transaction) ++ .await?; ++ } ++ transaction.commit().await?; ++ Ok(id) ++} ++ ++/// Continue a retained RPC result. A missing/expired snapshot is an explicit error; ++/// callers must never replace it with a newly aggregated RPC result. ++pub async fn snapshot_page( ++ pool: &SqlitePool, ++ scope: &str, ++ limit: i64, ++ encoded: Option<&str>, ++ first_snapshot: Option<&str>, ++) -> Result, PaginationError> { ++ crate::cursor_pagination::validate_page_request(limit, 0) ++ .map_err(PaginationError::InvalidCursor)?; ++ let cursor = parse_cursor(encoded, scope)?; ++ let snapshot = match (&cursor, first_snapshot) { ++ (Some(value), None) => value.snapshot.as_deref(), ++ (None, Some(id)) => Some(id), ++ _ => None, ++ } ++ .ok_or_else(|| { ++ PaginationError::InvalidCursor("Missing or ambiguous result snapshot".to_string()) ++ })?; ++ let mut transaction = pool.begin().await?; ++ let valid: i64 = sqlx::query_scalar( ++ "SELECT COUNT(*) FROM pagination_snapshots \ ++ WHERE id = ? AND scope = ? AND expires_at_ms > ?", ++ ) ++ .bind(snapshot) ++ .bind(scope) ++ .bind(Utc::now().timestamp_millis()) ++ .fetch_one(&mut *transaction) ++ .await?; ++ if valid != 1 { ++ return Err(PaginationError::ExpiredSnapshot); ++ } ++ let (total, ceiling): (i64, i64) = sqlx::query_as( ++ "SELECT COUNT(*), COALESCE(MAX(sequence), 0) \ ++ FROM pagination_snapshot_rows WHERE snapshot_id = ?", ++ ) ++ .bind(snapshot) ++ .fetch_one(&mut *transaction) ++ .await?; ++ if cursor ++ .as_ref() ++ .is_some_and(|value| value.ceiling != ceiling || value.total != total) ++ { ++ return Err(PaginationError::InvalidCursor( ++ "Snapshot ceiling changed".to_string(), ++ )); ++ } ++ let mut query = QueryBuilder::::new( ++ "SELECT payload, timestamp_ms AS page_timestamp, sequence AS page_sequence \ ++ FROM pagination_snapshot_rows WHERE snapshot_id = ", ++ ); ++ query.push_bind(snapshot); ++ if let Some(value) = &cursor { ++ query ++ .push(" AND (timestamp_ms, sequence) > (") ++ .push_bind(value.ts) ++ .push(", ") ++ .push_bind(value.id) ++ .push(")"); ++ } ++ query ++ .push(" ORDER BY timestamp_ms ASC, sequence ASC LIMIT ") ++ .push_bind(limit + 1); ++ let rows = query.build().fetch_all(&mut *transaction).await?; ++ transaction.commit().await?; ++ page_from_rows( ++ rows, ++ total, ++ limit, ++ encoded, ++ scope, ++ ceiling, ++ Some(snapshot), ++ |row| { ++ let payload: String = row.try_get("payload")?; ++ Ok(serde_json::from_str(&payload)?) ++ }, ++ ) ++} +diff --git a/tests/batch_collision_test.rs b/tests/batch_collision_test.rs +new file mode 100644 +index 0000000000000000000000000000000000000000..9821a9ad12f4ae57287a4f2320b1f20d1d3b0b4f +--- /dev/null ++++ b/tests/batch_collision_test.rs +@@ -0,0 +1,308 @@ ++use std::{ ++ collections::{HashMap, HashSet}, ++ error::Error, ++}; ++ ++use serde_json::{json, Value}; ++use sqlx::{sqlite::SqlitePoolOptions, Row, SqlitePool}; ++use stellar_analysis_backend::{ ++ cursor_pagination::{CompoundCursor, CursorPageMeta}, ++ pagination_queries::{list_anchors, list_ledger_transactions, snapshot_page, store_snapshot}, ++}; ++ ++const CREATED_AT: &str = "2026-01-22T10:00:00.123Z"; ++const BEFORE: [u8; 5] = [9, 1, 7, 0, 5]; ++const AFTER: [u8; 6] = [8, 2, 10, 4, 6, 3]; ++const LIMIT: i64 = 3; ++const SNAPSHOT_SCOPE: &str = "corridors:USD:active"; ++ ++#[derive(Debug, serde::Serialize, sqlx::FromRow)] ++struct AnchorRow { ++ id: String, ++} ++ ++async fn seed(pool: &SqlitePool, ids: &[u8]) -> Result<(), sqlx::Error> { ++ for id in ids { ++ sqlx::query( ++ "INSERT INTO anchors (id, name, stellar_account, created_at) VALUES (?, ?, ?, ?)", ++ ) ++ .bind(format!("anchor-{id:02}")) ++ .bind(format!("Anchor {id}")) ++ .bind(format!("account-{id:02}")) ++ .bind(CREATED_AT) ++ .execute(pool) ++ .await?; ++ sqlx::query( ++ "INSERT INTO transactions (hash, ledger_sequence, source_account, created_at) \ ++ VALUES (?, 1, ?, ?)", ++ ) ++ .bind(format!("tx-{id:02}")) ++ .bind(format!("account-{id:02}")) ++ .bind(CREATED_AT) ++ .execute(pool) ++ .await?; ++ } ++ Ok(()) ++} ++ ++async fn read_page( ++ pool: &SqlitePool, ++ scope: &str, ++ cursor: Option<&str>, ++ first_snapshot: Option<&str>, ++) -> Result<(Vec, CursorPageMeta), Box> { ++ Ok(match scope { ++ "anchors" => { ++ let page = list_anchors::(pool, LIMIT, cursor).await?; ++ ( ++ page.data.into_iter().map(|row| row.id).collect(), ++ page.pagination, ++ ) ++ } ++ "transactions" => { ++ let page = list_ledger_transactions(pool, LIMIT, cursor).await?; ++ ( ++ page.data.into_iter().map(|row| row.hash).collect(), ++ page.pagination, ++ ) ++ } ++ _ => { ++ let page = snapshot_page::(pool, scope, LIMIT, cursor, first_snapshot).await?; ++ let ids = page ++ .data ++ .iter() ++ .map(|row| { ++ row["id"] ++ .as_str() ++ .map(str::to_owned) ++ .ok_or("snapshot payload lacks an ID") ++ }) ++ .collect::, _>>()?; ++ (ids, page.pagination) ++ } ++ }) ++} ++ ++fn assert_seek_index(rows: Vec, index: &str) -> Result<(), sqlx::Error> { ++ let plan = rows ++ .iter() ++ .map(|row| row.try_get::("detail")) ++ .collect::, _>>()? ++ .join("\n"); ++ assert!( ++ plan.contains(index), ++ "Expected {index} in query plan:\n{plan}" ++ ); ++ assert!( ++ !plan.contains("USE TEMP B-TREE"), ++ "Unexpected pagination sort:\n{plan}" ++ ); ++ Ok(()) ++} ++ ++#[tokio::test] ++async fn identical_timestamps_preserve_every_row_across_live_selectors_and_rpc_snapshot( ++) -> Result<(), Box> { ++ let pool = SqlitePoolOptions::new() ++ .max_connections(1) ++ .connect("sqlite::memory:") ++ .await?; ++ sqlx::query("PRAGMA foreign_keys = ON") ++ .execute(&pool) ++ .await?; ++ for migration in [ ++ include_str!("../migrations/001_create_anchors.sql"), ++ include_str!("../migrations/002_create_metrics_corridors_snapshots.sql"), ++ include_str!("../migrations/007_create_ledger_ingestion_tables.sql"), ++ ] { ++ sqlx::raw_sql(migration).execute(&pool).await?; ++ } ++ sqlx::query("INSERT INTO ledgers (sequence, hash, close_time) VALUES (1, 'ledger-1', ?)") ++ .bind(CREATED_AT) ++ .execute(&pool) ++ .await?; ++ seed(&pool, &BEFORE).await?; ++ sqlx::raw_sql(include_str!( ++ "../migrations/035_collision_proof_pagination.sql" ++ )) ++ .execute(&pool) ++ .await?; ++ let backfilled: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pagination_keys") ++ .fetch_one(&pool) ++ .await?; ++ assert_eq!(backfilled, (BEFORE.len() * 2) as i64); ++ seed(&pool, &AFTER).await?; ++ ++ let timestamp_ms = chrono::DateTime::parse_from_rfc3339(CREATED_AT)?.timestamp_millis(); ++ let total = BEFORE.len() + AFTER.len(); ++ let snapshot = store_snapshot( ++ &pool, ++ SNAPSHOT_SCOPE, ++ (0..total) ++ .rev() ++ .map(|id| { ++ let row_id = format!("corridor-{id:02}"); ++ ( ++ timestamp_ms, ++ row_id.clone(), ++ json!({ "id": row_id, "status": "active" }), ++ ) ++ }) ++ .collect::>(), ++ ) ++ .await?; ++ ++ for scope in ["anchors", "transactions", SNAPSHOT_SCOPE] { ++ let is_snapshot = scope == SNAPSHOT_SCOPE; ++ let snapshot_id = is_snapshot.then_some(snapshot.as_str()); ++ let expected: Vec = if is_snapshot { ++ (0..total).map(|id| format!("corridor-{id:02}")).collect() ++ } else { ++ let prefix = if scope == "anchors" { "anchor" } else { "tx" }; ++ BEFORE ++ .iter() ++ .chain(AFTER.iter()) ++ .map(|id| format!("{prefix}-{id:02}")) ++ .collect() ++ }; ++ let key_rows: Vec<(String, i64, i64)> = if is_snapshot { ++ sqlx::query_as( ++ "SELECT row_id, timestamp_ms, sequence FROM pagination_snapshot_rows \ ++ WHERE snapshot_id = ?", ++ ) ++ .bind(&snapshot) ++ .fetch_all(&pool) ++ .await? ++ } else { ++ sqlx::query_as( ++ "SELECT row_id, timestamp_ms, sequence FROM pagination_keys \ ++ WHERE resource = ?", ++ ) ++ .bind(scope) ++ .fetch_all(&pool) ++ .await? ++ }; ++ assert_eq!( ++ key_rows.len(), ++ total, ++ "{scope} missed backfill or trigger registration" ++ ); ++ let ceiling = key_rows ++ .iter() ++ .map(|row| row.2) ++ .max() ++ .ok_or("missing pagination keys")?; ++ let keys: HashMap<_, _> = key_rows ++ .into_iter() ++ .map(|(id, ts, seq)| (id, (ts, seq))) ++ .collect(); ++ let mut cursor: Option = None; ++ let mut seen = Vec::new(); ++ let mut unique = HashSet::new(); ++ let mut previous_key = None; ++ let page_count = total.div_ceil(LIMIT as usize); ++ assert!(page_count > 2); ++ ++ for page_number in 0..page_count { ++ let (ids, meta) = read_page( ++ &pool, ++ scope, ++ cursor.as_deref(), ++ if page_number == 0 { snapshot_id } else { None }, ++ ) ++ .await?; ++ assert_eq!( ++ ids.len(), ++ (total - seen.len()).min(LIMIT as usize), ++ "{scope}" ++ ); ++ assert_eq!(meta.total, total as i64, "{scope} total changed"); ++ assert_eq!(meta.limit, LIMIT); ++ assert_eq!(meta.cursor.as_deref(), cursor.as_deref()); ++ assert_eq!(meta.has_next, page_number + 1 < page_count, "{scope}"); ++ assert_eq!(meta.next_cursor.is_some(), meta.has_next, "{scope}"); ++ for id in &ids { ++ assert!(unique.insert(id.clone()), "{scope} repeated {id}"); ++ let key = *keys.get(id).ok_or("returned ID has no pagination key")?; ++ assert_eq!( ++ key.0, timestamp_ms, ++ "{scope} timestamp normalization changed" ++ ); ++ assert!(key.1 > 0 && key.1 <= ceiling); ++ if let Some(previous) = previous_key { ++ assert!(key > previous, "{scope} tuple keys did not advance"); ++ } ++ previous_key = Some(key); ++ } ++ if let Some(next) = &meta.next_cursor { ++ let boundary = CompoundCursor::decode(next, scope)?; ++ assert_eq!( ++ Some((boundary.ts, boundary.id)), ++ previous_key, ++ "{scope} cursor must identify the last emitted row" ++ ); ++ assert_eq!(boundary.ceiling, ceiling); ++ assert_eq!(boundary.snapshot.as_deref(), snapshot_id); ++ } ++ seen.extend(ids); ++ cursor = meta.next_cursor; ++ } ++ assert_eq!(seen, expected, "{scope} lost rows or changed their order"); ++ assert_eq!(unique.len(), total); ++ assert!( ++ cursor.is_none(), ++ "{scope} advertised continuation beyond the last page" ++ ); ++ ++ let (ts, id) = previous_key.ok_or("no final row")?; ++ let terminal = CompoundCursor { ++ v: 1, ++ scope: scope.to_owned(), ++ ts, ++ id, ++ ceiling, ++ total: total as i64, ++ snapshot: snapshot_id.map(str::to_owned), ++ } ++ .encode()?; ++ let (ids, meta) = read_page(&pool, scope, Some(&terminal), None).await?; ++ assert!(ids.is_empty(), "{scope} returned rows past the final tuple"); ++ assert_eq!(meta.total, total as i64); ++ assert!(!meta.has_next); ++ assert!(meta.next_cursor.is_none()); ++ ++ if scope == "anchors" { ++ let rows = sqlx::query( ++ "EXPLAIN QUERY PLAN \ ++ SELECT r.*, p.timestamp_ms AS page_timestamp, p.sequence AS page_sequence \ ++ FROM pagination_keys p JOIN anchors r ON r.id = p.row_id \ ++ WHERE p.resource = ? AND p.sequence <= ? AND (p.timestamp_ms, p.sequence) > (?, ?) \ ++ ORDER BY p.timestamp_ms ASC, p.sequence ASC LIMIT ?", ++ ) ++ .bind(scope) ++ .bind(ceiling) ++ .bind(timestamp_ms) ++ .bind(0_i64) ++ .bind(LIMIT + 1) ++ .fetch_all(&pool) ++ .await?; ++ assert_seek_index(rows, "idx_pagination_keys_seek")?; ++ } else if is_snapshot { ++ let rows = sqlx::query( ++ "EXPLAIN QUERY PLAN \ ++ SELECT payload, timestamp_ms AS page_timestamp, sequence AS page_sequence \ ++ FROM pagination_snapshot_rows WHERE snapshot_id = ? \ ++ AND (timestamp_ms, sequence) > (?, ?) \ ++ ORDER BY timestamp_ms ASC, sequence ASC LIMIT ?", ++ ) ++ .bind(&snapshot) ++ .bind(timestamp_ms) ++ .bind(0_i64) ++ .bind(LIMIT + 1) ++ .fetch_all(&pool) ++ .await?; ++ assert_seek_index(rows, "idx_pagination_snapshot_rows_seek")?; ++ } ++ } ++ Ok(()) ++} +diff --git a/tests/concurrent_write_stability_test.rs b/tests/concurrent_write_stability_test.rs +new file mode 100644 +index 0000000000000000000000000000000000000000..a0784355653f4224aba9c4cf1bbea59b83b0eef2 +--- /dev/null ++++ b/tests/concurrent_write_stability_test.rs +@@ -0,0 +1,319 @@ ++use serde_json::{json, Value}; ++use sqlx::sqlite::SqlitePoolOptions; ++use stellar_analysis_backend::pagination_queries::{ ++ list_anchors, list_ledger_transactions, snapshot_page, store_snapshot, PaginationError, ++}; ++ ++#[derive(Debug, serde::Serialize, sqlx::FromRow)] ++struct AnchorRow { ++ id: String, ++} ++ ++#[tokio::test] ++async fn traversal_survives_concurrent_inserts_updates_and_new_rpc_snapshots( ++) -> Result<(), Box> { ++ const ROWS: i64 = 7; ++ const LIMIT: i64 = 2; ++ const CREATED: &str = "2026-01-02 03:04:05"; ++ const SCOPE: &str = "corridors:asset=USDC"; ++ ++ let pool = SqlitePoolOptions::new() ++ .max_connections(1) ++ .connect("sqlite::memory:") ++ .await?; ++ for migration in [ ++ include_str!("../migrations/001_create_anchors.sql"), ++ include_str!("../migrations/002_create_metrics_corridors_snapshots.sql"), ++ include_str!("../migrations/007_create_ledger_ingestion_tables.sql"), ++ include_str!("../migrations/035_collision_proof_pagination.sql"), ++ ] { ++ sqlx::raw_sql(migration).execute(&pool).await?; ++ } ++ sqlx::query("INSERT INTO ledgers (sequence, hash, close_time) VALUES (1, 'ledger', ?)") ++ .bind(CREATED) ++ .execute(&pool) ++ .await?; ++ for index in 0..ROWS { ++ sqlx::query( ++ "INSERT INTO anchors (id, name, stellar_account, reliability_score, created_at) \ ++ VALUES (?, ?, ?, ?, ?)", ++ ) ++ .bind(format!("anchor-{index}")) ++ .bind(format!("Anchor {index}")) ++ .bind(format!("account-{index}")) ++ .bind(50.0 + index as f64) ++ .bind(CREATED) ++ .execute(&pool) ++ .await?; ++ sqlx::query( ++ "INSERT INTO transactions (hash, ledger_sequence, source_account, successful, created_at) \ ++ VALUES (?, 1, ?, 1, ?)", ++ ) ++ .bind(format!("transaction-{index}")) ++ .bind(format!("account-{index}")) ++ .bind(CREATED) ++ .execute(&pool) ++ .await?; ++ } ++ let original_keys: Vec<(String, String, i64, i64)> = sqlx::query_as( ++ "SELECT resource, row_id, timestamp_ms, sequence FROM pagination_keys ORDER BY sequence", ++ ) ++ .fetch_all(&pool) ++ .await?; ++ let original_payloads: Vec = (0..ROWS) ++ .map(|index| json!({"id": format!("corridor-{index}"), "volume": index + 1})) ++ .collect(); ++ let snapshot = store_snapshot( ++ &pool, ++ SCOPE, ++ original_payloads ++ .iter() ++ .enumerate() ++ .map(|(index, payload)| { ++ ( ++ 1_767_323_045_000, ++ format!("corridor-{index}"), ++ payload.clone(), ++ ) ++ }) ++ .collect(), ++ ) ++ .await?; ++ ++ let anchor_first = list_anchors::(&pool, LIMIT, None).await?; ++ let transaction_first = list_ledger_transactions(&pool, LIMIT, None).await?; ++ let snapshot_first = snapshot_page::(&pool, SCOPE, LIMIT, None, Some(&snapshot)).await?; ++ let anchor_cursor = anchor_first ++ .pagination ++ .next_cursor ++ .clone() ++ .ok_or("expected an anchor continuation")?; ++ let transaction_cursor = transaction_first ++ .pagination ++ .next_cursor ++ .clone() ++ .ok_or("expected a transaction continuation")?; ++ let snapshot_cursor = snapshot_first ++ .pagination ++ .next_cursor ++ .clone() ++ .ok_or("expected a corridor continuation")?; ++ let anchor_before = list_anchors::(&pool, LIMIT, Some(&anchor_cursor)).await?; ++ let transaction_before = ++ list_ledger_transactions(&pool, LIMIT, Some(&transaction_cursor)).await?; ++ let snapshot_before = ++ snapshot_page::(&pool, SCOPE, LIMIT, Some(&snapshot_cursor), None).await?; ++ ++ // A separate writer commits between page one and its continuation. This ++ // controlled interleaving exercises concurrent traversal without timing sleeps. ++ let writer_pool = pool.clone(); ++ tokio::spawn(async move { ++ let mut write = writer_pool.begin().await?; ++ for (suffix, timestamp) in [("same", CREATED), ("backdated", "2020-01-02 03:04:05")] { ++ sqlx::query( ++ "INSERT INTO anchors (id, name, stellar_account, created_at) VALUES (?, ?, ?, ?)", ++ ) ++ .bind(format!("late-anchor-{suffix}")) ++ .bind(format!("Late {suffix}")) ++ .bind(format!("late-account-{suffix}")) ++ .bind(timestamp) ++ .execute(&mut *write) ++ .await?; ++ sqlx::query( ++ "INSERT INTO transactions (hash, ledger_sequence, created_at) VALUES (?, 1, ?)", ++ ) ++ .bind(format!("late-transaction-{suffix}")) ++ .bind(timestamp) ++ .execute(&mut *write) ++ .await?; ++ } ++ // Move one already returned row forward and one unseen row backward. ++ sqlx::query( ++ "UPDATE anchors SET created_at = CASE id WHEN 'anchor-0' \ ++ THEN '2099-01-01 00:00:00' ELSE '1900-01-01 00:00:00' END, \ ++ reliability_score = CASE id WHEN 'anchor-0' THEN 100 ELSE 0 END \ ++ WHERE id IN ('anchor-0', 'anchor-5')", ++ ) ++ .execute(&mut *write) ++ .await?; ++ sqlx::query( ++ "UPDATE transactions SET created_at = CASE hash WHEN 'transaction-0' \ ++ THEN '2099-01-01 00:00:00' ELSE '1900-01-01 00:00:00' END \ ++ WHERE hash IN ('transaction-0', 'transaction-5')", ++ ) ++ .execute(&mut *write) ++ .await?; ++ write.commit().await ++ }) ++ .await??; ++ ++ let keys_after: Vec<(String, String, i64, i64)> = sqlx::query_as( ++ "SELECT resource, row_id, timestamp_ms, sequence FROM pagination_keys \ ++ WHERE row_id NOT LIKE 'late-%' ORDER BY sequence", ++ ) ++ .fetch_all(&pool) ++ .await?; ++ assert_eq!( ++ keys_after, original_keys, ++ "updates must preserve both order keys" ++ ); ++ ++ let replacement = store_snapshot( ++ &pool, ++ SCOPE, ++ vec![ ++ ( ++ 1_767_323_045_000, ++ "corridor-1".to_string(), ++ json!({"id": "corridor-1", "volume": 999}), ++ ), ++ ( ++ 1_767_323_045_000, ++ "new-corridor".to_string(), ++ json!({"id": "new-corridor", "volume": 999}), ++ ), ++ ], ++ ) ++ .await?; ++ assert_ne!(replacement, snapshot); ++ let replacement_first = ++ snapshot_page::(&pool, SCOPE, LIMIT, None, Some(&replacement)).await?; ++ assert_eq!(replacement_first.pagination.total, 2); ++ assert_eq!(replacement_first.data[0]["volume"], json!(999)); ++ ++ let anchor_replay = list_anchors::(&pool, LIMIT, Some(&anchor_cursor)).await?; ++ assert_eq!( ++ anchor_replay ++ .data ++ .iter() ++ .map(|row| &row.id) ++ .collect::>(), ++ anchor_before ++ .data ++ .iter() ++ .map(|row| &row.id) ++ .collect::>() ++ ); ++ let transaction_replay = ++ list_ledger_transactions(&pool, LIMIT, Some(&transaction_cursor)).await?; ++ assert_eq!( ++ transaction_replay ++ .data ++ .iter() ++ .map(|row| &row.hash) ++ .collect::>(), ++ transaction_before ++ .data ++ .iter() ++ .map(|row| &row.hash) ++ .collect::>() ++ ); ++ let snapshot_replay = ++ snapshot_page::(&pool, SCOPE, LIMIT, Some(&snapshot_cursor), None).await?; ++ assert_eq!(snapshot_replay.data, snapshot_before.data); ++ assert_eq!( ++ anchor_replay.pagination.next_cursor, ++ anchor_before.pagination.next_cursor ++ ); ++ assert_eq!( ++ transaction_replay.pagination.next_cursor, ++ transaction_before.pagination.next_cursor ++ ); ++ assert_eq!( ++ snapshot_replay.pagination.next_cursor, ++ snapshot_before.pagination.next_cursor ++ ); ++ ++ let mut anchor_ids: Vec<_> = anchor_first.data.into_iter().map(|row| row.id).collect(); ++ let mut transaction_ids: Vec<_> = transaction_first ++ .data ++ .into_iter() ++ .map(|row| row.hash) ++ .collect(); ++ let mut payloads = snapshot_first.data; ++ let mut anchors_next = Some(anchor_cursor.clone()); ++ let mut transactions_next = Some(transaction_cursor.clone()); ++ let mut snapshots_next = Some(snapshot_cursor.clone()); ++ for _ in 0..ROWS { ++ if let Some(cursor) = anchors_next.take() { ++ let page = list_anchors::(&pool, LIMIT, Some(&cursor)).await?; ++ assert_eq!(page.pagination.total, ROWS); ++ assert_eq!( ++ page.pagination.has_next, ++ page.pagination.next_cursor.is_some() ++ ); ++ anchor_ids.extend(page.data.into_iter().map(|row| row.id)); ++ anchors_next = page.pagination.next_cursor; ++ } ++ if let Some(cursor) = transactions_next.take() { ++ let page = list_ledger_transactions(&pool, LIMIT, Some(&cursor)).await?; ++ assert_eq!(page.pagination.total, ROWS); ++ assert_eq!( ++ page.pagination.has_next, ++ page.pagination.next_cursor.is_some() ++ ); ++ transaction_ids.extend(page.data.into_iter().map(|row| row.hash)); ++ transactions_next = page.pagination.next_cursor; ++ } ++ if let Some(cursor) = snapshots_next.take() { ++ let page = snapshot_page::(&pool, SCOPE, LIMIT, Some(&cursor), None).await?; ++ assert_eq!(page.pagination.total, ROWS); ++ assert_eq!( ++ page.pagination.has_next, ++ page.pagination.next_cursor.is_some() ++ ); ++ payloads.extend(page.data); ++ snapshots_next = page.pagination.next_cursor; ++ } ++ } ++ assert!(anchors_next.is_none() && transactions_next.is_none() && snapshots_next.is_none()); ++ assert_eq!( ++ anchor_ids, ++ (0..ROWS) ++ .map(|index| format!("anchor-{index}")) ++ .collect::>() ++ ); ++ assert_eq!( ++ transaction_ids, ++ (0..ROWS) ++ .map(|index| format!("transaction-{index}")) ++ .collect::>() ++ ); ++ assert_eq!(payloads, original_payloads); ++ assert_eq!( ++ list_anchors::(&pool, 100, None) ++ .await? ++ .pagination ++ .total, ++ ROWS + 2 ++ ); ++ assert_eq!( ++ list_ledger_transactions(&pool, 100, None) ++ .await? ++ .pagination ++ .total, ++ ROWS + 2 ++ ); ++ ++ assert!(matches!( ++ list_anchors::(&pool, LIMIT, Some(&transaction_cursor)).await, ++ Err(PaginationError::InvalidCursor(_)) ++ )); ++ assert!(matches!( ++ list_ledger_transactions(&pool, LIMIT, Some("!!!")).await, ++ Err(PaginationError::InvalidCursor(_)) ++ )); ++ assert!(matches!( ++ snapshot_page::( ++ &pool, ++ "corridors:asset=XLM", ++ LIMIT, ++ Some(&snapshot_cursor), ++ None ++ ) ++ .await, ++ Err(PaginationError::InvalidCursor(_)) ++ )); ++ pool.close().await; ++ Ok(()) ++} From db0684bfd81bf945d69632e3b31abc3cca1a4f02 Mon Sep 17 00:00:00 2001 From: woahwhattheheck Date: Sun, 4 Oct 2026 08:29:11 -0400 Subject: [PATCH 6/6] docs: clarify the scope of backend pagination validation [skip ci] --- backend/integration/pagination/README.md | 5 +++-- backend/integration/pagination/manifest.json | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/backend/integration/pagination/README.md b/backend/integration/pagination/README.md index 32511ab6..09a4f98f 100644 --- a/backend/integration/pagination/README.md +++ b/backend/integration/pagination/README.md @@ -78,8 +78,9 @@ checks replay, original membership and retained corridor payloads. The run used a small manifest importing the unchanged production module files with `#[path]` and matching direct dependency versions. The test files use the -actual migrations. This was necessary because shared storage exhausted during -dependency compilation of the full backend crate. Whole-server compilation, +actual migrations. Full-backend dependency fetching hit shared storage +exhaustion; the focused compilation also hit that limit before the successful +two-test run. Whole-server compilation, HTTP composition and live-provider performance were not verified by that run. The maintained command for a backend environment with sufficient storage is: diff --git a/backend/integration/pagination/manifest.json b/backend/integration/pagination/manifest.json index 28d441bf..193790d4 100644 --- a/backend/integration/pagination/manifest.json +++ b/backend/integration/pagination/manifest.json @@ -121,7 +121,7 @@ "duration_seconds": 0.03 }, "execution_scope": "A small Rust manifest imported the exact production cursor_pagination.rs and pagination_queries.rs modules with #[path], using matching direct dependency versions including SQLx 0.8.6 and actual SQLite migrations. The maintained issue test files were executed without altered pagination logic.", - "whole_backend_build": "Not completed: shared storage exhausted during dependency compilation.", + "whole_backend_build": "Not executed: full-backend dependency fetching hit shared storage exhaustion. Focused module compilation also encountered the resource limit before the successful two-test run.", "http_runtime": "Not executed.", "patch_check": "Passed git apply --check against the complete clean backend checkout at 965e916227cf9173cf6ad254a6971429060c1d19; git diff --check also passed." }