add persistent transmission + message blend

network-runner/Cargo.toml

@@ -36,6 +36,10 @@ serde_json = "1.0"
 thiserror = "1"
 tracing = { version = "0.1", default-features = false, features = ["log", "attributes"] }
 tracing-subscriber = { version = "0.3", features = ["json", "env-filter", "tracing-log"]}
+nomos-mix = { git = "https://github.com/logos-co/nomos-node", rev = "e095964", package = "nomos-mix" }
+nomos-mix-message = { git = "https://github.com/logos-co/nomos-node", rev = "e095964", package = "nomos-mix-message" }
+rand_chacha = "0.3"
+multiaddr = "0.18"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 getrandom = { version = "0.2", features = ["js"] }

network-runner/src/bin/app/main.rs

@@ -7,6 +7,10 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use anyhow::Ok;
 use clap::Parser;
 use crossbeam::channel;
+use nomos_mix::message_blend::{
+    CryptographicProcessorSettings, MessageBlendSettings, TemporalSchedulerSettings,
+};
+use nomos_mix::persistent_transmission::PersistentTransmissionSettings;
 use nomos_simulations_network_runner::network::behaviour::create_behaviours;
 use nomos_simulations_network_runner::network::regions::{create_regions, RegionsData};
 use nomos_simulations_network_runner::network::{InMemoryNetworkInterface, Network};
@@ -94,6 +98,21 @@ impl SimulationApp {
                             .filter(|&id| id != &node_id)
                             .copied()
                             .choose_multiple(&mut rng, 3),
+                        seed: 0,
+                        persistent_transmission: PersistentTransmissionSettings {
+                            max_emission_frequency: 1.0,
+                            drop_message_probability: 0.5,
+                        },
+                        message_blend: MessageBlendSettings {
+                            cryptographic_processor: CryptographicProcessorSettings {
+                                private_key: node_id.into(),
+                                num_mix_layers: 1,
+                            },
+                            temporal_processor: TemporalSchedulerSettings {
+                                max_delay_seconds: 10,
+                            },
+                        },
+                        membership: node_ids.iter().map(|&id| id.into()).collect(),
                     },
                 )
             })

network-runner/src/node/mix/mod.rs

@@ -1,16 +1,35 @@
+mod scheduler;
 pub mod state;
-mod step_scheduler;
+mod stream_wrapper;
 
 use super::{Node, NodeId};
 use crate::network::{InMemoryNetworkInterface, NetworkInterface, PayloadSize};
+use crossbeam::channel;
+use futures::Stream;
+use multiaddr::Multiaddr;
+use nomos_mix::{
+    membership::Membership,
+    message_blend::{MessageBlendExt, MessageBlendSettings, MessageBlendStream},
+    persistent_transmission::{
+        PersistentTransmissionExt, PersistentTransmissionSettings, PersistentTransmissionStream,
+    },
+    MixOutgoingMessage,
+};
+use nomos_mix_message::mock::MockMixMessage;
+use rand::SeedableRng;
+use rand_chacha::ChaCha12Rng;
+use scheduler::{Interval, TemporalRelease};
 use serde::Deserialize;
 use state::MixnodeState;
-use std::time::Duration;
+use std::{
+    pin::{self},
+    task::Poll,
+    time::Duration,
+};
+use stream_wrapper::CrossbeamReceiverStream;
 
 #[derive(Debug, Clone)]
-pub enum MixMessage {
+pub struct MixMessage(Vec<u8>);
-    Dummy(String),
-}
 
 impl PayloadSize for MixMessage {
     fn size_bytes(&self) -> u32 {
@@ -18,9 +37,13 @@ impl PayloadSize for MixMessage {
     }
 }
 
-#[derive(Clone, Default, Deserialize)]
+#[derive(Clone, Deserialize)]
 pub struct MixnodeSettings {
     pub connected_peers: Vec<NodeId>,
+    pub seed: u64,
+    pub persistent_transmission: PersistentTransmissionSettings,
+    pub message_blend: MessageBlendSettings<MockMixMessage>,
+    pub membership: Vec<<MockMixMessage as nomos_mix_message::MixMessage>::PublicKey>,
 }
 
 /// This node implementation only used for testing different streaming implementation purposes.
@@ -29,6 +52,23 @@ pub struct MixNode {
     state: MixnodeState,
     settings: MixnodeSettings,
     network_interface: InMemoryNetworkInterface<MixMessage>,
+
+    persistent_sender: channel::Sender<Vec<u8>>,
+    persistent_update_time_sender: channel::Sender<Duration>,
+    persistent_transmission_messages: PersistentTransmissionStream<
+        CrossbeamReceiverStream<Vec<u8>>,
+        ChaCha12Rng,
+        MockMixMessage,
+        Interval,
+    >,
+    blend_sender: channel::Sender<Vec<u8>>,
+    blend_update_time_sender: channel::Sender<Duration>,
+    blend_messages: MessageBlendStream<
+        CrossbeamReceiverStream<Vec<u8>>,
+        ChaCha12Rng,
+        MockMixMessage,
+        TemporalRelease,
+    >,
 }
 
 impl MixNode {
@@ -37,6 +77,54 @@ impl MixNode {
         settings: MixnodeSettings,
         network_interface: InMemoryNetworkInterface<MixMessage>,
     ) -> Self {
+        let mut rng_generator = ChaCha12Rng::seed_from_u64(settings.seed);
+
+        // Init Tier-1: Persistent transmission
+        let (persistent_sender, persistent_receiver) = channel::unbounded();
+        let (persistent_update_time_sender, persistent_update_time_receiver) = channel::unbounded();
+        let persistent_transmission_messages = CrossbeamReceiverStream::new(persistent_receiver)
+            .persistent_transmission(
+                settings.persistent_transmission,
+                ChaCha12Rng::from_rng(&mut rng_generator).unwrap(),
+                Interval::new(
+                    Duration::from_secs_f64(
+                        1.0 / settings.persistent_transmission.max_emission_frequency,
+                    ),
+                    persistent_update_time_receiver,
+                ),
+            );
+
+        // Init Tier-2: message blend
+        let (blend_sender, blend_receiver) = channel::unbounded();
+        let (blend_update_time_sender, blend_update_time_receiver) = channel::unbounded();
+        let nodes: Vec<
+            nomos_mix::membership::Node<
+                <MockMixMessage as nomos_mix_message::MixMessage>::PublicKey,
+            >,
+        > = settings
+            .membership
+            .iter()
+            .map(|&public_key| nomos_mix::membership::Node {
+                address: Multiaddr::empty(),
+                public_key,
+            })
+            .collect();
+        let membership = Membership::<MockMixMessage>::new(nodes, id.into());
+        let temporal_release = TemporalRelease::new(
+            ChaCha12Rng::from_rng(&mut rng_generator).unwrap(),
+            blend_update_time_receiver,
+            (
+                1,
+                settings.message_blend.temporal_processor.max_delay_seconds,
+            ),
+        );
+        let blend_messages = CrossbeamReceiverStream::new(blend_receiver).blend(
+            settings.message_blend.clone(),
+            membership,
+            temporal_release,
+            ChaCha12Rng::from_rng(&mut rng_generator).unwrap(),
+        );
+
         Self {
             id,
             network_interface,
@@ -46,8 +134,26 @@ impl MixNode {
                 mock_counter: 0,
                 step_id: 0,
             },
+            persistent_sender,
+            persistent_update_time_sender,
+            persistent_transmission_messages,
+            blend_sender,
+            blend_update_time_sender,
+            blend_messages,
         }
     }
+
+    fn forward(&self, message: MixMessage) {
+        for node_id in self.settings.connected_peers.iter() {
+            self.network_interface
+                .send_message(*node_id, message.clone())
+        }
+    }
+
+    fn update_time(&mut self, elapsed: Duration) {
+        self.persistent_update_time_sender.send(elapsed).unwrap();
+        self.blend_update_time_sender.send(elapsed).unwrap();
+    }
 }
 
 impl Node for MixNode {
@@ -63,20 +169,44 @@ impl Node for MixNode {
         &self.state
     }
 
-    fn step(&mut self, _: Duration) {
+    fn step(&mut self, elapsed: Duration) {
+        self.update_time(elapsed);
+
+        let Self {
+            persistent_sender,
+            persistent_transmission_messages,
+            blend_sender,
+            blend_messages,
+            ..
+        } = self;
+
         let messages = self.network_interface.receive_messages();
         for message in messages {
             println!(">>>>> Node {}, message: {message:?}", self.id);
+            blend_sender.send(message.into_payload().0).unwrap();
+        }
+
+        let waker = futures::task::noop_waker();
+        let mut cx = futures::task::Context::from_waker(&waker);
+        // Proceed message blend
+        if let Poll::Ready(Some(msg)) = pin::pin!(blend_messages).poll_next(&mut cx) {
+            match msg {
+                MixOutgoingMessage::Outbound(msg) => {
+                    persistent_sender.send(msg).unwrap();
+                }
+                MixOutgoingMessage::FullyUnwrapped(_) => {
+                    //TODO: increase counter and create a tracing event
+                }
+            }
+        }
+        // Proceed persistent transmission
+        if let Poll::Ready(Some(msg)) =
+            pin::pin!(persistent_transmission_messages).poll_next(&mut cx)
+        {
+            self.forward(MixMessage(msg));
         }
 
         self.state.step_id += 1;
         self.state.mock_counter += 1;
-
-        for node_id in self.settings.connected_peers.iter() {
-            self.network_interface.send_message(
-                *node_id,
-                MixMessage::Dummy(format!("Hello from node: {}", self.id)),
-            )
-        }
     }
 }

network-runner/src/node/mix/scheduler.rs

@@ -1,18 +1,25 @@
-use chrono::format::Item;
+use crossbeam::channel;
 use futures::Stream;
 use rand::RngCore;
 use std::pin::Pin;
-use std::sync::mpsc;
 use std::task::{Context, Poll};
 use std::time::Duration;
 
-struct Interval {
+pub struct Interval {
     duration: Duration,
     current_elapsed: Duration,
-    update_time: mpsc::Receiver<Duration>,
+    update_time: channel::Receiver<Duration>,
 }
 
 impl Interval {
+    pub fn new(duration: Duration, update_time: channel::Receiver<Duration>) -> Self {
+        Self {
+            duration,
+            current_elapsed: Duration::from_secs(0),
+            update_time,
+        }
+    }
+
     pub fn update(&mut self, elapsed: Duration) -> bool {
         self.current_elapsed += elapsed;
         if self.current_elapsed >= self.duration {
@@ -37,17 +44,17 @@ impl Stream for Interval {
     }
 }
 
-struct TemporalRelease {
+pub struct TemporalRelease {
-    random_sleeps: Box<dyn Iterator<Item = Duration>>,
+    random_sleeps: Box<dyn Iterator<Item = Duration> + Send + Sync + 'static>,
     elapsed: Duration,
     current_sleep: Duration,
-    update_time: mpsc::Receiver<Duration>,
+    update_time: channel::Receiver<Duration>,
 }
 
 impl TemporalRelease {
-    pub fn new<Rng: RngCore + 'static>(
+    pub fn new<Rng: RngCore + Send + Sync + 'static>(
         mut rng: Rng,
-        update_time: mpsc::Receiver<Duration>,
+        update_time: channel::Receiver<Duration>,
         (min_delay, max_delay): (u64, u64),
     ) -> Self {
         let mut random_sleeps = Box::new(std::iter::repeat_with(move || {

network-runner/src/node/mix/stream_wrapper.rs

@@ -0,0 +1,29 @@
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use crossbeam::channel;
+use futures::Stream;
+
+pub struct CrossbeamReceiverStream<T> {
+    receiver: channel::Receiver<T>,
+}
+
+impl<T> CrossbeamReceiverStream<T> {
+    pub fn new(receiver: channel::Receiver<T>) -> Self {
+        Self { receiver }
+    }
+}
+
+impl<T> Stream for CrossbeamReceiverStream<T> {
+    type Item = T;
+
+    fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        match self.receiver.try_recv() {
+            Ok(item) => Poll::Ready(Some(item)),
+            Err(channel::TryRecvError::Empty) => Poll::Pending,
+            Err(channel::TryRecvError::Disconnected) => Poll::Ready(None),
+        }
+    }
+}