Fix/api hardening followups

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"

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::<i128>().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::<i128>() {
+        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<Holder>, ComplianceSummary, Vec<String>), IndexError> {
+    ) -> Result<(Vec<Holder>, ComplianceSummary, Vec<String>, 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<String, usize> = 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<Vec<Distribution>, IndexError> {
+    async fn index_dividends(
+        &self,
+        asset_id: u64,
+        token_contract: &str,
+    ) -> Result<(Vec<Distribution>, bool), IndexError> {
         let read = self
             .rpc
             .read(
@@ -812,11 +874,15 @@ impl Indexer {
             .await?;
         let raw: Vec<RawDistribution> =
             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.

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() {

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.

api/src/routes/assets.rs

@@ -25,8 +25,9 @@ pub struct AssetQuery {
 pub async fn list(
     State(state): State<AppState>,
     Query(query): Query<AssetQuery>,
-) -> Json<Vec<Asset>> {
+) -> Result<Json<Vec<Asset>>, 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<u64>,
 ) -> Result<Json<Asset>, ApiError> {
     let snap = state.snapshot();
+    super::require_ready(&snap)?;
     snap.asset(id)
         .cloned()
         .map(Json)

api/src/routes/compliance.rs

@@ -15,6 +15,7 @@ pub async fn summary(
     Path(id): Path<u64>,
 ) -> Result<Json<ComplianceSummary>, 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}")));
     }

api/src/routes/dividends.rs

@@ -15,6 +15,7 @@ pub async fn list(
     Path(id): Path<u64>,
 ) -> Result<Json<Vec<Distribution>>, 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}")));
     }

api/src/routes/holders.rs

@@ -15,6 +15,7 @@ pub async fn list(
     Path(id): Path<u64>,
 ) -> Result<Json<Vec<Holder>>, 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}")));
     }