diff --git a/Cargo.lock b/Cargo.lock index a8aae45c..814066c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4878,6 +4878,7 @@ version = "0.1.0" dependencies = [ "anyhow", "futures", + "prometheus-client 0.24.0", "sqd-data-client", "sqd-primitives", "tokio", diff --git a/crates/data-client/src/reqwest/client.rs b/crates/data-client/src/reqwest/client.rs index 93b662ec..9ca131bf 100644 --- a/crates/data-client/src/reqwest/client.rs +++ b/crates/data-client/src/reqwest/client.rs @@ -226,6 +226,46 @@ impl DataClient for ReqwestDataClient { fn is_retryable(&self, err: &anyhow::Error) -> bool { self.is_retryable(err) } + + fn error_kind(&self, err: &anyhow::Error) -> &'static str { + for cause in err.chain() { + if cause.downcast_ref::().is_some() { + return "http"; + } + if let Some(reqwest_error) = cause.downcast_ref::() { + if reqwest_error.is_timeout() { + return "timeout"; + } + if reqwest_error.is_connect() { + return "connect"; + } + if reqwest_error.is_status() { + return "http"; + } + if reqwest_error.is_body() || reqwest_error.is_decode() { + return "decode"; + } + if reqwest_error.is_request() { + return "request"; + } + } + if cause.downcast_ref::().is_some() { + return "io"; + } + } + "other" + } + + // Host alone collapses endpoints that differ only by port or dataset path. Built by hand + // rather than from `Url::authority()`, which carries userinfo straight into the label. + fn source_label(&self) -> String { + let host = self.url.host_str().unwrap_or("unknown"); + let path = self.url.path().trim_end_matches('/'); + match self.url.port() { + Some(port) => format!("{host}:{port}{path}"), + None => format!("{host}{path}") + } + } } #[derive(Debug)] diff --git a/crates/data-client/src/types.rs b/crates/data-client/src/types.rs index d3b26145..88679e21 100644 --- a/crates/data-client/src/types.rs +++ b/crates/data-client/src/types.rs @@ -45,4 +45,14 @@ pub trait DataClient: Send + Sync + Debug + Unpin { fn get_finalized_head(&self) -> BoxFuture<'static, anyhow::Result>>; fn is_retryable(&self, err: &anyhow::Error) -> bool; + + /// Error class for the `ingest_source_errors` metric label. + fn error_kind(&self, _err: &anyhow::Error) -> &'static str { + "other" + } + + /// Source identity (host) for the `ingest_source_errors` metric label. + fn source_label(&self) -> String { + "unknown".to_string() + } } diff --git a/crates/data-source/Cargo.toml b/crates/data-source/Cargo.toml index c2213b04..c58bd0dd 100644 --- a/crates/data-source/Cargo.toml +++ b/crates/data-source/Cargo.toml @@ -6,6 +6,7 @@ edition = "2021" [dependencies] anyhow = { workspace = true } futures = { workspace = true } +prometheus-client = { workspace = true } sqd-data-client = { path = "../data-client" } sqd-primitives = { path = "../primitives" } tokio = { workspace = true } diff --git a/crates/data-source/src/lib.rs b/crates/data-source/src/lib.rs index d0cdaf8a..8dcc78b5 100644 --- a/crates/data-source/src/lib.rs +++ b/crates/data-source/src/lib.rs @@ -1,4 +1,5 @@ mod map; +pub mod metrics; mod standard; mod types; diff --git a/crates/data-source/src/metrics.rs b/crates/data-source/src/metrics.rs new file mode 100644 index 00000000..a17f6ecd --- /dev/null +++ b/crates/data-source/src/metrics.rs @@ -0,0 +1,29 @@ +use std::sync::LazyLock; + +use prometheus_client::metrics::{counter::Counter, family::Family}; + +type Labels = Vec<(&'static str, String)>; + +/// Upstream data-source ingestion errors by `source` endpoint and `kind`. +/// Registered in the hotblocks metrics registry. +pub static INGEST_SOURCE_ERRORS: LazyLock> = LazyLock::new(Default::default); + +/// Fork signals by `source` and whether it held the contested position (`at_tip`/`above_tip`). +pub static INGEST_FORK_SIGNALS: LazyLock> = LazyLock::new(Default::default); + +/// Public because the pre-ingest head probe lives in `hotblocks` and must feed the same counter: +/// it runs before this crate's stream loop, so a total outage never reaches `on_error`. +pub fn record_ingest_source_error(source: &str, kind: &'static str) { + INGEST_SOURCE_ERRORS + .get_or_create(&vec![("source", source.to_string()), ("kind", kind.to_string())]) + .inc(); +} + +pub(crate) fn record_ingest_fork_signal(source: &str, standing: &'static str) { + INGEST_FORK_SIGNALS + .get_or_create(&vec![ + ("source", source.to_string()), + ("standing", standing.to_string()), + ]) + .inc(); +} diff --git a/crates/data-source/src/standard.rs b/crates/data-source/src/standard.rs index f896c2dd..682c2d77 100644 --- a/crates/data-source/src/standard.rs +++ b/crates/data-source/src/standard.rs @@ -77,11 +77,10 @@ impl DataSourceState { } } Poll::Ready(Ok(BlockStreamResponse::Fork(prev_blocks))) => { + let req = req.clone(); + ep.on_fork_signal(req.first_block, &prev_blocks); ep.error_counter = 0; - ep.state = EndpointState::Fork { - req: req.clone(), - prev_blocks - }; + ep.state = EndpointState::Fork { req, prev_blocks }; } Poll::Ready(Err(err)) => ep.on_error(err), Poll::Pending => return Poll::Pending @@ -220,7 +219,20 @@ impl Endpoint { } } + // A source answers 409 only at `head + 1 == from`, so in-spec hints top out at `from - 1`. + fn on_fork_signal(&self, from: BlockNumber, prev_blocks: &[BlockRef]) { + let standing = match prev_blocks.last() { + Some(top) if top.number < from.saturating_sub(1) => "above_tip", + // Empty hints are a malformed 409 the reqwest client already rejects; counting them + // as `at_tip` keeps the defect bucket clean of shapes it cannot speak to. + _ => "at_tip" + }; + crate::metrics::record_ingest_fork_signal(&self.client.source_label(), standing); + } + fn on_error(&mut self, error: anyhow::Error) { + crate::metrics::record_ingest_source_error(&self.client.source_label(), self.client.error_kind(&error)); + let backoff = [0, 100, 200, 500, 1000, 2000, 5000, 10000]; let pause = backoff[std::cmp::min(self.error_counter, backoff.len() - 1)]; if pause > 0 { diff --git a/crates/hotblocks/src/dataset_controller/dataset_controller.rs b/crates/hotblocks/src/dataset_controller/dataset_controller.rs index 1ec67edd..26b73e07 100644 --- a/crates/hotblocks/src/dataset_controller/dataset_controller.rs +++ b/crates/hotblocks/src/dataset_controller/dataset_controller.rs @@ -2,7 +2,7 @@ use std::{ops::Add, time::Duration}; use anyhow::{Context, anyhow}; use futures::{FutureExt, StreamExt, future::BoxFuture, stream::FuturesUnordered}; -use sqd_data_client::reqwest::ReqwestDataClient; +use sqd_data_client::{DataClient, reqwest::ReqwestDataClient}; use sqd_primitives::{BlockNumber, BlockRef, TransactionRef}; use sqd_storage::db::{CompactionStatus, DatasetId}; use tokio::{select, task::JoinHandle, time::Instant}; @@ -248,6 +248,7 @@ impl Ctl { match self.write_epoch(std::mem::take(&mut maybe_write)).await { Ok(_) => return, Err(err) => { + crate::metrics::report_dataset_epoch_failure(self.dataset_id, &err); error!(reason = ?err, "dataset update task failed, will restart it in 1 minute"); tokio::time::sleep(Duration::from_secs(60)).await } @@ -489,6 +490,14 @@ async fn fetch_chain_top(clients: Vec) -> BlockNumber { } }, Some((Err(err), ci)) => { + // The probe gates ingestion, and its only exit on total failure is + // `completed > 0`. With every source down it spins here forever, so + // `on_error` never runs and the counter would read zero through the + // whole stall — the one outage it exists to expose. + sqd_data_source::metrics::record_ingest_source_error( + &clients[ci].source_label(), + clients[ci].error_kind(&err) + ); if clients[ci].is_retryable(&err) { warn!( reason =? err, diff --git a/crates/hotblocks/src/dataset_controller/write_controller.rs b/crates/hotblocks/src/dataset_controller/write_controller.rs index 813c2c8c..44ce65ed 100644 --- a/crates/hotblocks/src/dataset_controller/write_controller.rs +++ b/crates/hotblocks/src/dataset_controller/write_controller.rs @@ -8,6 +8,7 @@ use tracing::{debug, field::valuable, info, instrument, warn}; use crate::{ dataset_controller::ingest_generic::{IngestMessage, NewChunk}, + errors::UnapplicableFork, metrics::{WriteStage, report_hash_index_write_metrics, report_write_duration}, types::{DBRef, DatasetKind} }; @@ -139,10 +140,17 @@ impl WriteController { if let Some(finalized_head) = label.finalized_head() { let pos = match prev.iter().position(|b| b.number >= finalized_head.number) { Some(pos) => pos, - None => bail!("all passed prev blocks lie below finalized head") + None => bail!(UnapplicableFork { + reason: "all passed prev blocks lie below finalized head" + }) }; if prev[pos].number == finalized_head.number { - ensure!(prev[pos].hash == finalized_head.hash); + ensure!( + prev[pos].hash == finalized_head.hash, + UnapplicableFork { + reason: "fork disagrees with the finalized head hash" + } + ); } prev = &prev[pos..] } diff --git a/crates/hotblocks/src/errors.rs b/crates/hotblocks/src/errors.rs index 02bc7472..03a74d4f 100644 --- a/crates/hotblocks/src/errors.rs +++ b/crates/hotblocks/src/errors.rs @@ -38,6 +38,20 @@ impl Display for QueryTaskPanicked { impl std::error::Error for QueryTaskPanicked {} +/// Divergence reaching below finalized data. Kills the update task, which then parks for a minute. +#[derive(Debug)] +pub struct UnapplicableFork { + pub reason: &'static str +} + +impl Display for UnapplicableFork { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "received fork can not be applied: {}", self.reason) + } +} + +impl std::error::Error for UnapplicableFork {} + #[derive(Debug)] pub struct UnknownDataset { pub dataset_id: DatasetId diff --git a/crates/hotblocks/src/metrics.rs b/crates/hotblocks/src/metrics.rs index f60a04ca..10ed3e0f 100644 --- a/crates/hotblocks/src/metrics.rs +++ b/crates/hotblocks/src/metrics.rs @@ -18,7 +18,7 @@ use sqd_storage::db::{ }; use tracing::error; -use crate::{query::QueryExecutorCollector, types::DBRef}; +use crate::{errors::UnapplicableFork, query::QueryExecutorCollector, types::DBRef}; #[derive(Copy, Clone, Hash, Debug, Default, Ord, PartialOrd, Eq, PartialEq, EncodeLabelSet)] struct DatasetLabel { @@ -129,6 +129,29 @@ struct WriteLabels { outcome: WriteOutcome } +#[derive(Copy, Clone, Hash, Debug, Eq, PartialEq, EncodeLabelSet)] +struct EpochFailureLabels { + dataset: DatasetValue, + reason: &'static str +} + +static DATASET_EPOCH_FAILURES: LazyLock> = LazyLock::new(Default::default); + +pub(crate) fn report_dataset_epoch_failure(dataset_id: DatasetId, err: &anyhow::Error) { + // Never label by message — it carries block numbers and hashes. + let reason = if err.chain().any(|e| e.is::()) { + "unapplicable_fork" + } else { + "other" + }; + DATASET_EPOCH_FAILURES + .get_or_create(&EpochFailureLabels { + dataset: DatasetValue(dataset_id), + reason + }) + .inc(); +} + static WRITE_DURATION: LazyLock> = LazyLock::new(|| Family::new_with_constructor(|| Histogram::new(buckets(0.0001, 28)))); @@ -464,6 +487,31 @@ pub fn build_metrics_registry() -> Registry { QUERY_ERROR_WORKER_PANIC.clone() ); + registry.register( + "ingest_source_errors", + "Upstream data source ingestion errors, by source endpoint (host:port/path) and kind \ + (connect/timeout/http/decode/io/request/other). Covers both the pre-ingest head probe \ + and the stream loop", + sqd_data_source::metrics::INGEST_SOURCE_ERRORS.clone() + ); + + registry.register( + "ingest_fork_signals", + "Fork signals from upstream data sources, by source endpoint and whether the source \ + held the contested position (standing=at_tip) or contested one above its own tip \ + (standing=above_tip). GAP-21: an honest source can also land in above_tip when its \ + hint window misses the parent across a hole wider than 100 positions", + sqd_data_source::metrics::INGEST_FORK_SIGNALS.clone() + ); + + registry.register( + "dataset_epoch_failures", + "Dataset update task failures, by dataset and cause; each one parks ingestion for \ + 60s before a full restart. reason=unapplicable_fork is a divergence reaching below \ + finalized data", + DATASET_EPOCH_FAILURES.clone() + ); + registry.register("http_status", "Number of sent HTTP responses", HTTP_STATUS.clone()); registry.register( "http_seconds_to_first_byte", diff --git a/crates/hotblocks/tests/ct4_lagging_source.rs b/crates/hotblocks/tests/ct4_lagging_source.rs index 61505b0a..81428a23 100644 --- a/crates/hotblocks/tests/ct4_lagging_source.rs +++ b/crates/hotblocks/tests/ct4_lagging_source.rs @@ -193,3 +193,87 @@ async fn ct4_a_caught_up_source_changes_nothing() -> Result<()> { Ok(()) } + +/// The wedge above is invisible twice over: the fork signal is an `Ok` that never reaches +/// `on_error`, and the park is an `error!` a crash-looping pod may never ship. Neither GAP-41 nor +/// GAP-5 is fixed here — this pins that both became countable, so a head held back by a lying +/// source reads differently from a hung service. +/// +/// Not `#[ignore]`d unlike its neighbours: observability holds, conformance does not. +#[tokio::test(flavor = "multi_thread")] +async fn ct4_a_fork_signal_above_the_tip_and_the_park_it_causes_are_observable() -> Result<()> { + let mut h = Harness::start(config(3)).await?; + + h.produce(20)?; + h.finalize_with_lag(5)?; + h.settle().await?; + + for peer in &h.peers { + peer.inject_fault(&h.dataset, |f| f.fork_signal_above_tip = true); + } + + for _ in 0..12 { + h.produce_ahead(10)?; + h.finalize_with_lag(5)?; + tokio::time::sleep(Duration::from_millis(50)).await; + } + + // No `settle` — the epoch is serving out `P-EPOCH-RETRY`, which is the thing being measured. + tokio::time::sleep(Duration::from_secs(1)).await; + let metrics = h.client.metrics().await?; + + let above_tip = metrics + .get("hotblocks_ingest_fork_signals_total", Some(("standing", "above_tip"))) + .unwrap_or_default(); + assert!(above_tip > 0.0, "a fork signalled above the source's own tip"); + + assert!( + metrics + .get("hotblocks_ingest_fork_signals_total", Some(("standing", "at_tip"))) + .is_none(), + "no source held the contested position" + ); + + let parked = metrics + .get( + "hotblocks_dataset_epoch_failures_total", + Some(("reason", "unapplicable_fork")) + ) + .unwrap_or_default(); + assert!(parked > 0.0, "the park must name the divergence as its cause"); + + Ok(()) +} + +/// The other half: an honest reorg must land in `at_tip`. Without it the label above is +/// unfalsifiable — a discriminator stuck on `above_tip` would pass that script and then cry +/// source defect on every ordinary reorg in production. +#[tokio::test(flavor = "multi_thread")] +async fn ct4_an_honest_reorg_is_not_attributed_to_the_source() -> Result<()> { + let mut h = Harness::start(config(1)).await?; + + h.produce(20)?; + h.settle().await?; + + h.fork(1_010, 10)?; + h.finalize_with_lag(5)?; + h.settle().await?; + h.assert_conforms().await?; + + let metrics = h.client.metrics().await?; + + let at_tip = metrics + .get("hotblocks_ingest_fork_signals_total", Some(("standing", "at_tip"))) + .unwrap_or_default(); + assert!(at_tip > 0.0, "the reorg reaches the service as a fork signal"); + + assert!( + metrics + .get("hotblocks_ingest_fork_signals_total", Some(("standing", "above_tip"))) + .is_none(), + "the source signalled where it stood — attributing this to a defect would make the \ + metric fire on every reorg" + ); + + Ok(()) +} diff --git a/crates/hotblocks/tests/ct9_source_faults.rs b/crates/hotblocks/tests/ct9_source_faults.rs index 016bc9f8..f80ab139 100644 --- a/crates/hotblocks/tests/ct9_source_faults.rs +++ b/crates/hotblocks/tests/ct9_source_faults.rs @@ -9,8 +9,10 @@ use std::{sync::Arc, time::Duration}; use anyhow::Result; use sqd_hotblocks_harness::{ - chain::HlFills, - harness::{Harness, HarnessConfig} + chain::{Chain, Evm, HlFills}, + driver::Client, + harness::{Harness, HarnessConfig}, + sut::{DatasetSpec, Retention, Sut, SutConfig} }; const START: u64 = 1_000; @@ -54,3 +56,58 @@ async fn run(h: &mut Harness) -> Result<()> { ); Ok(()) } + +/// Total outage is the one source fault the ingest counter could not see. The head probe gates +/// ingestion and its only exit on total failure is `completed > 0`, so with every source down it +/// spins there forever, `StandardDataSource` is never constructed and `on_error` never runs — the +/// counter would read zero through precisely the outage it exists to expose, while the service +/// stays up and scraped. +/// +/// The assertion keys on `source` rather than `kind` so it also pins the label as the full +/// endpoint: a host-only label collapses every source of every dataset into one series. +#[tokio::test(flavor = "multi_thread")] +async fn ct9_a_total_source_outage_is_counted_before_ingestion_starts() -> Result<()> { + const DS: &str = "outage"; + + // Bound to claim a port, then released: probes are refused rather than left hanging. + let dead = std::net::TcpListener::bind("127.0.0.1:0")?.local_addr()?.port(); + + let sut = Sut::start(SutConfig::new( + env!("CARGO_BIN_EXE_sqd-hotblocks"), + vec![DatasetSpec { + id: DS.to_string(), + kind: Evm.config_kind().to_string(), + // `Head` is what routes the dataset through the probe at all. + retention: Retention::Head(100), + sources: vec![format!("http://127.0.0.1:{dead}/{DS}")] + }] + )) + .await?; + + let client = Client::new(sut.base_url(), DS)?; + let source = format!("127.0.0.1:{dead}/{DS}"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(20); + + loop { + let metrics = client.metrics().await?; + if metrics + .get("hotblocks_ingest_source_errors_total", Some(("source", &source))) + .unwrap_or_default() + > 0.0 + { + assert!( + metrics + .get("hotblocks_ingest_source_errors_total", Some(("kind", "connect"))) + .unwrap_or_default() + > 0.0, + "a refused connection must classify as `connect`" + ); + return Ok(()); + } + assert!( + tokio::time::Instant::now() < deadline, + "ingestion stalled on an unreachable source without counting a single error" + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } +}