nodedb-cluster 0.2.0

// SPDX-License-Identifier: BUSL-1.1

//! Raft configuration-change propose/apply with learner semantics.
//!
//! `propose_conf_change` writes a `ConfChange` payload (see
//! `crate::conf_change::ConfChange`) into the group leader's Raft log as a
//! regular entry with a special prefix byte. The entry replicates via the
//! normal `AppendEntries` channel; no new transport is needed.
//!
//! `apply_conf_change` is called by the tick loop when a committed entry
//! is identified as a conf change. It updates both the in-memory
//! `RaftNode` peer set and the `RoutingTable`:
//!
//! - `AddNode` → voter added to `RaftNode.peers` and `routing.members`.
//! - `RemoveNode` → voter removed from both.
//! - `AddLearner` → learner added to `RaftNode.learners` and `routing.learners`.
//! - `PromoteLearner` → learner moved from `learners` to `members` in both;
//!   if the promoted peer is *this* node, also flips the local role from
//!   `Learner` to `Follower`.

use tracing::debug;

use crate::conf_change::{ConfChange, ConfChangeType};
use crate::error::{ClusterError, Result};

use super::core::MultiRaft;

impl MultiRaft {
    /// Propose a configuration change to a Raft group.
    ///
    /// The change is serialized into the group's Raft log as a
    /// regular entry with a distinguishing prefix byte. It
    /// replicates through the normal `AppendEntries` path and is
    /// applied by every follower replica when the entry commits
    /// (see `apply_conf_change`).
    ///
    /// # Single-voter vs. multi-voter groups
    ///
    /// Single-voter groups commit inside `node.propose` itself
    /// (see `nodedb_raft::node::RaftNode::propose` single-voter
    /// branch). In that case the commit has already happened by
    /// the time we return, so we safely apply the change inline:
    /// any caller that reads routing immediately after the
    /// propose sees the final state.
    ///
    /// Multi-voter groups commit asynchronously once enough
    /// followers have replicated the entry. The apply then
    /// happens on the tick loop after it observes the updated
    /// `commit_index`. We MUST NOT inline-apply in that case —
    /// if the leader steps down before replication completes, a
    /// new leader may truncate the log entry and the local state
    /// would be permanently ahead of the committed state with no
    /// rollback path. Callers that need to wait for the apply
    /// should poll the routing table (see
    /// `raft_loop::join::wait_for_routing_contains_learner`).
    ///
    /// Returns `(group_id, log_index)` on success.
    pub fn propose_conf_change(
        &mut self,
        group_id: u64,
        change: &ConfChange,
    ) -> Result<(u64, u64)> {
        let (log_index, committed_immediately) = {
            let node = self
                .groups
                .get_mut(&group_id)
                .ok_or(ClusterError::GroupNotFound { group_id })?;
            let data = change.to_entry_data();
            let log_index = node.propose(data)?;
            // A single-voter group self-commits inside `propose`:
            // its `commit_index` is bumped to the new `log_index`
            // before we return. Detecting this is the one safe
            // trigger for an inline apply.
            let committed_immediately = node.commit_index() >= log_index;
            (log_index, committed_immediately)
        };

        if committed_immediately {
            self.apply_conf_change(group_id, change)?;
        }
        Ok((group_id, log_index))
    }

    /// Apply a committed configuration change to this node's view of the
    /// given Raft group.
    ///
    /// This is called from the tick loop for every committed entry
    /// detected as a conf-change (via `ConfChange::from_entry_data`). It
    /// must be idempotent with respect to no-op changes so replaying the
    /// log after a crash does not double-apply.
    pub fn apply_conf_change(&mut self, group_id: u64, change: &ConfChange) -> Result<()> {
        let self_node_id = self.node_id;

        let node = self
            .groups
            .get_mut(&group_id)
            .ok_or(ClusterError::GroupNotFound { group_id })?;

        match change.change_type {
            ConfChangeType::AddNode => {
                // Direct voter add (used for legacy or bootstrap paths).
                node.add_peer(change.node_id);
                if let Some(info) = self.routing.group_info(group_id)
                    && !info.members.contains(&change.node_id)
                {
                    let mut new_members = info.members.clone();
                    new_members.push(change.node_id);
                    self.routing.set_group_members(group_id, new_members);
                }
            }
            ConfChangeType::RemoveNode => {
                node.remove_peer(change.node_id);
                if let Some(info) = self.routing.group_info(group_id) {
                    let new_members: Vec<u64> = info
                        .members
                        .iter()
                        .copied()
                        .filter(|&id| id != change.node_id)
                        .collect();
                    self.routing.set_group_members(group_id, new_members);
                }
            }
            ConfChangeType::AddLearner => {
                // Non-voting add: peer enters learners on both the
                // RaftNode and the routing table. Voting quorum does not
                // change.
                node.add_learner(change.node_id);
                self.routing.add_group_learner(group_id, change.node_id);
            }
            ConfChangeType::PromoteLearner => {
                // Learner → voter. RaftNode and routing both update.
                // If this is our own promotion, we also need to flip the
                // local role from `Learner` to `Follower` so subsequent
                // ticks run election timeouts normally.
                let promoted = node.promote_learner(change.node_id);
                if promoted {
                    self.routing.promote_group_learner(group_id, change.node_id);
                }
                if change.node_id == self_node_id {
                    node.promote_self_to_voter();
                }
            }
        }

        debug!(
            node = self.node_id,
            group = group_id,
            change_type = ?change.change_type,
            target_node = change.node_id,
            voters = ?self.groups.get(&group_id).map(|n| n.voters().to_vec()),
            learners = ?self.groups.get(&group_id).map(|n| n.learners().to_vec()),
            "applied conf change"
        );

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::routing::RoutingTable;
    use nodedb_raft::NodeRole;

    use super::super::core::MultiRaft;

    fn new_mr(node_id: u64, group_ids: &[u64]) -> MultiRaft {
        let dir = tempfile::tempdir().unwrap();
        let rt = RoutingTable::uniform(group_ids.len() as u64, &[node_id], 1);
        let mut mr = MultiRaft::new(node_id, rt, dir.path().to_path_buf());
        std::mem::forget(dir); // Keep temp dir alive for the duration of the test.
        for &gid in group_ids {
            mr.add_group(gid, vec![]).unwrap();
        }
        mr
    }

    #[test]
    fn apply_add_learner_updates_routing_and_raftnode() {
        let mut mr = new_mr(1, &[0]);
        let change = ConfChange {
            change_type: ConfChangeType::AddLearner,
            node_id: 2,
        };
        mr.apply_conf_change(0, &change).unwrap();

        // RaftNode: learner tracked, voters unchanged.
        let node = mr.groups.get(&0).unwrap();
        assert_eq!(node.learners(), &[2]);
        assert!(node.voters().is_empty());

        // Routing: learners populated, members untouched.
        let info = mr.routing.group_info(0).unwrap();
        assert_eq!(info.learners, vec![2]);
        assert_eq!(info.members, vec![1]); // Self.
    }

    #[test]
    fn apply_promote_learner_moves_peer_to_voters() {
        let mut mr = new_mr(1, &[0]);
        mr.apply_conf_change(
            0,
            &ConfChange {
                change_type: ConfChangeType::AddLearner,
                node_id: 2,
            },
        )
        .unwrap();
        mr.apply_conf_change(
            0,
            &ConfChange {
                change_type: ConfChangeType::PromoteLearner,
                node_id: 2,
            },
        )
        .unwrap();

        let node = mr.groups.get(&0).unwrap();
        assert_eq!(node.voters(), &[2]);
        assert!(node.learners().is_empty());

        let info = mr.routing.group_info(0).unwrap();
        assert_eq!(info.learners, Vec::<u64>::new());
        assert!(info.members.contains(&2));
    }

    #[test]
    fn apply_promote_self_flips_role() {
        // Simulate receiving PromoteLearner(self=2) after being added as
        // a learner to group 0.
        let dir = tempfile::tempdir().unwrap();
        let rt = RoutingTable::uniform(1, &[1, 2], 1);
        let mut mr = MultiRaft::new(2, rt, dir.path().to_path_buf());
        mr.add_group_as_learner(0, vec![1], vec![]).unwrap();

        // Inject ourselves into the learners list so promote_learner has
        // something to find. (In the real flow this happens via
        // `AddLearner` applied from the log; we short-circuit for the
        // unit test.)
        mr.groups.get_mut(&0).unwrap().add_learner(2);
        // Technically `add_learner(self_id)` is a no-op guard — force
        // config.learners manually via promoting through a faux path:
        // re-apply AddLearner from apply_conf_change, which tolerates
        // self-id collision.
        //
        // For this test, the simpler route is to construct a tiny fake
        // and check that `promote_self_to_voter` is called on self.
        // Since the guard in add_learner skips self, we can't stage that
        // state cleanly. Instead we directly verify the role flip path:
        let node = mr.groups.get_mut(&0).unwrap();
        assert_eq!(node.role(), NodeRole::Learner);
        node.promote_self_to_voter();
        assert_eq!(node.role(), NodeRole::Follower);
    }
}