feat(app): wire the monitoring API into pluto run

crates/app/src/monitoringapi/checker.rs

@@ -1,6 +1,10 @@
 //! Background readiness checker for `/readyz`.
 
-use std::{collections::HashSet, time::Duration};
+use std::{
+    collections::HashSet,
+    sync::{Arc, Mutex},
+    time::Duration,
+};
 
 use chrono::{DateTime, Utc};
 use pluto_cluster::helpers;
@@ -45,14 +49,16 @@ struct ChainConfig {
 /// Starts the background readiness checker and returns the shared readiness
 /// state served by `/readyz`.
 ///
-/// `seen_pubkeys` should receive validator public keys observed through the
-/// validator API. `validator_api_calls` should receive one item for each
-/// validator API call. The checker consumes both receivers until cancellation.
+/// `seen_pubkeys` is a shared, deduped set of DV root pubkeys the validator
+/// client has referenced through the validator API; the checker drains it each
+/// slot, so it stays bounded by the validator count regardless of request
+/// volume. `validator_api_calls` should receive one item for each validator API
+/// call. The checker consumes both until cancellation.
 pub fn start_ready_checker(
     p2p_context: P2PContext,
     beacon_node: EthBeaconNodeApiClient,
     pubkeys: Vec<PubKey>,
-    seen_pubkeys: mpsc::Receiver<PubKey>,
+    seen_pubkeys: Arc<Mutex<HashSet<PubKey>>>,
     validator_api_calls: mpsc::Receiver<()>,
     ct: CancellationToken,
 ) -> ReadyState {
@@ -161,7 +167,7 @@ async fn run_ready_checker(
     p2p_context: P2PContext,
     beacon_node: EthBeaconNodeApiClient,
     pubkeys: Vec<PubKey>,
-    mut seen_pubkeys: mpsc::Receiver<PubKey>,
+    seen_pubkeys: Arc<Mutex<HashSet<PubKey>>>,
     mut validator_api_calls: mpsc::Receiver<()>,
     ct: CancellationToken,
     readiness: ReadyState,
@@ -184,7 +190,6 @@ async fn run_ready_checker(
     slot_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
     let mut peer_count_interval = tokio::time::interval(PEER_COUNT_PERIOD);
     peer_count_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
-    let mut seen_pubkeys_open = true;
     let mut validator_api_calls_open = true;
 
     slot_interval.tick().await;
@@ -206,6 +211,13 @@ async fn run_ready_checker(
                         None
                     }
                 };
+                // Fold in the pubkeys the VC referenced since the last tick.
+                // The set is deduped and drained every slot, so it stays
+                // bounded by the validator count regardless of request volume.
+                let observed = std::mem::take(&mut *seen_pubkeys.lock().expect("seen pubkeys mutex"));
+                for pubkey in observed {
+                    checker.observe_pubkey(pubkey);
+                }
                 let evaluated_epoch = current_epoch(&config, Utc::now());
                 let status = checker.evaluate_round(
                     quorum_peers_connected(&p2p_context),
@@ -218,12 +230,6 @@ async fn run_ready_checker(
                     Err(error) => readiness.set_error(error),
                 }
             }
-            pubkey = seen_pubkeys.recv(), if seen_pubkeys_open => {
-                match pubkey {
-                    Some(pubkey) => checker.observe_pubkey(pubkey),
-                    None => seen_pubkeys_open = false,
-                }
-            }
             call = validator_api_calls.recv(), if validator_api_calls_open => {
                 match call {
                     Some(()) => checker.observe_validator_api_call(),

crates/app/src/monitoringapi/router.rs

@@ -5,10 +5,11 @@ use std::sync::Arc;
 use axum::{
     Router,
     extract::State,
-    http::StatusCode,
+    http::{StatusCode, header::CONTENT_TYPE},
     response::{IntoResponse, Response},
     routing::get,
 };
+use vise::{Format, MetricsCollection};
 
 use super::readiness::{ReadinessCheck, ReadinessError};
 
@@ -41,14 +42,35 @@ pub fn router(checker: impl ReadinessCheck) -> Router {
     router_with_state(MonitoringState::new(checker))
 }
 
-/// Builds a monitoring API router from preconstructed state.
+/// Builds a monitoring API router from preconstructed state, serving Prometheus
+/// `/metrics` plus the `/livez` and `/readyz` probes.
 pub fn router_with_state(state: MonitoringState) -> Router {
     Router::new()
+        .route("/metrics", get(metrics))
         .route("/livez", get(livez))
         .route("/readyz", get(readyz))
         .with_state(state)
 }
 
+/// Serves the process metrics in the OpenMetrics-for-Prometheus text format —
+/// the same exposition the `vise-exporter` produces. Encodes the global `vise`
+/// registry on each scrape.
+async fn metrics() -> Response {
+    let registry = MetricsCollection::default().collect();
+    let mut buffer = String::new();
+    if let Err(error) = registry.encode(&mut buffer, Format::OpenMetricsForPrometheus) {
+        // Encoding the in-process registry should never fail; surface a 500
+        // rather than a partial body if it somehow does.
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            format!("failed to encode metrics: {error}"),
+        )
+            .into_response();
+    }
+
+    ([(CONTENT_TYPE, Format::OPEN_METRICS_CONTENT_TYPE)], buffer).into_response()
+}
+
 async fn livez() -> impl IntoResponse {
     (StatusCode::OK, "ok")
 }
@@ -124,6 +146,46 @@ mod tests {
         assert_eq!(body, "beacon node down");
     }
 
+    #[tokio::test]
+    async fn metrics_serves_prometheus_exposition() {
+        // Touch a monitoring gauge so the global `vise` registry is initialised
+        // and its series appears in the exposition.
+        crate::monitoringapi::MONITORING_METRICS
+            .monitoring_readyz
+            .set(1);
+
+        let request = Request::builder()
+            .uri("/metrics")
+            .body(Body::empty())
+            .expect("build request");
+        let response = router(ReadyState::new())
+            .oneshot(request)
+            .await
+            .expect("request");
+
+        assert_eq!(response.status(), StatusCode::OK);
+        let content_type = response
+            .headers()
+            .get(axum::http::header::CONTENT_TYPE)
+            .expect("content-type header")
+            .to_str()
+            .expect("utf8 content-type")
+            .to_owned();
+        assert!(
+            content_type.contains("openmetrics-text"),
+            "unexpected content-type: {content_type}"
+        );
+
+        let body = to_bytes(response.into_body(), BODY_LIMIT)
+            .await
+            .expect("read body");
+        let body = String::from_utf8(body.to_vec()).expect("utf8 body");
+        assert!(
+            body.contains("app_monitoring_readyz"),
+            "metrics body missing readyz gauge: {body}"
+        );
+    }
+
     #[tokio::test]
     async fn readyz_observes_readiness_state_updates() {
         let state = ReadyState::new();

crates/app/src/node/behaviour.rs

@@ -62,6 +62,9 @@ pub struct CoreHandles {
     pub parsigex: parsigex::Handle,
     /// Outbound QBFT broadcast handle.
     pub consensus: qbft::p2p::Handle,
+    /// Shared P2P runtime context (known peers + live connections), used by the
+    /// monitoring API's readiness checker to compute quorum connectivity.
+    pub p2p_context: P2PContext,
 }
 
 /// Composes the core behaviours and builds the libp2p [`Node`].
@@ -139,6 +142,10 @@ pub(crate) async fn wire_p2p(
     .with_peers(peer_ids.clone());
     let peerinfo_comp = peerinfo::Behaviour::new(local_peer_id, peerinfo_config);
 
+    // Clone the context before it is moved into the node so the readiness
+    // checker observes the same shared peer/connection state the swarm updates.
+    let p2p_context_for_handle = p2p_context.clone();
+
     let node = Node::new(
         p2p_config,
         key,
@@ -160,6 +167,7 @@ pub(crate) async fn wire_p2p(
     let handles = CoreHandles {
         parsigex: parsigex_handle,
         consensus: consensus_handle,
+        p2p_context: p2p_context_for_handle,
     };
 
     Ok((node, handles))

crates/app/src/node/config.rs

@@ -8,11 +8,11 @@ use pluto_p2p::config::P2PConfig;
 /// Application configuration for running a distributed-validator node.
 ///
 /// This is the Rust analog of Charon's `app.Config` (`app/app.go`), reduced to
-/// the minimal set required to wire and run the core duty workflow.
-/// Observability (monitoring/debug API, tracing/OTLP) and simnet/mock-only
+/// the minimal set required to wire and run the core duty workflow plus the
+/// monitoring API. Debug/pprof API, OTLP/Jaeger tracing and simnet/mock-only
 /// fields are intentionally omitted for the minimal-runnable wiring.
-// TODO(#402 part B): add monitoring/debug addrs, OTLP/Jaeger tracing config,
-// simnet (beacon/validator mock) and `TestConfig`-style overrides.
+// TODO(#402 part B): add debug/pprof addr, OTLP/Jaeger tracing config, simnet
+// (beacon/validator mock) and `TestConfig`-style overrides.
 #[derive(Debug, Clone)]
 pub struct AppConfig {
     /// P2P networking configuration (listen/advertise addresses, relays, ...).
@@ -42,6 +42,11 @@ pub struct AppConfig {
     /// Address the validator API HTTP server binds to.
     pub validator_api_addr: SocketAddr,
 
+    /// Address the monitoring API HTTP server binds to. Serves the Prometheus
+    /// `/metrics` scrape endpoint plus the `/livez` and `/readyz` health
+    /// probes.
+    pub monitoring_addr: SocketAddr,
+
     /// Whether the builder API (MEV-boost) is enabled.
     pub builder_api: bool,