diff --git a/Cargo.lock b/Cargo.lock index 9f506dc6b8..21988cff4c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5624,7 +5624,6 @@ version = "0.1.0" dependencies = [ "bincode", "defguard_wireguard_rs", - "log", "nym-credentials-interface", "nym-gateway-requests", "nym-sphinx", diff --git a/common/gateway-storage/Cargo.toml b/common/gateway-storage/Cargo.toml index bd4d8b7256..810697f258 100644 --- a/common/gateway-storage/Cargo.toml +++ b/common/gateway-storage/Cargo.toml @@ -11,7 +11,6 @@ license.workspace = true [dependencies] bincode = { workspace = true } defguard_wireguard_rs = { workspace = true } -log = { workspace = true } sqlx = { workspace = true, features = [ "runtime-tokio-rustls", "sqlite", diff --git a/common/gateway-storage/migrations/20250106120000_add_timestamps.sql b/common/gateway-storage/migrations/20250106120000_add_timestamps.sql new file mode 100644 index 0000000000..6c3b3bcf85 --- /dev/null +++ b/common/gateway-storage/migrations/20250106120000_add_timestamps.sql @@ -0,0 +1,7 @@ +/* + * Copyright 2025 - Nym Technologies SA + * SPDX-License-Identifier: GPL-3.0-only + */ + +ALTER TABLE message_store + ADD COLUMN timestamp TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP; diff --git a/common/gateway-storage/src/inboxes.rs b/common/gateway-storage/src/inboxes.rs index bdf798aa02..35efe28e1a 100644 --- a/common/gateway-storage/src/inboxes.rs +++ b/common/gateway-storage/src/inboxes.rs @@ -2,9 +2,11 @@ // SPDX-License-Identifier: GPL-3.0-only use crate::models::StoredMessage; +use time::OffsetDateTime; +use tracing::debug; #[derive(Clone)] -pub(crate) struct InboxManager { +pub struct InboxManager { connection_pool: sqlx::SqlitePool, /// Maximum number of messages that can be obtained from the database per operation. /// It is used to prevent out of memory errors in the case of client receiving a lot of data while @@ -71,44 +73,22 @@ impl InboxManager { // get 1 additional message to check whether there will be more to grab // next time let limit = self.retrieval_limit + 1; - let mut res = if let Some(start_after) = start_after { - sqlx::query_as!( - StoredMessage, - r#" - SELECT - id as "id!", - client_address_bs58 as "client_address_bs58!", - content as "content!" - FROM message_store - WHERE client_address_bs58 = ? AND id > ? - ORDER BY id ASC - LIMIT ?; - "#, - client_address_bs58, - start_after, - limit - ) - .fetch_all(&self.connection_pool) - .await? - } else { - sqlx::query_as!( - StoredMessage, - r#" - SELECT - id as "id!", - client_address_bs58 as "client_address_bs58!", - content as "content!" - FROM message_store - WHERE client_address_bs58 = ? - ORDER BY id ASC - LIMIT ?; - "#, - client_address_bs58, - limit - ) - .fetch_all(&self.connection_pool) - .await? - }; + let start_after = start_after.unwrap_or(-1); + + let mut res: Vec = sqlx::query_as( + r#" + SELECT id, client_address_bs58, content, timestamp + FROM message_store + WHERE client_address_bs58 = ? AND id > ? + ORDER BY id ASC + LIMIT ?; + "#, + ) + .bind(client_address_bs58) + .bind(start_after) + .bind(limit) + .fetch_all(&self.connection_pool) + .await?; if res.len() > self.retrieval_limit as usize { res.truncate(self.retrieval_limit as usize); @@ -146,4 +126,13 @@ impl InboxManager { .await?; Ok(()) } + + pub async fn remove_stale(&self, cutoff: OffsetDateTime) -> Result<(), sqlx::Error> { + let affected = sqlx::query!("DELETE FROM message_store WHERE timestamp < ?", cutoff) + .execute(&self.connection_pool) + .await? + .rows_affected(); + debug!("Removed {affected} stale messages"); + Ok(()) + } } diff --git a/common/gateway-storage/src/lib.rs b/common/gateway-storage/src/lib.rs index d95581f88a..5f289ec106 100644 --- a/common/gateway-storage/src/lib.rs +++ b/common/gateway-storage/src/lib.rs @@ -3,7 +3,6 @@ use bandwidth::BandwidthManager; use clients::{ClientManager, ClientType}; -use inboxes::InboxManager; use models::{ Client, PersistedBandwidth, PersistedSharedKeys, RedemptionProposal, StoredMessage, VerifiedTicket, WireguardPeer, @@ -31,6 +30,7 @@ mod tickets; mod wireguard_peers; pub use error::GatewayStorageError; +pub use inboxes::InboxManager; // note that clone here is fine as upon cloning the same underlying pool will be used #[derive(Clone)] @@ -53,7 +53,7 @@ impl GatewayStorage { &self.shared_key_manager } - pub(crate) fn inbox_manager(&self) -> &InboxManager { + pub fn inbox_manager(&self) -> &InboxManager { &self.inbox_manager } diff --git a/common/gateway-storage/src/models.rs b/common/gateway-storage/src/models.rs index f1bacdcaea..c623f72602 100644 --- a/common/gateway-storage/src/models.rs +++ b/common/gateway-storage/src/models.rs @@ -48,11 +48,13 @@ impl TryFrom for SharedGatewayKey { } } +#[derive(FromRow)] pub struct StoredMessage { pub id: i64, #[allow(dead_code)] pub client_address_bs58: String, pub content: Vec, + pub timestamp: OffsetDateTime, } #[derive(Debug, Clone, FromRow)] diff --git a/gateway/src/config.rs b/gateway/src/config.rs index 7d62118daa..66d6180f31 100644 --- a/gateway/src/config.rs +++ b/gateway/src/config.rs @@ -1,15 +1,10 @@ // Copyright 2024 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only -use nym_network_defaults::TICKETBOOK_VALIDITY_DAYS; use std::net::SocketAddr; use std::time::Duration; use url::Url; -// TODO: can we move those away? -pub const DEFAULT_CLIENT_BANDWIDTH_MAX_FLUSHING_RATE: Duration = Duration::from_millis(5); -pub const DEFAULT_CLIENT_BANDWIDTH_MAX_DELTA_FLUSHING_AMOUNT: i64 = 512 * 1024; // 512kB - #[derive(Debug)] pub struct Config { pub gateway: Gateway, @@ -96,18 +91,13 @@ pub struct Debug { /// Defines a maximum change in client bandwidth before it gets flushed to the persistent storage. pub client_bandwidth_max_delta_flushing_amount: i64, - pub zk_nym_tickets: ZkNymTicketHandlerDebug, -} + /// Specifies how often the clean-up task should check for stale data. + pub stale_messages_cleaner_run_interval: Duration, -impl Default for Debug { - fn default() -> Self { - Debug { - client_bandwidth_max_flushing_rate: DEFAULT_CLIENT_BANDWIDTH_MAX_FLUSHING_RATE, - client_bandwidth_max_delta_flushing_amount: - DEFAULT_CLIENT_BANDWIDTH_MAX_DELTA_FLUSHING_AMOUNT, - zk_nym_tickets: Default::default(), - } - } + /// Specifies maximum age of stored messages before they are removed from the storage + pub stale_messages_max_age: Duration, + + pub zk_nym_tickets: ZkNymTicketHandlerDebug, } #[derive(Debug, Clone)] @@ -131,44 +121,3 @@ pub struct ZkNymTicketHandlerDebug { /// That's required as nym-apis will purge all ticket information for tickets older than maximum validity. pub maximum_time_between_redemption: Duration, } - -impl ZkNymTicketHandlerDebug { - pub const DEFAULT_REVOCATION_BANDWIDTH_PENALTY: f32 = 10.0; - pub const DEFAULT_PENDING_POLLER: Duration = Duration::from_secs(300); - pub const DEFAULT_MINIMUM_API_QUORUM: f32 = 0.8; - pub const DEFAULT_MINIMUM_REDEMPTION_TICKETS: usize = 100; - - // use min(4/5 of max validity, validity - 1), but making sure it's no greater than 1 day - // ASSUMPTION: our validity period is AT LEAST 2 days - // - // this could have been a constant, but it's more readable as a function - pub const fn default_maximum_time_between_redemption() -> Duration { - let desired_secs = TICKETBOOK_VALIDITY_DAYS * (86400 * 4) / 5; - let desired_secs_alt = (TICKETBOOK_VALIDITY_DAYS - 1) * 86400; - - // can't use `min` in const context - let target_secs = if desired_secs < desired_secs_alt { - desired_secs - } else { - desired_secs_alt - }; - - assert!( - target_secs > 86400, - "the maximum time between redemption can't be lower than 1 day!" - ); - Duration::from_secs(target_secs as u64) - } -} - -impl Default for ZkNymTicketHandlerDebug { - fn default() -> Self { - ZkNymTicketHandlerDebug { - revocation_bandwidth_penalty: Self::DEFAULT_REVOCATION_BANDWIDTH_PENALTY, - pending_poller: Self::DEFAULT_PENDING_POLLER, - minimum_api_quorum: Self::DEFAULT_MINIMUM_API_QUORUM, - minimum_redemption_tickets: Self::DEFAULT_MINIMUM_REDEMPTION_TICKETS, - maximum_time_between_redemption: Self::default_maximum_time_between_redemption(), - } - } -} diff --git a/gateway/src/node/mixnet_handling/mod.rs b/gateway/src/node/mixnet_handling/mod.rs deleted file mode 100644 index d6f05c6d1f..0000000000 --- a/gateway/src/node/mixnet_handling/mod.rs +++ /dev/null @@ -1,7 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// SPDX-License-Identifier: GPL-3.0-only - -pub(crate) mod receiver; - -pub(crate) use receiver::listener::Listener; -pub(crate) use receiver::packet_processing::PacketProcessor; diff --git a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs deleted file mode 100644 index 62e8274af8..0000000000 --- a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs +++ /dev/null @@ -1,247 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// 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, - 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, - ) -> Result<(), Vec> { - 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, - ) -> 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, - 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}") - } - } - } -} diff --git a/gateway/src/node/mixnet_handling/receiver/listener.rs b/gateway/src/node/mixnet_handling/receiver/listener.rs deleted file mode 100644 index 17dfe00cd8..0000000000 --- a/gateway/src/node/mixnet_handling/receiver/listener.rs +++ /dev/null @@ -1,56 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// 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 }) - } -} diff --git a/gateway/src/node/mixnet_handling/receiver/mod.rs b/gateway/src/node/mixnet_handling/receiver/mod.rs deleted file mode 100644 index 2304ed90f0..0000000000 --- a/gateway/src/node/mixnet_handling/receiver/mod.rs +++ /dev/null @@ -1,6 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// SPDX-License-Identifier: GPL-3.0-only - -pub(crate) mod connection_handler; -pub(crate) mod listener; -pub(crate) mod packet_processing; diff --git a/gateway/src/node/mixnet_handling/receiver/packet_processing.rs b/gateway/src/node/mixnet_handling/receiver/packet_processing.rs deleted file mode 100644 index 4f16c2451f..0000000000 --- a/gateway/src/node/mixnet_handling/receiver/packet_processing.rs +++ /dev/null @@ -1,52 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// 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 { - match process_framed_packet(received, sphinx_key)? { - MixProcessingResult::ForwardHop(..) => Err(GatewayProcessingError::ForwardHopReceived), - MixProcessingResult::FinalHop(processed_final) => Ok(processed_final), - } -} diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index 3304f223b3..29c0d31e7b 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -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, diff --git a/gateway/src/node/stale_data_cleaner.rs b/gateway/src/node/stale_data_cleaner.rs new file mode 100644 index 0000000000..271acfe1b5 --- /dev/null +++ b/gateway/src/node/stale_data_cleaner.rs @@ -0,0 +1,60 @@ +// Copyright 2025 - Nym Technologies SA +// 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 }) + } +} diff --git a/nym-node/src/config/gateway_tasks.rs b/nym-node/src/config/gateway_tasks.rs index 677d381155..4944c8d0ed 100644 --- a/nym-node/src/config/gateway_tasks.rs +++ b/nym-node/src/config/gateway_tasks.rs @@ -47,6 +47,10 @@ pub struct Debug { /// Number of messages from offline client that can be pulled at once (i.e. with a single SQL query) from the storage. pub message_retrieval_limit: i64, + pub stale_messages: StaleMessageDebug, + + pub client_bandwidth: ClientBandwidthDebug, + pub zk_nym_tickets: ZkNymTicketHandlerDebug, } @@ -58,6 +62,8 @@ impl Default for Debug { fn default() -> Self { Debug { message_retrieval_limit: Self::DEFAULT_MESSAGE_RETRIEVAL_LIMIT, + stale_messages: Default::default(), + client_bandwidth: Default::default(), zk_nym_tickets: Default::default(), } } @@ -129,6 +135,54 @@ impl Default for ZkNymTicketHandlerDebug { } } +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +pub struct ClientBandwidthDebug { + /// Defines maximum delay between client bandwidth information being flushed to the persistent storage. + pub max_flushing_rate: Duration, + + /// Defines a maximum change in client bandwidth before it gets flushed to the persistent storage. + pub max_delta_flushing_amount: i64, +} + +impl ClientBandwidthDebug { + const DEFAULT_CLIENT_BANDWIDTH_MAX_FLUSHING_RATE: Duration = Duration::from_millis(5); + const DEFAULT_CLIENT_BANDWIDTH_MAX_DELTA_FLUSHING_AMOUNT: i64 = 512 * 1024; // 512kB +} + +impl Default for ClientBandwidthDebug { + fn default() -> Self { + ClientBandwidthDebug { + max_flushing_rate: Self::DEFAULT_CLIENT_BANDWIDTH_MAX_FLUSHING_RATE, + max_delta_flushing_amount: Self::DEFAULT_CLIENT_BANDWIDTH_MAX_DELTA_FLUSHING_AMOUNT, + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +pub struct StaleMessageDebug { + /// Specifies how often the clean-up task should check for stale data. + #[serde(with = "humantime_serde")] + pub cleaner_run_interval: Duration, + + /// Specifies maximum age of stored messages before they are removed from the storage + #[serde(with = "humantime_serde")] + pub max_age: Duration, +} + +impl StaleMessageDebug { + const DEFAULT_STALE_MESSAGES_CLEANER_RUN_INTERVAL: Duration = Duration::from_secs(60 * 60); + const DEFAULT_STALE_MESSAGES_MAX_AGE: Duration = Duration::from_secs(24 * 60 * 60); +} + +impl Default for StaleMessageDebug { + fn default() -> Self { + StaleMessageDebug { + cleaner_run_interval: Self::DEFAULT_STALE_MESSAGES_CLEANER_RUN_INTERVAL, + max_age: Self::DEFAULT_STALE_MESSAGES_MAX_AGE, + } + } +} + impl GatewayTasksConfig { pub fn new_default>(data_dir: P) -> Self { GatewayTasksConfig { diff --git a/nym-node/src/config/helpers.rs b/nym-node/src/config/helpers.rs index 3ba91b5432..c6259ac375 100644 --- a/nym-node/src/config/helpers.rs +++ b/nym-node/src/config/helpers.rs @@ -24,10 +24,22 @@ fn ephemeral_gateway_config(config: &Config) -> nym_gateway::config::Config { enabled: config.service_providers.network_requester.debug.enabled, }, nym_gateway::config::Debug { - client_bandwidth_max_flushing_rate: - nym_gateway::config::DEFAULT_CLIENT_BANDWIDTH_MAX_FLUSHING_RATE, - client_bandwidth_max_delta_flushing_amount: - nym_gateway::config::DEFAULT_CLIENT_BANDWIDTH_MAX_DELTA_FLUSHING_AMOUNT, + client_bandwidth_max_flushing_rate: config + .gateway_tasks + .debug + .client_bandwidth + .max_flushing_rate, + client_bandwidth_max_delta_flushing_amount: config + .gateway_tasks + .debug + .client_bandwidth + .max_delta_flushing_amount, + stale_messages_cleaner_run_interval: config + .gateway_tasks + .debug + .stale_messages + .cleaner_run_interval, + stale_messages_max_age: config.gateway_tasks.debug.stale_messages.max_age, zk_nym_tickets: nym_gateway::config::ZkNymTicketHandlerDebug { revocation_bandwidth_penalty: config .gateway_tasks diff --git a/nym-node/src/config/old_configs/old_config_v6.rs b/nym-node/src/config/old_configs/old_config_v6.rs index b920bf035f..7e9eba426f 100644 --- a/nym-node/src/config/old_configs/old_config_v6.rs +++ b/nym-node/src/config/old_configs/old_config_v6.rs @@ -1095,6 +1095,7 @@ pub async fn try_upgrade_config_v6>( .zk_nym_tickets .maximum_time_between_redemption, }, + ..Default::default() }, }, service_providers: ServiceProvidersConfig { diff --git a/nym-node/src/node/mod.rs b/nym-node/src/node/mod.rs index f91fce211b..23fdd2adab 100644 --- a/nym-node/src/node/mod.rs +++ b/nym-node/src/node/mod.rs @@ -665,6 +665,10 @@ impl NymNode { info!("node not running with wireguard: authenticator service provider and wireguard will remain unavailable"); } + // start task for removing stale and un-retrieved client messages + let stale_messages_cleaner = gateway_tasks_builder.build_stale_messages_cleaner(); + stale_messages_cleaner.start(); + Ok(()) } diff --git a/nym-wallet/Cargo.lock b/nym-wallet/Cargo.lock index 39581d9cc1..a477ed953f 100644 --- a/nym-wallet/Cargo.lock +++ b/nym-wallet/Cargo.lock @@ -3194,6 +3194,7 @@ dependencies = [ "strum 0.26.3", "thiserror", "time", + "utoipa", ] [[package]] @@ -3290,6 +3291,7 @@ dependencies = [ "thiserror", "time", "ts-rs", + "utoipa", ] [[package]] @@ -3395,6 +3397,7 @@ dependencies = [ "serde", "sha2 0.10.8", "time", + "utoipa", ] [[package]] @@ -3422,6 +3425,7 @@ dependencies = [ "thiserror", "ts-rs", "url", + "utoipa", "x25519-dalek", ] @@ -6208,9 +6212,9 @@ checksum = "711b9620af191e0cdc7468a8d14e709c3dcdb115b36f838e601583af800a370a" [[package]] name = "utoipa" -version = "4.2.3" +version = "5.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c5afb1a60e207dca502682537fefcfd9921e71d0b83e9576060f09abc6efab23" +checksum = "435c6f69ef38c9017b4b4eea965dfb91e71e53d869e896db40d1cf2441dd75c0" dependencies = [ "indexmap 2.5.0", "serde", @@ -6220,11 +6224,10 @@ dependencies = [ [[package]] name = "utoipa-gen" -version = "4.3.0" +version = "5.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7bf0e16c02bc4bf5322ab65f10ab1149bdbcaa782cba66dc7057370a3f8190be" +checksum = "a77d306bc75294fd52f3e99b13ece67c02c1a2789190a6f31d32f736624326f7" dependencies = [ - "proc-macro-error", "proc-macro2", "quote", "syn 2.0.85",