GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature: periodically remove stale gateway messages (#5312)

Browse Source 
* add timestamp to stored client messages

* removed dead code

* starting node task to remove old messages

* added log for number of removed messages

* debug log on task finishing
This commit is contained in:
Jędrzej Stuczyński
2025-01-09 10:03:19 +01:00
committed by GitHub
parent 3d2914b3e5
commit 226c040a13
19 changed files with 199 additions and 477 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/src/node/mixnet_handling/mod.rs
-7
View File
@@ -1,7 +0,0 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
pub(crate) mod receiver;
pub(crate) use receiver::listener::Listener;
pub(crate) use receiver::packet_processing::PacketProcessor;
gateway/src/node/mixnet_handling/receiver/connection_handler.rs
-247
View File
@@ -1,247 +0,0 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::node::client_handling::active_clients::ActiveClientsStore;
use crate::node::client_handling::websocket::message_receiver::MixMessageSender;
use crate::node::mixnet_handling::receiver::packet_processing::PacketProcessor;
use futures::channel::mpsc::SendError;
use futures::StreamExt;
use nym_gateway_storage::{error::GatewayStorageError, GatewayStorage};
use nym_mixnet_client::forwarder::MixForwardingSender;
use nym_sphinx::forwarding::packet::MixPacket;
use nym_sphinx::framing::codec::NymCodec;
use nym_sphinx::framing::packet::FramedNymPacket;
use nym_sphinx::framing::processing::ProcessedFinalHop;
use nym_sphinx::DestinationAddressBytes;
use nym_task::TaskClient;
use std::collections::HashMap;
use std::net::SocketAddr;
use thiserror::Error;
use tokio::net::TcpStream;
use tokio_util::codec::Framed;
use tracing::*;
use super::packet_processing::process_packet;
// defines errors that warrant a panic if not thrown in the context of a shutdown
#[derive(Debug, Error)]
enum CriticalPacketProcessingError {
#[error("failed to forward an ack")]
AckForwardingFailure { source: SendError },
}
pub(crate) struct ConnectionHandler {
packet_processor: PacketProcessor,
// TODO: investigate performance trade-offs for whether this cache even makes sense
// at this point.
// keep the following in mind: each action on ActiveClientsStore requires going through RwLock
// and each `get` internally copies the channel, however, is it really that expensive?
clients_store_cache: HashMap<DestinationAddressBytes, MixMessageSender>,
active_clients_store: ActiveClientsStore,
storage: GatewayStorage,
ack_sender: MixForwardingSender,
}
impl Clone for ConnectionHandler {
fn clone(&self) -> Self {
// remove stale entries from the cache while cloning
let mut clients_store_cache = HashMap::with_capacity(self.clients_store_cache.capacity());
for (k, v) in self.clients_store_cache.iter() {
if !v.is_closed() {
clients_store_cache.insert(*k, v.clone());
}
}
ConnectionHandler {
packet_processor: self.packet_processor.clone(),
clients_store_cache,
active_clients_store: self.active_clients_store.clone(),
storage: self.storage.clone(),
ack_sender: self.ack_sender.clone(),
}
}
}
impl ConnectionHandler {
pub(crate) fn new(
packet_processor: PacketProcessor,
storage: GatewayStorage,
ack_sender: MixForwardingSender,
active_clients_store: ActiveClientsStore,
) -> Self {
ConnectionHandler {
packet_processor,
clients_store_cache: HashMap::new(),
storage,
active_clients_store,
ack_sender,
}
}
fn update_clients_store_cache_entry(&mut self, client_address: DestinationAddressBytes) {
if let Some(client_senders) = self.active_clients_store.get_sender(client_address) {
self.clients_store_cache
.insert(client_address, client_senders);
}
}
fn check_cache(&mut self, client_address: DestinationAddressBytes) {
match self.clients_store_cache.get(&client_address) {
None => self.update_clients_store_cache_entry(client_address),
Some(entry) => {
if entry.is_closed() {
self.update_clients_store_cache_entry(client_address)
}
}
}
}
fn try_push_message_to_client(
&mut self,
client_address: DestinationAddressBytes,
message: Vec<u8>,
) -> Result<(), Vec<u8>> {
self.check_cache(client_address);
match self.clients_store_cache.get(&client_address) {
None => Err(message),
Some(sender_channel) => {
if let Err(unsent) = sender_channel.unbounded_send(vec![message]) {
// the unwrap here is fine as the original message got returned;
// plus we're only ever sending 1 message at the time (for now)
#[allow(clippy::unwrap_used)]
return Err(unsent.into_inner().pop().unwrap());
} else {
Ok(())
}
}
}
}
pub(crate) async fn store_processed_packet_payload(
&self,
client_address: DestinationAddressBytes,
message: Vec<u8>,
) -> Result<(), GatewayStorageError> {
debug!("Storing received message for {client_address} on the disk...",);
self.storage.store_message(client_address, message).await
}
fn forward_ack(
&self,
forward_ack: Option<MixPacket>,
client_address: DestinationAddressBytes,
) -> Result<(), CriticalPacketProcessingError> {
if let Some(forward_ack) = forward_ack {
let next_hop = forward_ack.next_hop();
trace!("Sending ack from packet for {client_address} to {next_hop}",);
self.ack_sender
.forward_packet(forward_ack)
.map_err(|source| CriticalPacketProcessingError::AckForwardingFailure { source })?;
}
Ok(())
}
async fn handle_processed_packet(
&mut self,
processed_final_hop: ProcessedFinalHop,
) -> Result<(), CriticalPacketProcessingError> {
let client_address = processed_final_hop.destination;
let message = processed_final_hop.message;
let forward_ack = processed_final_hop.forward_ack;
// we failed to push message directly to the client - it's probably offline.
// we should store it on the disk instead.
match self.try_push_message_to_client(client_address, message) {
Err(unsent_plaintext) => match self
.store_processed_packet_payload(client_address, unsent_plaintext)
.await
{
Err(err) => error!("Failed to store client data - {err}"),
Ok(_) => trace!("Stored packet for {client_address}"),
},
Ok(_) => trace!("Pushed received packet to {client_address}"),
}
// if we managed to either push message directly to the [online] client or store it at
// its inbox, it means that it must exist at this gateway, hence we can send the
// received ack back into the network
self.forward_ack(forward_ack, client_address)
}
async fn handle_received_packet(
&mut self,
framed_sphinx_packet: FramedNymPacket,
) -> Result<(), CriticalPacketProcessingError> {
//
// TODO: here be replay attack detection - it will require similar key cache to the one in
// packet processor for vpn packets,
// question: can it also be per connection vs global?
//
let processed_final_hop =
match process_packet(framed_sphinx_packet, self.packet_processor.sphinx_key()) {
Err(err) => {
debug!("We failed to process received sphinx packet - {err}");
return Ok(());
}
Ok(processed_final_hop) => processed_final_hop,
};
self.handle_processed_packet(processed_final_hop).await
}
pub(crate) async fn handle_connection(
mut self,
conn: TcpStream,
remote: SocketAddr,
mut shutdown: TaskClient,
) {
debug!("Starting connection handler for {:?}", remote);
shutdown.disarm();
let mut framed_conn = Framed::new(conn, NymCodec);
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown.recv() => {
trace!("ConnectionHandler: received shutdown");
}
framed_sphinx_packet = framed_conn.next() => {
match framed_sphinx_packet {
Some(Ok(framed_sphinx_packet)) => {
// TODO: benchmark spawning tokio task with full processing vs just processing it
// synchronously under higher load in single and multi-threaded situation.
// in theory we could process multiple sphinx packet from the same connection in parallel,
// but we already handle multiple concurrent connections so if anything, making
// that change would only slow things down
if let Err(critical_err) = self.handle_received_packet(framed_sphinx_packet).await {
if !shutdown.is_shutdown() {
panic!("experienced critical failure when processing received packet: {critical_err}")
}
}
}
Some(Err(err)) => {
error!(
"The socket connection got corrupted with error: {err}. Closing the socket",
);
return;
}
None => break, // stream got closed by remote
}
}
}
}
match framed_conn.into_inner().peer_addr() {
Ok(peer_addr) => {
debug!("closing connection from {peer_addr}")
}
Err(err) => {
warn!("closing connection from an unknown peer: {err}")
}
}
}
}
gateway/src/node/mixnet_handling/receiver/listener.rs
-56
View File
@@ -1,56 +0,0 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::node::mixnet_handling::receiver::connection_handler::ConnectionHandler;
use nym_task::TaskClient;
use std::net::SocketAddr;
use std::process;
use tokio::task::JoinHandle;
use tracing::*;
pub(crate) struct Listener {
address: SocketAddr,
shutdown: TaskClient,
}
// TODO: this file is nearly identical to the one in mixnode
impl Listener {
pub(crate) fn new(address: SocketAddr, shutdown: TaskClient) -> Self {
Listener { address, shutdown }
}
pub(crate) async fn run(&mut self, connection_handler: ConnectionHandler) {
info!("Starting mixnet listener at {}", self.address);
let tcp_listener = match tokio::net::TcpListener::bind(self.address).await {
Ok(listener) => listener,
Err(err) => {
error!("Failed to bind to {} - {err}. Are you sure nothing else is running on the specified port and your user has sufficient permission to bind to the requested address?", self.address);
process::exit(1);
}
};
while !self.shutdown.is_shutdown() {
tokio::select! {
biased;
_ = self.shutdown.recv() => {
trace!("mixnet_handling::Listener: Received shutdown");
}
connection = tcp_listener.accept() => {
match connection {
Ok((socket, remote_addr)) => {
let handler = connection_handler.clone();
tokio::spawn(handler.handle_connection(socket, remote_addr, self.shutdown.clone().named(format!("MixnetConnectionHandler_{remote_addr}"))));
}
Err(err) => warn!("failed to get client: {err}"),
}
}
}
}
}
pub(crate) fn start(mut self, connection_handler: ConnectionHandler) -> JoinHandle<()> {
info!("Running mix listener on {:?}", self.address.to_string());
tokio::spawn(async move { self.run(connection_handler).await })
}
}
gateway/src/node/mixnet_handling/receiver/mod.rs
-6
View File
@@ -1,6 +0,0 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
pub(crate) mod connection_handler;
pub(crate) mod listener;
pub(crate) mod packet_processing;
gateway/src/node/mixnet_handling/receiver/packet_processing.rs
-52
View File
@@ -1,52 +0,0 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use nym_crypto::asymmetric::encryption;
use nym_mixnode_common::packet_processor::error::MixProcessingError;
use nym_mixnode_common::packet_processor::processor::SphinxPacketProcessor;
use nym_sphinx::framing::packet::FramedNymPacket;
use nym_sphinx::framing::processing::{
process_framed_packet, MixProcessingResult, PacketProcessingError, ProcessedFinalHop,
};
use nym_sphinx::PrivateKey;
use thiserror::Error;
#[derive(Error, Debug)]
pub enum GatewayProcessingError {
#[error("failed to process received mix packet - {0}")]
PacketProcessing(#[from] MixProcessingError),
#[error("received a forward hop mix packet")]
ForwardHopReceived,
#[error("failed to process received sphinx packet: {0}")]
NymPacketProcessing(#[from] PacketProcessingError),
}
// PacketProcessor contains all data required to correctly unwrap and store sphinx packets
#[derive(Clone)]
pub struct PacketProcessor {
inner_processor: SphinxPacketProcessor,
}
impl PacketProcessor {
pub fn sphinx_key(&self) -> &PrivateKey {
self.inner_processor.sphinx_key()
}
pub(crate) fn new(encryption_key: &encryption::PrivateKey) -> Self {
PacketProcessor {
inner_processor: SphinxPacketProcessor::new(encryption_key.into()),
}
}
}
pub(crate) fn process_packet(
received: FramedNymPacket,
sphinx_key: &nym_sphinx::PrivateKey,
) -> Result<ProcessedFinalHop, GatewayProcessingError> {
match process_framed_packet(received, sphinx_key)? {
MixProcessingResult::ForwardHop(..) => Err(GatewayProcessingError::ForwardHopReceived),
MixProcessingResult::FinalHop(processed_final) => Ok(processed_final),
}
}
gateway/src/node/mod.rs
+11
View File
@@ -33,7 +33,9 @@ use zeroize::Zeroizing;
pub(crate) mod client_handling;
mod internal_service_providers;
mod stale_data_cleaner;
use crate::node::stale_data_cleaner::StaleMessagesCleaner;
pub use client_handling::active_clients::ActiveClientsStore;
pub use nym_gateway_stats_storage::PersistentStatsStorage;
pub use nym_gateway_storage::{error::GatewayStorageError, GatewayStorage};
@@ -446,6 +448,15 @@ impl GatewayTasksBuilder {
))
}
pub fn build_stale_messages_cleaner(&self) -> StaleMessagesCleaner {
StaleMessagesCleaner::new(
&self.storage,
self.shutdown.fork("stale-messages-cleaner"),
self.config.debug.stale_messages_max_age,
self.config.debug.stale_messages_cleaner_run_interval,
)
}
#[cfg(not(target_os = "linux"))]
pub async fn try_start_wireguard(
&mut self,
gateway/src/node/stale_data_cleaner.rs
+60
View File
@@ -0,0 +1,60 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use nym_gateway_storage::{GatewayStorage, InboxManager};
use nym_task::TaskClient;
use std::error::Error;
use std::time::Duration;
use time::OffsetDateTime;
use tokio::task::JoinHandle;
use tracing::{debug, trace, warn};
pub struct StaleMessagesCleaner {
inbox_manager: InboxManager,
task_client: TaskClient,
max_message_age: Duration,
run_interval: Duration,
}
impl StaleMessagesCleaner {
pub(crate) fn new(
storage: &GatewayStorage,
task_client: TaskClient,
max_message_age: Duration,
run_interval: Duration,
) -> Self {
StaleMessagesCleaner {
inbox_manager: storage.inbox_manager().clone(),
task_client,
max_message_age,
run_interval,
}
}
async fn clean_up_stale_messages(&mut self) -> Result<(), impl Error> {
let cutoff = OffsetDateTime::now_utc() - self.max_message_age;
self.inbox_manager.remove_stale(cutoff).await
}
async fn run(&mut self) {
let mut interval = tokio::time::interval(self.run_interval);
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = self.task_client.recv() => {
trace!("StaleMessagesCleaner: received shutdown");
}
_ = interval.tick() => {
if let Err(err) = self.clean_up_stale_messages().await {
warn!("failed to clean up stale messages: {err}");
}
}
}
}
debug!("StaleMessagesCleaner: Exiting");
}
pub fn start(mut self) -> JoinHandle<()> {
tokio::spawn(async move { self.run().await })
}
}