diff --git a/LICENSE-3rdparty.csv b/LICENSE-3rdparty.csv index cf67252..9590c0e 100644 --- a/LICENSE-3rdparty.csv +++ b/LICENSE-3rdparty.csv @@ -2,43 +2,30 @@ Component,Origin,License,Copyright allocator-api2,https://github.com/zakarumych/allocator-api2,MIT OR Apache-2.0,Zakarum anyhow,https://github.com/dtolnay/anyhow,MIT OR Apache-2.0,David Tolnay async-trait,https://github.com/dtolnay/async-trait,MIT OR Apache-2.0,David Tolnay -bitflags,https://github.com/bitflags/bitflags,MIT OR Apache-2.0,The Rust Project Developers bytes,https://github.com/tokio-rs/bytes,MIT,"Carl Lerche , Sean McArthur " -cfg-if,https://github.com/rust-lang/cfg-if,MIT OR Apache-2.0,Alex Crichton +cfg-if,https://github.com/alexcrichton/cfg-if,MIT OR Apache-2.0,Alex Crichton chacha20,https://github.com/RustCrypto/stream-ciphers,MIT OR Apache-2.0,RustCrypto Developers cpufeatures,https://github.com/RustCrypto/utils,MIT OR Apache-2.0,RustCrypto Developers -either,https://github.com/rayon-rs/either,MIT OR Apache-2.0,The either Authors -equivalent,https://github.com/indexmap-rs/equivalent,Apache-2.0 OR MIT,The equivalent Authors +either,https://github.com/rayon-rs/either,MIT OR Apache-2.0,bluss +equivalent,https://github.com/cuviper/equivalent,Apache-2.0 OR MIT,The equivalent Authors foldhash,https://github.com/orlp/foldhash,Zlib,Orson Peters futures-core,https://github.com/rust-lang/futures-rs,MIT OR Apache-2.0,The futures-core Authors futures-sink,https://github.com/rust-lang/futures-rs,MIT OR Apache-2.0,The futures-sink Authors getrandom,https://github.com/rust-random/getrandom,MIT OR Apache-2.0,The Rand Project Developers -hashbrown,https://github.com/rust-lang/hashbrown,MIT OR Apache-2.0,Amanieu d'Antras hashbrown,https://github.com/rust-lang/hashbrown,MIT OR Apache-2.0,The hashbrown Authors -heck,https://github.com/withoutboats/heck,MIT OR Apache-2.0,The heck Authors -id-arena,https://github.com/fitzgen/id-arena,MIT OR Apache-2.0,"Nick Fitzgerald , Aleksey Kladov " -indexmap,https://github.com/indexmap-rs/indexmap,Apache-2.0 OR MIT,The indexmap Authors itertools,https://github.com/rust-itertools/itertools,MIT OR Apache-2.0,bluss -itoa,https://github.com/dtolnay/itoa,MIT OR Apache-2.0,David Tolnay -leb128fmt,https://github.com/bluk/leb128fmt,MIT OR Apache-2.0,Bryant Luk libc,https://github.com/rust-lang/libc,MIT OR Apache-2.0,The Rust Project Developers -log,https://github.com/rust-lang/log,MIT OR Apache-2.0,The Rust Project Developers lru,https://github.com/jeromefroe/lru-rs,MIT,Jerome Froelich -memchr,https://github.com/BurntSushi/memchr,Unlicense OR MIT,"Andrew Gallant , bluss" mio,https://github.com/tokio-rs/mio,MIT,"Carl Lerche , Thomas de Zeeuw , Tokio Contributors " once_cell,https://github.com/matklad/once_cell,MIT OR Apache-2.0,Aleksey Kladov pin-project-lite,https://github.com/taiki-e/pin-project-lite,Apache-2.0 OR MIT,The pin-project-lite Authors -prettyplease,https://github.com/dtolnay/prettyplease,MIT OR Apache-2.0,David Tolnay proc-macro2,https://github.com/dtolnay/proc-macro2,MIT OR Apache-2.0,"David Tolnay , Alex Crichton " quote,https://github.com/dtolnay/quote,MIT OR Apache-2.0,David Tolnay r-efi,https://github.com/r-efi/r-efi,MIT OR Apache-2.0 OR LGPL-2.1-or-later,The r-efi Authors rand,https://github.com/rust-random/rand,MIT OR Apache-2.0,"The Rand Project Developers, The Rust Project Developers" rand_core,https://github.com/rust-random/rand_core,MIT OR Apache-2.0,The Rand Project Developers -semver,https://github.com/dtolnay/semver,MIT OR Apache-2.0,David Tolnay serde,https://github.com/serde-rs/serde,MIT OR Apache-2.0,"Erick Tryzelaar , David Tolnay " -serde_core,https://github.com/serde-rs/serde,MIT OR Apache-2.0,"Erick Tryzelaar , David Tolnay " serde_derive,https://github.com/serde-rs/serde,MIT OR Apache-2.0,"Erick Tryzelaar , David Tolnay " -serde_json,https://github.com/serde-rs/json,MIT OR Apache-2.0,"Erick Tryzelaar , David Tolnay " socket2,https://github.com/rust-lang/socket2,MIT OR Apache-2.0,"Alex Crichton , Thomas de Zeeuw " syn,https://github.com/dtolnay/syn,MIT OR Apache-2.0,David Tolnay tokio,https://github.com/tokio-rs/tokio,MIT,Tokio Contributors @@ -49,23 +36,10 @@ tracing,https://github.com/tokio-rs/tracing,MIT,"Eliza Weisman , Eliza Weisman , David Barsky " tracing-core,https://github.com/tokio-rs/tracing,MIT,Tokio Contributors unicode-ident,https://github.com/dtolnay/unicode-ident,(MIT OR Apache-2.0) AND Unicode-3.0,David Tolnay -unicode-xid,https://github.com/unicode-rs/unicode-xid,MIT OR Apache-2.0,"erick.tryzelaar , kwantam , Manish Goregaokar " valuable,https://github.com/tokio-rs/valuable,MIT,The valuable Authors wasi,https://github.com/bytecodealliance/wasi,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,The Cranelift Project Developers -wasip2,https://github.com/bytecodealliance/wasi-rs,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,The wasip2 Authors -wasip3,https://github.com/bytecodealliance/wasi-rs,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,The wasip3 Authors -wasm-encoder,https://github.com/bytecodealliance/wasm-tools/tree/main/crates/wasm-encoder,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,Nick Fitzgerald -wasm-metadata,https://github.com/bytecodealliance/wasm-tools/tree/main/crates/wasm-metadata,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,The wasm-metadata Authors -wasmparser,https://github.com/bytecodealliance/wasm-tools/tree/main/crates/wasmparser,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,Yury Delendik windows-link,https://github.com/microsoft/windows-rs,MIT OR Apache-2.0,The windows-link Authors windows-sys,https://github.com/microsoft/windows-rs,MIT OR Apache-2.0,The windows-sys Authors -wit-bindgen,https://github.com/bytecodealliance/wit-bindgen,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,Alex Crichton -wit-bindgen-core,https://github.com/bytecodealliance/wit-bindgen,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,Alex Crichton -wit-bindgen-rust,https://github.com/bytecodealliance/wit-bindgen,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,Alex Crichton -wit-bindgen-rust-macro,https://github.com/bytecodealliance/wit-bindgen,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,Alex Crichton -wit-component,https://github.com/bytecodealliance/wasm-tools/tree/main/crates/wit-component,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,Peter Huene -wit-parser,https://github.com/bytecodealliance/wasm-tools/tree/main/crates/wit-parser,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,Alex Crichton -zmij,https://github.com/dtolnay/zmij,MIT,David Tolnay zstd,https://github.com/gyscos/zstd-rs,MIT,Alexandre Bury zstd-safe,https://github.com/gyscos/zstd-rs,MIT OR Apache-2.0,Alexandre Bury zstd-sys,https://github.com/gyscos/zstd-rs,MIT OR Apache-2.0,Alexandre Bury diff --git a/chitchat-test/src/main.rs b/chitchat-test/src/main.rs index 3bd3d14..4a26cee 100644 --- a/chitchat-test/src/main.rs +++ b/chitchat-test/src/main.rs @@ -3,7 +3,9 @@ use std::sync::Arc; use std::time::{Duration, SystemTime}; use chitchat::transport::UdpTransport; -use chitchat::{Chitchat, ChitchatConfig, ChitchatId, FailureDetectorConfig, spawn_chitchat}; +use chitchat::{ + Chitchat, ChitchatConfig, ChitchatId, FailureDetectorConfig, ProtocolVersion, spawn_chitchat, +}; use chitchat_test::{ApiResponse, SetKeyValueResponse}; use cool_id_generator::Size; use poem::listener::TcpListener; @@ -112,6 +114,7 @@ async fn main() -> anyhow::Result<()> { marked_for_deletion_grace_period: Duration::from_secs(60), catchup_callback: None, extra_liveness_predicate: None, + protocol_version: ProtocolVersion::V1, }; let chitchat_handler = spawn_chitchat(config, Vec::new(), &UdpTransport).await?; let chitchat = chitchat_handler.chitchat(); diff --git a/chitchat/src/configuration.rs b/chitchat/src/configuration.rs index cace341..de0e7c1 100644 --- a/chitchat/src/configuration.rs +++ b/chitchat/src/configuration.rs @@ -3,7 +3,7 @@ use std::net::SocketAddr; use std::time::Duration; -use crate::{ChitchatId, FailureDetectorConfig, NodeState}; +use crate::{ChitchatId, FailureDetectorConfig, NodeState, ProtocolVersion}; /// An optional user-defined callback executed when the self node is lagging behind. pub type CatchupCallback = Box; @@ -41,6 +41,10 @@ pub struct ChitchatConfig { // It can be used for instance, to only surface the nodes that are both alive according // to the failure detector, but also have a given set of required keys. pub extra_liveness_predicate: Option, + /// Protocol version used to exchange messages between peers. The latest version is V1, which + /// compresses the digest. This MUST stay `V0` until every node in the cluster runs a version + /// able to decode the newer format. + pub protocol_version: ProtocolVersion, } impl ChitchatConfig { @@ -58,6 +62,7 @@ impl ChitchatConfig { marked_for_deletion_grace_period: Duration::from_secs(10_000), catchup_callback: None, extra_liveness_predicate: None, + protocol_version: ProtocolVersion::V0, } } } @@ -77,6 +82,7 @@ impl Default for ChitchatConfig { marked_for_deletion_grace_period: Duration::from_secs(3_600 * 2), // 2h catchup_callback: None, extra_liveness_predicate: None, + protocol_version: ProtocolVersion::V0, } } } diff --git a/chitchat/src/digest.rs b/chitchat/src/digest.rs index 7a791e2..a074bdf 100644 --- a/chitchat/src/digest.rs +++ b/chitchat/src/digest.rs @@ -1,8 +1,12 @@ use std::collections::BTreeMap; +use crate::message::ProtocolVersion; use crate::serialize::*; use crate::{ChitchatId, Heartbeat, Version}; +/// Block size used by the compressed digest stream. +const DIGEST_BLOCK_THRESHOLD: u16 = 16_384u16; + #[derive(Debug, Clone, Copy, Default, Eq, PartialEq)] pub(crate) struct NodeDigest { pub(crate) heartbeat: Heartbeat, @@ -37,6 +41,15 @@ impl Deserializable for NodeDigest { } } +/// Decodes one `(ChitchatId, NodeDigest)` entry of the compressed digest stream. +impl Deserializable for (ChitchatId, NodeDigest) { + fn deserialize(buf: &mut &[u8]) -> anyhow::Result { + let chitchat_id = ChitchatId::deserialize(buf)?; + let node_digest = NodeDigest::deserialize(buf)?; + Ok((chitchat_id, node_digest)) + } +} + /// A digest represents is a piece of information summarizing /// the staleness of one peer's data. /// @@ -63,28 +76,81 @@ impl Digest { }; self.node_digests.insert(node, node_digest); } + + pub fn sample_for_test(num_nodes: u16) -> Digest { + let mut digest = Digest::default(); + for i in 0..num_nodes { + let node = ChitchatId::for_local_test(10_000 + i); + digest.add_node( + node, + Heartbeat(100 + i as u64), + i as u64 % 7, + 1_000 + i as u64, + ); + } + digest + } } -impl Serializable for Digest { - fn serialize(&self, buf: &mut Vec) { +impl Digest { + /// Serializes the digest using the wire format selected by `protocol_version`. + pub(crate) fn serialize(&self, protocol_version: ProtocolVersion, buf: &mut Vec) { + match protocol_version { + ProtocolVersion::V0 => self.serialize_v0(buf), + ProtocolVersion::V1 => self.serialize_v1(buf), + } + } + + pub(crate) fn serialized_len(&self, protocol_version: ProtocolVersion) -> usize { + match protocol_version { + ProtocolVersion::V0 => { + let mut len = (self.node_digests.len() as u16).serialized_len(); + for (chitchat_id, node_digest) in &self.node_digests { + len += chitchat_id.serialized_len(); + len += node_digest.serialized_len(); + } + len + } + ProtocolVersion::V1 => { + let mut buf = Vec::new(); + self.serialize_v1(&mut buf); + buf.len() + } + } + } + + pub(crate) fn deserialize( + protocol_version: ProtocolVersion, + buf: &mut &[u8], + ) -> anyhow::Result { + match protocol_version { + ProtocolVersion::V0 => Self::deserialize_v0(buf), + ProtocolVersion::V1 => Self::deserialize_v1(buf), + } + } + + /// Serializes the digest using the V0 (uncompressed) wire format. + fn serialize_v0(&self, buf: &mut Vec) { (self.node_digests.len() as u16).serialize(buf); for (chitchat_id, node_digest) in &self.node_digests { chitchat_id.serialize(buf); node_digest.serialize(buf); } } - fn serialized_len(&self) -> usize { - let mut len = (self.node_digests.len() as u16).serialized_len(); + + /// Serializes the digest using the V1 (compressed) wire format: a zstd block + /// stream of `(ChitchatId, NodeDigest)` entries. + fn serialize_v1(&self, buf: &mut Vec) { + let mut stream_writer = + CompressedStreamWriter::with_block_threshold(DIGEST_BLOCK_THRESHOLD); for (chitchat_id, node_digest) in &self.node_digests { - len += chitchat_id.serialized_len(); - len += node_digest.serialized_len(); + stream_writer.append(chitchat_id); + stream_writer.append(node_digest); } - len + buf.extend_from_slice(&stream_writer.finish()); } -} -impl Deserializable for Digest { - fn deserialize(buf: &mut &[u8]) -> anyhow::Result { + fn deserialize_v0(buf: &mut &[u8]) -> anyhow::Result { let num_nodes = u16::deserialize(buf)?; let mut node_digests: BTreeMap = Default::default(); @@ -95,14 +161,29 @@ impl Deserializable for Digest { } Ok(Digest { node_digests }) } + + fn deserialize_v1(buf: &mut &[u8]) -> anyhow::Result { + let node_digests = deserialize_stream::<(ChitchatId, NodeDigest)>(buf)? + .into_iter() + .collect(); + Ok(Digest { node_digests }) + } } #[cfg(test)] mod tests { use crate::digest::{Digest, NodeDigest}; + use crate::message::ProtocolVersion; use crate::serialize::test_serdeser_aux; use crate::{ChitchatId, Heartbeat}; + fn digest_serdeser_aux(digest: &Digest, protocol_version: ProtocolVersion) -> Digest { + let mut buf = Vec::new(); + digest.serialize(protocol_version, &mut buf); + assert_eq!(buf.len(), digest.serialized_len(protocol_version)); + Digest::deserialize(protocol_version, &mut &buf[..]).unwrap() + } + #[test] fn test_node_digest_serialization() { let node_digest = NodeDigest { @@ -122,6 +203,27 @@ mod tests { digest.add_node(node1, Heartbeat(101), 1, 11); digest.add_node(node2, Heartbeat(102), 20, 12); digest.add_node(node3, Heartbeat(103), 0, 13); - test_serdeser_aux(&digest, 104); + + let mut buf = Vec::new(); + digest.serialize(ProtocolVersion::V0, &mut buf); + assert_eq!(buf.len(), 104); + assert_eq!(digest_serdeser_aux(&digest, ProtocolVersion::V0), digest); + } + + #[test] + fn test_digest_compressed_roundtrip() { + for num_nodes in [0u16, 1, 3, 500] { + let digest = Digest::sample_for_test(num_nodes); + assert_eq!(digest_serdeser_aux(&digest, ProtocolVersion::V1), digest); + } + } + + #[test] + fn test_digest_v0_and_v1_decode_identically() { + let digest = Digest::sample_for_test(200); + assert_eq!( + digest_serdeser_aux(&digest, ProtocolVersion::V0), + digest_serdeser_aux(&digest, ProtocolVersion::V1), + ); } } diff --git a/chitchat/src/lib.rs b/chitchat/src/lib.rs index 7c1397b..664c80a 100644 --- a/chitchat/src/lib.rs +++ b/chitchat/src/lib.rs @@ -30,7 +30,7 @@ use tracing::{error, info, warn}; pub use self::configuration::{CatchupCallback, ChitchatConfig}; pub use self::state::{ClusterStateSnapshot, NodeState}; use crate::digest::Digest; -pub use crate::message::ChitchatMessage; +pub use crate::message::{ChitchatMessage, ProtocolVersion}; pub use crate::server::{ChitchatHandle, spawn_chitchat}; use crate::state::ClusterState; pub use crate::types::{ChitchatId, DeletionStatus, Heartbeat, Version, VersionedValue}; @@ -46,6 +46,15 @@ pub use crate::types::{ChitchatId, DeletionStatus, Heartbeat, Version, Versioned /// or so. pub(crate) const MAX_UDP_DATAGRAM_PAYLOAD_SIZE: usize = 65_507; +/// Number of bytes left for a delta payload in a single reply datagram, once +/// the message header and `reserved_len` bytes (e.g. a digest carried in the +/// same message) are accounted for. +pub(crate) const fn delta_mtu(reserved_len: usize) -> usize { + MAX_UDP_DATAGRAM_PAYLOAD_SIZE + .saturating_sub(ChitchatMessage::HEADER_LEN) + .saturating_sub(reserved_len) +} + /// To prevent dead nodes from being recorded again after deletion, /// we keep a local memory of the last nodes that were garbage collected. pub(crate) const GARBAGE_COLLECTED_NODE_HISTORY_SIZE: NonZeroUsize = @@ -95,6 +104,7 @@ impl Chitchat { let scheduled_for_deletion: HashSet<_> = self.scheduled_for_deletion_nodes().collect(); let digest = self.compute_digest(&scheduled_for_deletion); ChitchatMessage::Syn { + protocol_version: self.config.protocol_version, cluster_id: self.config.cluster_id.clone(), digest, } @@ -122,7 +132,9 @@ impl Chitchat { self.update_self_heartbeat(); match msg { - ChitchatMessage::Syn { cluster_id, digest } => { + ChitchatMessage::Syn { + cluster_id, digest, .. + } => { if cluster_id != self.cluster_id() { warn!( our_cluster_id=%self.cluster_id(), @@ -135,25 +147,26 @@ impl Chitchat { let scheduled_for_deletion: HashSet<_> = self.scheduled_for_deletion_nodes().collect(); let self_digest = self.compute_digest(&scheduled_for_deletion); - let delta_mtu = MAX_UDP_DATAGRAM_PAYLOAD_SIZE - 1 - self_digest.serialized_len(); + let protocol_version = self.config.protocol_version; let delta = self.cluster_state.compute_partial_delta_respecting_mtu( &digest, - delta_mtu, + delta_mtu(self_digest.serialized_len(protocol_version)), &scheduled_for_deletion, ); Some(ChitchatMessage::SynAck { + protocol_version, digest: self_digest, delta, }) } - ChitchatMessage::SynAck { digest, delta } => { + ChitchatMessage::SynAck { digest, delta, .. } => { self.report_heartbeats_in_digest(&digest); self.process_delta(delta); let scheduled_for_deletion = self.scheduled_for_deletion_nodes().collect::>(); let delta = self.cluster_state.compute_partial_delta_respecting_mtu( &digest, - MAX_UDP_DATAGRAM_PAYLOAD_SIZE - 1, + delta_mtu(0), &scheduled_for_deletion, ); Some(ChitchatMessage::Ack { delta }) @@ -486,6 +499,44 @@ mod tests { const DEAD_NODE_GRACE_PERIOD: Duration = Duration::from_secs(20); + /// Guards against the protocol drifting (a new header byte, a different + /// digest encoding, ...) without `delta_mtu`/`HEADER_LEN` being updated to + /// match: the number of bytes `delta_mtu` reserves for everything but the + /// delta must equal what a real reply message actually serializes to, + /// otherwise the delta could overflow the datagram. + #[test] + fn test_delta_mtu_matches_serialization() { + // The delta content is irrelevant here; we only measure the overhead + // around it. + let delta_len = Delta::default().serialize_to_vec().len(); + + // SYN-ACK carries a digest, whose serialized length must be reserved. + for protocol_version in [ProtocolVersion::V0, ProtocolVersion::V1] { + let digest = Digest::sample_for_test(5); + let reserved = digest.serialized_len(protocol_version); + let syn_ack = ChitchatMessage::SynAck { + protocol_version, + digest, + delta: Delta::default(), + }; + let message_len = syn_ack.serialize_to_vec().len(); + assert_eq!( + message_len - delta_len, + MAX_UDP_DATAGRAM_PAYLOAD_SIZE - delta_mtu(reserved), + ); + } + + // ACK reserves nothing beyond the header. + let ack = ChitchatMessage::Ack { + delta: Delta::default(), + }; + let message_len = ack.serialize_to_vec().len(); + assert_eq!( + message_len - delta_len, + MAX_UDP_DATAGRAM_PAYLOAD_SIZE - delta_mtu(0), + ); + } + fn run_chitchat_handshake(initiating_node: &mut Chitchat, peer_node: &mut Chitchat) { let syn_message = initiating_node.create_syn_message(); let syn_ack_message = peer_node.process_message(syn_message).unwrap(); @@ -547,6 +598,7 @@ mod tests { marked_for_deletion_grace_period: Duration::from_secs(3_600), catchup_callback: None, extra_liveness_predicate: None, + protocol_version: ProtocolVersion::V0, }; start_node_with_config(transport, config).await } @@ -727,6 +779,7 @@ mod tests { extra_liveness_predicate: Some(Box::new(|node_state| { node_state.get("READY") == Some("true") })), + protocol_version: ProtocolVersion::V0, }; let mut nodes = Vec::new(); for chitchat_id in &chitchat_ids { @@ -1186,7 +1239,11 @@ mod tests { let mut digest = Digest::default(); let peer_id = ChitchatId::for_local_test(10_002); digest.add_node(peer_id, Heartbeat::default(), 10, 30); - let _ = node.process_message(ChitchatMessage::Syn { cluster_id, digest }); + let _ = node.process_message(ChitchatMessage::Syn { + protocol_version: ProtocolVersion::V0, + cluster_id, + digest, + }); let delta = Delta::default(); node.process_delta(delta); @@ -1388,6 +1445,7 @@ mod tests { let ack = node .process_message(ChitchatMessage::Syn { + protocol_version: ProtocolVersion::V0, cluster_id: node.config.cluster_id.clone(), digest: Digest::default(), }) diff --git a/chitchat/src/message.rs b/chitchat/src/message.rs index ca94711..7edcecb 100644 --- a/chitchat/src/message.rs +++ b/chitchat/src/message.rs @@ -17,10 +17,18 @@ const MAGIC_NUMBER: u16 = 45_139; #[derive(Debug, Eq, PartialEq)] pub enum ChitchatMessage { /// Scuttlebutt SYN: node A initiates a handshake and sends its digest. - Syn { cluster_id: String, digest: Digest }, + Syn { + protocol_version: ProtocolVersion, + cluster_id: String, + digest: Digest, + }, /// Scuttlebutt SYN-ACK: node B returns a partial update as described in the Scuttlebutt /// reconciliation algorithm and its own digest. - SynAck { digest: Digest, delta: Delta }, + SynAck { + protocol_version: ProtocolVersion, + digest: Digest, + delta: Delta, + }, /// Scuttlebutt ACK: node A returns a partial update for B. Ack { delta: Delta }, /// Node B rejects the SYN message because node A and B belong to different clusters. @@ -30,16 +38,40 @@ pub enum ChitchatMessage { PanicForTest, } -#[derive(Clone, Copy, Eq, PartialEq)] +impl ChitchatMessage { + /// Bytes written before the message body: magic number (2) + protocol + /// version (1) + message type (1). + pub(crate) const HEADER_LEN: usize = 2 + 1 + 1; + + /// Protocol version stamped in the message header. Only messages carrying a + /// digest can select a version; the others are always V0. + fn protocol_version(&self) -> ProtocolVersion { + match self { + ChitchatMessage::Syn { + protocol_version, .. + } + | ChitchatMessage::SynAck { + protocol_version, .. + } => *protocol_version, + _ => ProtocolVersion::V0, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, PartialOrd, Ord)] #[repr(u8)] -enum ProtocolVersion { +pub enum ProtocolVersion { + /// The digest is serialized using the V0 (uncompressed) wire format. V0 = 0, + /// The digest is serialized using the V1 (compressed) wire format. + V1 = 1, } impl ProtocolVersion { pub fn from_code(code: u8) -> Option { match code { 0 => Some(Self::V0), + 1 => Some(Self::V1), _ => None, } } @@ -81,17 +113,25 @@ impl MessageType { impl Serializable for ChitchatMessage { fn serialize(&self, buf: &mut Vec) { buf.extend(MAGIC_NUMBER.to_le_bytes()); - ProtocolVersion::V0.to_code().serialize(buf); + self.protocol_version().to_code().serialize(buf); match self { - ChitchatMessage::Syn { cluster_id, digest } => { + ChitchatMessage::Syn { + protocol_version, + cluster_id, + digest, + } => { buf.push(MessageType::Syn.to_code()); - digest.serialize(buf); + digest.serialize(*protocol_version, buf); cluster_id.serialize(buf); } - ChitchatMessage::SynAck { digest, delta } => { + ChitchatMessage::SynAck { + protocol_version, + digest, + delta, + } => { buf.push(MessageType::SynAck.to_code()); - digest.serialize(buf); + digest.serialize(*protocol_version, buf); delta.serialize(buf); } ChitchatMessage::Ack { delta } => { @@ -111,12 +151,16 @@ impl Serializable for ChitchatMessage { fn serialized_len(&self) -> usize { 2 + 1 + match self { - ChitchatMessage::Syn { cluster_id, digest } => { - 1 + cluster_id.serialized_len() + digest.serialized_len() - } - ChitchatMessage::SynAck { digest, delta } => { - 1 + digest.serialized_len() + delta.serialized_len() - } + ChitchatMessage::Syn { + protocol_version, + cluster_id, + digest, + } => 1 + cluster_id.serialized_len() + digest.serialized_len(*protocol_version), + ChitchatMessage::SynAck { + protocol_version, + digest, + delta, + } => 1 + digest.serialized_len(*protocol_version) + delta.serialized_len(), ChitchatMessage::Ack { delta } => 1 + delta.serialized_len(), ChitchatMessage::BadCluster => 1, #[cfg(test)] @@ -136,13 +180,6 @@ impl Deserializable for ChitchatMessage { } let protocol_version = ProtocolVersion::from_code(buf[2]).context("invalid protocol version")?; - - if protocol_version != ProtocolVersion::V0 { - bail!( - "unsupported protocol version `{}`", - protocol_version.to_code() - ) - } buf.consume(3); let message_type = buf @@ -154,14 +191,22 @@ impl Deserializable for ChitchatMessage { match message_type { MessageType::Syn => { - let digest = Digest::deserialize(buf)?; + let digest = Digest::deserialize(protocol_version, buf)?; let cluster_id = String::deserialize(buf)?; - Ok(Self::Syn { cluster_id, digest }) + Ok(Self::Syn { + protocol_version, + cluster_id, + digest, + }) } MessageType::SynAck => { - let digest = Digest::deserialize(buf)?; + let digest = Digest::deserialize(protocol_version, buf)?; let delta = Delta::deserialize(buf)?; - Ok(Self::SynAck { digest, delta }) + Ok(Self::SynAck { + protocol_version, + digest, + delta, + }) } MessageType::Ack => { let delta = Delta::deserialize(buf)?; @@ -176,6 +221,7 @@ impl Deserializable for ChitchatMessage { #[cfg(test)] mod tests { + use crate::message::ProtocolVersion; use crate::serialize::test_serdeser_aux; use crate::{ChitchatId, ChitchatMessage, Delta, Digest, Heartbeat}; @@ -183,6 +229,7 @@ mod tests { fn test_syn() { { let syn = ChitchatMessage::Syn { + protocol_version: ProtocolVersion::V0, cluster_id: "cluster-a".to_string(), digest: Digest::default(), }; @@ -194,6 +241,7 @@ mod tests { digest.add_node(node, Heartbeat(0), 0, 0); let syn = ChitchatMessage::Syn { + protocol_version: ProtocolVersion::V0, cluster_id: "cluster-a".to_string(), digest, }; @@ -201,10 +249,32 @@ mod tests { } } + #[test] + fn test_syn_compressed_digest_roundtrips() { + use crate::serialize::{Deserializable, Serializable}; + + let mut digest = Digest::default(); + for i in 0..100 { + let node = ChitchatId::for_local_test(10_000 + i); + digest.add_node(node, Heartbeat(i as u64), 0, i as u64); + } + // Emit the compressed wire format. + let syn = ChitchatMessage::Syn { + protocol_version: ProtocolVersion::V1, + cluster_id: "cluster-a".to_string(), + digest, + }; + let buf = syn.serialize_to_vec(); + // A peer decodes it transparently, regardless of the encoding used. + let deserialized = ChitchatMessage::deserialize(&mut &buf[..]).unwrap(); + assert_eq!(syn, deserialized); + } + #[test] fn test_syn_ack() { { let syn_ack = ChitchatMessage::SynAck { + protocol_version: ProtocolVersion::V0, digest: Digest::default(), delta: Delta::default(), }; @@ -232,7 +302,11 @@ mod tests { // That's compression kicking in. delta.set_serialized_len(70); - let syn_ack = ChitchatMessage::SynAck { digest, delta }; + let syn_ack = ChitchatMessage::SynAck { + protocol_version: ProtocolVersion::V0, + digest, + delta, + }; // 1 byte (protocol version) + 1 byte (message tag) + 53 bytes (digest) + 60 bytes // (delta). test_serdeser_aux(&syn_ack, 2 + 1 + 1 + 53 + 70); diff --git a/chitchat/src/server.rs b/chitchat/src/server.rs index 95a7dfa..c107fba 100644 --- a/chitchat/src/server.rs +++ b/chitchat/src/server.rs @@ -499,7 +499,9 @@ mod tests { let (from, message) = timeout(peer_transport.recv()).await.unwrap(); assert_eq!(from, test_addr); match message { - ChitchatMessage::Syn { cluster_id, digest } => { + ChitchatMessage::Syn { + cluster_id, digest, .. + } => { assert_eq!(cluster_id, "default-cluster"); assert_eq!(digest.node_digests.len(), 1); } diff --git a/chitchat/src/transport/mod.rs b/chitchat/src/transport/mod.rs index 925dce7..75de31f 100644 --- a/chitchat/src/transport/mod.rs +++ b/chitchat/src/transport/mod.rs @@ -38,12 +38,13 @@ mod tests { use super::Transport; use crate::MAX_UDP_DATAGRAM_PAYLOAD_SIZE; use crate::digest::Digest; - use crate::message::ChitchatMessage; + use crate::message::{ChitchatMessage, ProtocolVersion}; use crate::serialize::Serializable; use crate::transport::{ChannelTransport, UdpTransport}; fn sample_syn_msg() -> ChitchatMessage { ChitchatMessage::Syn { + protocol_version: ProtocolVersion::V0, cluster_id: "cluster_id".to_string(), digest: Digest::default(), } diff --git a/chitchat/tests/cluster_test.rs b/chitchat/tests/cluster_test.rs index e2852db..d4189a9 100644 --- a/chitchat/tests/cluster_test.rs +++ b/chitchat/tests/cluster_test.rs @@ -7,7 +7,7 @@ use anyhow::anyhow; use chitchat::transport::ChannelTransport; use chitchat::{ Chitchat, ChitchatConfig, ChitchatHandle, ChitchatId, FailureDetectorConfig, NodeState, - spawn_chitchat, + ProtocolVersion, spawn_chitchat, }; use itertools::Itertools; use rand::prelude::IndexedRandom; @@ -434,6 +434,7 @@ impl Simulator { marked_for_deletion_grace_period: self.marked_for_deletion_key_grace_period, catchup_callback, extra_liveness_predicate: None, + protocol_version: ProtocolVersion::V0, }; let handle = spawn_chitchat(config, Vec::new(), &self.transport) .await diff --git a/chitchat/tests/perf_test.rs b/chitchat/tests/perf_test.rs index 51de277..401c3dd 100644 --- a/chitchat/tests/perf_test.rs +++ b/chitchat/tests/perf_test.rs @@ -5,7 +5,8 @@ use std::time::Duration; use chitchat::transport::{ChannelTransport, Transport, TransportExt}; use chitchat::{ - ChitchatConfig, ChitchatHandle, ChitchatId, FailureDetectorConfig, NodeState, spawn_chitchat, + ChitchatConfig, ChitchatHandle, ChitchatId, FailureDetectorConfig, NodeState, ProtocolVersion, + spawn_chitchat, }; use tokio::time::Instant; use tokio_stream::StreamExt; @@ -32,6 +33,7 @@ async fn spawn_one(chitchat_id: u16, transport: &dyn Transport) -> ChitchatHandl marked_for_deletion_grace_period: Duration::from_secs(10_000), catchup_callback: None, extra_liveness_predicate: None, + protocol_version: ProtocolVersion::V0, }; spawn_chitchat(config, Vec::new(), transport).await.unwrap() }