diff --git a/Cargo.lock b/Cargo.lock
index e3fb3126..46d13452 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4867,6 +4867,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..53813111 100644
--- a/crates/data-client/src/reqwest/client.rs
+++ b/crates/data-client/src/reqwest/client.rs
@@ -226,6 +226,39 @@ 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"
+    }
+
+    fn source_label(&self) -> String {
+        self.url.host_str().unwrap_or("unknown").to_string()
+    }
 }
 
 #[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<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()
+    }
 }
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..c4cefdeb
--- /dev/null
+++ b/crates/data-source/src/metrics.rs
@@ -0,0 +1,15 @@
+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` host and `kind`.
+/// Registered in the hotblocks metrics registry.
+pub static INGEST_SOURCE_ERRORS: LazyLock<Family<Labels, Counter>> = LazyLock::new(Default::default);
+
+pub(crate) 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();
+}
diff --git a/crates/data-source/src/standard.rs b/crates/data-source/src/standard.rs
index f896c2dd..26e8d1c2 100644
--- a/crates/data-source/src/standard.rs
+++ b/crates/data-source/src/standard.rs
@@ -221,6 +221,8 @@ impl<C: DataClient> Endpoint<C> {
     }
 
     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/metrics.rs b/crates/hotblocks/src/metrics.rs
index d5e08c0b..cc388251 100644
--- a/crates/hotblocks/src/metrics.rs
+++ b/crates/hotblocks/src/metrics.rs
@@ -184,6 +184,12 @@ pub fn build_metrics_registry() -> Registry {
         QUERY_ERROR_TOO_MANY_DATA_WAITERS.clone()
     );
 
+    registry.register(
+        "ingest_source_errors",
+        "Upstream data source ingestion errors, by source host and kind (connect/timeout/http/io/...)",
+        sqd_data_source::metrics::INGEST_SOURCE_ERRORS.clone()
+    );
+
     registry.register("http_status", "Number of sent HTTP responses", HTTP_STATUS.clone());
     registry.register(
         "http_seconds_to_first_byte",