[WIP] Prevent unbounded Raft log growth.

src/config.rs

@@ -30,6 +30,7 @@ use super::{
  errors::{Error, Result},
  INVALID_ID,
 };
+use raft_log::NO_SIZE_LIMIT;
 
 /// Config contains the parameters to start a raft.
 pub struct Config {
@@ -109,6 +110,11 @@ pub struct Config {
 
  /// A human-friendly tag used for logging.
  pub tag: String,
+
+ /// Limits the aggregate byte size of the uncommitted entries that may be appended to a leader's
+ /// log. Once this limit is exceeded, proposals will begin to return ProposalDropped errors.
+ /// Defaults to no limit.
+ pub max_uncommitted_entries_size: usize,
 }
 
 impl Default for Config {
@@ -130,6 +136,7 @@ impl Default for Config {
  read_only_option: ReadOnlyOption::Safe,
  skip_bcast_commit: false,
  tag: "".into(),
+ max_uncommitted_entries_size: NO_SIZE_LIMIT,
  }
  }
 }
@@ -204,6 +211,12 @@ impl Config {
  ));
  }
 
+ if self.max_uncommitted_entries_size == 0 {
+ return Err(Error::ConfigInvalid(
+ "max uncommitted entries size must be greater than 0".to_owned(),
+ ));
+ }
+
  Ok(())
  }
 }

src/lib.rs

@@ -301,7 +301,7 @@ pub use self::progress::{Inflights, Progress, ProgressSet, ProgressState};
 pub use self::raft::{
  quorum, vote_resp_msg_type, Raft, SoftState, StateRole, INVALID_ID, INVALID_INDEX,
 };
-pub use self::raft_log::{RaftLog, NO_LIMIT};
+pub use self::raft_log::{RaftLog, NO_LIMIT, NO_SIZE_LIMIT};
 pub use self::raw_node::{is_empty_snap, Peer, RawNode, Ready, SnapshotStatus};
 pub use self::read_only::{ReadOnlyOption, ReadState};
 pub use self::status::Status;

src/raft.rs

@@ -178,6 +178,11 @@ pub struct Raft<T: Storage> {
 
  /// Tag is only used for logging
  tag: String,
+
+ /// Limits the aggregate byte size of the uncommitted entries that may be appended to a leader's
+ /// log. Once this limit is exceeded, proposals will begin to return ProposalDropped errors.
+ /// Defaults to no limit.
+ max_uncommitted_entries_size: usize,
 }
 
 trait AssertSend: Send {}
@@ -258,6 +263,7 @@ impl<T: Storage> Raft<T> {
  max_election_timeout: c.max_election_tick(),
  skip_bcast_commit: c.skip_bcast_commit,
  tag: c.tag.to_owned(),
+ max_uncommitted_entries_size: c.max_uncommitted_entries_size,
  };
  for p in peers {
  let pr = Progress::new(1, r.max_inflight);
@@ -381,6 +387,14 @@ impl<T: Storage> Raft<T> {
  self.skip_bcast_commit = skip;
  }
 
+ #[inline]
+ fn max_uncommitted_size(&self) -> usize {
+ if self.state == StateRole::Leader {
+ return self.max_uncommitted_entries_size;
+ }
+ raft_log::NO_SIZE_LIMIT
+ }
+
  // send persists state to stable storage and then sends to its mailbox.
  fn send(&mut self, mut m: Message) {
  m.set_from(self.id);
@@ -625,20 +639,22 @@ impl<T: Storage> Raft<T> {
 
  /// Appends a slice of entries to the log. The entries are updated to match
  /// the current index and term.
- pub fn append_entry(&mut self, es: &mut [Entry]) {
+ pub fn append_entry(&mut self, es: &mut [Entry]) -> Result<()> {
  let mut li = self.raft_log.last_index();
  for (i, e) in es.iter_mut().enumerate() {
  e.set_term(self.term);
  e.set_index(li + 1 + i as u64);
  }
+ let max = self.max_uncommitted_size();
  // use latest "last" index after truncate/append
- li = self.raft_log.append(es);
+ li = self.raft_log.append(es, max)?;
 
  let self_id = self.id;
  self.mut_prs().get_mut(self_id).unwrap().maybe_update(li);
 
  // Regardless of maybe_commit's return, our caller will call bcastAppend.
  self.maybe_commit();
+ Ok(())
  }
 
  /// Returns true to indicate that there will probably be some readiness need to be handled.
@@ -763,8 +779,6 @@ impl<T: Storage> Raft<T> {
  );
  let term = self.term;
  self.reset(term);
- self.leader_id = self.id;
- self.state = StateRole::Leader;
 
  // Conservatively set the pending_conf_index to the last index in the
  // log. There may or may not be a pending config change, but it's
@@ -773,7 +787,13 @@ impl<T: Storage> Raft<T> {
  // could be expensive.
  self.pending_conf_index = self.raft_log.last_index();
 
- self.append_entry(&mut [Entry::new()]);
+ // This unwrap is safe, because append_entry only returns a ProposalDropped
+ // error if self.state is set to Leader, which it is not, yet.
+ self.append_entry(&mut [Entry::new()]).unwrap();
+
+ self.leader_id = self.id;
+ self.state = StateRole::Leader;
+
  info!("{} became leader at term {}", self.tag, self.term);
  }
 
@@ -1404,7 +1424,7 @@ impl<T: Storage> Raft<T> {
  }
  }
  }
- self.append_entry(&mut m.mut_entries());
+ self.append_entry(&mut m.mut_entries())?;
  self.bcast_append();
  return Ok(());
  }
@@ -1680,6 +1700,7 @@ impl<T: Storage> Raft<T> {
  m.get_log_term(),
  m.get_commit(),
  m.get_entries(),
+ raft_log::NO_SIZE_LIMIT,
  ) {
  Some(mlast_index) => {
  to_send.set_index(mlast_index);