feat(metrics): make ingest errors, fork signals and epoch parks countable

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::<UnexpectedHttpStatus>().is_some() {
+                return "http";
+            }
+            if let Some(reqwest_error) = cause.downcast_ref::<reqwest::Error>() {
+                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::<std::io::Error>().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)]

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

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 }

crates/data-source/src/lib.rs

@@ -1,4 +1,5 @@
 mod map;
+pub mod metrics;
 mod standard;
 mod types;
 

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<Family<Labels, Counter>> = 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<Family<Labels, Counter>> = 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();
+}

crates/data-source/src/standard.rs

@@ -77,11 +77,10 @@ impl<F> DataSourceState<F> {
                         }
                     }
                     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<C: DataClient> Endpoint<C> {
         }
     }
 
+    // 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 {

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<ReqwestDataClient>) -> 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,

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..]
         }

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

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<Family<EpochFailureLabels, Counter>> = 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::<UnapplicableFork>()) {
+        "unapplicable_fork"
+    } else {
+        "other"
+    };
+    DATASET_EPOCH_FAILURES
+        .get_or_create(&EpochFailureLabels {
+            dataset: DatasetValue(dataset_id),
+            reason
+        })
+        .inc();
+}
+
 static WRITE_DURATION: LazyLock<Family<WriteLabels, Histogram>> =
     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",

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