Add new message type MsgGroupBroadcast and corresponding handler

harness/tests/test_util/mod.rs

@@ -86,6 +86,26 @@ pub fn new_test_raft_with_prevote(
     new_test_raft_with_config(&config, storage, l)
 }
 
+pub fn new_test_raft_with_follower_replication(
+    id: u64,
+    peers: Vec<u64>,
+    election: usize,
+    heartbeat: usize,
+    storage: MemStorage,
+    follower_replication: bool,
+    l: &Logger,
+) -> Interface {
+    let mut config = new_test_config(id, election, heartbeat);
+    config.follower_replication = follower_replication;
+    if storage.initial_state().unwrap().initialized() && peers.is_empty() {
+        panic!("new_test_raft with empty peers on initialized store");
+    }
+    if !peers.is_empty() && !storage.initial_state().unwrap().initialized() {
+        storage.initialize_with_conf_state((peers, vec![]));
+    }
+    new_test_raft_with_config(&config, storage, l)
+}
+
 pub fn new_test_raft_with_logs(
     id: u64,
     peers: Vec<u64>,
@@ -172,6 +192,14 @@ pub fn new_snapshot(index: u64, term: u64, voters: Vec<u64>) -> Snapshot {
     s
 }
 
+pub fn new_forward(to: u64, term: u64, index: u64) -> Forward {
+    let mut f = Forward::default();
+    f.to = to;
+    f.log_term = term;
+    f.index = index;
+    f
+}
+
 pub fn conf_change(ty: ConfChangeType, node_id: u64) -> ConfChange {
     let mut cc = ConfChange::default();
     cc.node_id = node_id;

proto/proto/eraftpb.proto

@@ -46,6 +46,17 @@ message Snapshot {
     SnapshotMetadata metadata = 2;
 }
 
+// Forward is a type that tells the agent how to forward the MsgGroupBroadcast from the leader.
+//
+// Field to is the destination of forwarding.
+// Field log_term and index is the previous entry of log entries that should be forwarded.
+// Entries to be forwarded is the range (index, last_index].
+message Forward {
+    uint64 to = 1;
+    uint64 log_term = 2;
+    uint64 index = 3;
+}
+
 enum MessageType {
     MsgHup = 0;
     MsgBeat = 1;
@@ -66,6 +77,8 @@ enum MessageType {
     MsgReadIndexResp = 16;
     MsgRequestPreVote = 17;
     MsgRequestPreVoteResponse = 18;
+    MsgGroupBroadcast = 19;
+    MsgGroupBroadcastResponse = 20;
 }
 
 message Message {
@@ -89,6 +102,7 @@ message Message {
     uint64 reject_hint = 11;
     bytes context = 12;
     uint64 priority = 14;
+    repeated Forward forwards = 16;
 }
 
 message HardState {

src/config.rs

@@ -67,6 +67,10 @@ pub struct Config {
     /// rejoins the cluster.
     pub pre_vote: bool,
 
+    /// Enables follower replication.
+    /// This reduces the across-AZ traffic of cloud deployment.
+    pub follower_replication: bool,
+
     /// The range of election timeout. In some cases, we hope some nodes has less possibility
     /// to become leader. This configuration ensures that the randomized election_timeout
     /// will always be suit in [min_election_tick, max_election_tick).
@@ -112,6 +116,7 @@ impl Default for Config {
             max_inflight_msgs: 256,
             check_quorum: false,
             pre_vote: false,
+            follower_replication: false,
             min_election_tick: 0,
             max_election_tick: 0,
             read_only_option: ReadOnlyOption::Safe,