feat(app): beacon node SSE listener

Cargo.toml

@@ -46,6 +46,7 @@ crossbeam = "0.8.4"
 dyn-clone = "1.0"
 dyn-eq = "0.1.3"
 either = "1.13"
+eventsource-stream = "0.2"
 futures = "0.3"
 futures-timer = "3.0"
 backon = "1.6.0"

crates/app/Cargo.toml

@@ -9,10 +9,12 @@ publish.workspace = true
 [dependencies]
 backon.workspace = true
 chrono.workspace = true
+futures.workspace = true
 pluto-core.workspace = true
 pluto-eth2api.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true
+vise.workspace = true
 prost.workspace = true
 prost-types.workspace = true
 regex.workspace = true
@@ -24,6 +26,7 @@ serde.workspace = true
 serde_json.workspace = true
 hex.workspace = true
 k256.workspace = true
+serde_with.workspace = true
 bon.workspace = true
 flate2.workspace = true
 tar.workspace = true

crates/app/src/lib.rs

@@ -22,5 +22,8 @@ pub mod eth2wrap;
 /// Private key locking service.
 pub mod privkeylock;
 
+/// Listen for SSE from Beacon Node
+pub mod sse;
+
 /// Utility helpers for archiving, extracting, and comparing files/directories.
 pub mod utils;

crates/app/src/sse/metrics.rs

@@ -0,0 +1,45 @@
+//! Prometheus metrics for the beacon node SSE listener.
+//!
+//! The `app_beacon_node` prefix and bucket boundaries reproduce Charon's
+//! `app/beacon_node` SSE metrics. All metrics are labelled by beacon node
+//! address.
+
+use vise::{Gauge, Global, Histogram, LabeledFamily, Metrics};
+
+/// Head delay buckets in seconds.
+const HEAD_DELAY_BUCKETS: [f64; 6] = [2.0, 4.0, 6.0, 8.0, 10.0, 12.0];
+/// Chain reorg depth buckets in slots.
+const REORG_DEPTH_BUCKETS: [f64; 6] = [1.0, 2.0, 4.0, 6.0, 8.0, 16.0];
+/// Block reception delay buckets in seconds.
+const BLOCK_BUCKETS: [f64; 14] = [
+    0.5, 1.0, 1.5, 2.0, 2.5, 3.0, 3.5, 4.0, 4.5, 5.0, 6.0, 8.0, 10.0, 12.0,
+];
+
+/// Metrics for the beacon node SSE listener.
+#[derive(Debug, Metrics)]
+#[metrics(prefix = "app_beacon_node")]
+pub struct SseMetrics {
+    /// Current beacon node head slot, supplied by the SSE endpoint.
+    #[metrics(labels = ["addr"])]
+    pub sse_head_slot: LabeledFamily<String, Gauge<u64>>,
+
+    /// Delay in seconds between slot start and head update.
+    #[metrics(buckets = &HEAD_DELAY_BUCKETS, labels = ["addr"])]
+    pub sse_head_delay: LabeledFamily<String, Histogram>,
+
+    /// Chain reorg depth in slots.
+    #[metrics(buckets = &REORG_DEPTH_BUCKETS, labels = ["addr"])]
+    pub sse_chain_reorg_depth: LabeledFamily<String, Histogram>,
+
+    /// Block reception via gossip delay in seconds.
+    #[metrics(buckets = &BLOCK_BUCKETS, labels = ["addr"])]
+    pub sse_block_gossip: LabeledFamily<String, Histogram>,
+
+    /// Block imported into fork choice delay in seconds.
+    #[metrics(buckets = &BLOCK_BUCKETS, labels = ["addr"])]
+    pub sse_block: LabeledFamily<String, Histogram>,
+}
+
+/// Global metrics for the beacon node SSE listener.
+#[vise::register]
+pub static SSE_METRICS: Global<SseMetrics> = Global::new();