From 10ca5d324cb505bd067497e9b81275a36d119c84 Mon Sep 17 00:00:00 2001 From: Jakob Notland Date: Thu, 19 Mar 2026 12:30:34 +0100 Subject: [PATCH 1/9] Checkpoint --- Cargo.lock | 66 +++++ Cargo.toml | 1 + src/db/cauldron/mod.rs | 1 + src/db/cauldron/tx.rs | 17 +- src/main.rs | 179 ++++++++++++- src/rpc/candlesticks.rs | 552 +++++++++++++++++++++++++++++++++------- 6 files changed, 704 insertions(+), 112 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7f06f30..7567bfd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -73,6 +73,17 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" +[[package]] +name = "async-lock" +version = "3.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" +dependencies = [ + "event-listener", + "event-listener-strategy", + "pin-project-lite", +] + [[package]] name = "async-stream" version = "0.3.6" @@ -464,6 +475,15 @@ version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" +[[package]] +name = "crossbeam-channel" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-deque" version = "0.8.6" @@ -675,6 +695,16 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "event-listener-strategy" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" +dependencies = [ + "event-listener", + "pin-project-lite", +] + [[package]] name = "fastrand" version = "2.3.0" @@ -1560,6 +1590,26 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "moka" +version = "0.12.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85f8024e1c8e71c778968af91d43700ce1d11b219d127d79fb2934153b82b42b" +dependencies = [ + "async-lock", + "crossbeam-channel", + "crossbeam-epoch", + "crossbeam-utils", + "equivalent", + "event-listener", + "futures-util", + "parking_lot", + "portable-atomic", + "smallvec", + "tagptr", + "uuid", +] + [[package]] name = "multer" version = "3.1.0" @@ -1793,6 +1843,12 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" +[[package]] +name = "portable-atomic" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49" + [[package]] name = "potential_utf" version = "0.1.2" @@ -2178,6 +2234,7 @@ dependencies = [ "hex", "log", "malachite", + "moka", "rand 0.8.5", "rayon", "reqwest", @@ -2955,6 +3012,12 @@ dependencies = [ "syn 2.0.101", ] +[[package]] +name = "tagptr" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" + [[package]] name = "tap" version = "1.0.1" @@ -3426,6 +3489,9 @@ name = "uuid" version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "458f7a779bf54acc9f347480ac654f68407d3aab21269a6e3c9f922acd9e2da9" +dependencies = [ + "getrandom 0.3.3", +] [[package]] name = "valuable" diff --git a/Cargo.toml b/Cargo.toml index 85f42a3..19b1faf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,7 @@ reqwest = { version = "0.12", features = ["json", "rustls-tls"], default-feature rand = "0.8.5" configure_me = "0.4.0" futures = "0.3.32" +moka = { version = "0.12", features = ["future"] } [build-dependencies] configure_me_codegen = "0.4.8" diff --git a/src/db/cauldron/mod.rs b/src/db/cauldron/mod.rs index 6d21b0b..977c1d0 100644 --- a/src/db/cauldron/mod.rs +++ b/src/db/cauldron/mod.rs @@ -16,6 +16,7 @@ use crate::db::cauldron::tokenlist::db_utils::{ pub mod config; pub mod header; pub mod mempool; +pub mod ohlcv; pub mod pool; pub mod poolvisitor; pub mod tokenlist; diff --git a/src/db/cauldron/tx.rs b/src/db/cauldron/tx.rs index d2bba6c..e4afda7 100644 --- a/src/db/cauldron/tx.rs +++ b/src/db/cauldron/tx.rs @@ -75,12 +75,17 @@ pub async fn latest( ) -> Result, u64)>> { let (sql, token_blob) = match &token_id { Some(tid) => ( - "SELECT DISTINCT tx.txid, tx.blockhash, tx.mtp_timestamp, tx.first_seen_timestamp - FROM tx - JOIN utxo_funding ON tx.txid = utxo_funding.txid - WHERE utxo_funding.token_id = ?1 - ORDER BY tx.effective_timestamp DESC - LIMIT ?2 OFFSET ?3", + "SELECT tx.txid, tx.blockhash, tx.mtp_timestamp, tx.first_seen_timestamp + FROM ( + SELECT txid, MAX(effective_timestamp) AS ts + FROM pool_history_entry + WHERE token_id = ?1 + GROUP BY txid + ORDER BY ts DESC + LIMIT ?2 OFFSET ?3 + ) AS recent + JOIN tx ON tx.txid = recent.txid + ORDER BY recent.ts DESC", Some(tid.to_blob()), ), None => ( diff --git a/src/main.rs b/src/main.rs index 08f01ca..c9e0aa3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -14,7 +14,7 @@ use electrum_client_netagnostic::{Client, ElectrumApi, Param}; use log::{error, info, warn}; use rocket::{launch, routes}; use rocket_cors::{AllowedHeaders, AllowedOrigins}; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}; use std::{ backtrace::Backtrace, panic, process, @@ -23,6 +23,14 @@ use std::{ }; use stderrlog::LogLevelNum; +/// Tracks how far the OHLCV pre-aggregation table has been populated. +/// `materialized_end` is the exclusive upper bound: `ohlcv_1h` contains +/// data for all complete 1-hour buckets whose `bucket_ts + 3600 ≤ materialized_end`. +/// Value 0 means nothing has been materialised yet. +pub struct OhlcvState { + pub materialized_end: AtomicI64, +} + /// State tracking for Initial Block Download (IBD). /// Used to return 503 errors while the indexer is catching up. pub struct IbdState { @@ -64,6 +72,7 @@ const CHIPNET_START_BLOCK: &str = const KEY_LAST_INDEXED: &str = "last_indexed"; mod bcmr; +mod cache; mod cashaddr; mod chain; mod crc20; @@ -111,6 +120,7 @@ async fn start_program( WellKnownDownloader, CRC20Fetcher, Arc, + Arc, // indexing_in_progress )> { let network = match config.network.to_lowercase().as_str() { "mainnet" => Network::Bitcoin, @@ -264,13 +274,10 @@ async fn start_program( }; }; indexing_in_progress_clone.store(false, Ordering::Relaxed); - - ibd_state_clone - .initial_sync_complete - .store(true, Ordering::Relaxed); info!("Initial block download complete"); - // Update query planner statistics so joins pick optimal order + // Run ANALYZE before signalling initial_sync_complete so the ohlcv post-IBD + // backfill (which waits for that flag) doesn't race with this write. info!("Running ANALYZE on cauldron database..."); if let Err(e) = sqlx::query("ANALYZE;").execute(&db.cauldron_w).await { warn!("ANALYZE failed: {e}"); @@ -278,6 +285,10 @@ async fn start_program( info!("ANALYZE complete"); } + ibd_state_clone + .initial_sync_complete + .store(true, Ordering::Relaxed); + // Follow chain loop { if signal::shutdown_requested() { @@ -343,6 +354,7 @@ async fn start_program( wellknowndownloader, crc20fetcher, ibd_state, + indexing_in_progress, )) } @@ -362,7 +374,7 @@ async fn launch() -> _ { config }; - let (dbpool, bcmrdownloader, wellknowndownloader, crc20fetcher, ibd_state) = + let (dbpool, bcmrdownloader, wellknowndownloader, crc20fetcher, ibd_state, indexing_in_progress) = match start_program(config).await { Ok(db) => db, Err(e) => { @@ -391,11 +403,164 @@ async fn launch() -> _ { .await .expect("ensure cached_token_metrics exists"); + // Ensure the OHLCV pre-aggregation table exists (safe on both new and existing DBs). + db::cauldron::ohlcv::create_table(&dbpool.cauldron_w).await; + + // Bootstrap OhlcvState from whatever is already in the table (survives restarts). + let initial_ohlcv_end = db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r) + .await + .unwrap_or(None) + .map(|ts| ts + 3600) + .unwrap_or(0); + let ohlcv_state = Arc::new(OhlcvState { + materialized_end: AtomicI64::new(initial_ohlcv_end), + }); + + // Synchronous post-IBD backfill: run the full ohlcv_1h materialisation before + // allowing metrics_cache and other background writers to start. We reuse the + // indexing_in_progress flag so metrics_cache backs off during this window. + { + const BACKFILL_BATCH_SECS: i64 = 24 * 3600; + const BACKFILL_SAFETY_SECS: i64 = 3 * 3600; + + // Wait for IBD to finish — ohlcv_1h data is only useful for confirmed blocks. + while !ibd_state.initial_sync_complete.load(Ordering::Relaxed) { + tokio::time::sleep(Duration::from_secs(1)).await; + } + + // Gate metrics_cache so it doesn't compete for cauldron_w during backfill. + indexing_in_progress.store(true, Ordering::Relaxed); + info!("ohlcv: starting post-IBD full backfill"); + + let now = crate::timeutil::time_now() as i64; + let cutoff = (now - BACKFILL_SAFETY_SECS) / 3600 * 3600; + let since_opt = match db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r).await { + Ok(Some(ts)) => Some(ts + 3600), + Ok(None) => { + match db::cauldron::ohlcv::get_min_trade_bucket_ts(&dbpool.cauldron_r).await { + Ok(v) => v, + Err(e) => { + warn!("ohlcv backfill: could not read min trade ts: {e}"); + None + } + } + } + Err(e) => { warn!("ohlcv backfill: could not read max bucket ts: {e}"); None } + }; + if since_opt.is_none() { + info!("ohlcv backfill: no confirmed trades found, skipping"); + } + if let Some(mut batch_start) = since_opt { + while batch_start < cutoff { + let batch_end = (batch_start + BACKFILL_BATCH_SECS).min(cutoff); + match db::cauldron::ohlcv::rebuild_range( + &dbpool.cauldron_r, + &dbpool.cauldron_w, + batch_start, + batch_end, + ) + .await + { + Ok(n) => { + info!("ohlcv backfill: {n} buckets [{batch_start}, {batch_end})"); + ohlcv_state + .materialized_end + .store(batch_end, Ordering::Relaxed); + } + Err(e) => { + warn!("ohlcv backfill failed at [{batch_start}, {batch_end}): {e}"); + break; + } + } + batch_start = batch_end; + // Brief yield so new block writes are not starved. + tokio::time::sleep(Duration::from_millis(200)).await; + } + } + + info!("ohlcv: post-IBD backfill complete"); + indexing_in_progress.store(false, Ordering::Relaxed); + } + + // Background task: incrementally materialise new 1-hour OHLCV buckets as blocks arrive. + // The full historical backfill above already ran; this task only handles the tail. + // Only processes buckets older than 3 hours (well beyond BCH reorg depth). + { + let ohlcv_write = dbpool.cauldron_w.clone(); + let ohlcv_read = dbpool.cauldron_r.clone(); + let ohlcv_state_bg = ohlcv_state.clone(); + tokio::spawn(async move { + // Batch size: 1 day per SQL call to keep each write short. + const BATCH_SECS: i64 = 24 * 3600; + // Safety margin: only materialise buckets older than this many seconds. + const SAFETY_SECS: i64 = 3 * 3600; + + loop { + + let now = crate::timeutil::time_now() as i64; + // Floor to 1-hour boundary, 3 hours ago. + let cutoff = (now - SAFETY_SECS) / 3600 * 3600; + + let since = match db::cauldron::ohlcv::get_max_bucket_ts(&ohlcv_read).await { + Ok(Some(max_ts)) => max_ts + 3600, + Ok(None) => { + // Table is empty: start from the first confirmed trade rather than + // scanning from Unix epoch 0 through thousands of empty batches. + match db::cauldron::ohlcv::get_min_trade_bucket_ts(&ohlcv_read).await { + Ok(Some(min_ts)) => min_ts, + Ok(None) => { + // No confirmed trades yet; wait before retrying. + tokio::time::sleep(Duration::from_secs(60)).await; + continue; + } + Err(e) => { + error!("ohlcv rebuild (min trade ts): {e}"); + tokio::time::sleep(Duration::from_secs(60)).await; + continue; + } + } + } + Err(e) => { + error!("ohlcv rebuild: {e}"); + tokio::time::sleep(Duration::from_secs(60)).await; + continue; + } + }; + + let mut batch_start = since; + while batch_start < cutoff { + let batch_end = (batch_start + BATCH_SECS).min(cutoff); + match db::cauldron::ohlcv::rebuild_range(&ohlcv_read, &ohlcv_write, batch_start, batch_end) + .await + { + Ok(n) => { + info!("ohlcv: materialised {n} buckets [{batch_start}, {batch_end})"); + ohlcv_state_bg + .materialized_end + .store(batch_end, Ordering::Relaxed); + } + Err(e) => { + error!("ohlcv rebuild failed: {e}"); + break; + } + } + batch_start = batch_end; + // Yield between batches so block indexing writes can proceed. + tokio::time::sleep(Duration::from_millis(200)).await; + } + + tokio::time::sleep(Duration::from_secs(600)).await; + } + }); + } + rocket::build() .attach(signal::ShutdownFairing) .attach(signal::IbdCheckFairing) .manage(dbpool) .manage(ibd_state) + .manage(ohlcv_state) + .manage(cache::new_candlestick_cache()) // give rocket ownership of downloader to ensure thread isn't dropped .manage(bcmrdownloader) .manage(wellknowndownloader) diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs index 28987a3..8243305 100644 --- a/src/rpc/candlesticks.rs +++ b/src/rpc/candlesticks.rs @@ -3,11 +3,14 @@ // 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 crate::cache::CandlestickCache; use crate::db::blob::display_hex_to_blob; +use crate::db::cauldron::ohlcv; use crate::db::DB; use crate::rpc::err::{bad_request, ApiErrorCode, CachedApiResult}; use crate::rpc::response::{cached_ok, CACHE_IMMUTABLE, CACHE_NONE}; use crate::timeutil::time_now; +use crate::OhlcvState; use anyhow::{bail, Result}; use bitcoincash::TokenID; use rocket::{get, State}; @@ -15,6 +18,8 @@ use serde::Serialize; use serde_json::json; use serde_json::Value; use sqlx::{Row, SqlitePool}; +use std::sync::atomic::Ordering; +use std::sync::Arc; #[derive(Debug, Serialize)] pub struct CandlestickData { @@ -77,91 +82,15 @@ impl PriceInterval { } } -pub async fn candlesticks( - pool: &SqlitePool, - timestamp_start: i64, - timestamp_end: i64, +/// Aggregate a sorted list of raw trades into candlestick intervals. +/// Returns the filled candles and the last close price seen (for continuing into a tail query). +fn aggregate_raw_trades( + all_trades: &[(i64, i64, i64, i64, i64)], + intervals: Vec, step_size: i64, - token_id: &str, -) -> Result> { - if timestamp_start > timestamp_end { - bail!("Start cannot be higher than end"); - } - - let mut intervals = Vec::new(); - let mut current_start = timestamp_start; - while current_start < timestamp_end { - intervals.push(PriceInterval::new(current_start, step_size)); - current_start += step_size; - } - - let sql = r#" -WITH per_pool_tx_raw AS ( - SELECT - tx.txid, - COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) AS effective_timestamp, - phe.utxo, - phe.sats_delta, - phe.token_delta - FROM pool_history_entry AS phe - JOIN utxo_funding AS uf ON phe.utxo = uf.new_utxo_hash - JOIN tx ON tx.txid = phe.txid - WHERE uf.token_id = ? -), -per_pool_tx AS ( - SELECT - txid, - effective_timestamp, - utxo, - SUM(sats_delta) AS signed_sats_pool, - SUM(token_delta) AS signed_tokens_pool, - SUM(ABS(sats_delta)) AS volume_sats_pool, - SUM(ABS(token_delta)) AS volume_tokens_pool - FROM per_pool_tx_raw - WHERE effective_timestamp >= ? AND effective_timestamp < ? - GROUP BY txid, effective_timestamp, utxo -), -tx_trades AS ( - SELECT - txid, - effective_timestamp, - SUM(signed_sats_pool) AS signed_sats, - SUM(signed_tokens_pool) AS signed_tokens, - SUM(volume_sats_pool) AS volume_sats, - SUM(volume_tokens_pool) AS volume_tokens - FROM per_pool_tx - GROUP BY txid, effective_timestamp -) -SELECT - effective_timestamp, - signed_sats, - signed_tokens, - volume_sats, - volume_tokens -FROM tx_trades -ORDER BY effective_timestamp ASC; -"#; - // Use proper blob conversion for Bitcoin hash types (handles byte reversal) - let token_blob = display_hex_to_blob::(token_id)?; - let rows = sqlx::query(sql) - .bind(token_blob) - .bind(timestamp_start) - .bind(timestamp_end) - .fetch_all(pool) - .await?; - - let mut all_trades = Vec::new(); - for row in rows { - let timestamp: i64 = row.get(0); - let signed_sats: i64 = row.get(1); - let signed_tokens: i64 = row.get(2); - let vol_sats: i64 = row.get(3); - let vol_tokens: i64 = row.get(4); - all_trades.push((timestamp, signed_sats, signed_tokens, vol_sats, vol_tokens)); - } - - let mut found_first_trade = false; - let mut last_close_price: Option = None; + mut found_first_trade: bool, + mut last_close_price: Option, +) -> (Vec, bool, Option) { let mut result = Vec::with_capacity(intervals.len()); let mut trade_index = 0; @@ -183,9 +112,7 @@ ORDER BY effective_timestamp ASC; } if signed_tokens != 0 { - // Price from signed ratio; make it positive for OHLC let price = (signed_sats as f64 / signed_tokens as f64).abs(); - if first_trade_in_interval { pi.open = Some(price); pi.high = price; @@ -199,17 +126,13 @@ ORDER BY effective_timestamp ASC; } } - // Accumulate volume using ABS deltas (already computed in SQL) pi.volume_sats += vol_sats; pi.volume_tokens += vol_tokens; - pi.transaction_count += 1; trade_index += 1; } - // If there were trades in this interval (volume), but net signed_tokens was 0 - // so we never set a price, carry forward the previous close so the candle - // keeps its (correct) volume and doesn't get dropped. + // Carry forward last close when volume exists but net tokens are zero. if pi.transaction_count > 0 && (pi.open.is_none() || pi.close.is_none()) { if let Some(prev) = last_close_price { if pi.open.is_none() { @@ -249,6 +172,193 @@ ORDER BY effective_timestamp ASC; } } + (result, found_first_trade, last_close_price) +} + +/// Fill in 1-hour candles from pre-materialised `ohlcv_1h` rows, adding flat gap-filler +/// candles between active buckets just like the raw path does. +/// Returns the filled candles and the last close price for continuing into a raw tail. +fn fill_ohlcv_candles( + rows: Vec, + start: i64, + end: i64, + mut found_first_trade: bool, + mut last_close: Option, +) -> (Vec, bool, Option) { + let mut result = Vec::new(); + let mut row_iter = rows.into_iter().peekable(); + let mut bucket = start; + + while bucket < end { + if row_iter.peek().map(|r| r.bucket_ts) == Some(bucket) { + let r = row_iter.next().unwrap(); + found_first_trade = true; + last_close = Some(r.close); + result.push(CandlestickData { + time: r.bucket_ts, + open: r.open, + close: r.close, + high: r.high, + low: r.low, + volume_sats: r.volume_sats, + volume_tokens: r.volume_tokens, + transaction_count: r.tx_count, + }); + } else if found_first_trade { + if let Some(prev) = last_close { + result.push(CandlestickData { + time: bucket, + open: prev, + close: prev, + high: prev, + low: prev, + volume_sats: 0, + volume_tokens: 0, + transaction_count: 0, + }); + } + } + bucket += 3600; + } + + (result, found_first_trade, last_close) +} + +async fn fetch_raw_trades( + pool: &SqlitePool, + token_blob: &[u8], + timestamp_start: i64, + timestamp_end: i64, +) -> Result> { + let sql = r#" +WITH per_pool_tx_raw AS ( + SELECT + phe.txid, + phe.effective_timestamp, + phe.utxo, + phe.sats_delta, + phe.token_delta + FROM pool_history_entry AS phe + WHERE phe.token_id = ? + AND phe.effective_timestamp >= ? + AND phe.effective_timestamp < ? +), +per_pool_tx AS ( + SELECT + txid, + effective_timestamp, + utxo, + SUM(sats_delta) AS signed_sats_pool, + SUM(token_delta) AS signed_tokens_pool, + SUM(ABS(sats_delta)) AS volume_sats_pool, + SUM(ABS(token_delta)) AS volume_tokens_pool + FROM per_pool_tx_raw + GROUP BY txid, effective_timestamp, utxo +), +tx_trades AS ( + SELECT + txid, + effective_timestamp, + SUM(signed_sats_pool) AS signed_sats, + SUM(signed_tokens_pool) AS signed_tokens, + SUM(volume_sats_pool) AS volume_sats, + SUM(volume_tokens_pool) AS volume_tokens + FROM per_pool_tx + GROUP BY txid, effective_timestamp +) +SELECT + effective_timestamp, + signed_sats, + signed_tokens, + volume_sats, + volume_tokens +FROM tx_trades +ORDER BY effective_timestamp ASC; +"#; + let rows = sqlx::query(sql) + .bind(token_blob) + .bind(timestamp_start) + .bind(timestamp_end) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|r| (r.get(0), r.get(1), r.get(2), r.get(3), r.get(4))) + .collect()) +} + +/// `ohlcv_materialized_end`: exclusive upper bound of what is in `ohlcv_1h`. +/// Pass 0 to always use the raw CTE path. +pub async fn candlesticks( + pool: &SqlitePool, + timestamp_start: i64, + timestamp_end: i64, + step_size: i64, + token_id: &str, + ohlcv_materialized_end: i64, +) -> Result> { + if timestamp_start > timestamp_end { + bail!("Start cannot be higher than end"); + } + + let token_blob = display_hex_to_blob::(token_id)?; + + // Fast path: use pre-materialised ohlcv_1h when step_size is exactly 1 hour, + // ohlcv covers at least part of the range, AND the start is hour-aligned. + // ohlcv_1h buckets are always aligned to multiples of 3600, so a non-aligned + // start would produce candles whose boundaries disagree with the raw path. + if step_size == 3600 + && ohlcv_materialized_end > timestamp_start + && timestamp_start % 3600 == 0 + { + let ohlcv_end = ohlcv_materialized_end.min(timestamp_end); + + // Align start to 1-hour bucket boundary for the ohlcv query. + let ohlcv_start = timestamp_start / 3600 * 3600; + + let ohlcv_rows = + ohlcv::get_active_candles(pool, &token_blob, ohlcv_start, ohlcv_end).await?; + + let (mut result, found_first, last_close) = + fill_ohlcv_candles(ohlcv_rows, ohlcv_start, ohlcv_end, false, None); + + // If the requested start doesn't align to ohlcv_start, trim leading flat candles + // that fall before timestamp_start. + result.retain(|c| c.time >= timestamp_start); + + if ohlcv_end < timestamp_end { + // Tail: query raw for [ohlcv_end, timestamp_end) and append. + let raw_trades = + fetch_raw_trades(pool, &token_blob, ohlcv_end, timestamp_end).await?; + + let mut tail_intervals = Vec::new(); + let mut t = ohlcv_end; + while t < timestamp_end { + tail_intervals.push(PriceInterval::new(t, step_size)); + t += step_size; + } + + let (tail, _, _) = + aggregate_raw_trades(&raw_trades, tail_intervals, step_size, found_first, last_close); + result.extend(tail); + } + + return Ok(result); + } + + // Raw path: full CTE scan (all non-3600 step sizes, or when ohlcv is not ready). + let mut intervals = Vec::new(); + let mut current_start = timestamp_start; + while current_start < timestamp_end { + intervals.push(PriceInterval::new(current_start, step_size)); + current_start += step_size; + } + + let all_trades = + fetch_raw_trades(pool, &token_blob, timestamp_start, timestamp_end).await?; + + let (result, _, _) = aggregate_raw_trades(&all_trades, intervals, step_size, false, None); Ok(result) } @@ -285,6 +395,8 @@ pub async fn price_candlesticks( end: Option, stepsize: Option, conn: &State, + cache: &State, + ohlcv: &State>, ) -> CachedApiResult { let current_timestamp = time_now(); @@ -324,15 +436,49 @@ pub async fn price_candlesticks( )); } - let candlestick_data = candlesticks( - &conn.cauldron_r, + // Only cache windows that are fully in the past (identical to the HTTP cache rule). + let is_historical = end.is_some() && effective_end < current_timestamp - effective_stepsize; + let cache_key = ( + token.to_string(), effective_start, effective_end, effective_stepsize, - token, - ) - .await - .map_err(|e| bad_request(ApiErrorCode::InvalidParameters, &e.to_string()))?; + ); + + let ohlcv_end = ohlcv.materialized_end.load(Ordering::Relaxed); + + let candlestick_data: Arc> = if is_historical { + if let Some(hit) = cache.get(&cache_key).await { + hit + } else { + let data = candlesticks( + &conn.cauldron_r, + effective_start, + effective_end, + effective_stepsize, + token, + ohlcv_end, + ) + .await + .map_err(|e| bad_request(ApiErrorCode::InvalidParameters, &e.to_string()))?; + let arc = Arc::new(data); + cache.insert(cache_key, arc.clone()).await; + arc + } + } else { + Arc::new( + candlesticks( + &conn.cauldron_r, + effective_start, + effective_end, + effective_stepsize, + token, + ohlcv_end, + ) + .await + .map_err(|e| bad_request(ApiErrorCode::InvalidParameters, &e.to_string()))?, + ) + }; let candlesticks_json: Vec = candlestick_data .iter() @@ -350,8 +496,7 @@ pub async fn price_candlesticks( }) .collect(); - let cache_duration = if end.is_some() && effective_end < current_timestamp - effective_stepsize - { + let cache_duration = if is_historical { CACHE_IMMUTABLE } else { CACHE_NONE @@ -365,12 +510,15 @@ pub async fn price_candlesticks( #[cfg(test)] mod tests { + use crate::cache::new_candlestick_cache; use crate::db::cauldron::{ + ohlcv, pool::{self, dummy_init_seq, insert_new_pool}, tx::{self, insert_block_tx, insert_mempool_tx}, utxo_funding::{self, insert_utxo_funding}, }; use crate::utiltest::mock_db_pool; + use crate::OhlcvState; use crate::timeutil::time_now; use bitcoin_hashes::Hash; @@ -379,6 +527,14 @@ mod tests { use rocket::http::Status; use rocket::local::asynchronous::Client; use rocket::routes; + use std::sync::atomic::AtomicI64; + use std::sync::Arc; + + fn no_ohlcv() -> Arc { + Arc::new(OhlcvState { + materialized_end: AtomicI64::new(0), + }) + } /// Four trades, each 600-second bin is 10 minutes. /// We'll have two bins: @@ -562,6 +718,8 @@ mod tests { let mock_db = mock_db_pool(setup_mock_db).await; let rocket = rocket::build() .manage(mock_db) + .manage(new_candlestick_cache()) + .manage(no_ohlcv()) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket) .await @@ -593,6 +751,8 @@ mod tests { let mock_db = mock_db_pool(setup_mock_db).await; let rocket = rocket::build() .manage(mock_db) + .manage(new_candlestick_cache()) + .manage(no_ohlcv()) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket) .await @@ -624,6 +784,8 @@ mod tests { let rocket = rocket::build() .manage(mock_db) + .manage(new_candlestick_cache()) + .manage(no_ohlcv()) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket) @@ -760,6 +922,8 @@ mod tests { // Build Rocket instance with our endpoint let rocket = rocket::build() .manage(mock_db) + .manage(new_candlestick_cache()) + .manage(no_ohlcv()) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket) .await @@ -957,6 +1121,8 @@ mod tests { // Build Rocket and call endpoint across the two intervals let rocket = rocket::build() .manage(mock_db) + .manage(new_candlestick_cache()) + .manage(no_ohlcv()) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket) .await @@ -1004,4 +1170,192 @@ mod tests { assert_eq!(c2["volume_tokens"].as_i64().unwrap(), 400); assert_eq!(c2["transaction_count"].as_i64().unwrap(), 1); } + + // ── ohlcv_1h fast-path tests ───────────────────────────────────────────── + // + // Test data from setup_mock_db — four trades at: + // TIME_1=1727963400, TIME_2=1727963600, TIME_3=1727963900 → hour bucket 1727960400 + // TIME_4=1727964200 → hour bucket 1727964000 + // + // Expected OHLCV (bucket 1727960400): open=40, close=80, high=80, low=40 + // vol_sats=360_000 (80k+120k+160k), vol_tokens=6_000, tx_count=3 + // Expected OHLCV (bucket 1727964000): open=close=high=low=100 + // vol_sats=200_000, vol_tokens=2_000, tx_count=1 + + /// Full range served from ohlcv_1h (materialized_end covers everything). + #[rocket::async_test] + async fn test_ohlcv_fast_path_full_range() { + let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { + setup_mock_db(pool.clone()).await; + ohlcv::create_table(&pool).await; + ohlcv::rebuild_range(&pool, &pool, 0, 1727970000) + .await + .expect("rebuild_range"); + }) + .await; + + let ohlcv_state = Arc::new(OhlcvState { + materialized_end: AtomicI64::new(1727970000), + }); + + let rocket = rocket::build() + .manage(mock_db) + .manage(new_candlestick_cache()) + .manage(ohlcv_state) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket).await.expect("valid rocket"); + + let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; + + // Hour-aligned range covering both buckets. + let response = client + .get(format!( + "/api/price/{token_id_zero}/candlesticks\ + ?start=1727960400&end=1727967600&stepsize=3600" + )) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + let body = response.into_string().await.unwrap(); + let json: serde_json::Value = serde_json::from_str(&body).unwrap(); + let candles = json["candlesticks"].as_array().unwrap(); + + assert_eq!(candles.len(), 2, "expected two 1h candles"); + + let c1 = &candles[0]; + assert_eq!(c1["time"].as_i64().unwrap(), 1727960400); + assert!((c1["open"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); + assert!((c1["close"].as_f64().unwrap() - 80.0).abs() < f64::EPSILON); + assert!((c1["high"].as_f64().unwrap() - 80.0).abs() < f64::EPSILON); + assert!((c1["low"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); + assert_eq!(c1["volume_sats"].as_i64().unwrap(), 360_000); + assert_eq!(c1["volume_tokens"].as_i64().unwrap(), 6_000); + assert_eq!(c1["transaction_count"].as_i64().unwrap(), 3); + + let c2 = &candles[1]; + assert_eq!(c2["time"].as_i64().unwrap(), 1727964000); + assert!((c2["open"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); + assert!((c2["close"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); + assert_eq!(c2["volume_sats"].as_i64().unwrap(), 200_000); + assert_eq!(c2["volume_tokens"].as_i64().unwrap(), 2_000); + assert_eq!(c2["transaction_count"].as_i64().unwrap(), 1); + } + + /// First bucket served from ohlcv_1h, second bucket served from raw CTE tail. + /// Both sources should produce identical output to the full-ohlcv test above. + #[rocket::async_test] + async fn test_ohlcv_fast_path_with_raw_tail() { + let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { + setup_mock_db(pool.clone()).await; + ohlcv::create_table(&pool).await; + // Only materialise up to 1727964000 — leaves the second bucket (T4) for raw. + ohlcv::rebuild_range(&pool, &pool, 0, 1727964000) + .await + .expect("rebuild_range"); + }) + .await; + + let ohlcv_state = Arc::new(OhlcvState { + // materialized_end = 1727964000: ohlcv has bucket 1727960400 only. + materialized_end: AtomicI64::new(1727964000), + }); + + let rocket = rocket::build() + .manage(mock_db) + .manage(new_candlestick_cache()) + .manage(ohlcv_state) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket).await.expect("valid rocket"); + + let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; + + let response = client + .get(format!( + "/api/price/{token_id_zero}/candlesticks\ + ?start=1727960400&end=1727967600&stepsize=3600" + )) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + let body = response.into_string().await.unwrap(); + let json: serde_json::Value = serde_json::from_str(&body).unwrap(); + let candles = json["candlesticks"].as_array().unwrap(); + + assert_eq!(candles.len(), 2, "expected two 1h candles"); + + // Candle 1 came from ohlcv_1h. + let c1 = &candles[0]; + assert_eq!(c1["time"].as_i64().unwrap(), 1727960400); + assert!((c1["open"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); + assert!((c1["close"].as_f64().unwrap() - 80.0).abs() < f64::EPSILON); + assert_eq!(c1["volume_sats"].as_i64().unwrap(), 360_000); + assert_eq!(c1["volume_tokens"].as_i64().unwrap(), 6_000); + assert_eq!(c1["transaction_count"].as_i64().unwrap(), 3); + + // Candle 2 came from the raw CTE tail. + let c2 = &candles[1]; + assert_eq!(c2["time"].as_i64().unwrap(), 1727964000); + assert!((c2["open"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); + assert!((c2["close"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); + assert_eq!(c2["volume_sats"].as_i64().unwrap(), 200_000); + assert_eq!(c2["volume_tokens"].as_i64().unwrap(), 2_000); + assert_eq!(c2["transaction_count"].as_i64().unwrap(), 1); + } + + /// Non-hour-aligned start must NOT use the ohlcv fast path (alignment guard). + #[rocket::async_test] + async fn test_ohlcv_skipped_for_non_aligned_start() { + let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { + setup_mock_db(pool.clone()).await; + ohlcv::create_table(&pool).await; + ohlcv::rebuild_range(&pool, &pool, 0, 1727970000) + .await + .expect("rebuild_range"); + }) + .await; + + let ohlcv_state = Arc::new(OhlcvState { + materialized_end: AtomicI64::new(1727970000), + }); + + let rocket = rocket::build() + .manage(mock_db) + .manage(new_candlestick_cache()) + .manage(ohlcv_state) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket).await.expect("valid rocket"); + + let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; + + // Start is NOT hour-aligned (1727963300 % 3600 != 0) — must fall back to raw. + // With stepsize=3600 the first interval is [1727963300, 1727966900). + // All four trades (T1–T4) fall within this single interval: + // candle 1 at 1727963300: open=40 (T1 first), close=100 (T4 last), tx_count=4 + // candle 2 at 1727966900: flat carry-forward at 100 (no trades) + let response = client + .get(format!( + "/api/price/{token_id_zero}/candlesticks\ + ?start=1727963300&end=1727970500&stepsize=3600" + )) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + let body = response.into_string().await.unwrap(); + let json: serde_json::Value = serde_json::from_str(&body).unwrap(); + let candles = json["candlesticks"].as_array().unwrap(); + + // Key assertion: first candle starts at 1727963300, not 1727960400. + // If the ohlcv path were mistakenly used it would start at 1727960400. + assert_eq!(candles.len(), 2); + assert_eq!(candles[0]["time"].as_i64().unwrap(), 1727963300); + assert!((candles[0]["open"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); + assert!((candles[0]["close"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); + assert_eq!(candles[0]["transaction_count"].as_i64().unwrap(), 4); + // Flat carry-forward candle (no trades in second interval). + assert_eq!(candles[1]["time"].as_i64().unwrap(), 1727966900); + assert_eq!(candles[1]["transaction_count"].as_i64().unwrap(), 0); + } } From b0929a5dd4e5d104879bed483b4fd97dd1ed6b7e Mon Sep 17 00:00:00 2001 From: Jakob Notland Date: Thu, 19 Mar 2026 13:46:16 +0100 Subject: [PATCH 2/9] Checkpoint --- src/cache.rs | 24 ++ src/db/cauldron/ohlcv.rs | 465 +++++++++++++++++++++++++++++++++++++++ src/db/cauldron/tx.rs | 17 +- src/rpc/candlesticks.rs | 7 +- 4 files changed, 499 insertions(+), 14 deletions(-) create mode 100644 src/cache.rs create mode 100644 src/db/cauldron/ohlcv.rs diff --git a/src/cache.rs b/src/cache.rs new file mode 100644 index 0000000..f2adadc --- /dev/null +++ b/src/cache.rs @@ -0,0 +1,24 @@ +// 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 + +use crate::rpc::candlesticks::CandlestickData; +use moka::future::Cache; +use std::sync::Arc; + +/// Cache key: (token_display_hex, start_ts, end_ts, stepsize_secs) +type CandleCacheKey = (String, i64, i64, i64); + +pub type CandlestickCache = Cache>>; + +/// Max total candles stored across all entries. +/// Each CandlestickData is ~72 bytes; 500_000 candles ≈ 36 MB upper bound. +const MAX_CANDLE_CAPACITY: u64 = 500_000; + +pub fn new_candlestick_cache() -> CandlestickCache { + Cache::builder() + .max_capacity(MAX_CANDLE_CAPACITY) + .weigher(|_k, v: &Arc>| v.len().max(1) as u32) + .build() +} diff --git a/src/db/cauldron/ohlcv.rs b/src/db/cauldron/ohlcv.rs new file mode 100644 index 0000000..0848670 --- /dev/null +++ b/src/db/cauldron/ohlcv.rs @@ -0,0 +1,465 @@ +// 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 + +use anyhow::Result; +use sqlx::{Row, SqlitePool}; + +pub async fn create_table(pool: &SqlitePool) { + sqlx::query( + "CREATE TABLE IF NOT EXISTS ohlcv_1h ( + token_id BLOB NOT NULL, + bucket_ts INTEGER NOT NULL, + open REAL NOT NULL, + high REAL NOT NULL, + low REAL NOT NULL, + close REAL NOT NULL, + volume_sats INTEGER NOT NULL, + volume_tokens INTEGER NOT NULL, + tx_count INTEGER NOT NULL, + PRIMARY KEY (token_id, bucket_ts) + )", + ) + .execute(pool) + .await + .expect("failed to create ohlcv_1h table"); + + // Needed for fast joins in rebuild_range; use IF NOT EXISTS so this is safe on existing DBs. + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_phe_txid ON pool_history_entry(txid)", + ) + .execute(pool) + .await + .expect("failed to create idx_phe_txid index"); + + // Composite index for fast token+time range scans in the candlestick raw CTE. + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_phe_token_id_ts ON pool_history_entry(token_id, effective_timestamp)", + ) + .execute(pool) + .await + .expect("failed to create idx_phe_token_id_ts index"); + + // Index for tx_latest token-filtered query: JOIN utxo_funding WHERE token_id = ? → ORDER BY tx.effective_timestamp. + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_utxo_funding_token_txid ON utxo_funding(token_id, txid)", + ) + .execute(pool) + .await + .expect("failed to create idx_utxo_funding_token_txid index"); +} + +/// Returns the highest `bucket_ts` in `ohlcv_1h`, or `None` if the table is empty. +pub async fn get_max_bucket_ts(pool: &SqlitePool) -> Result> { + let row: Option<(Option,)> = sqlx::query_as("SELECT MAX(bucket_ts) FROM ohlcv_1h") + .fetch_optional(pool) + .await?; + Ok(row.and_then(|r| r.0)) +} + +/// Returns the earliest confirmed trade timestamp floored to the nearest 1-hour bucket, +/// or `None` if there are no confirmed trades. Used to seed the initial backfill start +/// so the background task doesn't scan from Unix epoch 0. +pub async fn get_min_trade_bucket_ts(pool: &SqlitePool) -> Result> { + let row: Option<(Option,)> = sqlx::query_as( + "SELECT (MIN(tx.effective_timestamp) / 3600) * 3600 + FROM pool_history_entry AS phe + JOIN tx ON tx.txid = phe.txid + WHERE tx.blockhash IS NOT NULL", + ) + .fetch_optional(pool) + .await?; + Ok(row.and_then(|r| r.0)) +} + +/// Materialise all 1-hour OHLCV buckets for confirmed trades whose effective timestamp falls +/// in `[since_ts, until_ts)`. +/// +/// Two-phase approach: the slow aggregation SELECT runs against `read_pool` (no write lock), +/// then the pre-computed rows are bulk-inserted via `write_pool` (write lock held briefly). +/// Uses INSERT OR IGNORE so existing rows are never overwritten. +/// Returns the number of rows inserted. +pub async fn rebuild_range( + read_pool: &SqlitePool, + write_pool: &SqlitePool, + since_ts: i64, + until_ts: i64, +) -> Result { + if since_ts >= until_ts { + return Ok(0); + } + + // Phase 1: aggregate using the read pool — no write lock held during the slow CTE. + let select_sql = r#" +WITH per_pool_tx_raw AS ( + SELECT + uf.token_id, + phe.txid, + tx.effective_timestamp AS ts, + phe.utxo, + phe.sats_delta, + phe.token_delta + FROM pool_history_entry AS phe + JOIN utxo_funding AS uf ON phe.utxo = uf.new_utxo_hash + JOIN tx ON tx.txid = phe.txid + WHERE tx.blockhash IS NOT NULL + AND tx.effective_timestamp >= ? + AND tx.effective_timestamp < ? +), +per_pool_tx AS ( + SELECT + token_id, + txid, + ts, + (ts / 3600) * 3600 AS bucket_ts, + utxo, + SUM(sats_delta) AS signed_sats, + SUM(token_delta) AS signed_tokens, + SUM(ABS(sats_delta)) AS vol_sats, + SUM(ABS(token_delta)) AS vol_tokens + FROM per_pool_tx_raw + GROUP BY token_id, txid, ts, utxo +), +tx_trades AS ( + SELECT + token_id, + txid, + ts, + bucket_ts, + SUM(signed_sats) AS signed_sats, + SUM(signed_tokens) AS signed_tokens, + SUM(vol_sats) AS vol_sats, + SUM(vol_tokens) AS vol_tokens + FROM per_pool_tx + GROUP BY token_id, txid, ts +), +priceable AS ( + SELECT + token_id, + bucket_ts, + ABS(CAST(signed_sats AS REAL) / CAST(signed_tokens AS REAL)) AS price, + ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts ASC, txid ASC) AS rn_asc, + ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts DESC, txid DESC) AS rn_desc + FROM tx_trades + WHERE signed_tokens != 0 +), +ohlc AS ( + SELECT + token_id, + bucket_ts, + MAX(CASE WHEN rn_asc = 1 THEN price END) AS open, + MAX(CASE WHEN rn_desc = 1 THEN price END) AS close, + MAX(price) AS high, + MIN(price) AS low + FROM priceable + GROUP BY token_id, bucket_ts +), +vol AS ( + SELECT + token_id, + bucket_ts, + SUM(vol_sats) AS volume_sats, + SUM(vol_tokens) AS volume_tokens, + COUNT(*) AS tx_count + FROM tx_trades + GROUP BY token_id, bucket_ts +) +SELECT + ohlc.token_id, + ohlc.bucket_ts, + ohlc.open, + ohlc.high, + ohlc.low, + ohlc.close, + vol.volume_sats, + vol.volume_tokens, + vol.tx_count +FROM ohlc +JOIN vol ON ohlc.token_id = vol.token_id AND ohlc.bucket_ts = vol.bucket_ts +"#; + + let rows = sqlx::query(select_sql) + .bind(since_ts) + .bind(until_ts) + .fetch_all(read_pool) + .await?; + + if rows.is_empty() { + return Ok(0); + } + + // Phase 2: insert pre-computed rows inside a single transaction. + // The write lock is held only for these fast INSERTs, not during aggregation. + let mut tx = write_pool.begin().await?; + let mut inserted = 0u64; + for row in &rows { + let token_id: Vec = row.get(0); + let bucket_ts: i64 = row.get(1); + let open: f64 = row.get(2); + let high: f64 = row.get(3); + let low: f64 = row.get(4); + let close: f64 = row.get(5); + let volume_sats: i64 = row.get(6); + let volume_tokens: i64 = row.get(7); + let tx_count: i64 = row.get(8); + + inserted += sqlx::query( + "INSERT OR IGNORE INTO ohlcv_1h + (token_id, bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind(token_id) + .bind(bucket_ts) + .bind(open) + .bind(high) + .bind(low) + .bind(close) + .bind(volume_sats) + .bind(volume_tokens) + .bind(tx_count) + .execute(&mut *tx) + .await? + .rows_affected(); + } + tx.commit().await?; + + Ok(inserted) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::cauldron::{pool as cauldron_pool, tx, utxo_funding}; + use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; + use std::sync::atomic::{AtomicU64, Ordering}; + + static OHLCV_TEST_COUNTER: AtomicU64 = AtomicU64::new(0); + + async fn test_pool() -> SqlitePool { + let id = OHLCV_TEST_COUNTER.fetch_add(1, Ordering::SeqCst); + let uri = format!("file:ohlcv_test_{}?mode=memory&cache=shared", id); + let opts = SqliteConnectOptions::new() + .filename(&uri) + .foreign_keys(false); + SqlitePoolOptions::new().connect_with(opts).await.unwrap() + } + + async fn setup_db(pool: &SqlitePool) { + tx::create_table(pool).await; + utxo_funding::create_table(pool).await; + cauldron_pool::create_table(pool).await; + create_table(pool).await; // ohlcv_1h + idx_phe_txid + } + + /// Insert a single confirmed trade via raw SQL (FK disabled in tests). + async fn insert_confirmed_trade( + pool: &SqlitePool, + txid: [u8; 32], + utxo: [u8; 32], + token_id: [u8; 32], + mtp_ts: i64, + sats_delta: i64, + token_delta: i64, + ) { + let blockhash = [0xAA_u8; 32]; + sqlx::query("INSERT INTO tx (txid, blockhash, mtp_timestamp) VALUES (?, ?, ?)") + .bind(txid.as_slice()) + .bind(blockhash.as_slice()) + .bind(mtp_ts) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO utxo_funding (new_utxo_hash, txid, spent_utxo_hash, new_utxo_txid, new_utxo_n, sats, token_amount, token_id) + VALUES (?, ?, ?, ?, 0, 1000, 1000, ?)", + ) + .bind(utxo.as_slice()) + .bind(txid.as_slice()) + .bind([0u8; 32].as_slice()) + .bind(txid.as_slice()) + .bind(token_id.as_slice()) + .execute(pool) + .await + .unwrap(); + let seq: i64 = + sqlx::query_scalar("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry") + .fetch_one(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO pool_history_entry + (utxo, pool, token_id, txid, tx_pos, mtp_timestamp, sequence, sats, token_amount, sats_delta, token_delta) + VALUES (?, ?, ?, ?, 0, ?, ?, 1000, 1000, ?, ?)", + ) + .bind(utxo.as_slice()) + .bind([0xBB_u8; 32].as_slice()) // dummy pool hash (FK disabled) + .bind(token_id.as_slice()) + .bind(txid.as_slice()) + .bind(mtp_ts) + .bind(seq) + .bind(sats_delta) + .bind(token_delta) + .execute(pool) + .await + .unwrap(); + } + + /// Insert a mempool-only trade (no blockhash on the tx row). + async fn insert_mempool_trade( + pool: &SqlitePool, + txid: [u8; 32], + utxo: [u8; 32], + token_id: [u8; 32], + first_seen_ts: i64, + ) { + sqlx::query("INSERT INTO tx (txid, first_seen_timestamp) VALUES (?, ?)") + .bind(txid.as_slice()) + .bind(first_seen_ts) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO utxo_funding (new_utxo_hash, txid, spent_utxo_hash, new_utxo_txid, new_utxo_n, sats, token_amount, token_id) + VALUES (?, ?, ?, ?, 0, 1000, 1000, ?)", + ) + .bind(utxo.as_slice()) + .bind(txid.as_slice()) + .bind([0u8; 32].as_slice()) + .bind(txid.as_slice()) + .bind(token_id.as_slice()) + .execute(pool) + .await + .unwrap(); + let seq: i64 = + sqlx::query_scalar("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry") + .fetch_one(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO pool_history_entry + (utxo, pool, token_id, txid, tx_pos, first_seen_timestamp, sequence, sats, token_amount, sats_delta, token_delta) + VALUES (?, ?, ?, ?, 0, ?, ?, 1000, 1000, -1000, 25)", + ) + .bind(utxo.as_slice()) + .bind([0xBB_u8; 32].as_slice()) + .bind(token_id.as_slice()) + .bind(txid.as_slice()) + .bind(first_seen_ts) + .bind(seq) + .execute(pool) + .await + .unwrap(); + } + + /// `get_min_trade_bucket_ts` should floor a mid-hour timestamp to the hour boundary. + #[tokio::test] + async fn test_get_min_trade_bucket_ts_floors_to_hour() { + let pool = test_pool().await; + setup_db(&pool).await; + + // Trade at 1727963400 — not hour-aligned; floor to 1727960400 + insert_confirmed_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25) + .await; + + let result = get_min_trade_bucket_ts(&pool).await.unwrap(); + assert_eq!( + result, + Some(1727960400), + "1727963400 should floor to 1727960400" + ); + } + + /// `get_min_trade_bucket_ts` returns `None` when only unconfirmed (mempool) trades exist. + #[tokio::test] + async fn test_get_min_trade_bucket_ts_no_confirmed_trades() { + let pool = test_pool().await; + setup_db(&pool).await; + + insert_mempool_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400).await; + + let result = get_min_trade_bucket_ts(&pool).await.unwrap(); + assert_eq!(result, None, "mempool-only trades must not be returned"); + } + + /// Running `rebuild_range` twice on the same range inserts nothing on the second call + /// because INSERT OR IGNORE skips rows that already exist. + #[tokio::test] + async fn test_rebuild_range_idempotent() { + let pool = test_pool().await; + setup_db(&pool).await; + + insert_confirmed_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25) + .await; + + let n1 = rebuild_range(&pool, &pool, 1727960400, 1727964000) + .await + .unwrap(); + assert!(n1 > 0, "first rebuild should insert at least one bucket"); + + let n2 = rebuild_range(&pool, &pool, 1727960400, 1727964000) + .await + .unwrap(); + assert_eq!(n2, 0, "second rebuild must insert nothing (INSERT OR IGNORE)"); + } + + /// Unconfirmed trades (tx.blockhash IS NULL) must not appear in `ohlcv_1h`. + #[tokio::test] + async fn test_rebuild_range_excludes_mempool() { + let pool = test_pool().await; + setup_db(&pool).await; + + insert_mempool_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400).await; + + let n = rebuild_range(&pool, &pool, 1727960400, 1727967600) + .await + .unwrap(); + assert_eq!(n, 0, "mempool trades must not be materialised"); + } +} + +pub struct OhlcvRow { + pub bucket_ts: i64, + pub open: f64, + pub high: f64, + pub low: f64, + pub close: f64, + pub volume_sats: i64, + pub volume_tokens: i64, + pub tx_count: i64, +} + +/// Return the materialised 1-hour candles for a single token in `[start, end)`. +/// Only buckets that had at least one trade are returned (gaps must be filled by the caller). +pub async fn get_active_candles( + pool: &SqlitePool, + token_blob: &[u8], + start: i64, + end: i64, +) -> Result> { + let rows = sqlx::query( + "SELECT bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count + FROM ohlcv_1h + WHERE token_id = ? AND bucket_ts >= ? AND bucket_ts < ? + ORDER BY bucket_ts ASC", + ) + .bind(token_blob) + .bind(start) + .bind(end) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|r| OhlcvRow { + bucket_ts: r.get(0), + open: r.get(1), + high: r.get(2), + low: r.get(3), + close: r.get(4), + volume_sats: r.get(5), + volume_tokens: r.get(6), + tx_count: r.get(7), + }) + .collect()) +} diff --git a/src/db/cauldron/tx.rs b/src/db/cauldron/tx.rs index e4afda7..d2bba6c 100644 --- a/src/db/cauldron/tx.rs +++ b/src/db/cauldron/tx.rs @@ -75,17 +75,12 @@ pub async fn latest( ) -> Result, u64)>> { let (sql, token_blob) = match &token_id { Some(tid) => ( - "SELECT tx.txid, tx.blockhash, tx.mtp_timestamp, tx.first_seen_timestamp - FROM ( - SELECT txid, MAX(effective_timestamp) AS ts - FROM pool_history_entry - WHERE token_id = ?1 - GROUP BY txid - ORDER BY ts DESC - LIMIT ?2 OFFSET ?3 - ) AS recent - JOIN tx ON tx.txid = recent.txid - ORDER BY recent.ts DESC", + "SELECT DISTINCT tx.txid, tx.blockhash, tx.mtp_timestamp, tx.first_seen_timestamp + FROM tx + JOIN utxo_funding ON tx.txid = utxo_funding.txid + WHERE utxo_funding.token_id = ?1 + ORDER BY tx.effective_timestamp DESC + LIMIT ?2 OFFSET ?3", Some(tid.to_blob()), ), None => ( diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs index 8243305..16b8b05 100644 --- a/src/rpc/candlesticks.rs +++ b/src/rpc/candlesticks.rs @@ -234,14 +234,15 @@ async fn fetch_raw_trades( WITH per_pool_tx_raw AS ( SELECT phe.txid, - phe.effective_timestamp, + tx.effective_timestamp, phe.utxo, phe.sats_delta, phe.token_delta FROM pool_history_entry AS phe + JOIN tx ON tx.txid = phe.txid WHERE phe.token_id = ? - AND phe.effective_timestamp >= ? - AND phe.effective_timestamp < ? + AND tx.effective_timestamp >= ? + AND tx.effective_timestamp < ? ), per_pool_tx AS ( SELECT From a58a6decde136cd7c5956897f32d7ef2a852cbd3 Mon Sep 17 00:00:00 2001 From: Jakob Notland Date: Fri, 20 Mar 2026 09:45:30 +0100 Subject: [PATCH 3/9] Checkpoint --- src/rpc/candlesticks.rs | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs index 16b8b05..8243305 100644 --- a/src/rpc/candlesticks.rs +++ b/src/rpc/candlesticks.rs @@ -234,15 +234,14 @@ async fn fetch_raw_trades( WITH per_pool_tx_raw AS ( SELECT phe.txid, - tx.effective_timestamp, + phe.effective_timestamp, phe.utxo, phe.sats_delta, phe.token_delta FROM pool_history_entry AS phe - JOIN tx ON tx.txid = phe.txid WHERE phe.token_id = ? - AND tx.effective_timestamp >= ? - AND tx.effective_timestamp < ? + AND phe.effective_timestamp >= ? + AND phe.effective_timestamp < ? ), per_pool_tx AS ( SELECT From f6dfa3158fca2e56284ed08f917829e5394c72ab Mon Sep 17 00:00:00 2001 From: Jakob Notland Date: Fri, 20 Mar 2026 11:21:42 +0100 Subject: [PATCH 4/9] Correcter checkpoint --- src/db/cauldron/ohlcv.rs | 9 ++++++--- src/rpc/candlesticks.rs | 18 +++++++++++------- 2 files changed, 17 insertions(+), 10 deletions(-) diff --git a/src/db/cauldron/ohlcv.rs b/src/db/cauldron/ohlcv.rs index 0848670..93ffaf3 100644 --- a/src/db/cauldron/ohlcv.rs +++ b/src/db/cauldron/ohlcv.rs @@ -99,7 +99,8 @@ WITH per_pool_tx_raw AS ( tx.effective_timestamp AS ts, phe.utxo, phe.sats_delta, - phe.token_delta + phe.token_delta, + phe.sequence FROM pool_history_entry AS phe JOIN utxo_funding AS uf ON phe.utxo = uf.new_utxo_hash JOIN tx ON tx.txid = phe.txid @@ -114,6 +115,7 @@ per_pool_tx AS ( ts, (ts / 3600) * 3600 AS bucket_ts, utxo, + MIN(sequence) AS min_sequence, SUM(sats_delta) AS signed_sats, SUM(token_delta) AS signed_tokens, SUM(ABS(sats_delta)) AS vol_sats, @@ -127,6 +129,7 @@ tx_trades AS ( txid, ts, bucket_ts, + MIN(min_sequence) AS min_sequence, SUM(signed_sats) AS signed_sats, SUM(signed_tokens) AS signed_tokens, SUM(vol_sats) AS vol_sats, @@ -139,8 +142,8 @@ priceable AS ( token_id, bucket_ts, ABS(CAST(signed_sats AS REAL) / CAST(signed_tokens AS REAL)) AS price, - ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts ASC, txid ASC) AS rn_asc, - ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts DESC, txid DESC) AS rn_desc + ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts ASC, min_sequence ASC) AS rn_asc, + ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts DESC, min_sequence DESC) AS rn_desc FROM tx_trades WHERE signed_tokens != 0 ), diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs index 8243305..1e5f237 100644 --- a/src/rpc/candlesticks.rs +++ b/src/rpc/candlesticks.rs @@ -233,32 +233,36 @@ async fn fetch_raw_trades( let sql = r#" WITH per_pool_tx_raw AS ( SELECT - phe.txid, - phe.effective_timestamp, + tx.txid, + COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) AS effective_timestamp, phe.utxo, phe.sats_delta, - phe.token_delta + phe.token_delta, + phe.sequence FROM pool_history_entry AS phe - WHERE phe.token_id = ? - AND phe.effective_timestamp >= ? - AND phe.effective_timestamp < ? + JOIN utxo_funding AS uf ON phe.utxo = uf.new_utxo_hash + JOIN tx ON tx.txid = phe.txid + WHERE uf.token_id = ? ), per_pool_tx AS ( SELECT txid, effective_timestamp, utxo, + MIN(sequence) AS min_sequence, SUM(sats_delta) AS signed_sats_pool, SUM(token_delta) AS signed_tokens_pool, SUM(ABS(sats_delta)) AS volume_sats_pool, SUM(ABS(token_delta)) AS volume_tokens_pool FROM per_pool_tx_raw + WHERE effective_timestamp >= ? AND effective_timestamp < ? GROUP BY txid, effective_timestamp, utxo ), tx_trades AS ( SELECT txid, effective_timestamp, + MIN(min_sequence) AS min_sequence, SUM(signed_sats_pool) AS signed_sats, SUM(signed_tokens_pool) AS signed_tokens, SUM(volume_sats_pool) AS volume_sats, @@ -273,7 +277,7 @@ SELECT volume_sats, volume_tokens FROM tx_trades -ORDER BY effective_timestamp ASC; +ORDER BY effective_timestamp ASC, min_sequence ASC; "#; let rows = sqlx::query(sql) .bind(token_blob) From cf573472403a885c34033ddc560194c2917ff759 Mon Sep 17 00:00:00 2001 From: Jakob Notland Date: Fri, 20 Mar 2026 11:44:34 +0100 Subject: [PATCH 5/9] cleanup --- src/db/cauldron/ohlcv.rs | 9 ++++----- src/rpc/candlesticks.rs | 23 ++++++++--------------- 2 files changed, 12 insertions(+), 20 deletions(-) diff --git a/src/db/cauldron/ohlcv.rs b/src/db/cauldron/ohlcv.rs index 93ffaf3..365b0f2 100644 --- a/src/db/cauldron/ohlcv.rs +++ b/src/db/cauldron/ohlcv.rs @@ -94,19 +94,18 @@ pub async fn rebuild_range( let select_sql = r#" WITH per_pool_tx_raw AS ( SELECT - uf.token_id, + phe.token_id, phe.txid, - tx.effective_timestamp AS ts, + phe.effective_timestamp AS ts, phe.utxo, phe.sats_delta, phe.token_delta, phe.sequence FROM pool_history_entry AS phe - JOIN utxo_funding AS uf ON phe.utxo = uf.new_utxo_hash JOIN tx ON tx.txid = phe.txid WHERE tx.blockhash IS NOT NULL - AND tx.effective_timestamp >= ? - AND tx.effective_timestamp < ? + AND phe.effective_timestamp >= ? + AND phe.effective_timestamp < ? ), per_pool_tx AS ( SELECT diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs index 1e5f237..d793579 100644 --- a/src/rpc/candlesticks.rs +++ b/src/rpc/candlesticks.rs @@ -233,16 +233,16 @@ async fn fetch_raw_trades( let sql = r#" WITH per_pool_tx_raw AS ( SELECT - tx.txid, - COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) AS effective_timestamp, + phe.txid, + phe.effective_timestamp, phe.utxo, phe.sats_delta, phe.token_delta, phe.sequence FROM pool_history_entry AS phe - JOIN utxo_funding AS uf ON phe.utxo = uf.new_utxo_hash - JOIN tx ON tx.txid = phe.txid - WHERE uf.token_id = ? + WHERE phe.token_id = ? + AND phe.effective_timestamp >= ? + AND phe.effective_timestamp < ? ), per_pool_tx AS ( SELECT @@ -255,7 +255,6 @@ per_pool_tx AS ( SUM(ABS(sats_delta)) AS volume_sats_pool, SUM(ABS(token_delta)) AS volume_tokens_pool FROM per_pool_tx_raw - WHERE effective_timestamp >= ? AND effective_timestamp < ? GROUP BY txid, effective_timestamp, utxo ), tx_trades AS ( @@ -318,18 +317,12 @@ pub async fn candlesticks( { let ohlcv_end = ohlcv_materialized_end.min(timestamp_end); - // Align start to 1-hour bucket boundary for the ohlcv query. - let ohlcv_start = timestamp_start / 3600 * 3600; - + // timestamp_start is guaranteed hour-aligned by the entry condition above. let ohlcv_rows = - ohlcv::get_active_candles(pool, &token_blob, ohlcv_start, ohlcv_end).await?; + ohlcv::get_active_candles(pool, &token_blob, timestamp_start, ohlcv_end).await?; let (mut result, found_first, last_close) = - fill_ohlcv_candles(ohlcv_rows, ohlcv_start, ohlcv_end, false, None); - - // If the requested start doesn't align to ohlcv_start, trim leading flat candles - // that fall before timestamp_start. - result.retain(|c| c.time >= timestamp_start); + fill_ohlcv_candles(ohlcv_rows, timestamp_start, ohlcv_end, false, None); if ohlcv_end < timestamp_end { // Tail: query raw for [ohlcv_end, timestamp_end) and append. From 2273e042e12f052b6f8f5cbd7e1ad63651832df7 Mon Sep 17 00:00:00 2001 From: Jakob Notland Date: Mon, 23 Mar 2026 15:45:44 +0100 Subject: [PATCH 6/9] FMT and clippy --- src/db/cauldron/ohlcv.rs | 27 ++++++++++++++---------- src/main.rs | 45 ++++++++++++++++++++++++++-------------- src/rpc/candlesticks.rs | 19 +++++++++-------- 3 files changed, 55 insertions(+), 36 deletions(-) diff --git a/src/db/cauldron/ohlcv.rs b/src/db/cauldron/ohlcv.rs index 365b0f2..f5bb937 100644 --- a/src/db/cauldron/ohlcv.rs +++ b/src/db/cauldron/ohlcv.rs @@ -26,12 +26,10 @@ pub async fn create_table(pool: &SqlitePool) { .expect("failed to create ohlcv_1h table"); // Needed for fast joins in rebuild_range; use IF NOT EXISTS so this is safe on existing DBs. - sqlx::query( - "CREATE INDEX IF NOT EXISTS idx_phe_txid ON pool_history_entry(txid)", - ) - .execute(pool) - .await - .expect("failed to create idx_phe_txid index"); + sqlx::query("CREATE INDEX IF NOT EXISTS idx_phe_txid ON pool_history_entry(txid)") + .execute(pool) + .await + .expect("failed to create idx_phe_txid index"); // Composite index for fast token+time range scans in the candlestick raw CTE. sqlx::query( @@ -361,8 +359,10 @@ mod tests { setup_db(&pool).await; // Trade at 1727963400 — not hour-aligned; floor to 1727960400 - insert_confirmed_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25) - .await; + insert_confirmed_trade( + &pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25, + ) + .await; let result = get_min_trade_bucket_ts(&pool).await.unwrap(); assert_eq!( @@ -391,8 +391,10 @@ mod tests { let pool = test_pool().await; setup_db(&pool).await; - insert_confirmed_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25) - .await; + insert_confirmed_trade( + &pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25, + ) + .await; let n1 = rebuild_range(&pool, &pool, 1727960400, 1727964000) .await @@ -402,7 +404,10 @@ mod tests { let n2 = rebuild_range(&pool, &pool, 1727960400, 1727964000) .await .unwrap(); - assert_eq!(n2, 0, "second rebuild must insert nothing (INSERT OR IGNORE)"); + assert_eq!( + n2, 0, + "second rebuild must insert nothing (INSERT OR IGNORE)" + ); } /// Unconfirmed trades (tx.blockhash IS NULL) must not appear in `ohlcv_1h`. diff --git a/src/main.rs b/src/main.rs index c9e0aa3..d78bd69 100644 --- a/src/main.rs +++ b/src/main.rs @@ -374,16 +374,22 @@ async fn launch() -> _ { config }; - let (dbpool, bcmrdownloader, wellknowndownloader, crc20fetcher, ibd_state, indexing_in_progress) = - match start_program(config).await { - Ok(db) => db, - Err(e) => { - let backtrace = Backtrace::capture(); - error!("Backtrace (if RUST_BACKTRACE=1):\n{backtrace}"); - error!("Error: {e}"); - panic!("Failed at program startup") - } - }; + let ( + dbpool, + bcmrdownloader, + wellknowndownloader, + crc20fetcher, + ibd_state, + indexing_in_progress, + ) = match start_program(config).await { + Ok(db) => db, + Err(e) => { + let backtrace = Backtrace::capture(); + error!("Backtrace (if RUST_BACKTRACE=1):\n{backtrace}"); + error!("Error: {e}"); + panic!("Failed at program startup") + } + }; let allowed_origins = AllowedOrigins::all(); let cors = rocket_cors::CorsOptions { @@ -432,7 +438,7 @@ async fn launch() -> _ { indexing_in_progress.store(true, Ordering::Relaxed); info!("ohlcv: starting post-IBD full backfill"); - let now = crate::timeutil::time_now() as i64; + let now = crate::timeutil::time_now(); let cutoff = (now - BACKFILL_SAFETY_SECS) / 3600 * 3600; let since_opt = match db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r).await { Ok(Some(ts)) => Some(ts + 3600), @@ -445,7 +451,10 @@ async fn launch() -> _ { } } } - Err(e) => { warn!("ohlcv backfill: could not read max bucket ts: {e}"); None } + Err(e) => { + warn!("ohlcv backfill: could not read max bucket ts: {e}"); + None + } }; if since_opt.is_none() { info!("ohlcv backfill: no confirmed trades found, skipping"); @@ -496,8 +505,7 @@ async fn launch() -> _ { const SAFETY_SECS: i64 = 3 * 3600; loop { - - let now = crate::timeutil::time_now() as i64; + let now = crate::timeutil::time_now(); // Floor to 1-hour boundary, 3 hours ago. let cutoff = (now - SAFETY_SECS) / 3600 * 3600; @@ -530,8 +538,13 @@ async fn launch() -> _ { let mut batch_start = since; while batch_start < cutoff { let batch_end = (batch_start + BATCH_SECS).min(cutoff); - match db::cauldron::ohlcv::rebuild_range(&ohlcv_read, &ohlcv_write, batch_start, batch_end) - .await + match db::cauldron::ohlcv::rebuild_range( + &ohlcv_read, + &ohlcv_write, + batch_start, + batch_end, + ) + .await { Ok(n) => { info!("ohlcv: materialised {n} buckets [{batch_start}, {batch_end})"); diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs index d793579..2c4a51f 100644 --- a/src/rpc/candlesticks.rs +++ b/src/rpc/candlesticks.rs @@ -311,9 +311,7 @@ pub async fn candlesticks( // ohlcv covers at least part of the range, AND the start is hour-aligned. // ohlcv_1h buckets are always aligned to multiples of 3600, so a non-aligned // start would produce candles whose boundaries disagree with the raw path. - if step_size == 3600 - && ohlcv_materialized_end > timestamp_start - && timestamp_start % 3600 == 0 + if step_size == 3600 && ohlcv_materialized_end > timestamp_start && timestamp_start % 3600 == 0 { let ohlcv_end = ohlcv_materialized_end.min(timestamp_end); @@ -326,8 +324,7 @@ pub async fn candlesticks( if ohlcv_end < timestamp_end { // Tail: query raw for [ohlcv_end, timestamp_end) and append. - let raw_trades = - fetch_raw_trades(pool, &token_blob, ohlcv_end, timestamp_end).await?; + let raw_trades = fetch_raw_trades(pool, &token_blob, ohlcv_end, timestamp_end).await?; let mut tail_intervals = Vec::new(); let mut t = ohlcv_end; @@ -336,8 +333,13 @@ pub async fn candlesticks( t += step_size; } - let (tail, _, _) = - aggregate_raw_trades(&raw_trades, tail_intervals, step_size, found_first, last_close); + let (tail, _, _) = aggregate_raw_trades( + &raw_trades, + tail_intervals, + step_size, + found_first, + last_close, + ); result.extend(tail); } @@ -352,8 +354,7 @@ pub async fn candlesticks( current_start += step_size; } - let all_trades = - fetch_raw_trades(pool, &token_blob, timestamp_start, timestamp_end).await?; + let all_trades = fetch_raw_trades(pool, &token_blob, timestamp_start, timestamp_end).await?; let (result, _, _) = aggregate_raw_trades(&all_trades, intervals, step_size, false, None); Ok(result) From d605a258fd4ef7c08e70e623fcbb4e3e3848da82 Mon Sep 17 00:00:00 2001 From: Jakob Notland Date: Mon, 23 Mar 2026 17:15:05 +0100 Subject: [PATCH 7/9] Optimise and remove unecessary moka in memory cache and dependency --- Cargo.lock | 66 ----------------------------------------- Cargo.toml | 1 - src/cache.rs | 24 --------------- src/main.rs | 2 -- src/rpc/candlesticks.rs | 63 ++++++++------------------------------- 5 files changed, 12 insertions(+), 144 deletions(-) delete mode 100644 src/cache.rs diff --git a/Cargo.lock b/Cargo.lock index 7567bfd..7f06f30 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -73,17 +73,6 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" -[[package]] -name = "async-lock" -version = "3.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" -dependencies = [ - "event-listener", - "event-listener-strategy", - "pin-project-lite", -] - [[package]] name = "async-stream" version = "0.3.6" @@ -475,15 +464,6 @@ version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" -[[package]] -name = "crossbeam-channel" -version = "0.5.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2" -dependencies = [ - "crossbeam-utils", -] - [[package]] name = "crossbeam-deque" version = "0.8.6" @@ -695,16 +675,6 @@ dependencies = [ "pin-project-lite", ] -[[package]] -name = "event-listener-strategy" -version = "0.5.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" -dependencies = [ - "event-listener", - "pin-project-lite", -] - [[package]] name = "fastrand" version = "2.3.0" @@ -1590,26 +1560,6 @@ dependencies = [ "windows-sys 0.52.0", ] -[[package]] -name = "moka" -version = "0.12.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85f8024e1c8e71c778968af91d43700ce1d11b219d127d79fb2934153b82b42b" -dependencies = [ - "async-lock", - "crossbeam-channel", - "crossbeam-epoch", - "crossbeam-utils", - "equivalent", - "event-listener", - "futures-util", - "parking_lot", - "portable-atomic", - "smallvec", - "tagptr", - "uuid", -] - [[package]] name = "multer" version = "3.1.0" @@ -1843,12 +1793,6 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" -[[package]] -name = "portable-atomic" -version = "1.13.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49" - [[package]] name = "potential_utf" version = "0.1.2" @@ -2234,7 +2178,6 @@ dependencies = [ "hex", "log", "malachite", - "moka", "rand 0.8.5", "rayon", "reqwest", @@ -3012,12 +2955,6 @@ dependencies = [ "syn 2.0.101", ] -[[package]] -name = "tagptr" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" - [[package]] name = "tap" version = "1.0.1" @@ -3489,9 +3426,6 @@ name = "uuid" version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "458f7a779bf54acc9f347480ac654f68407d3aab21269a6e3c9f922acd9e2da9" -dependencies = [ - "getrandom 0.3.3", -] [[package]] name = "valuable" diff --git a/Cargo.toml b/Cargo.toml index 19b1faf..85f42a3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,6 @@ reqwest = { version = "0.12", features = ["json", "rustls-tls"], default-feature rand = "0.8.5" configure_me = "0.4.0" futures = "0.3.32" -moka = { version = "0.12", features = ["future"] } [build-dependencies] configure_me_codegen = "0.4.8" diff --git a/src/cache.rs b/src/cache.rs deleted file mode 100644 index f2adadc..0000000 --- a/src/cache.rs +++ /dev/null @@ -1,24 +0,0 @@ -// 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 - -use crate::rpc::candlesticks::CandlestickData; -use moka::future::Cache; -use std::sync::Arc; - -/// Cache key: (token_display_hex, start_ts, end_ts, stepsize_secs) -type CandleCacheKey = (String, i64, i64, i64); - -pub type CandlestickCache = Cache>>; - -/// Max total candles stored across all entries. -/// Each CandlestickData is ~72 bytes; 500_000 candles ≈ 36 MB upper bound. -const MAX_CANDLE_CAPACITY: u64 = 500_000; - -pub fn new_candlestick_cache() -> CandlestickCache { - Cache::builder() - .max_capacity(MAX_CANDLE_CAPACITY) - .weigher(|_k, v: &Arc>| v.len().max(1) as u32) - .build() -} diff --git a/src/main.rs b/src/main.rs index d78bd69..8c591d8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -72,7 +72,6 @@ const CHIPNET_START_BLOCK: &str = const KEY_LAST_INDEXED: &str = "last_indexed"; mod bcmr; -mod cache; mod cashaddr; mod chain; mod crc20; @@ -573,7 +572,6 @@ async fn launch() -> _ { .manage(dbpool) .manage(ibd_state) .manage(ohlcv_state) - .manage(cache::new_candlestick_cache()) // give rocket ownership of downloader to ensure thread isn't dropped .manage(bcmrdownloader) .manage(wellknowndownloader) diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs index 2c4a51f..599dbdf 100644 --- a/src/rpc/candlesticks.rs +++ b/src/rpc/candlesticks.rs @@ -3,7 +3,6 @@ // 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 crate::cache::CandlestickCache; use crate::db::blob::display_hex_to_blob; use crate::db::cauldron::ohlcv; use crate::db::DB; @@ -393,7 +392,6 @@ pub async fn price_candlesticks( end: Option, stepsize: Option, conn: &State, - cache: &State, ohlcv: &State>, ) -> CachedApiResult { let current_timestamp = time_now(); @@ -434,52 +432,24 @@ pub async fn price_candlesticks( )); } - // Only cache windows that are fully in the past (identical to the HTTP cache rule). + // Windows fully in the past get immutable HTTP cache headers. let is_historical = end.is_some() && effective_end < current_timestamp - effective_stepsize; - let cache_key = ( - token.to_string(), - effective_start, - effective_end, - effective_stepsize, - ); let ohlcv_end = ohlcv.materialized_end.load(Ordering::Relaxed); - let candlestick_data: Arc> = if is_historical { - if let Some(hit) = cache.get(&cache_key).await { - hit - } else { - let data = candlesticks( - &conn.cauldron_r, - effective_start, - effective_end, - effective_stepsize, - token, - ohlcv_end, - ) - .await - .map_err(|e| bad_request(ApiErrorCode::InvalidParameters, &e.to_string()))?; - let arc = Arc::new(data); - cache.insert(cache_key, arc.clone()).await; - arc - } - } else { - Arc::new( - candlesticks( - &conn.cauldron_r, - effective_start, - effective_end, - effective_stepsize, - token, - ohlcv_end, - ) - .await - .map_err(|e| bad_request(ApiErrorCode::InvalidParameters, &e.to_string()))?, - ) - }; + let candlestick_data = candlesticks( + &conn.cauldron_r, + effective_start, + effective_end, + effective_stepsize, + token, + ohlcv_end, + ) + .await + .map_err(|e| bad_request(ApiErrorCode::InvalidParameters, &e.to_string()))?; let candlesticks_json: Vec = candlestick_data - .iter() + .into_iter() .map(|c| { json!({ "time": c.time, @@ -508,7 +478,6 @@ pub async fn price_candlesticks( #[cfg(test)] mod tests { - use crate::cache::new_candlestick_cache; use crate::db::cauldron::{ ohlcv, pool::{self, dummy_init_seq, insert_new_pool}, @@ -716,7 +685,6 @@ mod tests { let mock_db = mock_db_pool(setup_mock_db).await; let rocket = rocket::build() .manage(mock_db) - .manage(new_candlestick_cache()) .manage(no_ohlcv()) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket) @@ -749,7 +717,6 @@ mod tests { let mock_db = mock_db_pool(setup_mock_db).await; let rocket = rocket::build() .manage(mock_db) - .manage(new_candlestick_cache()) .manage(no_ohlcv()) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket) @@ -782,7 +749,6 @@ mod tests { let rocket = rocket::build() .manage(mock_db) - .manage(new_candlestick_cache()) .manage(no_ohlcv()) .mount("/api", routes![super::price_candlesticks]); @@ -920,7 +886,6 @@ mod tests { // Build Rocket instance with our endpoint let rocket = rocket::build() .manage(mock_db) - .manage(new_candlestick_cache()) .manage(no_ohlcv()) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket) @@ -1119,7 +1084,6 @@ mod tests { // Build Rocket and call endpoint across the two intervals let rocket = rocket::build() .manage(mock_db) - .manage(new_candlestick_cache()) .manage(no_ohlcv()) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket) @@ -1198,7 +1162,6 @@ mod tests { let rocket = rocket::build() .manage(mock_db) - .manage(new_candlestick_cache()) .manage(ohlcv_state) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket).await.expect("valid rocket"); @@ -1261,7 +1224,6 @@ mod tests { let rocket = rocket::build() .manage(mock_db) - .manage(new_candlestick_cache()) .manage(ohlcv_state) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket).await.expect("valid rocket"); @@ -1320,7 +1282,6 @@ mod tests { let rocket = rocket::build() .manage(mock_db) - .manage(new_candlestick_cache()) .manage(ohlcv_state) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket).await.expect("valid rocket"); From 51e73606b14691190bf7b5a9610605c34b973659 Mon Sep 17 00:00:00 2001 From: Jakob Notland Date: Wed, 25 Mar 2026 11:26:17 +0100 Subject: [PATCH 8/9] Review and clean up --- src/db/cauldron/ohlcv.rs | 26 ++------------------------ src/db/cauldron/pool.rs | 10 ++++++++++ src/db/cauldron/utxo_funding.rs | 6 ++++++ src/main.rs | 17 ++++++----------- src/rpc/candlesticks.rs | 10 +++++----- 5 files changed, 29 insertions(+), 40 deletions(-) diff --git a/src/db/cauldron/ohlcv.rs b/src/db/cauldron/ohlcv.rs index f5bb937..b345816 100644 --- a/src/db/cauldron/ohlcv.rs +++ b/src/db/cauldron/ohlcv.rs @@ -25,27 +25,6 @@ pub async fn create_table(pool: &SqlitePool) { .await .expect("failed to create ohlcv_1h table"); - // Needed for fast joins in rebuild_range; use IF NOT EXISTS so this is safe on existing DBs. - sqlx::query("CREATE INDEX IF NOT EXISTS idx_phe_txid ON pool_history_entry(txid)") - .execute(pool) - .await - .expect("failed to create idx_phe_txid index"); - - // Composite index for fast token+time range scans in the candlestick raw CTE. - sqlx::query( - "CREATE INDEX IF NOT EXISTS idx_phe_token_id_ts ON pool_history_entry(token_id, effective_timestamp)", - ) - .execute(pool) - .await - .expect("failed to create idx_phe_token_id_ts index"); - - // Index for tx_latest token-filtered query: JOIN utxo_funding WHERE token_id = ? → ORDER BY tx.effective_timestamp. - sqlx::query( - "CREATE INDEX IF NOT EXISTS idx_utxo_funding_token_txid ON utxo_funding(token_id, txid)", - ) - .execute(pool) - .await - .expect("failed to create idx_utxo_funding_token_txid index"); } /// Returns the highest `bucket_ts` in `ohlcv_1h`, or `None` if the table is empty. @@ -61,10 +40,9 @@ pub async fn get_max_bucket_ts(pool: &SqlitePool) -> Result> { /// so the background task doesn't scan from Unix epoch 0. pub async fn get_min_trade_bucket_ts(pool: &SqlitePool) -> Result> { let row: Option<(Option,)> = sqlx::query_as( - "SELECT (MIN(tx.effective_timestamp) / 3600) * 3600 + "SELECT (MIN(phe.effective_timestamp) / 3600) * 3600 FROM pool_history_entry AS phe - JOIN tx ON tx.txid = phe.txid - WHERE tx.blockhash IS NOT NULL", + WHERE phe.mtp_timestamp IS NOT NULL", ) .fetch_optional(pool) .await?; diff --git a/src/db/cauldron/pool.rs b/src/db/cauldron/pool.rs index 5cfd741..9c7d807 100644 --- a/src/db/cauldron/pool.rs +++ b/src/db/cauldron/pool.rs @@ -83,6 +83,16 @@ pub async fn create_table(pool: &SqlitePool) { .execute(pool) .await .unwrap(); + sqlx::query("CREATE INDEX IF NOT EXISTS idx_phe_txid ON pool_history_entry(txid)") + .execute(pool) + .await + .unwrap(); + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_phe_token_id_ts ON pool_history_entry(token_id, effective_timestamp)", + ) + .execute(pool) + .await + .unwrap(); } async fn get_pool_by_utxo( diff --git a/src/db/cauldron/utxo_funding.rs b/src/db/cauldron/utxo_funding.rs index cbc8452..3e3a43c 100644 --- a/src/db/cauldron/utxo_funding.rs +++ b/src/db/cauldron/utxo_funding.rs @@ -27,6 +27,12 @@ pub async fn create_table(pool: &SqlitePool) { .execute(pool) .await .unwrap(); + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_utxo_funding_token_txid ON utxo_funding(token_id, txid)", + ) + .execute(pool) + .await + .unwrap(); } pub async fn insert_utxo_funding( diff --git a/src/main.rs b/src/main.rs index 8c591d8..6cbc68f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -412,11 +412,10 @@ async fn launch() -> _ { db::cauldron::ohlcv::create_table(&dbpool.cauldron_w).await; // Bootstrap OhlcvState from whatever is already in the table (survives restarts). - let initial_ohlcv_end = db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r) + let max_bucket_ts = db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r) .await - .unwrap_or(None) - .map(|ts| ts + 3600) - .unwrap_or(0); + .unwrap_or(None); + let initial_ohlcv_end = max_bucket_ts.map(|ts| ts + 3600).unwrap_or(0); let ohlcv_state = Arc::new(OhlcvState { materialized_end: AtomicI64::new(initial_ohlcv_end), }); @@ -439,9 +438,9 @@ async fn launch() -> _ { let now = crate::timeutil::time_now(); let cutoff = (now - BACKFILL_SAFETY_SECS) / 3600 * 3600; - let since_opt = match db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r).await { - Ok(Some(ts)) => Some(ts + 3600), - Ok(None) => { + let since_opt = match max_bucket_ts { + Some(ts) => Some(ts + 3600), + None => { match db::cauldron::ohlcv::get_min_trade_bucket_ts(&dbpool.cauldron_r).await { Ok(v) => v, Err(e) => { @@ -450,10 +449,6 @@ async fn launch() -> _ { } } } - Err(e) => { - warn!("ohlcv backfill: could not read max bucket ts: {e}"); - None - } }; if since_opt.is_none() { info!("ohlcv backfill: no confirmed trades found, skipping"); diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs index 599dbdf..11a25ea 100644 --- a/src/rpc/candlesticks.rs +++ b/src/rpc/candlesticks.rs @@ -578,27 +578,27 @@ mod tests { .await .unwrap(); - // Insert tx rows + // Insert tx rows in realistic order: mempool first, then confirmed. let block_zero = BlockHash::all_zeros(); + insert_mempool_tx(&mut *conn, &txid1, TIME_1).await.unwrap(); insert_block_tx(&mut *conn, &txid1, &block_zero, TIME_1 as i64) .await .unwrap(); - insert_mempool_tx(&mut *conn, &txid1, TIME_1).await.unwrap(); + insert_mempool_tx(&mut *conn, &txid2, TIME_2).await.unwrap(); insert_block_tx(&mut *conn, &txid2, &block_zero, TIME_2 as i64) .await .unwrap(); - insert_mempool_tx(&mut *conn, &txid2, TIME_2).await.unwrap(); + insert_mempool_tx(&mut *conn, &txid3, TIME_3).await.unwrap(); insert_block_tx(&mut *conn, &txid3, &block_zero, TIME_3 as i64) .await .unwrap(); - insert_mempool_tx(&mut *conn, &txid3, TIME_3).await.unwrap(); + insert_mempool_tx(&mut *conn, &txid4, TIME_4).await.unwrap(); insert_block_tx(&mut *conn, &txid4, &block_zero, TIME_4 as i64) .await .unwrap(); - insert_mempool_tx(&mut *conn, &txid4, TIME_4).await.unwrap(); // Insert pool_history_entry let pool1 = OutPointHash::from_inner([0x0a; 32]); From 5f0498aa7350d0256337340eb3c876f938e951d6 Mon Sep 17 00:00:00 2001 From: Jakob Notland Date: Wed, 25 Mar 2026 11:35:55 +0100 Subject: [PATCH 9/9] Format --- src/db/cauldron/ohlcv.rs | 1 - src/main.rs | 14 ++++++-------- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/src/db/cauldron/ohlcv.rs b/src/db/cauldron/ohlcv.rs index b345816..fa32178 100644 --- a/src/db/cauldron/ohlcv.rs +++ b/src/db/cauldron/ohlcv.rs @@ -24,7 +24,6 @@ pub async fn create_table(pool: &SqlitePool) { .execute(pool) .await .expect("failed to create ohlcv_1h table"); - } /// Returns the highest `bucket_ts` in `ohlcv_1h`, or `None` if the table is empty. diff --git a/src/main.rs b/src/main.rs index 6cbc68f..a6612fe 100644 --- a/src/main.rs +++ b/src/main.rs @@ -440,15 +440,13 @@ async fn launch() -> _ { let cutoff = (now - BACKFILL_SAFETY_SECS) / 3600 * 3600; let since_opt = match max_bucket_ts { Some(ts) => Some(ts + 3600), - None => { - match db::cauldron::ohlcv::get_min_trade_bucket_ts(&dbpool.cauldron_r).await { - Ok(v) => v, - Err(e) => { - warn!("ohlcv backfill: could not read min trade ts: {e}"); - None - } + None => match db::cauldron::ohlcv::get_min_trade_bucket_ts(&dbpool.cauldron_r).await { + Ok(v) => v, + Err(e) => { + warn!("ohlcv backfill: could not read min trade ts: {e}"); + None } - } + }, }; if since_opt.is_none() { info!("ohlcv backfill: no confirmed trades found, skipping");