logos-co
/
nomos-node
mirror of
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Mix: Integrate cover traffic to mix service (#920)

This commit is contained in:
Youngjoon Lee 2024-11-28 10:36:43 +09:00 committed by GitHub
parent dc286a967a
commit 63c472e172
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 134 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
nodes/nomos-node/config.yaml
View File

@ -62,6 +62,13 @@ mix:
num_mix_layers: 1
temporal_processor:
max_delay_seconds: 5
cover_traffic:
epoch_duration:
secs: 432000
nanos: 0
slot_duration:
secs: 20
nanos: 0
membership:
- address: /ip4/127.0.0.1/udp/3001/quic-v1
public_key: [1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1]

2
nomos-mix/core/src/cover_traffic.rs
View File

@ -122,7 +122,7 @@ mod tests {
#[test]
fn test_ticket() {
generate_ticket(10u32.to_be_bytes(), 1123, 0);
for i in (0..1u32) {
for i in 0..1u32 {
let slots = select_slot(i.to_be_bytes(), 1234, 100, 21600, winning_probability(1));
println!("slots = {slots:?}");
}

25
nomos-mix/core/src/membership.rs
View File

@ -9,6 +9,7 @@ where
M: MixMessage,
{
remote_nodes: Vec<Node<M::PublicKey>>,
local_node: Node<M::PublicKey>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
@ -22,10 +23,20 @@ where
M: MixMessage,
M::PublicKey: PartialEq,
{
pub fn new(mut nodes: Vec<Node<M::PublicKey>>, local_public_key: M::PublicKey) -> Self {
nodes.retain(|node| node.public_key != local_public_key);
pub fn new(nodes: Vec<Node<M::PublicKey>>, local_public_key: M::PublicKey) -> Self {
let mut remote_nodes = Vec::with_capacity(nodes.len() - 1);
let mut local_node = None;
nodes.into_iter().for_each(|node| {
if node.public_key == local_public_key {
local_node = Some(node);
} else {
remote_nodes.push(node);
}
});
Self {
remote_nodes: nodes,
remote_nodes,
local_node: local_node.expect("Local node not found"),
}
}
@ -36,4 +47,12 @@ where
) -> Vec<&Node<M::PublicKey>> {
self.remote_nodes.choose_multiple(rng, amount).collect()
}
pub fn local_node(&self) -> &Node<M::PublicKey> {
&self.local_node
}
pub fn size(&self) -> usize {
self.remote_nodes.len() + 1
}
}

4
nomos-services/data-availability/tests/src/common.rs
View File

@ -229,6 +229,10 @@ pub fn new_node(
max_delay_seconds: 2,
},
},
cover_traffic: nomos_mix_service::CoverTrafficExtSettings {
epoch_duration: Duration::from_secs(432000),
slot_duration: Duration::from_secs(20),
},
membership: mix_config.membership.clone(),
},
da_network: DaNetworkConfig {

107
nomos-services/mix/src/lib.rs
View File

@ -6,15 +6,18 @@ use backends::MixBackend;
use futures::StreamExt;
use network::NetworkAdapter;
use nomos_core::wire;
use nomos_mix::membership::{Membership, Node};
use nomos_mix::message_blend::crypto::CryptographicProcessor;
use nomos_mix::message_blend::temporal::TemporalScheduler;
use nomos_mix::message_blend::{crypto::CryptographicProcessor, CryptographicProcessorSettings};
use nomos_mix::message_blend::{MessageBlendExt, MessageBlendSettings};
use nomos_mix::persistent_transmission::{
PersistentTransmissionExt, PersistentTransmissionSettings, PersistentTransmissionStream,
};
use nomos_mix::MixOutgoingMessage;
use nomos_mix_message::mock::MockMixMessage;
use nomos_mix::{
cover_traffic::{CoverTraffic, CoverTrafficSettings},
membership::{Membership, Node},
};
use nomos_mix_message::{mock::MockMixMessage, MixMessage};
use nomos_network::NetworkService;
use overwatch_rs::services::{
handle::ServiceStateHandle,
@ -127,12 +130,22 @@ where
ChaCha12Rng::from_entropy(),
);
let mut blend_messages = backend.listen_to_incoming_messages().blend(
mix_config.message_blend,
mix_config.message_blend.clone(),
membership.clone(),
temporal_scheduler,
ChaCha12Rng::from_entropy(),
);
// tier 3 cover traffic
let mut cover_traffic: CoverTraffic<_, _, MockMixMessage> = CoverTraffic::new(
mix_config.cover_traffic.cover_traffic_settings(
&membership,
&mix_config.message_blend.cryptographic_processor,
),
mix_config.cover_traffic.epoch_stream(),
mix_config.cover_traffic.slot_stream(),
);
// local messages, are bypassed and send immediately
let mut local_messages = service_state
.inbound_relay
@ -162,23 +175,17 @@ where
network_adapter.broadcast(msg.message, msg.broadcast_settings).await;
},
_ => {
tracing::error!("unrecognized message from mix backend");
tracing::debug!("unrecognized message from mix backend");
}
}
}
}
}
Some(msg) = cover_traffic.next() => {
Self::wrap_and_send_to_persistent_transmission(msg, &mut cryptographic_processor, &persistent_sender);
}
Some(msg) = local_messages.next() => {
match cryptographic_processor.wrap_message(&msg) {
Ok(wrapped_message) => {
if let Err(e) = persistent_sender.send(wrapped_message) {
tracing::error!("Error sending message to persistent stream: {e}");
}
}
Err(e) => {
tracing::error!("Failed to wrap message: {:?}", e);
}
}
Self::wrap_and_send_to_persistent_transmission(msg, &mut cryptographic_processor, &persistent_sender);
}
Some(msg) = lifecycle_stream.next() => {
if Self::should_stop_service(msg).await {
@ -214,6 +221,23 @@ where
}
}
}
fn wrap_and_send_to_persistent_transmission(
message: Vec<u8>,
cryptographic_processor: &mut CryptographicProcessor<ChaCha12Rng, MockMixMessage>,
persistent_sender: &mpsc::UnboundedSender<Vec<u8>>,
) {
match cryptographic_processor.wrap_message(&message) {
Ok(wrapped_message) => {
if let Err(e) = persistent_sender.send(wrapped_message) {
tracing::error!("Error sending message to persistent stream: {e}");
}
}
Err(e) => {
tracing::error!("Failed to wrap message: {:?}", e);
}
}
}
}
#[derive(Serialize, Deserialize, Clone, Debug)]
@ -221,11 +245,64 @@ pub struct MixConfig<BackendSettings> {
pub backend: BackendSettings,
pub message_blend: MessageBlendSettings<MockMixMessage>,
pub persistent_transmission: PersistentTransmissionSettings,
pub cover_traffic: CoverTrafficExtSettings,
pub membership: Vec<
Node<<nomos_mix_message::mock::MockMixMessage as nomos_mix_message::MixMessage>::PublicKey>,
>,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct CoverTrafficExtSettings {
pub epoch_duration: Duration,
pub slot_duration: Duration,
}
impl CoverTrafficExtSettings {
fn cover_traffic_settings(
&self,
membership: &Membership<MockMixMessage>,
cryptographic_processor_settings: &CryptographicProcessorSettings<
<MockMixMessage as MixMessage>::PrivateKey,
>,
) -> CoverTrafficSettings {
CoverTrafficSettings {
node_id: membership.local_node().public_key,
number_of_hops: cryptographic_processor_settings.num_mix_layers,
slots_per_epoch: self.slots_per_epoch(),
network_size: membership.size(),
}
}
fn slots_per_epoch(&self) -> usize {
(self.epoch_duration.as_secs() as usize)
.checked_div(self.slot_duration.as_secs() as usize)
.expect("Invalid epoch & slot duration")
}
fn epoch_stream(
&self,
) -> futures::stream::Map<
futures::stream::Enumerate<IntervalStream>,
impl FnMut((usize, time::Instant)) -> usize,
> {
IntervalStream::new(time::interval(self.epoch_duration))
.enumerate()
.map(|(i, _)| i)
}
fn slot_stream(
&self,
) -> futures::stream::Map<
futures::stream::Enumerate<IntervalStream>,
impl FnMut((usize, time::Instant)) -> usize,
> {
let slots_per_epoch = self.slots_per_epoch();
IntervalStream::new(time::interval(self.slot_duration))
.enumerate()
.map(move |(i, _)| i % slots_per_epoch)
}
}
impl<BackendSettings> MixConfig<BackendSettings> {
fn membership(&self) -> Membership<MockMixMessage> {
// We use private key as a public key because the `MockMixMessage` doesn't differentiate between them.

4
tests/src/nodes/executor.rs
View File

@ -166,6 +166,10 @@ pub fn create_executor_config(config: GeneralConfig) -> Config {
max_delay_seconds: 2,
},
},
cover_traffic: nomos_mix_service::CoverTrafficExtSettings {
epoch_duration: Duration::from_secs(432000),
slot_duration: Duration::from_secs(20),
},
membership: config.mix_config.membership,
},
cryptarchia: CryptarchiaSettings {

4
tests/src/nodes/validator.rs
View File

@ -252,6 +252,10 @@ pub fn create_validator_config(config: GeneralConfig) -> Config {
max_delay_seconds: 2,
},
},
cover_traffic: nomos_mix_service::CoverTrafficExtSettings {
epoch_duration: Duration::from_secs(432000),
slot_duration: Duration::from_secs(20),
},
membership: config.mix_config.membership,
},
cryptarchia: CryptarchiaSettings {