openraft 0.9.22

Advanced Raft consensus
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

use std::fmt;

use crate::display_ext::DisplayOptionExt;
use crate::replication::request_id::RequestId;
use crate::replication::ReplicationSessionId;
use crate::type_config::alias::InstantOf;
use crate::type_config::alias::LogIdOf;
use crate::MessageSummary;
use crate::RaftTypeConfig;
use crate::StorageError;
use crate::Vote;

/// The response of replication command.
#[derive(Debug)]
pub(crate) enum Response<C>
where C: RaftTypeConfig
{
    // /// Logs that are submitted to append has been persisted to disk.
    // LogPersisted {},
    /// Update the `matched` log id of a replication target.
    /// Sent by a replication task `ReplicationCore`.
    Progress {
        /// The ID of the target node for which the match index is to be updated.
        target: C::NodeId,

        /// The id of the subject that submit this replication action.
        request_id: RequestId,

        /// The request by this leader has been successfully handled by the target node,
        /// or an error in string.
        ///
        /// A successful result can still be log matching or log conflicting.
        /// In either case, the request is considered accepted, i.e., this leader is still valid to
        /// the target node.
        ///
        /// The result also track the time when this request is sent.
        result: Result<ReplicationResult<C>, String>,

        /// In which session this message is sent.
        ///
        /// This session id identifies a certain leader(by vote) that is replicating to a certain
        /// group of nodes.
        ///
        /// A message should be discarded if it does not match the present vote and
        /// membership_log_id.
        session_id: ReplicationSessionId<C::NodeId>,
    },

    /// [`StorageError`] error has taken place locally(not on remote node) when replicating, and
    /// [`RaftCore`](`crate::core::RaftCore`) needs to shutdown. Sent by a replication task
    /// [`crate::replication::ReplicationCore`].
    StorageError { error: StorageError<C::NodeId> },

    /// ReplicationCore has seen a higher `vote`.
    /// Sent by a replication task `ReplicationCore`.
    HigherVote {
        /// The ID of the target node from which the new term was observed.
        target: C::NodeId,

        /// The higher vote observed.
        higher: Vote<C::NodeId>,

        /// Which state(a Leader or Candidate) sent this message
        sender_vote: Vote<C::NodeId>,
        // TODO: need this?
        // /// The cluster this replication works for.
        // membership_log_id: Option<LogId<C::NodeId>>,
    },
}

impl<C> MessageSummary<Response<C>> for Response<C>
where C: RaftTypeConfig
{
    fn summary(&self) -> String {
        match self {
            Self::Progress {
                ref target,
                request_id: ref id,
                ref result,
                ref session_id,
            } => {
                format!(
                    "UpdateReplicationProgress: target: {}, id: {}, result: {:?}, session_id: {}",
                    target, id, result, session_id,
                )
            }

            Self::StorageError { error } => format!("ReplicationStorageError: {}", error),

            Self::HigherVote {
                target,
                higher,
                sender_vote,
            } => {
                format!(
                    "Seen a higher vote: target: {}, vote: {}, server_state_vote: {}",
                    target, higher, sender_vote
                )
            }
        }
    }
}

/// Result of an append-entries replication
#[derive(Clone, Debug)]
pub(crate) struct ReplicationResult<C: RaftTypeConfig> {
    /// The timestamp when this request is sent.
    ///
    /// It is used to update the lease for leader.
    pub(crate) sending_time: InstantOf<C>,

    /// Ok for matching, Err for conflict.
    pub(crate) result: Result<Option<LogIdOf<C>>, LogIdOf<C>>,
}

impl<C> fmt::Display for ReplicationResult<C>
where C: RaftTypeConfig
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{{time:{:?}, result:", self.sending_time)?;

        match &self.result {
            Ok(matching) => write!(f, "Match:{}", matching.display())?,
            Err(conflict) => write!(f, "Conflict:{}", conflict)?,
        }

        write!(f, "}}")
    }
}

impl<C> ReplicationResult<C>
where C: RaftTypeConfig
{
    pub(crate) fn new(sending_time: InstantOf<C>, result: Result<Option<LogIdOf<C>>, LogIdOf<C>>) -> Self {
        Self { sending_time, result }
    }
}

#[cfg(test)]
mod tests {
    use crate::engine::testing::UTConfig;
    use crate::replication::response::ReplicationResult;
    use crate::testing::log_id;
    use crate::type_config::TypeConfigExt;

    #[test]
    fn test_replication_result_display() {
        // NOTE that with single-term-leader, log id is `1-3`

        let result = ReplicationResult::<UTConfig>::new(UTConfig::now(), Ok(Some(log_id(1, 2, 3))));
        let want = format!(", result:Match:{}}}", log_id(1, 2, 3));
        assert!(result.to_string().ends_with(&want), "{}", result.to_string());

        let result = ReplicationResult::<UTConfig>::new(UTConfig::now(), Err(log_id(1, 2, 3)));
        let want = format!(", result:Conflict:{}}}", log_id(1, 2, 3));
        assert!(result.to_string().ends_with(&want), "{}", result.to_string());
    }
}