raft: use read/write instead of query/mutate for requests

src/raft/message.rs

@@ -104,19 +104,31 @@ pub type RequestID = Vec<u8>;
 /// A read sequence number, used to confirm leadership for linearizable reads.
 pub type ReadSequence = u64;
 
-/// A client request.
+/// A client request, typically passed through to the state machine.
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub enum Request {
- Query(Vec<u8>),
- Mutate(Vec<u8>),
+ /// A state machine read command. This is not replicated, and only evaluted
+ /// on the leader.
+ Read(Vec<u8>),
+ /// A state machine write command. This is replicated across all nodes, and
+ /// must result in a deterministic response.
+ Write(Vec<u8>),
+ /// Requests Raft cluster status from the leader.
  Status,
 }
 
-/// A client response.
+/// A client response. This will be wrapped in a Result to handle errors.
+///
+/// TODO: consider a separate error kind here, or a wrapped Result, to separate
+/// fallible state machine operations (returned to the caller) from apply errors
+/// (fatal).
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub enum Response {
- Query(Vec<u8>),
- Mutate(Vec<u8>),
+ /// A state machine read result.
+ Read(Vec<u8>),
+ /// A state machine write result.
+ Write(Vec<u8>),
+ /// The current Raft leader status.
  Status(Status),
 }
 

src/raft/node/candidate.rs

@@ -330,10 +330,7 @@ mod tests {
  from: 1,
  to: 1,
  term: 3,
- message: Message::ClientRequest {
- id: vec![0x01],
- request: Request::Mutate(vec![0xaf]),
- },
+ message: Message::ClientRequest { id: vec![0x01], request: Request::Write(vec![0xaf]) },
  })?;
  assert_node(&mut node).is_candidate().term(3);
  assert_messages(

src/raft/node/follower.rs

@@ -868,10 +868,7 @@ pub mod tests {
  from: 1,
  to: 1,
  term: 3,
- message: Message::ClientRequest {
- id: vec![0x01],
- request: Request::Mutate(vec![0xaf]),
- },
+ message: Message::ClientRequest { id: vec![0x01], request: Request::Write(vec![0xaf]) },
  })?;
  assert_node(&mut node).is_follower().term(3).leader(Some(2)).forwarded(vec![vec![0x01]]);
  assert_messages(
@@ -882,7 +879,7 @@ pub mod tests {
  term: 3,
  message: Message::ClientRequest {
  id: vec![0x01],
- request: Request::Mutate(vec![0xaf]),
+ request: Request::Write(vec![0xaf]),
  },
  }],
  );
@@ -893,7 +890,7 @@ pub mod tests {
  term: 3,
  message: Message::ClientResponse {
  id: vec![0x01],
- response: Ok(Response::Mutate(vec![0xaf])),
+ response: Ok(Response::Write(vec![0xaf])),
  },
  })?;
  assert_node(&mut node).is_follower().term(3).leader(Some(2)).forwarded(vec![]);
@@ -905,7 +902,7 @@ pub mod tests {
  term: 3,
  message: Message::ClientResponse {
  id: vec![0x01],
- response: Ok(Response::Mutate(vec![0xaf])),
+ response: Ok(Response::Write(vec![0xaf])),
  },
  }],
  );
@@ -923,10 +920,7 @@ pub mod tests {
  from: 1,
  to: 1,
  term: 3,
- message: Message::ClientRequest {
- id: vec![0x01],
- request: Request::Mutate(vec![0xaf]),
- },
+ message: Message::ClientRequest { id: vec![0x01], request: Request::Write(vec![0xaf]) },
  })?;
  assert_node(&mut node).is_follower().term(3).leader(None).forwarded(vec![]);
  assert_messages(
@@ -951,10 +945,7 @@ pub mod tests {
  from: 1,
  to: 1,
  term: 3,
- message: Message::ClientRequest {
- id: vec![0x01],
- request: Request::Mutate(vec![0xaf]),
- },
+ message: Message::ClientRequest { id: vec![0x01], request: Request::Write(vec![0xaf]) },
  })?;
  assert_node(&mut node).is_follower().term(3).leader(Some(2)).forwarded(vec![vec![0x01]]);
  assert_messages(
@@ -965,7 +956,7 @@ pub mod tests {
  term: 3,
  message: Message::ClientRequest {
  id: vec![0x01],
- request: Request::Mutate(vec![0xaf]),
+ request: Request::Write(vec![0xaf]),
  },
  }],
  );

src/raft/node/leader.rs

@@ -209,7 +209,7 @@ impl RawNode<Leader> {
  // must confirm that we are still the leader by sending a heartbeat
  // with the read's sequence number and wait for confirmation from a
  // quorum before executing the read.
- Message::ClientRequest { id, request: Request::Query(command) } => {
+ Message::ClientRequest { id, request: Request::Read(command) } => {
  self.role.read_seq += 1;
  self.role.reads.push_back(Read {
  seq: self.role.read_seq,
@@ -225,7 +225,7 @@ impl RawNode<Leader> {
 
  // A client submitted a write command. Propose it, and track it
  // until it's applied and the response is returned to the client.
- Message::ClientRequest { id, request: Request::Mutate(command) } => {
+ Message::ClientRequest { id, request: Request::Write(command) } => {
  let index = self.propose(Some(command))?;
  self.role.writes.insert(index, Write { from: msg.from, id: id.clone() });
  if self.peers.is_empty() {
@@ -340,7 +340,7 @@ impl RawNode<Leader> {
  term: self.term,
  message: Message::ClientResponse {
  id: write.id,
- response: result.map(Response::Mutate),
+ response: result.map(Response::Write),
  },
  })?;
  }
@@ -376,7 +376,7 @@ impl RawNode<Leader> {
  let result = self.state.query(read.command);
  self.send(
  read.from,
- Message::ClientResponse { id: read.id, response: result.map(Response::Query) },
+ Message::ClientResponse { id: read.id, response: result.map(Response::Read) },
  )?;
  }
 
@@ -673,7 +673,7 @@ mod tests {
  from: 1,
  to: 1,
  term: 3,
- message: Message::ClientRequest { id: vec![0x01], request: Request::Query(vec![0xaf]) },
+ message: Message::ClientRequest { id: vec![0x01], request: Request::Read(vec![0xaf]) },
  })?;
  assert_node(&mut node).is_leader().term(3).committed(2).last(5);
  for to in peers.iter().copied().sorted() {
@@ -701,10 +701,7 @@ mod tests {
  from: 1,
  to: 1,
  term: 3,
- message: Message::ClientRequest {
- id: vec![0x01],
- request: Request::Mutate(vec![0xaf]),
- },
+ message: Message::ClientRequest { id: vec![0x01], request: Request::Write(vec![0xaf]) },
  })?;
  assert_node(&mut node).is_leader().term(3).committed(2).last(6).entry(Entry {
  index: 6,

src/sql/engine/raft.rs

@@ -87,17 +87,17 @@ impl Client {
  /// Mutates the Raft state machine, deserializing the response into the
  /// return type.
  fn mutate<V: DeserializeOwned>(&self, mutation: Mutation) -> Result<V> {
- match self.execute(raft::Request::Mutate(bincode::serialize(&mutation)?))? {
- raft::Response::Mutate(response) => Ok(bincode::deserialize(&response)?),
+ match self.execute(raft::Request::Write(bincode::serialize(&mutation)?))? {
+ raft::Response::Write(response) => Ok(bincode::deserialize(&response)?),
  resp => Err(Error::Internal(format!("Unexpected Raft mutation response {:?}", resp))),
  }
  }
 
  /// Queries the Raft state machine, deserializing the response into the
  /// return type.
  fn query<V: DeserializeOwned>(&self, query: Query) -> Result<V> {
- match self.execute(raft::Request::Query(bincode::serialize(&query)?))? {
- raft::Response::Query(response) => Ok(bincode::deserialize(&response)?),
+ match self.execute(raft::Request::Read(bincode::serialize(&query)?))? {
+ raft::Response::Read(response) => Ok(bincode::deserialize(&response)?),
  resp => Err(Error::Internal(format!("Unexpected Raft query response {:?}", resp))),
  }
  }