diff --git a/linters/rpc_write_lock_check.py b/linters/rpc_write_lock_check.py new file mode 100755 index 0000000..4225ba3 --- /dev/null +++ b/linters/rpc_write_lock_check.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python3 +# Copyright (C) 2024-2026 Whiterun LLC, +# 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 + +import os +import sys +import re + +""" +Disallow blocking write pool access (_w.get()) in RPC handlers. + +Blocking .get() calls on write pools can exhaust Rocket's thread pool if a +background task holds the write lock for an extended period. All threads +waiting on .get() will block, causing the entire API to hang. + +Use try_get() instead, which returns None immediately if the lock is unavailable. + +OK (non-blocking): + if let Some(cw) = dbp.cauldron_w.try_get() { ... } + +NOT OK (blocks thread, can cause API hang): + let cw = dbp.cauldron_w.get()?; +""" + +OUR_PATH = os.path.dirname(os.path.realpath(__file__)) + +# Pattern to detect blocking write pool access: something_w.get() +FORBIDDEN_PATTERN = re.compile(r"\b\w+_w\.get\(\)") + + +def check_file_for_blocking_write_lock(file_path): + """Check if the file contains blocking _w.get() calls outside of test modules.""" + violations = [] + + with open(file_path, 'r', encoding='utf-8') as file: + for line_num, line in enumerate(file, 1): + # Stop checking if we enter test module + if re.search(r"\bmod tests \{", line): + break + + if FORBIDDEN_PATTERN.search(line): + violations.append((line_num, line.strip())) + + return violations + + +def check_rpc_files(): + """ + Traverse src/rpc and check each .rs file for blocking write lock access. + """ + rpc_dir = os.path.join(OUR_PATH, "..", "src", "rpc") + found_violations = False + + for root, dirs, files in os.walk(rpc_dir): + for file in files: + if file.endswith('.rs'): + file_path = os.path.join(root, file) + violations = check_file_for_blocking_write_lock(file_path) + + if violations: + found_violations = True + rel_path = os.path.relpath(file_path, os.path.join(OUR_PATH, "..")) + for line_num, line in violations: + print(f"{rel_path}:{line_num}: blocking _w.get() found") + print(f" {line}") + print(f" hint: use try_get() instead to avoid thread exhaustion") + print() + + if found_violations: + print("ERROR: Blocking write pool access found in RPC handlers.") + print("This can cause API hangs if a background task holds the write lock.") + sys.exit(1) + else: + print("OK") + sys.exit(0) + + +if __name__ == "__main__": + check_rpc_files() diff --git a/src/db/cauldron/tokenlist/metrics_cache.rs b/src/db/cauldron/tokenlist/metrics_cache.rs index 37104ba..cf90ea6 100644 --- a/src/db/cauldron/tokenlist/metrics_cache.rs +++ b/src/db/cauldron/tokenlist/metrics_cache.rs @@ -285,19 +285,105 @@ pub fn recompute_score_ranks(conn: &Connection) -> anyhow::Result<()> { ) } -// ---------- 5 min: price changes (24h/7d), score, volume + ranking ---------- -pub fn update_changes_score_volume_and_ranking( - cauldron_conn: &Connection, - bcmr_conn: &Connection, - crc20_conn: &Connection, - oracle_conn: &Connection, +/// Holds computed metrics for a single token, ready to be written to DB. +struct TokenMetricsUpdate { + token_blob: Vec, + trade_volume: i64, + score: i64, + price_24h: Option, + price_7d: Option, + change_24h_bp: Option, + change_7d_bp: Option, + price_now_usd: Option, + price_24h_usd: Option, + price_7d_usd: Option, + change_24h_usd_bp: Option, + change_7d_usd_bp: Option, + display_name: Option, + display_symbol: Option, +} + +/// Flush a batch of token metrics updates in a short transaction. +fn flush_metrics_update_batch( + conn: &Connection, + updates: &[TokenMetricsUpdate], ) -> anyhow::Result<()> { + with_busy_retry( + || { + let tx = conn.unchecked_transaction()?; + { + let mut upd = tx.prepare( + r#" + UPDATE cached_token_metrics + SET trade_volume = ?2, + score = ?3, + price_24h = ?4, + price_7d = ?5, + change_24h_bp = ?6, + change_7d_bp = ?7, + 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 + "#, + )?; + for u in updates { + upd.execute(params![ + &u.token_blob, + u.trade_volume, + u.score, + u.price_24h, + u.price_7d, + u.change_24h_bp, + u.change_7d_bp, + u.price_now_usd, + u.price_24h_usd, + u.price_7d_usd, + u.change_24h_usd_bp, + u.change_7d_usd_bp, + &u.display_name, + &u.display_symbol, + ])?; + } + } + tx.commit()?; + Ok(()) + }, + Duration::from_secs(30), + ) +} + +// ---------- 5 min: price changes (24h/7d), score, volume + ranking ---------- +pub fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result<()> { + // Acquire read connections for read/compute phases + let cauldron_r = db + .cauldron_r + .get() + .map_err(|e| anyhow::anyhow!("cauldron_r pool: {e}"))?; + let bcmr_r = db + .bcmr_r + .get() + .map_err(|e| anyhow::anyhow!("bcmr_r pool: {e}"))?; + let crc20_r = db + .crc20_r + .get() + .map_err(|e| anyhow::anyhow!("crc20_r pool: {e}"))?; + let oracle_r = db + .oracle_r + .get() + .map_err(|e| anyhow::anyhow!("oracle_r pool: {e}"))?; + // 30d volume window let now = time_now(); let since = (now - 30 * 24 * 60 * 60).max(0); - // volume per token - let mut vol_stmt = cauldron_conn.prepare( + // volume per token (using read connection) + let mut vol_stmt = cauldron_r.prepare( r#" SELECT p.token_id, COALESCE(SUM(ABS(phe.sats_delta)), 0) AS vol FROM pool_history_entry phe @@ -320,335 +406,283 @@ pub fn update_changes_score_volume_and_ranking( let ts_24h = now - 86_400; let ts_7d = now - 7 * 86_400; - // oracle (USD/BCH) → USD per sat + // oracle (USD/BCH) → USD per sat (using read connection) let sats_per_bch = Decimal::from_i64(SATS_PER_BCH).unwrap(); - let usd_per_sat_now = usd_per_bch_at_or_before(oracle_conn, now) / sats_per_bch; - 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 usd_per_sat_now = usd_per_bch_at_or_before(&oracle_r, now) / sats_per_bch; + let usd_per_sat_24h = usd_per_bch_at_or_before(&oracle_r, ts_24h) / sats_per_bch; + let usd_per_sat_7d = usd_per_bch_at_or_before(&oracle_r, ts_7d) / sats_per_bch; - // One write tx; scope statements so they drop before commit. - with_busy_retry( - || { - let tx = cauldron_conn.unchecked_transaction()?; - { - let mut upd = tx.prepare( - r#" - UPDATE cached_token_metrics - SET trade_volume = ?2, - score = ?3, + // ========== READ PHASE: collect all token data ========== + // Read token_ids and their current TVL (using read connection) + let mut tokens_data: Vec<(Vec, String, u64, u64)> = Vec::new(); + { + let mut stmt = cauldron_r.prepare( + "SELECT token_id, tvl_sats, tvl_tokens FROM cached_token_metrics WHERE tvl_sats > 0", + )?; + let mut rows = stmt.query([])?; + while let Some(row) = rows.next()? { + let token_blob: Vec = row.get(0)?; + let token_id = match blob_to_display_hex::(&token_blob) { + Ok(h) => h, + Err(_) => continue, + }; + let tvl_sats: i64 = row.get(1)?; + let tvl_tokens: i64 = row.get(2)?; + tokens_data.push(( + token_blob, + token_id, + tvl_sats.max(0) as u64, + tvl_tokens.max(0) as u64, + )); + } + } - price_24h = ?4, - price_7d = ?5, - change_24h_bp = ?6, - change_7d_bp = ?7, + // ========== COMPUTE PHASE: compute all updates (no transaction) ========== + let mut updates: Vec = Vec::with_capacity(tokens_data.len()); + let mut dec_cache: HashMap = HashMap::new(); - price_now_usd = ?8, - price_24h_usd = ?9, - price_7d_usd = ?10, - change_24h_usd_bp = ?11, - change_7d_usd_bp = ?12, + for (token_blob, token_id, tvl_sats, tvl_tokens) in tokens_data { + // decimals → factor + let decimals_u32 = *dec_cache + .entry(token_id.clone()) + .or_insert_with(|| resolve_decimals(&bcmr_r, &crc20_r, &token_id)); + let factor = pow10_dec(decimals_u32); - display_name = ?13, - display_symbol = ?14, - updated_at = CAST(strftime('%s','now') AS INTEGER) - WHERE token_id = ?1 - "#, - )?; + // price_now (base) & scaled with guards + let price_now_dec = price_from_tvl(tvl_sats, tvl_tokens).unwrap_or(Decimal::ZERO); - // 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", - )?; - // Read token_id as BLOB directly (not via SQL hex()) - let mut id_stmt = - tx.prepare("SELECT token_id FROM cached_token_metrics WHERE tvl_sats > 0")?; + 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 + ); - let mut id_rows = id_stmt.query([])?; - let mut dec_cache: HashMap = HashMap::new(); - - while let Some(row) = id_rows.next()? { - // Convert blob to display hex (handles byte reversal for Bitcoin hashes) - let token_blob: Vec = row.get(0)?; - let token_id = - match crate::db::blob::blob_to_display_hex::(&token_blob) { - Ok(h) => h, - Err(_) => continue, - }; - - // 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_blob], |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 with guards - 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!( - "core price_now overflow; token={}, base={}, factor=10^{}, tvl_sats={}, tvl_tokens={}", - token_id, price_now_dec, decimals_u32, tvl_sats, tvl_tokens - ); - - // 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) - } - }; - - 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)); - - upd.execute(params![ - &token_blob, - 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; - } - }; - - // 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))); - - // volume/score - let vol = vol_by_token.get(&token_id).copied().unwrap_or(0); - let score = compute_score(tvl_sats, vol); - - // 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_blob, - 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_blob, - 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; + // 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) } + }; - // --------- 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)); + 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_r, &crc20_r, &token_id).ok(); + let (display_name, display_symbol) = labels_opt + .map(|(dn, sym)| (Some(dn), Some(sym))) + .unwrap_or((None, None)); - // 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)); + updates.push(TokenMetricsUpdate { + token_blob, + trade_volume: vol as i64, + score, + price_24h: None, + price_7d: None, + change_24h_bp: None, + change_7d_bp: None, + price_now_usd: Some(p_now_usd_f), + price_24h_usd: None, + price_7d_usd: None, + change_24h_usd_bp: None, + change_7d_usd_bp: None, + display_name, + display_symbol, + }); + continue; + } + }; - // 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)); + // 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))); - // 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, - }; + // volume/score + let vol = vol_by_token.get(&token_id).copied().unwrap_or(0); + let score = compute_score(tvl_sats, vol); - 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); + // labels + let labels_opt = resolve_display_labels(&bcmr_r, &crc20_r, &token_id).ok(); + let (display_name, display_symbol): (Option, Option) = match labels_opt { + Some((dn, sym)) => (Some(dn), Some(sym)), + None => (None, None), + }; - 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, - }; + // 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 + updates.push(TokenMetricsUpdate { + token_blob, + trade_volume: vol as i64, + score, + price_24h: None, + price_7d: None, + change_24h_bp: None, + change_7d_bp: None, + price_now_usd: price_now_usd_f_opt, + price_24h_usd: None, + price_7d_usd: None, + change_24h_usd_bp: None, + change_7d_usd_bp: None, + display_name, + display_symbol, + }); + continue; + } + }; - upd.execute(params![ - &token_blob, - 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 + 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_r, ts_24h, &tid), + Ok((_ts, _)) + ); + let p7d_exists = matches!(price_at_or_before_2(&cauldron_r, ts_7d, &tid), Ok((_ts, _))); - // 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); - "#, - )?; + let price_24h_human_f = if p24_exists { + Some(dec_to_f64_bounded(dec_round(price_now_human_dec, 12))) + } else { + None + }; - tx.commit()?; - Ok(()) - }, - Duration::from_secs(120), - )?; + 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, + }; + + updates.push(TokenMetricsUpdate { + token_blob, + trade_volume: vol as i64, + score, + price_24h: price_24h_human_f, + price_7d: price_7d_human_f, + change_24h_bp: d24_bp_opt, + change_7d_bp: d7d_bp_opt, + price_now_usd: price_now_usd_f_opt, + price_24h_usd: price_24h_usd_f, + price_7d_usd: price_7d_usd_f, + change_24h_usd_bp: d24_usd_bp_opt, + change_7d_usd_bp: d7d_usd_bp_opt, + display_name, + display_symbol, + }); + continue; + } + + // --------- FULL PATH (vol > 0) ---------- + let p24_dec_opt = price_at_or_before_2(&cauldron_r, ts_24h, &tid) + .ok() + .and_then(|(_, p)| Decimal::from_f64(p)); + let p7d_dec_opt = price_at_or_before_2(&cauldron_r, 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, + }; + + updates.push(TokenMetricsUpdate { + token_blob, + trade_volume: vol as i64, + score, + price_24h: price_24h_human_f, + price_7d: price_7d_human_f, + change_24h_bp: d24_bp_opt, + change_7d_bp: d7d_bp_opt, + price_now_usd: price_now_usd_f_opt, + price_24h_usd: price_24h_usd_f, + price_7d_usd: price_7d_usd_f, + change_24h_usd_bp: d24_usd_bp_opt, + change_7d_usd_bp: d7d_usd_bp_opt, + display_name, + display_symbol, + }); + } + + // ========== WRITE PHASE: batch updates in chunks ========== + // Acquire single write connection for all batches + ranking + let cauldron_w = db + .cauldron_w + .get() + .map_err(|e| anyhow::anyhow!("cauldron_w pool: {e}"))?; + + for chunk in updates.chunks(WRITE_CHUNK) { + flush_metrics_update_batch(&cauldron_w, chunk)?; + } + + // Ranking update (separate short transaction, same connection) + recompute_score_ranks(&cauldron_w)?; Ok(()) } @@ -801,18 +835,9 @@ pub fn spawn_token_metrics_updater(db: DB, indexing_in_progress: Arc 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_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"); + match update_changes_score_volume_and_ranking(&db) { + Ok(_) => info!("✅ core(5m) in {:.3?}", t0.elapsed()), + Err(e) => error!("❌ core(5m) failed after {:.3?}: {e:?}", t0.elapsed()), } } diff --git a/src/db/cauldron/tokenlist/test.rs b/src/db/cauldron/tokenlist/test.rs index 6e14cc0..6615cc5 100644 --- a/src/db/cauldron/tokenlist/test.rs +++ b/src/db/cauldron/tokenlist/test.rs @@ -637,9 +637,6 @@ mod tests { fn core_updater_keeps_deltas_null_for_young_token() { let mock = mock_db_pool(|conn| setup_basic_schemas(conn)); let cw = mock.cauldron_w.get().unwrap(); - let bcmr = mock.bcmr_r.get().unwrap(); - let crc = mock.crc20_r.get().unwrap(); - let orc = mock.oracle_r.get().unwrap(); // Seed a YOUNG token with *volume* so we do the full path, but with no 24h anchor. // Make both events within the last hour. @@ -659,7 +656,7 @@ mod tests { .unwrap(); // Run core updater; for a "young" token with no historicals, deltas remain NULL - update_changes_score_volume_and_ranking(&cw, &bcmr, &crc, &orc).unwrap(); + update_changes_score_volume_and_ranking(&mock).unwrap(); let (c24, c7d, c24u, c7du): (Option, Option, Option, Option) = cw .query_row( @@ -1121,9 +1118,6 @@ mod tests { 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]); @@ -1184,7 +1178,7 @@ mod tests { ON CONFLICT(token_id) DO NOTHING", rusqlite::params![token.to_blob()]).unwrap(); // Should not panic - update_changes_score_volume_and_ranking(&cw, &bcmr_r, &crc_r, &orc_r).unwrap(); + update_changes_score_volume_and_ranking(&mock).unwrap(); // Prices likely NULL after overflow guard let (p_now_usd, p_24h, p_7d): (Option, Option, Option) = diff --git a/src/rpc/tokens.rs b/src/rpc/tokens.rs index 123718b..a81b11e 100644 --- a/src/rpc/tokens.rs +++ b/src/rpc/tokens.rs @@ -335,11 +335,13 @@ pub fn first_pool_creation(token: &str, dbp: &State) -> CachedApiResult { - // Opportunistically cache the timestamp forever - let cw = dbp.cauldron_w.get().map_err(db_error)?; - if let Err(e) = cache_first_pool_ts_if_empty(&cw, &token_hex, timestamp) { - // Non-fatal: return the data - log::warn!("Failed to cache first_pool_ts for {token_hex}: {e}"); + // Opportunistically cache the timestamp forever (non-blocking) + // Use try_get() to avoid blocking if write pool is busy + if let Some(cw) = dbp.cauldron_w.try_get() { + if let Err(e) = cache_first_pool_ts_if_empty(&cw, &token_hex, timestamp) { + // Non-fatal: return the data + log::warn!("Failed to cache first_pool_ts for {token_hex}: {e}"); + } } Ok(cached_ok(