feat(observability): validator health metrics + readiness probe

crates/consensus/primary-metrics/src/consensus.rs

@@ -53,6 +53,20 @@ 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` | `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,
+    /// 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 +140,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
+            )?,
         })
     }
 }

crates/consensus/primary/src/certifier.rs

@@ -403,6 +403,14 @@ impl<DB: Database> Certifier<DB> {
                             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!(
@@ -414,6 +422,11 @@ impl<DB: Database> Certifier<DB> {
                     );
                 }
                 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,
@@ -448,10 +461,22 @@ impl<DB: Database> Certifier<DB> {
                         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;
                 }
 
                 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,

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

crates/middleware/orchestrator/src/epoch_manager/core.rs

@@ -97,14 +97,22 @@ where
         let mut node_task_manager = TaskManager::new(NODE_TASK_MANAGER);
         let node_task_spawner = node_task_manager.get_spawner();
 
-        // Bind the liveness healthcheck before any boot-time cold work (crash reconcile + backlog
+        // Bind the healthcheck before any boot-time cold work (crash reconcile + backlog
         // migration). Both are synchronous and can be multi-minute on a large DB; binding the probe
         // first keeps it answering throughout (the migration is `spawn_blocking`'d so the runtime
         // stays free), so a short-deadline probe cannot blackout and restart-loop the node during
-        // recovery. NOTE: the probe is liveness-only by design (it does not report readiness), so
-        // it is green while these run.
+        // recovery. LIVENESS (any path but `/readyz`) stays green throughout — the process is up.
+        // READINESS (`/readyz`) reads the live node mode and reports `503` until the node is voting
+        // or observing, so it is correctly not-ready during this boot recovery. A bind failure
+        // (e.g. the port is already in use) is propagated rather than silently swallowed — the
+        // operator asked for this endpoint.
         if let Some(port) = self.builder.healthcheck {
-            let _ = HealthcheckServer::spawn(node_task_manager.get_spawner(), port).await;
+            HealthcheckServer::spawn(
+                node_task_manager.get_spawner(),
+                port,
+                self.consensus_bus.node_mode().subscribe(),
+            )
+            .await?;
         }
 
         // Heal any crash-interrupted archive before serving, while consensus and execution have not
@@ -495,9 +503,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.