From 9a610ead79c2373953c1498fa92795b65ec62092 Mon Sep 17 00:00:00 2001 From: chrischeng-c4 Date: Sat, 24 Oct 2026 20:04:18 +0800 Subject: [PATCH 1/2] feat(raft-runtime): expose public RaftHost::status() accessor --- crates/raft-runtime/src/application.rs | 4 +- crates/raft-runtime/src/application/host.rs | 1 + .../src/application/host/membership_ops.rs | 6 + .../src/application/host/status.rs | 57 ++++++++ .../src/interfaces/peer_http/status.rs | 61 +------- crates/raft-runtime/tests/it/host_status.rs | 133 ++++++++++++++++++ crates/raft-runtime/tests/it/main.rs | 1 + 7 files changed, 202 insertions(+), 61 deletions(-) create mode 100644 crates/raft-runtime/tests/it/host_status.rs diff --git a/crates/raft-runtime/src/application.rs b/crates/raft-runtime/src/application.rs index ca9244cd..51b22bd8 100644 --- a/crates/raft-runtime/src/application.rs +++ b/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; diff --git a/crates/raft-runtime/src/application/host.rs b/crates/raft-runtime/src/application/host.rs index 221bec3b..67e9fc16 100644 --- a/crates/raft-runtime/src/application/host.rs +++ b/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}; diff --git a/crates/raft-runtime/src/application/host/membership_ops.rs b/crates/raft-runtime/src/application/host/membership_ops.rs index 5bb84ef7..99212937 100644 --- a/crates/raft-runtime/src/application/host/membership_ops.rs +++ b/crates/raft-runtime/src/application/host/membership_ops.rs @@ -29,6 +29,12 @@ impl RaftHost { pub async fn leader(&self) -> Option { 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) = { diff --git a/crates/raft-runtime/src/application/host/status.rs b/crates/raft-runtime/src/application/host/status.rs index b974a09d..1d57a788 100644 --- a/crates/raft-runtime/src/application/host/status.rs +++ b/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, + } +} diff --git a/crates/raft-runtime/src/interfaces/peer_http/status.rs b/crates/raft-runtime/src/interfaces/peer_http/status.rs index ffd95375..c9aca90a 100644 --- a/crates/raft-runtime/src/interfaces/peer_http/status.rs +++ b/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>) -> Json { Json(host_status(&s).await) diff --git a/crates/raft-runtime/tests/it/host_status.rs b/crates/raft-runtime/tests/it/host_status.rs new file mode 100644 index 00000000..c3916547 --- /dev/null +++ b/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( + 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(3).await; + let leader_idx = await_leader(&nodes) + .await + .expect("a three-voter cluster elects a leader"); + let target_idx = (leader_idx + 1) % 3; + 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); +} diff --git a/crates/raft-runtime/tests/it/main.rs b/crates/raft-runtime/tests/it/main.rs index c13f6c14..0568a89b 100644 --- a/crates/raft-runtime/tests/it/main.rs +++ b/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; From 9ab26c1b377791074a0ad3d513e5af94476c8219 Mon Sep 17 00:00:00 2001 From: chrischeng-c4 Date: Sat, 24 Oct 2026 20:25:01 +0800 Subject: [PATCH 2/2] test(raft-runtime): use 4-voter cluster in learner_status test --- crates/raft-runtime/tests/it/host_status.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/raft-runtime/tests/it/host_status.rs b/crates/raft-runtime/tests/it/host_status.rs index c3916547..97bdb31a 100644 --- a/crates/raft-runtime/tests/it/host_status.rs +++ b/crates/raft-runtime/tests/it/host_status.rs @@ -102,11 +102,11 @@ async fn follower_status() { #[tokio::test] async fn learner_status() { - let nodes = cluster(3).await; + let nodes = cluster(4).await; let leader_idx = await_leader(&nodes) .await - .expect("a three-voter cluster elects a leader"); - let target_idx = (leader_idx + 1) % 3; + .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]