openraft 0.10.0-alpha.18

//! [`RaftLeaderId`] implementation that enforces standard Raft behavior of at most one leader per
//! term.

use std::cmp::Ordering;
use std::fmt;
use std::ops::Deref;
use std::ops::DerefMut;

use display_more::DisplayOptionExt;

use crate::NodeId;
use crate::vote::LeaderIdCompare;
use crate::vote::RaftLeaderId;
use crate::vote::RaftTerm;

/// ID of a `leader`, enforcing a single leader per term.
///
/// It includes the `term` and the `node_id`.
///
/// Raft specifies that in a term there is at most one leader, thus Leader ID is partially order as
/// defined below.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize), serde(bound = ""))]
pub struct LeaderId<Term, NID>
where
    Term: RaftTerm,
    NID: NodeId,
{
    /// The term of the leader.
    pub term: Term,

    /// The node ID that was voted for in this term.
    pub voted_for: Option<NID>,
}

impl<Term, NID> PartialOrd for LeaderId<Term, NID>
where
    Term: RaftTerm,
    NID: NodeId,
{
    #[inline]
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        LeaderIdCompare::std(self, other)
    }
}

impl<Term, NID> fmt::Display for LeaderId<Term, NID>
where
    Term: RaftTerm,
    NID: NodeId,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "T{}-N{}", self.term, self.voted_for.display())
    }
}

impl<Term, NID> PartialEq<CommittedLeaderId<Term>> for LeaderId<Term, NID>
where
    Term: RaftTerm,
    NID: NodeId,
{
    fn eq(&self, _other: &CommittedLeaderId<Term>) -> bool {
        // Committed and non-committed are never equal
        false
    }
}

impl<Term, NID> PartialOrd<CommittedLeaderId<Term>> for LeaderId<Term, NID>
where
    Term: RaftTerm,
    NID: NodeId,
{
    fn partial_cmp(&self, other: &CommittedLeaderId<Term>) -> Option<Ordering> {
        if self.term == other.term {
            // For the same term, committed Leader overrides non-committed
            Some(Ordering::Less)
        } else {
            self.term.partial_cmp(&other.term)
        }
    }
}

/// Reciprocal comparison: `CommittedLeaderId` compared with `LeaderId`.
///
/// Not required by [`RaftLeaderId`] trait bound, but provides symmetric comparison semantics.
impl<Term, NID> PartialEq<LeaderId<Term, NID>> for CommittedLeaderId<Term>
where
    Term: RaftTerm,
    NID: NodeId,
{
    fn eq(&self, _other: &LeaderId<Term, NID>) -> bool {
        false
    }
}

/// Reciprocal comparison: `CommittedLeaderId` compared with `LeaderId`.
///
/// Not required by [`RaftLeaderId`] trait bound, but provides symmetric comparison semantics.
impl<Term, NID> PartialOrd<LeaderId<Term, NID>> for CommittedLeaderId<Term>
where
    Term: RaftTerm,
    NID: NodeId,
{
    fn partial_cmp(&self, other: &LeaderId<Term, NID>) -> Option<Ordering> {
        if self.term == other.term {
            Some(Ordering::Greater)
        } else {
            self.term.partial_cmp(&other.term)
        }
    }
}

impl<Term, NID> RaftLeaderId for LeaderId<Term, NID>
where
    Term: RaftTerm,
    NID: NodeId,
{
    type Term = Term;
    type NodeId = NID;
    type Committed = CommittedLeaderId<Term>;

    fn new(term: Term, node_id: NID) -> Self {
        Self {
            term,
            voted_for: Some(node_id),
        }
    }

    fn term(&self) -> Term {
        self.term
    }

    fn node_id(&self) -> &NID {
        self.voted_for.as_ref().unwrap()
    }

    fn to_committed(&self) -> Self::Committed {
        CommittedLeaderId::new(self.term)
    }
}

/// The unique identifier of a leader that is already granted by a quorum in phase-1(voting).
///
/// [`CommittedLeaderId`] contains less information than [`LeaderId`], because it implies the
/// constraint that **a quorum has granted it**.
///
/// For a partial order `LeaderId`, we know that all the committed leader-id must be a total order
/// set. Therefore, once it is granted by a quorum, it only keeps the information that makes
/// leader-ids a correct total order set, e.g., in standard raft, `voted_for: Option<node_id>` can
/// be removed from `(term, voted_for)` once it is granted. This is why standard Raft stores just a
/// `term` in log entry to identify the Leader proposing the log entry.
///
/// In std mode, `CommittedLeaderId` is just a wrapper of `Term`, which is an integer in most
/// cases (u64, u32, u16, u8, i64, i32, i16, i8).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(PartialOrd, Ord)]
#[derive(derive_more::Display)]
#[display("{}", term)]
#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize), serde(bound = ""))]
#[cfg_attr(feature = "serde", serde(transparent))]
#[repr(transparent)]
pub struct CommittedLeaderId<Term>
where Term: RaftTerm
{
    /// The term of the committed leader.
    pub term: Term,
}

impl<Term> CommittedLeaderId<Term>
where Term: RaftTerm
{
    /// Create a new committed leader ID for the given term.
    pub fn new(term: Term) -> Self {
        Self { term }
    }
}

impl<Term> Deref for CommittedLeaderId<Term>
where Term: RaftTerm
{
    type Target = Term;

    fn deref(&self) -> &Self::Target {
        &self.term
    }
}

impl<Term> DerefMut for CommittedLeaderId<Term>
where Term: RaftTerm
{
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.term
    }
}

#[rustfmt::skip]
mod impl_from_int {
    use crate::vote::leader_id_std::CommittedLeaderId;

    impl From<u64> for CommittedLeaderId<u64> {fn from(term: u64) -> Self {Self {term}}}
    impl From<u32> for CommittedLeaderId<u32> {fn from(term: u32) -> Self {Self {term}}}
    impl From<u16> for CommittedLeaderId<u16> {fn from(term: u16) -> Self {Self {term}}}
    impl From<u8>  for CommittedLeaderId<u8>  {fn from(term: u8)  -> Self {Self {term}}}
    impl From<i64> for CommittedLeaderId<i64> {fn from(term: i64) -> Self {Self {term}}}
    impl From<i32> for CommittedLeaderId<i32> {fn from(term: i32) -> Self {Self {term}}}
    impl From<i16> for CommittedLeaderId<i16> {fn from(term: i16) -> Self {Self {term}}}
    impl From<i8>  for CommittedLeaderId<i8>  {fn from(term: i8)  -> Self {Self {term}}}
}

#[cfg(test)]
#[allow(clippy::nonminimal_bool)]
mod tests {
    use super::LeaderId;
    use crate::vote::RaftLeaderId;

    #[cfg(feature = "serde")]
    #[test]
    fn test_committed_leader_id_serde() -> anyhow::Result<()> {
        use super::CommittedLeaderId;

        let c = CommittedLeaderId::<u64>::new(5);
        let s = serde_json::to_string(&c)?;
        assert_eq!(r#"5"#, s);

        let c2: CommittedLeaderId<u64> = serde_json::from_str(&s)?;
        assert_eq!(CommittedLeaderId::new(5), c2);

        Ok(())
    }

    #[test]
    #[allow(clippy::neg_cmp_op_on_partial_ord)]
    fn test_std_leader_id_partial_order() -> anyhow::Result<()> {
        #[allow(clippy::redundant_closure)]
        let lid = |term, node_id| LeaderId::<u64, u64>::new(term, node_id);

        // Compare term first
        assert!(lid(2, 2) > lid(1, 2));
        assert!(lid(1, 2) < lid(2, 2));

        // Equal
        assert!(lid(2, 2) == lid(2, 2));
        assert!(lid(2, 2) >= lid(2, 2));
        assert!(lid(2, 2) <= lid(2, 2));

        // Incomparable
        assert!(!(lid(2, 2) > lid(2, 3)));
        assert!(!(lid(2, 2) < lid(2, 3)));
        assert!(!(lid(2, 2) == lid(2, 3)));

        Ok(())
    }

    #[test]
    #[allow(clippy::neg_cmp_op_on_partial_ord)]
    fn test_leader_id_vs_committed_partial_order() -> anyhow::Result<()> {
        use super::CommittedLeaderId;

        let lid = |term, node_id| LeaderId::<u64, u64>::new(term, node_id);
        let clid = |term| CommittedLeaderId::<u64>::new(term);

        // PartialEq: LeaderId and CommittedLeaderId are never equal
        assert!(lid(2, 2) != clid(2));

        // Same term: CommittedLeaderId > LeaderId
        assert!(lid(2, 2) < clid(2));
        assert!(!(lid(2, 2) > clid(2)));

        // Different terms: compare by term
        assert!(lid(1, 2) < clid(2));
        assert!(lid(3, 2) > clid(2));

        Ok(())
    }

    #[test]
    #[allow(clippy::neg_cmp_op_on_partial_ord)]
    fn test_committed_vs_leader_id_partial_order() -> anyhow::Result<()> {
        use super::CommittedLeaderId;

        let lid = |term, node_id| LeaderId::<u64, u64>::new(term, node_id);
        let clid = |term| CommittedLeaderId::<u64>::new(term);

        // PartialEq: CommittedLeaderId and LeaderId are never equal (symmetric)
        assert!(clid(2) != lid(2, 2));
        assert!(clid(2) != lid(2, 5));

        // Same term: CommittedLeaderId > LeaderId (symmetric)
        assert!(clid(2) > lid(2, 2));
        assert!(!(clid(2) < lid(2, 2)));
        assert!(!(clid(2) == lid(2, 2)));

        // Different terms: compare by term (symmetric)
        assert!(clid(2) > lid(1, 2));
        assert!(clid(2) < lid(3, 2));
        assert!(!(clid(2) > lid(3, 2)));
        assert!(!(clid(2) == lid(1, 2)));

        Ok(())
    }

    #[test]
    fn test_committed_leader_id_deref() -> anyhow::Result<()> {
        use super::CommittedLeaderId;

        let clid = CommittedLeaderId::<u64>::new(5);
        assert_eq!(5, *clid);

        let term_ref: &u64 = &clid;
        assert_eq!(&5, term_ref);

        Ok(())
    }

    #[test]
    fn test_committed_leader_id_deref_mut() -> anyhow::Result<()> {
        use super::CommittedLeaderId;

        let mut clid = CommittedLeaderId::<u64>::new(5);
        *clid = 10;
        assert_eq!(10, *clid);
        assert_eq!(CommittedLeaderId::new(10), clid);

        Ok(())
    }

    #[test]
    fn test_committed_leader_id_from_int() -> anyhow::Result<()> {
        use super::CommittedLeaderId;

        macro_rules! test_from {
            ($term_type:ty, $value:expr) => {{
                let clid: CommittedLeaderId<$term_type> = $value.into();
                assert_eq!(CommittedLeaderId::new($value), clid);

                let clid = CommittedLeaderId::<$term_type>::from($value);
                assert_eq!(CommittedLeaderId::new($value), clid);
            }};
        }

        test_from!(u64, 5u64);
        test_from!(u32, 5u32);
        test_from!(u16, 5u16);
        test_from!(u8, 5u8);
        test_from!(i64, 5i64);
        test_from!(i32, 5i32);
        test_from!(i16, 5i16);
        test_from!(i8, 5i8);

        Ok(())
    }
}