FMT and clippy

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(
+    sqlx::query("CREATE INDEX IF NOT EXISTS idx_phe_txid ON pool_history_entry(txid)")
-        "CREATE INDEX IF NOT EXISTS idx_phe_txid ON pool_history_entry(txid)",
+        .execute(pool)
-    )
+        .await
-    .execute(pool)
+        .expect("failed to create idx_phe_txid index");
-    .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)
+        insert_confirmed_trade(
-            .await;
+            &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)
+        insert_confirmed_trade(
-            .await;
+            &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`.

src/main.rs

@@ -374,16 +374,22 @@ async fn launch() -> _ {
         config
     };
 
-    let (dbpool, bcmrdownloader, wellknowndownloader, crc20fetcher, ibd_state, indexing_in_progress) =
+    let (
-        match start_program(config).await {
+        dbpool,
-            Ok(db) => db,
+        bcmrdownloader,
-            Err(e) => {
+        wellknowndownloader,
-                let backtrace = Backtrace::capture();
+        crc20fetcher,
-                error!("Backtrace (if RUST_BACKTRACE=1):\n{backtrace}");
+        ibd_state,
-                error!("Error: {e}");
+        indexing_in_progress,
-                panic!("Failed at program startup")
+    ) = 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();
-                let now = crate::timeutil::time_now() as i64;
                 // 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)
+                    match db::cauldron::ohlcv::rebuild_range(
-                        .await
+                        &ohlcv_read,
+                        &ohlcv_write,
+                        batch_start,
+                        batch_end,
+                    )
+                    .await
                     {
                         Ok(n) => {
                             info!("ohlcv: materialised {n} buckets [{batch_start}, {batch_end})");

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
+    if step_size == 3600 && ohlcv_materialized_end > timestamp_start && timestamp_start % 3600 == 0
-        && 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 =
+            let raw_trades = fetch_raw_trades(pool, &token_blob, ohlcv_end, timestamp_end).await?;
-                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, _, _) =
+            let (tail, _, _) = aggregate_raw_trades(
-                aggregate_raw_trades(&raw_trades, tail_intervals, step_size, found_first, last_close);
+                &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 =
+    let all_trades = fetch_raw_trades(pool, &token_blob, timestamp_start, timestamp_end).await?;
-        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)