BitcoinCash/riftenlabs-indexer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix potential API hangs from write lock contention

Browse source 
- Refactor update_changes_score_volume_and_ranking to use read connections
  for read/compute phases, only acquiring write connection for final writes
- Change first_pool_creation to use try_get() for opportunistic caching,
  preventing thread exhaustion if write pool is busy
- Add linter to flag blocking _w.get() calls in RPC handlers
This commit is contained in:
Dagur Valberg Johannsson 2026-02-02 21:01:50 +01:00
parent 055d0bee29
commit e1ebc853f5
4 changed files with 446 additions and 345 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

80
linters/rpc_write_lock_check.py Executable file
View file

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

689
src/db/cauldron/tokenlist/metrics_cache.rs
View file

@ -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<u8>,
trade_volume: i64,
score: i64,
price_24h: Option<f64>,
price_7d: Option<f64>,
change_24h_bp: Option<i64>,
change_7d_bp: Option<i64>,
price_now_usd: Option<f64>,
price_24h_usd: Option<f64>,
price_7d_usd: Option<f64>,
change_24h_usd_bp: Option<i64>,
change_7d_usd_bp: Option<i64>,
display_name: Option<String>,
display_symbol: Option<String>,
}
/// 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<u8>, 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<u8> = row.get(0)?;
let token_id = match blob_to_display_hex::<TokenID>(&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<TokenMetricsUpdate> = Vec::with_capacity(tokens_data.len());
let mut dec_cache: HashMap<String, u32> = 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<String, u32> = HashMap::new();
while let Some(row) = id_rows.next()? {
// Convert blob to display hex (handles byte reversal for Bitcoin hashes)
let token_blob: Vec<u8> = row.get(0)?;
let token_id =
match crate::db::blob::blob_to_display_hex::<TokenID>(&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::<f64>::None, // price_24h
Option::<f64>::None, // price_7d
Option::<i64>::None, // change_24h_bp
Option::<i64>::None, // change_7d_bp
p_now_usd_f, // fallback (finite) on this branch
Option::<f64>::None, // price_24h_usd
Option::<f64>::None, // price_7d_usd
Option::<i64>::None, // change_24h_usd_bp
Option::<i64>::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<String>, Option<String>) =
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::<f64>::None,
Option::<f64>::None,
Option::<i64>::None,
Option::<i64>::None,
price_now_usd_f_opt, // ← may be NULL if overflowed
Option::<f64>::None,
Option::<f64>::None,
Option::<i64>::None,
Option::<i64>::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<String>, Option<String>) = 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<AtomicBool>
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()),
}
}

10
src/db/cauldron/tokenlist/test.rs
View file

@ -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<i64>, Option<i64>, Option<i64>, Option<i64>) = 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<f64>, Option<f64>, Option<f64>) =

12
src/rpc/tokens.rs
View file

@ -335,11 +335,13 @@ pub fn first_pool_creation(token: &str, dbp: &State<DB>) -> CachedApiResult<Valu
match db_first_pool_creation_row(&conn, &token_hex) {
Ok(Some((creation_utxo, txid, timestamp, block_height))) => {
// 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(