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

Revert "Reestablish connection when mixclient fails (#445)" (#472) 

This reverts commit aa69eeca00.
This commit is contained in:
Giacomo Pasini 2023-10-23 16:53:20 +02:00 committed by GitHub
parent d528ebd2ea
commit 3ce8cacb30
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 15 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
nomos-services/network/src/backends/libp2p/mixnet.rs
View File

@ -1,6 +1,6 @@
use std::{ops::Range, time::Duration}; use std::{ops::Range, time::Duration};
use mixnet_client::{MessageStream, MixnetClient}; use mixnet_client::MixnetClient;
use nomos_core::wire; use nomos_core::wire;
use rand::{rngs::OsRng, thread_rng, Rng}; use rand::{rngs::OsRng, thread_rng, Rng};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@ -47,26 +47,11 @@ impl MixnetHandler {
} }
pub async fn run(&mut self) { pub async fn run(&mut self) {
const BASE_DELAY: Duration = Duration::from_secs(5); let Ok(mut stream) = self.client.run().await else {
// we need this loop to help us reestablish the connection in case tracing::error!("Could not quickstart mixnet stream");
// the mixnet client fails for whatever reason return;
let mut backoff = 0; };
loop {
match self.client.run().await {
Ok(stream) => {
backoff = 0;
Self::handle_stream(self.commands_tx.clone(), stream).await;
}
Err(e) => {
tracing::error!("mixnet client error: {e}");
backoff += 1;
tokio::time::sleep(BASE_DELAY * backoff).await;
}
}
}
}
async fn handle_stream(tx: mpsc::Sender<Command>, mut stream: MessageStream) {
while let Some(result) = stream.next().await { while let Some(result) = stream.next().await {
match result { match result {
Ok(msg) => { Ok(msg) => {
@ -77,7 +62,8 @@ impl MixnetHandler {
continue; continue;
}; };
tx.send(Command::DirectBroadcastAndRetry { self.commands_tx
.send(Command::DirectBroadcastAndRetry {
topic, topic,
message, message,
retry_count: 0, retry_count: 0,
@ -86,8 +72,7 @@ impl MixnetHandler {
.unwrap_or_else(|_| tracing::error!("could not schedule broadcast")); .unwrap_or_else(|_| tracing::error!("could not schedule broadcast"));
} }
Err(e) => { Err(e) => {
tracing::error!("mixnet client stream error: {e}"); todo!("Handle mixclient error: {e}");
// TODO: handle mixnet client stream error
} }
} }
} }

1
nomos-services/network/src/backends/libp2p/mod.rs
View File

@ -45,6 +45,7 @@ impl NetworkBackend for Libp2p {
fn new(config: Self::Settings, overwatch_handle: OverwatchHandle) -> Self { fn new(config: Self::Settings, overwatch_handle: OverwatchHandle) -> Self {
let (commands_tx, commands_rx) = tokio::sync::mpsc::channel(BUFFER_SIZE); let (commands_tx, commands_rx) = tokio::sync::mpsc::channel(BUFFER_SIZE);
let (events_tx, _) = tokio::sync::broadcast::channel(BUFFER_SIZE); let (events_tx, _) = tokio::sync::broadcast::channel(BUFFER_SIZE);
let mut mixnet_handler = MixnetHandler::new(&config, commands_tx.clone()); let mut mixnet_handler = MixnetHandler::new(&config, commands_tx.clone());
overwatch_handle.runtime().spawn(async move { overwatch_handle.runtime().spawn(async move {
mixnet_handler.run().await; mixnet_handler.run().await;