Review and clean up

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<Option<i64>> {
 /// 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(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 phe.mtp_timestamp IS NOT NULL",
-         WHERE tx.blockhash IS NOT NULL",
     )
     .fetch_optional(pool)
     .await?;

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(

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(

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)
+        .unwrap_or(None);
-        .map(|ts| ts + 3600)
+    let initial_ohlcv_end = max_bucket_ts.map(|ts| ts + 3600).unwrap_or(0);
-        .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 {
+        let since_opt = match max_bucket_ts {
-            Ok(Some(ts)) => Some(ts + 3600),
+            Some(ts) => Some(ts + 3600),
-            Ok(None) => {
+            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");

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