diff --git a/crates/data-source/src/metrics.rs b/crates/data-source/src/metrics.rs index a17f6ecd..d303f78a 100644 --- a/crates/data-source/src/metrics.rs +++ b/crates/data-source/src/metrics.rs @@ -1,6 +1,10 @@ -use std::sync::LazyLock; +use std::{sync::LazyLock, time::Duration}; -use prometheus_client::metrics::{counter::Counter, family::Family}; +use prometheus_client::metrics::{ + counter::Counter, + family::Family, + histogram::{exponential_buckets, Histogram} +}; type Labels = Vec<(&'static str, String)>; @@ -11,6 +15,10 @@ pub static INGEST_SOURCE_ERRORS: LazyLock> = LazyLock::n /// 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); +/// Time from the first fork signal until a fork decision, by decision path. +pub static INGEST_FORK_CONSENSUS_DURATION: LazyLock> = + LazyLock::new(|| Family::new_with_constructor(|| Histogram::new(exponential_buckets(0.001, 2.0, 15)))); + /// 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) { @@ -27,3 +35,9 @@ pub(crate) fn record_ingest_fork_signal(source: &str, standing: &'static str) { ]) .inc(); } + +pub(crate) fn record_ingest_fork_consensus_duration(decision: &'static str, duration: Duration) { + INGEST_FORK_CONSENSUS_DURATION + .get_or_create(&vec![("decision", decision.to_string())]) + .observe(duration.as_secs_f64()); +} diff --git a/crates/data-source/src/standard.rs b/crates/data-source/src/standard.rs index d1361323..2eb01dc1 100644 --- a/crates/data-source/src/standard.rs +++ b/crates/data-source/src/standard.rs @@ -4,7 +4,7 @@ use anyhow::Context; use futures::{future::BoxFuture, stream::BoxStream, FutureExt, Stream, StreamExt}; use sqd_data_client::{BlockStreamRequest, BlockStreamResponse, DataClient}; use sqd_primitives::{Block, BlockNumber, BlockRef}; -use tokio::time::Sleep; +use tokio::time::{Instant, Sleep}; use tracing::{info, warn}; use crate::types::{DataEvent, DataSource}; @@ -44,7 +44,8 @@ struct DataSourceState { position: BlockStreamRequest, position_is_canonical: bool, max_seen_finalized_block: BlockNumber, - fork_consensus_timeout: Option>> + fork_consensus_timeout: Option>>, + fork_consensus_started_at: Option } impl DataSourceState { @@ -78,6 +79,7 @@ impl DataSourceState { } Poll::Ready(Ok(BlockStreamResponse::Fork(prev_blocks))) => { let req = req.clone(); + self.fork_consensus_started_at.get_or_insert_with(Instant::now); ep.on_fork_signal(req.first_block, &prev_blocks); ep.error_counter = 0; ep.state = EndpointState::Fork { req, prev_blocks }; @@ -143,7 +145,7 @@ impl DataSourceState { } self.position.first_block = block.number() + 1; self.position_is_canonical = true; - self.fork_consensus_timeout = None; + self.reset_fork_consensus(); if is_final { set_head(&mut self.finalized_head, block.number(), block.hash()); @@ -152,6 +154,11 @@ impl DataSourceState { true } + fn reset_fork_consensus(&mut self) { + self.fork_consensus_timeout = None; + self.fork_consensus_started_at = None; + } + fn on_new_finalized_head(&mut self, new_head: Option<&BlockRef>) -> bool { let Some(new_head) = new_head else { return false }; @@ -285,7 +292,8 @@ where }, position_is_canonical: false, max_seen_finalized_block: 0, - fork_consensus_timeout: None + fork_consensus_timeout: None, + fork_consensus_started_at: None }; Self { endpoints, state } @@ -302,21 +310,38 @@ where let forks = self.endpoints.iter().filter(|ep| ep.is_on_fork()).count(); if forks > 0 { let active = self.endpoints.iter().filter(|ep| ep.is_active()).count(); - if forks > self.endpoints.len() / 2 || forks == active || self.fork_consensus_timeout(cx) { - let chain = self.extract_fork(); - info!( - forked_endpoints = forks, - active_endpoints = active, - total_endpoints = self.endpoints.len(), - hint_count = chain.len(), - oldest_hint =? chain.first().map(|b| b.number), - newest_hint =? chain.last().map(|b| b.number), - "fork consensus reached" - ); - return Poll::Ready(DataEvent::Fork(chain)); - } + let decision = if forks > self.endpoints.len() / 2 { + "majority" + } else if forks == active { + "all_active" + } else if self.fork_consensus_timeout(cx) { + "timeout" + } else { + return Poll::Pending; + }; + + let consensus_duration = self + .state + .fork_consensus_started_at + .expect("fork consensus must start with the first fork signal") + .elapsed(); + crate::metrics::record_ingest_fork_consensus_duration(decision, consensus_duration); + + let chain = self.extract_fork(); + info!( + decision = decision, + consensus_duration_seconds = consensus_duration.as_secs_f64(), + forked_endpoints = forks, + active_endpoints = active, + total_endpoints = self.endpoints.len(), + hint_count = chain.len(), + oldest_hint =? chain.first().map(|b| b.number), + newest_hint =? chain.last().map(|b| b.number), + "fork consensus reached" + ); + return Poll::Ready(DataEvent::Fork(chain)); } else { - self.state.fork_consensus_timeout = None + self.state.reset_fork_consensus() } Poll::Pending @@ -338,7 +363,7 @@ where } fn extract_fork(&mut self) -> Vec { - self.state.fork_consensus_timeout = None; + self.state.reset_fork_consensus(); let mut chain = Vec::new(); for ep in self.endpoints.iter_mut() { match std::mem::replace(&mut ep.state, EndpointState::Ready) { @@ -381,6 +406,7 @@ where self.state.position.set_parent_block_hash(parent_block_hash); self.state.position_is_canonical = false; self.state.finalized_head = None; + self.state.reset_fork_consensus(); for ep in self.endpoints.iter_mut() { ep.state = EndpointState::Ready; ep.last_committed_block = None; diff --git a/crates/hotblocks/src/metrics.rs b/crates/hotblocks/src/metrics.rs index 1460fc0b..1bf06e55 100644 --- a/crates/hotblocks/src/metrics.rs +++ b/crates/hotblocks/src/metrics.rs @@ -538,6 +538,13 @@ pub fn build_metrics_registry() -> Registry { sqd_data_source::metrics::INGEST_FORK_SIGNALS.clone() ); + registry.register( + "ingest_fork_consensus_duration_seconds", + "Time from the first upstream fork signal until fork consensus, by decision path \ + (majority/all_active/timeout)", + sqd_data_source::metrics::INGEST_FORK_CONSENSUS_DURATION.clone() + ); + registry.register( "dataset_epoch_failures", "Dataset update task failures, by dataset and cause; each one parks ingestion for \ diff --git a/crates/hotblocks/tests/ct4_lagging_source.rs b/crates/hotblocks/tests/ct4_lagging_source.rs index 81428a23..191b0af1 100644 --- a/crates/hotblocks/tests/ct4_lagging_source.rs +++ b/crates/hotblocks/tests/ct4_lagging_source.rs @@ -208,18 +208,13 @@ async fn ct4_a_fork_signal_above_the_tip_and_the_park_it_causes_are_observable() 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); - } + h.peers[0].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; - } + h.produce_lagging(&[0], 10)?; + h.finalize_with_lag(5)?; // No `settle` — the epoch is serving out `P-EPOCH-RETRY`, which is the thing being measured. - tokio::time::sleep(Duration::from_secs(1)).await; + tokio::time::sleep(Duration::from_secs(3)).await; let metrics = h.client.metrics().await?; let above_tip = metrics @@ -234,6 +229,28 @@ async fn ct4_a_fork_signal_above_the_tip_and_the_park_it_causes_are_observable() "no source held the contested position" ); + let consensus_count = metrics + .get( + "hotblocks_ingest_fork_consensus_duration_seconds_count", + Some(("decision", "timeout")) + ) + .unwrap_or_default(); + assert!( + consensus_count > 0.0, + "the lone fork signal must reach consensus by timeout" + ); + + let consensus_seconds = metrics + .get( + "hotblocks_ingest_fork_consensus_duration_seconds_sum", + Some(("decision", "timeout")) + ) + .unwrap_or_default(); + assert!( + consensus_seconds >= 2.0, + "the timeout path must spend at least 2 s in consensus, observed {consensus_seconds} s" + ); + let parked = metrics .get( "hotblocks_dataset_epoch_failures_total",