diff --git a/.orc/orc-cycle.js b/.orc/orc-cycle.js new file mode 100644 index 0000000..9b1e430 --- /dev/null +++ b/.orc/orc-cycle.js @@ -0,0 +1,139 @@ +// ORC review cycle — fan out four reviewer lanes over the ORC diff, collect +// raw findings. Verification and fixes happen in the parent session. +const cwd = '/home/erubboli/Work/nullPointerEnjoyer/rust-sdk'; + +const brief = ` +REVIEW SCOPE: branch feat/indexer-v2-keyset-pagination, i.e. git diff f18563d..HEAD (commits 13e4beb, fb4b93e, a5e9a86, f101a7b) in ${cwd} +(run: git show 13e4beb --stat; git show 13e4beb -- to read the diff; +also read full files as needed). + +The commit adapts the Mintlayer Rust SDK (crate mintlayer-sdk) to the +api-server v2 changes of mintlayer-core PR #2130. The upstream wire contract +it must satisfy (verified against mintlayer-core @ 2684078c546, +api-server/web-server/src/api/{v2.rs,cursor.rs,error.rs}): + +- Cursor envelope (paged_response): {"items":[...],"next_cursor":""|null}. + Endpoints that branch on cursor-param presence (GET /pool, GET /transaction) + return a PLAIN ARRAY when no cursor param is sent; an EMPTY "cursor=" query + parameter still counts as present and forces the envelope (and means "start + from beginning"). Holders and order-book endpoints ALWAYS return the envelope. +- GET /statistics/coin/holders and /statistics/token/{id}/holders: items are + {"address":"","amount":{atoms,decimal}} with the coin's or token's + decimals; ordered balance DESC; unknown token -> 404 {"error":"Token not found"}; + offset params still work when no cursor is sent; items cap 100, items=0 -> 400. +- GET /order/pair/{pair}/book: side=ask|bid REQUIRED (missing/other -> 400 + {"error":"Bad request"}); pair is {base}_{quote}, coin ticker matched + case-insensitively, token ids exact case-sensitive bech32, preserved as given. + Items: {"price":{"atoms":"numer/denom","decimal":""}, + "amount":{atoms,decimal}}. Ask ascending price, bid descending. Cursors are + side-specific (tags book-ask / book-bid): an ask cursor on a bid walk -> 400 + {"error":"Invalid cursor"}. When the per-request cap (10,000 orders) truncates + the aggregation the response adds "truncated":true AND next_cursor is null — + the walk cannot be continued. Book computed fresh per request (no snapshot). +- GET /pool: sort=by_height|by_pledge default by_height; a cursor is only valid + with by_height (cursor + other sort -> 400 {"error":"Bad request"}); with a + cursor the offset position is silently overridden server-side (items applies). +- GET /transaction: cursor+offset_mode together -> 400 Bad request; + offset_mode=legacy|absolute (default legacy; invalid -> 400 invalid offset + mode); offset path treats absolute offset as a global tx index; listing is + newest block first. transaction/{id} for a pending tx emits JSON null for + block_id/timestamp/confirmations (listing items for pending likewise null). +- ALL paginated v2 endpoints: offset u64 default 0, items u32 default 10, + items=0 and items>100 -> 400 {"error":"Invalid number of items"}. +- Error body shape: {"error":""}; exact messages include + "Bad request", "Invalid cursor", "Invalid number of items", "Invalid offset + mode", "Invalid pools sort order", "Token not found", "Transaction not found". + +SDK conventions in this repo (pre-existing, must be respected): one client +method per endpoint on indexer::Client; pub(crate) get(path, &[(&str,String)]) +query builder; page_query omits zero offset/items; validate_segment restricts +path segments to [A-Za-z0-9_-]; errors via thiserror enum indexer::Error; +NO new dependencies (Pager is hand-rolled on Pin>). +`; + +const outputInstruction = ` +CRITICAL OUTPUT RULE: your ENTIRE final message must be exactly one JSON +array: it must start with '[' and end with ']' and contain nothing else — +no headings, no prose sections, no fences. If you find no defects, your +entire message is exactly []. Do NOT write a human-readable review; the +parent parses your message with JSON.parse. +Each finding: +{"severity":"blocking"|"advisory","file":"","claim":"", + "detail":"", + "fix":""} +Report ONLY real defects: contract violations vs the upstream facts above, +broken invariants, logic bugs, doc/test claims that contradict the code, +convention violations. Style nitpicks are NOT findings. If everything is +correct return []. Findings you are not sure about must be marked advisory, +not blocking. +`; + +const lanes = [ + { + key: 'wire-contract', + task: `${brief}\n${outputInstruction} +DIMENSION: wire-contract fidelity. Verify every new/changed SDK request shape +and response type against the upstream contract above: query parameters and +their omission rules, the cursor= envelope forcing for pools/transactions, the +always-envelope holders/book endpoints, the order-book item shape and pair +handling, Transaction optionality, holders ordering/decimals, statistics docs. +Files: src/indexer/types.rs, src/indexer/statistics.rs, src/indexer/pool.rs, +src/indexer/transaction.rs, src/indexer/order.rs, src/indexer/mod.rs.`, + }, + { + key: 'async-pager', + task: `${brief}\n${outputInstruction} +DIMENSION: async/state-machine correctness of the pagination layer. Scrutinize +src/indexer/pager.rs and every *_pager constructor (statistics.rs, pool.rs, +transaction.rs, order.rs): FnMut closure capture/move semantics, Send bounds, +cursor propagation and take() logic, done/exhaustion conditions, buffered-item +semantics when mixing next()/next_page(), retry-after-error behavior, truncated +book termination, page-size clamping, per-call clones, accidental unbounded +loops on empty pages.`, + }, + { + key: 'errors-serde', + task: `${brief}\n${outputInstruction} +DIMENSION: error mapping and (de)serialization. Scrutinize src/indexer/error.rs +(from_status_body matching, fallback correctness, sanitization interaction), +src/indexer/mod.rs decode() change, and serde derives in src/indexer/types.rs +(Page, Holder, OrderBook* with #[serde(default)] truncated, Amount usage). +Check: a 400 body that is not JSON or has a different message must still fall +back to Error::Http; the typed variants trigger only on exact upstream +messages; no breaking change to existing variant semantics; clamp_items +boundary conditions (0, 1, 100, 101, u32::MAX).`, + }, + { + key: 'tests-docs', + task: `${brief}\n${outputInstruction} +DIMENSION: tests and documentation coherence. Scrutinize the new tests in +tests/indexer.rs (bottom section) and docs in README.md, docs/indexer.md, +examples/indexer-pagination.rs plus their doc comments in src/. Check: each +required scenario is actually pinned (first page/full walk for pools, holders, +global transactions; both book sides; truncated book stops; last page null; +invalid cursor / invalid num items / token-not-found; pending tx null fields; +offset_mode param; page-size clamp), mocks assert the right query params, the +documented claims match the code exactly (no overclaiming), and the example +compiles against the real API surface.`, + }, +]; + +const results = await runs.all(lanes.map(lane => ({ key: lane.key, agent: 'reviewer', task: lane.task }))); + +const findings = {}; +for (let i = 0; i < lanes.length; i++) { + const lane = lanes[i]; + const text = results[i] && results[i].output ? String(results[i].output) : ''; + const start = text.indexOf('['); + const end = text.lastIndexOf(']'); + let parsed = null; + if (start !== -1 && end > start) { + try { parsed = JSON.parse(text.slice(start, end + 1)); } catch (error) { parsed = null; } + } + findings[lane.key] = parsed === null + ? [{ severity: 'blocking', file: '(unparsed)', claim: `lane ${lane.key} did not return JSON`, detail: text.slice(0, 2000), fix: 're-run lane' }] + : parsed; +} +const total = Object.values(findings).reduce((n, list) => n + list.length, 0); +console.log(`ORC total raw findings: ${total}`); +return { findings, total }; diff --git a/.orc/orc-rerun.js b/.orc/orc-rerun.js new file mode 100644 index 0000000..f874b2b --- /dev/null +++ b/.orc/orc-rerun.js @@ -0,0 +1,118 @@ +// ORC review cycle — fan out four reviewer lanes over the ORC diff, collect +// raw findings. Verification and fixes happen in the parent session. +const cwd = '/home/erubboli/Work/nullPointerEnjoyer/rust-sdk'; + +const brief = ` +REVIEW SCOPE: branch feat/indexer-v2-keyset-pagination, i.e. git diff f18563d..HEAD (commits 13e4beb, fb4b93e, a5e9a86, f101a7b) in ${cwd} +(run: git show 13e4beb --stat; git show 13e4beb -- to read the diff; +also read full files as needed). + +The commit adapts the Mintlayer Rust SDK (crate mintlayer-sdk) to the +api-server v2 changes of mintlayer-core PR #2130. The upstream wire contract +it must satisfy (verified against mintlayer-core @ 2684078c546, +api-server/web-server/src/api/{v2.rs,cursor.rs,error.rs}): + +- Cursor envelope (paged_response): {"items":[...],"next_cursor":""|null}. + Endpoints that branch on cursor-param presence (GET /pool, GET /transaction) + return a PLAIN ARRAY when no cursor param is sent; an EMPTY "cursor=" query + parameter still counts as present and forces the envelope (and means "start + from beginning"). Holders and order-book endpoints ALWAYS return the envelope. +- GET /statistics/coin/holders and /statistics/token/{id}/holders: items are + {"address":"","amount":{atoms,decimal}} with the coin's or token's + decimals; ordered balance DESC; unknown token -> 404 {"error":"Token not found"}; + offset params still work when no cursor is sent; items cap 100, items=0 -> 400. +- GET /order/pair/{pair}/book: side=ask|bid REQUIRED (missing/other -> 400 + {"error":"Bad request"}); pair is {base}_{quote}, coin ticker matched + case-insensitively, token ids exact case-sensitive bech32, preserved as given. + Items: {"price":{"atoms":"numer/denom","decimal":""}, + "amount":{atoms,decimal}}. Ask ascending price, bid descending. Cursors are + side-specific (tags book-ask / book-bid): an ask cursor on a bid walk -> 400 + {"error":"Invalid cursor"}. When the per-request cap (10,000 orders) truncates + the aggregation the response adds "truncated":true AND next_cursor is null — + the walk cannot be continued. Book computed fresh per request (no snapshot). +- GET /pool: sort=by_height|by_pledge default by_height; a cursor is only valid + with by_height (cursor + other sort -> 400 {"error":"Bad request"}); with a + cursor the offset position is silently overridden server-side (items applies). +- GET /transaction: cursor+offset_mode together -> 400 Bad request; + offset_mode=legacy|absolute (default legacy; invalid -> 400 invalid offset + mode); offset path treats absolute offset as a global tx index; listing is + newest block first. transaction/{id} for a pending tx emits JSON null for + block_id/timestamp/confirmations (listing items for pending likewise null). +- ALL paginated v2 endpoints: offset u64 default 0, items u32 default 10, + items=0 and items>100 -> 400 {"error":"Invalid number of items"}. +- Error body shape: {"error":""}; exact messages include + "Bad request", "Invalid cursor", "Invalid number of items", "Invalid offset + mode", "Invalid pools sort order", "Token not found", "Transaction not found". + +SDK conventions in this repo (pre-existing, must be respected): one client +method per endpoint on indexer::Client; pub(crate) get(path, &[(&str,String)]) +query builder; page_query omits zero offset/items; validate_segment restricts +path segments to [A-Za-z0-9_-]; errors via thiserror enum indexer::Error; +NO new dependencies (Pager is hand-rolled on Pin>). +`; + +const outputInstruction = ` +CRITICAL OUTPUT RULE: your ENTIRE final message must be exactly one JSON +array: it must start with '[' and end with ']' and contain nothing else — +no headings, no prose sections, no fences. If you find no defects, your +entire message is exactly []. Do NOT write a human-readable review; the +parent parses your message with JSON.parse. +Each finding: +{"severity":"blocking"|"advisory","file":"","claim":"", + "detail":"", + "fix":""} +Report ONLY real defects: contract violations vs the upstream facts above, +broken invariants, logic bugs, doc/test claims that contradict the code, +convention violations. Style nitpicks are NOT findings. If everything is +correct return []. Findings you are not sure about must be marked advisory, +not blocking. +`; + +const lanes = [ + { + key: 'errors-serde', + task: `${brief}\n${outputInstruction} +DIMENSION: error mapping and (de)serialization. Scrutinize src/indexer/error.rs +(from_status_body matching, fallback correctness, sanitization interaction), +src/indexer/mod.rs decode() change, and serde derives in src/indexer/types.rs +(Page, Holder, OrderBook* with #[serde(default)] truncated, Amount usage). +Check: a 400 body that is not JSON or has a different message must still fall +back to Error::Http; the typed variants trigger only on exact upstream +messages; no breaking change to existing variant semantics; clamp_items +boundary conditions (0, 1, 100, 101, u32::MAX).`, + }, + { + key: 'tests-docs', + task: `${brief}\n${outputInstruction} +DIMENSION: tests and documentation coherence. Scrutinize the new tests in +tests/indexer.rs (bottom section) and docs in README.md, docs/indexer.md, +examples/indexer-pagination.rs plus their doc comments in src/. Check: each +required scenario is actually pinned (first page/full walk for pools, holders, +global transactions; both book sides; truncated book stops; last page null; +invalid cursor / invalid num items / token-not-found; pending tx null fields; +offset_mode param; page-size clamp; retry-resume after a failed fetch; +start_from rewind of an exhausted pager), mocks assert the right query params, +the documented claims match the code exactly (no overclaiming), and the +example compiles against the real API surface.`, + }, +]; + +const results = await runs.all(lanes.map(lane => ({ key: lane.key, agent: 'reviewer', task: lane.task }))); + +const findings = {}; +for (let i = 0; i < lanes.length; i++) { + const lane = lanes[i]; + const text = results[i] && results[i].output ? String(results[i].output) : ''; + const start = text.indexOf('['); + const end = text.lastIndexOf(']'); + let parsed = null; + if (start !== -1 && end > start) { + try { parsed = JSON.parse(text.slice(start, end + 1)); } catch (error) { parsed = null; } + } + findings[lane.key] = parsed === null + ? [{ severity: 'blocking', file: '(unparsed)', claim: `lane ${lane.key} did not return JSON`, detail: text.slice(0, 2000), fix: 're-run lane' }] + : parsed; +} +const total = Object.values(findings).reduce((n, list) => n + list.length, 0); +console.log(`ORC total raw findings: ${total}`); +return { findings, total }; diff --git a/CHANGELOG.md b/CHANGELOG.md index ea70186..9e51e02 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,21 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ### Added +- indexer: cursor (keyset) pagination for api-server v2 + (mintlayer-core PR #2130): `Page`, the reusable `Pager` walker + (`*_pager` constructors for pools, the global transaction listing, both + holders listings and the order book), the `*_paged`/`*_pager` listing + methods with explicit cursors, `list_transactions_with_offset_mode` + (`OffsetMode::Legacy` / `Absolute`), the holders listings + (`coin_holders`, `token_holders`, `Holder`) and the order book + (`order_pair_book`, `OrderBook`/`OrderBookLevel`/`OrderBookPrice` with + the `truncated` invariant and side-specific cursors). +- indexer: typed server errors `Error::InvalidCursor`, + `Error::InvalidNumItems`, `Error::BadRequest` and `Error::TokenNotFound`, + mapped from the api-server error bodies (`{"error": ...}`). +- `examples/indexer-pagination.rs`: runnable walks for holders, the order + book, the global transaction listing and pools. + - `node::Error::OutcomeUnknown` / `wallet::Error::OutcomeUnknown`: the request was delivered but its outcome could not be confirmed (lost, oversized, undecodable or mismatched response, or a timeout while @@ -18,6 +33,10 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ### Changed +- **Breaking:** indexer `Transaction.block_id`, `Transaction.timestamp` + and `Transaction.confirmations` are now `Option<...>`: a pending + (mempool) transaction has `null` for all three on the wire, which + previously failed to decode; update direct field uses. - **Breaking:** post-delivery transport failures surface as `OutcomeUnknown(...)` instead of the bare variant, so a retry on `Err` can no longer silently double a fund-moving mutation; update matches on diff --git a/Cargo.toml b/Cargo.toml index d03eb6e..231734f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -73,6 +73,10 @@ required-features = ["crypto", "indexer"] name = "issue-token" required-features = ["wallet"] +[[example]] +name = "indexer-pagination" +required-features = ["indexer"] + [[test]] name = "client" required-features = ["node", "indexer", "wallet", "crypto"] diff --git a/README.md b/README.md index 77be93c..757137e 100644 --- a/README.md +++ b/README.md @@ -134,7 +134,7 @@ REST client for `api-web-server`. All paths are relative to `/api/v2/`. The indexer API is unauthenticated. ```rust -use mintlayer_sdk::indexer::{self, PageOpts, PoolListOpts, PoolSort}; +use mintlayer_sdk::indexer::{self, OrderBookOpts, OrderBookSide, PageOpts, PoolListOpts, PoolSort}; let c = indexer::Client::builder("http://127.0.0.1:3000") .timeout(std::time::Duration::from_secs(15)) @@ -159,14 +159,33 @@ let tokens = c.tokens_by_ticker("MYTOKEN", PageOpts { items: 10, ..Default::defa // Orders let orders = c.list_orders(PageOpts::default()).await?; -// Statistics +// Order book (api-server v2): aggregated price levels, side required. +// Cursors are side-specific and a truncated book stops the walk. +let book = c.order_pair_book("ML", "mmltk1...", OrderBookSide::Ask, OrderBookOpts::default()).await?; +for level in &book.items { + println!("{} @ {}", level.amount.decimal, level.price.decimal); +} + +// Statistics — all four supply counters are always present. let stats = c.coin_statistics().await?; +// Cursor pagination: pools, global transactions, holders, order book. +// The pager follows next_cursor and stops on the last page. +let mut holders = c.coin_holders_pager(100); +while let Some(holder) = holders.next().await { + let holder = holder?; + println!("{}: {}", holder.address, holder.amount.decimal); +} + // Submit a signed transaction (requires --enable-post-routes) let tx_id = c.submit_transaction(&signed_tx_hex).await?; ``` -Non-2xx responses are returned as `indexer::Error::Http { status_code, body }`. +Non-2xx responses map to typed errors where the server's message is known +(`Error::InvalidCursor`, `Error::InvalidNumItems`, `Error::BadRequest`, +`Error::TokenNotFound`) and fall back to +`indexer::Error::Http { status_code, body }` otherwise. See +[docs/indexer.md](docs/indexer.md) for the full cursor-pagination rules. ## Wallet client (`wallet`) @@ -293,6 +312,7 @@ to reach into `crypto`. |---|---| | [examples/send-coins](examples/send-coins.rs) | Derive key → fetch UTXOs → build, sign and submit a transaction | | [examples/issue-token](examples/issue-token.rs) | Issue a fungible token and mint an initial supply via the wallet daemon | +| [examples/indexer-pagination](examples/indexer-pagination.rs) | Cursor-paginated walks: coin holders, order book (both sides), transactions (cursor and offset_mode), pools | ```text cargo run --example send-coins --features crypto,indexer -- --mnemonic "..." --to mtc1q... --amount 100000000000 --network testnet diff --git a/docs/indexer.md b/docs/indexer.md index 8b0a6d3..5662217 100644 --- a/docs/indexer.md +++ b/docs/indexer.md @@ -22,8 +22,18 @@ The indexer API is unauthenticated; there are no credential options. Errors are returned as `indexer::Error`: -- `Error::Http { status_code, body }` — any non-2xx response; the body is - trimmed and stripped of control characters. +- `Error::Http { status_code, body }` — any non-2xx response that does not + match a known server error below; the body is trimmed and stripped of + control characters. +- `Error::InvalidCursor` — `400 "Invalid cursor"`: the cursor is malformed + or oversized, belongs to a different listing, or was minted for the other + side of an order book. +- `Error::InvalidNumItems` — `400 "Invalid number of items"`: every + paginated v2 endpoint accepts 1..=100 items. +- `Error::BadRequest` — `400 "Bad request"`: mutually incompatible query + parameters (a cursor together with `offset_mode`, or a cursor with a + non-default pools sort). +- `Error::TokenNotFound` — `404 "Token not found"`. - `Error::Transport(reqwest::Error)` — the request failed. - `Error::Json(serde_json::Error)` — the response could not be decoded. - `Error::InvalidUrl { message }` — a path segment contained characters @@ -53,6 +63,66 @@ let opts = PageOpts { offset: 20, items: 10 }; `PoolListOpts` extends the same pagination with a `sort` field (`PoolSort::ByHeight`, the server default, or `PoolSort::ByPledge`). +### Cursor (keyset) pagination + +The pools listing, the global transaction listing, both holders listings, +and the order book also support keyset pagination: with a `cursor` in the +request the response is the envelope +`Page { items, next_cursor }`, where `next_cursor` is the opaque cursor +of the last returned item and is `None` on the last page. Without a cursor +parameter the pools and transaction listings return a plain array (the +offset methods), while the holders listings and the order book always +return the envelope. +Cursors are minted by the indexer — never construct or mutate one; pass it +back verbatim. A cursor silently overrides the `offset` page position on +the server (`items` still applies). + +```rust +use mintlayer_sdk::indexer::Client; + +let c = Client::new("http://127.0.0.1:3000"); + +// Page by page: +let mut cursor = None; +loop { + let page = c.list_pools_paged(cursor.as_deref(), 100).await?; + for pool in &page.items { /* ... */ } + match page.next_cursor { + Some(next) => cursor = Some(next), + None => break, + } +} + +// Or item by item with the shared pager: +let mut pools = c.pools_pager(100); +while let Some(pool) = pools.next().await { + let pool = pool?; +} +``` + +Per-endpoint rules (the server enforces them; the errors are typed): + +- **Pools** — the cursor walk only exists for the default `by_height` + sort; a cursor combined with any other sort is rejected with + `Error::BadRequest`. For `by_pledge` use `list_pools`. +- **Transactions** — a cursor cannot be combined with `offset_mode` + (`Error::BadRequest`). For offset-based listings use `list_transactions` + or `list_transactions_with_offset_mode` (`OffsetMode::Legacy`, the + server default, or `OffsetMode::Absolute`, which treats the offset as a + stable global transaction index). +- **Order book** — cursors are side-specific: an ask-side cursor on a bid + walk is rejected with `Error::InvalidCursor`. When the per-request cap + of 10,000 aggregated orders truncates the book, the response carries + `truncated: true` and `next_cursor: null` — the walk cannot be + continued, and the `*_pager` stops there (see `OrderBook`). + +Semantics worth knowing on every cursor walk: pages are only guaranteed +stable once the indexer's scanner is fully caught up — during catch-up or +a reorg a walk may skip or repeat an entry — and the order book is +computed fresh on every request, so a walk is not a consistent snapshot of +a moving book. Offset-based methods remain the simple alternative for +shallow listings. + --- ## Chain @@ -87,20 +157,26 @@ let opts = PageOpts { offset: 20, items: 10 }; | Method | REST path | Notes | |--------|-----------|-------| | `list_transactions(opts: PageOpts) -> Result, Error>` | `GET /transaction` | Newest first | -| `transaction(id: &str) -> Result` | `GET /transaction/{id}` | See the unconfirmed quirk below | +| `list_transactions_paged(cursor, items) -> Result, Error>` | `GET /transaction` | Cursor walk; `None` starts from the beginning | +| `list_transactions_with_offset_mode(mode, opts) -> Result, Error>` | `GET /transaction` | Explicit offset semantics, no cursor | +| `transactions_pager(items) -> Pager` | `GET /transaction` | Item-wise cursor walk | +| `transaction(id: &str) -> Result` | `GET /transaction/{id}` | Pending transactions carry `None` fields, see below | | `transaction_merkle_path(id: &str) -> Result` | `GET /transaction/{id}/merkle-path` | 404 before confirmation | | `transaction_output(tx_id: &str, index: u32) -> Result` | `GET /transaction/{tx_id}/output/{index}` | Raw JSON | | `submit_transaction(signed_tx_hex: &str) -> Result` | `POST /transaction` | Requires `--enable-post-routes`; returns the tx id | -**Quirk:** for unconfirmed transactions the `Transaction` fields -`block_id`, `timestamp` and `confirmations` are empty strings, not numbers -or nulls. +**Quirk:** for pending (mempool) transactions the `Transaction` fields +`block_id`, `timestamp` and `confirmations` are `None`; on the global +listing a pending transaction has `block_id: None` while confirmed ones +carry the block hash. ## Pools and delegations | Method | REST path | Notes | |--------|-----------|-------| | `list_pools(opts: PoolListOpts) -> Result, Error>` | `GET /pool` | `sort=by_height` (default) or `by_pledge` | +| `list_pools_paged(cursor, items) -> Result, Error>` | `GET /pool` | Cursor walk (creation-height sort only) | +| `pools_pager(items) -> Pager` | `GET /pool` | Item-wise cursor walk | | `pool(id: &str) -> Result` | `GET /pool/{id}` | | | `pool_block_stats(id: &str, from: i64, to: i64) -> Result` | `GET /pool/{id}/block-stats` | Block count in `[from, to)`, UNIX seconds | | `pool_delegations(id: &str) -> Result, Error>` | `GET /pool/{id}/delegations` | Includes `creation_block_height` | @@ -123,13 +199,18 @@ or nulls. | `list_orders(opts: PageOpts) -> Result, Error>` | `GET /order` | | | `order(id: &str) -> Result` | `GET /order/{id}` | | | `orders_by_pair(ask, give, opts) -> Result, Error>` | `GET /order/pair/{ask}_{give}` | Currencies are the coin ticker (e.g. `ML`) or a bech32 token id | +| `order_pair_book(base, quote, side, opts) -> Result` | `GET /order/pair/{base}_{quote}/book` | Aggregated price levels; `side` is required; see `OrderBook` invariants | +| `order_book_pager(base, quote, side, items) -> Result, Error>` | | Item-wise cursor walk; stops on a truncated book | ## Statistics and fees | Method | REST path | Notes | |--------|-----------|-------| -| `coin_statistics() -> Result` | `GET /statistics/coin` | Circulating, preminted, burned, staked | -| `token_statistics(token_id: &str) -> Result` | `GET /statistics/token/{token_id}` | | +| `coin_statistics() -> Result` | `GET /statistics/coin` | Circulating, preminted, burned, staked — all four counters are always present; unwritten counters are zero | +| `token_statistics(token_id: &str) -> Result` | `GET /statistics/token/{token_id}` | Same counters; 404 `Error::TokenNotFound` for unknown tokens | +| `coin_holders(opts: HoldersOpts) -> Result, Error>` | `GET /statistics/coin/holders` | Largest balance first; amounts in coin decimals | +| `token_holders(token_id, opts) -> Result, Error>` | `GET /statistics/token/{token_id}/holders` | Amounts in token decimals | +| `coin_holders_pager(items) / token_holders_pager(token_id, items)` | | Item-wise cursor walks | | `fee_rate(in_top_x_mb: u32) -> Result` | `GET /feerate` | Atoms per KB; `0` uses the server default of 5 MB | --- diff --git a/examples/indexer-pagination.rs b/examples/indexer-pagination.rs new file mode 100644 index 0000000..2296865 --- /dev/null +++ b/examples/indexer-pagination.rs @@ -0,0 +1,137 @@ +// Copyright (c) 2026 Mintlayer Institutional FZCO +// Contact: hello@mintlayer.org +// +// Use of this source code is governed by an MIT license +// that can be found in the LICENSE file. + +//! Cursor (keyset) pagination against the Mintlayer indexer +//! (api-server v2). Demonstrates cursor walks over the four paginated +//! listings (each bounded to the first few items for brevity): +//! +//! 1. the native coin holders, +//! 2. both sides of an order book, +//! 3. the global transaction listing (with cursor and with +//! `offset_mode`), +//! 4. the stake pools (creation-height cursor walk). +//! +//! Usage: +//! ```text +//! cargo run --example indexer-pagination -- http://127.0.0.1:3000 +//! ``` +//! +//! Point it at a local indexer (`indexer --chain-config mainnet` or a +//! testnet instance on port 13000). Requires no wallet or node. + +use mintlayer_sdk::indexer::{Client, OffsetMode, OrderBookOpts, OrderBookSide, PageOpts}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let mut args = std::env::args().skip(1); + let base_url = args.next().unwrap_or_else(|| "http://127.0.0.1:3000".to_owned()); + let indexer = Client::new(&base_url); + // The native coin ticker, e.g. "ML" (mainnet) or "TMCL" (testnet); the + // server matches the coin ticker case-insensitively while token ids in + // a pair must be exact bech32 strings. + let coin = std::env::var("COIN").unwrap_or_else(|_| "ML".to_owned()); + + // 1. Coin holders: largest balance first. `next()` yields items across + // pages and stops once the server sends `next_cursor: null`. + println!("== {coin} holders =="); + let mut holders = indexer.coin_holders_pager(50); + let mut count = 0u64; + while let Some(holder) = holders.next().await { + let holder = holder?; + println!( + "{:>3}. {} — {}", + count + 1, + holder.address, + holder.amount.decimal + ); + count += 1; + if count == 10 { + println!( + " ... stopping after 10 (cursor so far: {:?})", + holders.cursor() + ); + break; + } + } + + // 2. Order book, both sides. Levels are aggregated remaining balances; + // a truncated book (10,000-order server cap) ends the walk early. + for side in [OrderBookSide::Ask, OrderBookSide::Bid] { + println!("== order book {coin}/USDT, side {side:?} =="); + match indexer.order_book_pager(&coin, "USDT", side, 20) { + Ok(mut book) => { + let mut levels = 0u64; + while let Some(level) = book.next().await { + let level = level?; + println!( + "price {} amount {}", + level.price.decimal, level.amount.decimal + ); + levels += 1; + if levels == 10 { + break; + } + } + if levels == 0 { + println!("(empty book)"); + } + } + Err(error) => println!("book unavailable: {error}"), + } + } + + // 3. Global transaction listing: cursor walk first, then the + // offset-based listing with explicit `offset_mode`. + println!("== latest transactions (cursor walk) =="); + let mut transactions = indexer.transactions_pager(10); + for _ in 0..5 { + match transactions.next().await { + Some(Ok(tx)) => println!("tx {} in {:?} at {:?}", tx.id, tx.block_id, tx.timestamp), + Some(Err(error)) => return Err(error.into()), + None => break, + } + } + + println!("== latest transactions (offset_mode = absolute) =="); + for tx in indexer + .list_transactions_with_offset_mode( + OffsetMode::Absolute, + PageOpts { + offset: 0, + items: 5, + }, + ) + .await? + { + println!("tx {} in {:?}", tx.id, tx.block_id); + } + + // 4. Stake pools: the cursor walk only exists for the default + // creation-height sort (a cursor with `by_pledge` is rejected by + // the server). For a pledge-sorted listing use `list_pools`. + println!("== pools (creation-height walk) =="); + let mut pools = indexer.pools_pager(50); + for _ in 0..5 { + match pools.next().await { + Some(Ok(pool)) => println!( + "pool {} staked {}", + pool.pool_id, pool.staker_balance.decimal + ), + Some(Err(error)) => return Err(error.into()), + None => break, + } + } + + // One-off paged access (without the pager) is equally available: + let page = indexer.list_pools_paged(None, 10).await?; + println!( + "first pool page: {} pools, next_cursor {:?}", + page.items.len(), + page.next_cursor + ); + let _ = OrderBookOpts::default(); // typed request options also exist + Ok(()) +} diff --git a/src/indexer/error.rs b/src/indexer/error.rs index 66ba9bf..24ef30d 100644 --- a/src/indexer/error.rs +++ b/src/indexer/error.rs @@ -35,4 +35,48 @@ pub enum Error { /// The limit that was exceeded. limit: usize, }, + /// The indexer rejected a cursor (`400`, body `Invalid cursor`): the + /// cursor is malformed or oversized, belongs to a different listing, + /// or was minted for the other side of an order book. + #[error("invalid cursor")] + InvalidCursor, + /// The indexer rejected the requested page size (`400`, body `Invalid + /// number of items`); every paginated v2 endpoint accepts 1..=100 + /// items. + #[error("invalid number of items (must be 1..=100)")] + InvalidNumItems, + /// The indexer rejected the query (`400`, body `Bad request`), for + /// example mutually incompatible parameters (`cursor` together with + /// `offset_mode`, or a cursor with a non-default pools sort). + #[error("bad request (incompatible query parameters)")] + BadRequest, + /// The referenced token does not exist (`404`, body `Token not + /// found`). + #[error("token not found")] + TokenNotFound, +} + +impl Error { + /// Maps an indexer error response to a typed error when the + /// `{ "error": "" }` body matches a known server message; + /// falls back to [`Error::Http`] otherwise. + pub(crate) fn from_status_body(status_code: u16, body: &str) -> Error { + #[derive(serde::Deserialize)] + struct ErrorBody { + error: String, + } + if let Ok(parsed) = serde_json::from_str::(body) { + match (status_code, parsed.error.as_str()) { + (400, "Invalid cursor") => return Error::InvalidCursor, + (400, "Invalid number of items") => return Error::InvalidNumItems, + (400, "Bad request") => return Error::BadRequest, + (404, "Token not found") => return Error::TokenNotFound, + _ => {} + } + } + Error::Http { + status_code, + body: body.to_owned(), + } + } } diff --git a/src/indexer/mod.rs b/src/indexer/mod.rs index 27e5cd4..d147176 100644 --- a/src/indexer/mod.rs +++ b/src/indexer/mod.rs @@ -13,6 +13,7 @@ mod chain; mod delegation; mod error; mod order; +mod pager; mod pool; mod statistics; mod token; @@ -24,11 +25,13 @@ use std::time::Duration; use serde::de::DeserializeOwned; pub use error::Error; +pub use pager::{PagedFuture, Pager}; pub use types::{ AddressInfo, Amount, Block, BlockBody, BlockHeader, ChainTip, CoinStats, Delegation, - DelegationInfo, GenesisInfo, MerklePath, NftInfo, NftMetadata, Order, PageOpts, PerThousand, - Pool, PoolDelegation, PoolListOpts, PoolSort, Timestamp, TokenBalance, TokenInfo, TokenTx, - Transaction, Uint64, Utxo, UtxoOutpoint, + DelegationInfo, GenesisInfo, Holder, HoldersOpts, MerklePath, NftInfo, NftMetadata, OffsetMode, + Order, OrderBook, OrderBookLevel, OrderBookOpts, OrderBookPrice, OrderBookSide, Page, PageOpts, + PerThousand, Pool, PoolDelegation, PoolListOpts, PoolSort, Timestamp, TokenBalance, TokenInfo, + TokenTx, Transaction, Uint64, Utxo, UtxoOutpoint, }; use crate::limits::DEFAULT_TIMEOUT; @@ -104,10 +107,7 @@ impl Client { if status.is_client_error() || status.is_server_error() { let bytes = Self::read_capped(response).await.unwrap_or_default(); let body = crate::limits::sanitize_daemon_text(&String::from_utf8_lossy(&bytes)); - return Err(Error::Http { - status_code: status.as_u16(), - body, - }); + return Err(Error::from_status_body(status.as_u16(), &body)); } let bytes = Self::read_capped(response).await?; Ok(serde_json::from_slice(&bytes)?) @@ -119,7 +119,7 @@ impl Client { } /// Builds the pagination query parameters; zero values are omitted. -pub(crate) fn page_query(offset: u32, items: u32) -> Vec<(&'static str, String)> { +pub(crate) fn page_query(offset: u64, items: u32) -> Vec<(&'static str, String)> { let mut query = Vec::new(); if offset > 0 { query.push(("offset", offset.to_string())); @@ -130,6 +130,18 @@ pub(crate) fn page_query(offset: u32, items: u32) -> Vec<(&'static str, String)> query } +/// The largest page size every paginated v2 endpoint accepts (server-side +/// `MAX_NUM_ITEMS`; larger or zero page sizes are rejected with `400` +/// `"Invalid number of items"`). +pub(crate) const MAX_NUM_ITEMS: u32 = 100; + +/// Clamps a caller-supplied page size to the server-accepted 1..=[`MAX_NUM_ITEMS`] +/// range; used by the pager constructors, whose walks must not fail on a +/// page boundary. +pub(crate) fn clamp_items(items: u32) -> u32 { + items.clamp(1, MAX_NUM_ITEMS) +} + /// Validates that a caller-supplied path segment cannot alter the request /// path (ids and addresses are bech32 or hex; tickers are alphanumeric). pub(crate) fn validate_segment(segment: &str) -> Result<&str, Error> { diff --git a/src/indexer/order.rs b/src/indexer/order.rs index d475f2c..18f88b4 100644 --- a/src/indexer/order.rs +++ b/src/indexer/order.rs @@ -6,7 +6,9 @@ //! Indexer DEX order endpoints. -use super::{Error, Order, PageOpts}; +use super::{ + Error, Order, OrderBook, OrderBookLevel, OrderBookOpts, OrderBookSide, Page, PageOpts, Pager, +}; use crate::indexer::Client; impl Client { @@ -41,4 +43,84 @@ impl Client { ) .await } + + /// Returns one page of the order book of a trading pair + /// (`GET /order/pair/{base}_{quote}/book`). + /// + /// The native coin ticker in the pair matches case-insensitively on + /// the server, while token ids are exact bech32 strings — both + /// currencies are sent exactly as given. Levels are aggregated + /// remaining balances per price level, ascending on the + /// [`ask`](OrderBookSide::Ask) side and descending on the + /// [`bid`](OrderBookSide::Bid) side; `side` is required by the + /// server. + /// + /// Cursors are side-specific: reusing an ask cursor on a bid walk is + /// rejected with [`Error::InvalidCursor`]. A cursor silently overrides + /// the `offset` page position on the server. See [`OrderBook`] for the + /// truncation and snapshot invariants. + pub async fn order_pair_book( + &self, + base: &str, + quote: &str, + side: OrderBookSide, + opts: OrderBookOpts, + ) -> Result { + let base = super::validate_segment(base)?; + let quote = super::validate_segment(quote)?; + let mut query = crate::indexer::page_query(opts.offset, opts.items); + query.push(("side", side.as_str().to_owned())); + if let Some(cursor) = opts.cursor { + query.push(("cursor", cursor)); + } + self.get(&format!("/order/pair/{base}_{quote}/book"), &query).await + } + + /// Walks one side of an order book page by page, `items` levels per + /// page (clamped to 1..=100). + /// + /// The walk stops when the server reports the last page — including a + /// book that was truncated by the server's per-request order cap, + /// which is signalled with `next_cursor: null` (see + /// [`OrderBook::truncated`]). Note the walk is not a consistent + /// snapshot of a moving book. + /// + /// Truncation detail: a truncated page is mapped to a plain end-of-walk + /// here, so a pager consumer cannot distinguish a truncated book from a + /// fully walked one. When your caller must know (e.g. to warn about an + /// incomplete aggregation), call [`Client::order_pair_book`] directly + /// and inspect [`OrderBook::truncated`]. + pub fn order_book_pager( + &self, + base: &str, + quote: &str, + side: OrderBookSide, + items: u32, + ) -> Result, Error> { + let base = super::validate_segment(base)?.to_owned(); + let quote = super::validate_segment(quote)?.to_owned(); + let items = crate::indexer::clamp_items(items); + let client = self.clone(); + Ok(Pager::new(move |cursor| { + let (client, base, quote) = (client.clone(), base.clone(), quote.clone()); + Box::pin(async move { + let book = client + .order_pair_book( + &base, + "e, + side, + OrderBookOpts { + offset: 0, + items, + cursor, + }, + ) + .await?; + Ok(Page { + items: book.items, + next_cursor: book.next_cursor, + }) + }) + })) + } } diff --git a/src/indexer/pager.rs b/src/indexer/pager.rs new file mode 100644 index 0000000..acfba25 --- /dev/null +++ b/src/indexer/pager.rs @@ -0,0 +1,116 @@ +// Copyright (c) 2026 Mintlayer Institutional FZCO +// Contact: hello@mintlayer.org +// +// Use of this source code is governed by an MIT license +// that can be found in the LICENSE file. + +//! Cursor pagination helper for the indexer's keyset-paginated listings. + +use std::future::Future; +use std::pin::Pin; + +use crate::indexer::error::Error; +use crate::indexer::types::Page; + +/// The future produced by a [`Pager`]'s page fetch. +pub type PagedFuture = Pin, Error>> + Send>>; + +/// A stateful walk over a cursor-paginated indexer listing (pools, the +/// global transaction listing, the holders listings, and the order book). +/// +/// Constructed through the `*_pager` methods on [`crate::indexer::Client`] +/// (for example [`crate::indexer::Client::coin_holders_pager`]). Each fetch +/// transparently follows the listing's `next_cursor`, and the walk stops +/// once the server sends `next_cursor: null` — including an order book page +/// that was [`truncated`](crate::indexer::OrderBook::truncated), whose +/// cursor cannot be continued. +/// +/// Cursors are minted by the indexer; the pager only ever replays what the +/// server sent (or, via [`Pager::start_from`], a cursor a caller persisted +/// earlier). A failed fetch leaves the pager positioned at the same cursor, +/// so the walk may simply be retried. +/// +/// [`Pager::next`] and [`Pager::next_page`] are two views over the same +/// walk and should not be mixed: a manual [`Pager::next_page`] discards any +/// not-yet-yielded items buffered by [`Pager::next`]. +/// +/// Page stability is only guaranteed once the indexer's scanner is fully +/// caught up: during catch-up or a reorg a cursor walk may skip or repeat +/// an entry. +pub struct Pager { + fetch: Box) -> PagedFuture + Send>, + next_cursor: Option, + pending: std::vec::IntoIter, + done: bool, +} + +impl Pager { + pub(crate) fn new( + fetch: impl FnMut(Option) -> PagedFuture + Send + 'static, + ) -> Self { + Self { + fetch: Box::new(fetch), + next_cursor: None, + pending: Vec::new().into_iter(), + done: false, + } + } + + /// Resumes a walk from a cursor obtained earlier (typically persisted + /// after a previous [`Pager`] run). Also usable on an exhausted + /// [`Pager`]: it rewinds the walk to the given cursor. + #[must_use] + pub fn start_from(mut self, cursor: impl Into) -> Self { + self.next_cursor = Some(cursor.into()); + self.pending = Vec::new().into_iter(); + self.done = false; + self + } + + /// The cursor the next page fetch resumes from; `None` at the start of + /// a fresh walk and once the walk is exhausted. + #[must_use] + pub fn cursor(&self) -> Option<&str> { + self.next_cursor.as_deref() + } + + /// Fetches the next page, or `None` once the listing is exhausted. + /// Any items buffered by [`Pager::next`] but not yet yielded are + /// discarded — unless the walk is already exhausted, in which case + /// nothing is fetched and nothing is discarded (a later [`Pager::next`] + /// still yields the buffered items). + pub async fn next_page(&mut self) -> Result>, Error> { + if self.done { + return Ok(None); + } + self.pending = Vec::new().into_iter(); + // Cloned, not taken: on a failed fetch the cursor must survive so a + // retried call resumes from the same position (see the retry doc + // above). The success path overwrites it below anyway. + let cursor = self.next_cursor.clone(); + let Page { items, next_cursor } = (self.fetch)(cursor).await?; + self.done = next_cursor.is_none(); + self.next_cursor = next_cursor; + Ok(Some(Page { + items, + next_cursor: self.next_cursor.clone(), + })) + } + + /// Returns the next item, fetching pages as needed; `None` once the + /// listing is exhausted. A fetch error is yielded as `Some(Err(..))` + /// and leaves the pager positioned at the same cursor, so the walk can + /// be retried. + pub async fn next(&mut self) -> Option> { + loop { + if let Some(item) = self.pending.next() { + return Some(Ok(item)); + } + match self.next_page().await { + Ok(Some(page)) => self.pending = page.items.into_iter(), + Ok(None) => return None, + Err(error) => return Some(Err(error)), + } + } + } +} diff --git a/src/indexer/pool.rs b/src/indexer/pool.rs index 97b427f..6d727eb 100644 --- a/src/indexer/pool.rs +++ b/src/indexer/pool.rs @@ -8,7 +8,7 @@ use serde::Deserialize; -use super::{Error, Pool, PoolDelegation, PoolListOpts}; +use super::{Error, Page, Pager, Pool, PoolDelegation, PoolListOpts}; use crate::indexer::Client; impl Client { @@ -51,4 +51,39 @@ impl Client { let id = super::validate_segment(id)?; self.get(&format!("/pool/{id}/delegations"), &[]).await } + + /// Lists stake pools along the creation-height cursor walk + /// (`GET /pool` with `cursor`). + /// + /// The cursor walk only exists for the server's default `by_height` + /// sort: combining a cursor with any other sort is rejected with + /// [`Error::BadRequest`], so this method sends no `sort` parameter. + /// For a `by_pledge` listing use [`Client::list_pools`]. + /// + /// When `cursor` is `None` the walk starts from the beginning. A + /// cursor silently overrides the `offset` page position on the server, + /// so this method takes no offset. + pub async fn list_pools_paged( + &self, + cursor: Option<&str>, + items: u32, + ) -> Result, Error> { + let query = [ + ("cursor", cursor.unwrap_or("").to_owned()), + ("items", crate::indexer::clamp_items(items).to_string()), + ]; + self.get("/pool", &query).await + } + + /// Walks the stake pools along the creation-height cursor walk, + /// `items` pools per page (clamped to 1..=100). See + /// [`Client::list_pools_paged`] for the sort restriction. + pub fn pools_pager(&self, items: u32) -> Pager { + let items = crate::indexer::clamp_items(items); + let client = self.clone(); + Pager::new(move |cursor| { + let client = client.clone(); + Box::pin(async move { client.list_pools_paged(cursor.as_deref(), items).await }) + }) + } } diff --git a/src/indexer/statistics.rs b/src/indexer/statistics.rs index f38bfd7..66b3f69 100644 --- a/src/indexer/statistics.rs +++ b/src/indexer/statistics.rs @@ -6,7 +6,7 @@ //! Indexer statistics endpoints. -use super::{CoinStats, Error}; +use super::{CoinStats, Error, Holder, HoldersOpts, Page, Pager}; use crate::indexer::Client; impl Client { @@ -32,4 +32,75 @@ impl Client { }; self.get("/feerate", &query).await } + + /// Lists the holders of the native coin, largest balance first + /// (`GET /statistics/coin/holders`). Amounts are rendered with the + /// coin's decimals. + pub async fn coin_holders(&self, opts: HoldersOpts) -> Result, Error> { + let mut query = crate::indexer::page_query(opts.offset, opts.items); + if let Some(cursor) = opts.cursor { + query.push(("cursor", cursor)); + } + self.get("/statistics/coin/holders", &query).await + } + + /// Lists the holders of a token, largest balance first + /// (`GET /statistics/token/{token_id}/holders`). Amounts are rendered + /// with the token's decimals; an unknown token is rejected with + /// [`Error::TokenNotFound`]. + pub async fn token_holders( + &self, + token_id: &str, + opts: HoldersOpts, + ) -> Result, Error> { + let token_id = super::validate_segment(token_id)?; + let mut query = crate::indexer::page_query(opts.offset, opts.items); + if let Some(cursor) = opts.cursor { + query.push(("cursor", cursor)); + } + self.get(&format!("/statistics/token/{token_id}/holders"), &query).await + } + + /// Walks the native coin holders page by page, `items` holders per + /// page (clamped to 1..=100). + pub fn coin_holders_pager(&self, items: u32) -> Pager { + let items = crate::indexer::clamp_items(items); + let client = self.clone(); + Pager::new(move |cursor| { + let client = client.clone(); + Box::pin(async move { + client + .coin_holders(HoldersOpts { + offset: 0, + items, + cursor, + }) + .await + }) + }) + } + + /// Walks a token's holders page by page, `items` holders per page + /// (clamped to 1..=100). The token id is validated up front, so the + /// walk itself cannot fail with an invalid URL. + pub fn token_holders_pager(&self, token_id: &str, items: u32) -> Result, Error> { + let token_id = super::validate_segment(token_id)?.to_owned(); + let items = crate::indexer::clamp_items(items); + let client = self.clone(); + Ok(Pager::new(move |cursor| { + let (client, token_id) = (client.clone(), token_id.clone()); + Box::pin(async move { + client + .token_holders( + &token_id, + HoldersOpts { + offset: 0, + items, + cursor, + }, + ) + .await + }) + })) + } } diff --git a/src/indexer/transaction.rs b/src/indexer/transaction.rs index 9e24072..fd2e5d1 100644 --- a/src/indexer/transaction.rs +++ b/src/indexer/transaction.rs @@ -8,7 +8,7 @@ use serde::Deserialize; -use super::{Error, MerklePath, PageOpts, Transaction}; +use super::{Error, MerklePath, OffsetMode, Page, PageOpts, Pager, Transaction}; use crate::indexer::Client; impl Client { @@ -56,4 +56,53 @@ impl Client { let response: Response = self.post("/transaction", signed_tx_hex).await?; Ok(response.tx_id) } + + /// Lists transactions along the global keyset cursor walk + /// (`GET /transaction` with `cursor`), newest block first, in block + /// order within each block. + /// + /// When `cursor` is `None` the walk starts from the beginning. A + /// cursor cannot be combined with an `offset_mode` (the server rejects + /// the pair with [`Error::BadRequest`]), so this method takes no + /// offset; for offset-based listings use [`Client::list_transactions`] + /// or [`Client::list_transactions_with_offset_mode`]. A cursor + /// silently overrides the `offset` page position on the server, so + /// this method takes no offset either. + pub async fn list_transactions_paged( + &self, + cursor: Option<&str>, + items: u32, + ) -> Result, Error> { + let query = [ + ("cursor", cursor.unwrap_or("").to_owned()), + ("items", crate::indexer::clamp_items(items).to_string()), + ]; + self.get("/transaction", &query).await + } + + /// Lists transactions with explicit offset semantics and no cursor + /// (`GET /transaction` with `offset_mode`). [`OffsetMode::Legacy`] is + /// the server default (see [`Client::list_transactions`]); + /// [`OffsetMode::Absolute`] treats `offset` as a global transaction + /// index, stable across scanner catch-up. + pub async fn list_transactions_with_offset_mode( + &self, + mode: OffsetMode, + opts: PageOpts, + ) -> Result, Error> { + let mut query = crate::indexer::page_query(opts.offset, opts.items); + query.push(("offset_mode", mode.as_str().to_owned())); + self.get("/transaction", &query).await + } + + /// Walks the global transaction listing page by page, `items` + /// transactions per page (clamped to 1..=100). + pub fn transactions_pager(&self, items: u32) -> Pager { + let items = crate::indexer::clamp_items(items); + let client = self.clone(); + Pager::new(move |cursor| { + let client = client.clone(); + Box::pin(async move { client.list_transactions_paged(cursor.as_deref(), items).await }) + }) + } } diff --git a/src/indexer/types.rs b/src/indexer/types.rs index 1270555..0de9c08 100644 --- a/src/indexer/types.rs +++ b/src/indexer/types.rs @@ -137,8 +137,8 @@ pub struct Block { /// A transaction as reported by the indexer. /// /// [`block_id`](#structfield.block_id), [`timestamp`](#structfield.timestamp) -/// and [`confirmations`](#structfield.confirmations) are empty strings while -/// the transaction is unconfirmed. +/// and [`confirmations`](#structfield.confirmations) are `None` while the +/// transaction is pending (not yet included in a block). #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct Transaction { /// Hex-encoded transaction id. @@ -147,12 +147,13 @@ pub struct Transaction { pub inputs: serde_json::Value, /// The transaction outputs. pub outputs: serde_json::Value, - /// Hex-encoded id of the confirming block, empty when unconfirmed. - pub block_id: String, - /// Confirmation timestamp, empty when unconfirmed. - pub timestamp: String, - /// Number of confirmations, empty when unconfirmed. - pub confirmations: String, + /// Hex-encoded id of the confirming block; `None` for a pending + /// (mempool) transaction. + pub block_id: Option, + /// Confirmation timestamp; `None` while unconfirmed. + pub timestamp: Option, + /// Number of confirmations; `None` while unconfirmed. + pub confirmations: Option, } /// The merkle path of a confirmed transaction. @@ -384,8 +385,9 @@ pub struct CoinStats { /// The indexer defaults to offset 0 and 10 items per page. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct PageOpts { - /// Number of items to skip. - pub offset: u32, + /// Number of items to skip; the server accepts the full `u64` range + /// (`OffsetMode::Absolute` offsets are global transaction indexes). + pub offset: u64, /// Number of items to return. pub items: u32, } @@ -413,9 +415,160 @@ impl PoolSort { #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct PoolListOpts { /// Number of pools to skip. - pub offset: u32, + pub offset: u64, /// Number of pools to return. pub items: u32, /// Sort order; defaults to [`PoolSort::ByHeight`] on the server. pub sort: Option, } + +/// One page of a cursor-paginated listing. +/// +/// The indexer's cursor envelope is `{ "items": [...], "next_cursor": ... }`: +/// `next_cursor` is the opaque cursor of the last returned item, and is +/// `None` once the result set is exhausted. Cursors are minted by the +/// indexer; pass them back verbatim — the SDK never constructs or mutates +/// them. See [`crate::indexer::Pager`] for a ready-made walk. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Page { + /// The items of this page. + pub items: Vec, + /// Cursor resuming after the last item of this page; `None` on the last + /// page. + pub next_cursor: Option, +} + +/// One entry of a holders listing (`GET /statistics/coin/holders` or +/// `GET /statistics/token/{id}/holders`), ordered by balance, largest +/// first. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Holder { + /// Bech32 address of the holder. + pub address: String, + /// The holder's balance, rendered with the coin's or the token's + /// decimals. + pub amount: Amount, +} + +/// Parameters for the holders listings. +/// +/// The indexer defaults to offset 0 and 10 items per page. When `cursor` is +/// set the server resolves the page position from the cursor and silently +/// ignores `offset` (`items` still applies). +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct HoldersOpts { + /// Number of holders to skip (no-cursor walks only). + pub offset: u64, + /// Number of holders to return. + pub items: u32, + /// Opaque cursor resuming a previous walk. + pub cursor: Option, +} + +/// Offset semantics of the global transaction listing (`GET /transaction` +/// without a cursor). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OffsetMode { + /// `offset` counts transactions from the beginning of the listing as it + /// happens to be scanned right now (the historical behavior; the server + /// default). + Legacy, + /// `offset` is a global transaction index: the page holds the + /// transactions immediately preceding that index in the listing + /// (global indexes below it), stable across scanner catch-up. + Absolute, +} + +impl OffsetMode { + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Legacy => "legacy", + Self::Absolute => "absolute", + } + } +} + +/// Side of an order book (`GET /order/pair/{pair}/book`); the server +/// requires it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OrderBookSide { + /// Orders asking for the base currency (first in the pair) while giving + /// the quote currency; levels ordered by ascending price. + Ask, + /// Orders asking for the quote currency while giving the base currency + /// (the reverse); levels ordered by descending price. + Bid, +} + +impl OrderBookSide { + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Ask => "ask", + Self::Bid => "bid", + } + } +} + +/// The price of an order book level: quote currency per one base unit. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OrderBookPrice { + /// The exact price as a reduced rational number of atoms + /// (`"quote atoms/base atoms"`, e.g. `"3/2"`). + pub atoms: String, + /// The price in currency units, truncated (floored) toward zero; the + /// exact value sits alongside in + /// [`atoms`](#structfield.atoms). + pub decimal: String, +} + +/// One aggregated price level of an order book. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OrderBookLevel { + /// The level price. + pub price: OrderBookPrice, + /// The remaining base-currency amount resting at this level. + pub amount: Amount, +} + +/// The order book of a trading pair. +/// +/// Invariants (api-server v2): +/// - `next_cursor` is `None` on the last page. +/// - When +/// [`truncated`](#structfield.truncated) is `true` the per-request cap of +/// 10,000 aggregated orders was hit; the server then sends +/// `next_cursor: null` because the levels in hand are an incomplete +/// aggregation and the cursor walk cannot be continued. Never treat a +/// truncated book as complete. +/// - The book is computed fresh on every request, so a paginated walk is +/// not a consistent snapshot of a moving book. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OrderBook { + /// The price levels of this page (ascending on the ask side, + /// descending on the bid side). + pub items: Vec, + /// Cursor resuming after the last level of this page; `None` on the + /// last page and whenever [`truncated`](#structfield.truncated) is + /// `true`. + pub next_cursor: Option, + /// `true` when the server-side cap truncated the book; absent means + /// `false`. + #[serde(default)] + pub truncated: bool, +} + +/// Parameters for the order book listing. +/// +/// The indexer defaults to offset 0 and 10 items per page. When `cursor` is +/// set the server resolves the page position from the cursor and silently +/// ignores `offset` (`items` still applies). Cursors are side-specific: an +/// ask-side cursor on a bid walk is rejected with `400 "Invalid cursor"`. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct OrderBookOpts { + /// Number of levels to skip (no-cursor requests only). + pub offset: u64, + /// Number of levels to return. + pub items: u32, + /// Opaque cursor resuming a previous walk on the same side. + pub cursor: Option, +} diff --git a/tests/indexer.rs b/tests/indexer.rs index e460ff8..5a8d65b 100644 --- a/tests/indexer.rs +++ b/tests/indexer.rs @@ -12,10 +12,11 @@ mod common; use common::respond; use httpmock::prelude::*; use serde_json::json; +use std::sync::atomic::{AtomicUsize, Ordering}; use mintlayer_sdk::indexer::{ - Amount, Client, Error, PageOpts, PerThousand, PoolListOpts, PoolSort, Uint64, Utxo, - UtxoOutpoint, + Amount, Client, Error, HoldersOpts, OffsetMode, OrderBookOpts, OrderBookSide, PageOpts, + PerThousand, PoolListOpts, PoolSort, Uint64, Utxo, UtxoOutpoint, }; #[tokio::test] @@ -359,7 +360,7 @@ async fn utxo_wire_key_is_utxo() { } #[tokio::test] -async fn unconfirmed_transaction_empty_string_fields() { +async fn pending_transaction_null_fields() { let server = MockServer::start(); let mock = server.mock(|when, then| { when.method(GET).path("/api/v2/transaction/aabb"); @@ -370,9 +371,9 @@ async fn unconfirmed_transaction_empty_string_fields() { "id": "aabb", "inputs": [], "outputs": [], - "block_id": "", - "timestamp": "", - "confirmations": "" + "block_id": null, + "timestamp": null, + "confirmations": null }) .to_string(), ); @@ -382,9 +383,9 @@ async fn unconfirmed_transaction_empty_string_fields() { let tx = client.transaction("aabb").await.unwrap(); assert_eq!(tx.id, "aabb"); - assert_eq!(tx.block_id, ""); - assert_eq!(tx.timestamp, ""); - assert_eq!(tx.confirmations, ""); + assert_eq!(tx.block_id, None); + assert_eq!(tx.timestamp, None); + assert_eq!(tx.confirmations, None); mock.assert(); } @@ -807,3 +808,598 @@ async fn remaining_types_decode() { genesis_mock.assert(); all_utxos_mock.assert(); } + +// --------------------------------------------------------------------------- +// Cursor pagination, holders, and order book (api-server v2, PR #2130) +// --------------------------------------------------------------------------- + +fn holder(address: &str, atoms: &str, decimal: &str) -> serde_json::Value { + json!({"address": address, "amount": {"atoms": atoms, "decimal": decimal}}) +} + +fn full_pool(pool_id: &str) -> serde_json::Value { + json!({ + "pool_id": pool_id, + "decommission_destination": "mtct1x", + "staker_balance": {"atoms": "1000", "decimal": "0.00000001"}, + "margin_ratio_per_thousand": "3.5%", + "cost_per_block": {"atoms": "1000", "decimal": "0.00000001"}, + "vrf_public_key": "vrfpub1x", + "delegations_balance": {"atoms": "2000", "decimal": "0.00000002"} + }) +} + +#[tokio::test] +async fn coin_holders_cursor_walk() { + let server = MockServer::start(); + let first = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/statistics/coin/holders") + .query_param("items", "2") + .matches(|request| { + let params = request.query_params.as_deref().unwrap_or(&[]); + !params.iter().any(|(name, _)| *name == "cursor") + }); + respond( + then, + 200, + json!({ + "items": [ + holder("mt1qrich", "9000000000000", "90"), + holder("mt1qpoor", "1000000000000", "10"), + ], + "next_cursor": "cursor-1", + }) + .to_string(), + ); + }); + let second = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/statistics/coin/holders") + .query_param("cursor", "cursor-1") + .query_param("items", "2"); + respond( + then, + 200, + json!({"items": [holder("mt1qmid", "5000000000000", "50")], "next_cursor": null}) + .to_string(), + ); + }); + + let client = Client::new(server.url("")); + let mut pager = client.coin_holders_pager(2); + let mut walked = Vec::new(); + while let Some(item) = pager.next().await { + walked.push(item.unwrap().address); + } + assert_eq!(walked, ["mt1qrich", "mt1qpoor", "mt1qmid"]); + first.assert(); + second.assert(); +} + +#[tokio::test] +async fn token_holders_walk_and_last_page() { + let server = MockServer::start(); + let mock = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/statistics/token/tok1/holders") + .query_param("cursor", "resume"); + respond( + then, + 200, + json!({"items": [holder("mt1qone", "7", "0.00000007")], "next_cursor": null}) + .to_string(), + ); + }); + + let client = Client::new(server.url("")); + let page = client + .token_holders( + "tok1", + HoldersOpts { + offset: 0, + items: 0, + cursor: Some("resume".to_owned()), + }, + ) + .await + .unwrap(); + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].amount.atoms, 7); + assert_eq!(page.next_cursor, None); + mock.assert(); +} + +#[tokio::test] +async fn token_holders_unknown_token_is_token_not_found() { + let server = MockServer::start(); + let mock = server.mock(|when, then| { + when.method(GET).path("/api/v2/statistics/token/tok1/holders"); + respond(then, 404, json!({"error": "Token not found"}).to_string()); + }); + + let client = Client::new(server.url("")); + let error = client.token_holders("tok1", HoldersOpts::default()).await.unwrap_err(); + assert!(matches!(error, Error::TokenNotFound), "{error:?}"); + mock.assert(); +} + +#[tokio::test] +async fn pager_clamps_page_size_to_server_limits() { + let server = MockServer::start(); + let clamped_high = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/statistics/coin/holders") + .query_param("items", "100"); + respond( + then, + 200, + json!({"items": [], "next_cursor": null}).to_string(), + ); + }); + let clamped_low = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/statistics/coin/holders") + .query_param("items", "1"); + respond( + then, + 200, + json!({"items": [], "next_cursor": null}).to_string(), + ); + }); + + let client = Client::new(server.url("")); + client.coin_holders_pager(250).next().await; + client.coin_holders_pager(0).next().await; + clamped_high.assert(); + clamped_low.assert(); +} + +#[tokio::test] +async fn oversized_items_is_invalid_num_items() { + let server = MockServer::start(); + let mock = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/statistics/coin/holders") + .query_param("items", "150"); + respond( + then, + 400, + json!({"error": "Invalid number of items"}).to_string(), + ); + }); + + let client = Client::new(server.url("")); + let error = client + .coin_holders(HoldersOpts { + offset: 0, + items: 150, + cursor: None, + }) + .await + .unwrap_err(); + assert!(matches!(error, Error::InvalidNumItems), "{error:?}"); + mock.assert(); +} + +#[tokio::test] +async fn order_book_decodes_both_sides() { + let server = MockServer::start(); + let ask = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/order/pair/ML_tok1/book") + .query_param("side", "ask") + .query_param("items", "2"); + respond( + then, + 200, + json!({ + "items": [ + {"price": {"atoms": "3/2", "decimal": "1.5"}, + "amount": {"atoms": "200000000", "decimal": "2"}}, + {"price": {"atoms": "2/1", "decimal": "2"}, + "amount": {"atoms": "100000000", "decimal": "1"}}, + ], + "next_cursor": "ask-cursor", + }) + .to_string(), + ); + }); + let bid = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/order/pair/ML_tok1/book") + .query_param("side", "bid"); + respond( + then, + 200, + json!({ + "items": [ + {"price": {"atoms": "1/3", "decimal": "0.33"}, + "amount": {"atoms": "100000000", "decimal": "1"}}, + ], + "next_cursor": "bid-cursor", + }) + .to_string(), + ); + }); + + let client = Client::new(server.url("")); + let ask_book = client + .order_pair_book( + "ML", + "tok1", + OrderBookSide::Ask, + OrderBookOpts { + offset: 0, + items: 2, + cursor: None, + }, + ) + .await + .unwrap(); + assert!(!ask_book.truncated); + assert_eq!(ask_book.next_cursor.as_deref(), Some("ask-cursor")); + assert_eq!(ask_book.items[0].price.atoms, "3/2"); + assert_eq!(ask_book.items[0].price.decimal, "1.5"); + assert_eq!(ask_book.items[0].amount.atoms, 200_000_000); + + let bid_book = client + .order_pair_book("ML", "tok1", OrderBookSide::Bid, OrderBookOpts::default()) + .await + .unwrap(); + assert_eq!(bid_book.items[0].price.atoms, "1/3"); + ask.assert(); + bid.assert(); +} + +#[tokio::test] +async fn truncated_order_book_stops_the_pager() { + let server = MockServer::start(); + // A truncated book always arrives with `next_cursor: null`; a pager + // walk must stop instead of trying to continue. + let mock = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/order/pair/ML_tok1/book") + .query_param("side", "ask"); + respond( + then, + 200, + json!({ + "items": [ + {"price": {"atoms": "3/2", "decimal": "1.5"}, + "amount": {"atoms": "200000000", "decimal": "2"}}, + ], + "next_cursor": null, + "truncated": true, + }) + .to_string(), + ); + }); + + let client = Client::new(server.url("")); + let book = client + .order_pair_book("ML", "tok1", OrderBookSide::Ask, OrderBookOpts::default()) + .await + .unwrap(); + assert!(book.truncated); + assert_eq!(book.next_cursor, None); + + let mut pager = client.order_book_pager("ML", "tok1", OrderBookSide::Ask, 10).unwrap(); + let mut levels = 0; + while let Some(level) = pager.next().await { + level.unwrap(); + levels += 1; + } + assert_eq!(levels, 1); + mock.assert_hits(2); // one direct fetch, one pager fetch; no continuation +} + +#[tokio::test] +async fn wrong_side_cursor_is_invalid_cursor() { + let server = MockServer::start(); + let mock = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/order/pair/ML_tok1/book") + .query_param("side", "bid") + .query_param("cursor", "ask-cursor"); + respond(then, 400, json!({"error": "Invalid cursor"}).to_string()); + }); + + let client = Client::new(server.url("")); + let error = client + .order_pair_book( + "ML", + "tok1", + OrderBookSide::Bid, + OrderBookOpts { + offset: 0, + items: 0, + cursor: Some("ask-cursor".to_owned()), + }, + ) + .await + .unwrap_err(); + assert!(matches!(error, Error::InvalidCursor), "{error:?}"); + mock.assert(); +} + +#[tokio::test] +async fn pools_cursor_walk() { + let server = MockServer::start(); + let first = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/pool") + .query_param("cursor", "") + .query_param("items", "1"); + respond( + then, + 200, + json!({"items": [full_pool("p1")], "next_cursor": "pool-cursor"}).to_string(), + ); + }); + let second = server.mock(|when, then| { + when.method(GET).path("/api/v2/pool").query_param("cursor", "pool-cursor"); + respond( + then, + 200, + json!({"items": [full_pool("p2")], "next_cursor": null}).to_string(), + ); + }); + + let client = Client::new(server.url("")); + let mut pager = client.pools_pager(1); + let mut walked = Vec::new(); + while let Some(pool) = pager.next().await { + walked.push(pool.unwrap().pool_id); + } + assert_eq!(walked, ["p1", "p2"]); + first.assert(); + second.assert(); +} + +#[tokio::test] +async fn transactions_cursor_walk_with_pending_block_id() { + let server = MockServer::start(); + let first = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/transaction") + .query_param("cursor", "") + .query_param("items", "1"); + respond( + then, + 200, + json!({ + "items": [{ + "id": "tx1", + "inputs": [], + "outputs": [], + "block_id": "0000aa", + "timestamp": "1700000000", + "confirmations": "10", + "tx_global_index": 41, + }], + "next_cursor": "tx-cursor", + }) + .to_string(), + ); + }); + let second = server.mock(|when, then| { + when.method(GET).path("/api/v2/transaction").query_param("cursor", "tx-cursor"); + respond( + then, + 200, + json!({ + "items": [{ + "id": "tx2", + "inputs": [], + "outputs": [], + "block_id": null, + "timestamp": null, + "confirmations": null, + }], + "next_cursor": null, + }) + .to_string(), + ); + }); + + let client = Client::new(server.url("")); + let mut pager = client.transactions_pager(1); + let mut walked = Vec::new(); + while let Some(tx) = pager.next().await { + let tx = tx.unwrap(); + assert_eq!(tx.block_id.is_none(), tx.id == "tx2"); + walked.push(tx.id); + } + assert_eq!(walked, ["tx1", "tx2"]); + first.assert(); + second.assert(); +} + +#[tokio::test] +async fn transactions_offset_mode_is_sent() { + let server = MockServer::start(); + let mock = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/transaction") + .query_param("offset_mode", "absolute") + .query_param("offset", "5") + .query_param("items", "2"); + respond(then, 200, json!([]).to_string()); + }); + + let client = Client::new(server.url("")); + let transactions = client + .list_transactions_with_offset_mode( + OffsetMode::Absolute, + PageOpts { + offset: 5, + items: 2, + }, + ) + .await + .unwrap(); + assert!(transactions.is_empty()); + mock.assert(); +} + +#[tokio::test] +async fn incompatible_parameters_surface_as_bad_request() { + let server = MockServer::start(); + // A cursor combined with a non-default pools sort (or with + // `offset_mode`) is rejected server-side; the typed methods cannot + // produce that pair, but the mapping must surface the server's verdict + // verbatim. + let mock = server.mock(|when, then| { + when.method(GET).path("/api/v2/pool").query_param("cursor", "pool-cursor"); + respond(then, 400, json!({"error": "Bad request"}).to_string()); + }); + + let client = Client::new(server.url("")); + let error = client.list_pools_paged(Some("pool-cursor"), 10).await.unwrap_err(); + assert!(matches!(error, Error::BadRequest), "{error:?}"); + mock.assert(); +} + +#[tokio::test] +async fn unmatched_server_errors_fall_back_to_http() { + let server = MockServer::start(); + let mock = server.mock(|when, then| { + when.method(GET).path("/api/v2/statistics/coin/holders"); + respond( + then, + 500, + json!({"error": "Internal server error"}).to_string(), + ); + }); + + let client = Client::new(server.url("")); + let error = client.coin_holders(HoldersOpts::default()).await.unwrap_err(); + match error { + Error::Http { status_code, body } => { + assert_eq!(status_code, 500); + assert!(body.contains("Internal server error"), "{body}"); + } + other => panic!("expected Http, got {other:?}"), + } + mock.assert(); +} + +#[tokio::test] +async fn pager_retry_after_failed_fetch_resumes_from_same_cursor() { + let server = MockServer::start(); + let first = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/statistics/coin/holders") + .query_param("items", "1") + .matches(|request| { + let params = request.query_params.as_deref().unwrap_or(&[]); + !params.iter().any(|(name, _)| *name == "cursor") + }); + respond( + then, + 200, + json!({"items": [holder("mt1qa", "3000000000000", "30")], "next_cursor": "cursor-1"}) + .to_string(), + ); + }); + + // The first request carrying cursor-1 fails; the retry (same request) + // succeeds. httpmock's `matches` takes a plain fn pointer (no + // captures), so the two identical requests are routed via a static + // request counter; httpmock serves the first registered match. + static RETRY_HITS: AtomicUsize = AtomicUsize::new(0); + let failing = server.mock(|when, then| { + when.method(GET).path("/api/v2/statistics/coin/holders").matches(|request| { + let params = request.query_params.as_deref().unwrap_or(&[]); + if !params.iter().any(|(name, value)| name == "cursor" && value == "cursor-1") { + return false; + } + RETRY_HITS.fetch_add(1, Ordering::SeqCst) == 0 + }); + respond(then, 500, json!({"error": "boom"}).to_string()); + }); + let succeeding = server.mock(|when, then| { + when.method(GET).path("/api/v2/statistics/coin/holders").matches(|request| { + let params = request.query_params.as_deref().unwrap_or(&[]); + if !params.iter().any(|(name, value)| name == "cursor" && value == "cursor-1") { + return false; + } + RETRY_HITS.load(Ordering::SeqCst) >= 1 + }); + respond( + then, + 200, + json!({"items": [holder("mt1qb", "4000000000000", "40")], "next_cursor": null}) + .to_string(), + ); + }); + + let client = Client::new(server.url("")); + let mut pager = client.coin_holders_pager(1); + let mut walked = Vec::new(); + let mut errors = 0; + while let Some(item) = pager.next().await { + match item { + Ok(h) => walked.push(h.address), + Err(_) => errors += 1, + } + } + // The error is surfaced once, then the retry resumes from cursor-1: + // "mt1qa" is not re-emitted and the walk still reaches "mt1qb". + assert_eq!(errors, 1); + assert_eq!(walked, ["mt1qa", "mt1qb"]); + assert_eq!(pager.cursor(), None); + first.assert_hits(1); + failing.assert_hits(1); + succeeding.assert_hits(1); +} + +#[tokio::test] +async fn pager_start_from_rewinds_an_exhausted_walk() { + let server = MockServer::start(); + let first = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/statistics/coin/holders") + .query_param("items", "1") + .matches(|request| { + let params = request.query_params.as_deref().unwrap_or(&[]); + !params.iter().any(|(name, _)| *name == "cursor") + }); + respond( + then, + 200, + json!({"items": [holder("mt1qa", "3000000000000", "30")], "next_cursor": "cursor-1"}) + .to_string(), + ); + }); + let second = server.mock(|when, then| { + when.method(GET) + .path("/api/v2/statistics/coin/holders") + .query_param("cursor", "cursor-1") + .query_param("items", "1"); + respond( + then, + 200, + json!({"items": [holder("mt1qb", "4000000000000", "40")], "next_cursor": null}) + .to_string(), + ); + }); + + let client = Client::new(server.url("")); + let mut pager = client.coin_holders_pager(1); + while let Some(item) = pager.next().await { + item.unwrap(); + } + assert_eq!(pager.cursor(), None); + + // start_from on the exhausted pager rewinds it to cursor-1. + let mut pager = pager.start_from("cursor-1"); + assert_eq!(pager.cursor(), Some("cursor-1")); + let resumed = pager.next().await.unwrap().unwrap(); + assert_eq!(resumed.address, "mt1qb"); + assert!(pager.next().await.is_none()); + + first.assert_hits(1); + second.assert_hits(2); +}