From 86588414be2b2181f3762af17282c64d8033c17f Mon Sep 17 00:00:00 2001 From: Evgeny Formanenko Date: Thu, 15 Oct 2026 17:36:10 +0300 Subject: [PATCH] feat(hotblocks): count and log query tasks that hold a worker for long The queue histograms show that tasks wait behind other tasks, not which queries they wait behind. A slow task is now counted by dataset, client and step, and logged with its query at most once per query shape and up to a limit per five minutes. A worker that waits for its task's parallel work can run another queued task meanwhile, so that nested time is not held against the outer task. The worker reports the task, not the caller, so a task that fails or whose client disconnects is reported too. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/hotblocks/spec/13-interface-binding.md | 2 +- crates/hotblocks/src/api.rs | 4 +- crates/hotblocks/src/cli.rs | 24 ++- crates/hotblocks/src/log_window.rs | 102 ++++++++++ crates/hotblocks/src/main.rs | 1 + crates/hotblocks/src/metrics.rs | 20 +- crates/hotblocks/src/query/executor.rs | 71 ++++++- crates/hotblocks/src/query/mod.rs | 1 + crates/hotblocks/src/query/response.rs | 53 +++-- crates/hotblocks/src/query/running.rs | 1 + crates/hotblocks/src/query/service.rs | 27 ++- crates/hotblocks/src/query/slow_tasks.rs | 184 ++++++++++++++++++ crates/hotblocks/src/settings.rs | 29 ++- crates/hotblocks/src/shadow/mod.rs | 63 +----- crates/hotblocks/tests/query_task_metrics.rs | 60 ++++++ 15 files changed, 550 insertions(+), 92 deletions(-) create mode 100644 crates/hotblocks/src/log_window.rs create mode 100644 crates/hotblocks/src/query/slow_tasks.rs diff --git a/crates/hotblocks/spec/13-interface-binding.md b/crates/hotblocks/spec/13-interface-binding.md index 3df80257..240a1535 100644 --- a/crates/hotblocks/spec/13-interface-binding.md +++ b/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) | diff --git a/crates/hotblocks/src/api.rs b/crates/hotblocks/src/api.rs index 75d55515..b869a18a 100644 --- a/crates/hotblocks/src/api.rs +++ b/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 }; diff --git a/crates/hotblocks/src/cli.rs b/crates/hotblocks/src/cli.rs index 7607a9e6..5ff7034c 100644 --- a/crates/hotblocks/src/cli.rs +++ b/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 { diff --git a/crates/hotblocks/src/log_window.rs b/crates/hotblocks/src/log_window.rs new file mode 100644 index 00000000..7f9dc12c --- /dev/null +++ b/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::(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("…"); + } + 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) + ); + } +} diff --git a/crates/hotblocks/src/main.rs b/crates/hotblocks/src/main.rs index 311aea7d..5004e3d9 100644 --- a/crates/hotblocks/src/main.rs +++ b/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; diff --git a/crates/hotblocks/src/metrics.rs b/crates/hotblocks/src/metrics.rs index 1f10a045..9ee73583 100644 --- a/crates/hotblocks/src/metrics.rs +++ b/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> = static QUERY_TASK_RUN_DURATION: LazyLock> = LazyLock::new(|| Family::new_with_constructor(|| Histogram::new(exponential_buckets(0.0001, 2.0, 18)))); static QUERY_TASKS_QUEUED: LazyLock = LazyLock::new(Default::default); +static QUERY_SLOW_TASKS: LazyLock> = 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", diff --git a/crates/hotblocks/src/query/executor.rs b/crates/hotblocks/src/query/executor.rs index 57745f43..61e72de8 100644 --- a/crates/hotblocks/src/query/executor.rs +++ b/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> = 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(self, step: QueryStep, task: F) -> Result + /// 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(self, step: QueryStep, task: F, report: P) -> Result 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:?}"); + } } diff --git a/crates/hotblocks/src/query/mod.rs b/crates/hotblocks/src/query/mod.rs index b4718ada..e1b4810c 100644 --- a/crates/hotblocks/src/query/mod.rs +++ b/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; diff --git a/crates/hotblocks/src/query/response.rs b/crates/hotblocks/src/query/response.rs index 78378ffe..b9b6d436 100644 --- a/crates/hotblocks/src/query/response.rs +++ b/crates/hotblocks/src/query/response.rs @@ -1,4 +1,7 @@ -use std::time::{Duration, Instant}; +use std::{ + sync::Arc, + time::{Duration, Instant} +}; use anyhow::bail; use bytes::Bytes; @@ -8,7 +11,8 @@ use sqd_storage::db::DatasetId; use super::{ executor::{QueryExecutor, QuerySlot}, - running::{RunningQuery, RunningQueryStats} + running::{RunningQuery, RunningQueryStats}, + slow_tasks::{SlowTasksRef, TaskReporter} }; use crate::{ encoding::ContentEncoding, @@ -25,6 +29,7 @@ const DEFAULT_QUERY_LIMIT: Duration = Duration::from_secs(10); pub struct QueryResponse { executor: QueryExecutor, + reporter: Arc, runner: Option>, finalized_head: Option, dataset_id: DatasetId, @@ -83,9 +88,11 @@ impl QueryStreamStats { impl QueryResponse { pub(super) async fn new( executor: QueryExecutor, + slow_tasks: SlowTasksRef, db: DBRef, dataset_id: DatasetId, query: Query, + body: Bytes, only_finalized: bool, time_limit: Option, client_id: ClientId, @@ -95,18 +102,29 @@ impl QueryResponse { let Some(slot) = executor.get_slot() else { bail!(Busy) }; let stats = QueryStreamStats::new(); + let reporter = Arc::new(TaskReporter::new(slow_tasks, dataset_id, client_id.clone(), body)); + let first_reporter = reporter.clone(); let mut runner = slot - .run(QueryStep::First, move |slot| -> anyhow::Result<_> { - let mut runner = - RunningQuery::new(db, dataset_id, &query, only_finalized, encoding, shadow).map(Box::new)?; - next_run(&mut runner, slot)?; - Ok(runner) - }) + .run( + QueryStep::First, + move |slot| -> anyhow::Result<_> { + let mut runner = + RunningQuery::new(db, dataset_id, &query, only_finalized, encoding, shadow).map(Box::new)?; + next_run(&mut runner, slot)?; + Ok(runner) + }, + move |runner, time| { + let failed = runner.is_err(); + let runner = runner.as_deref().ok(); + first_reporter.record(QueryStep::First, time, runner, &RunningQueryStats::new(), failed); + } + ) .await??; let time_limit = time_limit.unwrap_or(DEFAULT_QUERY_LIMIT); let response = Self { executor, + reporter, finalized_head: runner.take_finalized_head(), runner: Some(runner), stats, @@ -148,12 +166,21 @@ impl QueryResponse { bail!(Busy); }; + let stats_before = *runner.stats(); + let reporter = self.reporter.clone(); let (mut runner, result) = slot - .run(QueryStep::Next, move |slot| { - let mut runner = runner; - let result = next_run(&mut runner, slot); - (runner, result) - }) + .run( + QueryStep::Next, + move |slot| { + let mut runner = runner; + let result = next_run(&mut runner, slot); + (runner, result) + }, + move |(runner, result), time| { + let failed = result.is_err(); + reporter.record(QueryStep::Next, time, Some(runner), &stats_before, failed); + } + ) .await?; if let Err(err) = result { diff --git a/crates/hotblocks/src/query/running.rs b/crates/hotblocks/src/query/running.rs index eafbe59f..1eed561d 100644 --- a/crates/hotblocks/src/query/running.rs +++ b/crates/hotblocks/src/query/running.rs @@ -22,6 +22,7 @@ struct LeftOver { next_block: BlockNumber } +#[derive(Clone, Copy)] pub struct RunningQueryStats { pub chunks_read: u64, pub blocks_read: u64, diff --git a/crates/hotblocks/src/query/service.rs b/crates/hotblocks/src/query/service.rs index 066a9eb9..2907bfd2 100644 --- a/crates/hotblocks/src/query/service.rs +++ b/crates/hotblocks/src/query/service.rs @@ -7,14 +7,20 @@ use std::{ }; use anyhow::{bail, ensure}; +use bytes::Bytes; use sqd_query::Query; -use super::{executor::QueryExecutor, response::QueryResponse}; +use super::{ + executor::QueryExecutor, + response::QueryResponse, + slow_tasks::{SlowTasks, SlowTasksRef} +}; use crate::{ dataset_controller::DatasetController, encoding::ContentEncoding, errors::{Busy, QueryIsAboveTheHead, QueryKindMismatch}, query::QueryExecutorCollector, + settings::SettingsRef, shadow::ShadowQuery, types::{ClientId, DBRef, DatasetKind} }; @@ -23,15 +29,17 @@ pub type QueryServiceRef = Arc; pub struct QueryServiceBuilder { db: DBRef, + settings: SettingsRef, max_data_waiters: usize, max_pending_tasks: usize, urgency: usize } impl QueryServiceBuilder { - pub fn new(db: DBRef) -> Self { + pub fn new(db: DBRef, settings: SettingsRef) -> Self { Self { db, + settings, max_data_waiters: 64_000, max_pending_tasks: sqd_polars::POOL.current_num_threads() * 200, urgency: 500 @@ -63,6 +71,7 @@ impl QueryServiceBuilder { QueryService { db: self.db.clone(), executor: QueryExecutor::new(self.max_pending_tasks, self.urgency), + slow_tasks: Arc::new(SlowTasks::new(self.settings.clone())), wait_slots: WaitSlots { waiters: AtomicUsize::new(0), limit: self.max_data_waiters @@ -74,23 +83,25 @@ impl QueryServiceBuilder { pub struct QueryService { db: DBRef, executor: QueryExecutor, + slow_tasks: SlowTasksRef, wait_slots: WaitSlots } impl QueryService { - pub fn builder(db: DBRef) -> QueryServiceBuilder { - QueryServiceBuilder::new(db) + pub fn builder(db: DBRef, settings: SettingsRef) -> QueryServiceBuilder { + QueryServiceBuilder::new(db, settings) } pub async fn query( &self, dataset: &DatasetController, query: Query, + body: Bytes, client_id: ClientId, encoding: ContentEncoding, shadow: Option ) -> anyhow::Result { - self.query_internal(dataset, query, false, client_id, encoding, shadow) + self.query_internal(dataset, query, body, false, client_id, encoding, shadow) .await } @@ -98,11 +109,12 @@ impl QueryService { &self, dataset: &DatasetController, query: Query, + body: Bytes, client_id: ClientId, encoding: ContentEncoding, shadow: Option ) -> anyhow::Result { - self.query_internal(dataset, query, true, client_id, encoding, shadow) + self.query_internal(dataset, query, body, true, client_id, encoding, shadow) .await } @@ -110,6 +122,7 @@ impl QueryService { &self, dataset: &DatasetController, query: Query, + body: Bytes, finalized: bool, client_id: ClientId, encoding: ContentEncoding, @@ -164,9 +177,11 @@ impl QueryService { QueryResponse::new( self.executor.clone(), + self.slow_tasks.clone(), self.db.clone(), dataset.dataset_id(), query, + body, finalized, None, client_id, diff --git a/crates/hotblocks/src/query/slow_tasks.rs b/crates/hotblocks/src/query/slow_tasks.rs new file mode 100644 index 00000000..9b161d09 --- /dev/null +++ b/crates/hotblocks/src/query/slow_tasks.rs @@ -0,0 +1,184 @@ +//! Query tasks that hold a query pool worker for long: every task queued +//! behind one waits for it. + +use std::sync::Arc; + +use bytes::Bytes; +use parking_lot::Mutex; +use sqd_storage::db::DatasetId; +use tracing::warn; + +use super::{ + executor::TaskTime, + running::{RunningQuery, RunningQueryStats} +}; +use crate::{ + log_window::{LogWindow, query_fingerprint, truncate_query}, + metrics::{QueryStep, report_slow_query_task}, + settings::SettingsRef, + types::ClientId +}; + +pub type SlowTasksRef = Arc; + +pub struct SlowTasks { + settings: SettingsRef, + log_window: Mutex +} + +/// A finished task of a query stream. The read counts are `None` for a task +/// that failed before it had a running query. +pub struct TaskReport<'a> { + pub dataset_id: DatasetId, + pub client_id: &'a ClientId, + /// The request as the client sent it. + pub body: &'a [u8], + pub step: QueryStep, + pub time: TaskTime, + pub failed: bool, + pub chunks_read: Option, + pub blocks_read: Option, + pub buffered_bytes: Option +} + +impl SlowTasks { + pub fn new(settings: SettingsRef) -> Self { + Self { + settings, + log_window: Mutex::new(LogWindow::new()) + } + } + + /// Count a slow task, and log it while its query shape and the log limit allow. + pub fn record(&self, task: &TaskReport) { + let settings = self.settings.slow_tasks(); + if task.time.own() < settings.threshold() { + return; + } + report_slow_query_task(task.dataset_id, task.client_id, task.step); + + let fingerprint = query_fingerprint(task.dataset_id, task.body); + if !self + .log_window + .lock() + .claim(fingerprint, "slow_task", settings.log_limit) + { + return; + } + + warn!( + dataset = %task.dataset_id, + client_id = task.client_id.as_str(), + step = task.step.as_str(), + own_ms = task.time.own().as_secs_f64() * 1000.0, + nested_ms = task.time.nested.as_secs_f64() * 1000.0, + failed = task.failed, + chunks_read = task.chunks_read, + blocks_read = task.blocks_read, + buffered_bytes = task.buffered_bytes, + query = truncate_query(String::from_utf8_lossy(task.body).into_owned()), + query_bytes = task.body.len(), + "query task held a query pool worker for long" + ); + } +} + +/// Reports the tasks of one query stream to [`SlowTasks`]. +pub struct TaskReporter { + slow_tasks: SlowTasksRef, + dataset_id: DatasetId, + client_id: ClientId, + body: Bytes +} + +impl TaskReporter { + pub fn new(slow_tasks: SlowTasksRef, dataset_id: DatasetId, client_id: ClientId, body: Bytes) -> Self { + Self { + slow_tasks, + dataset_id, + client_id, + body + } + } + + /// `stats_before` are the stream's read counts when the task started. + pub fn record( + &self, + step: QueryStep, + time: TaskTime, + runner: Option<&RunningQuery>, + stats_before: &RunningQueryStats, + failed: bool + ) { + let stats = runner.map(RunningQuery::stats); + self.slow_tasks.record(&TaskReport { + dataset_id: self.dataset_id, + client_id: &self.client_id, + body: &self.body, + step, + time, + failed, + chunks_read: stats.map(|stats| stats.chunks_read.saturating_sub(stats_before.chunks_read)), + blocks_read: stats.map(|stats| stats.blocks_read.saturating_sub(stats_before.blocks_read)), + buffered_bytes: runner.map(RunningQuery::buffered_bytes) + }); + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + use crate::{ + metrics::build_metrics_registry, + settings::{RuntimeSettings, Settings, ShadowSettings, SlowTaskSettings} + }; + + #[test] + fn time_of_nested_tasks_does_not_make_a_task_slow() { + let settings = RuntimeSettings::new(Settings { + shadow: ShadowSettings { + percent: 0.0, + slow_ratio: 2.0, + slow_margin_ms: 50, + log_limit: 0 + }, + slow_tasks: SlowTaskSettings { + threshold_ms: 500, + log_limit: 0 + } + }) + .unwrap(); + let slow_tasks = SlowTasks::new(Arc::new(settings)); + let client_id = ClientId::new("slow-task-test"); + let record = |own_ms: u64, nested_ms: u64| { + slow_tasks.record(&TaskReport { + dataset_id: DatasetId::from_str("slow-task-test"), + client_id: &client_id, + body: b"{}", + step: QueryStep::Next, + time: TaskTime { + total: Duration::from_millis(own_ms + nested_ms), + nested: Duration::from_millis(nested_ms) + }, + failed: false, + chunks_read: Some(1), + blocks_read: Some(1), + buffered_bytes: Some(0) + }) + }; + + record(499, 0); + record(400, 600); + record(500, 0); + + let mut output = String::new(); + prometheus_client::encoding::text::encode(&mut output, &build_metrics_registry()).unwrap(); + let counted = output + .lines() + .find(|line| line.starts_with("hotblocks_query_slow_tasks_total") && line.contains("slow-task-test")) + .unwrap_or_else(|| panic!("no slow task series:\n{output}")); + assert!(counted.ends_with(" 1"), "{counted}"); + } +} diff --git a/crates/hotblocks/src/settings.rs b/crates/hotblocks/src/settings.rs index d0657f12..57dc10b1 100644 --- a/crates/hotblocks/src/settings.rs +++ b/crates/hotblocks/src/settings.rs @@ -17,7 +17,8 @@ pub type SettingsRef = Arc; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct Settings { - pub shadow: ShadowSettings + pub shadow: ShadowSettings, + pub slow_tasks: SlowTaskSettings } #[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] @@ -39,6 +40,22 @@ impl ShadowSettings { } } +#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct SlowTaskSettings { + /// A query task is slow when it holds a query pool worker this long. + pub threshold_ms: u64, + /// Detailed log lines per 5 minutes, at most one per query shape; 0 turns + /// them off. + pub log_limit: usize +} + +impl SlowTaskSettings { + pub fn threshold(&self) -> Duration { + Duration::from_millis(self.threshold_ms) + } +} + impl Settings { fn validate(&self) -> anyhow::Result<()> { let percent = self.shadow.percent; @@ -71,6 +88,10 @@ impl RuntimeSettings { self.current.read().shadow } + pub fn slow_tasks(&self) -> SlowTaskSettings { + self.current.read().slow_tasks + } + /// Apply a JSON Merge Patch (RFC 7386). A patch that leaves the settings /// invalid, names an unknown key or removes a key changes nothing. pub fn update(&self, patch: Value) -> anyhow::Result { @@ -142,6 +163,10 @@ mod tests { slow_ratio: 2.0, slow_margin_ms: 50, log_limit: 20 + }, + slow_tasks: SlowTaskSettings { + threshold_ms: 500, + log_limit: 20 } }) .unwrap() @@ -150,6 +175,7 @@ mod tests { #[test] fn a_patch_changes_only_the_keys_it_names() { let settings = settings(); + let before = settings.get(); let updated = settings .update(json!({"shadow": {"percent": 5, "logLimit": 0}})) @@ -163,6 +189,7 @@ mod tests { }; assert_eq!(updated.shadow, expected); assert_eq!(settings.shadow(), expected); + assert_eq!(updated.slow_tasks, before.slow_tasks); } #[test] diff --git a/crates/hotblocks/src/shadow/mod.rs b/crates/hotblocks/src/shadow/mod.rs index 1264644d..74ad4def 100644 --- a/crates/hotblocks/src/shadow/mod.rs +++ b/crates/hotblocks/src/shadow/mod.rs @@ -14,8 +14,6 @@ mod compare; use std::{ - collections::HashSet, - hash::{Hash, Hasher}, panic::{AssertUnwindSafe, catch_unwind}, sync::{ Arc, @@ -35,6 +33,7 @@ use tracing::{error, warn}; use self::compare::Difference; use crate::{ + log_window::{LogWindow, query_fingerprint, truncate_query}, metrics::{ ShadowOutcome, report_shadow_query, report_shadow_run, report_shadow_run_finished, report_shadow_run_started, report_shadow_timing @@ -45,12 +44,6 @@ use crate::{ types::DatasetKind }; -/// 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 [`ShadowSettings::log_limit`] counts log lines in. -const LOG_WINDOW: Duration = Duration::from_secs(300); - pub type ShadowRef = Arc; pub struct Shadow { @@ -73,10 +66,7 @@ impl Shadow { .context("failed to start the shadow threads")?; Ok(Self { in_flight: AtomicUsize::new(0), - log_window: Mutex::new(LogWindow { - started: Instant::now(), - logged: HashSet::new() - }), + log_window: Mutex::new(LogWindow::new()), settings, max_in_flight, catalogs, @@ -136,7 +126,7 @@ impl Shadow { ShadowOutcome::Match if slow => "slow", _ => return }; - let fingerprint = run.query.fingerprint(); + let fingerprint = query_fingerprint(dataset_id, &run.query.body); if !self.log_window.lock().claim(fingerprint, reason, settings.log_limit) { return; } @@ -201,23 +191,6 @@ impl ShadowQuery { drop(slot); }); } - - /// The query with its block range and parent hash removed: what stays the - /// same while a client follows the chain. - fn fingerprint(&self) -> u64 { - let mut hasher = std::collections::hash_map::DefaultHasher::new(); - self.dataset_id.as_str().hash(&mut hasher); - match serde_json::from_slice::(&self.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); - } - _ => self.body.hash(&mut hasher) - } - hasher.finish() - } } struct InFlight(ShadowRef); @@ -484,18 +457,6 @@ fn guarded(run: impl FnOnce() -> anyhow::Result) -> Result { } } -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("…"); - } - query -} - fn panic_message(panic: &Box) -> &str { panic .downcast_ref::() @@ -503,21 +464,3 @@ fn panic_message(panic: &Box) -> &str { .or_else(|| panic.downcast_ref::<&str>().copied()) .unwrap_or("non-string panic payload") } - -/// The queries logged in detail in the current [`LOG_WINDOW`]. -struct LogWindow { - started: Instant, - /// Query fingerprints and reasons. - logged: HashSet<(u64, &'static str)> -} - -impl LogWindow { - /// Whether this query may be logged in detail for this reason now. - 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)) - } -} diff --git a/crates/hotblocks/tests/query_task_metrics.rs b/crates/hotblocks/tests/query_task_metrics.rs index 3e5643f1..2ab434e7 100644 --- a/crates/hotblocks/tests/query_task_metrics.rs +++ b/crates/hotblocks/tests/query_task_metrics.rs @@ -4,6 +4,7 @@ use std::sync::Arc; use anyhow::{Result, ensure}; +use serde_json::json; use sqd_hotblocks_harness::{ Harness, HarnessConfig, Outcome, chain::{Chain, Evm}, @@ -13,6 +14,7 @@ use sqd_hotblocks_harness::{ const START: u64 = 1_000; const QUEUE: &str = "hotblocks_query_task_queue_seconds_count"; const RUN: &str = "hotblocks_query_task_run_seconds_count"; +const SLOW: &str = "hotblocks_query_slow_tasks_total"; #[tokio::test(flavor = "multi_thread")] async fn a_query_task_is_timed_in_the_queue_and_on_a_worker() -> Result<()> { @@ -42,3 +44,61 @@ async fn a_query_task_is_timed_in_the_queue_and_on_a_worker() -> Result<()> { assert_eq!(after.get("hotblocks_query_tasks_queued", None), Some(0.0)); Ok(()) } + +#[tokio::test(flavor = "multi_thread")] +async fn a_task_over_the_slow_threshold_is_counted_and_the_threshold_changes_at_runtime() -> Result<()> { + let mut cfg = HarnessConfig::from_block(env!("CARGO_BIN_EXE_sqd-hotblocks"), Arc::new(Evm), START); + cfg.sut_args + .extend(["--slow-task-threshold-ms", "0"].into_iter().map(String::from)); + let mut h = Harness::start(cfg).await?; + h.produce(10)?; + h.settle().await?; + let query = Evm.scan_query(START, None, None); + + let slow_first_tasks = async |h: &Harness| -> Result { + let metrics = h.client.metrics().await?; + Ok(metrics.get(SLOW, Some(("step", "first"))).unwrap_or(0.0)) + }; + + let outcome = h.client.query(&query).await?; + ensure!( + matches!(outcome, Outcome::Ok { .. }), + "{query} did not return data: {outcome:?}" + ); + assert_eq!(slow_first_tasks(&h).await?, 1.0); + + assert_eq!( + h.client + .update_settings(&json!({"slowTasks": {"thresholdMs": 60_000}})) + .await?, + 200 + ); + let outcome = h.client.query(&query).await?; + ensure!( + matches!(outcome, Outcome::Ok { .. }), + "{query} did not return data: {outcome:?}" + ); + assert_eq!(slow_first_tasks(&h).await?, 1.0, "a fast task was counted"); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_first_task_that_fails_is_still_counted() -> Result<()> { + let mut cfg = HarnessConfig::from_block(env!("CARGO_BIN_EXE_sqd-hotblocks"), Arc::new(Evm), START); + cfg.sut_args + .extend(["--slow-task-threshold-ms", "0"].into_iter().map(String::from)); + let mut h = Harness::start(cfg).await?; + h.produce(10)?; + h.settle().await?; + + let below_retained = Evm.scan_query(START - 1, None, None); + let outcome = h.client.query(&below_retained).await?; + ensure!( + matches!(outcome, Outcome::Error { status: 400, .. }), + "{below_retained} did not fail in its first task: {outcome:?}" + ); + + let metrics = h.client.metrics().await?; + assert_eq!(metrics.get(SLOW, Some(("step", "first"))), Some(1.0)); + Ok(()) +}