remove protobuf dependency with prost-codec feature

.github/workflows/ci.yml

@@ -26,6 +26,7 @@ jobs:
     - if: matrix.os == 'ubuntu-latest'
       name: Install dependencies (ubuntu-latest)
       run: |
+        sudo apt-get update
         sudo apt-get install protobuf-compiler
     - if: matrix.os == 'macos-latest'
       name: Install dependencies (macos-latest)

Cargo.toml

@@ -19,17 +19,15 @@ members = ["proto", "harness", "datadriven"]
 default = ["protobuf-codec", "default-logger"]
 # Enable failpoints
 failpoints = ["fail/failpoints"]
-protobuf-codec = ["raft-proto/protobuf-codec", "bytes"]
+protobuf-codec = ["raft-proto/protobuf-codec"]
 prost-codec = ["raft-proto/prost-codec"]
 default-logger = ["slog-stdlog", "slog-envlogger", "slog-term"]
 
 # Make sure to synchronize updates with Harness.
 [dependencies]
-bytes = { version = "1", optional = true }
 fxhash = "0.2.1"
 fail = { version = "0.4", optional = true }
 getset = "0.1.1"
-protobuf = "2"
 thiserror = "1.0"
 raft-proto = { path = "proto", version = "0.7.0", default-features = false }
 rand = "0.8"

examples/five_mem_node/main.rs

@@ -11,7 +11,7 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 use std::{str, thread};
 
-use protobuf::Message as PbMessage;
+use raft::protocompat::*;
 use raft::storage::MemStorage;
 use raft::{prelude::*, StateRole};
 use regex::Regex;

harness/tests/integration_cases/test_raft.rs

@@ -19,8 +19,8 @@ use std::collections::HashMap;
 use std::panic::{self, AssertUnwindSafe};
 
 use harness::*;
-use protobuf::Message as PbMessage;
 use raft::eraftpb::*;
+use raft::protocompat::*;
 use raft::storage::MemStorage;
 use raft::*;
 use raft_proto::*;
@@ -397,7 +397,7 @@ fn test_progress_flow_control() {
     // election, and the first proposal (only one proposal gets sent
     // because we're in probe state).
     assert_eq!(ms.len(), 1);
-    assert_eq!(ms[0].msg_type, MessageType::MsgAppend);
+    assert_eq!(ms[0].get_msg_type(), MessageType::MsgAppend);
     assert_eq!(ms[0].entries.len(), 2);
     assert_eq!(ms[0].entries[0].data.len(), 0);
     assert_eq!(ms[0].entries[1].data.len(), 1000);
@@ -410,7 +410,7 @@ fn test_progress_flow_control() {
     ms = r.read_messages();
     assert_eq!(ms.len(), 3);
     for (i, m) in ms.iter().enumerate() {
-        if m.msg_type != MessageType::MsgAppend {
+        if m.get_msg_type() != MessageType::MsgAppend {
             panic!("{}: expected MsgAppend, got {:?}", i, m.msg_type);
         }
         if m.entries.len() != 2 {
@@ -426,7 +426,7 @@ fn test_progress_flow_control() {
     ms = r.read_messages();
     assert_eq!(ms.len(), 2);
     for (i, m) in ms.iter().enumerate() {
-        if m.msg_type != MessageType::MsgAppend {
+        if m.get_msg_type() != MessageType::MsgAppend {
             panic!("{}: expected MsgAppend, got {:?}", i, m.msg_type);
         }
     }
@@ -3342,7 +3342,7 @@ fn test_commit_after_remove_node() -> Result<()> {
     let ents = next_ents(&mut r, &s);
     assert_eq!(ents.len(), 1);
     assert_eq!(ents[0].get_entry_type(), EntryType::EntryNormal);
-    assert_eq!(ents[0].data.as_ref(), b"hello");
+    assert_eq!(&ents[0].data[..], b"hello");
 
     Ok(())
 }
@@ -4458,7 +4458,7 @@ fn test_conf_change_check_before_campaign() {
     let mut cc = ConfChange::default();
     cc.set_change_type(ConfChangeType::RemoveNode);
     cc.node_id = 3;
-    e.data = protobuf::Message::write_to_bytes(&cc).unwrap().into();
+    e.data = cc.write_to_bytes().unwrap().into();
     m.mut_entries().push(e);
     nt.send(vec![m]);
 

harness/tests/integration_cases/test_raw_node.rs

@@ -15,8 +15,8 @@
 // limitations under the License.
 
 use harness::Network;
-use protobuf::{Message as PbMessage, ProtobufEnum as _};
 use raft::eraftpb::*;
+use raft::protocompat::*;
 use raft::storage::MemStorage;
 use raft::*;
 use raft_proto::*;
@@ -887,7 +887,7 @@ fn prepare_async_entries(raw_node: &mut RawNode<MemStorage>, s: &MemStorage) {
     // election, and the first proposal (only one proposal gets sent
     // because we're in probe state).
     assert_eq!(msgs.len(), 1);
-    assert_eq!(msgs[0].msg_type, MessageType::MsgAppend);
+    assert_eq!(msgs[0].get_msg_type(), MessageType::MsgAppend);
     assert_eq!(msgs[0].entries.len(), 2);
     let _ = raw_node.advance_append(rd);
 
@@ -928,7 +928,7 @@ fn test_raw_node_with_async_entries() {
     s.wl().append(&entries).unwrap();
     let msgs = rd.messages();
     assert_eq!(msgs.len(), 5);
-    assert_eq!(msgs[0].msg_type, MessageType::MsgAppend);
+    assert_eq!(msgs[0].get_msg_type(), MessageType::MsgAppend);
     assert_eq!(msgs[0].entries.len(), 2);
     let _ = raw_node.advance_append(rd);
 }
@@ -969,7 +969,7 @@ fn test_raw_node_with_async_entries_on_follower() {
 
     // Set recent inactive to step down leader
     raw_node.raft.mut_prs().get_mut(2).unwrap().recent_active = false;
-    let mut msg = Message::new();
+    let mut msg = Message::default();
     msg.set_to(1);
     msg.set_msg_type(MessageType::MsgCheckQuorum);
     raw_node.raft.step(msg).unwrap();
@@ -1014,7 +1014,7 @@ fn test_raw_node_async_entries_with_leader_change() {
     // election, and the first proposal (only one proposal gets sent
     // because we're in probe state).
     assert_eq!(msgs.len(), 1);
-    assert_eq!(msgs[0].msg_type, MessageType::MsgAppend);
+    assert_eq!(msgs[0].get_msg_type(), MessageType::MsgAppend);
     assert_eq!(msgs[0].entries.len(), 2);
     let _ = raw_node.advance_append(rd);
 

harness/tests/test_util/mod.rs

@@ -129,6 +129,8 @@ pub fn soft_state(leader_id: u64, raft_state: StateRole) -> SoftState {
 pub const SOME_DATA: Option<&'static str> = Some("somedata");
 
 pub fn new_message_with_entries(from: u64, to: u64, ty: MessageType, ents: Vec<Entry>) -> Message {
+    #[cfg(feature = "prost-codec")]
+    let ty = ty as i32;
     let mut m = Message {
         msg_type: ty,
         to,
@@ -178,6 +180,8 @@ pub fn new_snapshot(index: u64, term: u64, voters: Vec<u64>) -> Snapshot {
 }
 
 pub fn conf_change(ty: ConfChangeType, node_id: u64) -> ConfChange {
+    #[cfg(feature = "prost-codec")]
+    let ty = ty as i32;
     ConfChange {
         change_type: ty,
         node_id,

proto/Cargo.toml

@@ -24,4 +24,4 @@ protobuf-build = { version = "0.15.1", default-features = false }
 bytes = { version = "1", optional = true }
 lazy_static = { version = "1", optional = true }
 prost = { version = "0.11", optional = true }
-protobuf = "2"
+protobuf = { version = "2", optional = true }

proto/build.rs

@@ -4,9 +4,12 @@ use protobuf_build::Builder;
 
 fn main() {
     let base = std::env::var("CARGO_MANIFEST_DIR").unwrap_or_else(|_| ".".to_string());
-    Builder::new()
+    let mut builder = Builder::new();
+    builder
         .search_dir_for_protos(&format!("{base}/proto"))
         .includes(&[format!("{base}/include"), format!("{base}/proto")])
-        .include_google_protos()
-        .generate()
+        .include_google_protos();
+    #[cfg(feature = "prost-codec")]
+    builder.wrapper_options(protobuf_build::GenOpt::all() - protobuf_build::GenOpt::MESSAGE);
+    builder.generate()
 }

proto/src/lib.rs

@@ -15,6 +15,78 @@ pub use crate::confchange::{
 pub use crate::confstate::conf_state_eq;
 pub use crate::protos::eraftpb;
 
+/// compatibility utils between `protobuf` and `prost`
+///
+/// To write code that is compatible with both the `protobuf` and `prost` codecs the following
+/// import is often enough:
+/// ```
+/// use raft_proto::protocompat::*;
+/// ```
+pub mod protocompat {
+    /// A reexport of `[::protobuf]` or `[::prost]` if the `prost-codec` feature is
+    /// enabled.
+    pub use crate::protoimpl::pbcrate;
+    /// A reexport of `[::protobuf::ProtobufError]` or the equivalent type in `[prost]` if the
+    /// `prost-codec` feature is enabled.
+    pub use crate::protoimpl::PbError;
+    /// A reexport of `[::protobuf::Message]` or `[::prost::Message]` if the `prost-codec` feature is
+    /// enabled.
+    pub use crate::protoimpl::PbMessage;
+
+    /// A compatibility layer to bring some protobuf specific methods to the prost message trait
+    ///
+    /// See the documentation of the method of the same name in `[::protobuf::Message]`
+    pub use crate::protoimpl::PbMessageExt;
+
+    /// A reexport of `[::protobuf::ProtobufEnum]` if the `protobuf-codec` feature is
+    /// enabled.
+    ///
+    /// It is an empty trait when using `prost-codec`.
+    pub use crate::protoimpl::ProtobufEnum;
+}
+
+#[cfg(all(feature = "prost-codec", feature = "protobuf-codec"))]
+compile_error!("`prost-codec` and `protobuf-codec` are mutually exclusive. Enable exactly one.");
+
+#[cfg(not(any(feature = "prost-codec", feature = "protobuf-codec")))]
+compile_error!("Enable one codec feature: `prost-codec` or `protobuf-codec`.");
+
+#[cfg(feature = "prost-codec")]
+mod protoimpl {
+    pub use ::prost as pbcrate;
+    pub use ::prost::DecodeError as PbError;
+    pub use ::prost::Message as PbMessage;
+    pub trait PbMessageExt: PbMessage {
+        fn compute_size(&self) -> u32;
+        fn merge_from_bytes(&mut self, bytes: &[u8]) -> Result<(), PbError>;
+        fn write_to_bytes(&self) -> Result<Vec<u8>, PbError>;
+    }
+    impl<T: PbMessage> PbMessageExt for T {
+        #[inline]
+        fn compute_size(&self) -> u32 {
+            self.encoded_len() as u32
+        }
+        #[inline]
+        fn merge_from_bytes(&mut self, bytes: &[u8]) -> Result<(), PbError> {
+            self.merge(bytes.as_ref())
+        }
+        #[inline]
+        fn write_to_bytes(&self) -> Result<Vec<u8>, PbError> {
+            Ok(self.encode_to_vec())
+        }
+    }
+    pub trait ProtobufEnum {}
+}
+#[cfg(feature = "protobuf-codec")]
+mod protoimpl {
+    pub use ::protobuf as pbcrate;
+    pub use ::protobuf::Message as PbMessage;
+    pub use ::protobuf::ProtobufEnum;
+    pub use ::protobuf::ProtobufError as PbError;
+    pub trait PbMessageExt: PbMessage {}
+    impl<T: PbMessage> super::protocompat::PbMessageExt for T {}
+}
+
 #[allow(dead_code)]
 #[allow(unknown_lints)]
 #[allow(clippy::all)]

src/errors.rs

@@ -24,7 +24,7 @@ pub enum Error {
     ConfigInvalid(String),
     /// A protobuf message codec failed in some manner.
     #[error("protobuf codec error {0:?}")]
-    CodecError(#[from] protobuf::ProtobufError),
+    CodecError(#[from] crate::protocompat::PbError),
     /// The node exists, but should not.
     #[error("The node {id} already exists in the {set} set.")]
     Exists {

src/lib.rs

@@ -448,7 +448,7 @@ the following:
 For example to promote a learner 4 and demote an existing voter 3:
 ```no_run
 # use raft::{Config, storage::MemStorage, raw_node::RawNode, eraftpb::*};
-# use protobuf::Message as PbMessage;
+# use raft_proto::protocompat::*;
 # use slog::{Drain, o};
 #
 # let mut config = Config { id: 1, ..Default::default() };
@@ -529,6 +529,7 @@ pub use quorum::joint::Configuration as JointConfig;
 pub use quorum::majority::Configuration as MajorityConfig;
 pub use raft_log::{RaftLog, NO_LIMIT};
 pub use raft_proto::eraftpb;
+pub use raft_proto::protocompat;
 #[allow(deprecated)]
 pub use raw_node::is_empty_snap;
 pub use raw_node::{LightReady, Peer, RawNode, Ready, SnapshotStatus};

src/raft.rs

@@ -21,7 +21,7 @@ use crate::eraftpb::{
     ConfChange, ConfChangeV2, ConfState, Entry, EntryType, HardState, Message, MessageType,
     Snapshot,
 };
-use protobuf::Message as _;
+use raft_proto::protocompat::*;
 use raft_proto::ConfChangeI;
 use rand::Rng;
 use slog::Logger;

src/raft_log.rs

@@ -729,12 +729,11 @@ mod test {
         panic::{self, AssertUnwindSafe},
     };
 
-    use protobuf::Message as PbMessage;
-
     use crate::config::Config;
     use crate::default_logger;
     use crate::eraftpb;
     use crate::errors::{Error, StorageError};
+    use crate::protocompat::*;
     use crate::raft_log::{self, RaftLog};
     use crate::storage::{GetEntriesContext, MemStorage};
     use crate::NO_LIMIT;

src/raw_node.rs

@@ -22,7 +22,7 @@
 
 use std::{collections::VecDeque, mem};
 
-use protobuf::Message as PbMessage;
+use raft_proto::protocompat::*;
 use raft_proto::ConfChangeI;
 use slog::Logger;
 

src/storage.rs

@@ -522,7 +522,7 @@ impl Storage for MemStorage {
 mod test {
     use std::panic::{self, AssertUnwindSafe};
 
-    use protobuf::Message as PbMessage;
+    use crate::protocompat::*;
 
     use crate::eraftpb::{ConfState, Entry, Snapshot};
     use crate::errors::{Error as RaftError, StorageError};

src/util.rs

@@ -10,7 +10,7 @@ use slog::{OwnedKVList, Record, KV};
 
 use crate::eraftpb::{Entry, Message};
 use crate::HashSet;
-use protobuf::Message as PbMessage;
+use raft_proto::protocompat::*;
 
 use slog::{b, record_static};