raft-runtime: public RaftHost::status() accessor

crates/raft-runtime/src/application.rs

@@ -36,7 +36,7 @@ pub use state_machine::{
 pub use topology::{peer_ordinal, ClusterDims, ClusterTopology};
 
 pub(crate) use host::{
-    apply_ready, cold_start, decode_backpressure, persist_node, take_reply, HostStorage, HostStore,
-    PeerLaneQueue, PeerWiring, Shared,
+    apply_ready, cold_start, decode_backpressure, host_status, persist_node, take_reply,
+    HostStorage, HostStore, PeerLaneQueue, PeerWiring, Shared,
 };
 pub(crate) use topology::check_peer_scheme;

crates/raft-runtime/src/application/host.rs

@@ -61,6 +61,7 @@ pub(crate) use apply::{apply_ready, cold_start, persist_node};
 pub(crate) use backpressure::decode_backpressure;
 pub(crate) use peer_lane::PeerLaneQueue;
 pub(crate) use spawn::{HostStore, PeerWiring};
+pub(crate) use status::host_status;
 
 use apply::tick_then_maybe_persist;
 use backpressure::{encode_backpressure, rejected_admission};

crates/raft-runtime/src/application/host/membership_ops.rs

@@ -29,6 +29,12 @@ impl RaftHost {
     pub async fn leader(&self) -> Option<NodeId> {
         self.shared.node.lock().await.leader()
     }
+    /// Return the host's current [`RaftStatus`].
+    ///
+    /// Returns the same value as the `/raftz` body and is a point-in-time snapshot.
+    pub async fn status(&self) -> RaftStatus {
+        host_status(&self.shared).await
+    }
     /// Hand leadership to an eligible caught-up voter before shutdown (#3664).
     pub async fn handoff_leadership(&self) -> LeadershipHandoff {
         let (outcome, transferred) = {

crates/raft-runtime/src/application/host/status.rs

@@ -38,3 +38,60 @@ pub struct RaftStatus {
     #[serde(default)]
     pub max_resident_log_bytes: u64,
 }
+
+pub(crate) async fn host_status(s: &Shared) -> RaftStatus {
+    let n = s.node.lock().await;
+    let durability_error = s
+        .latched_failure
+        .lock()
+        .unwrap()
+        .as_ref()
+        .map(|e| e.to_string());
+    let conf = n.conf_state();
+    let (committed_voters, incoming_voters, membership_phase) = match &conf.outgoing {
+        Some(outgoing) => (
+            outgoing.clone(),
+            Some(conf.membership.voters().to_vec()),
+            MembershipPhase::Joint,
+        ),
+        None => (
+            conf.membership.voters().to_vec(),
+            None,
+            MembershipPhase::Stable,
+        ),
+    };
+    let learners = conf.membership.learners().to_vec();
+    let role = if !n.is_voter() {
+        "Learner".to_string()
+    } else {
+        format!("{:?}", n.role())
+    };
+    RaftStatus {
+        group_id: s.group_id.as_str().to_owned(),
+        id: s.id,
+        role,
+        term: n.current_term().get(),
+        commit_index: n.commit_index().get(),
+        last_index: n.last_index().get(),
+        snapshot_index: n.snapshot_index().get(),
+        applied_index: s.completed_applied_index().get(),
+        leader: n.leader(),
+        is_leader: n.is_leader(),
+        durability_error,
+        committed_voters,
+        incoming_voters,
+        learners,
+        membership_phase,
+        undeliverable_never_addressed: s.undeliverable_never_addressed.load(Ordering::Relaxed),
+        undeliverable_withdrawn_address: s.undeliverable_withdrawn_address(),
+        proposal_rejected_before_routing: s
+            .proposal_rejected_before_routing
+            .load(Ordering::Relaxed),
+        proposal_rejected_before_append: s.proposal_rejected_before_append.load(Ordering::Relaxed),
+        proposal_admission_closed: s.lifecycle_generation.load(Ordering::Acquire) > 0,
+        lifecycle_generation: s.lifecycle_generation.load(Ordering::Acquire),
+        snapshot_capability: s.sm.snapshot_capability().map(str::to_string),
+        resident_log_bytes: n.resident_log_bytes() as u64,
+        max_resident_log_bytes: s.max_resident_log_bytes as u64,
+    }
+}

crates/raft-runtime/src/interfaces/peer_http/status.rs

@@ -1,67 +1,10 @@
-use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
 use axum::extract::State;
 use axum::Json;
 
-use crate::application::{MembershipPhase, RaftStatus, Shared};
-
-pub(crate) async fn host_status(s: &Shared) -> RaftStatus {
-    let n = s.node.lock().await;
-    let durability_error = s
-        .latched_failure
-        .lock()
-        .unwrap()
-        .as_ref()
-        .map(|e| e.to_string());
-    let conf = n.conf_state();
-    let (committed_voters, incoming_voters, membership_phase) = match &conf.outgoing {
-        Some(outgoing) => (
-            outgoing.clone(),
-            Some(conf.membership.voters().to_vec()),
-            MembershipPhase::Joint,
-        ),
-        None => (
-            conf.membership.voters().to_vec(),
-            None,
-            MembershipPhase::Stable,
-        ),
-    };
-    let learners = conf.membership.learners().to_vec();
-    let role = if !n.is_voter() {
-        "Learner".to_string()
-    } else {
-        format!("{:?}", n.role())
-    };
-    RaftStatus {
-        group_id: s.group_id.as_str().to_owned(),
-        id: s.id,
-        role,
-        term: n.current_term().get(),
-        commit_index: n.commit_index().get(),
-        last_index: n.last_index().get(),
-        snapshot_index: n.snapshot_index().get(),
-        applied_index: s.completed_applied_index().get(),
-        leader: n.leader(),
-        is_leader: n.is_leader(),
-        durability_error,
-        committed_voters,
-        incoming_voters,
-        learners,
-        membership_phase,
-        undeliverable_never_addressed: s.undeliverable_never_addressed.load(Ordering::Relaxed),
-        undeliverable_withdrawn_address: s.undeliverable_withdrawn_address(),
-        proposal_rejected_before_routing: s
-            .proposal_rejected_before_routing
-            .load(Ordering::Relaxed),
-        proposal_rejected_before_append: s.proposal_rejected_before_append.load(Ordering::Relaxed),
-        proposal_admission_closed: s.lifecycle_generation.load(Ordering::Acquire) > 0,
-        lifecycle_generation: s.lifecycle_generation.load(Ordering::Acquire),
-        snapshot_capability: s.sm.snapshot_capability().map(str::to_string),
-        resident_log_bytes: n.resident_log_bytes() as u64,
-        max_resident_log_bytes: s.max_resident_log_bytes as u64,
-    }
-}
+pub(crate) use crate::application::host_status;
+use crate::application::{RaftStatus, Shared};
 
 pub(crate) async fn raftz(State(s): State<Arc<Shared>>) -> Json<RaftStatus> {
     Json(host_status(&s).await)

crates/raft-runtime/tests/it/host_status.rs

@@ -0,0 +1,133 @@
+//! Integration tests for `RaftHost::status()` and its equivalence to `/raftz`.
+
+use std::time::{Duration, Instant};
+
+use raft_runtime::{NodeId, RaftStatus};
+
+use crate::support::cluster;
+use cluster::{await_leader, cluster};
+
+fn h2c_client() -> reqwest::Client {
+    reqwest::Client::builder()
+        .http2_prior_knowledge()
+        .build()
+        .unwrap()
+}
+
+async fn raftz_status(client: &reqwest::Client, url: &str) -> RaftStatus {
+    client
+        .get(format!("{url}/raftz"))
+        .send()
+        .await
+        .expect("a host serves its own status")
+        .json()
+        .await
+        .expect("the status is the published shape")
+}
+
+async fn poll_status_until<F>(
+    client: &reqwest::Client,
+    url: &str,
+    predicate: F,
+    timeout: Duration,
+    desc: &str,
+) -> RaftStatus
+where
+    F: Fn(&RaftStatus) -> bool,
+{
+    let deadline = Instant::now() + timeout;
+    loop {
+        let s = raftz_status(client, url).await;
+        if predicate(&s) {
+            return s;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "timed out waiting for condition: {desc} at {url}"
+        );
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+}
+
+fn assert_stable_fields_equal(a: &RaftStatus, b: &RaftStatus) {
+    assert_eq!(a.group_id, b.group_id);
+    assert_eq!(a.id, b.id);
+    assert_eq!(a.role, b.role);
+    assert_eq!(a.is_leader, b.is_leader);
+    assert_eq!(a.leader, b.leader);
+    assert_eq!(a.committed_voters, b.committed_voters);
+    assert_eq!(a.incoming_voters, b.incoming_voters);
+    assert_eq!(a.learners, b.learners);
+    assert_eq!(a.membership_phase, b.membership_phase);
+}
+
+#[tokio::test]
+async fn leader_status() {
+    let nodes = cluster(3).await;
+    let leader_idx = await_leader(&nodes)
+        .await
+        .expect("a three-voter cluster elects a leader");
+    let leader = &nodes[leader_idx];
+    let own_id = NodeId::new(leader_idx as u64);
+
+    let status = leader.host.status().await;
+    assert_eq!(status.role, "Leader");
+    assert!(status.is_leader);
+    assert_eq!(status.leader, Some(own_id));
+
+    let client = h2c_client();
+    let http_status = raftz_status(&client, &leader.url).await;
+    assert_stable_fields_equal(&http_status, &status);
+}
+
+#[tokio::test]
+async fn follower_status() {
+    let nodes = cluster(3).await;
+    let leader_idx = await_leader(&nodes)
+        .await
+        .expect("a three-voter cluster elects a leader");
+    let leader_id = NodeId::new(leader_idx as u64);
+    let follower_idx = (leader_idx + 1) % 3;
+    let follower = &nodes[follower_idx];
+
+    let status = follower.host.status().await;
+    assert_eq!(status.role, "Follower");
+    assert!(!status.is_leader);
+    assert_eq!(status.leader, Some(leader_id));
+
+    let client = h2c_client();
+    let http_status = raftz_status(&client, &follower.url).await;
+    assert_stable_fields_equal(&http_status, &status);
+}
+
+#[tokio::test]
+async fn learner_status() {
+    let nodes = cluster(4).await;
+    let leader_idx = await_leader(&nodes)
+        .await
+        .expect("a four-voter cluster elects a leader");
+    let target_idx = (leader_idx + 1) % 4;
+    let target_id = NodeId::new(target_idx as u64);
+
+    nodes[leader_idx]
+        .host
+        .demote_voter(target_id)
+        .await
+        .expect("demoting a voter succeeds");
+
+    let client = h2c_client();
+    let demoted_http_status = poll_status_until(
+        &client,
+        &nodes[target_idx].url,
+        |s| s.role == "Learner" && s.learners.contains(&target_id),
+        Duration::from_secs(5),
+        "demoted node reports Learner role and target in learners",
+    )
+    .await;
+
+    let status = nodes[target_idx].host.status().await;
+    assert_eq!(status.role, "Learner");
+    assert!(status.learners.contains(&target_id));
+
+    assert_stable_fields_equal(&demoted_http_status, &status);
+}

crates/raft-runtime/tests/it/main.rs

@@ -11,6 +11,7 @@ mod group_identity;
 mod group_membership_isolation;
 mod group_registry;
 mod host_shutdown_deadline;
+mod host_status;
 mod leadership_transfer;
 mod legacy_store_migration;
 mod member_admission;