From f415cd4bee24e2c7462f3a1a784a7275c9b75585 Mon Sep 17 00:00:00 2001 From: elina-chertova Date: Thu, 11 Jun 2026 10:25:34 +0300 Subject: [PATCH 1/3] Improve failure observability for the incident-resolution agent The agent has to diagnose portal alerts from pod logs alone. Several failure classes left no usable trace there: - worker-query failures were metrics-only: emit one warn per failure with peer_id and failure kind (timeout / transport_error / invalid_response / read_error) - the query_worker span records peer_id, dataset, chunk, block range and request_id, so every per-query event is attributable - "Stream got interrupted" carries dataset, request_id and stream_index - "Stream finished" carries request_id so it can be joined with the access-log line of the same request - panics are logged through tracing (message + location) before the default handler runs; task panics used to die as bare stderr text - the head-lookup task panic no longer fails the whole stream response via bare unwrap: it is logged and the head header is omitted - blockchain-state fetch failures carry a consecutive_failures counter and escalate warn -> error after 5 in a row --- src/controller/task_manager.rs | 13 ++++++++++- src/endpoints/stream.rs | 13 +++++++++-- src/main.rs | 24 +++++++++++++++++++ src/network/client.rs | 42 +++++++++++++++++++++++++++++++--- src/utils/logging.rs | 3 +++ 5 files changed, 89 insertions(+), 6 deletions(-) diff --git a/src/controller/task_manager.rs b/src/controller/task_manager.rs index 569ab24..3800d95 100644 --- a/src/controller/task_manager.rs +++ b/src/controller/task_manager.rs @@ -74,6 +74,8 @@ impl TaskManager { } let stream_index = self.next_stream_index.fetch_add(1, Ordering::Relaxed); + let dataset = request.dataset_id.clone(); + let request_id = request.request_id.clone(); let mut streamer = StreamController::new( request, self.network_client.clone(), @@ -93,7 +95,16 @@ impl TaskManager { None => break, Some(Ok(chunk)) => yield chunk, Some(Err(e)) => { - tracing::warn!("Stream got interrupted: {:?}", e); + // This warn is often the only signal of a systemic + // degradation (all streams breaking at once), so it + // must be attributable: which dataset, which request. + tracing::warn!( + dataset = %dataset, + request_id, + stream_index, + error = ?e, + "Stream got interrupted" + ); // There is no way to pass the error to the client break; } diff --git a/src/endpoints/stream.rs b/src/endpoints/stream.rs index 619bdef..6637597 100644 --- a/src/endpoints/stream.rs +++ b/src/endpoints/stream.rs @@ -317,8 +317,17 @@ async fn stream_from_network( res = res.header(FINALIZED_NUMBER_HEADER, head.number); res = res.header(FINALIZED_HASH_HEADER, head.hash); } - if let Some(head) = head_task.await.unwrap() { - res = res.header(HEAD_NUMBER_HEADER, head); + match head_task.await { + Ok(Some(head)) => { + res = res.header(HEAD_NUMBER_HEADER, head); + } + Ok(None) => {} + Err(e) => { + // A panicked head lookup must not 500 the whole stream (the header + // is best-effort), but it must leave a trace — a bare `.unwrap()` + // here used to kill the handler with no error-level log at all. + tracing::error!(error = %e, "head lookup task panicked; omitting head header"); + } } let body = response_body(stream, compression, config.use_gzjoin); res.header(header::CONTENT_TYPE, "application/jsonl") diff --git a/src/main.rs b/src/main.rs index bdde86e..b2787b5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -101,6 +101,29 @@ fn setup_tracing(json: bool, log_span_durations: bool) { .init(); } +/// Panics in spawned tokio tasks bypass `tracing` entirely and die as bare +/// stderr text — invisible to JSON log pipelines, so a crash-looping pod shows +/// no structured cause. Log the panic through `tracing` before the default +/// handler runs (the default still prints the backtrace and aborts as before). +fn install_panic_hook() { + let default_hook = std::panic::take_hook(); + std::panic::set_hook(Box::new(move |info| { + let message = if let Some(s) = info.payload().downcast_ref::<&str>() { + (*s).to_string() + } else if let Some(s) = info.payload().downcast_ref::() { + s.clone() + } else { + "unknown panic payload".to_string() + }; + let location = info + .location() + .map(|l| l.to_string()) + .unwrap_or_else(|| "unknown".to_string()); + tracing::error!(panic = true, message, location, "thread panicked"); + default_hook(info); + })); +} + #[tokio::main] async fn main() -> anyhow::Result<()> { dotenv::dotenv().ok(); @@ -113,6 +136,7 @@ async fn main() -> anyhow::Result<()> { .then(|| setup_sentry(&args.config, &args)); setup_tracing(args.json_log, args.log_span_durations); + install_panic_hook(); let datasets = Arc::new(RwLock::new(Datasets::load(&args.config).await?, "datasets")); diff --git a/src/network/client.rs b/src/network/client.rs index a82359e..215dd29 100644 --- a/src/network/client.rs +++ b/src/network/client.rs @@ -318,6 +318,7 @@ impl NetworkClient { let mut first_fetch = true; let mut current_epoch: u32 = 0; let mut operator; + let mut consecutive_failures: u32 = 0; loop { if first_iteration { first_iteration = false; @@ -339,9 +340,24 @@ impl NetworkClient { epoch_started, compute_units_per_epoch, ) = match self.fetch_blockchain_state().await { - Ok(data) => data, + Ok(data) => { + consecutive_failures = 0; + data + } Err(e) => { - tracing::warn!("Couldn't get blockchain data: {e}"); + // Fires once per polling interval while the chain RPC is + // unavailable; the counter separates a one-off blip from a + // persistent outage (contracts_state silently going stale). + consecutive_failures += 1; + if consecutive_failures >= 5 { + tracing::error!( + consecutive_failures, + error = %e, + "Couldn't get blockchain data; contracts state is stale" + ); + } else { + tracing::warn!(consecutive_failures, error = %e, "Couldn't get blockchain data"); + } continue; } }; @@ -471,7 +487,15 @@ impl NetworkClient { self.network_state.get_height(dataset) } - #[instrument(skip_all, level = "debug", fields(query_id))] + #[instrument(skip_all, level = "debug", fields( + query_id, + peer_id = %lease.worker(), + dataset = %chunk_id.dataset, + chunk = %chunk_id.chunk, + first_block = *block_range.start(), + last_block = *block_range.end(), + request_id, + ))] pub async fn query_worker( self: Arc, lease: WorkerLease, @@ -695,16 +719,25 @@ impl NetworkClient { }) } + // Worker-query failures used to be metrics-only: the most common cause of + // elevated 5xx (worker timeouts / transport errors) left no trace in logs. + // Emit one warn per failure so log readers can localize the failing peer + // and failure kind; events inherit the `query_worker` span fields + // (dataset/chunk/request_id). Known risk: volume scales with the failure + // rate, so a large incident produces a noticeable stream of warns — if + // that proves too noisy, rate-limit here rather than dropping the events. fn convert_query_failure(&self, peer_id: PeerId, failure: QueryFailure) -> QueryError { match failure { QueryFailure::InvalidRequest(e) => { metrics::report_query_result(&peer_id, "invalid"); self.network_state.report_query_success(peer_id, None); + tracing::warn!(peer_id = %peer_id, kind = "invalid_request", error = %e, "worker query failed"); QueryError::Failure(format!("portal tried to send invalid request: {e}")) } QueryFailure::InvalidResponse(e) => { metrics::report_query_result(&peer_id, "invalid"); self.network_state.report_query_error(peer_id); + tracing::warn!(peer_id = %peer_id, kind = "invalid_response", error = %e, "worker query failed"); QueryError::Retriable(format!("couldn't decode response: {e}")) } QueryFailure::Timeout(t) => { @@ -714,11 +747,13 @@ impl NetworkClient { StreamClientTimeout::Connect => "timed out connecting to the peer", StreamClientTimeout::Request => "timed out reading response", }; + tracing::warn!(peer_id = %peer_id, kind = "timeout", phase = msg, "worker query failed"); QueryError::Retriable(msg.to_owned()) } QueryFailure::TransportError(e) => { metrics::report_query_result(&peer_id, "transport_error"); self.network_state.report_query_failure(peer_id); + tracing::warn!(peer_id = %peer_id, kind = "transport_error", error = %e, "worker query failed"); QueryError::Retriable(format!("transport error: {e}")) } } @@ -731,6 +766,7 @@ impl NetworkClient { ReadError::TooLarge => "response too large".to_owned(), ReadError::Transport(e) => format!("transport error: {e}"), }; + tracing::warn!(peer_id = %peer_id, kind = "read_error", error = %msg, "worker query failed"); QueryError::Retriable(msg) } diff --git a/src/utils/logging.rs b/src/utils/logging.rs index 180a01a..ccd963c 100644 --- a/src/utils/logging.rs +++ b/src/utils/logging.rs @@ -86,6 +86,9 @@ impl StreamStats { // ); tracing::info!( dataset = %request.dataset_id, + // The summary is written on Drop, outside the `http_request` span — + // without an explicit field it cannot be joined with the access log. + request_id = %request.request_id, first_block = request.query.first_block(), last_block = request.query.last_block(), queries_sent = self.queries_sent, From 575fb8270b9f933b4522f29eed6ab1c21aff25fa Mon Sep 17 00:00:00 2001 From: elina-chertova Date: Thu, 2 Jul 2026 11:02:30 +0300 Subject: [PATCH 2/3] Aggregate worker failure logging --- src/controller/stream.rs | 15 +++++ src/network/client.rs | 36 ++++++------ src/types/errors.rs | 27 +++++++++ src/types/mod.rs | 2 +- src/utils/logging.rs | 120 ++++++++++++++++++++++++++++++++++++++- 5 files changed, 180 insertions(+), 20 deletions(-) diff --git a/src/controller/stream.rs b/src/controller/stream.rs index fc7550d..09cf43d 100644 --- a/src/controller/stream.rs +++ b/src/controller/stream.rs @@ -454,6 +454,10 @@ impl StreamController { } }; + if let Err(error) = &response { + self.stats.worker_query_failed(running.worker, error); + } + if retriable(&response) { tracing::debug!( "Got retriable error in {}ms from {}: {}", @@ -1109,6 +1113,7 @@ fn retriable(result: &QueryResult) -> bool { Ok(_) => false, Err(QueryError::BadRequest(_)) => false, Err(QueryError::Retriable(_)) => true, + Err(QueryError::WorkerFailure { .. }) => true, Err(QueryError::Failure(_)) => false, Err(QueryError::RateLimitExceeded) => true, Err(QueryError::BaseBlockMismatch(_)) => false, @@ -1255,6 +1260,16 @@ mod tests { assert!(!continuation_query.contains("parentBlockHash")); } + #[test] + fn worker_failures_remain_retriable() { + let result = Err(QueryError::WorkerFailure { + kind: crate::types::WorkerFailureKind::Timeout, + message: "timed out reading response".to_string(), + }); + + assert!(retriable(&result)); + } + #[test] fn chunk_slot_keeps_pending_responses_before_active_slot() { let mut chunk_slot = ChunkSlot::new(Slot { diff --git a/src/network/client.rs b/src/network/client.rs index 215dd29..2b1775b 100644 --- a/src/network/client.rs +++ b/src/network/client.rs @@ -34,7 +34,7 @@ use crate::utils::{RwLock, UseOnce}; use crate::{ config::Config, metrics, - types::{generate_query_id, DatasetId, QueryError}, + types::{generate_query_id, DatasetId, QueryError, WorkerFailureKind}, }; #[derive(Debug)] @@ -719,26 +719,22 @@ impl NetworkClient { }) } - // Worker-query failures used to be metrics-only: the most common cause of - // elevated 5xx (worker timeouts / transport errors) left no trace in logs. - // Emit one warn per failure so log readers can localize the failing peer - // and failure kind; events inherit the `query_worker` span fields - // (dataset/chunk/request_id). Known risk: volume scales with the failure - // rate, so a large incident produces a noticeable stream of warns — if - // that proves too noisy, rate-limit here rather than dropping the events. + // Keep per-attempt failures structured so the stream summary can report + // totals by kind and top failing worker without emitting a warn per retry. fn convert_query_failure(&self, peer_id: PeerId, failure: QueryFailure) -> QueryError { match failure { QueryFailure::InvalidRequest(e) => { metrics::report_query_result(&peer_id, "invalid"); self.network_state.report_query_success(peer_id, None); - tracing::warn!(peer_id = %peer_id, kind = "invalid_request", error = %e, "worker query failed"); QueryError::Failure(format!("portal tried to send invalid request: {e}")) } QueryFailure::InvalidResponse(e) => { metrics::report_query_result(&peer_id, "invalid"); self.network_state.report_query_error(peer_id); - tracing::warn!(peer_id = %peer_id, kind = "invalid_response", error = %e, "worker query failed"); - QueryError::Retriable(format!("couldn't decode response: {e}")) + QueryError::WorkerFailure { + kind: WorkerFailureKind::InvalidResponse, + message: format!("couldn't decode response: {e}"), + } } QueryFailure::Timeout(t) => { metrics::report_query_result(&peer_id, "timeout"); @@ -747,14 +743,18 @@ impl NetworkClient { StreamClientTimeout::Connect => "timed out connecting to the peer", StreamClientTimeout::Request => "timed out reading response", }; - tracing::warn!(peer_id = %peer_id, kind = "timeout", phase = msg, "worker query failed"); - QueryError::Retriable(msg.to_owned()) + QueryError::WorkerFailure { + kind: WorkerFailureKind::Timeout, + message: msg.to_owned(), + } } QueryFailure::TransportError(e) => { metrics::report_query_result(&peer_id, "transport_error"); self.network_state.report_query_failure(peer_id); - tracing::warn!(peer_id = %peer_id, kind = "transport_error", error = %e, "worker query failed"); - QueryError::Retriable(format!("transport error: {e}")) + QueryError::WorkerFailure { + kind: WorkerFailureKind::TransportError, + message: format!("transport error: {e}"), + } } } } @@ -766,8 +766,10 @@ impl NetworkClient { ReadError::TooLarge => "response too large".to_owned(), ReadError::Transport(e) => format!("transport error: {e}"), }; - tracing::warn!(peer_id = %peer_id, kind = "read_error", error = %msg, "worker query failed"); - QueryError::Retriable(msg) + QueryError::WorkerFailure { + kind: WorkerFailureKind::ReadError, + message: msg, + } } #[instrument(skip_all, level = "debug")] diff --git a/src/types/errors.rs b/src/types/errors.rs index 90f80b5..99cdd41 100644 --- a/src/types/errors.rs +++ b/src/types/errors.rs @@ -5,6 +5,25 @@ use serde::Serialize; use sqd_contract_client::PeerId; use tokio::time::Instant; +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub enum WorkerFailureKind { + InvalidResponse, + Timeout, + TransportError, + ReadError, +} + +impl WorkerFailureKind { + pub fn as_str(self) -> &'static str { + match self { + Self::InvalidResponse => "invalid_response", + Self::Timeout => "timeout", + Self::TransportError => "transport_error", + Self::ReadError => "read_error", + } + } +} + #[derive(thiserror::Error, Debug)] pub enum RequestError { #[error("{0}")] @@ -31,6 +50,11 @@ pub enum QueryError { BadRequest(String), #[error("{0}")] Retriable(String), + #[error("{message}")] + WorkerFailure { + kind: WorkerFailureKind, + message: String, + }, #[error("{0}")] Failure(String), #[error("rate limit exceeded")] @@ -54,6 +78,9 @@ impl RequestError { QueryError::Retriable(s) => { RequestError::InternalError(format!("received an error from worker {worker}: {s}")) } + QueryError::WorkerFailure { message, .. } => RequestError::InternalError(format!( + "received an error from worker {worker}: {message}" + )), QueryError::Failure(s) => RequestError::Failure(format!("worker {worker} failed: {s}")), QueryError::RateLimitExceeded => RequestError::RateLimitExceeded, QueryError::BaseBlockMismatch(block_ref) => RequestError::BaseBlockMismatch(block_ref), diff --git a/src/types/mod.rs b/src/types/mod.rs index 4882949..a7d8af2 100644 --- a/src/types/mod.rs +++ b/src/types/mod.rs @@ -9,5 +9,5 @@ pub use api_types::DatasetRef; pub use chunk::*; pub use common::*; pub use dataset::*; -pub use errors::{GenericError, QueryError, RequestError, SendQueryError}; +pub use errors::{GenericError, QueryError, RequestError, SendQueryError, WorkerFailureKind}; pub use request::*; diff --git a/src/utils/logging.rs b/src/utils/logging.rs index ccd963c..5129446 100644 --- a/src/utils/logging.rs +++ b/src/utils/logging.rs @@ -3,13 +3,17 @@ use axum::{ response::{IntoResponse, Response}, routing::MethodRouter, }; +use std::collections::BTreeMap; use std::task::{Context, Poll}; use tokio::time::{Duration, Instant}; use tower::{Layer, Service}; use tower_http::request_id::RequestId; use tracing::Instrument; -use crate::{metrics, types::StreamRequest}; +use crate::{ + metrics, + types::{QueryError, StreamRequest, WorkerFailureKind}, +}; const LOG_INTERVAL: Duration = Duration::from_secs(5); const NO_DATA_SOURCE: &str = "none"; @@ -23,6 +27,14 @@ pub struct StreamStats { pub start_time: Instant, pub last_log: Instant, pub throttled_for: Duration, + worker_failures: WorkerFailureStats, +} + +#[derive(Debug, Default)] +struct WorkerFailureStats { + total: u64, + by_kind: BTreeMap<&'static str, u64>, + by_worker: BTreeMap, } impl Default for StreamStats { @@ -43,6 +55,7 @@ impl StreamStats { start_time: now, last_log: now, throttled_for: Duration::from_secs(0), + worker_failures: WorkerFailureStats::default(), } } @@ -64,14 +77,26 @@ impl StreamStats { self.throttled_for += duration; } + pub fn worker_query_failed(&mut self, worker: impl ToString, error: &QueryError) { + if let QueryError::WorkerFailure { kind, .. } = error { + self.worker_failures.record(worker.to_string(), *kind); + } + } + pub fn maybe_write_log(&mut self) { if self.last_log.elapsed() >= LOG_INTERVAL { + let (top_failure_worker, top_failure_worker_failures) = + self.worker_failures.top_worker(); tracing::info!( queries_sent = self.queries_sent, chunks_downloaded = self.chunks_downloaded, max_chunk_parts = self.max_chunk_parts, blocks_streamed = self.response_blocks, bytes_streamed = self.response_bytes, + worker_failures = self.worker_failures.total, + worker_failure_counts = %self.worker_failures.counts_summary(), + top_failure_worker = %top_failure_worker, + top_failure_worker_failures, "Streaming..." ); self.last_log = Instant::now(); @@ -84,6 +109,7 @@ impl StreamStats { // query = request.query.to_string(), // "Query processed" // ); + let (top_failure_worker, top_failure_worker_failures) = self.worker_failures.top_worker(); tracing::info!( dataset = %request.dataset_id, // The summary is written on Drop, outside the `http_request` span — @@ -98,6 +124,10 @@ impl StreamStats { bytes_streamed = self.response_bytes, total_time = ?self.start_time.elapsed(), throttled_for = ?self.throttled_for, + worker_failures = self.worker_failures.total, + worker_failure_counts = %self.worker_failures.counts_summary(), + top_failure_worker = %top_failure_worker, + top_failure_worker_failures, error = error.unwrap_or_else(|| "-".to_string()), "Stream finished" ); @@ -105,6 +135,43 @@ impl StreamStats { } } +impl WorkerFailureStats { + fn record(&mut self, worker: String, kind: WorkerFailureKind) { + self.total += 1; + *self.by_kind.entry(kind.as_str()).or_default() += 1; + *self.by_worker.entry(worker).or_default() += 1; + } + + fn counts_summary(&self) -> String { + if self.total == 0 { + return "-".to_string(); + } + let mut counts = self.by_kind.iter().collect::>(); + counts.sort_by(|(left_kind, left_count), (right_kind, right_count)| { + right_count + .cmp(left_count) + .then_with(|| left_kind.cmp(right_kind)) + }); + counts + .iter() + .map(|(kind, count)| format!("{kind}={count}")) + .collect::>() + .join(" ") + } + + fn top_worker(&self) -> (String, u64) { + self.by_worker + .iter() + .max_by(|(left_worker, left_count), (right_worker, right_count)| { + left_count + .cmp(right_count) + .then_with(|| right_worker.cmp(left_worker)) + }) + .map(|(worker, count)| (worker.clone(), *count)) + .unwrap_or_else(|| ("-".to_string(), 0)) + } +} + pub async fn middleware(req: Request, next: axum::middleware::Next) -> impl IntoResponse { let method = req.method().to_string(); let path = req.uri().path().to_string(); @@ -182,8 +249,57 @@ where #[cfg(test)] mod tests { - use super::data_source_metric_label; + use super::{data_source_metric_label, StreamStats}; use crate::endpoints::stream::{DATA_SOURCE_NETWORK_METRIC, DATA_SOURCE_REALTIME_METRIC}; + use crate::types::{QueryError, WorkerFailureKind}; + + #[test] + fn worker_failure_stats_aggregate_by_kind_and_top_worker() { + let mut stats = StreamStats::new(); + + stats.worker_query_failed( + "worker-a", + &QueryError::WorkerFailure { + kind: WorkerFailureKind::Timeout, + message: "timed out reading response".to_string(), + }, + ); + stats.worker_query_failed( + "worker-b", + &QueryError::WorkerFailure { + kind: WorkerFailureKind::TransportError, + message: "transport error: closed".to_string(), + }, + ); + stats.worker_query_failed( + "worker-a", + &QueryError::WorkerFailure { + kind: WorkerFailureKind::Timeout, + message: "timed out connecting to the peer".to_string(), + }, + ); + stats.worker_query_failed( + "worker-a", + &QueryError::Retriable("not counted".to_string()), + ); + stats.worker_query_failed( + "worker-c", + &QueryError::WorkerFailure { + kind: WorkerFailureKind::InvalidResponse, + message: "couldn't decode response".to_string(), + }, + ); + + assert_eq!(stats.worker_failures.total, 4); + assert_eq!( + stats.worker_failures.counts_summary(), + "timeout=2 invalid_response=1 transport_error=1" + ); + assert_eq!( + stats.worker_failures.top_worker(), + ("worker-a".to_string(), 2) + ); + } #[test] fn data_source_metric_label_keeps_network() { From 816728c494da14aa7112a7510dadface5113396b Mon Sep 17 00:00:00 2001 From: elina-chertova Date: Thu, 2 Jul 2026 11:06:12 +0300 Subject: [PATCH 3/3] Revert "Aggregate worker failure logging" This reverts commit 575fb8270b9f933b4522f29eed6ab1c21aff25fa. --- src/controller/stream.rs | 15 ----- src/network/client.rs | 36 ++++++------ src/types/errors.rs | 27 --------- src/types/mod.rs | 2 +- src/utils/logging.rs | 120 +-------------------------------------- 5 files changed, 20 insertions(+), 180 deletions(-) diff --git a/src/controller/stream.rs b/src/controller/stream.rs index 09cf43d..fc7550d 100644 --- a/src/controller/stream.rs +++ b/src/controller/stream.rs @@ -454,10 +454,6 @@ impl StreamController { } }; - if let Err(error) = &response { - self.stats.worker_query_failed(running.worker, error); - } - if retriable(&response) { tracing::debug!( "Got retriable error in {}ms from {}: {}", @@ -1113,7 +1109,6 @@ fn retriable(result: &QueryResult) -> bool { Ok(_) => false, Err(QueryError::BadRequest(_)) => false, Err(QueryError::Retriable(_)) => true, - Err(QueryError::WorkerFailure { .. }) => true, Err(QueryError::Failure(_)) => false, Err(QueryError::RateLimitExceeded) => true, Err(QueryError::BaseBlockMismatch(_)) => false, @@ -1260,16 +1255,6 @@ mod tests { assert!(!continuation_query.contains("parentBlockHash")); } - #[test] - fn worker_failures_remain_retriable() { - let result = Err(QueryError::WorkerFailure { - kind: crate::types::WorkerFailureKind::Timeout, - message: "timed out reading response".to_string(), - }); - - assert!(retriable(&result)); - } - #[test] fn chunk_slot_keeps_pending_responses_before_active_slot() { let mut chunk_slot = ChunkSlot::new(Slot { diff --git a/src/network/client.rs b/src/network/client.rs index 2b1775b..215dd29 100644 --- a/src/network/client.rs +++ b/src/network/client.rs @@ -34,7 +34,7 @@ use crate::utils::{RwLock, UseOnce}; use crate::{ config::Config, metrics, - types::{generate_query_id, DatasetId, QueryError, WorkerFailureKind}, + types::{generate_query_id, DatasetId, QueryError}, }; #[derive(Debug)] @@ -719,22 +719,26 @@ impl NetworkClient { }) } - // Keep per-attempt failures structured so the stream summary can report - // totals by kind and top failing worker without emitting a warn per retry. + // Worker-query failures used to be metrics-only: the most common cause of + // elevated 5xx (worker timeouts / transport errors) left no trace in logs. + // Emit one warn per failure so log readers can localize the failing peer + // and failure kind; events inherit the `query_worker` span fields + // (dataset/chunk/request_id). Known risk: volume scales with the failure + // rate, so a large incident produces a noticeable stream of warns — if + // that proves too noisy, rate-limit here rather than dropping the events. fn convert_query_failure(&self, peer_id: PeerId, failure: QueryFailure) -> QueryError { match failure { QueryFailure::InvalidRequest(e) => { metrics::report_query_result(&peer_id, "invalid"); self.network_state.report_query_success(peer_id, None); + tracing::warn!(peer_id = %peer_id, kind = "invalid_request", error = %e, "worker query failed"); QueryError::Failure(format!("portal tried to send invalid request: {e}")) } QueryFailure::InvalidResponse(e) => { metrics::report_query_result(&peer_id, "invalid"); self.network_state.report_query_error(peer_id); - QueryError::WorkerFailure { - kind: WorkerFailureKind::InvalidResponse, - message: format!("couldn't decode response: {e}"), - } + tracing::warn!(peer_id = %peer_id, kind = "invalid_response", error = %e, "worker query failed"); + QueryError::Retriable(format!("couldn't decode response: {e}")) } QueryFailure::Timeout(t) => { metrics::report_query_result(&peer_id, "timeout"); @@ -743,18 +747,14 @@ impl NetworkClient { StreamClientTimeout::Connect => "timed out connecting to the peer", StreamClientTimeout::Request => "timed out reading response", }; - QueryError::WorkerFailure { - kind: WorkerFailureKind::Timeout, - message: msg.to_owned(), - } + tracing::warn!(peer_id = %peer_id, kind = "timeout", phase = msg, "worker query failed"); + QueryError::Retriable(msg.to_owned()) } QueryFailure::TransportError(e) => { metrics::report_query_result(&peer_id, "transport_error"); self.network_state.report_query_failure(peer_id); - QueryError::WorkerFailure { - kind: WorkerFailureKind::TransportError, - message: format!("transport error: {e}"), - } + tracing::warn!(peer_id = %peer_id, kind = "transport_error", error = %e, "worker query failed"); + QueryError::Retriable(format!("transport error: {e}")) } } } @@ -766,10 +766,8 @@ impl NetworkClient { ReadError::TooLarge => "response too large".to_owned(), ReadError::Transport(e) => format!("transport error: {e}"), }; - QueryError::WorkerFailure { - kind: WorkerFailureKind::ReadError, - message: msg, - } + tracing::warn!(peer_id = %peer_id, kind = "read_error", error = %msg, "worker query failed"); + QueryError::Retriable(msg) } #[instrument(skip_all, level = "debug")] diff --git a/src/types/errors.rs b/src/types/errors.rs index 99cdd41..90f80b5 100644 --- a/src/types/errors.rs +++ b/src/types/errors.rs @@ -5,25 +5,6 @@ use serde::Serialize; use sqd_contract_client::PeerId; use tokio::time::Instant; -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] -pub enum WorkerFailureKind { - InvalidResponse, - Timeout, - TransportError, - ReadError, -} - -impl WorkerFailureKind { - pub fn as_str(self) -> &'static str { - match self { - Self::InvalidResponse => "invalid_response", - Self::Timeout => "timeout", - Self::TransportError => "transport_error", - Self::ReadError => "read_error", - } - } -} - #[derive(thiserror::Error, Debug)] pub enum RequestError { #[error("{0}")] @@ -50,11 +31,6 @@ pub enum QueryError { BadRequest(String), #[error("{0}")] Retriable(String), - #[error("{message}")] - WorkerFailure { - kind: WorkerFailureKind, - message: String, - }, #[error("{0}")] Failure(String), #[error("rate limit exceeded")] @@ -78,9 +54,6 @@ impl RequestError { QueryError::Retriable(s) => { RequestError::InternalError(format!("received an error from worker {worker}: {s}")) } - QueryError::WorkerFailure { message, .. } => RequestError::InternalError(format!( - "received an error from worker {worker}: {message}" - )), QueryError::Failure(s) => RequestError::Failure(format!("worker {worker} failed: {s}")), QueryError::RateLimitExceeded => RequestError::RateLimitExceeded, QueryError::BaseBlockMismatch(block_ref) => RequestError::BaseBlockMismatch(block_ref), diff --git a/src/types/mod.rs b/src/types/mod.rs index a7d8af2..4882949 100644 --- a/src/types/mod.rs +++ b/src/types/mod.rs @@ -9,5 +9,5 @@ pub use api_types::DatasetRef; pub use chunk::*; pub use common::*; pub use dataset::*; -pub use errors::{GenericError, QueryError, RequestError, SendQueryError, WorkerFailureKind}; +pub use errors::{GenericError, QueryError, RequestError, SendQueryError}; pub use request::*; diff --git a/src/utils/logging.rs b/src/utils/logging.rs index 5129446..ccd963c 100644 --- a/src/utils/logging.rs +++ b/src/utils/logging.rs @@ -3,17 +3,13 @@ use axum::{ response::{IntoResponse, Response}, routing::MethodRouter, }; -use std::collections::BTreeMap; use std::task::{Context, Poll}; use tokio::time::{Duration, Instant}; use tower::{Layer, Service}; use tower_http::request_id::RequestId; use tracing::Instrument; -use crate::{ - metrics, - types::{QueryError, StreamRequest, WorkerFailureKind}, -}; +use crate::{metrics, types::StreamRequest}; const LOG_INTERVAL: Duration = Duration::from_secs(5); const NO_DATA_SOURCE: &str = "none"; @@ -27,14 +23,6 @@ pub struct StreamStats { pub start_time: Instant, pub last_log: Instant, pub throttled_for: Duration, - worker_failures: WorkerFailureStats, -} - -#[derive(Debug, Default)] -struct WorkerFailureStats { - total: u64, - by_kind: BTreeMap<&'static str, u64>, - by_worker: BTreeMap, } impl Default for StreamStats { @@ -55,7 +43,6 @@ impl StreamStats { start_time: now, last_log: now, throttled_for: Duration::from_secs(0), - worker_failures: WorkerFailureStats::default(), } } @@ -77,26 +64,14 @@ impl StreamStats { self.throttled_for += duration; } - pub fn worker_query_failed(&mut self, worker: impl ToString, error: &QueryError) { - if let QueryError::WorkerFailure { kind, .. } = error { - self.worker_failures.record(worker.to_string(), *kind); - } - } - pub fn maybe_write_log(&mut self) { if self.last_log.elapsed() >= LOG_INTERVAL { - let (top_failure_worker, top_failure_worker_failures) = - self.worker_failures.top_worker(); tracing::info!( queries_sent = self.queries_sent, chunks_downloaded = self.chunks_downloaded, max_chunk_parts = self.max_chunk_parts, blocks_streamed = self.response_blocks, bytes_streamed = self.response_bytes, - worker_failures = self.worker_failures.total, - worker_failure_counts = %self.worker_failures.counts_summary(), - top_failure_worker = %top_failure_worker, - top_failure_worker_failures, "Streaming..." ); self.last_log = Instant::now(); @@ -109,7 +84,6 @@ impl StreamStats { // query = request.query.to_string(), // "Query processed" // ); - let (top_failure_worker, top_failure_worker_failures) = self.worker_failures.top_worker(); tracing::info!( dataset = %request.dataset_id, // The summary is written on Drop, outside the `http_request` span — @@ -124,10 +98,6 @@ impl StreamStats { bytes_streamed = self.response_bytes, total_time = ?self.start_time.elapsed(), throttled_for = ?self.throttled_for, - worker_failures = self.worker_failures.total, - worker_failure_counts = %self.worker_failures.counts_summary(), - top_failure_worker = %top_failure_worker, - top_failure_worker_failures, error = error.unwrap_or_else(|| "-".to_string()), "Stream finished" ); @@ -135,43 +105,6 @@ impl StreamStats { } } -impl WorkerFailureStats { - fn record(&mut self, worker: String, kind: WorkerFailureKind) { - self.total += 1; - *self.by_kind.entry(kind.as_str()).or_default() += 1; - *self.by_worker.entry(worker).or_default() += 1; - } - - fn counts_summary(&self) -> String { - if self.total == 0 { - return "-".to_string(); - } - let mut counts = self.by_kind.iter().collect::>(); - counts.sort_by(|(left_kind, left_count), (right_kind, right_count)| { - right_count - .cmp(left_count) - .then_with(|| left_kind.cmp(right_kind)) - }); - counts - .iter() - .map(|(kind, count)| format!("{kind}={count}")) - .collect::>() - .join(" ") - } - - fn top_worker(&self) -> (String, u64) { - self.by_worker - .iter() - .max_by(|(left_worker, left_count), (right_worker, right_count)| { - left_count - .cmp(right_count) - .then_with(|| right_worker.cmp(left_worker)) - }) - .map(|(worker, count)| (worker.clone(), *count)) - .unwrap_or_else(|| ("-".to_string(), 0)) - } -} - pub async fn middleware(req: Request, next: axum::middleware::Next) -> impl IntoResponse { let method = req.method().to_string(); let path = req.uri().path().to_string(); @@ -249,57 +182,8 @@ where #[cfg(test)] mod tests { - use super::{data_source_metric_label, StreamStats}; + use super::data_source_metric_label; use crate::endpoints::stream::{DATA_SOURCE_NETWORK_METRIC, DATA_SOURCE_REALTIME_METRIC}; - use crate::types::{QueryError, WorkerFailureKind}; - - #[test] - fn worker_failure_stats_aggregate_by_kind_and_top_worker() { - let mut stats = StreamStats::new(); - - stats.worker_query_failed( - "worker-a", - &QueryError::WorkerFailure { - kind: WorkerFailureKind::Timeout, - message: "timed out reading response".to_string(), - }, - ); - stats.worker_query_failed( - "worker-b", - &QueryError::WorkerFailure { - kind: WorkerFailureKind::TransportError, - message: "transport error: closed".to_string(), - }, - ); - stats.worker_query_failed( - "worker-a", - &QueryError::WorkerFailure { - kind: WorkerFailureKind::Timeout, - message: "timed out connecting to the peer".to_string(), - }, - ); - stats.worker_query_failed( - "worker-a", - &QueryError::Retriable("not counted".to_string()), - ); - stats.worker_query_failed( - "worker-c", - &QueryError::WorkerFailure { - kind: WorkerFailureKind::InvalidResponse, - message: "couldn't decode response".to_string(), - }, - ); - - assert_eq!(stats.worker_failures.total, 4); - assert_eq!( - stats.worker_failures.counts_summary(), - "timeout=2 invalid_response=1 transport_error=1" - ); - assert_eq!( - stats.worker_failures.top_worker(), - ("worker-a".to_string(), 2) - ); - } #[test] fn data_source_metric_label_keeps_network() {