diff --git a/Cargo.lock b/Cargo.lock
index 9974f92fa7..188924826d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4180,6 +4180,7 @@ dependencies = [
  "tokio",
  "tokio-stream",
  "tokio-tungstenite",
+ "tokio-util",
  "tungstenite",
  "url",
  "wasm-bindgen",
@@ -4956,6 +4957,7 @@ dependencies = [
  "time",
  "tokio",
  "tokio-tungstenite",
+ "tokio-util",
  "url",
  "zeroize",
 ]
@@ -5300,6 +5302,7 @@ dependencies = [
  "tap",
  "thiserror",
  "tokio",
+ "tokio-util",
  "url",
 ]
 
diff --git a/common/client-core/Cargo.toml b/common/client-core/Cargo.toml
index 6462e98154..773a7402ba 100644
--- a/common/client-core/Cargo.toml
+++ b/common/client-core/Cargo.toml
@@ -26,6 +26,7 @@ tap = { workspace = true }
 thiserror = { workspace = true }
 url = { workspace = true, features = ["serde"] }
 tokio = { workspace = true, features = ["macros"] }
+tokio-util = { workspace = true }
 time = { workspace = true }
 zeroize = { workspace = true }
 
@@ -47,7 +48,9 @@ nym-validator-client = { path = "../client-libs/validator-client", default-featu
 nym-task = { path = "../task" }
 nym-credential-storage = { path = "../credential-storage" }
 nym-network-defaults = { path = "../network-defaults" }
-nym-client-core-config-types = { path = "./config-types", features = ["disk-persistence"] }
+nym-client-core-config-types = { path = "./config-types", features = [
+    "disk-persistence",
+] }
 nym-client-core-surb-storage = { path = "./surb-storage" }
 nym-client-core-gateways-storage = { path = "./gateways-storage" }
 
diff --git a/common/client-core/src/client/base_client/mod.rs b/common/client-core/src/client/base_client/mod.rs
index 3d159fb485..c50e7ec7f8 100644
--- a/common/client-core/src/client/base_client/mod.rs
+++ b/common/client-core/src/client/base_client/mod.rs
@@ -59,6 +59,7 @@ use std::fmt::Debug;
 use std::os::raw::c_int as RawFd;
 use std::path::Path;
 use std::sync::Arc;
+use tokio_util::sync::PollSender;
 use url::Url;
 
 #[cfg(all(
@@ -82,7 +83,11 @@ impl ClientInput {
         &self,
         message: InputMessage,
     ) -> Result<(), tokio::sync::mpsc::error::SendError<InputMessage>> {
-        self.input_sender.send(message).await
+        if let Some(channel) = self.input_sender.get_ref() {
+            channel.send(message).await
+        } else {
+            Err(tokio::sync::mpsc::error::SendError(message))
+        }
     }
 }
 
@@ -804,7 +809,7 @@ where
             client_input: ClientInputStatus::AwaitingProducer {
                 client_input: ClientInput {
                     connection_command_sender: client_connection_tx,
-                    input_sender,
+                    input_sender: PollSender::new(input_sender),
                 },
             },
             client_output: ClientOutputStatus::AwaitingConsumer {
diff --git a/common/client-core/src/client/inbound_messages.rs b/common/client-core/src/client/inbound_messages.rs
index baf163913f..9ddc981d64 100644
--- a/common/client-core/src/client/inbound_messages.rs
+++ b/common/client-core/src/client/inbound_messages.rs
@@ -7,7 +7,7 @@ use nym_sphinx::forwarding::packet::MixPacket;
 use nym_sphinx::params::PacketType;
 use nym_task::connections::TransmissionLane;
 
-pub type InputMessageSender = tokio::sync::mpsc::Sender<InputMessage>;
+pub type InputMessageSender = tokio_util::sync::PollSender<InputMessage>;
 pub type InputMessageReceiver = tokio::sync::mpsc::Receiver<InputMessage>;
 
 #[derive(Debug)]
diff --git a/common/socks5-client-core/Cargo.toml b/common/socks5-client-core/Cargo.toml
index 344b60cd64..17194a729e 100644
--- a/common/socks5-client-core/Cargo.toml
+++ b/common/socks5-client-core/Cargo.toml
@@ -19,6 +19,7 @@ serde = { workspace = true, features = ["derive"] } # for config serialization/d
 tap = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "net", "signal"] }
+tokio-util = { workspace = true }
 url = { workspace = true }
 
 nym-bandwidth-controller = { path = "../../common/bandwidth-controller" }
diff --git a/common/socks5-client-core/src/socks/client.rs b/common/socks5-client-core/src/socks/client.rs
index cca9cd6b53..d7908a1508 100644
--- a/common/socks5-client-core/src/socks/client.rs
+++ b/common/socks5-client-core/src/socks/client.rs
@@ -7,6 +7,7 @@ use super::{SocksVersion, RESERVED, SOCKS4_VERSION, SOCKS5_VERSION};
 use crate::config;
 use futures::channel::mpsc;
 use futures::task::{Context, Poll};
+use futures::SinkExt;
 use log::*;
 use nym_client_core::client::inbound_messages::{InputMessage, InputMessageSender};
 use nym_service_providers_common::interface::{ProviderInterfaceVersion, RequestVersion};
diff --git a/common/socks5/proxy-helpers/src/ordered_sender.rs b/common/socks5/proxy-helpers/src/ordered_sender.rs
index d71f843558..a94cfe97dc 100644
--- a/common/socks5/proxy-helpers/src/ordered_sender.rs
+++ b/common/socks5/proxy-helpers/src/ordered_sender.rs
@@ -7,7 +7,7 @@ use log::{debug, error};
 use nym_socks5_requests::{ConnectionId, SocketData};
 use std::io;
 
-pub(crate) struct OrderedMessageSender<F, S> {
+pub(crate) struct OrderedMessageSender<F, S: Send> {
     connection_id: ConnectionId,
     // addresses are provided for better logging
     local_destination_address: String,
@@ -18,7 +18,7 @@ pub(crate) struct OrderedMessageSender<F, S> {
     mix_message_adapter: F,
 }
 
-impl<F, S> OrderedMessageSender<F, S>
+impl<F, S: Send> OrderedMessageSender<F, S>
 where
     F: Fn(SocketData) -> S,
 {
@@ -56,8 +56,10 @@ where
     }
 
     async fn send_message(&self, message: S) {
-        if self.mixnet_sender.send(message).await.is_err() {
-            panic!("BatchRealMessageReceiver has stopped receiving!")
+        if let Some(sender) = self.mixnet_sender.get_ref() {
+            if sender.send(message).await.is_err() {
+                panic!("BatchRealMessageReceiver has stopped receiving!")
+            }
         }
     }
 
diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs
index 82e6396a5f..60bd0bb786 100644
--- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs
+++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs
@@ -74,7 +74,7 @@ async fn wait_for_lane(
     }
 }
 
-pub(super) async fn run_inbound<F, S>(
+pub(super) async fn run_inbound<F, S: Send>(
     mut reader: OwnedReadHalf,
     mut message_sender: OrderedMessageSender<F, S>,
     connection_id: ConnectionId,
diff --git a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs
index 38499ea59c..730e4b8eb0 100644
--- a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs
+++ b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs
@@ -6,6 +6,7 @@ use crate::ordered_sender::OrderedMessageSender;
 use nym_socks5_requests::{ConnectionId, SocketData};
 use nym_task::connections::LaneQueueLengths;
 use nym_task::TaskClient;
+use tokio_util::sync::PollSender;
 use std::fmt::Debug;
 use std::{sync::Arc, time::Duration};
 use tokio::{net::TcpStream, sync::Notify};
@@ -35,7 +36,7 @@ impl From<(Vec<u8>, bool)> for ProxyMessage {
     }
 }
 
-pub type MixProxySender<S> = tokio::sync::mpsc::Sender<S>;
+pub type MixProxySender<S> = PollSender<S>;
 pub type MixProxyReader<S> = tokio::sync::mpsc::Receiver<S>;
 
 // TODO: when we finally get to implementing graceful shutdown,
diff --git a/common/statistics/src/collector.rs b/common/statistics/src/collector.rs
index ee418a2028..9a995befb5 100644
--- a/common/statistics/src/collector.rs
+++ b/common/statistics/src/collector.rs
@@ -19,7 +19,7 @@ pub trait StatisticsCollector {
         interval: Duration,
         timestamp: DateTime<Utc>,
     ) -> StatsMessage;
-    async fn send_stats_message(&self, stats_message: StatsMessage) -> Result<(), StatsError>;
+    async fn send_stats_message(&mut self, stats_message: StatsMessage) -> Result<(), StatsError>;
     async fn reset_stats(&mut self);
 }
 
diff --git a/gateway/src/node/statistics/collector.rs b/gateway/src/node/statistics/collector.rs
index d82847509f..889343e806 100644
--- a/gateway/src/node/statistics/collector.rs
+++ b/gateway/src/node/statistics/collector.rs
@@ -52,7 +52,7 @@ impl StatisticsCollector for GatewayStatisticsCollector {
         }
     }
 
-    async fn send_stats_message(&self, stats_message: StatsMessage) -> Result<(), StatsError> {
+    async fn send_stats_message(&mut self, stats_message: StatsMessage) -> Result<(), StatsError> {
         build_and_send_statistics_request(stats_message, self.statistics_service_url.to_string())
             .await
     }
diff --git a/service-providers/network-requester/Cargo.toml b/service-providers/network-requester/Cargo.toml
index e023108192..2ad5ff163a 100644
--- a/service-providers/network-requester/Cargo.toml
+++ b/service-providers/network-requester/Cargo.toml
@@ -20,7 +20,7 @@ anyhow = { workspace = true }
 addr = { workspace = true }
 async-trait = { workspace = true }
 bs58 = { workspace = true }
-clap = { workspace = true, features = ["cargo", "derive"]}
+clap = { workspace = true, features = ["cargo", "derive"] }
 dirs = "4.0"
 futures = { workspace = true }
 humantime-serde = { workspace = true }
@@ -33,19 +33,26 @@ regex = { workspace = true }
 reqwest = { workspace = true, features = ["json"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
-sqlx = { workspace = true, features = ["runtime-tokio-rustls", "chrono"]}
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "chrono"] }
 tap = { workspace = true }
 thiserror = { workspace = true }
-tokio = { workspace = true, features = [ "net", "rt-multi-thread", "macros" ] }
+tokio = { workspace = true, features = ["net", "rt-multi-thread", "macros"] }
 tokio-tungstenite = { workspace = true }
+tokio-util = { workspace = true }
 url = { workspace = true }
 time = { workspace = true }
 zeroize = { workspace = true }
 
 # internal
 nym-async-file-watcher = { path = "../../common/async-file-watcher" }
-nym-bin-common = { path = "../../common/bin-common", features = ["output_format"] }
-nym-client-core = { path = "../../common/client-core", features = ["cli", "fs-gateways-storage", "fs-surb-storage"] }
+nym-bin-common = { path = "../../common/bin-common", features = [
+    "output_format",
+] }
+nym-client-core = { path = "../../common/client-core", features = [
+    "cli",
+    "fs-gateways-storage",
+    "fs-surb-storage",
+] }
 nym-client-websocket-requests = { path = "../../clients/native/websocket-requests" }
 nym-config = { path = "../../common/config" }
 nym-credentials = { path = "../../common/credentials" }
diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs
index 01c395365c..ff05e08a49 100644
--- a/service-providers/network-requester/src/core.rs
+++ b/service-providers/network-requester/src/core.rs
@@ -10,6 +10,7 @@ use crate::{reply, socks5};
 use async_trait::async_trait;
 use futures::channel::{mpsc, oneshot};
 use futures::stream::StreamExt;
+use futures::SinkExt;
 use log::{debug, warn};
 use nym_bin_common::bin_info_owned;
 use nym_client_core::client::mix_traffic::transceiver::GatewayTransceiver;
@@ -38,6 +39,7 @@ use nym_statistics_common::collector::StatisticsSender;
 use nym_task::connections::LaneQueueLengths;
 use nym_task::manager::TaskHandle;
 use nym_task::TaskClient;
+use tokio_util::sync::PollSender;
 use std::path::Path;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
@@ -284,6 +286,8 @@ impl NRServiceProviderBuilder {
         // going to be used by `mixnet_response_listener`
         let (mix_input_sender, mix_input_receiver) = tokio::sync::mpsc::channel::<MixnetMessage>(1);
 
+        let mix_input_sender = PollSender::new(mix_input_sender);
+
         // Controller for managing all active connections.
         let (mut active_connections_controller, controller_sender) = Controller::new(
             mixnet_client.connection_command_sender(),
@@ -443,7 +447,7 @@ impl NRServiceProvider {
         return_address: reply::MixnetAddress,
         biggest_packet_size: PacketSize,
         controller_sender: ControllerSender,
-        mix_input_sender: MixProxySender<MixnetMessage>,
+        mut mix_input_sender: MixProxySender<MixnetMessage>,
         lane_queue_lengths: LaneQueueLengths,
         mut shutdown: TaskClient,
     ) {
@@ -537,7 +541,7 @@ impl NRServiceProvider {
             .unwrap_or(traffic_config.primary_packet_size);
 
         let controller_sender_clone = self.controller_sender.clone();
-        let mix_input_sender_clone = self.mix_input_sender.clone();
+        let mut mix_input_sender_clone = self.mix_input_sender.clone();
         let lane_queue_lengths_clone = self.mixnet_client.shared_lane_queue_lengths();
         let mut shutdown = self.shutdown.get_handle();
 
diff --git a/service-providers/network-requester/src/statistics/collector.rs b/service-providers/network-requester/src/statistics/collector.rs
index 95bc4cce71..2e114ccff7 100644
--- a/service-providers/network-requester/src/statistics/collector.rs
+++ b/service-providers/network-requester/src/statistics/collector.rs
@@ -5,6 +5,7 @@ use super::error::StatsError;
 use crate::core::new_legacy_request_version;
 use crate::reply::MixnetMessage;
 use async_trait::async_trait;
+use futures::SinkExt;
 use log::*;
 use nym_service_providers_common::interface::RequestVersion;
 use nym_socks5_proxy_helpers::proxy_runner::MixProxySender;
@@ -165,7 +166,7 @@ impl StatisticsCollector for ServiceStatisticsCollector {
     }
 
     async fn send_stats_message(
-        &self,
+        &mut self,
         stats_message: StatsMessage,
     ) -> Result<(), CommonStatsError> {
         let msg = build_statistics_request_bytes(stats_message)?;