diff --git a/api/Cargo.toml b/api/Cargo.toml index 0f0bee7..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" @@ -21,6 +22,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..43370e5 100644 --- a/api/src/indexer/mod.rs +++ b/api/src/indexer/mod.rs @@ -17,6 +17,9 @@ 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; use reqwest::StatusCode; use serde::Deserialize; @@ -38,6 +41,22 @@ 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; + +/// 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 { @@ -137,6 +156,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)); } @@ -158,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 { @@ -166,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(_) + ) } } @@ -220,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. @@ -244,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); @@ -574,6 +618,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; @@ -606,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 @@ -617,75 +683,100 @@ 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; + let mut failed_assets = 0usize; - 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, + for (idx, result) in tagged { + let asset_id = raw_entries[idx].id; + let indexed = match result { + Ok(indexed) => Some(indexed), 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() + 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 + } + } } }; - total_distributions += dists.len(); + let Some(AssetIndexed { + asset, + holders, + compliance, + dividends, + }) = indexed + else { + continue; + }; - if raw.active { - tvl += valuation; + 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); } + 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() { @@ -716,6 +807,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( @@ -731,54 +888,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(), @@ -800,6 +951,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 @@ -870,6 +1055,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()); 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() { 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 {