logos-co/nomos-node
1
0
Fork 0
mirror of https://github.com/logos-co/nomos-node.git synced 2026-08-31 03:21:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: remove drop messages from blend code (#1340)

Browse Source 
* Remove mentions of drop messages

* Add comment for ignored cover messages upon deserialization failure
This commit is contained in:
Antonio
2025-06-10 08:52:52 +02:00
committed by GitHub
parent 4110eaae15
commit 07032a52ed
10 changed files with 72 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nomos-blend/core/Cargo.toml
+1 -2
View File
@@ -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"] }
nomos-blend/core/src/conn_maintenance.rs
+35 -88
View File
@@ -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<Box<dyn futures::Stream<Item = ()> + 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,
nomos-blend/core/src/persistent_transmission.rs
+14 -92
View File
@@ -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<MsgStream, Rng, Scheduler>
pub struct PersistentTransmissionStream<MsgStream, Scheduler>
where
MsgStream: Stream,
Rng: RngCore,
{
coin: Coin<Rng>,
stream: MsgStream,
scheduler: Scheduler,
drop_message: MsgStream::Item,
}
impl<MsgStream, Rng, Scheduler> PersistentTransmissionStream<MsgStream, Rng, Scheduler>
impl<MsgStream, Scheduler> PersistentTransmissionStream<MsgStream, Scheduler>
where
MsgStream: Stream,
Rng: RngCore,
Scheduler: Stream<Item = ()>,
{
pub fn new(
settings: PersistentTransmissionSettings,
stream: MsgStream,
scheduler: Scheduler,
drop_message: MsgStream::Item,
rng: Rng,
) -> Self {
let coin = Coin::<Rng>::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<MessageStream, Rng, Scheduler> Stream
for PersistentTransmissionStream<MessageStream, Rng, Scheduler>
impl<MessageStream, Scheduler> Stream for PersistentTransmissionStream<MessageStream, Scheduler>
where
MessageStream: Stream + Unpin,
MessageStream::Item: Clone + Unpin,
Rng: RngCore + Unpin,
Scheduler: Stream<Item = ()> + 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<Rng, Scheduler>: Stream
pub trait PersistentTransmissionExt<Scheduler>: Stream
where
Rng: RngCore,
Scheduler: Stream<Item = ()>,
{
fn persistent_transmission(
self,
settings: PersistentTransmissionSettings,
rng: Rng,
scheduler: Scheduler,
drop_message: Self::Item,
) -> PersistentTransmissionStream<Self, Rng, Scheduler>
) -> PersistentTransmissionStream<Self, Scheduler>
where
Self: Sized + Unpin,
{
PersistentTransmissionStream::new(settings, self, scheduler, drop_message, rng)
PersistentTransmissionStream::new(self, scheduler)
}
}
impl<MessageStream, Rng, Scheduler> PersistentTransmissionExt<Rng, Scheduler> for MessageStream
impl<MessageStream, Scheduler> PersistentTransmissionExt<Scheduler> for MessageStream
where
MessageStream: Stream,
Rng: RngCore,
Scheduler: Stream<Item = ()>,
{
}
struct Coin<R: Rng> {
rng: R,
distribution: Uniform<f64>,
probability: f64,
}
impl<R: Rng> Coin<R> {
fn new(rng: R, probability: f64) -> Result<Self, CoinError> {
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!(
nomos-blend/message/src/lib.rs
-5
View File
@@ -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<Vec<u8>, Self::Error>;
/// Unwrap the message one layer.
@@ -20,10 +19,6 @@ pub trait BlendMessage {
message: &[u8],
private_key: &Self::PrivateKey,
) -> Result<(Vec<u8>, bool), MessageUnwrapError<Self::Error>>;
#[must_use]
fn is_drop(message: &[u8]) -> bool {
message == Self::DROP_MESSAGE
}
}
#[derive(thiserror::Error, Debug)]
nomos-blend/message/src/mock/mod.rs
-1
View File
@@ -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
nomos-blend/message/src/sphinx/mod.rs
-3
View File
@@ -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<Vec<u8>, Self::Error> {
let packet = Packet::build(
&public_keys
nomos-blend/network/src/behaviour.rs
+2 -7
View File
@@ -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<Self>,
) {
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
nomos-blend/network/src/handler.rs
+5 -11
View File
@@ -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!(
nomos-blend/network/src/lib.rs
+8 -14
View File
@@ -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 {
nomos-services/blend/src/lib.rs
+7 -7
View File
@@ -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::<NetworkMessage<Network::BroadcastSettings>>(&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.");
}
}
}