From d96144e1e36410719ba6926d612833647fb95427 Mon Sep 17 00:00:00 2001 From: euniceamoni Date: Mon, 27 Jul 2026 08:26:12 +0000 Subject: [PATCH] feat(api): add per-asset indexed_at_ledger and index_error (#91) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add `indexed_at_ledger: u32` and `index_error: Option` to the Asset model so consumers can tell which assets are stale. - Make per-asset indexing fully best-effort: a failed get_metadata or compliance read no longer aborts the whole refresh cycle. The previous snapshot's data is re-emitted with index_error set instead. - indexed_at_ledger is taken from the latest_ledger returned by the per-asset get_metadata simulation — the tightest freshness signal available without a database. - Dividend fallback now also re-uses previous data rather than Vec::new(). Closes #91 --- api/src/indexer/mod.rs | 115 ++++++++++++++++++++++++++++++++++++++--- api/src/models/mod.rs | 6 +++ 2 files changed, 114 insertions(+), 7 deletions(-) diff --git a/api/src/indexer/mod.rs b/api/src/indexer/mod.rs index cd4d5e6..e1aa9b8 100644 --- a/api/src/indexer/mod.rs +++ b/api/src/indexer/mod.rs @@ -617,6 +617,10 @@ impl Indexer { let raw_entries: Vec = serde_json::from_value(entries_read.value) .map_err(|e| IndexError::Decode(e.to_string()))?; + // Grab the previous snapshot so we can carry forward per-asset + // freshness info for assets that fail this cycle. + let prev = self.state.snapshot(); + let mut assets = Vec::new(); let mut holders_map: HashMap> = HashMap::new(); let mut compliance_map: HashMap = HashMap::new(); @@ -625,25 +629,120 @@ impl Indexer { let mut tvl: i128 = 0; for raw in &raw_entries { - let meta = self + // ── per-asset metadata + compliance reads (best-effort) ─────────── + // A failure here is recorded and the asset is emitted with its + // previous data (if any) plus an `index_error`. This mirrors the + // dividend handling below and means one broken asset contract can + // never abort the whole refresh cycle. + let meta_read = 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()))?; + .inspect_err(|_| record_asset_read_error(raw.id, "get_metadata")); + + let meta_read = match meta_read { + Ok(r) => r, + Err(e) => { + tracing::warn!( + asset_id = raw.id, + error = %e, + "get_metadata failed; keeping previous asset data" + ); + // Re-emit the previous Asset (if we have one) with the + // updated error, so consumers can detect staleness. + if let Some(prev_asset) = prev.asset(raw.id) { + let mut stale = prev_asset.clone(); + stale.index_error = Some(e.to_string()); + if let Some(prev_holders) = prev.holders.get(&raw.id) { + holders_map.insert(raw.id, prev_holders.clone()); + total_distributions += prev + .dividends + .get(&raw.id) + .map(|d| d.len()) + .unwrap_or(0); + if stale.active { + tvl += parse_i128(&stale.valuation_cents); + } + } + if let Some(prev_comp) = prev.compliance.get(&raw.id) { + compliance_map.insert(raw.id, prev_comp.clone()); + } + if let Some(prev_dists) = prev.dividends.get(&raw.id) { + dividends_map.insert(raw.id, prev_dists.clone()); + } + assets.push(stale); + } + continue; + } + }; + + let asset_ledger = meta_read.latest_ledger; + let meta: RawMetadata = match serde_json::from_value(meta_read.value) { + Ok(m) => m, + Err(e) => { + let err = IndexError::Decode(e.to_string()); + record_asset_read_error(raw.id, "get_metadata"); + tracing::warn!( + asset_id = raw.id, + error = %err, + "metadata decode failed; keeping previous asset data" + ); + if let Some(prev_asset) = prev.asset(raw.id) { + let mut stale = prev_asset.clone(); + stale.index_error = Some(err.to_string()); + if stale.active { + tvl += parse_i128(&stale.valuation_cents); + } + if let Some(prev_holders) = prev.holders.get(&raw.id) { + holders_map.insert(raw.id, prev_holders.clone()); + } + if let Some(prev_comp) = prev.compliance.get(&raw.id) { + compliance_map.insert(raw.id, prev_comp.clone()); + } + if let Some(prev_dists) = prev.dividends.get(&raw.id) { + total_distributions += prev_dists.len(); + dividends_map.insert(raw.id, prev_dists.clone()); + } + assets.push(stale); + } + continue; + } + }; 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 + // Also best-effort: fall back to previous holders on failure. + let (holders, summary, compliance_err) = match self .index_compliance_and_holders( &meta.compliance_contract, &raw.token_contract, total_supply, ) - .await?; + .await + { + Ok(result) => (result.0, result.1, None), + Err(e) => { + record_asset_read_error(raw.id, "compliance"); + tracing::warn!( + asset_id = raw.id, + error = %e, + "compliance/holders read failed; using previous data" + ); + let holders = prev + .holders + .get(&raw.id) + .cloned() + .unwrap_or_default(); + let summary = prev + .compliance + .get(&raw.id) + .cloned() + .unwrap_or_default(); + (holders, summary, Some(e.to_string())) + } + }; // Dividends for this asset token; treated as empty on failure so one // asset's dividend contract issue doesn't abort the whole refresh. @@ -652,7 +751,7 @@ impl Indexer { 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() + prev.dividends.get(&raw.id).cloned().unwrap_or_default() } }; total_distributions += dists.len(); @@ -678,6 +777,8 @@ impl Indexer { paused: meta.paused, compliance_contract: meta.compliance_contract, created_at_ledger: raw.created_at, + indexed_at_ledger: asset_ledger, + index_error: compliance_err, }; holders_map.insert(raw.id, holders); diff --git a/api/src/models/mod.rs b/api/src/models/mod.rs index 26f1a5a..1241ef5 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, + /// Ledger number at which this asset was last successfully indexed. + pub indexed_at_ledger: u32, + /// Non-null when the most recent per-asset index attempt failed. The + /// global fields on [`Stats`] still reflect the last successful full + /// refresh; this tells consumers which individual assets may be stale. + pub index_error: Option, } /// A single holder of an asset token.