diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ff5bd7d21..83775db29 100644 --- a/.github/workflows/ci.yml +++ b/.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) diff --git a/Cargo.toml b/Cargo.toml index 2bcf72720..2476e6352 100644 --- a/Cargo.toml +++ b/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" diff --git a/examples/five_mem_node/main.rs b/examples/five_mem_node/main.rs index d2326aa98..99f6fe87b 100644 --- a/examples/five_mem_node/main.rs +++ b/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; diff --git a/harness/tests/integration_cases/test_raft.rs b/harness/tests/integration_cases/test_raft.rs index 727fa9624..59a9cdae8 100644 --- a/harness/tests/integration_cases/test_raft.rs +++ b/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]); diff --git a/harness/tests/integration_cases/test_raw_node.rs b/harness/tests/integration_cases/test_raw_node.rs index c7e62a6be..f506906d2 100644 --- a/harness/tests/integration_cases/test_raw_node.rs +++ b/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, 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); diff --git a/harness/tests/test_util/mod.rs b/harness/tests/test_util/mod.rs index 076f7c5bf..d02898ad2 100644 --- a/harness/tests/test_util/mod.rs +++ b/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) -> 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) -> 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, diff --git a/proto/Cargo.toml b/proto/Cargo.toml index 5efbe0dd3..fbe350290 100644 --- a/proto/Cargo.toml +++ b/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 } diff --git a/proto/build.rs b/proto/build.rs index bd173334a..c8d0b4691 100644 --- a/proto/build.rs +++ b/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() } diff --git a/proto/src/lib.rs b/proto/src/lib.rs index 6b88d14fd..6a11b1089 100644 --- a/proto/src/lib.rs +++ b/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, PbError>; + } + impl 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, 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 super::protocompat::PbMessageExt for T {} +} + #[allow(dead_code)] #[allow(unknown_lints)] #[allow(clippy::all)] diff --git a/src/errors.rs b/src/errors.rs index 21c303d47..752f56c31 100644 --- a/src/errors.rs +++ b/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 { diff --git a/src/lib.rs b/src/lib.rs index b1413c211..ff149c4a0 100644 --- a/src/lib.rs +++ b/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}; diff --git a/src/raft.rs b/src/raft.rs index 466c7a8d3..f744238a4 100644 --- a/src/raft.rs +++ b/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; diff --git a/src/raft_log.rs b/src/raft_log.rs index 8b25b56ce..8733a182a 100644 --- a/src/raft_log.rs +++ b/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; diff --git a/src/raw_node.rs b/src/raw_node.rs index 8828cfa29..463f5721f 100644 --- a/src/raw_node.rs +++ b/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; diff --git a/src/storage.rs b/src/storage.rs index a99fcddd5..bd870f180 100644 --- a/src/storage.rs +++ b/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}; diff --git a/src/util.rs b/src/util.rs index 04d89427e..23f2e5ba9 100644 --- a/src/util.rs +++ b/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};