omnipaxos/
messages.rs

1use crate::{
2    messages::{ballot_leader_election::BLEMessage, sequence_paxos::PaxosMessage},
3    storage::Entry,
4    util::NodeId,
5};
6#[cfg(feature = "serde")]
7use serde::{Deserialize, Serialize};
8
9/// Internal component for log replication
10pub mod sequence_paxos {
11    use crate::{
12        ballot_leader_election::Ballot,
13        storage::{Entry, SnapshotType, StopSign},
14        util::{NodeId, SequenceNumber},
15    };
16    #[cfg(feature = "serde")]
17    use serde::{Deserialize, Serialize};
18    use std::fmt::Debug;
19
20    /// Message sent by a follower on crash-recovery or dropped messages to request its leader to re-prepare them.
21    #[derive(Copy, Clone, Debug)]
22    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
23    pub struct PrepareReq {
24        /// The current round.
25        pub n: Ballot,
26    }
27
28    /// Prepare message sent by a newly-elected leader to initiate the Prepare phase.
29    #[derive(Copy, Clone, Debug)]
30    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
31    pub struct Prepare {
32        /// The current round.
33        pub n: Ballot,
34        /// The decided index of this leader.
35        pub decided_idx: u64,
36        /// The latest round in which an entry was accepted.
37        pub n_accepted: Ballot,
38        /// The log length of this leader.
39        pub accepted_idx: u64,
40    }
41
42    /// Promise message sent by a follower in response to a [`Prepare`] sent by the leader.
43    #[derive(Clone, Debug)]
44    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
45    pub struct Promise<T>
46    where
47        T: Entry,
48    {
49        /// The current round.
50        pub n: Ballot,
51        /// The latest round in which an entry was accepted.
52        pub n_accepted: Ballot,
53        /// The decided snapshot.
54        pub decided_snapshot: Option<SnapshotType<T>>,
55        /// The log suffix.
56        pub suffix: Vec<T>,
57        /// The decided index of this follower.
58        pub decided_idx: u64,
59        /// The log length of this follower.
60        pub accepted_idx: u64,
61        /// The StopSign accepted by this follower
62        pub stopsign: Option<StopSign>,
63    }
64
65    /// AcceptSync message sent by the leader to synchronize the logs of all replicas in the prepare phase.
66    #[derive(Clone, Debug)]
67    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
68    pub struct AcceptSync<T>
69    where
70        T: Entry,
71    {
72        /// The current round.
73        pub n: Ballot,
74        /// The sequence number of this message in the leader-to-follower accept sequence
75        pub seq_num: SequenceNumber,
76        /// The decided snapshot.
77        pub decided_snapshot: Option<SnapshotType<T>>,
78        /// The log suffix.
79        pub suffix: Vec<T>,
80        /// The index of the log where the entries from `suffix` should be applied at (also the compacted idx of `decided_snapshot` if it exists)
81        pub sync_idx: u64,
82        /// The decided index
83        pub decided_idx: u64,
84        /// StopSign to be accepted
85        pub stopsign: Option<StopSign>,
86        #[cfg(feature = "unicache")]
87        /// The UniCache of the leader
88        pub unicache: T::UniCache,
89    }
90
91    /// Message with entries to be replicated and the latest decided index sent by the leader in the accept phase.
92    #[derive(Clone, Debug)]
93    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
94    pub struct AcceptDecide<T>
95    where
96        T: Entry,
97    {
98        /// The current round.
99        pub n: Ballot,
100        /// The sequence number of this message in the leader-to-follower accept sequence
101        pub seq_num: SequenceNumber,
102        /// The decided index.
103        pub decided_idx: u64,
104        /// Entries to be replicated.
105        pub entries: Vec<T>,
106    }
107
108    /// TODO
109    #[derive(Clone, Debug)]
110    #[cfg(feature = "unicache")]
111    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
112    pub struct EncodedAcceptDecide<T>
113    where
114        T: Entry,
115    {
116        /// The current round.
117        pub n: Ballot,
118        /// The sequence number of this message in the leader-to-follower accept sequence
119        pub seq_num: SequenceNumber,
120        /// The decided index.
121        pub decided_idx: u64,
122        /// Entries to be replicated.
123        pub entries: Vec<T::EncodeResult>,
124    }
125
126    /// Message sent by follower to leader when entries has been accepted.
127    #[derive(Copy, Clone, Debug)]
128    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
129    pub struct Accepted {
130        /// The current round.
131        pub n: Ballot,
132        /// The accepted index.
133        pub accepted_idx: u64,
134    }
135
136    /// Message sent by leader to followers to decide up to a certain index in the log.
137    #[derive(Copy, Clone, Debug)]
138    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
139    pub struct Decide {
140        /// The current round.
141        pub n: Ballot,
142        /// The sequence number of this message in the leader-to-follower accept sequence
143        pub seq_num: SequenceNumber,
144        /// The decided index.
145        pub decided_idx: u64,
146    }
147
148    /// Message sent by leader to followers to accept a StopSign
149    #[derive(Clone, Debug)]
150    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
151    pub struct AcceptStopSign {
152        /// The current round.
153        pub n: Ballot,
154        /// The sequence number of this message in the leader-to-follower accept sequence
155        pub seq_num: SequenceNumber,
156        /// The decided index.
157        pub ss: StopSign,
158    }
159
160    /// Message sent by follower to leader when accepting an entry is rejected.
161    /// This happens when the follower is promised to a greater leader.
162    #[derive(Clone, Debug)]
163    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
164    pub struct NotAccepted {
165        /// The follower's current ballot
166        pub n: Ballot,
167    }
168
169    /// Compaction Request
170    #[allow(missing_docs)]
171    #[derive(Clone, Debug)]
172    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
173    pub enum Compaction {
174        Trim(u64),
175        Snapshot(Option<u64>),
176    }
177
178    /// An enum for all the different message types.
179    #[allow(missing_docs)]
180    #[derive(Clone, Debug)]
181    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
182    pub enum PaxosMsg<T>
183    where
184        T: Entry,
185    {
186        /// Request a [`Prepare`] to be sent from the leader. Used for fail-recovery.
187        PrepareReq(PrepareReq),
188        #[allow(missing_docs)]
189        Prepare(Prepare),
190        Promise(Promise<T>),
191        AcceptSync(AcceptSync<T>),
192        AcceptDecide(AcceptDecide<T>),
193        Accepted(Accepted),
194        NotAccepted(NotAccepted),
195        Decide(Decide),
196        /// Forward client proposals to the leader.
197        ProposalForward(Vec<T>),
198        Compaction(Compaction),
199        AcceptStopSign(AcceptStopSign),
200        ForwardStopSign(StopSign),
201        #[cfg(feature = "unicache")]
202        EncodedAcceptDecide(EncodedAcceptDecide<T>),
203    }
204
205    /// A struct for a Paxos message that also includes sender and receiver.
206    #[derive(Clone, Debug)]
207    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
208    pub struct PaxosMessage<T>
209    where
210        T: Entry,
211    {
212        /// Sender of `msg`.
213        pub from: NodeId,
214        /// Receiver of `msg`.
215        pub to: NodeId,
216        /// The message content.
217        pub msg: PaxosMsg<T>,
218    }
219}
220
221/// The different messages BLE uses to communicate with other servers.
222pub mod ballot_leader_election {
223
224    use crate::{ballot_leader_election::Ballot, util::NodeId};
225    #[cfg(feature = "serde")]
226    use serde::{Deserialize, Serialize};
227
228    /// An enum for all the different BLE message types.
229    #[allow(missing_docs)]
230    #[derive(Clone, Debug)]
231    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
232    pub enum HeartbeatMsg {
233        Request(HeartbeatRequest),
234        Reply(HeartbeatReply),
235    }
236
237    /// Requests a reply from all the other servers.
238    #[derive(Clone, Debug)]
239    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
240    pub struct HeartbeatRequest {
241        /// Number of the current round.
242        pub round: u32,
243    }
244
245    /// Replies
246    #[derive(Clone, Debug)]
247    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
248    pub struct HeartbeatReply {
249        /// Number of the current heartbeat round.
250        pub round: u32,
251        /// Ballot of replying server.
252        pub ballot: Ballot,
253        /// Leader this server is following
254        pub leader: Ballot,
255        /// Whether the replying server sees a need for a new leader
256        pub happy: bool,
257    }
258
259    /// A struct for a Paxos message that also includes sender and receiver.
260    #[derive(Clone, Debug)]
261    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
262    pub struct BLEMessage {
263        /// Sender of `msg`.
264        pub from: NodeId,
265        /// Receiver of `msg`.
266        pub to: NodeId,
267        /// The message content.
268        pub msg: HeartbeatMsg,
269    }
270}
271
272#[allow(missing_docs)]
273/// Message in OmniPaxos. Can be either a `SequencePaxos` message (for log replication) or `BLE` message (for leader election)
274#[derive(Clone, Debug)]
275#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
276pub enum Message<T>
277where
278    T: Entry,
279{
280    SequencePaxos(PaxosMessage<T>),
281    BLE(BLEMessage),
282}
283
284impl<T> Message<T>
285where
286    T: Entry,
287{
288    /// Get the sender id of the message
289    pub fn get_sender(&self) -> NodeId {
290        match self {
291            Message::SequencePaxos(p) => p.from,
292            Message::BLE(b) => b.from,
293        }
294    }
295
296    /// Get the receiver id of the message
297    pub fn get_receiver(&self) -> NodeId {
298        match self {
299            Message::SequencePaxos(p) => p.to,
300            Message::BLE(b) => b.to,
301        }
302    }
303}
omnipaxos/messages.rs

omnipaxos/
messages.rs
