From bc75f42ba558fdb26a9362af5dca6be22570aca3 Mon Sep 17 00:00:00 2001 From: Luis Soares Date: Thu, 6 Aug 2026 13:26:36 +0100 Subject: [PATCH 1/4] feat(observability): validator health metrics + readiness probe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2 of the Network Observability Initiative (axyl-private#428). Adds the validator/ network-health signals the audit found missing — the node already exports Narwhal/ Mysten's Prometheus suite, but the highest-value per-validator and readiness signals were tracked in-memory and never exposed. Readiness probe: the healthcheck now serves /readyz — 200 only when the node is voting (CvvActive) or an operational observer, 503 while a validator is still catching up (CvvInactive). Liveness (any other path) stays 200. Reads the live NodeMode via a watch::Receiver from the consensus bus, threaded at the spawn site. Lets a load balancer / on-call distinguish "process up" from "actually voting in the current epoch". New metrics (registered into the consensus default_registry, served on --metrics): - validator_participation{authority} — committed certificates per validator, counted at the bullshark commit site. A direct per-validator liveness signal, consistent across nodes (every node commits the same sub-dags). This is the audit's biggest gap ("did validator X vote/certify?"). - vote_request_rejections{authority, reason} — this node's header vote requests rejected by a peer (from VoteFailureTracker), by rejecting validator and reason (too_old | epoch_mismatch). Node-local: rising totals mean this node is falling behind. - committee_size — committee size for the current epoch, set per epoch. Dashboard: etc/monitoring/.../rayls-consensus.json — a starter Grafana dashboard (epoch, committee size, bad nodes, round progress, commit-latency p95, leader election, the two new per-validator series, peers) with auto-provisioning, portable via a datasource variable. Composes with the etc/monitoring stack from axyl#99. Refs: raylsnetwork/axyl-private#428 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../primary-metrics/src/consensus.rs | 27 +++ crates/consensus/primary/src/certifier.rs | 10 + .../primary/src/consensus/bullshark.rs | 13 ++ .../orchestrator/src/epoch_manager/core.rs | 16 +- .../orchestrator/src/types/health.rs | 198 ++++++++++-------- .../provisioning/dashboards/dashboards.yml | 13 ++ .../dashboards/rayls-consensus.json | 149 +++++++++++++ 7 files changed, 338 insertions(+), 88 deletions(-) create mode 100644 etc/monitoring/grafana/provisioning/dashboards/dashboards.yml create mode 100644 etc/monitoring/grafana/provisioning/dashboards/rayls-consensus.json diff --git a/crates/consensus/primary-metrics/src/consensus.rs b/crates/consensus/primary-metrics/src/consensus.rs index 94f35687..b50f1b54 100644 --- a/crates/consensus/primary-metrics/src/consensus.rs +++ b/crates/consensus/primary-metrics/src/consensus.rs @@ -53,6 +53,16 @@ pub struct ConsensusMetrics { pub leader_commits: IntCounterVec, /// number of bad nodes in the committee pub num_of_bad_nodes: IntGauge, + /// This node's header vote requests rejected by a peer, labeled by the rejecting validator + /// (`authority`) and `reason` (`too_old` | `epoch_mismatch`). Node-local: high totals mean + /// this node is falling behind and being rejected by the committee. + pub vote_request_rejections: IntCounterVec, + /// Size of the committee for the current epoch. + pub committee_size: IntGauge, + /// Committed certificates per validator (by `authority`) — a direct per-validator liveness + /// signal: which validators' certificates actually land in committed sub-dags. Consistent + /// across nodes (every node commits the same sub-dags), unlike node-local rejection counts. + pub validator_participation: IntCounterVec, } impl ConsensusMetrics { @@ -126,6 +136,23 @@ impl ConsensusMetrics { "The number of bad nodes in the new leader schedule", registry )?, + vote_request_rejections: register_int_counter_vec_with_registry!( + "vote_request_rejections", + "This node's header vote requests rejected by a peer, by rejecting validator (authority) and reason", + &["authority", "reason"], + registry + )?, + committee_size: register_int_gauge_with_registry!( + "committee_size", + "Size of the committee for the current epoch", + registry + )?, + validator_participation: register_int_counter_vec_with_registry!( + "validator_participation", + "Committed certificates per validator (by authority) - a per-validator liveness signal", + &["authority"], + registry + )?, }) } } diff --git a/crates/consensus/primary/src/certifier.rs b/crates/consensus/primary/src/certifier.rs index 9aee1c42..a943cdb9 100644 --- a/crates/consensus/primary/src/certifier.rs +++ b/crates/consensus/primary/src/certifier.rs @@ -414,6 +414,11 @@ impl Certifier { ); } let outcome = self.vote_failures.record_too_old(peer_id.clone()); + self.consensus_bus + .consensus_metrics() + .vote_request_rejections + .with_label_values(&[&peer_id.to_string(), "too_old"]) + .inc(); warn!( target: "primary::certifier", auth=?self.authority_id, @@ -452,6 +457,11 @@ impl Certifier { } let outcome = self.vote_failures.record_epoch_mismatch(peer_id.clone()); + self.consensus_bus + .consensus_metrics() + .vote_request_rejections + .with_label_values(&[&peer_id.to_string(), "epoch_mismatch"]) + .inc(); warn!( target: "primary::certifier", auth=?self.authority_id, diff --git a/crates/consensus/primary/src/consensus/bullshark.rs b/crates/consensus/primary/src/consensus/bullshark.rs index f3a1d4ff..6417cb87 100644 --- a/crates/consensus/primary/src/consensus/bullshark.rs +++ b/crates/consensus/primary/src/consensus/bullshark.rs @@ -188,6 +188,19 @@ impl Bullshark { self.metrics.committed_certificates.report(total_committed_certificates); + // Per-validator participation: count each committed certificate by its author. Distinct + // from the leader-only series (`leader_election` / `leader_commits`), this shows which + // validators are actually contributing certificates each commit — a direct liveness + // signal, and consistent across nodes since every node commits the same sub-dags. + for sub_dag in &committed_sub_dags { + for certificate in &sub_dag.certificates { + self.metrics + .validator_participation + .with_label_values(&[&certificate.origin().to_string()]) + .inc(); + } + } + Ok((Outcome::Commit, committed_sub_dags)) } diff --git a/crates/middleware/orchestrator/src/epoch_manager/core.rs b/crates/middleware/orchestrator/src/epoch_manager/core.rs index 96b35871..54a39549 100644 --- a/crates/middleware/orchestrator/src/epoch_manager/core.rs +++ b/crates/middleware/orchestrator/src/epoch_manager/core.rs @@ -270,9 +270,14 @@ where info!(target: "epoch-manager", tasks=?node_task_manager, "NODE TASKS\n"); - // spawn node healthcheck service if enabled + // spawn node healthcheck + readiness service if enabled if let Some(port) = self.builder.healthcheck { - let _ = HealthcheckServer::spawn(node_task_manager.get_spawner(), port).await; + let _ = HealthcheckServer::spawn( + node_task_manager.get_spawner(), + port, + self.consensus_bus.node_mode().subscribe(), + ) + .await; } // Catch the termination signal ourselves so we can drive a graceful, ORDERED @@ -471,9 +476,12 @@ where // This needs to be created early so required machinery for other tasks exists when needed. let mut worker = worker_node.new_worker().await?; worker.set_batch_tracker(self.consensus_bus.batch_tracker().clone()); - let current_epoch = primary.current_committee().await.epoch(); + let current_committee = primary.current_committee().await; + let current_epoch = current_committee.epoch(); - self.consensus_bus.consensus_metrics().current_epoch.set(current_epoch as i64); + let consensus_metrics = self.consensus_bus.consensus_metrics(); + consensus_metrics.current_epoch.set(current_epoch as i64); + consensus_metrics.committee_size.set(current_committee.size() as i64); // Produce a "dummy" epoch 0 EpochRecord if missing. // This will let us use simple code to find any epoch including 0 at startup. diff --git a/crates/middleware/orchestrator/src/types/health.rs b/crates/middleware/orchestrator/src/types/health.rs index 54c517c2..181a0e42 100644 --- a/crates/middleware/orchestrator/src/types/health.rs +++ b/crates/middleware/orchestrator/src/types/health.rs @@ -1,52 +1,79 @@ -//! Simple TCP healthcheck endpoint for monitoring service availability. +//! TCP health + readiness endpoint for monitoring service availability. //! -//! Implements a minimal HTTP/1.1 server that responds with status 200 to all requests. -//! This is designed for integration with GCP load balancers and similar health monitoring systems. +//! A minimal HTTP/1.1 server exposing two probes: +//! - **liveness** (any path, e.g. `GET /`): always `200 OK` while the process is accepting +//! connections — the pre-existing behaviour. +//! - **readiness** (`GET /readyz`): `200 OK` only when the node is actually participating — +//! voting (`CvvActive`) or an operational observer — and `503 Service Unavailable` while a +//! validator is still catching up (`CvvInactive`). This lets a load balancer / on-call +//! distinguish "process up" from "actually voting in the current epoch". +//! +//! Designed for integration with GCP load balancers and similar health monitoring systems. use std::{io::ErrorKind, net::SocketAddr, time::Duration}; +use rayls_consensus_primary::NodeMode; use rayls_infrastructure_types::TaskSpawner; -use tokio::{io::AsyncWriteExt, net::TcpListener, time::sleep}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, + sync::watch, + time::sleep, +}; use tracing::info; -/// Minimal HTTP health check responder for service monitoring. +/// Liveness / readiness HTTP responder for service monitoring. /// -/// Binds to a TCP port and responds with HTTP 200 to any connection. -/// Uses raw TCP sockets for minimal overhead and dependencies. +/// Binds a TCP port and answers: +/// - liveness (any path): `200 OK` — the process is up. +/// - readiness (`/readyz`): `200` when voting/observing, `503` while catching up. /// /// # Security Considerations /// -/// This endpoint accepts connections from any source and responds unconditionally. -/// -/// Node operators must ensure the endpoint is protected by a firewall. -/// This service is off by default, but can be enabled through the CLI node command. -/// Each connection is handled synchronously in the main accept loop. -/// No connection limits or rate limiting are implemented. -/// Connections are immediately closed after sending response. +/// This endpoint accepts connections from any source. Node operators must protect it with a +/// firewall. It is off by default and enabled via the CLI node command. Each connection is +/// handled in the accept loop and closed immediately after the response. /// -/// To enable on node startup, use `rayls-network node --enable-healthcheck`. +/// To enable on node startup, use `rayls-network node --healthcheck `. /// See `rayls-network-cli::node` for more info. #[derive(Debug)] pub(crate) struct HealthcheckServer; +const LIVE_200: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK"; +const READY_200: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nREADY"; +const NOT_READY_503: &[u8] = + b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 9\r\n\r\nNOT_READY"; + +/// True if the HTTP request line targets the readiness path (`/readyz` or `/ready`). +fn request_targets_readiness(req: &[u8]) -> bool { + // Request line looks like `GET /readyz HTTP/1.1`; match the path token, ignoring the + // method and version. + let line = req.split(|&b| b == b'\r' || b == b'\n').next().unwrap_or(req); + let mut tokens = line.split(|&b| b == b' ').filter(|t| !t.is_empty()); + let _method = tokens.next(); + matches!(tokens.next(), Some(path) if path == b"/readyz" || path == b"/ready") +} + +/// Readiness response for a node mode: ready (`200`) when voting (`CvvActive`) or an operational +/// observer; not ready (`503`) while a validator is still catching up (`CvvInactive`). +fn readiness_response(mode: NodeMode) -> &'static [u8] { + if mode.is_active_cvv() || mode.is_observer() { + READY_200 + } else { + NOT_READY_503 + } +} + impl HealthcheckServer { - /// Spawns the health check server task and returns the bound address. - /// - /// Binds to port specified by `HEALTHCHECK_PORT` environment variable, - /// or lets the OS assign a port if unset or set to 0. - /// - /// # Network Binding - /// - /// Binds to 0.0.0.0 (all interfaces) to allow external health checkers. - /// This makes the service accessible on all network interfaces including public IPs. + /// Spawn the health/readiness server, returning the bound address. /// - /// # Protocol - /// - /// Implements minimal HTTP/1.1 with a fixed response: - /// - Status: 200 OK - /// - Body: "OK" (2 bytes) - /// - No request parsing or validation - /// - No custom headers to avoid information disclosure - pub(crate) async fn spawn(task_spawner: TaskSpawner, port: u16) -> eyre::Result { + /// Binds `0.0.0.0:port` (all interfaces, for external health checkers). `node_mode` is a + /// watch receiver of the node's consensus mode (from the consensus bus); the readiness probe + /// reads its current value per request. + pub(crate) async fn spawn( + task_spawner: TaskSpawner, + port: u16, + node_mode: watch::Receiver, + ) -> eyre::Result { // IMPORTANT: use firewall to protect this endpoint let addr: SocketAddr = ([0, 0, 0, 0], port).into(); let listener = TcpListener::bind(addr).await?; @@ -54,17 +81,29 @@ impl HealthcheckServer { info!(target: "epoch-manager", ?listen_on, "healthcheck listening"); task_spawner.spawn_task("healthcheck", async move { - // minimal valid HTTP - let response = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK"; - let mut backoff = Duration::from_millis(100); let max_backoff = Duration::from_secs(5); loop { match listener.accept().await { Ok((mut socket, _)) => { - // write response, ignore errors (client disconnect, etc.) - // then drop connection + // Read the request line to route by path. Bounded read with a short + // timeout so a slow/partial client can't wedge the accept loop. + let mut buf = [0u8; 256]; + let wants_readiness = matches!( + tokio::time::timeout(Duration::from_millis(200), socket.read(&mut buf)) + .await, + Ok(Ok(n)) if n > 0 && request_targets_readiness(&buf[..n]) + ); + + let response = if wants_readiness { + readiness_response(*node_mode.borrow()) + } else { + // Any other path (incl. `/`) is a liveness probe: the process is up. + LIVE_200 + }; + + // write response, ignore errors (client disconnect, etc.), then drop. if let Err(e) = socket.write_all(response).await { tracing::error!(target: "healthcheck", ?e, "error writing healthcheck response"); } @@ -76,15 +115,14 @@ impl HealthcheckServer { | ErrorKind::ConnectionAborted | ErrorKind::ConnectionReset | ErrorKind::Other - )=> { + ) => { // transient errors that can be ignored tracing::warn!(target: "healthcheck", ?e, "transient error accepting healthcheck connection"); sleep(backoff).await; backoff = (backoff * 2).min(max_backoff); - } Err(e) => { - // unexpected errors should be logged and break the loop to avoid spinning on fatal errors + // unexpected errors: log and break to avoid spinning on fatal errors tracing::error!(target: "healthcheck", ?e, "error accepting healthcheck connection"); break; } @@ -98,60 +136,52 @@ impl HealthcheckServer { #[cfg(test)] mod tests { + use super::*; use rayls_infrastructure_types::{get_available_tcp_port, TaskManager}; - use std::time::Duration; - use tokio::{ - io::{AsyncReadExt, AsyncWriteExt}, - net::TcpStream, - }; + use tokio::net::TcpStream; + + async fn request(addr: SocketAddr, path: &str) -> String { + let mut stream = TcpStream::connect(addr).await.unwrap(); + stream.write_all(format!("GET {path} HTTP/1.1\r\n\r\n").as_bytes()).await.unwrap(); + let mut response = vec![0u8; 1024]; + let n = stream.read(&mut response).await.unwrap(); + String::from_utf8_lossy(&response[..n]).into_owned() + } + + #[test] + fn readiness_path_is_matched() { + assert!(request_targets_readiness(b"GET /readyz HTTP/1.1\r\n\r\n")); + assert!(request_targets_readiness(b"GET /ready HTTP/1.1\r\n\r\n")); + assert!(!request_targets_readiness(b"GET / HTTP/1.1\r\n\r\n")); + assert!(!request_targets_readiness(b"GET /healthz HTTP/1.1\r\n\r\n")); + } - use crate::types::HealthcheckServer; + #[test] + fn readiness_response_tracks_mode() { + assert_eq!(readiness_response(NodeMode::CvvActive), READY_200); + assert_eq!(readiness_response(NodeMode::Observer), READY_200); + assert_eq!(readiness_response(NodeMode::CvvInactive), NOT_READY_503); + } #[tokio::test] - async fn test_tcp_healthcheck() -> eyre::Result<()> { + async fn liveness_always_ok_and_readiness_follows_mode() -> eyre::Result<()> { let task_manager = TaskManager::default(); let task_spawner = task_manager.get_spawner(); + let (tx, rx) = watch::channel(NodeMode::CvvInactive); let port = get_available_tcp_port("127.0.0.1").expect("tcp port assigned by host"); - // spawn server and get the bound address - let addr = HealthcheckServer::spawn(task_spawner.clone(), port).await?; - - // give server time to start listening + let addr = HealthcheckServer::spawn(task_spawner, port, rx).await?; tokio::time::sleep(Duration::from_millis(10)).await; - tokio::time::timeout(Duration::from_millis(500), async move { - // request healthcheck - let mut stream = TcpStream::connect(addr).await?; - - // send minimal HTTP request - stream.write_all(b"GET / HTTP/1.1\r\n\r\n").await?; - - // read response - let mut response = vec![0u8; 1024]; - let n = stream.read(&mut response).await?; - response.truncate(n); - let response_str = String::from_utf8_lossy(&response); - - // verify http status line - assert!( - response_str.starts_with("HTTP/1.1 200 OK"), - "Expected 200 OK, got: {}", - response_str - ); - - // verify body - assert!(response_str.ends_with("OK"), "Expected body 'OK', got: {}", response_str); - - // verify content-length header - assert!( - response_str.contains("Content-Length: 2"), - "Missing or incorrect Content-Length header" - ); - - Ok::<(), eyre::Error>(()) - }) - .await - .expect("response received")?; + // Liveness: always 200, even while catching up. + assert!(request(addr, "/").await.starts_with("HTTP/1.1 200 OK")); + + // Readiness: 503 while CvvInactive (catching up)... + assert!(request(addr, "/readyz").await.starts_with("HTTP/1.1 503")); + + // ...and 200 once the node is voting. + tx.send(NodeMode::CvvActive).unwrap(); + assert!(request(addr, "/readyz").await.starts_with("HTTP/1.1 200 OK")); Ok(()) } diff --git a/etc/monitoring/grafana/provisioning/dashboards/dashboards.yml b/etc/monitoring/grafana/provisioning/dashboards/dashboards.yml new file mode 100644 index 00000000..8c95269a --- /dev/null +++ b/etc/monitoring/grafana/provisioning/dashboards/dashboards.yml @@ -0,0 +1,13 @@ +# Grafana dashboard provisioning — auto-loads the JSON dashboards in this directory on start. +apiVersion: 1 + +providers: + - name: rayls + orgId: 1 + type: file + disableDeletion: false + updateIntervalSeconds: 30 + allowUiUpdates: true + options: + path: /etc/grafana/provisioning/dashboards + foldersFromFilesStructure: false diff --git a/etc/monitoring/grafana/provisioning/dashboards/rayls-consensus.json b/etc/monitoring/grafana/provisioning/dashboards/rayls-consensus.json new file mode 100644 index 00000000..73e20538 --- /dev/null +++ b/etc/monitoring/grafana/provisioning/dashboards/rayls-consensus.json @@ -0,0 +1,149 @@ +{ + "annotations": { "list": [] }, + "editable": true, + "graphTooltip": 1, + "title": "Rayls — Consensus & Validator Health", + "uid": "rayls-consensus", + "tags": ["rayls", "consensus"], + "timezone": "browser", + "schemaVersion": 39, + "version": 1, + "time": { "from": "now-1h", "to": "now" }, + "refresh": "30s", + "templating": { + "list": [ + { + "name": "DS", + "label": "Prometheus", + "type": "datasource", + "query": "prometheus", + "current": {}, + "hide": 0, + "refresh": 1 + } + ] + }, + "panels": [ + { + "id": 1, + "title": "Current epoch", + "type": "stat", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 4, "w": 6, "x": 0, "y": 0 }, + "targets": [{ "expr": "current_epoch", "refId": "A" }] + }, + { + "id": 2, + "title": "Committee size", + "type": "stat", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 4, "w": 6, "x": 6, "y": 0 }, + "targets": [{ "expr": "committee_size", "refId": "A" }] + }, + { + "id": 3, + "title": "Bad nodes (leader schedule)", + "type": "stat", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 4, "w": 6, "x": 12, "y": 0 }, + "fieldConfig": { + "defaults": { + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "red", "value": 1 } + ] + } + }, + "overrides": [] + }, + "targets": [{ "expr": "num_of_bad_nodes", "refId": "A" }] + }, + { + "id": 4, + "title": "Consensus rounds (proposer vs. committed)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 4 }, + "description": "Gap between current_round and last_committed_round is the commit lag.", + "targets": [ + { "expr": "current_round", "legendFormat": "proposer round", "refId": "A" }, + { "expr": "last_committed_round", "legendFormat": "committed round", "refId": "B" } + ] + }, + { + "id": 5, + "title": "Commit round latency (p95)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 4 }, + "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }, + "targets": [ + { + "expr": "histogram_quantile(0.95, sum(rate(consensus_commit_rounds_latency_bucket[5m])) by (le))", + "legendFormat": "p95", + "refId": "A" + } + ] + }, + { + "id": 6, + "title": "Leader election outcomes (rate)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 12 }, + "targets": [ + { + "expr": "sum by (outcome) (rate(leader_election[5m]))", + "legendFormat": "{{outcome}}", + "refId": "A" + } + ] + }, + { + "id": 7, + "title": "Vote-request rejections by validator (rate)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 12 }, + "description": "This node's header vote requests rejected by a peer. Rising totals mean this node is falling behind.", + "targets": [ + { + "expr": "sum by (authority, reason) (rate(vote_request_rejections[5m]))", + "legendFormat": "{{authority}} — {{reason}}", + "refId": "A" + } + ] + }, + { + "id": 8, + "title": "Connected peers", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 20 }, + "targets": [ + { + "expr": "connected_peers_count", + "legendFormat": "{{kad_type}}", + "refId": "A" + } + ] + }, + { + "id": 9, + "title": "Validator participation (committed certs/s)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 20 }, + "description": "Rate of committed certificates per validator — a direct per-validator liveness signal, consistent across nodes. A validator whose line drops to zero has stopped contributing certificates.", + "targets": [ + { + "expr": "sum by (authority) (rate(validator_participation[5m]))", + "legendFormat": "{{authority}}", + "refId": "A" + } + ] + } + ] +} From 65bef89594d5ac12daad5b813f44a6dd0e1d8291 Mon Sep 17 00:00:00 2001 From: Luis Soares Date: Thu, 6 Aug 2026 13:43:20 +0100 Subject: [PATCH 2/4] fix(observability): address PR review on the readiness probe + rejection metric MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the P2 review on #102: - Propagate a healthcheck bind failure (core.rs) instead of `let _ =`-swallowing it. A port already in use would previously start the node with no health endpoint and no signal — defeating the purpose of an explicitly-requested probe. Now returns the error. - Handle each healthcheck connection in its own task (health.rs). The accept loop bounded each request read with a 200ms timeout but still ran it inline, so a client that connects and sends nothing held the loop for the full timeout and delayed every other probe. A per-connection task keeps the accept loop free; the watch::Receiver is cloned per conn. - Reset the accept-loop backoff on a healthy accept (health.rs) so a past transient error doesn't keep the retry delay elevated. - Count cert-covered-skipped too-old rejections under a distinct `too_old_skipped` reason (certifier.rs) rather than dropping them from vote_request_rejections. Otherwise the metric reads zero while a node continuously trips the skip path, masking real trouble. Doc comment on the metric updated to list the three reason values. - Document the `/ready` alias alongside `/readyz` in the health module docs. Reviewer's other two points verified as non-issues, no change: the `authority` label is a ~44-char bs58 of a 32-byte AuthorityIdentifier with cardinality bounded by committee size (not a raw BLS key); and `current_round` (dashboard panel 4) is a real registered PrimaryMetrics gauge on the same default_registry. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../primary-metrics/src/consensus.rs | 7 ++- crates/consensus/primary/src/certifier.rs | 8 +++ .../orchestrator/src/epoch_manager/core.rs | 8 ++- .../orchestrator/src/types/health.rs | 62 ++++++++++++------- 4 files changed, 56 insertions(+), 29 deletions(-) diff --git a/crates/consensus/primary-metrics/src/consensus.rs b/crates/consensus/primary-metrics/src/consensus.rs index b50f1b54..ea8c642a 100644 --- a/crates/consensus/primary-metrics/src/consensus.rs +++ b/crates/consensus/primary-metrics/src/consensus.rs @@ -54,8 +54,11 @@ pub struct ConsensusMetrics { /// number of bad nodes in the committee pub num_of_bad_nodes: IntGauge, /// This node's header vote requests rejected by a peer, labeled by the rejecting validator - /// (`authority`) and `reason` (`too_old` | `epoch_mismatch`). Node-local: high totals mean - /// this node is falling behind and being rejected by the committee. + /// (`authority`) and `reason` (`too_old` | `too_old_skipped` | `epoch_mismatch`). Node-local: + /// high totals mean this node is falling behind and being rejected by the committee. + /// `too_old_skipped` is a too-old rejection that was not counted toward peer demotion (the + /// cert store already covered the limit round and the DAG was still progressing) — tracked + /// separately so the raw rejection rate stays visible. pub vote_request_rejections: IntCounterVec, /// Size of the committee for the current epoch. pub committee_size: IntGauge, diff --git a/crates/consensus/primary/src/certifier.rs b/crates/consensus/primary/src/certifier.rs index a943cdb9..19197820 100644 --- a/crates/consensus/primary/src/certifier.rs +++ b/crates/consensus/primary/src/certifier.rs @@ -403,6 +403,14 @@ impl Certifier { header_round, limit_round, cert_store_round, committed_round, "ignoring too-old rejection: cert store covers limit round and DAG still progressing (transient proposer lag)" ); + // Count the rejection under a distinct reason even though it isn't held + // against the peer for demotion — otherwise the metric would show zero + // while a node continuously trips the skip path, masking real trouble. + self.consensus_bus + .consensus_metrics() + .vote_request_rejections + .with_label_values(&[&peer_id.to_string(), "too_old_skipped"]) + .inc(); return VoteErrorAction::Continue; } warn!( diff --git a/crates/middleware/orchestrator/src/epoch_manager/core.rs b/crates/middleware/orchestrator/src/epoch_manager/core.rs index 54a39549..b6b39276 100644 --- a/crates/middleware/orchestrator/src/epoch_manager/core.rs +++ b/crates/middleware/orchestrator/src/epoch_manager/core.rs @@ -270,14 +270,16 @@ where info!(target: "epoch-manager", tasks=?node_task_manager, "NODE TASKS\n"); - // spawn node healthcheck + readiness service if enabled + // spawn node healthcheck + readiness service if enabled. Propagate a bind failure (e.g. + // the port is already in use) rather than silently starting the node without the endpoint + // an operator explicitly asked for. if let Some(port) = self.builder.healthcheck { - let _ = HealthcheckServer::spawn( + HealthcheckServer::spawn( node_task_manager.get_spawner(), port, self.consensus_bus.node_mode().subscribe(), ) - .await; + .await?; } // Catch the termination signal ourselves so we can drive a graceful, ORDERED diff --git a/crates/middleware/orchestrator/src/types/health.rs b/crates/middleware/orchestrator/src/types/health.rs index 181a0e42..f8275b12 100644 --- a/crates/middleware/orchestrator/src/types/health.rs +++ b/crates/middleware/orchestrator/src/types/health.rs @@ -3,10 +3,10 @@ //! A minimal HTTP/1.1 server exposing two probes: //! - **liveness** (any path, e.g. `GET /`): always `200 OK` while the process is accepting //! connections — the pre-existing behaviour. -//! - **readiness** (`GET /readyz`): `200 OK` only when the node is actually participating — -//! voting (`CvvActive`) or an operational observer — and `503 Service Unavailable` while a -//! validator is still catching up (`CvvInactive`). This lets a load balancer / on-call -//! distinguish "process up" from "actually voting in the current epoch". +//! - **readiness** (`GET /readyz`, alias `GET /ready`): `200 OK` only when the node is actually +//! participating — voting (`CvvActive`) or an operational observer — and `503 Service +//! Unavailable` while a validator is still catching up (`CvvInactive`). This lets a load +//! balancer / on-call distinguish "process up" from "actually voting in the current epoch". //! //! Designed for integration with GCP load balancers and similar health monitoring systems. use std::{io::ErrorKind, net::SocketAddr, time::Duration}; @@ -81,32 +81,46 @@ impl HealthcheckServer { info!(target: "epoch-manager", ?listen_on, "healthcheck listening"); task_spawner.spawn_task("healthcheck", async move { - let mut backoff = Duration::from_millis(100); + let initial_backoff = Duration::from_millis(100); + let mut backoff = initial_backoff; let max_backoff = Duration::from_secs(5); loop { match listener.accept().await { Ok((mut socket, _)) => { - // Read the request line to route by path. Bounded read with a short - // timeout so a slow/partial client can't wedge the accept loop. - let mut buf = [0u8; 256]; - let wants_readiness = matches!( - tokio::time::timeout(Duration::from_millis(200), socket.read(&mut buf)) + // Reset backoff after a healthy accept so a past transient error doesn't + // keep the delay elevated. + backoff = initial_backoff; + + // Handle each connection in its own task: the request read is bounded by a + // short timeout, but a client that connects and sends nothing would still + // hold the accept loop for the full timeout, delaying every other probe. + // A per-connection task keeps the accept loop free. `node_mode` is a cheap + // watch receiver, cloned per connection. + let node_mode = node_mode.clone(); + tokio::spawn(async move { + let mut buf = [0u8; 256]; + let wants_readiness = matches!( + tokio::time::timeout( + Duration::from_millis(200), + socket.read(&mut buf), + ) .await, - Ok(Ok(n)) if n > 0 && request_targets_readiness(&buf[..n]) - ); - - let response = if wants_readiness { - readiness_response(*node_mode.borrow()) - } else { - // Any other path (incl. `/`) is a liveness probe: the process is up. - LIVE_200 - }; - - // write response, ignore errors (client disconnect, etc.), then drop. - if let Err(e) = socket.write_all(response).await { - tracing::error!(target: "healthcheck", ?e, "error writing healthcheck response"); - } + Ok(Ok(n)) if n > 0 && request_targets_readiness(&buf[..n]) + ); + + let response = if wants_readiness { + readiness_response(*node_mode.borrow()) + } else { + // Any other path (incl. `/`) is a liveness probe: process is up. + LIVE_200 + }; + + // write response, ignore errors (client disconnect, etc.), then drop. + if let Err(e) = socket.write_all(response).await { + tracing::error!(target: "healthcheck", ?e, "error writing healthcheck response"); + } + }); } Err(ref e) if matches!( e.kind(), From 087c49953f61cecf2461ff09a368dddf70215e81 Mon Sep 17 00:00:00 2001 From: Luis Soares Date: Thu, 6 Aug 2026 13:48:21 +0100 Subject: [PATCH 3/4] fix(observability): count stale epoch-mismatch rejections too (symmetric with too_old_skipped) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The re-review noted the epoch-mismatch stale-peer early-return dropped the rejection from vote_request_rejections — the same gap just fixed for too_old. Count it under a distinct `epoch_mismatch_stale` reason so the raw rejection rate stays visible across all four paths. Metric doc updated to list all reason values. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/consensus/primary-metrics/src/consensus.rs | 11 ++++++----- crates/consensus/primary/src/certifier.rs | 7 +++++++ 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/crates/consensus/primary-metrics/src/consensus.rs b/crates/consensus/primary-metrics/src/consensus.rs index ea8c642a..edaecb3d 100644 --- a/crates/consensus/primary-metrics/src/consensus.rs +++ b/crates/consensus/primary-metrics/src/consensus.rs @@ -54,11 +54,12 @@ pub struct ConsensusMetrics { /// number of bad nodes in the committee pub num_of_bad_nodes: IntGauge, /// This node's header vote requests rejected by a peer, labeled by the rejecting validator - /// (`authority`) and `reason` (`too_old` | `too_old_skipped` | `epoch_mismatch`). Node-local: - /// high totals mean this node is falling behind and being rejected by the committee. - /// `too_old_skipped` is a too-old rejection that was not counted toward peer demotion (the - /// cert store already covered the limit round and the DAG was still progressing) — tracked - /// separately so the raw rejection rate stays visible. + /// (`authority`) and `reason` (`too_old` | `too_old_skipped` | `epoch_mismatch` | + /// `epoch_mismatch_stale`). Node-local: high totals mean this node is falling behind and being + /// rejected by the committee. The `*_skipped` / `*_stale` variants are rejections that were + /// *not* counted toward peer demotion (a too-old rejection the cert store already covered + /// while the DAG progressed, or an epoch rejection from a peer that is itself stale) — tracked + /// under distinct reasons so the raw rejection rate stays visible. pub vote_request_rejections: IntCounterVec, /// Size of the committee for the current epoch. pub committee_size: IntGauge, diff --git a/crates/consensus/primary/src/certifier.rs b/crates/consensus/primary/src/certifier.rs index 19197820..4995583c 100644 --- a/crates/consensus/primary/src/certifier.rs +++ b/crates/consensus/primary/src/certifier.rs @@ -461,6 +461,13 @@ impl Certifier { peer_epoch, our_epoch, "ignoring epoch rejection from stale peer" ); + // Count it under a distinct reason even though a stale peer can't demote us — + // mirrors `too_old_skipped`, keeping the raw rejection rate visible. + self.consensus_bus + .consensus_metrics() + .vote_request_rejections + .with_label_values(&[&peer_id.to_string(), "epoch_mismatch_stale"]) + .inc(); return VoteErrorAction::Continue; } From 6b55dd52e5bfdf7eac9b4163da3b8fc40e47d4cb Mon Sep 17 00:00:00 2001 From: Luis Soares Date: Thu, 6 Aug 2026 14:13:15 +0100 Subject: [PATCH 4/4] style(observability): wrap health.rs doc comment to nightly rustfmt comment_width CI's Format job runs `cargo +nightly fmt --all --check`, whose rustfmt.toml sets comment_width=100 + wrap_comments (nightly-only). One doc-comment line in the readiness module wrapped one word early; rewrap it so the nightly check passes. No code change. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/middleware/orchestrator/src/types/health.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/middleware/orchestrator/src/types/health.rs b/crates/middleware/orchestrator/src/types/health.rs index f8275b12..554b9336 100644 --- a/crates/middleware/orchestrator/src/types/health.rs +++ b/crates/middleware/orchestrator/src/types/health.rs @@ -5,8 +5,8 @@ //! connections — the pre-existing behaviour. //! - **readiness** (`GET /readyz`, alias `GET /ready`): `200 OK` only when the node is actually //! participating — voting (`CvvActive`) or an operational observer — and `503 Service -//! Unavailable` while a validator is still catching up (`CvvInactive`). This lets a load -//! balancer / on-call distinguish "process up" from "actually voting in the current epoch". +//! Unavailable` while a validator is still catching up (`CvvInactive`). This lets a load balancer +//! / on-call distinguish "process up" from "actually voting in the current epoch". //! //! Designed for integration with GCP load balancers and similar health monitoring systems. use std::{io::ErrorKind, net::SocketAddr, time::Duration};