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(()) +}