From 9aa5b98465a149e7838e252091ff581adcd86367 Mon Sep 17 00:00:00 2001 From: Drazen Urch Date: Thu, 7 Apr 2022 18:13:29 +0200 Subject: [PATCH] Debugging validator (#1198) * Checkpoint * Replace Stream logic with StreamMap * Ignore blacklisted mixnodes and gateways * Moar logging * Remove version checks * Cleanup * Some more cleanup --- Cargo.lock | 1 + .../gateway-client/src/packet_router.rs | 7 +- .../gateway-client/src/socket_state.rs | 2 +- validator-api/Cargo.toml | 1 + .../src/network_monitor/gateways_reader.rs | 176 ++---------------- .../monitor/gateways_pinger.rs | 24 ++- .../src/network_monitor/monitor/mod.rs | 22 ++- .../src/network_monitor/monitor/preparer.rs | 66 ++----- .../src/network_monitor/monitor/receiver.rs | 14 +- .../src/network_monitor/monitor/sender.rs | 8 +- 10 files changed, 84 insertions(+), 237 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 39552decd0..a79522613a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3100,6 +3100,7 @@ dependencies = [ "thiserror", "time 0.3.7", "tokio", + "tokio-stream", "topology", "url", "validator-api-requests", diff --git a/common/client-libs/gateway-client/src/packet_router.rs b/common/client-libs/gateway-client/src/packet_router.rs index 8fb35aed5f..7d650ad695 100644 --- a/common/client-libs/gateway-client/src/packet_router.rs +++ b/common/client-libs/gateway-client/src/packet_router.rs @@ -72,7 +72,12 @@ impl PacketRouter { if !received_acks.is_empty() { trace!("routing acks"); - self.ack_sender.unbounded_send(received_acks).unwrap(); + match self.ack_sender.unbounded_send(received_acks) { + Ok(_) => {} + Err(e) => { + error!("failed to send ack: {:?}", e); + } + }; } } } diff --git a/common/client-libs/gateway-client/src/socket_state.rs b/common/client-libs/gateway-client/src/socket_state.rs index 9fba57208f..5de854921c 100644 --- a/common/client-libs/gateway-client/src/socket_state.rs +++ b/common/client-libs/gateway-client/src/socket_state.rs @@ -74,7 +74,7 @@ impl PartiallyDelegated { // This would also require NOT discarding any text responses here. // TODO: those can return the "send confirmations" - perhaps it should be somehow worked around? - Message::Text(text) => debug!( + Message::Text(text) => trace!( "received a text message - probably a response to some previous query! - {}", text ), diff --git a/validator-api/Cargo.toml b/validator-api/Cargo.toml index c28d62d1af..c9f88f076a 100644 --- a/validator-api/Cargo.toml +++ b/validator-api/Cargo.toml @@ -15,6 +15,7 @@ rust-version = "1.56" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +tokio-stream = "0.1.8" clap = "2.33.0" dirs = "3.0" dotenv = "0.15.0" diff --git a/validator-api/src/network_monitor/gateways_reader.rs b/validator-api/src/network_monitor/gateways_reader.rs index 2d3b140885..53c0fd224c 100644 --- a/validator-api/src/network_monitor/gateways_reader.rs +++ b/validator-api/src/network_monitor/gateways_reader.rs @@ -2,181 +2,41 @@ // SPDX-License-Identifier: Apache-2.0 use crypto::asymmetric::identity; -use futures::stream::Stream; -use futures::task::Context; use gateway_client::{AcknowledgementReceiver, MixnetMessageReceiver}; -use std::pin::Pin; -use std::task::{Poll, Waker}; +use tokio_stream::StreamMap; -/// Constant used to determine maximum number of times the GatewayReader can poll. It basically -/// tries to solve the same problem that `FuturesUnordered` has: https://github.com/rust-lang/futures-rs/issues/2047 -const YIELD_EVERY: usize = 32; - -// TODO: Originally I set it to (identity::PublicKey, Vec>) and I definitely -// had a reason for doing so, but right now I can't remember what that was... pub(crate) type GatewayMessages = Vec>; -pub(crate) struct GatewayChannel { - id: identity::PublicKey, - message_receiver: MixnetMessageReceiver, - ack_receiver: AcknowledgementReceiver, - is_closed: bool, -} - -impl GatewayChannel { - pub(crate) fn new( - id: identity::PublicKey, - message_receiver: MixnetMessageReceiver, - ack_receiver: AcknowledgementReceiver, - ) -> Self { - GatewayChannel { - id, - message_receiver, - ack_receiver, - is_closed: false, - } - } -} - -impl Stream for GatewayChannel { - type Item = Vec>; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - if self.is_closed { - return Poll::Ready(None); - } - - let mut polled = 0; - - // empty the ack channel if anything is on it (we don't care about the content at all at the - // moment) - while let Poll::Ready(_ack) = Pin::new(&mut self.ack_receiver).poll_next(cx) { - polled += 1; - } - - let item = futures::ready!(Pin::new(&mut self.message_receiver).poll_next(cx)); - - match item { - None => Poll::Ready(None), - // if we managed to get an item, try to also read additional ones - Some(mut messages) => { - while let Poll::Ready(new_item) = Pin::new(&mut self.message_receiver).poll_next(cx) - { - polled += 1; - match new_item { - None => { - self.is_closed = true; - cx.waker().wake_by_ref(); - return Poll::Ready(Some(messages)); - } - Some(mut additional_messages) => { - messages.append(&mut additional_messages); - - // it is fine enough to use the same constant here as in the main GatewayReader - // as only a single channel, i.e. the main gateway will be capable - // of returning more than 2 values - if polled >= YIELD_EVERY { - cx.waker().wake_by_ref(); - break; - } - } - } - } - Poll::Ready(Some(messages)) - } - } - } -} - pub(crate) struct GatewaysReader { - latest_read: usize, - // channels: FuturesUnordered, - channels: Vec, - waker: Option, + ack_map: StreamMap, + stream_map: StreamMap, } impl GatewaysReader { pub(crate) fn new() -> Self { GatewaysReader { - latest_read: 0, - channels: Vec::new(), - waker: None, + ack_map: StreamMap::new(), + stream_map: StreamMap::new(), } } - fn remove_nth(&mut self, i: usize) { - self.channels.remove(i); + pub fn stream_map(&mut self) -> &mut StreamMap { + &mut self.stream_map } - // todo: if we find that this method is called frequently, perhaps the vector should get - // replaced with different data structure - pub(crate) fn remove_by_key(&mut self, key: identity::PublicKey) { - match self.channels.iter().position(|item| item.id == key) { - Some(i) => { - self.channels.remove(i); - } - // this shouldn't ever get thrown, so perhaps a panic would be more in order? - None => error!( - "tried to remove gateway reader {} but it doesn't exist!", - key.to_base58_string() - ), - } - } - - fn poll_nth( + pub fn add_recievers( &mut self, - cx: &mut Context<'_>, - i: usize, - ) -> Option>> { - if let Poll::Ready(item) = Pin::new(&mut self.channels[i]).poll_next(cx) { - self.latest_read = i; - match item { - // Some(messages) => return Some(Poll::Ready(Some((self.channels[i].id, messages)))), - Some(messages) => return Some(Poll::Ready(Some(messages))), - // remove dead channel - None => self.remove_nth(i), - } - } - None + id: identity::PublicKey, + message_receiver: MixnetMessageReceiver, + ack_receiver: AcknowledgementReceiver, + ) { + let channel_id = id.to_string(); + self.stream_map.insert(channel_id.clone(), message_receiver); + self.ack_map.insert(channel_id, ack_receiver); } - pub(crate) fn insert_channel(&mut self, channel: GatewayChannel) { - self.channels.push(channel); - if let Some(waker) = self.waker.take() { - waker.wake() - } - } -} - -// TODO: not sure if this will scale well, but I don't know what would be a good alternative, -// perhaps try to somehow incorporate FuturesUnordered? -// also, perhaps reading should be done in parallel? -impl Stream for GatewaysReader { - // item represents gateway that returned messages alongside the actual messages - type Item = GatewayMessages; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - if self.latest_read >= self.channels.len() { - self.latest_read = 0; - } - - // don't start reading from beginning each time to at least slightly help with the bias - for i in self.latest_read..self.channels.len() { - if let Some(item) = self.poll_nth(cx, i) { - return item; - } - } - - for i in 0..self.latest_read { - if let Some(item) = self.poll_nth(cx, i) { - return item; - } - } - - // if we have no channels available, store the waker to be woken when a new one is pushed - if self.channels.is_empty() { - self.waker = Some(cx.waker().clone()) - } - Poll::Pending + pub fn remove_recievers(&mut self, id: &str) { + self.stream_map.remove(id); + self.ack_map.remove(id); } } diff --git a/validator-api/src/network_monitor/monitor/gateways_pinger.rs b/validator-api/src/network_monitor/monitor/gateways_pinger.rs index b62c780db0..f7d47ca302 100644 --- a/validator-api/src/network_monitor/monitor/gateways_pinger.rs +++ b/validator-api/src/network_monitor/monitor/gateways_pinger.rs @@ -44,14 +44,17 @@ impl GatewayPinger { } async fn ping_and_cleanup_all_gateways(&self) { - info!(target: "GatewayPinger", "Pinging all active gateways"); + info!("Pinging all active gateways"); let lock_acquire_start = Instant::now(); let active_gateway_clients_guard = self.gateway_clients.lock().await; - trace!(target: "GatewayPinger", "Acquiring lock took {:?}", Instant::now().duration_since(lock_acquire_start)); + trace!( + "Acquiring lock took {:?}", + Instant::now().duration_since(lock_acquire_start) + ); if active_gateway_clients_guard.is_empty() { - debug!(target: "GatewayPinger", "no gateways to ping"); + debug!("no gateways to ping"); return; } @@ -89,7 +92,6 @@ impl GatewayPinger { { Err(_timeout) => { warn!( - target: "GatewayPinger", "we timed out trying to ping {} - assuming the connection is dead.", active_client.gateway_identity().to_base58_string(), ); @@ -97,7 +99,6 @@ impl GatewayPinger { } Ok(Err(err)) => { warn!( - target: "GatewayPinger", "failed to send ping message to gateway {} - {} - assuming the connection is dead.", active_client.gateway_identity().to_base58_string(), err, @@ -112,25 +113,34 @@ impl GatewayPinger { } } + info!( + "Purging {} gateways, acquiring lock", + clients_to_purge.len() + ); // purge all dead connections // reacquire the guard let lock_acquire_start = Instant::now(); let mut active_gateway_clients_guard = self.gateway_clients.lock().await; - trace!(target: "GatewayPinger", "Acquiring lock took {:?}", Instant::now().duration_since(lock_acquire_start)); + info!( + "Acquiring lock took {:?}", + Instant::now().duration_since(lock_acquire_start) + ); for gateway_id in clients_to_purge.into_iter() { if let Some(removed_handle) = active_gateway_clients_guard.remove(&gateway_id) { if !removed_handle.is_invalid().await { + info!("Handle is invalid, purging"); // it was not invalidated by the packet sender meaning it probably was some unbonded node // that was never cleared self.notify_connection_failure(gateway_id); } + info!("Handle is not invalid, not purged") } } let ping_end = Instant::now(); let time_taken = ping_end.duration_since(ping_start); - debug!(target: "GatewayPinger", "Pinging all active gateways took {:?}", time_taken); + info!("Pinging all active gateways took {:?}", time_taken); } pub(crate) async fn run(&self) { diff --git a/validator-api/src/network_monitor/monitor/mod.rs b/validator-api/src/network_monitor/monitor/mod.rs index ce3df86727..efd034710c 100644 --- a/validator-api/src/network_monitor/monitor/mod.rs +++ b/validator-api/src/network_monitor/monitor/mod.rs @@ -177,9 +177,9 @@ impl Monitor { for entry in results.iter() { if *entry.1 == self.route_test_packets { - debug!("✔️ {} succeeded", entry.0) + info!("✔️ {} succeeded", entry.0) } else { - debug!( + info!( "❌️ {} failed ({}/{} received)", entry.0, entry.1, self.route_test_packets ) @@ -200,7 +200,7 @@ impl Monitor { } async fn prepare_test_routes(&mut self) -> Option> { - info!(target: "Monitor", "Generating test routes..."); + info!("Generating test routes..."); // keep track of nodes that should not be used for route construction let mut blacklist = HashSet::new(); @@ -209,7 +209,7 @@ impl Monitor { let mut current_attempt = 0; // todo: tweak this to something more appropriate - let max_attempts = self.test_routes * 2; + let max_attempts = self.test_routes * 10; 'outer: loop { if current_attempt >= max_attempts { @@ -277,13 +277,12 @@ impl Monitor { .set_new_test_nonce(self.test_nonce) .await; - debug!("Sending packets to all gateways..."); + info!("Sending packets to all gateways..."); self.packet_sender .send_packets(prepared_packets.packets) .await; - debug!( - target: "Monitor", + info!( "Sending is over, waiting for {:?} before checking what we received", self.packet_delivery_timeout ); @@ -293,6 +292,8 @@ impl Monitor { let received = self.received_processor.return_received().await; let total_received = received.len(); + info!("Test routes: {:?}", routes); + info!("Received {}/{} packets", total_received, total_sent); let summary = self.summary_producer.produce_summary( prepared_packets.tested_mixnodes, @@ -310,11 +311,14 @@ impl Monitor { } async fn test_run(&mut self) { - info!(target: "Monitor", "Starting test run no. {}", self.test_nonce); + info!("Starting test run no. {}", self.test_nonce); let start = Instant::now(); if let Some(test_routes) = self.prepare_test_routes().await { - debug!(target: "Monitor", "Determined reliable routes to test all other nodes against. : {:?}", test_routes); + info!( + "Determined reliable routes to test all other nodes against. : {:?}", + test_routes + ); self.test_network_against(&test_routes).await; } else { error!("We failed to construct sufficient number of test routes to test the network against") diff --git a/validator-api/src/network_monitor/monitor/preparer.rs b/validator-api/src/network_monitor/monitor/preparer.rs index 2b48c8da1d..2169fc3cba 100644 --- a/validator-api/src/network_monitor/monitor/preparer.rs +++ b/validator-api/src/network_monitor/monitor/preparer.rs @@ -25,6 +25,7 @@ type Id = String; type Owner = Addr; #[derive(Clone)] +#[allow(dead_code)] pub(crate) enum InvalidNode { Outdated(Id, Owner, Version), Malformed(Id, Owner), @@ -188,8 +189,8 @@ impl PacketPreparer { info!("Waiting for minimal topology to be online"); let initialisation_backoff = Duration::from_secs(30); loop { - let gateways = self.validator_cache.gateways().await; - let mixnodes = self.validator_cache.rewarded_set().await; + let gateways = self.validator_cache.gateways_all().await; + let mixnodes = self.validator_cache.mixnodes_all().await; if gateways.len() < minimum_full_routes { info!( @@ -201,7 +202,7 @@ impl PacketPreparer { } let mut layered_mixes = HashMap::new(); - for mix in mixnodes.into_inner() { + for mix in mixnodes { let layer = mix.layer; let mixes = layered_mixes.entry(layer).or_insert_with(Vec::new); mixes.push(mix) @@ -229,7 +230,7 @@ impl PacketPreparer { } async fn all_mixnodes_and_gateways(&self) -> (Vec, Vec) { - info!(target: "Monitor", "Obtaining network topology..."); + info!("Obtaining network topology..."); let mixnodes = self.validator_cache.mixnodes_all().await; let gateways = self.validator_cache.gateways_all().await; @@ -239,9 +240,6 @@ impl PacketPreparer { pub(crate) fn try_parse_mix_bond(&self, mix: &MixNodeBond) -> Result { let identity = mix.mix_node.identity_key.clone(); - if !self.check_version_compatibility(&mix.mix_node.version) { - return Err(identity); - } mix.try_into().map_err(|_| identity) } @@ -250,9 +248,6 @@ impl PacketPreparer { gateway: &GatewayBond, ) -> Result { let identity = gateway.gateway.identity_key.clone(); - if !self.check_version_compatibility(&gateway.gateway.version) { - return Err(identity); - } gateway.try_into().map_err(|_| identity) } @@ -270,19 +265,10 @@ impl PacketPreparer { // separate mixes into layers for easier selection let mut layered_mixes = HashMap::new(); for mix in mixnodes { - // filter out mixes on the blacklist - if blacklist.contains(&mix.mix_node.identity_key) { - continue; - } let layer = mix.layer; let mixes = layered_mixes.entry(layer).or_insert_with(Vec::new); mixes.push(mix) } - // filter out gateways on the blacklist - let gateways = gateways - .into_iter() - .filter(|gateway| !blacklist.contains(&gateway.gateway.identity_key)) - .collect::>(); // get all nodes from each layer... let l1 = layered_mixes.get(&Layer::One)?; @@ -297,9 +283,6 @@ impl PacketPreparer { let rand_l3 = l3.choose_multiple(&mut rng, n).collect::>(); let rand_gateways = gateways.choose_multiple(&mut rng, n).collect::>(); - // let rand_gateways = gateways.iter().filter(|g| g.gateway.host == "109.74.196.254".to_string()).collect::>(); - // let rand_gateways = gateways.iter().filter(|g| g.gateway.host == "185.3.94.33".to_string()).collect::>(); - // the unwrap on `min()` is fine as we know the iterator is not empty let most_available = *[ rand_l1.len(), @@ -312,9 +295,10 @@ impl PacketPreparer { .unwrap(); if most_available == 0 { - // it's impossible to generate a single route + error!("Cannot construct test routes. No nodes or gateways available"); None } else { + trace!("Generating test routes..."); let mut routes = Vec::new(); for i in 0..most_available { let node_1 = match self.try_parse_mix_bond(rand_l1[i]) { @@ -358,14 +342,11 @@ impl PacketPreparer { gateway, )) } + info!("{:?}", routes); Some(routes) } } - fn check_version_compatibility(&self, node_version: &str) -> bool { - version_checker::is_minor_version_compatible(node_version, &self.system_version) - } - fn create_packet_sender(&self, gateway: &gateway::Node) -> Recipient { Recipient::new( self.self_public_identity, @@ -404,14 +385,6 @@ impl PacketPreparer { let mut parsed_nodes = Vec::new(); let mut invalid_nodes = Vec::new(); for mixnode in nodes { - if !self.check_version_compatibility(&mixnode.mix_node.version) { - invalid_nodes.push(InvalidNode::Outdated( - mixnode.mix_node.identity_key, - mixnode.owner, - mixnode.mix_node.version, - )); - continue; - } if let Ok(parsed_node) = (&mixnode).try_into() { parsed_nodes.push(parsed_node) } else { @@ -431,14 +404,6 @@ impl PacketPreparer { let mut parsed_nodes = Vec::new(); let mut invalid_nodes = Vec::new(); for gateway in nodes { - if !self.check_version_compatibility(&gateway.gateway.version) { - invalid_nodes.push(InvalidNode::Outdated( - gateway.gateway.identity_key, - gateway.owner, - gateway.gateway.version, - )); - continue; - } if let Ok(parsed_node) = (&gateway).try_into() { parsed_nodes.push(parsed_node) } else { @@ -460,18 +425,17 @@ impl PacketPreparer { // (remember that "idle" nodes are still part of that set) // we don't care about other nodes, i.e. nodes that are bonded but will not get // any reward during the current rewarding interval - let (rewarded_set, all_gateways) = self.all_mixnodes_and_gateways().await; + let (mixnodes, gateways) = self.all_mixnodes_and_gateways().await; - let (mixes, invalid_mixnodes) = self.filter_outdated_and_malformed_mixnodes(rewarded_set); - let (gateways, invalid_gateways) = - self.filter_outdated_and_malformed_gateways(all_gateways); + let (mixnodes, invalid_mixnodes) = self.filter_outdated_and_malformed_mixnodes(mixnodes); + let (gateways, invalid_gateways) = self.filter_outdated_and_malformed_gateways(gateways); - let tested_mixnodes = mixes.iter().map(|node| node.into()).collect::>(); + let tested_mixnodes = mixnodes.iter().map(|node| node.into()).collect::>(); let tested_gateways = gateways.iter().map(|node| node.into()).collect::>(); let packets_to_create = (test_routes.len() * self.per_node_test_packets) * (tested_mixnodes.len() + tested_gateways.len()); - info!(target: "TestPreparer", "Need to create {} mix packets", packets_to_create); + info!("Need to create {} mix packets", packets_to_create); let mut all_gateway_packets = HashMap::new(); @@ -483,10 +447,10 @@ impl PacketPreparer { let gateway_owner = test_route.gateway_owner(); // it's actually going to be a tiny bit more due to gateway testing, but it's a good enough approximation - let mut mix_packets = Vec::with_capacity(mixes.len() * self.per_node_test_packets); + let mut mix_packets = Vec::with_capacity(mixnodes.len() * self.per_node_test_packets); // and for each mixnode... - for mixnode in &mixes { + for mixnode in &mixnodes { let test_packet = TestPacket::from_mixnode(mixnode, test_route.id(), test_nonce); let topology = test_route.substitute_mix(mixnode); // produce n mix packets diff --git a/validator-api/src/network_monitor/monitor/receiver.rs b/validator-api/src/network_monitor/monitor/receiver.rs index 5cefb3894d..f837cb9f80 100644 --- a/validator-api/src/network_monitor/monitor/receiver.rs +++ b/validator-api/src/network_monitor/monitor/receiver.rs @@ -1,7 +1,7 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::network_monitor::gateways_reader::{GatewayChannel, GatewayMessages, GatewaysReader}; +use crate::network_monitor::gateways_reader::{GatewayMessages, GatewaysReader}; use crate::network_monitor::monitor::processor::ReceivedProcessorSender; use crypto::asymmetric::identity; use futures::channel::mpsc; @@ -40,10 +40,12 @@ impl PacketReceiver { fn process_gateway_update(&mut self, update: GatewayClientUpdate) { match update { GatewayClientUpdate::New(id, (message_receiver, ack_receiver)) => { - let channel = GatewayChannel::new(id, message_receiver, ack_receiver); - self.gateways_reader.insert_channel(channel); + self.gateways_reader + .add_recievers(id, message_receiver, ack_receiver); + } + GatewayClientUpdate::Failure(id) => { + self.gateways_reader.remove_recievers(&id.to_string()); } - GatewayClientUpdate::Failure(id) => self.gateways_reader.remove_by_key(id), } } @@ -62,7 +64,9 @@ impl PacketReceiver { // similarly gateway reader will never return a `None` as it's implemented // as an infinite stream that returns Poll::Pending if it doesn't have anything // to return - messages = self.gateways_reader.next() => self.process_gateway_messages(messages.unwrap()), + Some((_gateway_id, message)) = self.gateways_reader.stream_map().next() => { + self.process_gateway_messages(message) + } } } } diff --git a/validator-api/src/network_monitor/monitor/sender.rs b/validator-api/src/network_monitor/monitor/sender.rs index f650cc4b04..29fd943321 100644 --- a/validator-api/src/network_monitor/monitor/sender.rs +++ b/validator-api/src/network_monitor/monitor/sender.rs @@ -232,14 +232,13 @@ impl PacketSender { ) -> Result<(), GatewayClientError> { let gateway_id = client.gateway_identity().to_base58_string(); info!( - target: "MessageSender", "Got {} packets to send to gateway {}", mix_packets.len(), gateway_id ); if mix_packets.len() <= max_sending_rate { - debug!(target: "MessageSender","Everything is going to get sent as one."); + debug!("Everything is going to get sent as one."); client.batch_send_mix_packets(mix_packets).await?; } else { let packets_per_time_chunk = @@ -248,7 +247,6 @@ impl PacketSender { let total_expected_time = Duration::from_secs_f64(mix_packets.len() as f64 / max_sending_rate as f64); info!( - target: "MessageSender", "With our rate of {} packets/s it should take around {:?} to send it all to {} ...", max_sending_rate, total_expected_time, gateway_id ); @@ -268,7 +266,7 @@ impl PacketSender { // this way we won't have to do reallocations in here as they're unavoidable when // splitting a vector into multiple vectors while let Some(retained) = split_off_vec(&mut mix_packets, packets_per_time_chunk) { - trace!(target: "MessageSender","Sending {} packets...", mix_packets.len()); + trace!("Sending {} packets...", mix_packets.len()); if mix_packets.len() == 1 { client.send_mix_packet(mix_packets.pop().unwrap()).await?; @@ -280,7 +278,7 @@ impl PacketSender { mix_packets = retained; } - debug!(target: "MessageSender", "Done sending"); + debug!("Done sending"); } Ok(())