diff --git a/api/Cargo.toml b/api/Cargo.toml index 0f0bee7..fceeb48 100644 --- a/api/Cargo.toml +++ b/api/Cargo.toml @@ -8,12 +8,15 @@ license = "Apache-2.0" [dependencies] axum = "0.7" tokio = { version = "1", features = ["full"] } -tower-http = { version = "0.5", features = ["cors", "trace"] } +tower = { version = "0.5", default-features = false, features = ["util", "limit", "timeout", "load-shed"] } +tower-http = { version = "0.5", features = ["cors", "trace", "limit"] } tower_governor = "0.5" reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } serde = { version = "1", features = ["derive"] } serde_json = "1" arc-swap = "1" +metrics = "0.24" +metrics-exporter-prometheus = { version = "0.18", default-features = false } stellar-xdr = { version = "23", default-features = false, features = ["curr", "std", "base64"] } stellar-strkey = "0.0.13" thiserror = "1" diff --git a/api/src/indexer/mod.rs b/api/src/indexer/mod.rs index cd4d5e6..10c348b 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 metrics_exporter_prometheus::PrometheusHandle; use reqwest::header::RETRY_AFTER; use reqwest::StatusCode; use serde::Deserialize; @@ -112,6 +113,13 @@ impl Snapshot { pub fn asset(&self, id: u64) -> Option<&Asset> { self.assets.iter().find(|a| a.id == id) } + + /// Whether the indexer has completed at least one successful refresh. + /// Before that, every field is a zero-value default rather than real + /// data, so routes should surface this instead of serving it silently. + pub fn is_ready(&self) -> bool { + self.stats.last_updated.is_some() + } } /// Shared, hot-swappable state handed to the Axum routes. @@ -137,6 +145,12 @@ impl AppState { (*guard).clone() } + /// The last successfully indexed ledger, without cloning the full + /// snapshot. Used to derive the cache-control ETag on every request. + pub fn last_indexed_ledger(&self) -> u32 { + self.inner.load().stats.last_indexed_ledger + } + fn replace(&self, next: Snapshot) { self.inner.store(Arc::new(next)); } @@ -535,10 +549,36 @@ struct RawDistribution { completed: bool, } +/// Parse a decimal i128 string, defaulting to 0 on failure without logging. +/// Only for re-parsing strings this process already produced and validated +/// (e.g. sorting by a balance we just formatted), where failure isn't a real +/// possibility. fn parse_i128(s: &str) -> i128 { s.parse::().unwrap_or(0) } +/// Parse a decimal i128 string that came straight off a contract read. A +/// failure here means real on-chain data (supply, valuation, a balance, a +/// distribution amount) silently became zero, understating whatever it feeds +/// into -- so it's logged loudly, and the second return value tells the +/// caller to flag the containing asset as having a decode error rather than +/// treating the 0 as trustworthy. +fn parse_i128_flagged(s: &str, asset_id: u64, field: &str) -> (i128, bool) { + match s.parse::() { + Ok(v) => (v, false), + Err(e) => { + tracing::warn!( + asset_id, + field, + raw = s, + error = %e, + "failed to parse i128 from contract data; defaulting to 0" + ); + (0, true) + } + } +} + fn cents_to_usd(cents: i128) -> f64 { cents as f64 / 100.0 } @@ -633,12 +673,15 @@ impl Indexer { 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 (total_supply, total_supply_err) = + parse_i128_flagged(&meta.total_supply, raw.id, "total_supply"); + let (valuation, valuation_err) = + parse_i128_flagged(&raw.valuation, raw.id, "valuation"); // Holders: every allowlisted address with a positive balance. - let (holders, summary, _) = self + let (holders, summary, _, holders_decode_err) = self .index_compliance_and_holders( + raw.id, &meta.compliance_contract, &raw.token_contract, total_supply, @@ -647,14 +690,19 @@ impl Indexer { // 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() - } - }; + let (dists, dividends_decode_err) = + match self.index_dividends(raw.id, &raw.token_contract).await { + Ok(result) => result, + 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(), false) + } + }; total_distributions += dists.len(); if raw.active { @@ -678,6 +726,10 @@ impl Indexer { paused: meta.paused, compliance_contract: meta.compliance_contract, created_at_ledger: raw.created_at, + decode_errors: total_supply_err + || valuation_err + || holders_decode_err + || dividends_decode_err, }; holders_map.insert(raw.id, holders); @@ -720,10 +772,11 @@ impl Indexer { /// list (allowlisted ∩ positive balance) and the non-PII summary. async fn index_compliance_and_holders( &self, + asset_id: u64, compliance_contract: &str, token_contract: &str, total_supply: i128, - ) -> Result<(Vec, ComplianceSummary, Vec), IndexError> { + ) -> Result<(Vec, ComplianceSummary, Vec, bool), IndexError> { let allowlist = self .rpc .read(compliance_contract, "get_allowlist", vec![]) @@ -735,6 +788,7 @@ impl Indexer { let mut summary = ComplianceSummary::default(); let mut jurisdictions: BTreeMap = BTreeMap::new(); let mut approved_addresses = Vec::new(); + let mut decode_error = false; for address in &addresses { summary.total_records += 1; @@ -775,7 +829,11 @@ impl Indexer { .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::String(s) => { + let (v, err) = parse_i128_flagged(&s, asset_id, "holder_balance"); + decode_error |= err; + v + } serde_json::Value::Number(n) => n.as_i64().unwrap_or(0) as i128, _ => 0, }; @@ -797,11 +855,15 @@ impl Indexer { }) .collect(); - Ok((holders, summary, approved_addresses)) + Ok((holders, summary, approved_addresses, decode_error)) } /// Read all distributions for an asset token from the dividend contract. - async fn index_dividends(&self, token_contract: &str) -> Result, IndexError> { + async fn index_dividends( + &self, + asset_id: u64, + token_contract: &str, + ) -> Result<(Vec, bool), IndexError> { let read = self .rpc .read( @@ -812,11 +874,15 @@ impl Indexer { .await?; let raw: Vec = serde_json::from_value(read.value).map_err(|e| IndexError::Decode(e.to_string()))?; - Ok(raw + let mut decode_error = false; + let dists = raw .into_iter() .map(|d| { - let total = parse_i128(&d.total_amount); - let distributed = parse_i128(&d.distributed); + let (total, total_err) = + parse_i128_flagged(&d.total_amount, asset_id, "distribution_total_amount"); + let (distributed, distributed_err) = + parse_i128_flagged(&d.distributed, asset_id, "distribution_distributed"); + decode_error |= total_err || distributed_err; Distribution { id: d.id, asset_token: d.asset_token, @@ -829,7 +895,8 @@ impl Indexer { created_at_ledger: d.created_at, } }) - .collect()) + .collect(); + Ok((dists, decode_error)) } } @@ -887,6 +954,12 @@ mod tests { assert_eq!(ratio_percent(150, 100), 100.0); } + #[test] + fn parse_i128_flagged_reports_decode_failures() { + assert_eq!(parse_i128_flagged("500000000", 1, "total_supply"), (500_000_000, false)); + assert_eq!(parse_i128_flagged("not-a-number", 1, "total_supply"), (0, true)); + } + #[test] fn normalizes_unit_enum_status() { // Soroban encodes a unit-variant enum as a vec of one symbol. diff --git a/api/src/main.rs b/api/src/main.rs index 21c5f7a..7612f3d 100644 --- a/api/src/main.rs +++ b/api/src/main.rs @@ -11,6 +11,8 @@ mod routes; use std::net::SocketAddr; +use metrics_exporter_prometheus::PrometheusBuilder; + use indexer::{AppState, Config, ConfigError, Indexer}; #[tokio::main] @@ -70,12 +72,34 @@ async fn main() { tracing::info!("shut down cleanly"); } -/// Resolve when the process receives Ctrl-C, for graceful shutdown. +/// Resolve on Ctrl-C or SIGTERM (the signal container runtimes send on +/// `docker stop`/pod termination), for graceful shutdown. async fn shutdown_signal() { - if let Err(e) = tokio::signal::ctrl_c().await { - tracing::error!(error = %e, "failed to install Ctrl-C handler"); + let ctrl_c = async { + if let Err(e) = tokio::signal::ctrl_c().await { + tracing::error!(error = %e, "failed to install Ctrl-C handler"); + } + }; + + #[cfg(unix)] + let terminate = async { + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) { + Ok(mut sig) => { + sig.recv().await; + } + Err(e) => { + tracing::error!(error = %e, "failed to install SIGTERM handler"); + std::future::pending::<()>().await; + } + } + }; + #[cfg(not(unix))] + let terminate = std::future::pending::<()>(); + + tokio::select! { + _ = ctrl_c => tracing::info!("shutdown signal received (ctrl-c)"), + _ = terminate => tracing::info!("shutdown signal received (SIGTERM)"), } - tracing::info!("shutdown signal received"); } fn init_tracing() { diff --git a/api/src/models/mod.rs b/api/src/models/mod.rs index 26f1a5a..40847bf 100644 --- a/api/src/models/mod.rs +++ b/api/src/models/mod.rs @@ -31,6 +31,12 @@ pub struct Asset { pub paused: bool, pub compliance_contract: String, pub created_at_ledger: u32, + /// True if any of this asset's raw numeric fields -- valuation, supply, + /// a holder balance, or a distribution amount -- failed to decode as an + /// integer and was defaulted to zero. When set, treat the numbers above + /// (and platform-wide TVL/holder counts they feed into) as a possible + /// understatement rather than exact. + pub decode_errors: bool, } /// A single holder of an asset token. diff --git a/api/src/routes/assets.rs b/api/src/routes/assets.rs index e8099b3..aeda922 100644 --- a/api/src/routes/assets.rs +++ b/api/src/routes/assets.rs @@ -25,8 +25,9 @@ pub struct AssetQuery { pub async fn list( State(state): State, Query(query): Query, -) -> Json> { +) -> Result>, ApiError> { let snap = state.snapshot(); + super::require_ready(&snap)?; let assets = snap .assets .into_iter() @@ -38,7 +39,7 @@ pub async fn list( }) .filter(|a| query.active.is_none_or(|active| a.active == active)) .collect(); - Json(assets) + Ok(Json(assets)) } /// Full detail for a single asset by its registry id. @@ -47,6 +48,7 @@ pub async fn detail( Path(id): Path, ) -> Result, ApiError> { let snap = state.snapshot(); + super::require_ready(&snap)?; snap.asset(id) .cloned() .map(Json) diff --git a/api/src/routes/compliance.rs b/api/src/routes/compliance.rs index e32010a..c4a04ce 100644 --- a/api/src/routes/compliance.rs +++ b/api/src/routes/compliance.rs @@ -15,6 +15,7 @@ pub async fn summary( Path(id): Path, ) -> Result, ApiError> { let snap = state.snapshot(); + super::require_ready(&snap)?; if snap.asset(id).is_none() { return Err(ApiError::NotFound(format!("no asset with id {id}"))); } diff --git a/api/src/routes/dividends.rs b/api/src/routes/dividends.rs index ac3e851..bd1f9de 100644 --- a/api/src/routes/dividends.rs +++ b/api/src/routes/dividends.rs @@ -15,6 +15,7 @@ pub async fn list( Path(id): Path, ) -> Result>, ApiError> { let snap = state.snapshot(); + super::require_ready(&snap)?; if snap.asset(id).is_none() { return Err(ApiError::NotFound(format!("no asset with id {id}"))); } diff --git a/api/src/routes/holders.rs b/api/src/routes/holders.rs index a01c90d..3dbf3b7 100644 --- a/api/src/routes/holders.rs +++ b/api/src/routes/holders.rs @@ -15,6 +15,7 @@ pub async fn list( Path(id): Path, ) -> Result>, ApiError> { let snap = state.snapshot(); + super::require_ready(&snap)?; if snap.asset(id).is_none() { return Err(ApiError::NotFound(format!("no asset with id {id}"))); } diff --git a/api/src/routes/mod.rs b/api/src/routes/mod.rs index edfe9c3..5dfe2af 100644 --- a/api/src/routes/mod.rs +++ b/api/src/routes/mod.rs @@ -7,37 +7,64 @@ pub mod holders; pub mod stats; use std::sync::Arc; +use std::time::Duration; use axum::{ body::Body, + error_handling::HandleErrorLayer, extract::{Request, State}, http::{header, HeaderMap, HeaderValue, StatusCode}, middleware::{self, Next}, response::{IntoResponse, Response}, routing::get, - Json, Router, + BoxError, Json, Router, }; use serde_json::json; +use tower::{ + limit::ConcurrencyLimitLayer, load_shed::LoadShedLayer, timeout::TimeoutLayer, ServiceBuilder, +}; use tower_governor::{governor::GovernorConfigBuilder, GovernorLayer}; -use tower_http::cors::{Any, CorsLayer}; +use tower_http::{ + cors::{Any, CorsLayer}, + limit::RequestBodyLimitLayer, +}; -use crate::indexer::{AppState, POLL_INTERVAL}; +use crate::indexer::{AppState, Snapshot, POLL_INTERVAL}; use crate::models::ApiErrorBody; /// Sustained requests-per-second allowed per client IP, with bursting. const RATE_LIMIT_PER_SECOND: u64 = 5; const RATE_LIMIT_BURST: u32 = 20; +/// Hard ceiling on how long a single request may take before it's aborted. +const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +/// Max requests handled concurrently across the whole server. Once this many +/// are in flight, `LoadShedLayer` rejects new requests with 503 immediately +/// instead of queuing them, so a burst of slow or abusive clients can't pile +/// up unbounded work. +const MAX_CONCURRENT_REQUESTS: usize = 256; +/// No route accepts a request body; cap it small rather than unbounded. +const MAX_REQUEST_BODY_BYTES: usize = 16 * 1024; + /// Errors surfaced to API clients as a JSON body with an appropriate status. #[derive(Debug)] pub enum ApiError { NotFound(String), + /// The indexer hasn't produced a usable snapshot (yet), or a downstream + /// dependency is temporarily unavailable. Clients should retry later. + ServiceUnavailable(String), + /// An unexpected, non-retryable server-side failure. + Internal(String), } impl IntoResponse for ApiError { fn into_response(self) -> Response { let (status, error, message) = match self { ApiError::NotFound(msg) => (StatusCode::NOT_FOUND, "not_found", msg), + ApiError::ServiceUnavailable(msg) => { + (StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", msg) + } + ApiError::Internal(msg) => (StatusCode::INTERNAL_SERVER_ERROR, "internal", msg), }; ( status, @@ -50,6 +77,19 @@ impl IntoResponse for ApiError { } } +/// Fail fast with 503 if the indexer hasn't completed its first successful +/// refresh yet, rather than letting routes serve an empty/zero-value +/// snapshot as if it were real data. +fn require_ready(snap: &Snapshot) -> Result<(), ApiError> { + if snap.is_ready() { + Ok(()) + } else { + Err(ApiError::ServiceUnavailable( + "indexer has not completed an initial refresh yet".into(), + )) + } +} + /// Build the application router with CORS enabled for the docs/web app. pub fn router(state: AppState) -> Router { let cors = CorsLayer::new() @@ -88,7 +128,28 @@ pub fn router(state: AppState) -> Router { .route("/metrics", get(metrics)) .merge(data_routes) .with_state(state) - .layer(cors) + .layer( + ServiceBuilder::new() + .layer(HandleErrorLayer::new(handle_middleware_error)) + .layer(TimeoutLayer::new(REQUEST_TIMEOUT)) + .layer(LoadShedLayer::new()) + .layer(ConcurrencyLimitLayer::new(MAX_CONCURRENT_REQUESTS)) + .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_BYTES)) + .layer(cors), + ) +} + +/// Converts errors from the timeout/load-shed layers into the same +/// structured JSON body the rest of the API uses, rather than letting tower +/// terminate the connection or return a bare status code. +async fn handle_middleware_error(err: BoxError) -> ApiError { + if err.is::() { + ApiError::ServiceUnavailable("request exceeded the timeout".into()) + } else if err.is::() { + ApiError::ServiceUnavailable("server is at capacity; try again shortly".into()) + } else { + ApiError::Internal(format!("unhandled middleware error: {err}")) + } } /// Attach `Cache-Control` and `ETag` to snapshot-backed responses, and answer @@ -98,7 +159,7 @@ pub fn router(state: AppState) -> Router { /// derived from `last_indexed_ledger`: two requests against the same indexed /// ledger are guaranteed to have identical bodies. async fn cache_headers(State(state): State, req: Request, next: Next) -> Response { - let ledger = state.last_indexed_ledger().await; + let ledger = state.last_indexed_ledger(); let etag = format!("\"ledger-{ledger}\""); let fresh = req diff --git a/api/src/routes/stats.rs b/api/src/routes/stats.rs index 2c6c72f..90aa4fc 100644 --- a/api/src/routes/stats.rs +++ b/api/src/routes/stats.rs @@ -2,11 +2,13 @@ use axum::{extract::State, Json}; +use super::ApiError; use crate::indexer::AppState; use crate::models::Stats; /// Platform-wide statistics: asset count, TVL, holders, distributions. -pub async fn get(State(state): State) -> Json { +pub async fn get(State(state): State) -> Result, ApiError> { let snap = state.snapshot(); - Json(snap.stats) + super::require_ready(&snap)?; + Ok(Json(snap.stats)) }