style: apply cargo fmt

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: zzylol

asap-query-engine/src/bin/test_e2e_precompute.rs

@@ -353,7 +353,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         }
     }
     let batch_body = build_remote_write_body(batch_timeseries);
-    println!("  Payload size: {} bytes (snappy-compressed)", batch_body.len());
+    println!(
+        "  Payload size: {} bytes (snappy-compressed)",
+        batch_body.len()
+    );
 
     let t0 = std::time::Instant::now();
     let resp = client
@@ -374,12 +377,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
 
     // Verify samples landed in the store
-    let batch_results = store.query_precomputed_output(
-        "fake_metric",
-        raw_agg_id,
-        200_000,
-        210_000,
-    )?;
+    let batch_results =
+        store.query_precomputed_output("fake_metric", raw_agg_id, 200_000, 210_000)?;
     let batch_buckets: usize = batch_results.values().map(|v| v.len()).sum();
     println!("  Stored {batch_buckets} buckets from batch (expected 1000)");
     assert!(
@@ -438,12 +437,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let drain_deadline = std::time::Instant::now() + std::time::Duration::from_secs(60);
     let mut tp_buckets: usize;
     loop {
-        let tp_results = store.query_precomputed_output(
-            "fake_metric",
-            raw_agg_id,
-            300_000,
-            max_ts,
-        )?;
+        let tp_results =
+            store.query_precomputed_output("fake_metric", raw_agg_id, 300_000, max_ts)?;
         tp_buckets = tp_results.values().map(|v| v.len()).sum();
         if tp_buckets as u64 >= total_samples || std::time::Instant::now() > drain_deadline {
             break;

asap-query-engine/src/precompute_engine/mod.rs

@@ -15,9 +15,9 @@ use crate::precompute_engine::worker::Worker;
 use axum::{body::Bytes, extract::State, http::StatusCode, routing::post, Router};
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Instant;
 use tokio::net::TcpListener;
 use tokio::sync::mpsc;
-use std::time::Instant;
 use tracing::{debug_span, info, warn, Instrument};
 
 /// Shared state for the ingest HTTP handler.
@@ -67,10 +67,8 @@ impl PrecomputeEngine {
         let router = SeriesRouter::new(senders);
 
         // Build aggregation config map from streaming config
-        let agg_configs: HashMap<u64, _> = self
-            .streaming_config
-            .get_all_aggregation_configs()
-            .clone();
+        let agg_configs: HashMap<u64, _> =
+            self.streaming_config.get_all_aggregation_configs().clone();
 
         // Spawn workers
         let mut worker_handles = Vec::with_capacity(num_workers);
@@ -138,10 +136,7 @@ impl PrecomputeEngine {
 }
 
 /// Axum handler for Prometheus remote write.
-async fn handle_ingest(
-    State(state): State<Arc<IngestState>>,
-    body: Bytes,
-) -> StatusCode {
+async fn handle_ingest(State(state): State<Arc<IngestState>>, body: Bytes) -> StatusCode {
     let ingest_span = debug_span!("ingest", body_len = body.len());
     let ingest_received_at = Instant::now();
 
@@ -179,7 +174,11 @@ async fn handle_ingest(
             .collect();
 
         // Route all series to workers concurrently
-        if let Err(e) = state.router.route_batch(by_series_owned, ingest_received_at).await {
+        if let Err(e) = state
+            .router
+            .route_batch(by_series_owned, ingest_received_at)
+            .await
+        {
             warn!("Batch routing error: {}", e);
             return StatusCode::INTERNAL_SERVER_ERROR;
         }

asap-query-engine/src/precompute_engine/series_router.rs

@@ -1,6 +1,6 @@
+use futures::future::try_join_all;
 use std::collections::HashMap;
 use std::time::Instant;
-use futures::future::try_join_all;
 use tokio::sync::mpsc;
 use xxhash_rust::xxh64::xxh64;
 
@@ -82,15 +82,18 @@ impl SeriesRouter {
             let sender = &self.senders[worker_idx];
             async move {
                 for msg in messages {
-                    sender.send(msg).await.map_err(|e| {
-                        format!("Failed to send to worker {}: {}", worker_idx, e)
-                    })?;
+                    sender
+                        .send(msg)
+                        .await
+                        .map_err(|e| format!("Failed to send to worker {}: {}", worker_idx, e))?;
                 }
                 Ok::<(), String>(())
             }
         }))
         .await
-        .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { Box::new(std::io::Error::new(std::io::ErrorKind::Other, e)) })?;
+        .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> {
+            Box::new(std::io::Error::new(std::io::ErrorKind::Other, e))
+        })?;
 
         Ok(())
     }