From 885dde9534483ec06e1031d3fe5cd86c64ff9181 Mon Sep 17 00:00:00 2001 From: jakobsn Date: Fri, 3 Oct 2025 13:32:37 +0000 Subject: [PATCH] Resolve "Panic occurred: attempt to multiply with overflow" --- src/chain.rs | 2 +- src/db/cauldron/tokenlist/metrics_cache.rs | 1024 ++++++++++++-------- src/db/cauldron/tokenlist/test.rs | 268 ++++- src/db/cauldron/tokenlist/token_utils.rs | 145 ++- src/main.rs | 43 +- src/rpc/apy/poolperiod.rs | 85 +- 6 files changed, 1087 insertions(+), 480 deletions(-) diff --git a/src/chain.rs b/src/chain.rs index a86ed92..120a252 100644 --- a/src/chain.rs +++ b/src/chain.rs @@ -401,7 +401,7 @@ pub fn get_new_headers( let mut blockhash = *new_tip; while blockhash != null_hash { - if new_headers.len() % 1000 == 0 { + if new_headers.len().is_multiple_of(1000) { info!( "Downloading headers progress: {} fetched... ", new_headers.len() diff --git a/src/db/cauldron/tokenlist/metrics_cache.rs b/src/db/cauldron/tokenlist/metrics_cache.rs index 7ff6659..3e56062 100644 --- a/src/db/cauldron/tokenlist/metrics_cache.rs +++ b/src/db/cauldron/tokenlist/metrics_cache.rs @@ -11,12 +11,10 @@ use crate::db::cauldron::poolvisitor::{db_visit_pool_entries, PoolFilters}; use crate::db::cauldron::tokenlist::db_utils::{ cache_first_pool_ts_if_empty, db_first_pool_creation_row, }; -use crate::db::cauldron::tokenlist::token_utils::apy_30d_bp_for_token; -use crate::db::cauldron::tokenlist::token_utils::pct_change_bp_dec; -use crate::db::cauldron::tokenlist::token_utils::resolve_display_labels; use crate::db::cauldron::tokenlist::token_utils::{ - compute_score, pow10_dec, price_from_tvl, resolve_decimals, usd_per_bch_at_or_before, - SATS_PER_BCH, + apy_30d_bp_for_token, compute_score, dec_round, dec_to_f64_bounded, overflow_f64_fallback, + pct_change_bp_dec, pow10_dec, price_from_tvl, resolve_decimals, resolve_display_labels, + usd_per_bch_at_or_before, SATS_PER_BCH, }; use crate::db::DB; use crate::rpc::price::price_at_or_before_2; @@ -27,15 +25,52 @@ use bitcoin_hashes::hex::FromHex; use bitcoincash::TokenID; use log::{error, info, warn}; use std::collections::HashMap; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; -// ---------- 2 min: TVL + price_now (and price_now_usd) ---------- +const WRITE_CHUNK_RELEASE: usize = 300; +const WRITE_CHUNK: usize = WRITE_CHUNK_RELEASE; + +/// Retry a DB op on SQLITE_BUSY with jittered backoff (no extra deps). +fn with_busy_retry(mut f: F, max_wait: Duration) -> anyhow::Result +where + F: FnMut() -> rusqlite::Result, +{ + use rusqlite::{Error as SqliteError, ErrorCode as SqliteCode}; + let start = std::time::Instant::now(); + let mut backoff_ms: u64 = 25; + loop { + match f() { + Ok(v) => return Ok(v), + Err(SqliteError::SqliteFailure(err, _)) if err.code == SqliteCode::DatabaseBusy => { + if start.elapsed() >= max_wait { + return Err(SqliteError::SqliteFailure(err, Some("busy-timeout".into())).into()); + } + let seed_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_else(|_| Duration::from_millis(0)) + .subsec_millis() as u64; + let sleep_ms = (seed_ms % backoff_ms.max(1)).min(2000); + std::thread::sleep(Duration::from_millis(sleep_ms)); + backoff_ms = (backoff_ms * 2).min(2000); + continue; + } + Err(e) => return Err(e.into()), + } + } +} + +// ---------- 2.5 min: TVL + price_now (+ price_now_usd), chunked ---------- pub fn update_tvl_and_price_now( cauldron_conn: &Connection, bcmr_conn: &Connection, crc20_conn: &Connection, oracle_conn: &Connection, ) -> anyhow::Result<()> { - // TVL + // TVL snapshot (read-only) let mut tvl_vis = TvlByTokenVisitor::default(); db_visit_pool_entries(cauldron_conn, &mut tvl_vis, PoolFilters::new())?; let tvl_by_token = tvl_vis.into_map(); @@ -44,95 +79,203 @@ pub fn update_tvl_and_price_now( let sats_per_bch = Decimal::from_i64(SATS_PER_BCH).unwrap(); let usd_per_sat_now = usd_per_bch_at_or_before(oracle_conn, time_now()) / sats_per_bch; - let tx = cauldron_conn.unchecked_transaction()?; - { - // Mark all cached rows stale for this tick so we can drop zero-liquidity rows deterministically - tx.execute_batch( - r#" - UPDATE cached_token_metrics - SET tvl_sats = 0, - tvl_tokens = 0; - "#, - )?; + // Upsert in short chunks + let mut batch: Vec<(&str, i64, i64, f64, f64)> = Vec::with_capacity(WRITE_CHUNK); + for (token_id, (tvl_sats, tvl_tokens)) in tvl_by_token.iter() { + if *tvl_sats == 0 && *tvl_tokens == 0 { + continue; + } - let mut upsert = tx.prepare( - r#" - INSERT INTO cached_token_metrics - (token_id, trade_volume, tvl_sats, tvl_tokens, score, - display_name, display_symbol, - price_now, price_24h, price_7d, change_24h_bp, change_7d_bp, - price_now_usd, price_24h_usd, price_7d_usd, change_24h_usd_bp, change_7d_usd_bp, - apy_30d_bp, updated_at) - VALUES (?1, 0, ?2, ?3, 0, - NULL, NULL, - ?4, NULL, NULL, NULL, NULL, - ?5, NULL, NULL, NULL, NULL, - NULL, CAST(strftime('%s','now') AS INTEGER)) - ON CONFLICT(token_id) DO UPDATE SET - tvl_sats = excluded.tvl_sats, - tvl_tokens = excluded.tvl_tokens, - price_now = excluded.price_now, - price_now_usd = excluded.price_now_usd, -- ← only this USD field - updated_at = excluded.updated_at - "#, - )?; + let decimals = resolve_decimals(bcmr_conn, crc20_conn, token_id); + let factor = pow10_dec(decimals); - for (token_id, (tvl_sats, tvl_tokens)) in tvl_by_token { - // Skip on no liquidity - if tvl_sats == 0 && tvl_tokens == 0 { + let price_now_dec = price_from_tvl(*tvl_sats, *tvl_tokens).unwrap_or(Decimal::ZERO); + let price_now_human_dec = match price_now_dec.checked_mul(factor) { + Some(v) => v, + None => { + warn!( + "price_now overflow mul; token={} base={} factor=10^{} tvl_sats={} tvl_tokens={}", + token_id, price_now_dec, decimals, tvl_sats, tvl_tokens + ); + let price_now_human = overflow_f64_fallback(price_now_dec, decimals); + batch.push(( + token_id.as_str(), + *tvl_sats as i64, + *tvl_tokens as i64, + price_now_human, + 0.0_f64, + )); + if batch.len() >= WRITE_CHUNK { + flush_fast_batch(cauldron_conn, &batch)?; + batch.clear(); + } continue; } + }; - let decimals = resolve_decimals(bcmr_conn, crc20_conn, &token_id); - let factor = pow10_dec(decimals); + let price_now_human = dec_to_f64_bounded(dec_round(price_now_human_dec, 12)); + let price_now_usd = match price_now_human_dec.checked_mul(usd_per_sat_now) { + Some(d) => dec_to_f64_bounded(dec_round(d, 12)), + None => { + // Decimal overflow → fallback in f64 + let usdps_f = dec_to_f64_bounded(usd_per_sat_now); + let usd_f = price_now_human * usdps_f; + if usd_f.is_finite() { + usd_f + } else { + f64::MAX.copysign(usd_f) + } + } + }; - let price_now_dec = price_from_tvl(tvl_sats, tvl_tokens).unwrap_or(Decimal::ZERO); - let price_now_human = (price_now_dec * factor).to_f64().unwrap_or(0.0); - let price_now_usd = (price_now_dec * factor * usd_per_sat_now) - .to_f64() - .unwrap_or(0.0); - - upsert.execute(rusqlite::params![ - token_id, - tvl_sats as i64, - tvl_tokens as i64, - price_now_human, - price_now_usd, - ])?; + batch.push(( + token_id.as_str(), + *tvl_sats as i64, + *tvl_tokens as i64, + price_now_human, + price_now_usd, + )); + if batch.len() >= WRITE_CHUNK { + flush_fast_batch(cauldron_conn, &batch)?; + batch.clear(); } - // Remove rows that ended up with *no* liquidity (both sats AND tokens zero) - tx.execute( - "DELETE FROM cached_token_metrics WHERE tvl_sats = 0 AND tvl_tokens = 0", - [], - )?; } - tx.commit()?; + if !batch.is_empty() { + flush_fast_batch(cauldron_conn, &batch)?; + } + + // Drop rows that vanished in this snapshot (short tx + temp table) + flush_delete_absent(cauldron_conn, tvl_by_token.keys().map(|s| s.as_str()))?; + Ok(()) } +fn to_usd_f64_with_fallback( + human_dec_opt: Option, + usd_per_sat_dec: Decimal, + decimals: u32, +) -> Option { + human_dec_opt.map(|h| { + if let Some(usd_dec) = h.checked_mul(usd_per_sat_dec) { + dec_to_f64_bounded(dec_round(usd_dec, 12)) + } else { + let human_f = to_human_f64_with_fallback(h, decimals); + let usd_f = dec_to_f64_bounded(usd_per_sat_dec) * human_f; + if usd_f.is_finite() { + usd_f + } else { + f64::MAX.copysign(usd_f) + } + } + }) +} + +#[inline] +fn to_human_f64_with_fallback(base_dec: Decimal, decimals: u32) -> f64 { + if let Some(h) = base_dec.checked_mul(pow10_dec(decimals)) { + dec_to_f64_bounded(dec_round(h, 12)) + } else { + overflow_f64_fallback(base_dec, decimals) + } +} + +fn flush_fast_batch(conn: &Connection, rows: &[(&str, i64, i64, f64, f64)]) -> anyhow::Result<()> { + with_busy_retry( + || { + let tx = conn.unchecked_transaction()?; + { + let mut upsert = tx.prepare( + r#" + INSERT INTO cached_token_metrics + (token_id, trade_volume, tvl_sats, tvl_tokens, score, + display_name, display_symbol, + price_now, price_24h, price_7d, change_24h_bp, change_7d_bp, + price_now_usd, price_24h_usd, price_7d_usd, change_24h_usd_bp, change_7d_usd_bp, + apy_30d_bp, updated_at) + VALUES (?1, 0, ?2, ?3, 0, + NULL, NULL, + ?4, NULL, NULL, NULL, NULL, + ?5, NULL, NULL, NULL, NULL, + NULL, CAST(strftime('%s','now') AS INTEGER)) + ON CONFLICT(token_id) DO UPDATE SET + tvl_sats = excluded.tvl_sats, + tvl_tokens = excluded.tvl_tokens, + price_now = excluded.price_now, + price_now_usd = excluded.price_now_usd, + updated_at = excluded.updated_at + "#, + )?; + for (tid, sats, toks, p_now, p_now_usd) in rows { + upsert.execute(params![*tid, *sats, *toks, *p_now, *p_now_usd])?; + } + } // drop(upsert) + tx.commit()?; + Ok(()) + }, + Duration::from_secs(120), + ) +} + +fn flush_delete_absent<'a>( + conn: &Connection, + present_ids: impl Iterator, +) -> anyhow::Result<()> { + let ids: Vec = present_ids.map(|s| s.to_owned()).collect(); + with_busy_retry( + || { + let tx = conn.unchecked_transaction()?; + tx.execute_batch( + "CREATE TEMP TABLE IF NOT EXISTS __present (token_id TEXT PRIMARY KEY);", + )?; + tx.execute_batch("DELETE FROM __present;")?; + { + let mut ins = tx.prepare("INSERT INTO __present(token_id) VALUES (?1)")?; + for tid in &ids { + ins.execute(params![tid])?; + } + } // drop(ins) + tx.execute( + r#" + DELETE FROM cached_token_metrics + WHERE token_id NOT IN (SELECT token_id FROM __present) + "#, + [], + )?; + tx.commit()?; + Ok(()) + }, + Duration::from_secs(120), + ) +} + pub fn recompute_score_ranks(conn: &Connection) -> anyhow::Result<()> { - conn.execute_batch( - r#" - WITH ranked AS ( - SELECT token_id, - ROW_NUMBER() OVER ( - ORDER BY - score DESC, - trade_volume DESC, - tvl_sats DESC, - token_id ASC - ) AS rnk - FROM cached_token_metrics - WHERE tvl_sats > 0 - ) - UPDATE cached_token_metrics - SET score_rank = (SELECT rnk - FROM ranked - WHERE ranked.token_id = cached_token_metrics.token_id) - WHERE token_id IN (SELECT token_id FROM ranked); - "#, - )?; - Ok(()) + with_busy_retry( + || { + conn.execute_batch( + r#" + WITH ranked AS ( + SELECT token_id, + ROW_NUMBER() OVER ( + ORDER BY + score DESC, + trade_volume DESC, + tvl_sats DESC, + token_id ASC + ) AS rnk + FROM cached_token_metrics + WHERE tvl_sats > 0 + ) + UPDATE cached_token_metrics + SET score_rank = (SELECT rnk + FROM ranked + WHERE ranked.token_id = cached_token_metrics.token_id) + WHERE token_id IN (SELECT token_id FROM ranked); + "#, + )?; + Ok(()) + }, + Duration::from_secs(120), + ) } // ---------- 5 min: price changes (24h/7d), score, volume + ranking ---------- @@ -146,7 +289,7 @@ pub fn update_changes_score_volume_and_ranking( let now = time_now(); let since = (now - 30 * 24 * 60 * 60).max(0); - // volume per token (unchanged) + // volume per token let mut vol_stmt = cauldron_conn.prepare( r#" SELECT p.token_id, COALESCE(SUM(ABS(phe.sats_delta)), 0) AS vol @@ -165,7 +308,7 @@ pub fn update_changes_score_volume_and_ranking( vol_by_token.insert(token_id, vol.max(0) as u64); } - // define window anchors + // window anchors let ts_24h = now - 86_400; let ts_7d = now - 7 * 86_400; @@ -175,229 +318,323 @@ pub fn update_changes_score_volume_and_ranking( let usd_per_sat_24h = usd_per_bch_at_or_before(oracle_conn, ts_24h) / sats_per_bch; let usd_per_sat_7d = usd_per_bch_at_or_before(oracle_conn, ts_7d) / sats_per_bch; - let tx = cauldron_conn.unchecked_transaction()?; - { - let mut upd = tx.prepare( - r#" - UPDATE cached_token_metrics - SET trade_volume = ?2, - score = ?3, - - -- BCH/human historicals (nullable) - price_24h = ?4, - price_7d = ?5, - change_24h_bp = ?6, - change_7d_bp = ?7, - - -- USD now always set; historical USD nullable - price_now_usd = ?8, - price_24h_usd = ?9, - price_7d_usd = ?10, - change_24h_usd_bp = ?11, - change_7d_usd_bp = ?12, - - display_name = ?13, - display_symbol = ?14, - updated_at = CAST(strftime('%s','now') AS INTEGER) - WHERE token_id = ?1 - "#, - )?; - - // we need tvl & decimals for price_now and scaling - let mut cur_stmt = tx - .prepare("SELECT tvl_sats, tvl_tokens FROM cached_token_metrics WHERE token_id = ?1")?; - let mut id_stmt = - tx.prepare("SELECT token_id FROM cached_token_metrics WHERE tvl_sats > 0")?; - - let mut id_rows = id_stmt.query([])?; - let mut dec_cache: HashMap = HashMap::new(); - - while let Some(row) = id_rows.next()? { - let token_id: String = row.get(0)?; - - // decimals → factor - let decimals_u32 = *dec_cache - .entry(token_id.clone()) - .or_insert_with(|| resolve_decimals(bcmr_conn, crc20_conn, &token_id)); - let factor = pow10_dec(decimals_u32); - - // tvl - let (tvl_sats_i64, tvl_tokens_i64): (i64, i64) = - cur_stmt.query_row([&token_id], |r| Ok((r.get(0)?, r.get(1)?)))?; - let tvl_sats = tvl_sats_i64.max(0) as u64; - let tvl_tokens = tvl_tokens_i64.max(0) as u64; - - // price_now (base) & scaled - let price_now_dec = price_from_tvl(tvl_sats, tvl_tokens).unwrap_or(Decimal::ZERO); - let price_now_human_dec = price_now_dec * factor; - let price_now_usd_dec = price_now_human_dec * usd_per_sat_now; - - // volume/score - let vol = vol_by_token.get(&token_id).copied().unwrap_or(0); - let score = compute_score(tvl_sats, vol); - - // labels (unchanged) - let labels_opt = resolve_display_labels(bcmr_conn, crc20_conn, &token_id).ok(); - let (display_name, display_symbol): (Option, Option) = match labels_opt + // One write tx; scope statements so they drop before commit. + with_busy_retry( + || { + let tx = cauldron_conn.unchecked_transaction()?; { - Some((dn, sym)) => (Some(dn), Some(sym)), - None => (None, None), - }; + let mut upd = tx.prepare( + r#" + UPDATE cached_token_metrics + SET trade_volume = ?2, + score = ?3, - // Common conversions - let price_now_human_f = price_now_human_dec.to_f64().unwrap_or(0.0); - let price_now_usd_f = price_now_usd_dec.to_f64().unwrap_or(0.0); + price_24h = ?4, + price_7d = ?5, + change_24h_bp = ?6, + change_7d_bp = ?7, - // We'll need TokenID for historical probes - let tid = match TokenID::from_hex(&token_id) { - Ok(t) => t, - Err(_) => continue, - }; + price_now_usd = ?8, + price_24h_usd = ?9, + price_7d_usd = ?10, + change_24h_usd_bp = ?11, + change_7d_usd_bp = ?12, - if vol == 0 { - // --------- FAST PATH (no trades in 30d) ---------- - // Only gate by *existence* of a historical point; if none → keep NULLs. - let p24_exists = matches!( - price_at_or_before_2(cauldron_conn, ts_24h, &tid), - Ok((_ts, _p)) - ); - let p7d_exists = matches!( - price_at_or_before_2(cauldron_conn, ts_7d, &tid), - Ok((_ts, _p)) - ); + display_name = ?13, + display_symbol = ?14, + updated_at = CAST(strftime('%s','now') AS INTEGER) + WHERE token_id = ?1 + "#, + )?; - let price_24h_human_f = if p24_exists { - Some(price_now_human_f) - } else { - None - }; - let price_7d_human_f = if p7d_exists { - Some(price_now_human_f) - } else { - None - }; + // need tvl for price_now and scaling + let mut cur_stmt = tx.prepare( + "SELECT tvl_sats, tvl_tokens FROM cached_token_metrics WHERE token_id = ?1", + )?; + let mut id_stmt = + tx.prepare("SELECT token_id FROM cached_token_metrics WHERE tvl_sats > 0")?; - let d24_bp_opt = if p24_exists { Some(0) } else { None }; - let d7d_bp_opt = if p7d_exists { Some(0) } else { None }; + let mut id_rows = id_stmt.query([])?; + let mut dec_cache: HashMap = HashMap::new(); - let price_24h_usd_f = if p24_exists && !usd_per_sat_24h.is_zero() { - (price_now_human_dec * usd_per_sat_24h).to_f64() - } else { - None - }; - let price_7d_usd_f = if p7d_exists && !usd_per_sat_7d.is_zero() { - (price_now_human_dec * usd_per_sat_7d).to_f64() - } else { - None - }; + while let Some(row) = id_rows.next()? { + let token_id: String = row.get(0)?; - let d24_usd_bp_opt = price_24h_usd_f - .and_then(Decimal::from_f64) - .map(|old| pct_change_bp_dec(price_now_usd_dec, old)); - let d7d_usd_bp_opt = price_7d_usd_f - .and_then(Decimal::from_f64) - .map(|old| pct_change_bp_dec(price_now_usd_dec, old)); + // decimals → factor + let decimals_u32 = *dec_cache + .entry(token_id.clone()) + .or_insert_with(|| resolve_decimals(bcmr_conn, crc20_conn, &token_id)); + let factor = pow10_dec(decimals_u32); - upd.execute(params![ - token_id, - vol as i64, - score, - price_24h_human_f, - price_7d_human_f, - d24_bp_opt, - d7d_bp_opt, - price_now_usd_f, - price_24h_usd_f, - price_7d_usd_f, - d24_usd_bp_opt, - d7d_usd_bp_opt, - display_name, - display_symbol, - ])?; - continue; - } + // tvl + let (tvl_sats_i64, tvl_tokens_i64): (i64, i64) = + cur_stmt.query_row([&token_id], |r| Ok((r.get(0)?, r.get(1)?)))?; + let tvl_sats = tvl_sats_i64.max(0) as u64; + let tvl_tokens = tvl_tokens_i64.max(0) as u64; - // --------- FULL PATH (vol > 0) ---------- - let p24_dec_opt = price_at_or_before_2(cauldron_conn, ts_24h, &tid) - .ok() - .and_then(|(_, p)| Decimal::from_f64(p)); - let p7d_dec_opt = price_at_or_before_2(cauldron_conn, ts_7d, &tid) - .ok() - .and_then(|(_, p)| Decimal::from_f64(p)); + // price_now (base) & scaled with guards + let price_now_dec = + price_from_tvl(tvl_sats, tvl_tokens).unwrap_or(Decimal::ZERO); - // scale to “human” (nullable) - let price_24h_human_opt = p24_dec_opt.map(|p| p * factor); - let price_7d_human_opt = p7d_dec_opt.map(|p| p * factor); + let price_now_human_dec = match price_now_dec.checked_mul(factor) { + Some(v) => v, + None => { + warn!( + "core price_now overflow; token={}, base={}, factor=10^{}, tvl_sats={}, tvl_tokens={}", + token_id, price_now_dec, decimals_u32, tvl_sats, tvl_tokens + ); - // BCH deltas (nullable) - let d24_bp_opt = p24_dec_opt.map(|p| pct_change_bp_dec(price_now_dec, p)); - let d7d_bp_opt = p7d_dec_opt.map(|p| pct_change_bp_dec(price_now_dec, p)); + // Keep existing fallback behavior here. + let p_now_human_f = overflow_f64_fallback(price_now_dec, decimals_u32); + let p_now_usd_f = { + let usd_per_sat_f = dec_to_f64_bounded(usd_per_sat_now); + let usd = p_now_human_f * usd_per_sat_f; + if usd.is_finite() { + usd + } else { + f64::MAX.copysign(usd) + } + }; - // USD historicals (nullable if oracle missing) - let price_24h_usd_opt = match (price_24h_human_opt, !usd_per_sat_24h.is_zero()) { - (Some(h), true) => Some(h * usd_per_sat_24h), - _ => None, - }; - let price_7d_usd_opt = match (price_7d_human_opt, !usd_per_sat_7d.is_zero()) { - (Some(h), true) => Some(h * usd_per_sat_7d), - _ => None, - }; + let vol = vol_by_token.get(&token_id).copied().unwrap_or(0); + let score = compute_score(tvl_sats, vol); + let labels_opt = + resolve_display_labels(bcmr_conn, crc20_conn, &token_id).ok(); + let (display_name, display_symbol) = labels_opt + .map(|(dn, sym)| (Some(dn), Some(sym))) + .unwrap_or((None, None)); - let price_24h_human_f = price_24h_human_opt.and_then(|d| d.to_f64()); - let price_7d_human_f = price_7d_human_opt.and_then(|d| d.to_f64()); - let price_24h_usd_f = price_24h_usd_opt.and_then(|d| d.to_f64()); - let price_7d_usd_f = price_7d_usd_opt.and_then(|d| d.to_f64()); + upd.execute(params![ + token_id, + vol as i64, + score, + Option::::None, // price_24h + Option::::None, // price_7d + Option::::None, // change_24h_bp + Option::::None, // change_7d_bp + p_now_usd_f, // fallback (finite) on this branch + Option::::None, // price_24h_usd + Option::::None, // price_7d_usd + Option::::None, // change_24h_usd_bp + Option::::None, // change_7d_usd_bp + display_name, + display_symbol, + ])?; + continue; + } + }; - let d24_usd_bp_opt = match (Some(price_now_usd_dec), price_24h_usd_opt) { - (Some(now_u), Some(old_u)) => Some(pct_change_bp_dec(now_u, old_u)), - _ => None, - }; - let d7d_usd_bp_opt = match (Some(price_now_usd_dec), price_7d_usd_opt) { - (Some(now_u), Some(old_u)) => Some(pct_change_bp_dec(now_u, old_u)), - _ => None, - }; + // NOTE: If multiplying human * usd_per_sat overflows, write NULL (unknown). + let price_now_usd_dec_opt = price_now_human_dec.checked_mul(usd_per_sat_now); + let price_now_usd_f_opt = + price_now_usd_dec_opt.map(|d| dec_to_f64_bounded(dec_round(d, 12))); - upd.execute(params![ - token_id, - vol as i64, - score, - price_24h_human_f, - price_7d_human_f, - d24_bp_opt, - d7d_bp_opt, - price_now_usd_f, - price_24h_usd_f, - price_7d_usd_f, - d24_usd_bp_opt, - d7d_usd_bp_opt, - display_name, - display_symbol, - ])?; - } - } + // volume/score + let vol = vol_by_token.get(&token_id).copied().unwrap_or(0); + let score = compute_score(tvl_sats, vol); - // ranking - // ranking - tx.execute_batch( - r#" - WITH ranked AS ( - SELECT token_id, - ROW_NUMBER() OVER ( - ORDER BY - score DESC, - trade_volume DESC, - tvl_sats DESC, -- <-- same tie-breaker here - token_id ASC - ) AS rnk - FROM cached_token_metrics - WHERE tvl_sats > 0 - ) - UPDATE cached_token_metrics - SET score_rank = (SELECT rnk FROM ranked WHERE ranked.token_id = cached_token_metrics.token_id) - WHERE token_id IN (SELECT token_id FROM ranked); -"#, + // labels + let labels_opt = resolve_display_labels(bcmr_conn, crc20_conn, &token_id).ok(); + let (display_name, display_symbol): (Option, Option) = + match labels_opt { + Some((dn, sym)) => (Some(dn), Some(sym)), + None => (None, None), + }; + + // We'll need TokenID for historical probes + let tid = match TokenID::from_hex(&token_id) { + Ok(t) => t, + Err(_) => { + // still update the "now" fields and basics + upd.execute(params![ + token_id, + vol as i64, + score, + Option::::None, + Option::::None, + Option::::None, + Option::::None, + price_now_usd_f_opt, // ← may be NULL if overflowed + Option::::None, + Option::::None, + Option::::None, + Option::::None, + display_name, + display_symbol, + ])?; + continue; + } + }; + + if vol == 0 { + // --------- FAST PATH (no trades in 30d) ---------- + // Only gate by *existence* of a historical point; if none → keep NULLs + let p24_exists = matches!( + price_at_or_before_2(cauldron_conn, ts_24h, &tid), + Ok((_ts, _)) + ); + let p7d_exists = matches!( + price_at_or_before_2(cauldron_conn, ts_7d, &tid), + Ok((_ts, _)) + ); + + let price_24h_human_f = if p24_exists { + Some(dec_to_f64_bounded(dec_round(price_now_human_dec, 12))) + } else { + None + }; + + let price_7d_human_f = if p7d_exists { + Some(dec_to_f64_bounded(dec_round(price_now_human_dec, 12))) + } else { + None + }; + + let d24_bp_opt = if p24_exists { Some(0) } else { None }; + let d7d_bp_opt = if p7d_exists { Some(0) } else { None }; + + let price_24h_usd_f = if p24_exists && !usd_per_sat_24h.is_zero() { + price_now_human_dec + .checked_mul(usd_per_sat_24h) + .map(|d| dec_to_f64_bounded(dec_round(d, 12))) + } else { + None + }; + let price_7d_usd_f = if p7d_exists && !usd_per_sat_7d.is_zero() { + price_now_human_dec + .checked_mul(usd_per_sat_7d) + .map(|d| dec_to_f64_bounded(dec_round(d, 12))) + } else { + None + }; + + let old24_u_dec_opt = price_24h_usd_f.and_then(Decimal::from_f64); + let old7d_u_dec_opt = price_7d_usd_f.and_then(Decimal::from_f64); + + let d24_usd_bp_opt = match (price_now_usd_dec_opt, old24_u_dec_opt) { + (Some(now_u), Some(old_u)) => Some(pct_change_bp_dec(now_u, old_u)), + _ => None, + }; + let d7d_usd_bp_opt = match (price_now_usd_dec_opt, old7d_u_dec_opt) { + (Some(now_u), Some(old_u)) => Some(pct_change_bp_dec(now_u, old_u)), + _ => None, + }; + + upd.execute(params![ + token_id, + vol as i64, + score, + price_24h_human_f, + price_7d_human_f, + d24_bp_opt, + d7d_bp_opt, + price_now_usd_f_opt, // ← may be NULL + price_24h_usd_f, + price_7d_usd_f, + d24_usd_bp_opt, + d7d_usd_bp_opt, + display_name, + display_symbol, + ])?; + continue; + } + + // --------- FULL PATH (vol > 0) ---------- + let p24_dec_opt = price_at_or_before_2(cauldron_conn, ts_24h, &tid) + .ok() + .and_then(|(_, p)| Decimal::from_f64(p)); + let p7d_dec_opt = price_at_or_before_2(cauldron_conn, ts_7d, &tid) + .ok() + .and_then(|(_, p)| Decimal::from_f64(p)); + + // scale to “human” with overflow guard (nullable) + let price_24h_human_opt = p24_dec_opt.and_then(|p| p.checked_mul(factor)); + let price_7d_human_opt = p7d_dec_opt.and_then(|p| p.checked_mul(factor)); + + // BCH deltas (nullable) + let d24_bp_opt = p24_dec_opt.map(|p| pct_change_bp_dec(price_now_dec, p)); + let d7d_bp_opt = p7d_dec_opt.map(|p| pct_change_bp_dec(price_now_dec, p)); + + // USD historicals (nullable if oracle missing or overflow) + let price_24h_usd_opt = match (price_24h_human_opt, !usd_per_sat_24h.is_zero()) + { + (Some(h), true) => h.checked_mul(usd_per_sat_24h), + _ => None, + }; + let price_7d_usd_opt = match (price_7d_human_opt, !usd_per_sat_7d.is_zero()) { + (Some(h), true) => h.checked_mul(usd_per_sat_7d), + _ => None, + }; + + let price_24h_human_f = + p24_dec_opt.map(|p24| to_human_f64_with_fallback(p24, decimals_u32)); + let price_7d_human_f = + p7d_dec_opt.map(|p7d| to_human_f64_with_fallback(p7d, decimals_u32)); + let price_24h_usd_f = to_usd_f64_with_fallback( + price_24h_human_opt, + usd_per_sat_24h, + decimals_u32, + ); + let price_7d_usd_f = + to_usd_f64_with_fallback(price_7d_human_opt, usd_per_sat_7d, decimals_u32); + + let d24_usd_bp_opt = match (price_now_usd_dec_opt, price_24h_usd_opt) { + (Some(now_u), Some(old_u)) => Some(pct_change_bp_dec(now_u, old_u)), + _ => None, + }; + let d7d_usd_bp_opt = match (price_now_usd_dec_opt, price_7d_usd_opt) { + (Some(now_u), Some(old_u)) => Some(pct_change_bp_dec(now_u, old_u)), + _ => None, + }; + + upd.execute(params![ + token_id, + vol as i64, + score, + price_24h_human_f, + price_7d_human_f, + d24_bp_opt, + d7d_bp_opt, + price_now_usd_f_opt, // ← may be NULL on overflow + price_24h_usd_f, + price_7d_usd_f, + d24_usd_bp_opt, + d7d_usd_bp_opt, + display_name, + display_symbol, + ])?; + } // while + } // scope statements + + // ranking (after statements dropped) + tx.execute_batch( + r#" + WITH ranked AS ( + SELECT token_id, + ROW_NUMBER() OVER ( + ORDER BY + score DESC, + trade_volume DESC, + tvl_sats DESC, + token_id ASC + ) AS rnk + FROM cached_token_metrics + WHERE tvl_sats > 0 + ) + UPDATE cached_token_metrics + SET score_rank = (SELECT rnk + FROM ranked + WHERE ranked.token_id = cached_token_metrics.token_id) + WHERE token_id IN (SELECT token_id FROM ranked); + "#, + )?; + + tx.commit()?; + Ok(()) + }, + Duration::from_secs(120), )?; - tx.commit()?; + Ok(()) } @@ -417,11 +654,10 @@ pub fn update_apy_only(cauldron_conn: &Connection) -> anyhow::Result<()> { } } - // Compute/set APY; when vol==0, write 0 directly (no heavy compute) + // Compute/set APY; when vol==0, write 0 directly let mut apys = Vec::<(String, Option)>::with_capacity(ids.len()); for (token_id, vol30) in ids { let apy_opt = if vol30 == 0 { - // optimization: don't compute; by definition over 30d there were no trades → 0 Some(0) } else { match apy_30d_bp_for_token(cauldron_conn, &token_id, now) { @@ -435,151 +671,139 @@ pub fn update_apy_only(cauldron_conn: &Connection) -> anyhow::Result<()> { apys.push((token_id, apy_opt)); } - let tx = cauldron_conn.unchecked_transaction()?; - { - let mut upd = tx.prepare( - "UPDATE cached_token_metrics - SET apy_30d_bp = ?2, - updated_at = CAST(strftime('%s','now') AS INTEGER) - WHERE token_id = ?1", - )?; - for (tid, apy_opt) in apys { - upd.execute(params![tid, apy_opt])?; - } - } - tx.commit()?; + with_busy_retry( + || { + let tx = cauldron_conn.unchecked_transaction()?; + { + let mut upd = tx.prepare( + "UPDATE cached_token_metrics + SET apy_30d_bp = ?2, + updated_at = CAST(strftime('%s','now') AS INTEGER) + WHERE token_id = ?1", + )?; + for (tid, apy_opt) in apys.iter() { + upd.execute(params![tid, apy_opt])?; + } + } + tx.commit()?; + Ok(()) + }, + Duration::from_secs(120), + )?; + Ok(()) } -pub fn spawn_token_metrics_updater(db: DB) { +pub fn spawn_token_metrics_updater(db: DB, indexing_in_progress: Arc) { std::thread::spawn(move || { - use std::time::{Duration, Instant}; + use std::time::Instant; // ---- PRODUCTION CADENCE ---- const BASE_SECS: u64 = 150; // 2.5 min const CORE_SECS: u64 = 300; // 5 min const APY_SECS: u64 = 900; // 15 min - // ---- DEV QUICK CADENCE (uncomment to test locally) ---- - // const BASE_SECS: u64 = 60; - // const CORE_SECS: u64 = 120; - // const APY_SECS: u64 = 320; - let base_d = Duration::from_secs(BASE_SECS); let core_every = (CORE_SECS / BASE_SECS).max(1); let apy_every = (APY_SECS / BASE_SECS).max(1); - // anchor to a stable base schedule let mut tick: u64 = 0; let mut next_deadline = Instant::now(); // first tick runs immediately loop { - // Wait until the next base boundary (but don't sleep negative) + // Back off if the indexer is mutating the DB + if indexing_in_progress.load(Ordering::Relaxed) { + std::thread::sleep(Duration::from_millis(250)); + continue; + } + + // Wait until the next base boundary let now = Instant::now(); if now < next_deadline { std::thread::sleep(next_deadline - now); } - // advance schedule for the *next* loop before doing work (prevents drift) tick = tick.saturating_add(1); next_deadline += base_d; - // Acquire connections once per tick; reuse across the cascade - let cw = db.cauldron_w.get(); - let bcmr = db.bcmr_r.get(); - let crc = db.crc20_r.get(); - let orc = db.oracle_r.get(); - // ---------- FAST (every base tick): TVL + price_now ---------- - { - let t0: Instant = Instant::now(); - match (&cw, &bcmr, &crc, &orc) { - (Ok(cw), Ok(bcmr), Ok(crc), Ok(orc)) => { - match update_tvl_and_price_now(cw, bcmr, crc, orc) { - Ok(_) => info!("✅ fast(2.5m) TVL+price_now in {:.3?}", t0.elapsed()), - Err(e) => { - error!("❌ fast(2.5m) failed after {:.3?}: {e:?}", t0.elapsed()) - } - } - - // Compact ranks immediately on fast-only ticks (avoid gap after deletions) - if tick % core_every != 0 { - if let Err(e) = recompute_score_ranks(cw) { - error!("rank recompute (fast-only tick) failed: {e:?}"); - } - } - - // Batched, idempotent backfill (does nothing once populated) - match backfill_first_pool_ts_batch(cw, 500) { - Ok(n) if n > 0 => info!("first_pool_ts backfill: set {n} tokens"), - Ok(_) => {} - Err(e) => warn!("first_pool_ts backfill failed: {e:?}"), - } - } - _ => { - if let Err(e) = &cw { - error!("DB unavailable (cauldron_w fast): {e}"); - } - if let Err(e) = &bcmr { - error!("DB unavailable (bcmr_r fast): {e}"); - } - if let Err(e) = &crc { - error!("DB unavailable (crc20_r fast): {e}"); - } - if let Err(e) = &orc { - error!("DB unavailable (oracle_r fast): {e}"); - } - } - } + if indexing_in_progress.load(Ordering::Relaxed) { + continue; } + { + let t0 = Instant::now(); + if let (Ok(cw), Ok(bcmr), Ok(crc), Ok(orc)) = ( + db.cauldron_w.get(), + db.bcmr_r.get(), + db.crc20_r.get(), + db.oracle_r.get(), + ) { + match update_tvl_and_price_now(&cw, &bcmr, &crc, &orc) { + Ok(_) => info!("✅ fast(2.5m) TVL+price_now in {:.3?}", t0.elapsed()), + Err(e) => error!("❌ fast(2.5m) failed after {:.3?}: {e:?}", t0.elapsed()), + } + + if indexing_in_progress.load(Ordering::Relaxed) { + continue; + } + if !tick.is_multiple_of(core_every) { + if let Err(e) = recompute_score_ranks(&cw) { + error!("rank recompute (fast-only tick) failed: {e:?}"); + } + } + + if indexing_in_progress.load(Ordering::Relaxed) { + continue; + } + match backfill_first_pool_ts_batch(&cw, WRITE_CHUNK as i64) { + Ok(n) if n > 0 => info!("first_pool_ts backfill: set {n} tokens"), + Ok(_) => {} + Err(e) => warn!("first_pool_ts backfill failed: {e:?}"), + } + } else { + error!("Pool checkout failed for fast tick"); + } + } // pooled conns dropped here // ---------- CORE (every 2 base ticks): fast + core ---------- - if tick % core_every == 0 { + if tick.is_multiple_of(core_every) { + if indexing_in_progress.load(Ordering::Relaxed) { + continue; + } let t0 = Instant::now(); - match (&cw, &bcmr, &crc, &orc) { - (Ok(cw), Ok(bcmr), Ok(crc), Ok(orc)) => { - match update_changes_score_volume_and_ranking(cw, bcmr, crc, orc) { - Ok(_) => info!( - "✅ core(5m) changes+score+volume(+rank) in {:.3?}", - t0.elapsed() - ), - Err(e) => { - error!("❌ core(5m) failed after {:.3?}: {e:?}", t0.elapsed()) - } - } - } - _ => { - if let Err(e) = &cw { - error!("DB unavailable (cauldron_w core): {e}"); - } - if let Err(e) = &bcmr { - error!("DB unavailable (bcmr_r core): {e}"); - } - if let Err(e) = &crc { - error!("DB unavailable (crc20_r core): {e}"); - } - if let Err(e) = &orc { - error!("DB unavailable (oracle_r core): {e}"); - } + if let (Ok(cw), Ok(bcmr), Ok(crc), Ok(orc)) = ( + db.cauldron_w.get(), + db.bcmr_r.get(), + db.crc20_r.get(), + db.oracle_r.get(), + ) { + match update_changes_score_volume_and_ranking(&cw, &bcmr, &crc, &orc) { + Ok(_) => info!("✅ core(5m) in {:.3?}", t0.elapsed()), + Err(e) => error!("❌ core(5m) failed after {:.3?}: {e:?}", t0.elapsed()), } + } else { + error!("Pool checkout failed for core tick"); } } // ---------- APY (every 6 base ticks): fast + core + apy ---------- - if tick % apy_every == 0 { + if tick.is_multiple_of(apy_every) { + if indexing_in_progress.load(Ordering::Relaxed) { + continue; + } let t0 = Instant::now(); - match &cw { - Ok(cw) => match update_apy_only(cw) { + if let Ok(cw) = db.cauldron_w.get() { + match update_apy_only(&cw) { Ok(_) => info!("✅ apy(15m) in {:.3?}", t0.elapsed()), Err(e) => error!("❌ apy(15m) failed after {:.3?}: {e:?}", t0.elapsed()), - }, - Err(e) => error!("DB unavailable (cauldron_w apy): {e}"), + } + } else { + error!("Pool checkout failed for apy tick"); } } - // If we overran the next base boundary (work took > base), skip ahead - // to the next future boundary to avoid back-to-back catch-up storms. - let now2 = Instant::now(); + // Skip ahead if we overran the boundary + let now2 = std::time::Instant::now(); while now2 >= next_deadline { next_deadline += base_d; tick = tick.saturating_add(1); @@ -589,13 +813,13 @@ pub fn spawn_token_metrics_updater(db: DB) { } pub fn backfill_first_pool_ts_batch(conn: &Connection, limit: i64) -> anyhow::Result { - // 1) Read the batch WITHOUT keeping a borrow alive past this block + // 1) Read the batch WITHOUT holding borrows over the tx let token_ids: Vec = { let mut stmt = conn.prepare( r#" SELECT token_id FROM cached_token_metrics - WHERE first_pool_ts IS NULL OR first_pool_ts = 0 -- ← include 0 as “unset” + WHERE first_pool_ts IS NULL OR first_pool_ts = 0 ORDER BY token_id LIMIT ?1; "#, @@ -608,14 +832,14 @@ pub fn backfill_first_pool_ts_batch(conn: &Connection, limit: i64) -> anyhow::Re v }; - // 2) Do the updates inside a short transaction + // 2) Short write tx; all fallible helpers here return anyhow::Result, so `?` is fine. let tx = conn.unchecked_transaction()?; let mut updated = 0usize; for token_id in token_ids { if let Some((_creation_utxo, _txid, ts, _height)) = db_first_pool_creation_row(&tx, &token_id)? { - cache_first_pool_ts_if_empty(&tx, &token_id, ts)?; // will also replace 0; see below + cache_first_pool_ts_if_empty(&tx, &token_id, ts)?; updated += 1; } } diff --git a/src/db/cauldron/tokenlist/test.rs b/src/db/cauldron/tokenlist/test.rs index c0b13ac..78dc1b4 100644 --- a/src/db/cauldron/tokenlist/test.rs +++ b/src/db/cauldron/tokenlist/test.rs @@ -23,8 +23,8 @@ mod tests { }; use crate::db::cauldron::tokenlist::list_tokens_volume::db_list_tokens_by_volume; use crate::db::cauldron::tokenlist::token_utils::{ - apy_30d_bp_for_token, compute_score, pct_change_bp_dec, pow10_dec, price_from_tvl, - resolve_decimals, resolve_display_labels, + apy_30d_bp_for_token, compute_score, overflow_f64_fallback, pct_change_bp_dec, pow10_dec, + price_from_tvl, resolve_decimals, resolve_display_labels, }; use crate::db::cauldron::tx::{insert_block_tx, insert_mempool_tx}; use crate::db::cauldron::utxo_funding::insert_utxo_funding; @@ -150,18 +150,15 @@ mod tests { fn test_helper_math() { // pow10 assert_eq!(pow10_dec(0), Decimal::ONE); - assert_eq!(pow10_dec(2), Decimal::from_i32(100).unwrap()); + assert_eq!(pow10_dec(2), dec!(100)); // price_from_tvl - assert_eq!( - price_from_tvl(100, 10).unwrap(), - Decimal::from_i32(10).unwrap() - ); + assert_eq!(price_from_tvl(100, 10).unwrap(), dec!(10)); assert!(price_from_tvl(100, 0).is_none()); // pct_change (Decimal) - let a = Decimal::from_i32(110).unwrap(); - let b = Decimal::from_i32(100).unwrap(); + let a = dec!(110); + let b = dec!(100); assert_eq!(pct_change_bp_dec(a, b), 1000); // compute_score @@ -1003,4 +1000,257 @@ mod tests { .unwrap(); assert!(cached_ts.is_none()); } + + #[test] + fn fast_updater_handles_extreme_decimals_without_panicking() { + // Arrange: fresh DBs/tables + let mock = mock_db_pool(|conn| setup_basic_schemas(conn)); + let cw = mock.cauldron_w.get().expect("cauldron_w"); + let bcmr_w = mock.bcmr_w.get().expect("bcmr_w"); + let bcmr_r = mock.bcmr_r.get().expect("bcmr_r"); + let crc_r = mock.crc20_r.get().expect("crc20_r"); + let orc_r = mock.oracle_r.get().expect("oracle_r"); + + // Seed BCMR with absurd decimals (should trigger overflow pre-patch) + let token = TokenID::from_inner([0xDE; 32]); + let token_hex = token.to_hex(); + let utxo = OutPointHash::from_inner([0xEE; 32]); + let txid = Txid::from_inner([0xEF; 32]); + + let row = BCMRRow { + name: "OverflowCoin".into(), + description: "trigger pow10 overflow".into(), + token: BCMRToken { + category: "cat".into(), + symbol: "OF".into(), + decimals: 30, // <-- extreme + }, + uris: Uris { + icon: None, + web: None, + }, + filemeta: FileMeta { + expected_hash: Some("x".into()), + actual_hash: Some("y".into()), + source: "test".into(), + }, + }; + insert_bcmr_data(&bcmr_w, &utxo, &row).unwrap(); + insert_authheader( + &bcmr_w, + &utxo, + &BlockHash::all_zeros(), + &txid, + &token, + 100, + Some(Vec::from("bcmr".as_bytes())), + ) + .unwrap(); + + // Sanity: ensure resolve_decimals sees 30 via the same code path the updater uses + let dec = resolve_decimals(&bcmr_r, &crc_r, &token_hex); + assert_eq!( + dec, 30, + "resolve_decimals must read the extreme value for this test" + ); + + // Ensure TVL so the updater scales price by 10^decimals + let now = crate::timeutil::time_now(); + super::tests::seed_minimal_token_history( + &cw, + token, + now - 120, + now - 60, + 1_000, + 500, + 2_000, + 1_000, + ); + + update_tvl_and_price_now(&cw, &bcmr_r, &crc_r, &orc_r) + .expect("fast updater should not panic on extreme decimals"); + } + + #[test] + fn pow10_dec_clamps_and_does_not_overflow() { + use rust_decimal_macros::dec; + + // Exact 10^28 (1 followed by 28 zeros) + let e28 = crate::db::cauldron::tokenlist::token_utils::pow10_dec(28); + let e28_expected = dec!(10000000000000000000000000000); // ← 29 digits total + assert_eq!(e28, e28_expected); + + // Asking for 30 should clamp to 28 and not panic + let e30 = crate::db::cauldron::tokenlist::token_utils::pow10_dec(30); + assert_eq!(e30, e28_expected); + } + + #[test] + fn core_overflow_paths_set_prices_null_not_panic() { + let mock = mock_db_pool(|conn| setup_basic_schemas(conn)); + let cw = mock.cauldron_w.get().unwrap(); + let bcmr_w = mock.bcmr_w.get().unwrap(); + let bcmr_r = mock.bcmr_r.get().unwrap(); + let crc_r = mock.crc20_r.get().unwrap(); + let orc_r = mock.oracle_r.get().unwrap(); + + let token = TokenID::from_inner([0xEE; 32]); + let utxo = OutPointHash::from_inner([0xCD; 32]); + let txid = Txid::from_inner([0xAB; 32]); + + // decimals = 28 OK, but we’ll make price huge by tiny tokens + insert_bcmr_data( + &bcmr_w, + &utxo, + &BCMRRow { + name: "HugePrice".into(), + description: "".into(), + token: BCMRToken { + category: "c".into(), + symbol: "HP".into(), + decimals: 28, + }, + uris: Uris { + icon: None, + web: None, + }, + filemeta: FileMeta { + expected_hash: Some("x".into()), + actual_hash: Some("y".into()), + source: "test".into(), + }, + }, + ) + .unwrap(); + insert_authheader( + &bcmr_w, + &utxo, + &BlockHash::all_zeros(), + &txid, + &token, + 100, + Some(b"bcmr".to_vec()), + ) + .unwrap(); + + // seed: huge sats, tiny tokens (1) to push price_now_dec very large + let now = crate::timeutil::time_now(); + super::tests::seed_minimal_token_history( + &cw, + token, + now - 120, + now - 60, + 9_000_000_000_000_000, + 1, + 9_000_000_000_000_000, + 1, + ); + + // Create presence in cache + cw.execute("INSERT INTO cached_token_metrics(token_id, trade_volume, tvl_sats, tvl_tokens, score, updated_at) + VALUES(?1,0,1,1,0,strftime('%s','now')) + ON CONFLICT(token_id) DO NOTHING", rusqlite::params![token.to_hex()]).unwrap(); + + // Should not panic + update_changes_score_volume_and_ranking(&cw, &bcmr_r, &crc_r, &orc_r).unwrap(); + + // Prices likely NULL after overflow guard + let (p_now_usd, p_24h, p_7d): (Option, Option, Option) = + cw.query_row("SELECT price_now_usd, price_24h, price_7d FROM cached_token_metrics WHERE token_id=?1", + rusqlite::params![token.to_hex()], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?))).unwrap(); + assert!(p_now_usd.is_none() || p_now_usd.unwrap().is_finite()); + // allow None here; the point is: no panic + let _ = (p_24h, p_7d); + } + + #[test] + fn pct_change_bp_handles_huge_ratio_without_panic() { + use crate::db::cauldron::tokenlist::token_utils::{pct_change_bp_dec, pow10_dec}; + use rust_decimal::Decimal; + + // Arrange: absurd values that would overflow in current code + let current: Decimal = pow10_dec(28); // ~1e28 + let past: Decimal = Decimal::ONE / pow10_dec(18); // ~1e-18 + + // Act: call pct_change_bp_dec + let result = std::panic::catch_unwind(|| pct_change_bp_dec(current, past)); + + // Assert: should NOT panic once patched + assert!( + result.is_ok(), + "pct_change_bp_dec panicked on huge ratio (needs overflow guard)" + ); + } + + #[test] + fn overflow_fallback_uses_f64_and_does_not_panic() { + // simulate absurdly high decimals that would overflow Decimal multiplication + let base = Decimal::new(123456789, 0); // 123456789 + let decimals: u32 = 1000; + + let price_now_human = overflow_f64_fallback(base, decimals); + + assert!(price_now_human.is_finite() || price_now_human.is_infinite()); + } + #[test] + fn price_from_tvl_handles_u64_extremes_without_overflow() { + use std::u64; + + // Max sats, 1 token → result should be exactly MAX as Decimal + let p = price_from_tvl(u64::MAX, 1).expect("not None"); + assert_eq!(p, Decimal::from_u128(u64::MAX as u128).unwrap()); + + // Symmetric max → 1 + let p = price_from_tvl(u64::MAX, u64::MAX).expect("not None"); + assert_eq!(p, Decimal::ONE); + + // Max sats, 2 tokens → roughly MAX/2 + let p = price_from_tvl(u64::MAX, 2).expect("not None"); + // Use a tolerant comparison because Decimal division can normalize scale + let half = Decimal::from_u128((u64::MAX as u128) / 2).unwrap(); + // Allow a difference of 1 ulp at integer scale if needed + assert!(p >= half && p <= half + Decimal::ONE); + + // Tiny sats, huge tokens → small decimal > 0 + let p = price_from_tvl(1, u64::MAX).expect("not None"); + assert!(p > Decimal::ZERO && p < Decimal::ONE); + } + + #[test] + fn price_from_tvl_zero_tokens_is_none() { + assert!(price_from_tvl(0, 0).is_none()); + assert!(price_from_tvl(100, 0).is_none()); + } + + #[test] + fn price_from_tvl_no_u64_intermediate_multiply() { + // This test is about behavior: if an accidental u64 multiply had crept in, + // some extreme combos would panic or wrap. We assert it doesn't. + let cases = [ + (std::u64::MAX, 1u64), + (std::u64::MAX, std::u64::MAX), + (1u64, std::u64::MAX), + (9_000_000_000_000_000u64, 1u64), + ]; + for (sats, toks) in cases { + let _ = price_from_tvl(sats, toks); // should not panic + } + } + #[test] + fn scaling_price_never_requires_integer_multiply() { + // Simulate a very large ratio, then scale by decimals via Decimal first + let p = price_from_tvl(std::u64::MAX, 1).unwrap(); // huge base + // Use your clamped pow10_dec; should not panic + let factor = crate::db::cauldron::tokenlist::token_utils::pow10_dec(28); + let scaled = p.checked_mul(factor).unwrap_or_else(|| Decimal::ZERO); + // Convert to f64 bounded; must be finite or clamped per your helper + let f = crate::db::cauldron::tokenlist::token_utils::dec_to_f64_bounded(scaled); + assert!(f.is_finite() || f.abs() == f64::MAX); + } + #[test] + fn compute_score_never_overflows() { + let s = compute_score(u64::MAX, u64::MAX); + assert!(s >= 0); // and, importantly, no panic occurred + } } diff --git a/src/db/cauldron/tokenlist/token_utils.rs b/src/db/cauldron/tokenlist/token_utils.rs index 91a2863..1c87f90 100644 --- a/src/db/cauldron/tokenlist/token_utils.rs +++ b/src/db/cauldron/tokenlist/token_utils.rs @@ -3,7 +3,7 @@ // This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later. // A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html -use anyhow::{anyhow, Result}; +use anyhow::Result; use log::warn; use rusqlite::Connection; @@ -11,9 +11,45 @@ use crate::db::cauldron::pool::get_pool_period_snapshot; use crate::db::oracle::get_closest; use crate::rpc::apy::apyaggregator::APYAggregator; use crate::rpc::apy::poolperiod::PoolPeriod; +use malachite::base::num::arithmetic::traits::FloorSqrt; +use malachite::Integer; use rust_decimal::prelude::*; use rust_decimal::Decimal; +use rust_decimal_macros::dec; + +/// Fallback path when Decimal scaling overflows: convert to f64 with clamped exponent. +pub fn overflow_f64_fallback(base: Decimal, decimals: u32) -> f64 { + let base_f = dec_to_f64_bounded(base); + let exp = (decimals as i32).clamp(0, 308); // max safe for f64::powi + let factor_f = 10f64.powi(exp); + let p = base_f * factor_f; + if p.is_finite() { + p + } else { + f64::MAX.copysign(p) + } +} + +/// Round to a reasonable scale before conversion (tunable). +pub fn dec_round(d: Decimal, scale: u32) -> Decimal { + d.round_dp_with_strategy(scale, rust_decimal::RoundingStrategy::MidpointNearestEven) +} + +/// Convert Decimal -> f64, guaranteed finite (never NaN/∞). Clamps if needed. +/// Optional: enforce a floor so tiny non-zero values don't collapse to 0. +pub fn dec_to_f64_bounded(d: Decimal) -> f64 { + if let Some(v) = d.to_f64() { + if v.is_finite() { + return v; + } + } + if d.is_sign_negative() { + -f64::MAX + } else { + f64::MAX + } +} pub const SATS_PER_BCH: i64 = 100_000_000; pub const ORACLE_SCALE: i64 = 1_000_000; // match the delphi scale @@ -31,31 +67,62 @@ pub fn usd_per_bch_at_or_before(conn: &Connection, ts: i64) -> Decimal { #[inline] pub fn pct_change_bp_dec(current: Decimal, past: Decimal) -> i64 { if past.is_zero() { - 0 - } else { - ((current - past) / past * Decimal::from_i32(10_000).unwrap()) - .round() - .to_i64() - .unwrap_or(0) + return 0; + } + + let tenk = dec!(10000); + + // Compute: ((current / past) - 1) * 10_000 in a safer order: + // => (current * 10_000 / past) - 10_000 + let scaled_current = match current.checked_mul(tenk) { + Some(v) => v, + None => return if current >= past { i64::MAX } else { i64::MIN }, + }; + + let ratio_bp = match scaled_current.checked_div(past) { + Some(v) => v, + None => return 0, // should not happen (past==0 handled), but be defensive + }; + + let delta_bp = match ratio_bp.checked_sub(tenk) { + Some(v) => v, + None => return if current >= past { i64::MAX } else { i64::MIN }, + }; + + // Round and convert; clamp if it wouldn't fit in i64 + match delta_bp.round().to_i64() { + Some(v) => v, + None => { + if delta_bp.is_sign_negative() { + i64::MIN + } else { + i64::MAX + } + } } } #[inline] -pub fn compute_score(tvl_sats: u64, volume: u64) -> i64 { - if tvl_sats == 0 || volume == 0 { +pub fn compute_score(tvl_sats: u64, vol_30d: u64) -> i64 { + if tvl_sats == 0 || vol_30d == 0 { return 0; } - let s = (volume as f64) * (tvl_sats as f64).sqrt(); // same as BigNumber sqrt + pow(2) pattern - if !s.is_finite() { - return 0; - } - if s >= i64::MAX as f64 { - i64::MAX - } else { - s.round() as i64 - } // ROUND_HALF_UP for positives -} + // score = vol_30d * floor_sqrt(tvl_sats), all in big-int to avoid overflow + let sqrt_tvl: Integer = Integer::from(tvl_sats).floor_sqrt(); + let score_big: Integer = Integer::from(vol_30d) * sqrt_tvl; + + // Clamp to i64 range (non-negative by construction) + if score_big > i64::MAX { + return i64::MAX; + } + + // Convert safely; fall back to MAX on unexpected parse failure + match score_big.to_string().parse::() { + Ok(v) => v, + Err(_) => i64::MAX, + } +} pub fn resolve_decimals(bcmr_conn: &Connection, crc20_conn: &Connection, token_id: &str) -> u32 { // On-chain BCMR (latest by height) if let Ok(Some(v)) = bcmr_conn.query_row( @@ -105,10 +172,25 @@ pub fn resolve_decimals(bcmr_conn: &Connection, crc20_conn: &Connection, token_i 0 } +/// Max safe exponent for 10^d that fits in `rust_decimal` (1e28 fits; 1e29 does not). +pub const MAX_DECIMALS_FOR_SCALING: u32 = 28; + +/// Clamp a metadata decimals value to something rust_decimal can represent. +#[inline] +pub fn clamp_decimals(decimals: u32) -> u32 { + if decimals > MAX_DECIMALS_FOR_SCALING { + warn!( + "Clamping decimals {} -> {} to avoid overflow in pow10_dec()", + decimals, MAX_DECIMALS_FOR_SCALING + ); + } + decimals.min(MAX_DECIMALS_FOR_SCALING) +} + #[inline] pub fn pow10_dec(decimals: u32) -> Decimal { - // rust_decimal supports integer powers - Decimal::from_i32(10).unwrap().powu(decimals as u64) + // After clamping, powu is safe and exact. + dec!(10).powu(clamp_decimals(decimals) as u64) } #[inline] @@ -151,11 +233,22 @@ pub fn apy_30d_bp_for_token(conn: &Connection, token_id: &str, now: i64) -> Resu e })?; - // Percent → basis points (100 bp = 1%). If it won’t fit into i64, surface an error. - let bp_dec = (apy_dec * Decimal::from_i32(100).unwrap()).round(); - let apy_bp = bp_dec - .to_i64() - .ok_or_else(|| anyhow!("APY bp overflow for {}", token_id))?; + // Percent → basis points (100 bp = 1%) + let hundred = dec!(100); + + let bp_dec = match apy_dec.checked_mul(hundred) { + Some(v) => v.round(), + None => { + // clamp on overflow instead of panicking + return Ok(i64::MAX); + } + }; + + let apy_bp = match bp_dec.to_i64() { + Some(v) => v, + None if bp_dec.is_sign_negative() => i64::MIN, + None => i64::MAX, + }; Ok(apy_bp) } diff --git a/src/main.rs b/src/main.rs index a116290..5f864ff 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,7 +15,7 @@ use rocket::{launch, routes}; use rocket_cors::{AllowedHeaders, AllowedOrigins}; use rpc::ResponseCache; use rusqlite::OpenFlags; -use std::sync::atomic::AtomicBool; +use std::sync::atomic::{AtomicBool, Ordering}; use std::{ backtrace::Backtrace, collections::HashMap, @@ -23,7 +23,7 @@ use std::{ path::Path, process, sync::{Arc, Mutex}, - thread::{self, sleep}, + thread::sleep, time::{Duration, Instant}, }; use stderrlog::LogLevelNum; @@ -136,7 +136,6 @@ fn start_program( ) -> Result<(DB, BCMRDownloader, WellKnownDownloader, CRC20Fetcher)> { let create_db_pool = |db_path| -> (bool, DBPool, DBPool) { let db_exists = Path::new(db_path).exists(); - info!("Initializing connection to {db_path}"); let write_manager = r2d2_sqlite::SqliteConnectionManager::file(db_path) @@ -258,12 +257,12 @@ fn start_program( let indexing_in_progress_clone = indexing_in_progress.clone(); - thread::spawn(move || { + std::thread::spawn(move || { let db = db_cpy; + // Initial full index let mut tip: BlockHash = loop { - indexing_in_progress_clone.store(true, std::sync::atomic::Ordering::Relaxed); - + indexing_in_progress_clone.store(true, Ordering::Relaxed); break match index_blocks(chain.clone(), db.clone(), client.clone(), true) { Ok(tip) => tip, Err(e) => { @@ -275,22 +274,21 @@ fn start_program( } }; }; + indexing_in_progress_clone.store(false, Ordering::Relaxed); - indexing_in_progress_clone.store(false, std::sync::atomic::Ordering::Relaxed); - + // Follow chain loop { let new_tip = match electrum_get_tip(&client.lock().unwrap()) { Ok(t) => t.0.block_hash(), Err(e) => { - warn!("Failed to get block chain tip from electrum: {e}"); - thread::sleep(Duration::from_secs(5)); + warn!("Failed to get chain tip from electrum: {e}"); + std::thread::sleep(Duration::from_secs(5)); continue; } }; if new_tip != tip { - indexing_in_progress_clone.store(true, std::sync::atomic::Ordering::Relaxed); - + indexing_in_progress_clone.store(true, Ordering::Relaxed); tip = match index_blocks(chain.clone(), db.clone(), client.clone(), true) { Ok(t) => t, Err(e) => { @@ -298,15 +296,18 @@ fn start_program( tip } }; + indexing_in_progress_clone.store(false, Ordering::Relaxed); + } - indexing_in_progress_clone.store(false, std::sync::atomic::Ordering::Relaxed); + // Avoid overlapping writer while indexer is on + if !indexing_in_progress_clone.load(Ordering::Relaxed) { + if let Err(e) = + update_mempool(db.cauldron_w.clone(), db.oracle_w.clone(), client.clone()) + { + error!("Failed to update mempool: {e}"); + } } - if let Err(e) = - update_mempool(db.cauldron_w.clone(), db.oracle_w.clone(), client.clone()) - { - error!("Failed to update mempool: {e}"); - } - thread::sleep(Duration::from_secs(5)); + std::thread::sleep(Duration::from_secs(5)); } }); @@ -316,6 +317,8 @@ fn start_program( let mut wellknowndownloader = WellKnownDownloader::new(db.bcmr_w.clone()); wellknowndownloader.start()?; + spawn_token_metrics_updater(db.clone(), indexing_in_progress.clone()); + Ok((db, bcmrdownloader, wellknowndownloader, crc20fetcher)) } @@ -364,8 +367,6 @@ fn launch() -> _ { create_cached_token_metrics_table(&conn).expect("ensure cached_token_metrics exists"); } - spawn_token_metrics_updater(dbpool.clone()); - rocket::build() .manage(dbpool) .manage(response_cache) diff --git a/src/rpc/apy/poolperiod.rs b/src/rpc/apy/poolperiod.rs index b3540cc..25ac798 100644 --- a/src/rpc/apy/poolperiod.rs +++ b/src/rpc/apy/poolperiod.rs @@ -3,7 +3,9 @@ // This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later. // A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html -use anyhow::{bail, Context, Result}; +use anyhow::{anyhow, bail, Context, Result}; +use malachite::base::num::arithmetic::traits::FloorSqrt; +use malachite::Integer; use rust_decimal::MathematicalOps; use rust_decimal::{prelude::FromPrimitive, Decimal}; use rust_decimal_macros::dec; @@ -18,8 +20,8 @@ pub struct PoolPeriod { pub start: PoolSnapshot, pub end: PoolSnapshot, - start_k: Decimal, - end_k: Decimal, + start_k: Integer, + end_k: Integer, } impl PoolPeriod { @@ -29,14 +31,12 @@ impl PoolPeriod { "start timestamp ({}) > end timestamp ({})", start.timestamp, end.timestamp - ) + ); } - let start_k = Decimal::from_u64(start.sats * start.token_amount) - .context("failed to convert inital_k to decimal")?; - - let end_k = Decimal::from_u64(end.sats * end.token_amount) - .context("failed to convert final_k to decimal")?; + // Multiply with big-int to avoid u64 overflow, then downcast to Decimal. + let start_k = Integer::from(start.sats) * Integer::from(start.token_amount); + let end_k = Integer::from(end.sats) * Integer::from(end.token_amount); Ok(Self { start, @@ -72,40 +72,79 @@ impl PoolPeriod { } pub fn days_over_year(&self, starting: &Option) -> Result { + // If duration_days is 0, checked_div returns None → default to 0 Ok(DAYS_IN_YEAR .checked_div(self.duration_days(starting)?) .unwrap_or_default()) } - pub fn end_k_sqrt(&self) -> Result { - self.end_k.sqrt().context("failed to sqrt end") + fn sqrt_integer_as_decimal(k: &Integer) -> anyhow::Result { + use std::str::FromStr; + let s: Integer = k.clone().floor_sqrt(); + if s == 0u8 { + return Ok(Decimal::ZERO); + } + let r: Integer = k - &(&s * &s); // r = k - s^2 (exact) + let s_dec = Decimal::from_str(&s.to_string())?; + let r_dec = Decimal::from_str(&r.to_string())?; + // sqrt(k) ≈ s + r/(2s) (first-order correction) + let corr = r_dec + .checked_div(s_dec * dec!(2)) + .ok_or_else(|| anyhow::anyhow!("div"))?; + s_dec + .checked_add(corr) + .ok_or_else(|| anyhow::anyhow!("add")) } - pub fn pool_yield(&self) -> Result { - let start_k_sr = self.start_k.sqrt().context("failed to sqrt start")?; - let end_k_sr = self.end_k.sqrt().context("failed to sqrt end")?; + pub fn end_k_sqrt(&self) -> anyhow::Result { + Self::sqrt_integer_as_decimal(&self.end_k) + } - Ok(((end_k_sr - start_k_sr) / start_k_sr) * Decimal::ONE_HUNDRED) + pub fn pool_yield(&self) -> anyhow::Result { + let start_k_sr = Self::sqrt_integer_as_decimal(&self.start_k)?; + let end_k_sr = Self::sqrt_integer_as_decimal(&self.end_k)?; + if start_k_sr.is_zero() { + anyhow::bail!("start sqrt is zero; division by zero"); + } + let num = end_k_sr + .checked_sub(start_k_sr) + .ok_or_else(|| anyhow::anyhow!("sub"))?; + let ratio = num + .checked_div(start_k_sr) + .ok_or_else(|| anyhow::anyhow!("div"))?; + ratio + .checked_mul(Decimal::ONE_HUNDRED) + .ok_or_else(|| anyhow::anyhow!("mul")) } pub fn yield_and_apy(&self, starting: &Option) -> Result<(Decimal, Decimal)> { let pool_yield = self.pool_yield()?; + let years_elapsed = self.days_over_year(starting)?; // > 0 means at least some duration - let years_elapsed = self.days_over_year(starting)?; - - // to avoid powd overflow; don't calculate for pools < 6 hour old - if self.duration(starting) < 3600 * 6 { + // avoid powd blowups for very short periods (< 6h) + if self.duration(starting) < 6 * 3600 { return Ok((pool_yield, Decimal::ZERO)); } let apy = if years_elapsed.is_zero() { Decimal::ZERO } else { - (((pool_yield / Decimal::ONE_HUNDRED) + Decimal::ONE) + // APY = ((1 + y)^years - 1) * 100, with overflow guards + let one_plus = (pool_yield + .checked_div(Decimal::ONE_HUNDRED) + .ok_or_else(|| anyhow!("divide by 100 overflow"))?) + .checked_add(Decimal::ONE) + .ok_or_else(|| anyhow!("1 + yield overflow"))?; + + let powd = one_plus .checked_powd(years_elapsed) - .context("powd overflow")? - - Decimal::ONE) - * Decimal::ONE_HUNDRED + .context("powd overflow")?; + let minus_one = powd + .checked_sub(Decimal::ONE) + .ok_or_else(|| anyhow!("pow - 1 underflow"))?; + minus_one + .checked_mul(Decimal::ONE_HUNDRED) + .ok_or_else(|| anyhow!("apy * 100 overflow"))? }; Ok((pool_yield, apy))