feat(hotblocks): count and log query tasks that hold a worker for long

crates/hotblocks/spec/13-interface-binding.md

@@ -33,7 +33,7 @@ annotated with the relevant GAP.
 | METADATA | `GET /datasets/{id}/metadata` | start block, real-time flag, aliases |
 | GET-RETENTION | `GET /datasets/{id}/retention` | current policy JSON |
 | SET-RETENTION | `POST /datasets/{id}/retention` | policy JSON; only for `External` datasets, else `FORBIDDEN` (403) |
-| GET-SETTINGS | `GET /settings` | runtime settings JSON (`{"shadow":{…}}`) |
+| GET-SETTINGS | `GET /settings` | runtime settings JSON (`{"shadow":{…},"slowTasks":{…}}`) |
 | UPDATE-SETTINGS | `PATCH /settings` | JSON Merge Patch (RFC 7386); answers the settings after it. An unknown key, a removed key or an invalid value is a 400 and changes nothing. In memory and per process: a restart returns to the flags |
 | observability | `GET /metrics` (+ engine-diagnostic routes) | OB surface, text formats |
 | readiness | `GET /ready` | rotation gate (OB-8), distinct from the `/` liveness signal: 503 for the whole pre-drain grace window so the orchestrator withdraws the endpoint before anything closes (LIV-12). Process-level only — per-dataset readability (LIV-5c) is still absent (GAP-7) |

crates/hotblocks/src/api.rs

@@ -354,11 +354,11 @@ async fn stream_internal(
 
     let query_result = if finalized {
         app.query_service
-            .query_finalized(&dataset, query, client_id, encoding, shadow)
+            .query_finalized(&dataset, query, body.clone(), client_id, encoding, shadow)
             .await
     } else {
         app.query_service
-            .query(&dataset, query, client_id, encoding, shadow)
+            .query(&dataset, query, body.clone(), client_id, encoding, shadow)
             .await
     };
 

crates/hotblocks/src/cli.rs

@@ -16,7 +16,7 @@ use crate::{
     metrics::{DatasetMetricsCollector, RocksDbCollector},
     query::{QueryService, QueryServiceRef},
     query_engine::MetadataCatalogs,
-    settings::{RuntimeSettings, Settings, SettingsRef, ShadowSettings},
+    settings::{RuntimeSettings, Settings, SettingsRef, ShadowSettings, SlowTaskSettings},
     shadow::{Shadow, ShadowRef},
     types::DBRef
 };
@@ -186,7 +186,21 @@ pub struct CLI {
     /// Shadow log lines for differences, failures and slow runs per 5 minutes,
     /// at most one per query shape; 0 turns them off.
     #[arg(long, value_name = "N", env = "SQD_SHADOW_LOG_LIMIT", default_value = "20")]
-    pub shadow_log_limit: usize
+    pub shadow_log_limit: usize,
+
+    /// A query task is slow, and counted in `hotblocks_query_slow_tasks_total`, when it
+    /// holds a query pool worker this many milliseconds, not counting other
+    /// tasks the worker ran inside it.
+    ///
+    /// This and `--slow-task-log-limit` are starting values: `PATCH /settings`
+    /// changes them until a restart.
+    #[arg(long, value_name = "MS", default_value = "500")]
+    pub slow_task_threshold_ms: u64,
+
+    /// Slow-task log lines per 5 minutes, at most one per query shape; 0 turns
+    /// them off.
+    #[arg(long, value_name = "N", default_value = "20")]
+    pub slow_task_log_limit: usize
 }
 
 pub struct App {
@@ -210,6 +224,10 @@ impl CLI {
                 slow_ratio: self.shadow_slow_ratio,
                 slow_margin_ms: self.shadow_slow_margin_ms,
                 log_limit: self.shadow_log_limit
+            },
+            slow_tasks: SlowTaskSettings {
+                threshold_ms: self.slow_task_threshold_ms,
+                log_limit: self.slow_task_log_limit
             }
         })
         .map(Arc::new)?;
@@ -259,7 +277,7 @@ impl CLI {
             .map(Arc::new)?;
 
         let query_service = {
-            let mut builder = QueryService::builder(db.clone());
+            let mut builder = QueryService::builder(db.clone(), runtime_settings.clone());
             builder.set_max_data_waiters(self.query_max_data_waiters);
 
             if let Some(size) = self.query_task_queue {

crates/hotblocks/src/log_window.rs

@@ -0,0 +1,102 @@
+//! Rate limits for log lines that carry a client's query.
+
+use std::{
+    collections::HashSet,
+    hash::{Hash, Hasher},
+    time::{Duration, Instant}
+};
+
+use sqd_storage::db::DatasetId;
+
+/// How much of a query a log line carries: a query can list tens of thousands
+/// of addresses.
+const MAX_LOGGED_QUERY_BYTES: usize = 16 * 1024;
+/// The window a log limit counts lines in.
+const LOG_WINDOW: Duration = Duration::from_secs(300);
+
+/// The queries logged in detail in the current [`LOG_WINDOW`].
+pub struct LogWindow {
+    started: Instant,
+    /// Query fingerprints and reasons.
+    logged: HashSet<(u64, &'static str)>
+}
+
+impl LogWindow {
+    pub fn new() -> Self {
+        Self {
+            started: Instant::now(),
+            logged: HashSet::new()
+        }
+    }
+
+    /// Whether this query may be logged in detail for this reason now: at most
+    /// `limit` lines per window, one per query shape and reason.
+    pub fn claim(&mut self, fingerprint: u64, reason: &'static str, limit: usize) -> bool {
+        if self.started.elapsed() >= LOG_WINDOW {
+            self.started = Instant::now();
+            self.logged.clear();
+        }
+        self.logged.len() < limit && self.logged.insert((fingerprint, reason))
+    }
+}
+
+/// The query with its block range and parent hash removed: what stays the
+/// same while a client follows the chain.
+pub fn query_fingerprint(dataset_id: DatasetId, body: &[u8]) -> u64 {
+    let mut hasher = std::collections::hash_map::DefaultHasher::new();
+    dataset_id.as_str().hash(&mut hasher);
+    match serde_json::from_slice::<serde_json::Value>(body) {
+        Ok(serde_json::Value::Object(mut query)) => {
+            for key in ["fromBlock", "toBlock", "parentBlockHash"] {
+                query.remove(key);
+            }
+            serde_json::Value::Object(query).to_string().hash(&mut hasher);
+        }
+        _ => body.hash(&mut hasher)
+    }
+    hasher.finish()
+}
+
+pub fn truncate_query(mut query: String) -> String {
+    if query.len() > MAX_LOGGED_QUERY_BYTES {
+        let mut cut = MAX_LOGGED_QUERY_BYTES;
+        while !query.is_char_boundary(cut) {
+            cut -= 1;
+        }
+        query.truncate(cut);
+        query.push_str("…<truncated>");
+    }
+    query
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_window_logs_each_query_shape_once_up_to_the_limit() {
+        let mut window = LogWindow::new();
+
+        assert!(window.claim(1, "slow", 2));
+        assert!(!window.claim(1, "slow", 2), "the same shape was logged twice");
+        assert!(window.claim(1, "mismatch", 2), "another reason is another line");
+        assert!(!window.claim(2, "slow", 2), "the limit was exceeded");
+    }
+
+    #[test]
+    fn the_block_range_does_not_change_the_fingerprint() {
+        let dataset_id = DatasetId::from_str("fingerprint-test");
+        let first = br#"{"type":"evm","fromBlock":1,"toBlock":9,"parentBlockHash":"0x01","logs":[{}]}"#;
+        let next = br#"{"type":"evm","fromBlock":10,"logs":[{}]}"#;
+        let other = br#"{"type":"evm","fromBlock":10,"transactions":[{}]}"#;
+
+        assert_eq!(
+            query_fingerprint(dataset_id, first),
+            query_fingerprint(dataset_id, next)
+        );
+        assert_ne!(
+            query_fingerprint(dataset_id, next),
+            query_fingerprint(dataset_id, other)
+        );
+    }
+}

crates/hotblocks/src/main.rs

@@ -5,6 +5,7 @@ mod dataset_config;
 mod dataset_controller;
 mod encoding;
 mod errors;
+mod log_window;
 mod metrics;
 mod query;
 mod query_engine;

crates/hotblocks/src/metrics.rs

@@ -23,7 +23,7 @@ use tracing::error;
 use crate::{
     errors::{DataAvailabilityChangedDuringFinalizedReplay, UnapplicableFork},
     query::QueryExecutorCollector,
-    types::DBRef
+    types::{ClientId, DBRef}
 };
 
 #[derive(Copy, Clone, Hash, Debug, Default, Ord, PartialOrd, Eq, PartialEq, EncodeLabelSet)]
@@ -303,7 +303,7 @@ pub(crate) enum QueryStep {
 }
 
 impl QueryStep {
-    const fn as_str(self) -> &'static str {
+    pub(crate) const fn as_str(self) -> &'static str {
         match self {
             Self::First => "first",
             Self::Next => "next"
@@ -327,6 +327,7 @@ static QUERY_TASK_QUEUE_DURATION: LazyLock<Family<QueryStepLabel, Histogram>> =
 static QUERY_TASK_RUN_DURATION: LazyLock<Family<QueryStepLabel, Histogram>> =
     LazyLock::new(|| Family::new_with_constructor(|| Histogram::new(exponential_buckets(0.0001, 2.0, 18))));
 static QUERY_TASKS_QUEUED: LazyLock<Gauge> = LazyLock::new(Default::default);
+static QUERY_SLOW_TASKS: LazyLock<Family<Labels, Counter>> = LazyLock::new(Default::default);
 
 pub(crate) fn report_query_task_spawned() {
     QUERY_TASKS_QUEUED.inc();
@@ -345,6 +346,15 @@ pub(crate) fn report_query_task_finished(step: QueryStep, ran: Duration) {
         .observe(ran.as_secs_f64());
 }
 
+pub(crate) fn report_slow_query_task(dataset_id: DatasetId, client_id: &ClientId, step: QueryStep) {
+    let labels = vec![
+        ("client_id", client_id.as_str().to_owned()),
+        ("dataset_name", dataset_id.as_str().to_owned()),
+        ("step", step.as_str().to_owned()),
+    ];
+    QUERY_SLOW_TASKS.get_or_create(&labels).inc();
+}
+
 pub fn report_query_too_many_tasks_error() {
     QUERY_ERROR_TOO_MANY_TASKS.inc();
 }
@@ -750,6 +760,12 @@ pub fn build_metrics_registry() -> Registry {
          of running tasks",
         QUERY_TASKS_QUEUED.clone()
     );
+    registry.register(
+        "query_slow_tasks",
+        "Query tasks that held a query pool worker longer than the slow-task threshold, not \
+         counting other tasks the worker ran inside them, by dataset, client and step",
+        QUERY_SLOW_TASKS.clone()
+    );
     registry.register(
         "shadow_queries",
         "Queries sampled for the new query engine's shadow comparison",

crates/hotblocks/src/query/executor.rs

@@ -1,10 +1,11 @@
 use std::{
+    cell::Cell,
     panic::{AssertUnwindSafe, catch_unwind},
     sync::{
         Arc,
         atomic::{AtomicUsize, Ordering}
     },
-    time::Instant
+    time::{Duration, Instant}
 };
 
 use crate::{
@@ -15,6 +16,26 @@ use crate::{
     }
 };
 
+thread_local! {
+    /// Time of the tasks nested in the one this worker runs; `None` between tasks.
+    static NESTED: Cell<Option<Duration>> = const { Cell::new(None) };
+}
+
+/// How long a query task held its worker.
+#[derive(Copy, Clone, Debug)]
+pub struct TaskTime {
+    pub total: Duration,
+    /// Other query tasks this worker ran inside the task while it waited for
+    /// the task's parallel work. Tasks nested on other workers are not seen.
+    pub nested: Duration
+}
+
+impl TaskTime {
+    pub fn own(&self) -> Duration {
+        self.total.saturating_sub(self.nested)
+    }
+}
+
 #[derive(Clone)]
 pub struct QueryExecutor {
     // number of concurrent queries
@@ -74,9 +95,12 @@ impl QuerySlot {
         time.min(100)
     }
 
-    pub async fn run<R, F>(self, step: QueryStep, task: F) -> Result<R, QueryTaskPanicked>
+    /// Run `task` on the query pool, then `report` its result and time on the
+    /// same worker. A task that panics is not reported.
+    pub async fn run<R, F, P>(self, step: QueryStep, task: F, report: P) -> Result<R, QueryTaskPanicked>
     where
         F: FnOnce(&Self) -> R + Send + 'static,
+        P: FnOnce(&R, TaskTime) + Send + 'static,
         R: Send + 'static
     {
         let (tx, rx) = tokio::sync::oneshot::channel();
@@ -86,14 +110,22 @@ impl QuerySlot {
         sqd_polars::POOL.spawn(move || {
             let started = Instant::now();
             report_query_task_started(step, started.duration_since(spawned));
+            let outer = NESTED.replace(Some(Duration::ZERO));
 
             let slot = self;
             let result = catch_unwind(AssertUnwindSafe(|| task(&slot))).map_err(|_| QueryTaskPanicked);
-            report_query_task_finished(step, started.elapsed());
+
+            let total = started.elapsed();
+            let nested = NESTED.replace(outer.map(|nested| nested + total)).unwrap_or_default();
+            report_query_task_finished(step, total);
 
             // Release before waking the caller: sending first lets it observe the slot still
             // taken and be refused admission for a query that has already finished.
             drop(slot);
+            // Here and not in the caller, which is gone when the client disconnects.
+            if let Ok(value) = &result {
+                report(value, TaskTime { total, nested });
+            }
             let _ = tx.send(result);
         });
 
@@ -126,10 +158,41 @@ mod tests {
         let result = executor
             .get_slot()
             .expect("the first task must get a slot")
-            .run(QueryStep::First, |_| panic!("injected query panic"))
+            .run(QueryStep::First, |_| panic!("injected query panic"), |_: &(), _| {})
             .await;
 
         assert!(result.is_err());
         assert!(executor.get_slot().is_some(), "the panicked task must release its slot");
     }
+
+    #[tokio::test]
+    async fn a_task_is_reported_after_its_caller_is_gone() {
+        let executor = QueryExecutor::new(1, 1);
+        let (start, started) = tokio::sync::oneshot::channel();
+        let (finish, finished) = std::sync::mpsc::channel::<()>();
+        let (report, reported) = std::sync::mpsc::channel();
+
+        let run = executor.get_slot().expect("the first task must get a slot").run(
+            QueryStep::Next,
+            move |_| {
+                start.send(()).unwrap();
+                finished.recv().unwrap()
+            },
+            move |_, time| report.send(time).unwrap()
+        );
+        let caller = tokio::spawn(run);
+        started.await.unwrap();
+        caller.abort();
+        assert!(
+            matches!(caller.await, Err(e) if e.is_cancelled()),
+            "the caller was not dropped while the task ran"
+        );
+
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        finish.send(()).unwrap();
+        let time = reported
+            .recv_timeout(Duration::from_secs(10))
+            .expect("the task was not reported");
+        assert!(time.total >= Duration::from_millis(50), "{time:?}");
+    }
 }

crates/hotblocks/src/query/mod.rs

@@ -2,6 +2,7 @@ mod executor;
 mod response;
 mod running;
 mod service;
+mod slow_tasks;
 mod static_snapshot;
 
 pub use executor::QueryExecutorCollector;