bmux_cluster_plugin 0.0.1-alpha.4

//! `OpenRaft` control-plane runtime wiring.
//!
//! This module owns node construction and lifecycle. Network implementations
//! remain injectable so production authenticated endpoint RPC and deterministic
//! in-memory tests exercise the same Raft/storage integration.

use crate::consensus_network::EndpointRaftNetworkFactory;
use crate::consensus_storage::{
    ConsensusLogStore, ConsensusStateMachine, ConsensusStorageError, ControlRaftConfig,
    ControlRequest,
};
use crate::control_state::ControlState;
use crate::membership::NodeId;
use bmux_cluster_plugin_api::cluster_types::{
    ControlReadConsistency, ControlResponse, ControlServiceError, ControlStateView,
};
use openraft::error::{CheckIsLeaderError, Fatal, ForwardToLeader, RaftError};
use openraft::network::RaftNetworkFactory;
use openraft::raft::ClientWriteResponse;
use openraft::{BasicNode, Config, SnapshotPolicy};
use std::collections::BTreeMap;
use std::path::Path;
use std::sync::Arc;

const SNAPSHOT_LOG_THRESHOLD: u64 = 1_024;

#[derive(Debug)]
pub enum ConsensusRuntimeError {
    Storage(ConsensusStorageError),
    Configuration(String),
    Fatal(Box<Fatal<NodeId>>),
}

impl std::fmt::Display for ConsensusRuntimeError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Storage(error) => write!(formatter, "consensus storage failed: {error}"),
            Self::Configuration(error) => {
                write!(
                    formatter,
                    "consensus runtime configuration is invalid: {error}"
                )
            }
            Self::Fatal(error) => write!(formatter, "consensus runtime failed: {error}"),
        }
    }
}

impl std::error::Error for ConsensusRuntimeError {}

impl From<ConsensusStorageError> for ConsensusRuntimeError {
    fn from(error: ConsensusStorageError) -> Self {
        Self::Storage(error)
    }
}

impl From<Fatal<NodeId>> for ConsensusRuntimeError {
    fn from(error: Fatal<NodeId>) -> Self {
        Self::Fatal(Box::new(error))
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NotLeader {
    pub leader_id: Option<NodeId>,
    pub leader_endpoint: Option<String>,
}

impl From<&ForwardToLeader<NodeId, BasicNode>> for NotLeader {
    fn from(forward: &ForwardToLeader<NodeId, BasicNode>) -> Self {
        Self {
            leader_id: forward.leader_id,
            leader_endpoint: forward.leader_node.as_ref().map(|node| node.addr.clone()),
        }
    }
}

#[derive(Debug)]
pub enum ConsensusWriteError {
    NotLeader(NotLeader),
    QuorumUnavailable(String),
    Fatal(Box<Fatal<NodeId>>),
    Membership(String),
}

impl std::fmt::Display for ConsensusWriteError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::NotLeader(forward) => match (forward.leader_id, &forward.leader_endpoint) {
                (Some(id), Some(endpoint)) => {
                    write!(formatter, "not leader; forward to {id} at {endpoint}")
                }
                (Some(id), None) => write!(formatter, "not leader; forward to {id}"),
                (None, _) => formatter.write_str("not leader; current leader is unknown"),
            },
            Self::QuorumUnavailable(reason) => {
                write!(formatter, "control-plane quorum is unavailable: {reason}")
            }
            Self::Fatal(error) => write!(formatter, "consensus runtime failed: {error}"),
            Self::Membership(error) => formatter.write_str(error),
        }
    }
}

impl std::error::Error for ConsensusWriteError {}

#[derive(Debug)]
pub enum ConsensusReadError {
    NotLeader(NotLeader),
    QuorumUnavailable(String),
    Fatal(Box<Fatal<NodeId>>),
    Storage(ConsensusStorageError),
}

impl std::fmt::Display for ConsensusReadError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::NotLeader(forward) => match (forward.leader_id, &forward.leader_endpoint) {
                (Some(id), Some(endpoint)) => {
                    write!(formatter, "not leader; forward to {id} at {endpoint}")
                }
                (Some(id), None) => write!(formatter, "not leader; forward to {id}"),
                (None, _) => formatter.write_str("not leader; current leader is unknown"),
            },
            Self::QuorumUnavailable(error) => {
                write!(formatter, "linearizable read unavailable: {error}")
            }
            Self::Fatal(error) => write!(formatter, "consensus runtime failed: {error}"),
            Self::Storage(error) => write!(formatter, "consensus storage failed: {error}"),
        }
    }
}

impl std::error::Error for ConsensusReadError {}

/// One running `OpenRaft` node backed by cluster-owned durable storage.
#[derive(Clone)]
pub struct ConsensusNode {
    raft: openraft::Raft<ControlRaftConfig>,
    storage: ConsensusLogStore,
    membership_submission: std::sync::Arc<tokio::sync::Mutex<()>>,
    decoder_admission: Option<Arc<dyn crate::consensus_network::DecoderAdmission>>,
    node_id: NodeId,
    cluster_id: String,
}

fn validate_admission_term(expected: u64, current: u64) -> Result<(), ConsensusWriteError> {
    if expected != current {
        return Err(ConsensusWriteError::Membership(
            "leadership term changed during admission; retry compatibility checks".into(),
        ));
    }
    Ok(())
}

impl ConsensusNode {
    /// Starts one Raft node over the durable storage for `cluster_id`.
    ///
    /// The supplied network factory determines transport only. All consensus,
    /// snapshot, compaction, and persistence behavior is identical across the
    /// production endpoint transport and deterministic test substitutions.
    /// # Errors
    ///
    /// Returns storage, configuration, or fatal Raft startup failures.
    pub async fn start<N>(
        state_dir: &Path,
        cluster_id: &str,
        node_id: NodeId,
        network: N,
    ) -> Result<Self, ConsensusRuntimeError>
    where
        N: RaftNetworkFactory<ControlRaftConfig>,
    {
        let storage = ConsensusLogStore::open(state_dir, cluster_id)?;
        let state_machine = storage.state_machine()?;
        let raft = openraft::Raft::new(
            node_id,
            consensus_config(cluster_id)?,
            network,
            storage.clone(),
            state_machine,
        )
        .await?;
        Ok(Self {
            raft,
            storage,
            membership_submission: std::sync::Arc::new(tokio::sync::Mutex::new(())),
            decoder_admission: None,
            node_id,
            cluster_id: cluster_id.to_string(),
        })
    }

    /// Starts one production node using the authenticated connections-backed
    /// peer transport.
    ///
    /// # Errors
    ///
    /// Returns storage, configuration, or fatal Raft startup failures.
    pub async fn start_endpoint<C>(
        state_dir: &Path,
        cluster_id: &str,
        identity: crate::membership::NodeIdentity,
        caller: Arc<C>,
    ) -> Result<Self, ConsensusRuntimeError>
    where
        C: bmux_plugin::ServiceCaller + Send + Sync + 'static,
    {
        let network = EndpointRaftNetworkFactory::new(caller.clone(), identity.clone());
        let mut node = Self::start(
            state_dir,
            cluster_id,
            *identity.node_id(),
            EndpointRaftNetworkFactory::new(caller, identity),
        )
        .await?;
        node.decoder_admission = Some(Arc::new(network));
        Ok(node)
    }

    #[must_use]
    pub const fn node_id(&self) -> NodeId {
        self.node_id
    }

    #[must_use]
    pub fn cluster_id(&self) -> &str {
        &self.cluster_id
    }

    #[must_use]
    pub fn current_term(&self) -> u64 {
        self.raft.metrics().borrow().current_term
    }

    #[must_use]
    pub fn committed_member_ids(&self) -> std::collections::BTreeSet<NodeId> {
        self.raft
            .metrics()
            .borrow()
            .membership_config
            .membership()
            .nodes()
            .map(|(node_id, _)| *node_id)
            .collect()
    }

    #[must_use]
    pub fn committed_voter_ids(&self) -> std::collections::BTreeSet<NodeId> {
        self.raft
            .metrics()
            .borrow()
            .membership_config
            .membership()
            .voter_ids()
            .collect()
    }

    #[must_use]
    pub const fn raft(&self) -> &openraft::Raft<ControlRaftConfig> {
        &self.raft
    }

    /// Initializes a pristine node as a single-voter cluster.
    ///
    /// # Errors
    ///
    /// Returns a Raft initialization error when durable state is non-pristine
    /// or the node cannot commit the initial membership.
    pub async fn initialize_single(
        &self,
        node_id: NodeId,
        endpoint: impl Into<String>,
    ) -> Result<(), RaftError<NodeId, openraft::error::InitializeError<NodeId, BasicNode>>> {
        self.raft
            .initialize(BTreeMap::from([(node_id, BasicNode::new(endpoint.into()))]))
            .await
    }

    /// Executes learner-first catch-up followed by `OpenRaft`'s joint and uniform
    /// voter transition. Every requested voter must have an authenticated
    /// endpoint address.
    ///
    /// The transition intentionally retains removed servers as learners so
    /// leave/revoke can first commit the safe voter-set change and then commit
    /// the replicated member-state update without making either step depend on
    /// an already-removed voter.
    ///
    /// # Errors
    ///
    /// Returns leader, quorum, membership, or fatal runtime failures.
    pub async fn change_voters(
        &self,
        voters: std::collections::BTreeMap<NodeId, BasicNode>,
    ) -> Result<ClientWriteResponse<ControlRaftConfig>, ConsensusWriteError> {
        let submission_started = std::time::Instant::now();
        let _submission = self.membership_submission.lock().await;
        tracing::debug!(
            node_id = %self.node_id,
            wait_ms = submission_started.elapsed().as_millis(),
            "consensus membership submission admitted"
        );
        let authenticated = voters
            .into_iter()
            .map(|(node_id, node)| {
                if node.addr.trim().is_empty() {
                    Err(ConsensusWriteError::Membership(format!(
                        "consensus voter {node_id} has no authenticated endpoint"
                    )))
                } else {
                    Ok((node_id, node))
                }
            })
            .collect::<Result<std::collections::BTreeMap<_, _>, _>>()?;
        if authenticated.is_empty() {
            return Err(ConsensusWriteError::Membership(
                "consensus voter set cannot be empty".to_string(),
            ));
        }
        match self.raft.ensure_linearizable().await {
            Ok(_) => {}
            Err(RaftError::Fatal(error)) => {
                return Err(ConsensusWriteError::Fatal(Box::new(error)));
            }
            Err(RaftError::APIError(CheckIsLeaderError::ForwardToLeader(forward))) => {
                return Err(ConsensusWriteError::NotLeader((&forward).into()));
            }
            Err(RaftError::APIError(CheckIsLeaderError::QuorumNotEnough(error))) => {
                return Err(ConsensusWriteError::QuorumUnavailable(error.to_string()));
            }
        }
        let admission_term = self.current_term();
        // Retained publication commands/snapshots require their decoder even
        // before activation advances the schema floor. Endpoint authentication
        // does not prove that a newly added learner can consume this history.
        let state = self.persisted_control_state().map_err(|error| {
            ConsensusWriteError::Membership(format!(
                "cannot verify retained compatibility: {error}"
            ))
        })?;
        if state.requires_publication_decoder() {
            for (id, node) in &authenticated {
                if *id == self.node_id {
                    continue;
                }
                // Probe every remote recipient, including endpoint replacements.
                // A previous membership entry is not evidence of the current decoder.
                let verifier = self.decoder_admission.as_ref().ok_or_else(|| {
                    ConsensusWriteError::Membership(
                        "new recipients require capability-publication decoder admission proof"
                            .into(),
                    )
                })?;
                verifier
                    .verify(*id, &node.addr)
                    .await
                    .map_err(ConsensusWriteError::Membership)?;
            }
            // Network waits may span a leadership transition. Never start admission
            // on the authority of the earlier read alone.
            self.raft.ensure_linearizable().await.map_err(|error| {
                ConsensusWriteError::Membership(format!("admission authority changed: {error}"))
            })?;
        }
        let voter_ids = authenticated
            .keys()
            .copied()
            .collect::<std::collections::BTreeSet<_>>();
        for (node_id, node) in authenticated {
            validate_admission_term(admission_term, self.current_term())?;
            tracing::info!(node_id = %self.node_id, learner_id = %node_id, "consensus learner admission started");
            let started = std::time::Instant::now();
            self.raft
                .add_learner(node_id, node, true)
                .await
                .map_err(client_write_error)?;
            tracing::info!(node_id = %self.node_id, learner_id = %node_id, elapsed_ms = started.elapsed().as_millis(), "consensus learner admission completed");
        }
        validate_admission_term(admission_term, self.current_term())?;
        tracing::info!(node_id = %self.node_id, voter_count = voter_ids.len(), "consensus voter-set commit started");
        let result = self
            .raft
            .change_membership(voter_ids, false)
            .await
            .map_err(client_write_error);
        tracing::info!(node_id = %self.node_id, succeeded = result.is_ok(), elapsed_ms = submission_started.elapsed().as_millis(), "consensus voter-set commit finished");
        result
    }

    /// Executes a generated control mutation and returns its typed response.
    ///
    /// # Errors
    ///
    /// Returns structured leader, quorum, rejection, runtime, or internal errors.
    pub async fn mutate(
        &self,
        command: bmux_cluster_plugin_api::cluster_types::ControlCommand,
    ) -> Result<ControlResponse, ControlServiceError> {
        let command_id = command.command_id.clone();
        let encoded = crate::control_codec::encode_control_command(&command);
        Self::decode_write_response(&command_id, self.write(encoded).await)
    }

    /// Activates one durable schema feature after membership eligibility checks.
    ///
    /// # Errors
    ///
    /// Returns structured leader, quorum, rejection, runtime, or internal errors.
    pub async fn activate_feature(
        &self,
        command: crate::control_codec::FeatureActivationCommand,
    ) -> Result<ControlResponse, ControlServiceError> {
        let command_id = command.command_id.clone();
        let encoded = crate::control_codec::encode_feature_activation(&command);
        Self::decode_write_response(&command_id, self.write(encoded).await)
    }

    /// Publishes supported capabilities for an isolated single-member authority.
    /// Multi-member publication requires reconnect-bound decoder negotiation first.
    /// # Errors
    /// Rejects incomplete authority, incompatible evidence and failed durable writes.
    pub async fn publish_capabilities(
        &self,
        reports: Vec<bmux_cluster_plugin_api::cluster_capability_types::CapabilityReport>,
    ) -> Result<ControlResponse, ControlServiceError> {
        let _submission = self.membership_submission.lock().await;
        self.linearizable_control_state()
            .await
            .map_err(control_read_error)?;
        let snapshot =
            self.storage
                .state_machine()
                .map_err(|error| ControlServiceError::Internal {
                    reason: error.to_string(),
                })?;
        let membership = snapshot.committed_membership();
        let effective = self.raft.metrics().borrow().membership_config.clone();
        if &*effective != membership
            || membership.membership().nodes().count() != 1
            || membership.membership().voter_ids().collect::<Vec<_>>() != vec![self.node_id]
        {
            return Err(ControlServiceError::RuntimeUnavailable { reason: "publication requires an isolated single-member authority until peer decoder negotiation is available".into() });
        }
        let command = crate::capability_publication::PublicationCommand {
            reports,
            verified_at_unix_ms: crate::now_unix_ms(),
        };
        let bytes = command
            .encode()
            .map_err(|reason| ControlServiceError::RuntimeUnavailable { reason })?;
        let mut state = snapshot.control_state().clone();
        state
            .apply_publication(&command, membership)
            .map_err(|reason| ControlServiceError::Rejected {
                error:
                    bmux_cluster_plugin_api::cluster_types::ControlCommandError::InvalidTransition {
                        reason,
                    },
            })?;
        // No remote recipient exists and membership changes share this lock.
        // Do not use the raw writer, which deliberately rejects publication.
        let result = self
            .raft
            .client_write(ControlRequest(bytes))
            .await
            .map_err(client_write_error);
        Self::decode_write_response(&command.reports[0].command_id, result)
    }

    /// Submits a signed protocol refresh using quorum-confirmed authority.
    /// The caller supplies proofs, never the time used to validate credentials.
    /// Apply rechecks preconditions in log order and persistence precedes the reply.
    ///
    /// # Errors
    /// Rejects missing quorum, inactive features, invalid proofs and failed commits.
    pub async fn refresh_member_protocol(
        &self,
        command_id: bmux_cluster_plugin_api::cluster_types::CommandId,
        replacement: bmux_cluster_plugin_api::cluster_types::ClusterMember,
        expected_serial: String,
        expected_revision: u64,
        node_signature: Vec<u8>,
    ) -> Result<ControlResponse, ControlServiceError> {
        let mut state = self
            .linearizable_control_state()
            .await
            .map_err(control_read_error)?;
        let command = crate::control_codec::ProtocolRefreshCommand {
            command_id: command_id.value,
            replacement,
            expected_serial,
            expected_revision,
            node_signature,
            verified_at_unix_ms: crate::now_unix_ms(),
        };
        let encoded = command
            .encode()
            .map_err(|error| ControlServiceError::Internal {
                reason: error.to_string(),
            })?;
        let preflight = state.apply_refresh_command(&command);
        if let bmux_cluster_plugin_api::cluster_types::ControlCommandResult::Rejected { error } =
            preflight.result
        {
            return Err(ControlServiceError::Rejected { error });
        }
        Self::decode_write_response(&command_id, self.write(encoded).await)
    }

    /// Validates bootstrap proof using current authority and submits it durably.
    /// Verification time is assigned here, never accepted from a service caller.
    ///
    /// # Errors
    /// Rejects missing quorum, invalid proof/preconditions and failed commits.
    pub async fn bootstrap_principal(
        &self,
        proof: bmux_cluster_plugin_api::cluster_principal_bootstrap_types::BootstrapProof,
        expected_control_revision: u64,
    ) -> Result<ControlResponse, ControlServiceError> {
        self.linearizable_control_state()
            .await
            .map_err(control_read_error)?;
        let snapshot =
            self.storage
                .state_machine()
                .map_err(|error| ControlServiceError::Internal {
                    reason: error.to_string(),
                })?;
        let command = crate::principal_bootstrap::BootstrapCommand {
            proof,
            expected_control_revision,
            verified_at_unix_ms: crate::now_unix_ms(),
        };
        // The preflight uses one persisted state/membership snapshot. Apply checks
        // again in log order, so intervening membership or revision changes reject.
        let mut state = snapshot.control_state().clone();
        let preflight = state.apply_bootstrap_command(&command, snapshot.committed_membership());
        if let bmux_cluster_plugin_api::cluster_types::ControlCommandResult::Rejected { error } =
            preflight.result
        {
            return Err(ControlServiceError::Rejected { error });
        }
        let encoded = command
            .encode()
            .map_err(|reason| ControlServiceError::Internal { reason })?;
        Self::decode_write_response(
            &command.proof.statement.command_id,
            self.write(encoded).await,
        )
    }

    fn decode_write_response(
        command_id: &bmux_cluster_plugin_api::cluster_types::CommandId,
        result: Result<ClientWriteResponse<ControlRaftConfig>, ConsensusWriteError>,
    ) -> Result<ControlResponse, ControlServiceError> {
        match result {
            Ok(response) => {
                let response = crate::control_codec::decode_control_response(&response.data.0)
                    .map_err(|error| ControlServiceError::Internal {
                        reason: format!("committed control response is invalid: {error}"),
                    })?;
                match &response.result {
                    bmux_cluster_plugin_api::cluster_types::ControlCommandResult::Rejected {
                        error,
                    } => Err(ControlServiceError::Rejected {
                        error: error.clone(),
                    }),
                    bmux_cluster_plugin_api::cluster_types::ControlCommandResult::Accepted {
                        ..
                    } => Ok(response),
                }
            }
            Err(ConsensusWriteError::NotLeader(not_leader)) => {
                Err(ControlServiceError::NotLeader {
                    leader_node_id: not_leader.leader_id.map(|id| id.to_string()),
                    leader_endpoint: not_leader.leader_endpoint,
                })
            }
            Err(ConsensusWriteError::QuorumUnavailable(reason)) => {
                Err(ControlServiceError::QuorumUnavailable { reason })
            }
            Err(ConsensusWriteError::Fatal(error)) => Err(ControlServiceError::Internal {
                reason: format!(
                    "consensus runtime failed for command {}: {error}",
                    command_id.value
                ),
            }),
            Err(ConsensusWriteError::Membership(reason)) => {
                Err(ControlServiceError::RuntimeUnavailable { reason })
            }
        }
    }

    /// Returns a quorum-confirmed generated state view.
    ///
    /// # Errors
    ///
    /// Returns structured leader, quorum, runtime, or storage errors.
    pub async fn read_linearizable_view(&self) -> Result<ControlStateView, ControlServiceError> {
        self.linearizable_control_state()
            .await
            .map(|state| state.to_view(ControlReadConsistency::Linearizable))
            .map_err(control_read_error)
    }

    /// Returns an explicitly stale generated state view.
    ///
    /// # Errors
    ///
    /// Returns an internal error when durable state cannot be decoded.
    pub fn read_stale_view(&self) -> Result<ControlStateView, ControlServiceError> {
        self.persisted_control_state()
            .map(|state| state.to_view(ControlReadConsistency::Stale))
            .map_err(|error| ControlServiceError::Internal {
                reason: error.to_string(),
            })
    }

    /// Submits one canonical control command to the current leader.
    ///
    /// # Errors
    ///
    /// Returns actionable leader guidance, quorum unavailability, membership
    /// errors, or fatal Raft failures. A command is never replayed internally
    /// after dispatch.
    pub async fn write(
        &self,
        command: Vec<u8>,
    ) -> Result<ClientWriteResponse<ControlRaftConfig>, ConsensusWriteError> {
        // Publication has a decoder, but no membership-serialized proposal path
        // yet. Never let the generic raw writer bypass that compatibility boundary:
        // apply-time rejection cannot protect an older recipient's decoder.
        if command.starts_with(b"BMCAP001") {
            return Err(ConsensusWriteError::Membership(
                "capability publication requires membership-safe negotiated submission".into(),
            ));
        }
        match self.raft.ensure_linearizable().await {
            Ok(_) => {}
            Err(RaftError::Fatal(error)) => {
                return Err(ConsensusWriteError::Fatal(Box::new(error)));
            }
            Err(RaftError::APIError(CheckIsLeaderError::ForwardToLeader(forward))) => {
                return Err(ConsensusWriteError::NotLeader((&forward).into()));
            }
            Err(RaftError::APIError(CheckIsLeaderError::QuorumNotEnough(error))) => {
                return Err(ConsensusWriteError::QuorumUnavailable(error.to_string()));
            }
        }
        match self.raft.client_write(ControlRequest(command)).await {
            Ok(response) => Ok(response),
            Err(error) => Err(client_write_error(error)),
        }
    }

    /// Confirms leadership and quorum before returning the durable state.
    ///
    /// # Errors
    ///
    /// Returns actionable leader guidance, quorum unavailability, fatal Raft
    /// failures, or durable state read failures.
    pub async fn linearizable_control_state(&self) -> Result<ControlState, ConsensusReadError> {
        match self.raft.ensure_linearizable().await {
            Ok(_) => self
                .persisted_control_state()
                .map_err(ConsensusReadError::Storage),
            Err(error) => match error {
                RaftError::Fatal(error) => Err(ConsensusReadError::Fatal(Box::new(error))),
                RaftError::APIError(CheckIsLeaderError::ForwardToLeader(forward)) => {
                    Err(ConsensusReadError::NotLeader((&forward).into()))
                }
                RaftError::APIError(CheckIsLeaderError::QuorumNotEnough(error)) => {
                    Err(ConsensusReadError::QuorumUnavailable(error.to_string()))
                }
            },
        }
    }

    /// Returns a durable state-machine view without a quorum confirmation.
    ///
    /// Callers must explicitly label this result stale when presenting it.
    ///
    /// # Errors
    ///
    /// Returns an error when the durable state machine cannot be decoded.
    pub fn persisted_control_state(&self) -> Result<ControlState, ConsensusStorageError> {
        self.storage
            .state_machine()
            .map(|state_machine: ConsensusStateMachine| state_machine.control_state().clone())
    }

    /// Gracefully stops the node and waits for its Raft task to exit.
    ///
    /// # Errors
    ///
    /// Returns the host runtime join failure if the Raft task cannot be joined.
    pub async fn shutdown(
        &self,
    ) -> Result<(), <openraft::TokioRuntime as openraft::AsyncRuntime>::JoinError> {
        self.raft.shutdown().await
    }
}

fn client_write_error(
    error: RaftError<NodeId, openraft::error::ClientWriteError<NodeId, BasicNode>>,
) -> ConsensusWriteError {
    error.forward_to_leader::<BasicNode>().cloned().map_or_else(
        || match error {
            RaftError::Fatal(error) => ConsensusWriteError::Fatal(Box::new(error)),
            RaftError::APIError(error) => ConsensusWriteError::Membership(error.to_string()),
        },
        |forward| ConsensusWriteError::NotLeader((&forward).into()),
    )
}

fn control_read_error(error: ConsensusReadError) -> ControlServiceError {
    match error {
        ConsensusReadError::NotLeader(not_leader) => ControlServiceError::NotLeader {
            leader_node_id: not_leader.leader_id.map(|id| id.to_string()),
            leader_endpoint: not_leader.leader_endpoint,
        },
        ConsensusReadError::QuorumUnavailable(reason) => {
            ControlServiceError::QuorumUnavailable { reason }
        }
        ConsensusReadError::Fatal(error) => ControlServiceError::Internal {
            reason: error.to_string(),
        },
        ConsensusReadError::Storage(error) => ControlServiceError::Internal {
            reason: error.to_string(),
        },
    }
}

fn consensus_config(cluster_id: &str) -> Result<Arc<Config>, ConsensusRuntimeError> {
    let config = Config {
        cluster_name: cluster_id.to_owned(),
        // Authenticated peer RPCs include service dispatch and durable storage;
        // allow those operations to finish before treating a healthy leader as lost.
        heartbeat_interval: 100,
        election_timeout_min: 750,
        election_timeout_max: 1500,
        snapshot_policy: SnapshotPolicy::LogsSinceLast(SNAPSHOT_LOG_THRESHOLD),
        max_in_snapshot_log_to_keep: 0,
        purge_batch_size: 64,
        ..Config::default()
    }
    .validate()
    .map_err(|error| ConsensusRuntimeError::Configuration(error.to_string()))?;
    Ok(Arc::new(config))
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use crate::control_codec::{
        FeatureActivationCommand, decode_control_response, encode_control_command,
    };
    use crate::membership::{ClusterId, NodeIdentity};
    use bmux_cluster_plugin_api::cluster_types::{
        ClusterConsensusRole, ClusterNodeCapabilities, CommandId, ControlCommand,
        ControlCommandRequest, WorkspaceId,
    };
    use openraft::error::{RPCError, RemoteError, Unreachable};
    use openraft::network::{RPCOption, RaftNetwork};
    use openraft::raft::{
        AppendEntriesRequest, AppendEntriesResponse, InstallSnapshotRequest,
        InstallSnapshotResponse, VoteRequest, VoteResponse,
    };
    use std::collections::{BTreeMap, BTreeSet};
    use std::io;
    use std::sync::RwLock;
    use std::time::Duration;
    use tempfile::TempDir;

    type RegisteredNode = (BasicNode, openraft::Raft<ControlRaftConfig>);

    #[derive(Clone, Default)]
    pub struct InMemoryNetworkFactory {
        nodes: Arc<RwLock<BTreeMap<NodeId, RegisteredNode>>>,
    }

    impl InMemoryNetworkFactory {
        pub fn unregister(&self, id: NodeId) {
            self.nodes
                .write()
                .expect("network registry lock poisoned")
                .remove(&id);
        }

        pub fn register(
            &self,
            id: NodeId,
            node: BasicNode,
            raft: openraft::Raft<ControlRaftConfig>,
        ) {
            self.nodes
                .write()
                .expect("network registry lock poisoned")
                .insert(id, (node, raft));
        }
    }

    pub struct InMemoryNetwork {
        target: NodeId,
        expected: BasicNode,
        nodes: Arc<RwLock<BTreeMap<NodeId, RegisteredNode>>>,
    }

    #[allow(clippy::result_large_err)] // OpenRaft's transport error preserves typed remote context.
    impl InMemoryNetwork {
        fn remote<E: std::error::Error>(
            &self,
        ) -> Result<openraft::Raft<ControlRaftConfig>, RPCError<NodeId, BasicNode, E>> {
            let Some((actual, raft)) = self
                .nodes
                .read()
                .expect("network registry lock poisoned")
                .get(&self.target)
                .cloned()
            else {
                return Err(Unreachable::new(&io::Error::new(
                    io::ErrorKind::NotConnected,
                    "target node is not registered",
                ))
                .into());
            };
            if actual != self.expected {
                return Err(Unreachable::new(&io::Error::new(
                    io::ErrorKind::PermissionDenied,
                    "target node identity does not match its registered endpoint",
                ))
                .into());
            }
            Ok(raft)
        }
    }

    impl RaftNetworkFactory<ControlRaftConfig> for InMemoryNetworkFactory {
        type Network = InMemoryNetwork;

        async fn new_client(&mut self, target: NodeId, node: &BasicNode) -> Self::Network {
            InMemoryNetwork {
                target,
                expected: node.clone(),
                nodes: self.nodes.clone(),
            }
        }
    }

    impl RaftNetwork<ControlRaftConfig> for InMemoryNetwork {
        async fn append_entries(
            &mut self,
            rpc: AppendEntriesRequest<ControlRaftConfig>,
            _option: RPCOption,
        ) -> Result<AppendEntriesResponse<NodeId>, RPCError<NodeId, BasicNode, RaftError<NodeId>>>
        {
            let remote = self.remote()?;
            remote.append_entries(rpc).await.map_err(|error| {
                RemoteError::new_with_node(self.target, self.expected.clone(), error).into()
            })
        }

        async fn install_snapshot(
            &mut self,
            rpc: InstallSnapshotRequest<ControlRaftConfig>,
            _option: RPCOption,
        ) -> Result<
            InstallSnapshotResponse<NodeId>,
            RPCError<NodeId, BasicNode, RaftError<NodeId, openraft::error::InstallSnapshotError>>,
        > {
            let remote = self.remote()?;
            remote.install_snapshot(rpc).await.map_err(|error| {
                RemoteError::new_with_node(self.target, self.expected.clone(), error).into()
            })
        }

        async fn vote(
            &mut self,
            rpc: VoteRequest<NodeId>,
            _option: RPCOption,
        ) -> Result<VoteResponse<NodeId>, RPCError<NodeId, BasicNode, RaftError<NodeId>>> {
            let remote = self.remote()?;
            remote.vote(rpc).await.map_err(|error| {
                RemoteError::new_with_node(self.target, self.expected.clone(), error).into()
            })
        }
    }

    #[tokio::test]
    async fn bootstrap_submission_rejects_uninitialized_authority() {
        use bmux_cluster_plugin_api::cluster_principal_bootstrap_types::{
            BootstrapProof, BootstrapStatement,
        };
        let root = TempDir::new().unwrap();
        let node = ConsensusNode::start(
            root.path(),
            "bootstrap-test",
            NodeId::from(91),
            InMemoryNetworkFactory::default(),
        )
        .await
        .unwrap();
        let result = node
            .bootstrap_principal(
                BootstrapProof {
                    statement: BootstrapStatement {
                        schema_version: 1,
                        cluster_id: "bootstrap-test".into(),
                        membership_term: 0,
                        membership_leader_id: NodeId::from(91).to_string(),
                        membership_log_index: 0,
                        principal_id: uuid::Uuid::new_v4(),
                        principal_public_key: String::new(),
                        command_id: bmux_cluster_plugin_api::cluster_types::CommandId {
                            value: uuid::Uuid::new_v4(),
                        },
                    },
                    principal_signature: Vec::new(),
                    voter_approvals: Vec::new(),
                },
                0,
            )
            .await;
        assert!(matches!(result, Err(ControlServiceError::NotLeader { .. })));
        assert!(
            node.persisted_control_state()
                .unwrap()
                .principal_bootstrap
                .is_none()
        );
        node.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn refresh_submission_requires_quorum_and_activation() {
        use crate::membership::{
            ClusterId, NodeIdentity, initializer_capabilities, issue_test_member,
        };
        let root = TempDir::new().unwrap();
        let identity = NodeIdentity::new_for_test(92);
        let cluster: ClusterId = "cluster:00000000-0000-0000-0000-000000000092"
            .parse()
            .unwrap();
        let node = ConsensusNode::start(
            root.path(),
            &cluster.to_string(),
            *identity.node_id(),
            InMemoryNetworkFactory::default(),
        )
        .await
        .unwrap();
        let member = issue_test_member(
            &identity,
            cluster,
            &identity,
            "tls://127.0.0.1:49992",
            initializer_capabilities(),
            42,
        );
        let submit = || {
            node.refresh_member_protocol(
                CommandId {
                    value: uuid::Uuid::new_v4(),
                },
                member.clone(),
                member.credential_serial.clone(),
                0,
                vec![0; 64],
            )
        };
        assert!(matches!(
            submit().await,
            Err(ControlServiceError::NotLeader { .. })
        ));
        node.initialize_single(*identity.node_id(), "node")
            .await
            .unwrap();
        wait_for_leader(&[&node]).await;
        assert!(matches!(
            submit().await,
            Err(ControlServiceError::Rejected { .. })
        ));
        assert_eq!(node.persisted_control_state().unwrap().revision, 0);
        node.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn raw_publication_write_cannot_bypass_compatibility_gate() {
        let root = TempDir::new().unwrap();
        let node = ConsensusNode::start(
            root.path(),
            "publication-gate",
            NodeId::from(95),
            InMemoryNetworkFactory::default(),
        )
        .await
        .unwrap();
        node.initialize_single(NodeId::from(95), "node")
            .await
            .unwrap();
        wait_for_leader(&[&node]).await;
        assert!(matches!(
            node.publish_capabilities(Vec::new()).await,
            Err(ControlServiceError::RuntimeUnavailable { .. })
        ));
        let before = node.persisted_control_state().unwrap();
        let result = node.write(b"BMCAP001".to_vec()).await;
        assert!(
            matches!(result, Err(ConsensusWriteError::Membership(reason)) if reason == "capability publication requires membership-safe negotiated submission")
        );
        assert_eq!(node.persisted_control_state().unwrap(), before);
        node.shutdown().await.unwrap();
    }

    pub async fn wait_for_leader(nodes: &[&ConsensusNode]) -> NodeId {
        tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                for node in nodes {
                    // Election hints may precede a working quorum after links recover.
                    if node.raft().current_leader().await == Some(node.node_id())
                        && node.raft().ensure_linearizable().await.is_ok()
                    {
                        return node.node_id();
                    }
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("cluster should elect a leader")
    }

    pub async fn wait_for_leader_excluding(nodes: &[&ConsensusNode], excluded: NodeId) -> NodeId {
        tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                for node in nodes {
                    if node.node_id() != excluded
                        && node.raft().current_leader().await == Some(node.node_id())
                        && node.raft().ensure_linearizable().await.is_ok()
                    {
                        return node.node_id();
                    }
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("surviving quorum should elect a replacement leader")
    }

    fn command(id: u128, workspace: u128) -> ControlCommand {
        ControlCommand {
            schema_version: 1,
            principal_id: "principal:test".to_string(),
            command_id: CommandId {
                value: uuid::Uuid::from_u128(id),
            },
            issued_at_unix_ms: u64::try_from(id).unwrap(),
            request: ControlCommandRequest::CreateWorkspace {
                workspace_id: WorkspaceId {
                    value: uuid::Uuid::from_u128(workspace),
                },
                name: Some(format!("workspace-{workspace}")),
            },
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    #[allow(clippy::too_many_lines)] // One scenario intentionally spans formation through failover.
    async fn three_voters_replicate_dedup_and_resume_after_leader_shutdown() {
        let network = InMemoryNetworkFactory::default();
        let roots = [
            TempDir::new().unwrap(),
            TempDir::new().unwrap(),
            TempDir::new().unwrap(),
        ];
        let ids = [NodeId::from(1), NodeId::from(2), NodeId::from(3)];
        let endpoints = ["memory://one", "memory://two", "memory://three"];
        let mut nodes = Vec::new();
        for index in 0..3 {
            let node = ConsensusNode::start(
                roots[index].path(),
                "cluster-runtime-test",
                ids[index],
                network.clone(),
            )
            .await
            .unwrap();
            network.register(
                ids[index],
                BasicNode::new(endpoints[index]),
                node.raft().clone(),
            );
            nodes.push(node);
        }

        nodes[0]
            .initialize_single(ids[0], endpoints[0])
            .await
            .unwrap();
        assert_eq!(wait_for_leader(&[&nodes[0]]).await, ids[0]);
        nodes[0]
            .raft()
            .add_learner(ids[1], BasicNode::new(endpoints[1]), true)
            .await
            .unwrap();
        nodes[0]
            .raft()
            .add_learner(ids[2], BasicNode::new(endpoints[2]), true)
            .await
            .unwrap();
        nodes[0]
            .raft()
            .change_membership(BTreeSet::from(ids), false)
            .await
            .unwrap();

        let first = command(1, 10);
        let encoded = encode_control_command(&first);
        let response = nodes[0].write(encoded.clone()).await.unwrap();
        let decoded = decode_control_response(&response.data.0).unwrap();
        assert_eq!(decoded.command_id, first.command_id);
        let replay = nodes[0].write(encoded).await.unwrap();
        assert_eq!(replay.data, response.data);

        let follower = &nodes[1];
        assert!(matches!(
            follower.write(encode_control_command(&command(99, 99))).await,
            Err(ConsensusWriteError::NotLeader(NotLeader {
                leader_id: Some(id),
                leader_endpoint: Some(endpoint),
            })) if id == ids[0] && endpoint == endpoints[0]
        ));
        let linear = nodes[0].read_linearizable_view().await.unwrap();
        assert_eq!(linear.consistency, ControlReadConsistency::Linearizable);
        assert!(
            linear
                .workspaces
                .iter()
                .any(|workspace| workspace.workspace_id.value == first_workspace())
        );
        let stale = follower.read_stale_view().unwrap();
        assert_eq!(stale.consistency, ControlReadConsistency::Stale);
        assert!(matches!(
            follower.linearizable_control_state().await,
            Err(ConsensusReadError::NotLeader(NotLeader {
                leader_id: Some(id),
                leader_endpoint: Some(endpoint),
            })) if id == ids[0] && endpoint == endpoints[0]
        ));

        tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                if nodes.iter().all(|node| {
                    node.persisted_control_state()
                        .is_ok_and(|state| state.workspaces.contains_key(&first_workspace()))
                }) {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("all voters should apply the first command");

        nodes[0].shutdown().await.unwrap();
        let replacement = wait_for_leader_excluding(&[&nodes[1], &nodes[2]], ids[0]).await;
        assert_ne!(replacement, ids[0]);
        let leader = if replacement == ids[1] {
            &nodes[1]
        } else {
            &nodes[2]
        };
        let second = command(2, 20);
        leader.write(encode_control_command(&second)).await.unwrap();
        let linear = leader.read_linearizable_view().await.unwrap();
        assert!(
            linear
                .workspaces
                .iter()
                .any(|workspace| workspace.workspace_id.value == second_workspace())
        );

        tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                if nodes[1..].iter().all(|node| {
                    node.persisted_control_state()
                        .is_ok_and(|state| state.workspaces.contains_key(&second_workspace()))
                }) {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("surviving quorum should apply a command after leader loss");

        nodes[1].shutdown().await.unwrap();
        nodes[2].shutdown().await.unwrap();
    }

    #[allow(clippy::too_many_lines)] // Integration test keeps one three-voter lifecycle explicit.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn three_voters_activate_feature_floor_and_continue_v1_writes() {
        let network = InMemoryNetworkFactory::default();
        let roots = [
            TempDir::new().unwrap(),
            TempDir::new().unwrap(),
            TempDir::new().unwrap(),
        ];
        let ids = [NodeId::from(31), NodeId::from(32), NodeId::from(33)];
        let endpoints = ["memory://f1", "memory://f2", "memory://f3"];
        let mut nodes = Vec::new();
        for index in 0..3 {
            let node = ConsensusNode::start(
                roots[index].path(),
                "cluster-feature-floor-test",
                ids[index],
                network.clone(),
            )
            .await
            .unwrap();
            network.register(
                ids[index],
                BasicNode::new(endpoints[index]),
                node.raft().clone(),
            );
            nodes.push(node);
        }
        nodes[0]
            .initialize_single(ids[0], endpoints[0])
            .await
            .unwrap();
        // Initialization admits election; membership writes require a ready leader.
        assert_eq!(wait_for_leader(&[&nodes[0]]).await, ids[0]);
        nodes[0]
            .change_voters(BTreeMap::from([
                (ids[0], BasicNode::new(endpoints[0])),
                (ids[1], BasicNode::new(endpoints[1])),
                (ids[2], BasicNode::new(endpoints[2])),
            ]))
            .await
            .unwrap();
        let leader_id = wait_for_leader(&nodes.iter().collect::<Vec<_>>()).await;
        let leader = nodes
            .iter()
            .find(|node| node.node_id() == leader_id)
            .unwrap();

        let issuer = NodeIdentity::new_for_test(70);
        let cluster_id = ClusterId::generate();
        let now = crate::now_unix_ms();
        for (index, id) in ids.into_iter().enumerate() {
            let identity = NodeIdentity::new_for_test(80 + u64::try_from(index).unwrap());
            let mut member = crate::membership::issue_test_member(
                &issuer,
                cluster_id,
                &identity,
                endpoints[index],
                ClusterNodeCapabilities {
                    consensus_role: ClusterConsensusRole::Voter,
                    worker: true,
                    ingress: true,
                },
                now,
            );
            member.node_id = id.to_string();
            member.public_key = identity.public_key().to_string();
            member.credential_signature.clear();
            member = crate::membership::issue_test_member_with_protocol(
                &issuer,
                cluster_id,
                &identity,
                endpoints[index],
                member.capabilities,
                member.negotiated_protocol,
                now,
            );
            // Test NodeId values are not derived from the test identities, so
            // re-key the signed record to the committed Raft id and skip the
            // service eligibility layer here; the state-machine integration
            // below proves replication and migration of the schema-2 envelope.
            member.node_id = id.to_string();
            let response = leader
                .mutate(ControlCommand {
                    schema_version: 1,
                    principal_id: "principal:test".to_string(),
                    command_id: CommandId {
                        value: uuid::Uuid::from_u128(300 + u128::try_from(index).unwrap()),
                    },
                    issued_at_unix_ms: now,
                    request: ControlCommandRequest::UpsertMember { member },
                })
                .await;
            assert!(
                response.is_err(),
                "re-keyed unsigned fixture must fail closed"
            );
        }

        let activation = FeatureActivationCommand {
            principal_id: "principal:test".to_string(),
            command_id: CommandId {
                value: uuid::Uuid::from_u128(400),
            },
            issued_at_unix_ms: now,
            expected_control_revision: 0,
            read_schema_floor: 2,
            write_schema_floor: 2,
            feature: crate::membership::ATOMIC_LAYOUT_FEATURE.to_string(),
        };
        leader.activate_feature(activation).await.unwrap();
        leader.mutate(command(401, 401)).await.unwrap();
        tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                if nodes.iter().all(|node| {
                    node.persisted_control_state().is_ok_and(|state| {
                        state.read_schema_floor == 2
                            && state.write_schema_floor == 2
                            && state
                                .activated_features
                                .contains(crate::membership::ATOMIC_LAYOUT_FEATURE)
                            && state.workspaces.contains_key(&uuid::Uuid::from_u128(401))
                    })
                }) {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .unwrap();
        for node in nodes {
            node.shutdown().await.unwrap();
        }
    }

    const fn first_workspace() -> uuid::Uuid {
        uuid::Uuid::from_u128(10)
    }

    const fn second_workspace() -> uuid::Uuid {
        uuid::Uuid::from_u128(20)
    }

    struct FailingCaller;

    impl bmux_plugin::ServiceCaller for FailingCaller {
        fn call_service_raw(
            &self,
            _capability: &str,
            _kind: bmux_plugin_sdk::ServiceKind,
            _interface_id: &str,
            _operation: &str,
            _payload: Vec<u8>,
        ) -> bmux_plugin_sdk::Result<Vec<u8>> {
            Err(bmux_plugin_sdk::PluginError::UnsupportedHostOperation {
                operation: "call_service_raw",
            })
        }

        fn execute_kernel_request(
            &self,
            _request: bmux_ipc::Request,
        ) -> bmux_plugin_sdk::Result<bmux_ipc::ResponsePayload> {
            Err(bmux_plugin_sdk::PluginError::UnsupportedHostOperation {
                operation: "execute_kernel_request",
            })
        }
    }

    #[tokio::test]
    async fn change_voters_catches_up_learners_and_removes_old_voters() {
        let network = InMemoryNetworkFactory::default();
        let roots = [
            TempDir::new().unwrap(),
            TempDir::new().unwrap(),
            TempDir::new().unwrap(),
        ];
        let ids = [NodeId::from(81), NodeId::from(82), NodeId::from(83)];
        let endpoints = ["memory://m1", "memory://m2", "memory://m3"];
        let mut nodes = Vec::new();
        for index in 0..3 {
            let node = ConsensusNode::start(
                roots[index].path(),
                "cluster-voter-change-test",
                ids[index],
                network.clone(),
            )
            .await
            .unwrap();
            network.register(
                ids[index],
                BasicNode::new(endpoints[index]),
                node.raft().clone(),
            );
            nodes.push(node);
        }
        nodes[0]
            .initialize_single(ids[0], endpoints[0])
            .await
            .unwrap();
        wait_for_leader(&[&nodes[0]]).await;
        nodes[0]
            .change_voters(BTreeMap::from([
                (ids[0], BasicNode::new(endpoints[0])),
                (ids[1], BasicNode::new(endpoints[1])),
            ]))
            .await
            .unwrap();
        assert_eq!(wait_for_leader(&[&nodes[0], &nodes[1]]).await, ids[0]);
        let prospective = BTreeMap::from([
            (ids[1], BasicNode::new(endpoints[1])),
            (ids[2], BasicNode::new(endpoints[2])),
        ]);
        network.unregister(ids[1]);
        network.unregister(ids[2]);
        let failed = tokio::time::timeout(
            Duration::from_secs(5),
            nodes[0].change_voters(prospective.clone()),
        )
        .await
        .expect("voter transition failure should be bounded")
        .expect_err("minority must reject voter transition");
        assert!(matches!(failed, ConsensusWriteError::QuorumUnavailable(_)));
        assert_eq!(
            nodes[0].committed_voter_ids(),
            BTreeSet::from([ids[0], ids[1]])
        );
        for index in 1..3 {
            network.register(
                ids[index],
                BasicNode::new(endpoints[index]),
                nodes[index].raft().clone(),
            );
        }
        let recovered_leader = wait_for_leader(&[&nodes[0], &nodes[1], &nodes[2]]).await;
        let recovered_index = ids
            .iter()
            .position(|node_id| *node_id == recovered_leader)
            .unwrap();
        nodes[recovered_index]
            .change_voters(prospective.clone())
            .await
            .unwrap();
        nodes[0].shutdown().await.unwrap();
        let replacement = wait_for_leader(&[&nodes[1], &nodes[2]]).await;
        let replacement_index = if replacement == ids[1] { 1 } else { 2 };
        nodes[replacement_index]
            .change_voters(prospective)
            .await
            .unwrap();
        assert!(replacement == ids[1] || replacement == ids[2]);
        let leader = if replacement == ids[1] {
            &nodes[1]
        } else {
            &nodes[2]
        };
        leader
            .mutate(command(82, 820))
            .await
            .expect("new voter set should commit after old voter removal");
        nodes[1].shutdown().await.unwrap();
        nodes[2].shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn quorum_loss_rejects_writes_and_preserves_explicit_stale_reads() {
        let network = InMemoryNetworkFactory::default();
        let roots = [
            TempDir::new().unwrap(),
            TempDir::new().unwrap(),
            TempDir::new().unwrap(),
        ];
        let ids = [NodeId::from(41), NodeId::from(42), NodeId::from(43)];
        let endpoints = ["memory://q1", "memory://q2", "memory://q3"];
        let mut nodes = Vec::new();
        for index in 0..3 {
            let node = ConsensusNode::start(
                roots[index].path(),
                "cluster-quorum-loss-test",
                ids[index],
                network.clone(),
            )
            .await
            .unwrap();
            network.register(
                ids[index],
                BasicNode::new(endpoints[index]),
                node.raft().clone(),
            );
            nodes.push(node);
        }

        nodes[0]
            .initialize_single(ids[0], endpoints[0])
            .await
            .unwrap();
        wait_for_leader(&[&nodes[0]]).await;
        for index in 1..3 {
            nodes[0]
                .raft()
                .add_learner(ids[index], BasicNode::new(endpoints[index]), true)
                .await
                .unwrap();
        }
        nodes[0]
            .raft()
            .change_membership(BTreeSet::from(ids), false)
            .await
            .unwrap();
        nodes[0]
            .mutate(command(70, 700))
            .await
            .expect("quorum should commit baseline command");

        nodes[1].shutdown().await.unwrap();
        nodes[2].shutdown().await.unwrap();

        let error = tokio::time::timeout(Duration::from_secs(5), nodes[0].mutate(command(71, 701)))
            .await
            .expect("quorum failure should be bounded")
            .expect_err("minority leader must reject writes");
        assert!(matches!(
            error,
            ControlServiceError::QuorumUnavailable { .. }
        ));

        let linear =
            tokio::time::timeout(Duration::from_secs(5), nodes[0].read_linearizable_view())
                .await
                .expect("linearizable read failure should be bounded")
                .expect_err("minority must reject linearizable reads");
        assert!(matches!(
            linear,
            ControlServiceError::QuorumUnavailable { .. }
        ));

        let stale = nodes[0].read_stale_view().unwrap();
        assert_eq!(stale.consistency, ControlReadConsistency::Stale);
        assert!(
            stale
                .workspaces
                .iter()
                .any(|workspace| workspace.workspace_id.value == uuid::Uuid::from_u128(700))
        );
        assert!(
            !stale
                .workspaces
                .iter()
                .any(|workspace| workspace.workspace_id.value == uuid::Uuid::from_u128(701))
        );

        nodes[0].shutdown().await.unwrap();
    }

    #[tokio::test]
    #[allow(clippy::too_many_lines)]
    async fn membership_publication_retries_after_quorum_restoration() {
        let network = InMemoryNetworkFactory::default();
        let roots = [
            TempDir::new().unwrap(),
            TempDir::new().unwrap(),
            TempDir::new().unwrap(),
        ];
        let identities = [
            NodeIdentity::new_for_test(51),
            NodeIdentity::new_for_test(52),
            NodeIdentity::new_for_test(53),
        ];
        let ids = identities.each_ref().map(|identity| *identity.node_id());
        let endpoints = ["memory://p1", "memory://p2", "memory://p3"];
        let cluster_id: ClusterId = "cluster:00000000-0000-0000-0000-000000000001"
            .parse()
            .unwrap();
        let mut nodes = Vec::new();
        for index in 0..3 {
            let node = ConsensusNode::start(
                roots[index].path(),
                &cluster_id.to_string(),
                ids[index],
                network.clone(),
            )
            .await
            .unwrap();
            network.register(
                ids[index],
                BasicNode::new(endpoints[index]),
                node.raft().clone(),
            );
            nodes.push(node);
        }
        nodes[0]
            .initialize_single(ids[0], endpoints[0])
            .await
            .unwrap();
        wait_for_leader(&[&nodes[0]]).await;
        for index in 1..3 {
            nodes[0]
                .raft()
                .add_learner(ids[index], BasicNode::new(endpoints[index]), true)
                .await
                .unwrap();
        }
        nodes[0]
            .raft()
            .change_membership(BTreeSet::from(ids), false)
            .await
            .unwrap();

        let member = crate::membership::issue_test_member(
            &identities[0],
            cluster_id,
            &identities[0],
            "node-a",
            ClusterNodeCapabilities {
                consensus_role: ClusterConsensusRole::Voter,
                worker: true,
                ingress: true,
            },
            crate::now_unix_ms(),
        );
        network.unregister(ids[1]);
        network.unregister(ids[2]);
        let failed = tokio::time::timeout(
            Duration::from_secs(5),
            crate::consensus_membership::publish_members(
                nodes[0].clone(),
                &member.node_id,
                [member.clone()],
            ),
        )
        .await
        .expect("publication failure should be bounded")
        .expect_err("minority must reject membership publication");
        assert!(failed.contains("QuorumUnavailable"));
        assert!(nodes[0].read_stale_view().unwrap().members.is_empty());

        for index in 1..3 {
            network.register(
                ids[index],
                BasicNode::new(endpoints[index]),
                nodes[index].raft().clone(),
            );
        }
        // A successful quorum probe cannot prevent a subsequent leadership change.
        // Retry the idempotent publication sequence, not arbitrary errors, while
        // the restored cluster settles. The deadline keeps non-recovery fatal.
        let view = tokio::time::timeout(Duration::from_secs(15), async {
            loop {
                let leader_id = wait_for_leader(&nodes.iter().collect::<Vec<_>>()).await;
                let leader = &nodes[ids.iter().position(|id| *id == leader_id).unwrap()];
                let attempt = async {
                    for _ in 0..2 {
                        crate::consensus_membership::publish_members(
                            leader.clone(),
                            &member.node_id,
                            [member.clone()],
                        )
                        .await?;
                    }
                    leader
                        .read_linearizable_view()
                        .await
                        .map_err(|error| format!("{error:?}"))
                }
                .await;
                match attempt {
                    Ok(view) => break view,
                    Err(error) if error.contains("NotLeader") => {
                        tokio::time::sleep(Duration::from_millis(20)).await;
                    }
                    Err(error) => panic!("publication after quorum restoration failed: {error}"),
                }
            }
        })
        .await
        .expect("restored quorum should accept membership publication");
        assert_eq!(view.members, vec![member]);

        for node in nodes {
            node.shutdown().await.unwrap();
        }
    }

    async fn assert_bootstrap_activation_service(node: &ConsensusNode, id: NodeId) {
        use bmux_cluster_plugin_api::cluster_control_command_v2::ClusterControlCommandService;
        let registry = crate::consensus_network::ConsensusNodeRegistry::default();
        registry.insert(id, node.clone()).unwrap();
        let service = crate::consensus_network::ControlServiceHandle::new(
            Arc::new(FailingCaller),
            id,
            registry.clone(),
        );
        let revision = node.linearizable_control_state().await.unwrap().revision;
        let command_id = bmux_cluster_plugin_api::cluster_types::CommandId {
            value: uuid::Uuid::new_v4(),
        };
        let now = crate::now_unix_ms();
        let first = service
            .activate_feature(
                command_id.clone(),
                id.to_string(),
                now,
                revision,
                3,
                3,
                "principal-bootstrap-v1".into(),
            )
            .await
            .unwrap();
        let retry = service
            .activate_feature(
                command_id.clone(),
                id.to_string(),
                now,
                revision,
                3,
                3,
                "principal-bootstrap-v1".into(),
            )
            .await
            .unwrap();
        assert_eq!(first, retry);
        let conflict = service
            .activate_feature(
                command_id,
                id.to_string(),
                now,
                revision,
                3,
                3,
                "different-feature".into(),
            )
            .await;
        assert!(matches!(
            conflict,
            Err(ControlServiceError::Rejected {
                error:
                    bmux_cluster_plugin_api::cluster_types::ControlCommandError::CommandIdConflict
            })
        ));
        registry.remove(id).unwrap();
    }

    #[tokio::test]
    async fn bootstrap_commit_survives_restart_and_preserves_retry_revision() {
        use crate::membership::{
            ClusterId, NodeIdentity, initializer_capabilities, issue_test_member,
        };
        use bmux_cluster_plugin_api::cluster_principal_bootstrap_types::{
            BootstrapProof, BootstrapStatement, VoterApproval,
        };
        let root = TempDir::new().unwrap();
        let network = InMemoryNetworkFactory::default();
        let identity = NodeIdentity::new_for_test(95);
        let id = *identity.node_id();
        let cluster: ClusterId = "cluster:00000000-0000-0000-0000-000000000095"
            .parse()
            .unwrap();
        let node = ConsensusNode::start(root.path(), &cluster.to_string(), id, network.clone())
            .await
            .unwrap();
        node.initialize_single(id, "tls://127.0.0.1:49995")
            .await
            .unwrap();
        wait_for_leader(&[&node]).await;
        let member = issue_test_member(
            &identity,
            cluster,
            &identity,
            "tls://127.0.0.1:49995",
            initializer_capabilities(),
            crate::now_unix_ms(),
        );
        crate::consensus_membership::publish_members(node.clone(), &id.to_string(), [member])
            .await
            .unwrap();
        assert_bootstrap_activation_service(&node, id).await;
        let snapshot = node.storage.state_machine().unwrap();
        let membership = snapshot.committed_membership().log_id().unwrap();
        let key = iroh::SecretKey::from_bytes(&[95; 32]);
        let statement = BootstrapStatement {
            schema_version: 1,
            cluster_id: cluster.to_string(),
            membership_term: membership.leader_id.term,
            membership_leader_id: membership.leader_id.node_id.to_string(),
            membership_log_index: membership.index,
            principal_id: uuid::Uuid::new_v4(),
            principal_public_key: key.public().to_string(),
            command_id: bmux_cluster_plugin_api::cluster_types::CommandId {
                value: uuid::Uuid::new_v4(),
            },
        };
        let payload = crate::principal_bootstrap::signing_payload(&statement).unwrap();
        let proof = BootstrapProof {
            statement,
            principal_signature: key.sign(&payload).to_bytes().to_vec(),
            voter_approvals: vec![VoterApproval {
                node_id: id.to_string(),
                signature: identity.sign(&payload),
            }],
        };
        let revision = snapshot.control_state().revision;
        drop(snapshot);
        let result = node
            .bootstrap_principal(proof.clone(), revision)
            .await
            .unwrap();
        assert_eq!(result.control_revision, revision + 1);
        node.shutdown().await.unwrap();
        drop(node);
        let recovered = ConsensusNode::start(root.path(), &cluster.to_string(), id, network)
            .await
            .unwrap();
        wait_for_leader(&[&recovered]).await;
        assert_eq!(
            recovered
                .bootstrap_principal(proof.clone(), revision)
                .await
                .unwrap(),
            result
        );
        let mut conflict = proof;
        conflict.statement.principal_id = uuid::Uuid::new_v4();
        assert!(
            recovered
                .bootstrap_principal(conflict, revision)
                .await
                .is_err()
        );
        recovered.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn publication_commit_survives_restart_and_rejects_unqualified_admission() {
        use crate::membership::{
            ClusterId, NodeIdentity, initializer_capabilities, issue_test_member,
        };
        use bmux_cluster_plugin_api::cluster_capability_types::CapabilityReport;
        let root = TempDir::new().unwrap();
        let identity = NodeIdentity::new_for_test(96);
        let id = *identity.node_id();
        let cluster: ClusterId = "cluster:00000000-0000-0000-0000-000000000096"
            .parse()
            .unwrap();
        let network = InMemoryNetworkFactory::default();
        let node = ConsensusNode::start(root.path(), &cluster.to_string(), id, network.clone())
            .await
            .unwrap();
        node.initialize_single(id, "node").await.unwrap();
        wait_for_leader(&[&node]).await;
        let member = issue_test_member(
            &identity,
            cluster,
            &identity,
            "tls://127.0.0.1:49996",
            initializer_capabilities(),
            crate::now_unix_ms(),
        );
        crate::consensus_membership::publish_members(
            node.clone(),
            &id.to_string(),
            [member.clone()],
        )
        .await
        .unwrap();
        let snapshot = node.storage.state_machine().unwrap();
        let position = snapshot.committed_membership().log_id().unwrap();
        let mut report = CapabilityReport {
            report_version: 1,
            cluster_id: cluster.to_string(),
            node_id: id.to_string(),
            credential_serial: member.credential_serial.clone(),
            membership_term: position.leader_id.term,
            membership_leader_id: position.leader_id.node_id.to_string(),
            membership_log_index: position.index,
            expected_report_revision: 0,
            command_id: CommandId {
                value: uuid::Uuid::new_v4(),
            },
            schema_min: 1,
            schema_max: 4,
            features: vec![
                "capability-publication-v1".into(),
                "protocol-refresh-v1".into(),
            ],
            signature: Vec::new(),
        };
        report.signature =
            identity.sign(&crate::capability_publication::signing_payload(&report).unwrap());
        let revision = snapshot.control_state().revision;
        drop(snapshot);
        let result = node
            .publish_capabilities(vec![report.clone()])
            .await
            .unwrap();
        assert_eq!(result.control_revision, revision + 1);
        assert!(
            node.persisted_control_state()
                .unwrap()
                .requires_publication_decoder()
        );
        node.shutdown().await.unwrap();
        drop(node);
        let recovered = ConsensusNode::start(root.path(), &cluster.to_string(), id, network)
            .await
            .unwrap();
        wait_for_leader(&[&recovered]).await;
        assert_eq!(
            recovered
                .publish_capabilities(vec![report.clone()])
                .await
                .unwrap(),
            result
        );
        assert_unqualified_admission_rejected(&recovered, id).await;
        let mut conflict = report;
        conflict.schema_max = 5;
        conflict.signature =
            identity.sign(&crate::capability_publication::signing_payload(&conflict).unwrap());
        assert!(
            recovered
                .publish_capabilities(vec![conflict])
                .await
                .is_err()
        );
        assert_eq!(
            recovered.persisted_control_state().unwrap().revision,
            result.control_revision
        );
        recovered.shutdown().await.unwrap();
    }

    #[test]
    fn admission_evidence_does_not_survive_a_leadership_term_change() {
        assert!(validate_admission_term(7, 7).is_ok());
        assert!(
            matches!(validate_admission_term(7, 8), Err(ConsensusWriteError::Membership(reason)) if reason.contains("retry compatibility checks"))
        );
    }

    async fn assert_unqualified_admission_rejected(node: &ConsensusNode, id: NodeId) {
        struct RejectDecoder;
        impl crate::consensus_network::DecoderAdmission for RejectDecoder {
            fn verify<'a>(
                &'a self,
                target: NodeId,
                endpoint: &'a str,
            ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), String>> + Send + 'a>>
            {
                Box::pin(async move {
                    assert_eq!(target, NodeId::from(97));
                    assert_eq!(endpoint, "new-node");
                    Err("unsupported remote decoder".into())
                })
            }
        }
        let admission = node
            .change_voters(BTreeMap::from([
                (id, BasicNode::new("node")),
                (NodeId::from(97), BasicNode::new("new-node")),
            ]))
            .await;
        assert!(
            matches!(admission, Err(ConsensusWriteError::Membership(reason)) if reason.contains("decoder admission proof"))
        );
        let mut guarded = node.clone();
        guarded.decoder_admission = Some(Arc::new(RejectDecoder));
        let before = node.committed_member_ids();
        let admission = guarded
            .change_voters(BTreeMap::from([
                (id, BasicNode::new("node")),
                (NodeId::from(97), BasicNode::new("new-node")),
            ]))
            .await;
        assert!(
            matches!(admission, Err(ConsensusWriteError::Membership(reason)) if reason == "unsupported remote decoder")
        );
        assert_eq!(node.committed_member_ids(), before);
    }

    #[tokio::test]
    async fn network_rejects_wrong_endpoint_for_expected_node() {
        let network = InMemoryNetworkFactory::default();
        let root = TempDir::new().unwrap();
        let id = NodeId::from(9);
        let node = ConsensusNode::start(
            root.path(),
            "cluster-runtime-identity-test",
            id,
            network.clone(),
        )
        .await
        .unwrap();
        network.register(id, BasicNode::new("memory://actual"), node.raft().clone());
        let mut factory = network;
        let client = factory
            .new_client(id, &BasicNode::new("memory://forged"))
            .await;
        assert!(client.remote::<RaftError<NodeId>>().is_err());
        let service = crate::consensus_network::ControlServiceHandle::new(
            Arc::new(FailingCaller),
            id,
            crate::consensus_network::global_consensus_nodes(),
        );
        assert!(matches!(
            service.active(),
            Err(ControlServiceError::RuntimeUnavailable { .. })
        ));
        node.shutdown().await.unwrap();
    }
}