From 971876993427f4cae9aee1e166cb47f0b46b1685 Mon Sep 17 00:00:00 2001 From: laddyr141-ui Date: Mon, 27 Jul 2026 11:13:47 +0000 Subject: [PATCH 1/5] fix(api): restore dependencies and methods dropped in bad merge Merging main into feat/api-hardening (59c9fdd) silently dropped the rand, metrics, and metrics-exporter-prometheus dependencies from Cargo.toml along with the PrometheusHandle/PrometheusBuilder imports and AppState::last_indexed_ledger, leaving main uncompilable. Restore the dependencies and imports, and re-add last_indexed_ledger adapted to the ArcSwap-backed snapshot (routes/mod.rs::cache_headers already calls it). Co-Authored-By: Claude Sonnet 5 --- api/Cargo.toml | 3 +++ api/src/indexer/mod.rs | 8 ++++++++ api/src/main.rs | 1 + 3 files changed, 12 insertions(+) diff --git a/api/Cargo.toml b/api/Cargo.toml index 0f0bee7..bc2e8ef 100644 --- a/api/Cargo.toml +++ b/api/Cargo.toml @@ -21,6 +21,9 @@ tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] } url = "2" +metrics = "0.24" +metrics-exporter-prometheus = { version = "0.18", default-features = false } +rand = "0.9" [[bin]] name = "stellar-rwa-api" diff --git a/api/src/indexer/mod.rs b/api/src/indexer/mod.rs index cd4d5e6..8738d83 100644 --- a/api/src/indexer/mod.rs +++ b/api/src/indexer/mod.rs @@ -17,6 +17,8 @@ use std::sync::Arc; use std::time::Duration; use arc_swap::ArcSwap; +use metrics_exporter_prometheus::PrometheusHandle; +use rand::Rng; use reqwest::header::RETRY_AFTER; use reqwest::StatusCode; use serde::Deserialize; @@ -137,6 +139,12 @@ impl AppState { (*guard).clone() } + /// The last ledger the indexer successfully read from, without cloning + /// the whole snapshot. + pub async fn last_indexed_ledger(&self) -> u32 { + self.inner.load().stats.last_indexed_ledger + } + fn replace(&self, next: Snapshot) { self.inner.store(Arc::new(next)); } diff --git a/api/src/main.rs b/api/src/main.rs index 21c5f7a..b52f0d0 100644 --- a/api/src/main.rs +++ b/api/src/main.rs @@ -12,6 +12,7 @@ mod routes; use std::net::SocketAddr; use indexer::{AppState, Config, ConfigError, Indexer}; +use metrics_exporter_prometheus::PrometheusBuilder; #[tokio::main] async fn main() { From 133b6cc2d5edcfc2682b4f3670b6233e819701d0 Mon Sep 17 00:00:00 2001 From: laddyr141-ui Date: Mon, 27 Jul 2026 11:16:36 +0000 Subject: [PATCH 2/5] perf(indexer): parallelize per-asset and per-address RPC reads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each refresh cycle read A assets fully sequentially, and within each asset issued two sequential RPC calls (get_record, balance) per allowlisted address — roughly A*(1 + 2N) sequential simulate calls every 10s poll. This doesn't scale as assets/addresses grow and risks tripping public-RPC rate limits. Index assets concurrently via a bounded futures::stream::buffer_unordered (MAX_CONCURRENT_ASSETS), and within each asset read every address's compliance record and balance concurrently as well (bounded by MAX_CONCURRENT_ADDRESS_READS), running the two per-address calls themselves in parallel via tokio::join!. Asset order is preserved by tagging each result with its original position and re-sorting after the unordered stream completes. Error-handling semantics are unchanged: a metadata/holder read failure still fails the asset (and, as before, the whole refresh) — that's addressed separately. Co-Authored-By: Claude Sonnet 5 --- api/Cargo.toml | 1 + api/src/indexer/mod.rs | 273 +++++++++++++++++++++++++++-------------- 2 files changed, 181 insertions(+), 93 deletions(-) diff --git a/api/Cargo.toml b/api/Cargo.toml index bc2e8ef..508c824 100644 --- a/api/Cargo.toml +++ b/api/Cargo.toml @@ -11,6 +11,7 @@ tokio = { version = "1", features = ["full"] } tower-http = { version = "0.5", features = ["cors", "trace"] } tower_governor = "0.5" reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } +futures = "0.3" serde = { version = "1", features = ["derive"] } serde_json = "1" arc-swap = "1" diff --git a/api/src/indexer/mod.rs b/api/src/indexer/mod.rs index 8738d83..d61062d 100644 --- a/api/src/indexer/mod.rs +++ b/api/src/indexer/mod.rs @@ -17,6 +17,7 @@ use std::sync::Arc; use std::time::Duration; use arc_swap::ArcSwap; +use futures::stream::{self, StreamExt}; use metrics_exporter_prometheus::PrometheusHandle; use rand::Rng; use reqwest::header::RETRY_AFTER; @@ -40,6 +41,14 @@ const RETRY_BASE_DELAY: Duration = Duration::from_millis(150); /// Ceiling on backoff growth between retries. const RETRY_MAX_DELAY: Duration = Duration::from_secs(2); +/// Upper bound on assets indexed concurrently. Bounded so a platform with +/// many assets doesn't open unbounded concurrent connections to the RPC node +/// in one refresh cycle. +const MAX_CONCURRENT_ASSETS: usize = 8; +/// Upper bound on allowlisted addresses read concurrently per asset, for the +/// same reason. +const MAX_CONCURRENT_ADDRESS_READS: usize = 16; + /// Static configuration for a network's contracts and RPC endpoint. #[derive(Debug, Clone)] pub struct Config { @@ -582,6 +591,20 @@ pub struct Indexer { state: AppState, } +/// Everything derived from indexing a single asset's contracts. +struct AssetIndexed { + asset: Asset, + holders: Vec, + compliance: ComplianceSummary, + dividends: Vec, +} + +/// The compliance record and token balance for a single allowlisted address. +struct AddressRead { + record: Option, + balance: i128, +} + impl Indexer { pub fn new(state: AppState) -> Self { let cfg = &state.config; @@ -625,72 +648,42 @@ impl Indexer { let raw_entries: Vec = serde_json::from_value(entries_read.value) .map_err(|e| IndexError::Decode(e.to_string()))?; - let mut assets = Vec::new(); + // Index every asset concurrently (bounded — see MAX_CONCURRENT_ASSETS) + // rather than one at a time; a platform with many assets would + // otherwise pay A*(1 + 2N) sequential RPC round trips per refresh. + // buffer_unordered doesn't preserve completion order, so each result + // is tagged with its position in raw_entries and re-sorted after. + let mut tagged: Vec<(usize, Result)> = + stream::iter(raw_entries.iter().enumerate()) + .map(|(idx, raw)| async move { (idx, self.index_asset(raw).await) }) + .buffer_unordered(MAX_CONCURRENT_ASSETS) + .collect() + .await; + tagged.sort_unstable_by_key(|(idx, _)| *idx); + + let mut assets = Vec::with_capacity(tagged.len()); let mut holders_map: HashMap> = HashMap::new(); let mut compliance_map: HashMap = HashMap::new(); let mut dividends_map: HashMap> = HashMap::new(); let mut total_distributions = 0usize; let mut tvl: i128 = 0; - for raw in &raw_entries { - let meta = self - .rpc - .read(&raw.token_contract, "get_metadata", vec![]) - .await - .inspect_err(|_| record_asset_read_error(raw.id, "get_metadata"))?; - let meta: RawMetadata = serde_json::from_value(meta.value) - .map_err(|e| IndexError::Decode(e.to_string()))?; - - let total_supply = parse_i128(&meta.total_supply); - let valuation = parse_i128(&raw.valuation); - - // Holders: every allowlisted address with a positive balance. - let (holders, summary, _) = self - .index_compliance_and_holders( - &meta.compliance_contract, - &raw.token_contract, - total_supply, - ) - .await?; - - // Dividends for this asset token; treated as empty on failure so one - // asset's dividend contract issue doesn't abort the whole refresh. - let dists = match self.index_dividends(&raw.token_contract).await { - Ok(dists) => dists, - Err(e) => { - record_asset_read_error(raw.id, "dividends"); - tracing::warn!(asset_id = raw.id, error = %e, "dividends read failed; treating as empty"); - Vec::new() - } - }; - total_distributions += dists.len(); - - if raw.active { - tvl += valuation; + for (_, result) in tagged { + let AssetIndexed { + asset, + holders, + compliance, + dividends, + } = result?; + + total_distributions += dividends.len(); + if asset.active { + tvl += parse_i128(&asset.valuation_cents); } - let asset = Asset { - id: raw.id, - token_contract: raw.token_contract.clone(), - issuer: raw.issuer.clone(), - name: raw.name.clone(), - symbol: meta.symbol, - asset_type: raw.asset_type.clone(), - description: meta.asset_description, - valuation_cents: valuation.to_string(), - valuation_usd: cents_to_usd(valuation), - decimals: meta.decimals, - total_supply: total_supply.to_string(), - holders: holders.len(), - active: raw.active, - paused: meta.paused, - compliance_contract: meta.compliance_contract, - created_at_ledger: raw.created_at, - }; - - holders_map.insert(raw.id, holders); - compliance_map.insert(raw.id, summary); - dividends_map.insert(raw.id, dists); + holders_map.insert(asset.id, holders); + compliance_map.insert(asset.id, compliance); + dividends_map.insert(asset.id, dividends); assets.push(asset); } @@ -724,6 +717,72 @@ impl Indexer { Ok(count) } + /// Read and derive everything for one asset: metadata, then its + /// compliance/holders and dividends concurrently (they hit independent + /// contracts, so there's no reason to wait on one before starting the + /// other). + async fn index_asset(&self, raw: &RawAssetEntry) -> Result { + let meta = self + .rpc + .read(&raw.token_contract, "get_metadata", vec![]) + .await + .inspect_err(|_| record_asset_read_error(raw.id, "get_metadata"))?; + let meta: RawMetadata = serde_json::from_value(meta.value) + .map_err(|e| IndexError::Decode(e.to_string()))?; + + let total_supply = parse_i128(&meta.total_supply); + let valuation = parse_i128(&raw.valuation); + + let (holders_result, dists_result) = tokio::join!( + self.index_compliance_and_holders( + &meta.compliance_contract, + &raw.token_contract, + total_supply, + ), + self.index_dividends(&raw.token_contract), + ); + + // Holders: every allowlisted address with a positive balance. + let (holders, summary, _) = holders_result?; + + // Dividends for this asset token; treated as empty on failure so one + // asset's dividend contract issue doesn't fail the whole asset. + let dividends = match dists_result { + Ok(dists) => dists, + Err(e) => { + record_asset_read_error(raw.id, "dividends"); + tracing::warn!(asset_id = raw.id, error = %e, "dividends read failed; treating as empty"); + Vec::new() + } + }; + + let asset = Asset { + id: raw.id, + token_contract: raw.token_contract.clone(), + issuer: raw.issuer.clone(), + name: raw.name.clone(), + symbol: meta.symbol, + asset_type: raw.asset_type.clone(), + description: meta.asset_description, + valuation_cents: valuation.to_string(), + valuation_usd: cents_to_usd(valuation), + decimals: meta.decimals, + total_supply: total_supply.to_string(), + holders: holders.len(), + active: raw.active, + paused: meta.paused, + compliance_contract: meta.compliance_contract, + created_at_ledger: raw.created_at, + }; + + Ok(AssetIndexed { + asset, + holders, + compliance: summary, + dividends, + }) + } + /// Read the compliance allowlist for an asset and derive both the holder /// list (allowlisted ∩ positive balance) and the non-PII summary. async fn index_compliance_and_holders( @@ -739,54 +798,48 @@ impl Indexer { let addresses: Vec = serde_json::from_value(allowlist.value) .map_err(|e| IndexError::Decode(e.to_string()))?; + // Read every address's record + balance concurrently (bounded — see + // MAX_CONCURRENT_ADDRESS_READS) instead of two sequential RPC calls + // per address, one address at a time. + let outcomes: Vec<(String, Result)> = stream::iter(addresses) + .map(|address| async move { + let result = self + .read_address(compliance_contract, token_contract, &address) + .await; + (address, result) + }) + .buffer_unordered(MAX_CONCURRENT_ADDRESS_READS) + .collect() + .await; + let mut holders = Vec::new(); let mut summary = ComplianceSummary::default(); let mut jurisdictions: BTreeMap = BTreeMap::new(); let mut approved_addresses = Vec::new(); - for address in &addresses { + for (address, outcome) in outcomes { summary.total_records += 1; + let AddressRead { record, balance } = outcome?; // Record status → summary counts. - if let Ok(rec) = self - .rpc - .read( - compliance_contract, - "get_record", - vec![address_scval(address)?], - ) - .await - { - if !rec.value.is_null() { - if let Ok(kyc) = serde_json::from_value::(rec.value) { - match normalize_status(&kyc.status).as_str() { - "Approved" => { - summary.approved += 1; - approved_addresses.push(address.clone()); - } - "Suspended" => summary.suspended += 1, - "Rejected" => summary.rejected += 1, - "Pending" => summary.pending += 1, - _ => {} - } - if kyc.expires_at != 0 { - summary.with_expiry += 1; - } - *jurisdictions.entry(kyc.jurisdiction).or_insert(0) += 1; + if let Some(kyc) = record { + match normalize_status(&kyc.status).as_str() { + "Approved" => { + summary.approved += 1; + approved_addresses.push(address.clone()); } + "Suspended" => summary.suspended += 1, + "Rejected" => summary.rejected += 1, + "Pending" => summary.pending += 1, + _ => {} + } + if kyc.expires_at != 0 { + summary.with_expiry += 1; } + *jurisdictions.entry(kyc.jurisdiction).or_insert(0) += 1; } // Balance → holder list. - let bal = self - .rpc - .read(token_contract, "balance", vec![address_scval(address)?]) - .await?; - let balance = match bal.value { - serde_json::Value::String(s) => parse_i128(&s), - serde_json::Value::Number(n) => n.as_i64().unwrap_or(0) as i128, - _ => 0, - }; if balance > 0 { holders.push(Holder { address: address.clone(), @@ -808,6 +861,40 @@ impl Indexer { Ok((holders, summary, approved_addresses)) } + /// Read one allowlisted address's compliance record and token balance + /// concurrently. A `get_record` failure is treated as "no record" (as + /// before parallelization, an unreadable/absent record just excludes the + /// address from the compliance summary); a `balance` failure propagates, + /// since holder totals must not silently omit a holder. + async fn read_address( + &self, + compliance_contract: &str, + token_contract: &str, + address: &str, + ) -> Result { + let record_arg = address_scval(address)?; + let balance_arg = address_scval(address)?; + + let (record_result, balance_result) = tokio::join!( + self.rpc.read(compliance_contract, "get_record", vec![record_arg]), + self.rpc.read(token_contract, "balance", vec![balance_arg]), + ); + + let record = match record_result { + Ok(rec) if !rec.value.is_null() => serde_json::from_value::(rec.value).ok(), + _ => None, + }; + + let bal = balance_result?; + let balance = match bal.value { + serde_json::Value::String(s) => parse_i128(&s), + serde_json::Value::Number(n) => n.as_i64().unwrap_or(0) as i128, + _ => 0, + }; + + Ok(AddressRead { record, balance }) + } + /// Read all distributions for an asset token from the dividend contract. async fn index_dividends(&self, token_contract: &str) -> Result, IndexError> { let read = self From 6c9fb65c2c1b00542bedd4804226be28596f62a4 Mon Sep 17 00:00:00 2001 From: laddyr141-ui Date: Mon, 27 Jul 2026 11:17:29 +0000 Subject: [PATCH 3/5] fix(indexer): add HTTP and per-read timeouts to the RPC client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reqwest client was built with reqwest::Client::new() — no timeout or connect_timeout. A stalled RPC (no response, half-open connection) made the await in a read hang indefinitely, freezing the single-task refresh loop and serving an increasingly stale snapshot with no recovery. Build the client with .timeout()/.connect_timeout(), and additionally wrap each attempt in tokio::time::timeout as a second bound. A timeout is now a distinct IndexError::Timeout variant, treated as transient so it participates in the existing retry/backoff logic rather than immediately failing the read. Co-Authored-By: Claude Sonnet 5 --- api/src/indexer/mod.rs | 42 +++++++++++++++++++++++++++++++++++------- 1 file changed, 35 insertions(+), 7 deletions(-) diff --git a/api/src/indexer/mod.rs b/api/src/indexer/mod.rs index d61062d..fbe2786 100644 --- a/api/src/indexer/mod.rs +++ b/api/src/indexer/mod.rs @@ -49,6 +49,14 @@ const MAX_CONCURRENT_ASSETS: usize = 8; /// same reason. const MAX_CONCURRENT_ADDRESS_READS: usize = 16; +/// Ceiling on a single RPC HTTP request (connecting, sending, and reading the +/// full response). Without this, a stalled RPC node (no response, half-open +/// connection) hangs the `await` forever, freezing the single-task refresh +/// loop and leaving the API serving an increasingly stale snapshot. +const REQUEST_TIMEOUT: Duration = Duration::from_secs(8); +/// Ceiling on establishing the TCP/TLS connection to the RPC endpoint. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); + /// Static configuration for a network's contracts and RPC endpoint. #[derive(Debug, Clone)] pub struct Config { @@ -175,6 +183,8 @@ pub enum IndexError { HttpStatus { status: u16, body: String }, #[error("rate limited or unavailable (status {status}); retry after {retry_after:?}")] RateLimited { status: u16, retry_after: Option, body: String }, + #[error("rpc read timed out after {0:?}")] + Timeout(Duration), } impl IndexError { @@ -183,7 +193,10 @@ impl IndexError { /// transient. XDR, strkey and decode errors stem from our own request or /// response handling and will fail identically on every attempt. fn is_transient(&self) -> bool { - matches!(self, IndexError::Http(_) | IndexError::Rpc(_)) + matches!( + self, + IndexError::Http(_) | IndexError::Rpc(_) | IndexError::Timeout(_) + ) } } @@ -237,11 +250,12 @@ struct ReadOutcome { impl Rpc { fn new(url: String, source: String) -> Self { - Rpc { - http: reqwest::Client::new(), - url, - source, - } + let http = reqwest::Client::builder() + .timeout(REQUEST_TIMEOUT) + .connect_timeout(CONNECT_TIMEOUT) + .build() + .expect("rpc client config (timeouts only) is always valid"); + Rpc { http, url, source } } /// Simulate `contract.method(args)` and decode the return value to JSON. @@ -261,7 +275,20 @@ impl Rpc { let mut attempt = 0; loop { attempt += 1; - match self.read_once(contract, method, args.clone()).await { + // Belt-and-suspenders alongside the client's own request timeout: + // this also bounds time spent should read_once ever grow a step + // that isn't covered by the HTTP client (e.g. future retries or + // processing added around the request). + let outcome = match tokio::time::timeout( + REQUEST_TIMEOUT, + self.read_once(contract, method, args.clone()), + ) + .await + { + Ok(result) => result, + Err(_) => Err(IndexError::Timeout(REQUEST_TIMEOUT)), + }; + match outcome { Ok(outcome) => return Ok(outcome), Err(e) if attempt < MAX_READ_ATTEMPTS && e.is_transient() => { let delay = retry_delay(attempt); @@ -965,6 +992,7 @@ mod tests { #[test] fn only_http_and_rpc_errors_are_transient() { assert!(IndexError::Rpc("busy".into()).is_transient()); + assert!(IndexError::Timeout(Duration::from_secs(1)).is_transient()); assert!(!IndexError::Decode("bad json".into()).is_transient()); assert!(!IndexError::Xdr("bad xdr".into()).is_transient()); assert!(!IndexError::Strkey("bad key".into()).is_transient()); From 9b76c5df19916d01c96a28a9a3540f7a188306f2 Mon Sep 17 00:00:00 2001 From: laddyr141-ui Date: Mon, 27 Jul 2026 11:18:50 +0000 Subject: [PATCH 4/5] fix(indexer): make per-asset indexing best-effort MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit refresh() used `?` on the per-asset metadata/holders reads, so a single asset's token contract failing to decode or an RPC read erroring (after retries) failed the whole refresh and left every asset stale — not just the bad one. index_dividends was already resilient (unwrap_or_default); the rest wasn't, so recovery behavior was inconsistent. A failing asset is now collected as a per-asset error (logged and counted via rwa_indexer_asset_read_errors_total{read="asset"}) and falls back to that asset's data from the previous snapshot when one exists, rather than aborting the refresh. Every other asset still publishes fresh. An asset with no prior snapshot (e.g. first refresh after it's registered) and no successful read is omitted for that cycle rather than fabricated. Co-Authored-By: Claude Sonnet 5 --- api/src/indexer/mod.rs | 69 ++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 66 insertions(+), 3 deletions(-) diff --git a/api/src/indexer/mod.rs b/api/src/indexer/mod.rs index fbe2786..43370e5 100644 --- a/api/src/indexer/mod.rs +++ b/api/src/indexer/mod.rs @@ -664,8 +664,16 @@ impl Indexer { } /// Read the full current state of all contracts and rebuild the snapshot. + /// + /// Per-asset indexing is best-effort: a failing asset (bad token + /// contract, an RPC error that exhausted retries, …) doesn't fail the + /// whole refresh. It falls back to that asset's data from the previous + /// snapshot if there is one — logged and counted via + /// `rwa_indexer_asset_read_errors_total{read="asset"}` as a last-good + /// marker — so every other asset still gets published with fresh data. async fn refresh(&self) -> Result { let cfg = &self.state.config; + let previous = self.state.snapshot(); let entries_read = self .rpc @@ -694,14 +702,61 @@ impl Indexer { let mut dividends_map: HashMap> = HashMap::new(); let mut total_distributions = 0usize; let mut tvl: i128 = 0; + let mut failed_assets = 0usize; - for (_, result) in tagged { - let AssetIndexed { + for (idx, result) in tagged { + let asset_id = raw_entries[idx].id; + let indexed = match result { + Ok(indexed) => Some(indexed), + Err(e) => { + failed_assets += 1; + record_asset_read_error(asset_id, "asset"); + match previous.asset(asset_id) { + Some(prev) => { + tracing::warn!( + asset_id, + error = %e, + "asset read failed; serving last-good snapshot for this asset" + ); + Some(AssetIndexed { + asset: prev.clone(), + holders: previous + .holders + .get(&asset_id) + .cloned() + .unwrap_or_default(), + compliance: previous + .compliance + .get(&asset_id) + .cloned() + .unwrap_or_default(), + dividends: previous + .dividends + .get(&asset_id) + .cloned() + .unwrap_or_default(), + }) + } + None => { + tracing::warn!( + asset_id, + error = %e, + "asset read failed with no prior snapshot; omitting from this refresh" + ); + None + } + } + } + }; + let Some(AssetIndexed { asset, holders, compliance, dividends, - } = result?; + }) = indexed + else { + continue; + }; total_distributions += dividends.len(); if asset.active { @@ -714,6 +769,14 @@ impl Indexer { assets.push(asset); } + if failed_assets > 0 { + tracing::warn!( + failed_assets, + total_assets = raw_entries.len(), + "refresh completed with some assets failing; published with best-effort/last-good data" + ); + } + let active_assets = assets.iter().filter(|a| a.active).count(); let mut distinct_holders = HashSet::new(); for holders in holders_map.values() { From 5b0d8375b79f7157ef9302c59ce9eeea8c7a497a Mon Sep 17 00:00:00 2001 From: laddyr141-ui Date: Mon, 27 Jul 2026 11:19:30 +0000 Subject: [PATCH 5/5] feat(api): add /ready endpoint with a snapshot-staleness threshold health returned {"status":"ok"} unconditionally: on boot the snapshot is Snapshot::default() (empty, last_updated: None), and if the indexer never succeeds the API still reported healthy while serving empty data. There was no readiness signal or staleness threshold. Add GET /ready: 200 once the snapshot has been updated within the last STALE_AFTER_SECS (3 missed poll intervals), 503 otherwise, with the snapshot age and threshold in the body. /health stays a pure liveness probe (process is up); /ready reflects whether the served data is trustworthy. Co-Authored-By: Claude Sonnet 5 --- api/src/routes/mod.rs | 46 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 45 insertions(+), 1 deletion(-) diff --git a/api/src/routes/mod.rs b/api/src/routes/mod.rs index edfe9c3..2c2872c 100644 --- a/api/src/routes/mod.rs +++ b/api/src/routes/mod.rs @@ -28,6 +28,11 @@ use crate::models::ApiErrorBody; const RATE_LIMIT_PER_SECOND: u64 = 5; const RATE_LIMIT_BURST: u32 = 20; +/// How many missed poll intervals before the snapshot is considered stale +/// for readiness purposes. A single slow/failed refresh shouldn't flip +/// readiness — a run of them should. +const STALE_AFTER_SECS: u64 = POLL_INTERVAL.as_secs() * 3; + /// Errors surfaced to API clients as a JSON body with an appropriate status. #[derive(Debug)] pub enum ApiError { @@ -85,6 +90,7 @@ pub fn router(state: AppState) -> Router { Router::new() .route("/", get(index)) .route("/health", get(health)) + .route("/ready", get(ready)) .route("/metrics", get(metrics)) .merge(data_routes) .with_state(state) @@ -145,17 +151,55 @@ async fn index() -> Json { "GET /assets/:id/compliance", "GET /assets/:id/dividends", "GET /health", + "GET /ready", "GET /metrics" ], "docs": "https://github.com/your-org/stellar-rwa-api-docs" })) } -/// Liveness probe. +/// Liveness probe: the process is up and serving HTTP. Always 200 — it does +/// not reflect whether the served data is fresh or even present, see +/// `/ready` for that. async fn health() -> Json { Json(json!({ "status": "ok" })) } +/// Readiness probe: 200 once the indexer has produced a snapshot within the +/// last [`STALE_AFTER_SECS`], 503 otherwise. On boot the snapshot is +/// `Snapshot::default()` (`last_updated: None`), which is correctly +/// unready rather than reported as healthy with empty data. +async fn ready(State(state): State) -> impl IntoResponse { + let stats = state.snapshot().stats; + let age_seconds = stats + .last_updated + .as_deref() + .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) + .map(|last| { + chrono::Utc::now() + .signed_duration_since(last) + .num_seconds() + .max(0) as u64 + }); + + let is_ready = age_seconds.is_some_and(|age| age <= STALE_AFTER_SECS); + let status = if is_ready { + StatusCode::OK + } else { + StatusCode::SERVICE_UNAVAILABLE + }; + + ( + status, + Json(json!({ + "ready": is_ready, + "last_updated": stats.last_updated, + "age_seconds": age_seconds, + "stale_after_seconds": STALE_AFTER_SECS, + })), + ) +} + /// Prometheus scrape endpoint: indexer refresh latency, failure counts, last /// success timestamp, and per-asset read errors. async fn metrics(State(state): State) -> impl IntoResponse {