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]);