feat(hotblocks): measure fork consensus duration

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<Family<Labels, Counter>> = LazyLock::n
 /// Fork signals by `source` and whether it held the contested position (`at_tip`/`above_tip`).
 pub static INGEST_FORK_SIGNALS: LazyLock<Family<Labels, Counter>> = LazyLock::new(Default::default);
 
+/// Time from the first fork signal until a fork decision, by decision path.
+pub static INGEST_FORK_CONSENSUS_DURATION: LazyLock<Family<Labels, Histogram>> =
+    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());
+}

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<F> {
     position: BlockStreamRequest,
     position_is_canonical: bool,
     max_seen_finalized_block: BlockNumber,
-    fork_consensus_timeout: Option<Pin<Box<Sleep>>>
+    fork_consensus_timeout: Option<Pin<Box<Sleep>>>,
+    fork_consensus_started_at: Option<Instant>
 }
 
 impl<F> DataSourceState<F> {
@@ -78,6 +79,7 @@ impl<F> DataSourceState<F> {
                     }
                     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<F> DataSourceState<F> {
         }
         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<F> DataSourceState<F> {
         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<BlockRef> {
-        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;

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 \

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",