Merge branch 'cacheCandles' into 'master'
Cachecandles See merge request riftenlabs/riftenlabs-indexer!78
This commit is contained in:
jakobsn
commit
59747221a6
6 changed files with 1063 additions and 115 deletions
|
|
@ -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;
|
||||
|
|
|
|||
449
src/db/cauldron/ohlcv.rs
Normal file
449
src/db/cauldron/ohlcv.rs
Normal file
|
|
@ -0,0 +1,449 @@
|
|||
// 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");
|
||||
}
|
||||
|
||||
/// 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<Option<i64>> {
|
||||
let row: Option<(Option<i64>,)> = 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<Option<i64>> {
|
||||
let row: Option<(Option<i64>,)> = sqlx::query_as(
|
||||
"SELECT (MIN(phe.effective_timestamp) / 3600) * 3600
|
||||
FROM pool_history_entry AS phe
|
||||
WHERE phe.mtp_timestamp 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<u64> {
|
||||
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
|
||||
phe.token_id,
|
||||
phe.txid,
|
||||
phe.effective_timestamp AS ts,
|
||||
phe.utxo,
|
||||
phe.sats_delta,
|
||||
phe.token_delta,
|
||||
phe.sequence
|
||||
FROM pool_history_entry AS phe
|
||||
JOIN tx ON tx.txid = phe.txid
|
||||
WHERE tx.blockhash IS NOT NULL
|
||||
AND phe.effective_timestamp >= ?
|
||||
AND phe.effective_timestamp < ?
|
||||
),
|
||||
per_pool_tx AS (
|
||||
SELECT
|
||||
token_id,
|
||||
txid,
|
||||
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,
|
||||
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,
|
||||
MIN(min_sequence) AS min_sequence,
|
||||
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, 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
|
||||
),
|
||||
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<u8> = 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<Vec<OhlcvRow>> {
|
||||
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())
|
||||
}
|
||||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
185
src/main.rs
185
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 {
|
||||
|
|
@ -111,6 +119,7 @@ async fn start_program(
|
|||
WellKnownDownloader,
|
||||
CRC20Fetcher,
|
||||
Arc<IbdState>,
|
||||
Arc<AtomicBool>, // indexing_in_progress
|
||||
)> {
|
||||
let network = match config.network.to_lowercase().as_str() {
|
||||
"mainnet" => Network::Bitcoin,
|
||||
|
|
@ -264,13 +273,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 +284,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 +353,7 @@ async fn start_program(
|
|||
wellknowndownloader,
|
||||
crc20fetcher,
|
||||
ibd_state,
|
||||
indexing_in_progress,
|
||||
))
|
||||
}
|
||||
|
||||
|
|
@ -362,8 +373,14 @@ async fn launch() -> _ {
|
|||
config
|
||||
};
|
||||
|
||||
let (dbpool, bcmrdownloader, wellknowndownloader, crc20fetcher, ibd_state) =
|
||||
match start_program(config).await {
|
||||
let (
|
||||
dbpool,
|
||||
bcmrdownloader,
|
||||
wellknowndownloader,
|
||||
crc20fetcher,
|
||||
ibd_state,
|
||||
indexing_in_progress,
|
||||
) = match start_program(config).await {
|
||||
Ok(db) => db,
|
||||
Err(e) => {
|
||||
let backtrace = Backtrace::capture();
|
||||
|
|
@ -391,11 +408,163 @@ 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 max_bucket_ts = db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r)
|
||||
.await
|
||||
.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),
|
||||
});
|
||||
|
||||
// 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();
|
||||
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
|
||||
}
|
||||
},
|
||||
};
|
||||
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();
|
||||
// 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)
|
||||
// give rocket ownership of downloader to ensure thread isn't dropped
|
||||
.manage(bcmrdownloader)
|
||||
.manage(wellknowndownloader)
|
||||
|
|
|
|||
|
|
@ -4,10 +4,12 @@
|
|||
// 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::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 +17,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 +81,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<PriceInterval>,
|
||||
step_size: i64,
|
||||
token_id: &str,
|
||||
) -> Result<Vec<CandlestickData>> {
|
||||
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::<TokenID>(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<f64> = None;
|
||||
mut found_first_trade: bool,
|
||||
mut last_close_price: Option<f64>,
|
||||
) -> (Vec<CandlestickData>, bool, Option<f64>) {
|
||||
let mut result = Vec::with_capacity(intervals.len());
|
||||
let mut trade_index = 0;
|
||||
|
||||
|
|
@ -183,9 +111,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 +125,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 +171,191 @@ 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<ohlcv::OhlcvRow>,
|
||||
start: i64,
|
||||
end: i64,
|
||||
mut found_first_trade: bool,
|
||||
mut last_close: Option<f64>,
|
||||
) -> (Vec<CandlestickData>, bool, Option<f64>) {
|
||||
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<Vec<(i64, i64, i64, i64, i64)>> {
|
||||
let sql = r#"
|
||||
WITH per_pool_tx_raw AS (
|
||||
SELECT
|
||||
phe.txid,
|
||||
phe.effective_timestamp,
|
||||
phe.utxo,
|
||||
phe.sats_delta,
|
||||
phe.token_delta,
|
||||
phe.sequence
|
||||
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,
|
||||
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
|
||||
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,
|
||||
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, min_sequence 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<Vec<CandlestickData>> {
|
||||
if timestamp_start > timestamp_end {
|
||||
bail!("Start cannot be higher than end");
|
||||
}
|
||||
|
||||
let token_blob = display_hex_to_blob::<TokenID>(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);
|
||||
|
||||
// timestamp_start is guaranteed hour-aligned by the entry condition above.
|
||||
let ohlcv_rows =
|
||||
ohlcv::get_active_candles(pool, &token_blob, timestamp_start, ohlcv_end).await?;
|
||||
|
||||
let (mut result, found_first, last_close) =
|
||||
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.
|
||||
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 +392,7 @@ pub async fn price_candlesticks(
|
|||
end: Option<i64>,
|
||||
stepsize: Option<i64>,
|
||||
conn: &State<DB>,
|
||||
ohlcv: &State<Arc<OhlcvState>>,
|
||||
) -> CachedApiResult<Value> {
|
||||
let current_timestamp = time_now();
|
||||
|
||||
|
|
@ -324,18 +432,24 @@ pub async fn price_candlesticks(
|
|||
));
|
||||
}
|
||||
|
||||
// Windows fully in the past get immutable HTTP cache headers.
|
||||
let is_historical = end.is_some() && effective_end < current_timestamp - effective_stepsize;
|
||||
|
||||
let ohlcv_end = ohlcv.materialized_end.load(Ordering::Relaxed);
|
||||
|
||||
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<Value> = candlestick_data
|
||||
.iter()
|
||||
.into_iter()
|
||||
.map(|c| {
|
||||
json!({
|
||||
"time": c.time,
|
||||
|
|
@ -350,8 +464,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
|
||||
|
|
@ -366,11 +479,13 @@ pub async fn price_candlesticks(
|
|||
#[cfg(test)]
|
||||
mod tests {
|
||||
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 +494,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<OhlcvState> {
|
||||
Arc::new(OhlcvState {
|
||||
materialized_end: AtomicI64::new(0),
|
||||
})
|
||||
}
|
||||
|
||||
/// Four trades, each 600-second bin is 10 minutes.
|
||||
/// We'll have two bins:
|
||||
|
|
@ -455,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]);
|
||||
|
|
@ -562,6 +685,7 @@ mod tests {
|
|||
let mock_db = mock_db_pool(setup_mock_db).await;
|
||||
let rocket = rocket::build()
|
||||
.manage(mock_db)
|
||||
.manage(no_ohlcv())
|
||||
.mount("/api", routes![super::price_candlesticks]);
|
||||
let client = Client::tracked(rocket)
|
||||
.await
|
||||
|
|
@ -593,6 +717,7 @@ mod tests {
|
|||
let mock_db = mock_db_pool(setup_mock_db).await;
|
||||
let rocket = rocket::build()
|
||||
.manage(mock_db)
|
||||
.manage(no_ohlcv())
|
||||
.mount("/api", routes![super::price_candlesticks]);
|
||||
let client = Client::tracked(rocket)
|
||||
.await
|
||||
|
|
@ -624,6 +749,7 @@ mod tests {
|
|||
|
||||
let rocket = rocket::build()
|
||||
.manage(mock_db)
|
||||
.manage(no_ohlcv())
|
||||
.mount("/api", routes![super::price_candlesticks]);
|
||||
|
||||
let client = Client::tracked(rocket)
|
||||
|
|
@ -760,6 +886,7 @@ mod tests {
|
|||
// Build Rocket instance with our endpoint
|
||||
let rocket = rocket::build()
|
||||
.manage(mock_db)
|
||||
.manage(no_ohlcv())
|
||||
.mount("/api", routes![super::price_candlesticks]);
|
||||
let client = Client::tracked(rocket)
|
||||
.await
|
||||
|
|
@ -957,6 +1084,7 @@ mod tests {
|
|||
// Build Rocket and call endpoint across the two intervals
|
||||
let rocket = rocket::build()
|
||||
.manage(mock_db)
|
||||
.manage(no_ohlcv())
|
||||
.mount("/api", routes![super::price_candlesticks]);
|
||||
let client = Client::tracked(rocket)
|
||||
.await
|
||||
|
|
@ -1004,4 +1132,189 @@ 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(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(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(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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue