From 07032a52ed75d8abcb55adc19a58b221e288ba24 Mon Sep 17 00:00:00 2001 From: Antonio Date: Tue, 10 Jun 2025 08:52:52 +0200 Subject: [PATCH] chore: remove drop messages from blend code (#1340) * Remove mentions of drop messages * Add comment for ignored cover messages upon deserialization failure --- nomos-blend/core/Cargo.toml | 3 +- nomos-blend/core/src/conn_maintenance.rs | 123 +++++------------- .../core/src/persistent_transmission.rs | 106 ++------------- nomos-blend/message/src/lib.rs | 5 - nomos-blend/message/src/mock/mod.rs | 1 - nomos-blend/message/src/sphinx/mod.rs | 3 - nomos-blend/network/src/behaviour.rs | 9 +- nomos-blend/network/src/handler.rs | 16 +-- nomos-blend/network/src/lib.rs | 22 ++-- nomos-services/blend/src/lib.rs | 14 +- 10 files changed, 72 insertions(+), 230 deletions(-) diff --git a/nomos-blend/core/Cargo.toml b/nomos-blend/core/Cargo.toml index 883af0a6b..d7c463c35 100644 --- a/nomos-blend/core/Cargo.toml +++ b/nomos-blend/core/Cargo.toml @@ -19,5 +19,4 @@ tokio-stream = "0.1" tracing = { workspace = true } [dev-dependencies] -rand_chacha = "0.3" -tokio = { version = "1", features = ["rt-multi-thread", "time"] } +tokio = { version = "1", features = ["rt-multi-thread", "time"] } diff --git a/nomos-blend/core/src/conn_maintenance.rs b/nomos-blend/core/src/conn_maintenance.rs index bb7a02043..68e9decc4 100644 --- a/nomos-blend/core/src/conn_maintenance.rs +++ b/nomos-blend/core/src/conn_maintenance.rs @@ -9,14 +9,13 @@ use nomos_utils::bounded_duration::{MinimalBoundedDuration, SECOND}; use serde::{Deserialize, Serialize}; use serde_with::serde_as; -/// Counts the number of effective and drop messages received from a peer during +/// Counts the number of messages received from a peer during /// an interval. `interval` is a field that implements [`futures::Stream`] to /// support both sync and async environments. pub struct ConnectionMonitor { settings: ConnectionMonitorSettings, interval: Pin + Send>>, - effective_messages: U57F7, - drop_messages: U57F7, + messages: U57F7, } #[serde_as] @@ -26,26 +25,16 @@ pub struct ConnectionMonitorSettings { /// peer. #[serde_as(as = "MinimalBoundedDuration<1, SECOND>")] pub interval: Duration, - /// The number of effective (data or cover) messages that a peer is expected + /// The number of (data or cover) messages that a peer is expected /// to send in a given time window. /// /// If the count is greater than (expected * (1 + `malicious_tolerance`)), /// the peer is considered malicious. /// If the count is less than (expected * (1 - `unhealthy_tolerance`)), the /// peer is considered unhealthy. - pub expected_effective_messages: U57F7, - pub effective_message_malicious_tolerance: U57F7, - pub effective_message_unhealthy_tolerance: U57F7, - /// The number of drop messages that a peer is expected to send - /// in a given time window. - /// - /// If the count is greater than (expected * (1 + `malicious_tolerance`)), - /// the peer is considered malicious. - /// If the count is less than (expected * (1 - `unhealthy_tolerance`)), - /// the peer is considered unhealthy. - pub expected_drop_messages: U57F7, - pub drop_message_malicious_tolerance: U57F7, - pub drop_message_unhealthy_tolerance: U57F7, + pub expected_messages: U57F7, + pub message_malicious_tolerance: U57F7, + pub message_unhealthy_tolerance: U57F7, } /// A result of connection monitoring during an interval. @@ -64,22 +53,16 @@ impl ConnectionMonitor { Self { settings, interval: Box::pin(interval), - effective_messages: U57F7::ZERO, - drop_messages: U57F7::ZERO, + messages: U57F7::ZERO, } } - /// Record an effective message received from the peer. - pub fn record_effective_message(&mut self) { - self.effective_messages = Self::record_message(self.effective_messages); + /// Record a message received from the peer. + pub fn record_message(&mut self) { + self.messages = Self::_record_message(self.messages); } - /// Record a drop effective message received from the peer. - pub fn record_drop_message(&mut self) { - self.drop_messages = Self::record_message(self.drop_messages); - } - - fn record_message(value: U57F7) -> U57F7 { + fn _record_message(value: U57F7) -> U57F7 { value.checked_add(U57F7::ONE).unwrap_or_else(|| { tracing::warn!("Skipping recording a message due to overflow"); value @@ -107,28 +90,21 @@ impl ConnectionMonitor { } const fn reset(&mut self) { - self.effective_messages = U57F7::ZERO; - self.drop_messages = U57F7::ZERO; + self.messages = U57F7::ZERO; } - /// Check if the peer is malicious based on the number of effective and drop - /// messages sent + /// Check if the peer is malicious based on the number of messages sent fn is_malicious(&self) -> bool { - let effective_threshold = self.settings.expected_effective_messages - * (U57F7::ONE + self.settings.effective_message_malicious_tolerance); - let drop_threshold = self.settings.expected_drop_messages - * (U57F7::ONE + self.settings.drop_message_malicious_tolerance); - self.effective_messages > effective_threshold || self.drop_messages > drop_threshold + let threshold = self.settings.expected_messages + * (U57F7::ONE + self.settings.message_malicious_tolerance); + self.messages > threshold } - /// Check if the peer is unhealthy based on the number of effective and drop - /// messages sent + /// Check if the peer is unhealthy based on the number of messages sent fn is_unhealthy(&self) -> bool { - let effective_threshold = self.settings.expected_effective_messages - * (U57F7::ONE - self.settings.effective_message_unhealthy_tolerance); - let drop_threshold = self.settings.expected_drop_messages - * (U57F7::ONE - self.settings.drop_message_unhealthy_tolerance); - effective_threshold > self.effective_messages || drop_threshold > self.drop_messages + let threshold = self.settings.expected_messages + * (U57F7::ONE - self.settings.message_unhealthy_tolerance); + threshold > self.messages } } @@ -144,62 +120,36 @@ mod tests { let mut monitor = ConnectionMonitor::new( ConnectionMonitorSettings { interval: Duration::from_secs(1), - expected_effective_messages: U57F7::from_num(2.0), - effective_message_malicious_tolerance: U57F7::from_num(0.5), - effective_message_unhealthy_tolerance: U57F7::from_num(0.1), - expected_drop_messages: U57F7::from_num(1.0), - drop_message_malicious_tolerance: U57F7::from_num(0.0), - drop_message_unhealthy_tolerance: U57F7::from_num(0.0), + expected_messages: U57F7::from_num(2.0), + message_malicious_tolerance: U57F7::from_num(0.5), + message_unhealthy_tolerance: U57F7::from_num(0.1), }, futures::stream::iter(std::iter::repeat(())), ); // Recording the expected number of messages, // expecting the peer to be healthy - monitor.record_effective_message(); - monitor.record_effective_message(); - monitor.record_drop_message(); + monitor.record_message(); + monitor.record_message(); assert_eq!( monitor.poll(&mut Context::from_waker(&noop_waker())), Poll::Ready(ConnectionMonitorOutput::Healthy) ); - // Recording more than the expected number of effective messages, + // Recording more than the expected number of messages, // expecting the peer to be malicious - monitor.record_effective_message(); - monitor.record_effective_message(); - monitor.record_effective_message(); - monitor.record_effective_message(); - monitor.record_drop_message(); + monitor.record_message(); + monitor.record_message(); + monitor.record_message(); + monitor.record_message(); assert_eq!( monitor.poll(&mut Context::from_waker(&noop_waker())), Poll::Ready(ConnectionMonitorOutput::Malicious) ); - // Recording less than the expected number of effective messages, + // Recording less than the expected number of messages, // expecting the peer to be unhealthy - monitor.record_effective_message(); - monitor.record_drop_message(); - assert_eq!( - monitor.poll(&mut Context::from_waker(&noop_waker())), - Poll::Ready(ConnectionMonitorOutput::Unhealthy) - ); - - // Recording more than the expected number of drop messages, - // expecting the peer to be malicious - monitor.record_effective_message(); - monitor.record_effective_message(); - monitor.record_drop_message(); - monitor.record_drop_message(); - assert_eq!( - monitor.poll(&mut Context::from_waker(&noop_waker())), - Poll::Ready(ConnectionMonitorOutput::Malicious) - ); - - // Recording less than the expected number of drop messages, - // expecting the peer to be unhealthy - monitor.record_effective_message(); - monitor.record_effective_message(); + monitor.record_message(); assert_eq!( monitor.poll(&mut Context::from_waker(&noop_waker())), Poll::Ready(ConnectionMonitorOutput::Unhealthy) @@ -212,12 +162,9 @@ mod tests { let mut monitor = ConnectionMonitor::new( ConnectionMonitorSettings { interval: Duration::from_secs(1), - expected_effective_messages: U57F7::from_num(2.0), - effective_message_malicious_tolerance: U57F7::from_num(0.1), - effective_message_unhealthy_tolerance: U57F7::from_num(0.1), - expected_drop_messages: U57F7::from_num(1.0), - drop_message_malicious_tolerance: U57F7::from_num(0.0), - drop_message_unhealthy_tolerance: U57F7::from_num(0.0), + expected_messages: U57F7::from_num(2.0), + message_malicious_tolerance: U57F7::from_num(0.1), + message_unhealthy_tolerance: U57F7::from_num(0.1), }, tokio_stream::wrappers::IntervalStream::new(tokio::time::interval_at( tokio::time::Instant::now() + interval, diff --git a/nomos-blend/core/src/persistent_transmission.rs b/nomos-blend/core/src/persistent_transmission.rs index 32a090c4e..358bf78c7 100644 --- a/nomos-blend/core/src/persistent_transmission.rs +++ b/nomos-blend/core/src/persistent_transmission.rs @@ -4,67 +4,45 @@ use std::{ }; use futures::{Stream, StreamExt as _}; -use rand::{distributions::Uniform, prelude::Distribution as _, Rng, RngCore}; use serde::{Deserialize, Serialize}; #[derive(Clone, Copy, Debug, Serialize, Deserialize)] pub struct PersistentTransmissionSettings { /// The maximum number of messages that can be emitted per second pub max_emission_frequency: f64, - /// The probability of emitting a drop message by coin flipping - pub drop_message_probability: f64, } impl Default for PersistentTransmissionSettings { fn default() -> Self { Self { max_emission_frequency: 1.0, - drop_message_probability: 0.5, } } } /// Transmit scheduled messages with a persistent rate as a stream. -pub struct PersistentTransmissionStream +pub struct PersistentTransmissionStream where MsgStream: Stream, - Rng: RngCore, { - coin: Coin, stream: MsgStream, scheduler: Scheduler, - drop_message: MsgStream::Item, } -impl PersistentTransmissionStream +impl PersistentTransmissionStream where MsgStream: Stream, - Rng: RngCore, Scheduler: Stream, { - pub fn new( - settings: PersistentTransmissionSettings, - stream: MsgStream, - scheduler: Scheduler, - drop_message: MsgStream::Item, - rng: Rng, - ) -> Self { - let coin = Coin::::new(rng, settings.drop_message_probability).unwrap(); - Self { - coin, - stream, - scheduler, - drop_message, - } + pub const fn new(stream: MsgStream, scheduler: Scheduler) -> Self { + Self { stream, scheduler } } } -impl Stream - for PersistentTransmissionStream +impl Stream for PersistentTransmissionStream where MessageStream: Stream + Unpin, MessageStream::Item: Clone + Unpin, - Rng: RngCore + Unpin, Scheduler: Stream + Unpin, { type Item = MessageStream::Item; @@ -73,8 +51,6 @@ where let Self { ref mut scheduler, ref mut stream, - ref mut coin, - ref drop_message, .. } = self.get_mut(); if pin!(scheduler).poll_next_unpin(cx).is_pending() { @@ -82,78 +58,39 @@ where } if let Poll::Ready(Some(item)) = pin!(stream).poll_next(cx) { Poll::Ready(Some(item)) - } else if coin.flip() { - Poll::Ready(Some(drop_message.clone())) } else { Poll::Pending } } } -pub trait PersistentTransmissionExt: Stream +pub trait PersistentTransmissionExt: Stream where - Rng: RngCore, Scheduler: Stream, { fn persistent_transmission( self, - settings: PersistentTransmissionSettings, - rng: Rng, scheduler: Scheduler, - drop_message: Self::Item, - ) -> PersistentTransmissionStream + ) -> PersistentTransmissionStream where Self: Sized + Unpin, { - PersistentTransmissionStream::new(settings, self, scheduler, drop_message, rng) + PersistentTransmissionStream::new(self, scheduler) } } -impl PersistentTransmissionExt for MessageStream +impl PersistentTransmissionExt for MessageStream where MessageStream: Stream, - Rng: RngCore, Scheduler: Stream, { } -struct Coin { - rng: R, - distribution: Uniform, - probability: f64, -} - -impl Coin { - fn new(rng: R, probability: f64) -> Result { - if !(0.0..=1.0).contains(&probability) { - return Err(CoinError::InvalidProbability); - } - Ok(Self { - rng, - distribution: Uniform::from(0.0..1.0), - probability, - }) - } - - // Flip the coin based on the given probability. - fn flip(&mut self) -> bool { - self.distribution.sample(&mut self.rng) < self.probability - } -} - -#[derive(Debug)] -enum CoinError { - InvalidProbability, -} - #[cfg(test)] mod tests { use std::time::Duration; use futures::StreamExt as _; - use nomos_blend_message::{mock::MockBlendMessage, BlendMessage as _}; - use rand::SeedableRng as _; - use rand_chacha::ChaCha8Rng; use tokio::{sync::mpsc, time}; use tokio_stream::wrappers::IntervalStream; @@ -187,22 +124,17 @@ mod tests { let stream = tokio_stream::wrappers::UnboundedReceiverStream::new(schedule_receiver); let settings = PersistentTransmissionSettings { max_emission_frequency: 1.0, - // Set to always emit drop messages if no scheduled messages for easy testing - drop_message_probability: 1.0, }; - // Prepare the expected emission interval with torelance + // Prepare the expected emission interval with tolerance let expected_emission_interval = Duration::from_secs_f64(1.0 / settings.max_emission_frequency); - let torelance = expected_emission_interval / 10; // 10% torelance - let lower_bound = expected_emission_interval - torelance; - let upper_bound = expected_emission_interval + torelance; + let tolerance = expected_emission_interval / 10; // 10% tolerance + let lower_bound = expected_emission_interval - tolerance; + let upper_bound = expected_emission_interval + tolerance; // prepare stream - let mut persistent_transmission_stream: PersistentTransmissionStream<_, _, _> = stream + let mut persistent_transmission_stream: PersistentTransmissionStream<_, _> = stream .persistent_transmission( - settings, - ChaCha8Rng::from_entropy(), IntervalStream::new(time::interval(expected_emission_interval)).map(|_| ()), - MockBlendMessage::DROP_MESSAGE.to_vec(), ); // Messages must be scheduled in non-blocking manner. schedule_sender.send(vec![1]).unwrap(); @@ -228,16 +160,6 @@ mod tests { ); assert_interval!(&mut last_time, lower_bound, upper_bound); - assert!(MockBlendMessage::is_drop( - &persistent_transmission_stream.next().await.unwrap() - )); - assert_interval!(&mut last_time, lower_bound, upper_bound); - - assert!(MockBlendMessage::is_drop( - &persistent_transmission_stream.next().await.unwrap() - )); - assert_interval!(&mut last_time, lower_bound, upper_bound); - // Schedule a new message and check if it is emitted at the next interval schedule_sender.send(vec![4]).unwrap(); assert_eq!( diff --git a/nomos-blend/message/src/lib.rs b/nomos-blend/message/src/lib.rs index 4968a3a64..4f903b864 100644 --- a/nomos-blend/message/src/lib.rs +++ b/nomos-blend/message/src/lib.rs @@ -5,7 +5,6 @@ pub trait BlendMessage { type PublicKey; type PrivateKey; type Error: std::error::Error; - const DROP_MESSAGE: &'static [u8]; fn build(payload: &[u8], public_keys: &[Self::PublicKey]) -> Result, Self::Error>; /// Unwrap the message one layer. @@ -20,10 +19,6 @@ pub trait BlendMessage { message: &[u8], private_key: &Self::PrivateKey, ) -> Result<(Vec, bool), MessageUnwrapError>; - #[must_use] - fn is_drop(message: &[u8]) -> bool { - message == Self::DROP_MESSAGE - } } #[derive(thiserror::Error, Debug)] diff --git a/nomos-blend/message/src/mock/mod.rs b/nomos-blend/message/src/mock/mod.rs index 43e27de3b..d4712e448 100644 --- a/nomos-blend/message/src/mock/mod.rs +++ b/nomos-blend/message/src/mock/mod.rs @@ -23,7 +23,6 @@ impl BlendMessage for MockBlendMessage { type PublicKey = [u8; NODE_ID_SIZE]; type PrivateKey = [u8; NODE_ID_SIZE]; type Error = Error; - const DROP_MESSAGE: &'static [u8] = &[0; MESSAGE_SIZE]; /// The length of the encoded message is fixed to [`MESSAGE_SIZE`] bytes. /// The [`MAX_LAYERS`] number of [`NodeId`]s are concatenated in front of diff --git a/nomos-blend/message/src/sphinx/mod.rs b/nomos-blend/message/src/sphinx/mod.rs index 68269ef99..424a55151 100644 --- a/nomos-blend/message/src/sphinx/mod.rs +++ b/nomos-blend/message/src/sphinx/mod.rs @@ -20,9 +20,6 @@ impl BlendMessage for SphinxMessage { type PublicKey = [u8; ASYM_KEY_SIZE]; type PrivateKey = [u8; ASYM_KEY_SIZE]; type Error = Error; - - const DROP_MESSAGE: &'static [u8] = &[0; Packet::size(MAX_LAYERS, MAX_PAYLOAD_SIZE)]; - fn build(payload: &[u8], public_keys: &[Self::PublicKey]) -> Result, Self::Error> { let packet = Packet::build( &public_keys diff --git a/nomos-blend/network/src/behaviour.rs b/nomos-blend/network/src/behaviour.rs index 06b397611..5471b701d 100644 --- a/nomos-blend/network/src/behaviour.rs +++ b/nomos-blend/network/src/behaviour.rs @@ -91,13 +91,8 @@ where } } - /// Publish a message (data or drop) to all connected peers + /// Publish a message to all connected peers pub fn publish(&mut self, message: &[u8]) -> Result<(), Error> { - if M::is_drop(message) { - // Bypass deduplication for the drop message - return self.forward_message(message, None); - } - let msg_id = Self::message_id(message); // If the message was already seen, don't forward it again if self.seen_message_cache.cache_get(&msg_id).is_some() { @@ -236,7 +231,7 @@ where event: THandlerOutEvent, ) { match event { - // A non-drop message was forwarded from the peer. + // A message was forwarded from the peer. ToBehaviour::Message(message) => { // Add the message to the cache. If it was already seen, ignore it. if self diff --git a/nomos-blend/network/src/handler.rs b/nomos-blend/network/src/handler.rs index 2db23b59b..1330ff7d2 100644 --- a/nomos-blend/network/src/handler.rs +++ b/nomos-blend/network/src/handler.rs @@ -202,22 +202,16 @@ where // Record the message to the monitor. if let Some(monitor) = &mut self.monitor { - if Msg::is_drop(&msg) { - monitor.record_drop_message(); - } else { - monitor.record_effective_message(); - } + monitor.record_message(); } self.inbound_substream = Some(InboundSubstreamState::PendingRecv(recv_msg(stream).boxed())); - // Notify behaviour only on non-drop messages. - if !Msg::is_drop(&msg) { - return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour( - ToBehaviour::Message(msg), - )); - } + // Notify behaviour. + return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour( + ToBehaviour::Message(msg), + )); } Poll::Ready(Err(e)) => { tracing::error!( diff --git a/nomos-blend/network/src/lib.rs b/nomos-blend/network/src/lib.rs index 0638b1a5d..660ba83f9 100644 --- a/nomos-blend/network/src/lib.rs +++ b/nomos-blend/network/src/lib.rs @@ -130,12 +130,9 @@ mod test { // Init two swarms with connection monitoring enabled. let conn_monitor_settings = ConnectionMonitorSettings { interval: Duration::from_secs(1), - expected_effective_messages: U57F7::from_num(0.0), - effective_message_malicious_tolerance: U57F7::from_num(0.0), - effective_message_unhealthy_tolerance: U57F7::from_num(0.0), - expected_drop_messages: U57F7::from_num(0.0), - drop_message_malicious_tolerance: U57F7::from_num(0.0), - drop_message_unhealthy_tolerance: U57F7::from_num(0.0), + expected_messages: U57F7::from_num(0.0), + message_malicious_tolerance: U57F7::from_num(0.0), + message_unhealthy_tolerance: U57F7::from_num(0.0), }; let (mut nodes, mut keypairs) = nodes(2, 8290); let node1_addr = nodes.next().unwrap().address; @@ -151,7 +148,7 @@ mod test { ); swarm2.dial(node1_addr).unwrap(); - // Swarm2 sends a message to Swarm1, even though expected_effective_messages is + // Swarm2 sends a message to Swarm1, even though `expected_messages` is // 0. Then, Swarm1 should detect Swarm2 as a malicious peer. let task = async { let mut num_events_waiting = 2; @@ -204,12 +201,9 @@ mod test { // Init two swarms with connection monitoring enabled. let conn_monitor_settings = ConnectionMonitorSettings { interval: Duration::from_secs(1), - expected_effective_messages: U57F7::from_num(1.0), - effective_message_malicious_tolerance: U57F7::from_num(0.0), - effective_message_unhealthy_tolerance: U57F7::from_num(0.0), - expected_drop_messages: U57F7::from_num(0.0), - drop_message_malicious_tolerance: U57F7::from_num(0.0), - drop_message_unhealthy_tolerance: U57F7::from_num(0.0), + expected_messages: U57F7::from_num(1.0), + message_malicious_tolerance: U57F7::from_num(0.0), + message_unhealthy_tolerance: U57F7::from_num(0.0), }; let (mut nodes, mut keypairs) = nodes(2, 8390); let node1_addr = nodes.next().unwrap().address; @@ -225,7 +219,7 @@ mod test { ); swarm2.dial(node1_addr).unwrap(); - // Swarms don't send anything, even though expected_effective_messages is 1. + // Swarms don't send anything, even though `expected_messages` is 1. // Then, both should detect the other as unhealthy. // Swarms shouldn't close the connection of the unhealthy peers. let task = async { diff --git a/nomos-services/blend/src/lib.rs b/nomos-services/blend/src/lib.rs index 5447c59d0..69a1ac315 100644 --- a/nomos-services/blend/src/lib.rs +++ b/nomos-services/blend/src/lib.rs @@ -128,15 +128,12 @@ where // tier 1 persistent transmission let (persistent_sender, persistent_receiver) = mpsc::unbounded_channel(); - let mut persistent_transmission_messages: PersistentTransmissionStream<_, _, _> = + let mut persistent_transmission_messages: PersistentTransmissionStream<_, _> = UnboundedReceiverStream::new(persistent_receiver).persistent_transmission( - blend_config.persistent_transmission, - ChaCha12Rng::from_entropy(), IntervalStream::new(time::interval(Duration::from_secs_f64( 1.0 / blend_config.persistent_transmission.max_emission_frequency, ))) .map(|_| ()), - SphinxMessage::DROP_MESSAGE.to_vec(), ); // tier 2 blend @@ -187,15 +184,18 @@ where tracing::error!("Error sending message to persistent stream: {e}"); } } - // If the message is fully unwrapped, broadcast it (unencrypted) to the rest of the network. + // If the message is fully unwrapped, broadcast it (unencrypted) to the rest of the network if it's not a cover message. BlendOutgoingMessage::FullyUnwrapped(msg) => { - tracing::debug!("Broadcasting fully unwrapped message"); + tracing::debug!("Processing a fully unwrapped message."); + // TODO: Change deserialization logic to return the actual type of message to the service, instead of assuming that a failed deserialization can mean a cover message as well as a malformed message. match wire::deserialize::>(&msg) { Ok(msg) => { + // Message is a valid network message, broadcast it to the entire network. network_adapter.broadcast(msg.message, msg.broadcast_settings).await; }, _ => { - tracing::debug!("unrecognized message from blend backend"); + // Message failed to be deserialized. It means that it was either malformed, or a cover message. + tracing::debug!("Unrecognized message from blend backend. Either malformed or a cover message. Dropping."); } } }