From 2c4611044c82cb6203fa1f8d2d08db1bf5436bc4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 7 Dec 2022 14:19:41 +0000 Subject: [PATCH] Feature/multi surbs invalidation (#1858) * Cleaned up RealMessagesController constructor * introduced config field for maximum_reply_surb_age * Handling edge-case reply-surb failures * invalidating old reply surbs * Removing old reply keys from cache * Invalidating old reply keys * missing config changes --- .../20221130120000_create_initial_tables.sql | 5 +- .../real_messages_control/message_handler.rs | 103 +++++- .../src/client/real_messages_control/mod.rs | 118 ++++--- .../real_traffic_stream.rs | 30 +- .../client-core/src/client/received_buffer.rs | 2 +- .../src/client/replies/reply_controller.rs | 299 +++++++++++++----- .../reply_storage/backend/fs_backend/mod.rs | 5 +- .../backend/fs_backend/models.rs | 57 +--- .../client/replies/reply_storage/combined.rs | 3 - .../replies/reply_storage/key_storage.rs | 50 ++- .../replies/reply_storage/surb_storage.rs | 34 +- clients/client-core/src/config/mod.rs | 26 ++ clients/webassembly/src/client/config.rs | 12 + 13 files changed, 515 insertions(+), 229 deletions(-) diff --git a/clients/client-core/fs_surbs_migrations/20221130120000_create_initial_tables.sql b/clients/client-core/fs_surbs_migrations/20221130120000_create_initial_tables.sql index 80c157a035..0143475a37 100644 --- a/clients/client-core/fs_surbs_migrations/20221130120000_create_initial_tables.sql +++ b/clients/client-core/fs_surbs_migrations/20221130120000_create_initial_tables.sql @@ -19,8 +19,9 @@ CREATE TABLE sender_tag CREATE TABLE reply_key ( - key_digest BLOB NOT NULL UNIQUE, - reply_key BLOB NOT NULL UNIQUE + key_digest BLOB NOT NULL UNIQUE, + reply_key BLOB NOT NULL UNIQUE, + sent_at_timestamp INTEGER NOT NULL ); CREATE TABLE reply_surb_sender diff --git a/clients/client-core/src/client/real_messages_control/message_handler.rs b/clients/client-core/src/client/real_messages_control/message_handler.rs index 5b09755ecd..73dc13df2e 100644 --- a/clients/client-core/src/client/real_messages_control/message_handler.rs +++ b/clients/client-core/src/client/real_messages_control/message_handler.rs @@ -16,9 +16,11 @@ use nymsphinx::anonymous_replies::requests::{AnonymousSenderTag, RepliableMessag use nymsphinx::anonymous_replies::{ReplySurb, SurbEncryptionKey}; use nymsphinx::chunking::fragment::Fragment; use nymsphinx::message::NymMessage; +use nymsphinx::params::{PacketSize, DEFAULT_NUM_MIX_HOPS}; use nymsphinx::preparer::{MessagePreparer, PreparedFragment}; use rand::{CryptoRng, Rng}; use std::sync::Arc; +use std::time::Duration; use thiserror::Error; use topology::{NymTopology, NymTopologyError}; @@ -88,10 +90,63 @@ impl SurbWrappedPreparationError { } #[derive(Clone)] -pub(crate) struct MessageHandler { - rng: R, +pub(crate) struct Config { + /// Key used to decrypt contents of received SURBAcks ack_key: Arc, - self_address: Recipient, + + /// Address of this client which also represent an address to which all acknowledgements + /// and surb-based are going to be sent. + sender_address: Recipient, + + /// Average delay a data packet is going to get delay at a single mixnode. + average_packet_delay: Duration, + + /// Average delay an acknowledgement packet is going to get delay at a single mixnode. + average_ack_delay: Duration, + + /// Number of mix hops each packet ('real' message, ack, reply) is expected to take. + /// Note that it does not include gateway hops. + num_mix_hops: u8, + + /// Predefined packet size used for the encapsulated messages. + packet_size: PacketSize, +} + +impl Config { + pub fn new( + ack_key: Arc, + sender_address: Recipient, + average_packet_delay: Duration, + average_ack_delay: Duration, + ) -> Self { + Config { + ack_key, + sender_address, + average_packet_delay, + average_ack_delay, + num_mix_hops: DEFAULT_NUM_MIX_HOPS, + packet_size: PacketSize::default(), + } + } + + /// Allows setting non-default number of expected mix hops in the network. + #[allow(dead_code)] + pub fn with_mix_hops(mut self, hops: u8) -> Self { + self.num_mix_hops = hops; + self + } + + /// Allows setting non-default size of the sphinx packets sent out. + pub fn with_custom_packet_size(mut self, packet_size: PacketSize) -> Self { + self.packet_size = packet_size; + self + } +} + +#[derive(Clone)] +pub(crate) struct MessageHandler { + config: Config, + rng: R, message_preparer: MessagePreparer, action_sender: AckActionSender, real_message_sender: BatchRealMessageSender, @@ -104,22 +159,30 @@ impl MessageHandler where R: CryptoRng + Rng, { - #[allow(clippy::too_many_arguments)] pub(crate) fn new( + config: Config, rng: R, - ack_key: Arc, - self_address: Recipient, - message_preparer: MessagePreparer, action_sender: AckActionSender, real_message_sender: BatchRealMessageSender, topology_access: TopologyAccessor, reply_key_storage: SentReplyKeys, tag_storage: UsedSenderTags, - ) -> Self { - MessageHandler { + ) -> Self + where + R: Copy, + { + let message_preparer = MessagePreparer::new( + rng, + config.sender_address, + config.average_packet_delay, + config.average_ack_delay, + ) + .with_custom_real_message_packet_size(config.packet_size) + .with_mix_hops(config.num_mix_hops); + + MessageHandler { + config, rng, - ack_key, - self_address, message_preparer, action_sender, real_message_sender, @@ -145,7 +208,7 @@ where &self, permit: &'a TopologyReadPermit<'a>, ) -> Result<&'a NymTopology, PreparationError> { - match permit.try_get_valid_topology_ref(&self.self_address, None) { + match permit.try_get_valid_topology_ref(&self.config.sender_address, None) { Ok(topology_ref) => Ok(topology_ref), Err(err) => { warn!("Could not process the packet - the network topology is invalid - {err}"); @@ -224,7 +287,8 @@ where ) -> Result<(), SurbWrappedPreparationError> { debug!("requesting {amount} reply SURBs from {from:?}"); - let surbs_request = ReplyMessage::new_surb_request_message(self.self_address, amount); + let surbs_request = + ReplyMessage::new_surb_request_message(self.config.sender_address, amount); self.try_send_single_surb_message(from, surbs_request, reply_surb, true) .await } @@ -267,7 +331,12 @@ where let chunk_clone = fragment.clone(); let prepared_fragment = self .message_preparer - .prepare_reply_chunk_for_sending(chunk_clone, topology, &self.ack_key, reply_surb) + .prepare_reply_chunk_for_sending( + chunk_clone, + topology, + &self.config.ack_key, + reply_surb, + ) .unwrap(); let real_message = @@ -319,7 +388,7 @@ where let prepared_fragment = self.message_preparer.prepare_chunk_for_sending( chunk_clone, topology, - &self.ack_key, + &self.config.ack_key, &recipient, )?; @@ -399,7 +468,7 @@ where let prepared_fragment = self .message_preparer - .prepare_chunk_for_sending(chunk, topology, &self.ack_key, &recipient) + .prepare_chunk_for_sending(chunk, topology, &self.config.ack_key, &recipient) .unwrap(); Ok(prepared_fragment) @@ -418,7 +487,7 @@ where let prepared_fragment = self .message_preparer - .prepare_reply_chunk_for_sending(chunk, topology, &self.ack_key, reply_surb) + .prepare_reply_chunk_for_sending(chunk, topology, &self.config.ack_key, reply_surb) .unwrap(); Ok(prepared_fragment) diff --git a/clients/client-core/src/client/real_messages_control/mod.rs b/clients/client-core/src/client/real_messages_control/mod.rs index a189016f53..736c0c54ae 100644 --- a/clients/client-core/src/client/real_messages_control/mod.rs +++ b/clients/client-core/src/client/real_messages_control/mod.rs @@ -28,11 +28,11 @@ use log::*; use nymsphinx::acknowledgements::AckKey; use nymsphinx::addressing::clients::Recipient; use nymsphinx::params::PacketSize; -use nymsphinx::preparer::MessagePreparer; use rand::{rngs::OsRng, CryptoRng, Rng}; use std::sync::Arc; use std::time::Duration; +use crate::client::replies::reply_controller; use crate::config; pub(crate) use acknowledgement_control::{AckActionSender, Action}; @@ -85,6 +85,64 @@ pub struct Config { /// Defines maximum amount of time the client is going to wait for reply surbs before explicitly asking /// for more even though in theory they wouldn't need to. maximum_reply_surb_waiting_period: Duration, + + /// Defines maximum amount of time given reply surb is going to be valid for. + /// This is going to be superseded by key rotation once implemented. + maximum_reply_surb_age: Duration, + + /// Defines maximum amount of time given reply key is going to be valid for. + /// This is going to be superseded by key rotation once implemented. + maximum_reply_key_age: Duration, +} + +impl<'a> From<&'a Config> for acknowledgement_control::Config { + fn from(cfg: &'a Config) -> Self { + acknowledgement_control::Config::new( + cfg.ack_wait_addition, + cfg.ack_wait_multiplier, + cfg.retransmission_reply_surb_request_size, + ) + .with_custom_packet_size(cfg.packet_size) + } +} + +impl<'a> From<&'a Config> for real_traffic_stream::Config { + fn from(cfg: &'a Config) -> Self { + real_traffic_stream::Config::new( + Arc::clone(&cfg.ack_key), + cfg.self_recipient, + cfg.average_ack_delay_duration, + cfg.average_packet_delay_duration, + cfg.average_message_sending_delay, + cfg.disable_main_poisson_packet_distribution, + ) + .with_custom_cover_packet_size(cfg.packet_size) + } +} + +impl<'a> From<&'a Config> for reply_controller::Config { + fn from(cfg: &'a Config) -> Self { + reply_controller::Config::new( + cfg.minimum_reply_surb_request_size, + cfg.maximum_reply_surb_request_size, + cfg.maximum_allowed_reply_surb_request_size, + cfg.maximum_reply_surb_waiting_period, + cfg.maximum_reply_surb_age, + cfg.maximum_reply_key_age, + ) + } +} + +impl<'a> From<&'a Config> for message_handler::Config { + fn from(cfg: &'a Config) -> Self { + message_handler::Config::new( + Arc::clone(&cfg.ack_key), + cfg.self_recipient, + cfg.average_packet_delay_duration, + cfg.average_ack_delay_duration, + ) + .with_custom_packet_size(cfg.packet_size) + } } impl Config { @@ -114,6 +172,8 @@ impl Config { .retransmission_reply_surb_request_size, maximum_reply_surb_waiting_period: base_client_debug_config .maximum_reply_surb_waiting_period, + maximum_reply_surb_age: base_client_debug_config.maximum_reply_surb_age, + maximum_reply_key_age: base_client_debug_config.maximum_reply_key_age, } } @@ -150,10 +210,10 @@ impl RealMessagesController { ) -> Self { let rng = OsRng; + // create channels for inter-task communication let (real_message_sender, real_message_receiver) = tokio::sync::mpsc::channel(1); let (sent_notifier_tx, sent_notifier_rx) = mpsc::unbounded(); let (ack_action_tx, ack_action_rx) = mpsc::unbounded(); - let ack_controller_connectors = AcknowledgementControllerConnectors::new( input_receiver, sent_notifier_rx, @@ -162,26 +222,16 @@ impl RealMessagesController { ack_action_rx, ); - let ack_control_config = acknowledgement_control::Config::new( - config.ack_wait_addition, - config.ack_wait_multiplier, - config.retransmission_reply_surb_request_size, - ) - .with_custom_packet_size(config.packet_size); + // create all configs for the components + let ack_control_config = (&config).into(); + let out_queue_config = (&config).into(); + let reply_controller_config = (&config).into(); + let message_handler_config = (&config).into(); - // TODO: construct MessagePreparer itself inside the MessageHandler - let message_preparer = MessagePreparer::new( - rng, - config.self_recipient, - config.average_packet_delay_duration, - config.average_ack_delay_duration, - ) - .with_custom_real_message_packet_size(config.packet_size); + // create the actual components let message_handler = MessageHandler::new( + message_handler_config, rng, - Arc::clone(&config.ack_key), - config.self_recipient, - message_preparer, ack_action_tx, real_message_sender, topology_access.clone(), @@ -189,42 +239,28 @@ impl RealMessagesController { reply_storage.tags_storage(), ); - let reply_control = ReplyController::new( - message_handler.clone(), - reply_storage.surbs_storage(), - reply_storage.tags_storage(), - reply_controller_receiver, - config.minimum_reply_surb_request_size, - config.maximum_reply_surb_request_size, - config.maximum_allowed_reply_surb_request_size, - config.maximum_reply_surb_waiting_period, - ); - let ack_control = AcknowledgementController::new( ack_control_config, Arc::clone(&config.ack_key), ack_controller_connectors, - message_handler, + message_handler.clone(), reply_controller_sender, reply_storage.surbs_storage(), ); - let out_queue_config = real_traffic_stream::Config::new( - config.average_ack_delay_duration, - config.average_packet_delay_duration, - config.average_message_sending_delay, - config.disable_main_poisson_packet_distribution, - ) - .with_custom_cover_packet_size(config.packet_size); + let reply_control = ReplyController::new( + reply_controller_config, + message_handler, + reply_storage, + reply_controller_receiver, + ); let out_queue_control = OutQueueControl::new( out_queue_config, - config.ack_key, + rng, sent_notifier_tx, mix_sender, real_message_receiver, - rng, - config.self_recipient, topology_access, lane_queue_lengths, client_connection_rx, diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs index a1eb8d2d8c..a4937b4493 100644 --- a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -48,6 +48,12 @@ fn get_time_now() -> wasm_timer::Instant { /// Configurable parameters of the `OutQueueControl` pub(crate) struct Config { + /// Key used to encrypt and decrypt content of an ACK packet. + ack_key: Arc, + + /// Represents full address of this client. + our_full_destination: Recipient, + /// Average delay an acknowledgement packet is going to get delay at a single mixnode. average_ack_delay: Duration, @@ -67,12 +73,16 @@ pub(crate) struct Config { impl Config { pub(crate) fn new( + ack_key: Arc, + our_full_destination: Recipient, average_ack_delay: Duration, average_packet_delay: Duration, average_message_sending_delay: Duration, disable_poisson_packet_distribution: bool, ) -> Self { Config { + ack_key, + our_full_destination, average_ack_delay, average_packet_delay, average_message_sending_delay, @@ -94,9 +104,6 @@ where /// Configurable parameters of the `ActionController` config: Config, - /// Key used to encrypt and decrypt content of an ACK packet. - ack_key: Arc, - /// Channel used for notifying of a real packet being sent out. Used to start up retransmission timer. sent_notifier: SentPacketNotificationSender, @@ -120,9 +127,6 @@ where /// before being sent out into the network. real_receiver: BatchRealMessageReceiver, - /// Represents full address of this client. - our_full_destination: Recipient, - /// Instance of a cryptographically secure random number generator. rng: R, @@ -186,25 +190,21 @@ where #[allow(clippy::too_many_arguments)] pub(crate) fn new( config: Config, - ack_key: Arc, + rng: R, sent_notifier: SentPacketNotificationSender, mix_tx: BatchMixMessageSender, real_receiver: BatchRealMessageReceiver, - rng: R, - our_full_destination: Recipient, topology_access: TopologyAccessor, lane_queue_lengths: LaneQueueLengths, client_connection_rx: ConnectionCommandReceiver, ) -> Self { OutQueueControl { config, - ack_key, sent_notifier, next_delay: None, sending_delay_controller: Default::default(), mix_tx, real_receiver, - our_full_destination, rng, topology_access, transmission_buffer: Default::default(), @@ -232,8 +232,8 @@ where let topology_permit = self.topology_access.get_read_permit().await; // the ack is sent back to ourselves (and then ignored) let topology_ref = match topology_permit.try_get_valid_topology_ref( - &self.our_full_destination, - Some(&self.our_full_destination), + &self.config.our_full_destination, + Some(&self.config.our_full_destination), ) { Ok(topology) => topology, Err(err) => { @@ -246,8 +246,8 @@ where generate_loop_cover_packet( &mut self.rng, topology_ref, - &self.ack_key, - &self.our_full_destination, + &self.config.ack_key, + &self.config.our_full_destination, self.config.average_ack_delay, self.config.average_packet_delay, self.config.cover_packet_size, diff --git a/clients/client-core/src/client/received_buffer.rs b/clients/client-core/src/client/received_buffer.rs index 3a3c760888..ebcd40170b 100644 --- a/clients/client-core/src/client/received_buffer.rs +++ b/clients/client-core/src/client/received_buffer.rs @@ -328,7 +328,7 @@ impl ReceivedMessagesBuffer { .try_pop(possible_key_digest) .map(|reply_encryption_key| { ( - reply_encryption_key, + *reply_encryption_key, &mut raw_message[reply_surb_digest_size..], ) }) diff --git a/clients/client-core/src/client/replies/reply_controller.rs b/clients/client-core/src/client/replies/reply_controller.rs index cd2e3c1d96..0ee5799c63 100644 --- a/clients/client-core/src/client/replies/reply_controller.rs +++ b/clients/client-core/src/client/replies/reply_controller.rs @@ -2,11 +2,11 @@ // SPDX-License-Identifier: Apache-2.0 use crate::client::real_messages_control::message_handler::{MessageHandler, PreparationError}; -use crate::client::replies::reply_storage::{ReceivedReplySurbsMap, UsedSenderTags}; +use crate::client::replies::reply_storage::CombinedReplyStorage; use client_connections::TransmissionLane; use futures::channel::mpsc; use futures::StreamExt; -use log::{debug, trace, warn}; +use log::{debug, error, info, trace, warn}; use nymsphinx::addressing::clients::Recipient; use nymsphinx::anonymous_replies::requests::AnonymousSenderTag; use nymsphinx::anonymous_replies::ReplySurb; @@ -15,12 +15,13 @@ use rand::{CryptoRng, Rng}; use std::cmp::{max, min}; use std::collections::{HashMap, VecDeque}; use std::time::Duration; +use time::OffsetDateTime; #[cfg(not(target_arch = "wasm32"))] -use tokio::time::Instant; +type IntervalStream = tokio_stream::wrappers::IntervalStream; #[cfg(target_arch = "wasm32")] -use wasm_timer::Instant; +type IntervalStream = gloo_timers::future::IntervalStream; pub fn new_control_channels() -> (ReplyControllerSender, ReplyControllerReceiver) { let (tx, rx) = mpsc::unbounded(); @@ -101,6 +102,35 @@ pub enum ReplyControllerMessage { }, } +pub struct Config { + min_surb_request_size: u32, + max_surb_request_size: u32, + maximum_allowed_reply_surb_request_size: u32, + max_surb_waiting_period: Duration, + max_reply_surb_age: Duration, + max_reply_key_age: Duration, +} + +impl Config { + pub(crate) fn new( + min_surb_request_size: u32, + max_surb_request_size: u32, + maximum_allowed_reply_surb_request_size: u32, + max_surb_waiting_period: Duration, + max_reply_surb_age: Duration, + max_reply_key_age: Duration, + ) -> Self { + Self { + min_surb_request_size, + max_surb_request_size, + maximum_allowed_reply_surb_request_size, + max_surb_waiting_period, + max_reply_surb_age, + max_reply_key_age, + } + } +} + // the purpose of this task: // - buffers split messages from input message listener if there were insufficient surbs to send them // - upon getting extra surbs, resends them @@ -111,6 +141,8 @@ pub enum ReplyControllerMessage { // TODO: this should be split into ingress and egress controllers // because currently its trying to perform two distinct jobs pub struct ReplyController { + config: Config, + // TODO: incorporate that field at some point // and use binomial distribution to determine the expected required number // of surbs required to send the message through @@ -118,44 +150,33 @@ pub struct ReplyController { request_receiver: ReplyControllerReceiver, pending_replies: HashMap>, message_handler: MessageHandler, - received_reply_surbs: ReceivedReplySurbsMap, - tag_storage: UsedSenderTags, - - min_surb_request_size: u32, - max_surb_request_size: u32, - maximum_allowed_reply_surb_request_size: u32, - max_surb_waiting_period: Duration, + full_reply_storage: CombinedReplyStorage, } impl ReplyController where R: CryptoRng + Rng, { - #[allow(clippy::too_many_arguments)] pub(crate) fn new( + config: Config, message_handler: MessageHandler, - received_reply_surbs: ReceivedReplySurbsMap, - tag_storage: UsedSenderTags, + full_reply_storage: CombinedReplyStorage, request_receiver: ReplyControllerReceiver, - min_surb_request_size: u32, - max_surb_request_size: u32, - maximum_allowed_reply_surb_request_size: u32, - max_surb_waiting_period: Duration, ) -> Self { ReplyController { + config, request_receiver, - pending_replies: Default::default(), + pending_replies: HashMap::new(), message_handler, - received_reply_surbs, - tag_storage, - min_surb_request_size, - max_surb_request_size, - maximum_allowed_reply_surb_request_size, - max_surb_waiting_period, + full_reply_storage, } } - fn insert_pending_replies(&mut self, recipient: &AnonymousSenderTag, fragments: Vec) { + fn insert_pending_replies>>( + &mut self, + recipient: &AnonymousSenderTag, + fragments: V, + ) { if let Some(existing) = self.pending_replies.get_mut(recipient) { existing.append(&mut fragments.into()) } else { @@ -173,10 +194,22 @@ where None => return false, }; - let available_surbs = self.received_reply_surbs.available_surbs(target); - let pending_surbs = self.received_reply_surbs.pending_reception(target) as usize; - let min_surbs_threshold = self.received_reply_surbs.min_surb_threshold(); - let max_surbs_threshold = self.received_reply_surbs.max_surb_threshold(); + let available_surbs = self + .full_reply_storage + .surbs_storage_ref() + .available_surbs(target); + let pending_surbs = self + .full_reply_storage + .surbs_storage_ref() + .pending_reception(target) as usize; + let min_surbs_threshold = self + .full_reply_storage + .surbs_storage_ref() + .min_surb_threshold(); + let max_surbs_threshold = self + .full_reply_storage + .surbs_storage_ref() + .max_surb_threshold(); debug!("queue size: {queue_size}, available surbs: {available_surbs} pending surbs: {pending_surbs} threshold range: {min_surbs_threshold}..{max_surbs_threshold}"); @@ -190,7 +223,11 @@ where data: Vec, lane: TransmissionLane, ) { - if !self.received_reply_surbs.contains_surbs_for(&recipient_tag) { + if !self + .full_reply_storage + .surbs_storage_ref() + .contains_surbs_for(&recipient_tag) + { warn!("received reply request for {:?} but we don't have any surbs stored for that recipient!", recipient_tag); return; } @@ -206,7 +243,8 @@ where // (but at some point we run out of surbs for surb requests) let (surbs, _surbs_left) = self - .received_reply_surbs + .full_reply_storage + .surbs_storage_ref() .get_reply_surbs(&recipient_tag, required_surbs); if let Some(reply_surbs) = surbs { @@ -215,8 +253,10 @@ where .try_send_reply_chunks(recipient_tag, fragments, reply_surbs, lane) .await { - // TODO: perhaps there should be some timer here to repeat the request once topology recovers - let err = err.return_unused_surbs(&self.received_reply_surbs, &recipient_tag); + let err = err.return_unused_surbs( + self.full_reply_storage.surbs_storage_ref(), + &recipient_tag, + ); warn!("failed to send reply to {:?} - {err}", recipient_tag); } } else { @@ -236,7 +276,8 @@ where amount: u32, ) -> Result<(), PreparationError> { let reply_surb = self - .received_reply_surbs + .full_reply_storage + .surbs_storage_ref() .get_reply_surb_ignoring_threshold(&target) .and_then(|(reply_surb, _)| reply_surb) .ok_or(PreparationError::NotEnoughSurbs { @@ -249,15 +290,15 @@ where .try_request_additional_reply_surbs(target, reply_surb, amount) .await { - let err = err.return_unused_surbs(&self.received_reply_surbs, &target); + let err = err.return_unused_surbs(self.full_reply_storage.surbs_storage_ref(), &target); warn!( "failed to request additional surbs from {:?} - {err}", target ); - // TODO: perhaps there should be some timer here to repeat the request once topology recovers return Err(err); } else { - self.received_reply_surbs + self.full_reply_storage + .surbs_storage_ref() .increment_pending_reception(&target, amount); } @@ -289,8 +330,14 @@ where async fn try_clear_pending_queue(&mut self, target: AnonymousSenderTag) { trace!("trying to clear pending queue"); - let available_surbs = self.received_reply_surbs.available_surbs(&target); - let min_surbs_threshold = self.received_reply_surbs.min_surb_threshold(); + let available_surbs = self + .full_reply_storage + .surbs_storage_ref() + .available_surbs(&target); + let min_surbs_threshold = self + .full_reply_storage + .surbs_storage_ref() + .min_surb_threshold(); let max_to_clear = if available_surbs > min_surbs_threshold { available_surbs - min_surbs_threshold @@ -302,8 +349,7 @@ where // we're guaranteed to not get more entries than we have reply surbs for if let Some(to_send) = self.pop_at_most_pending_replies(&target, max_to_clear) { - // TODO: optimise: we're cloning the fragments every time to re-insert them into the buffer in case of failure - let to_send_vec = to_send.into_iter().collect::>(); + let to_send_vec = to_send.iter().cloned().collect::>(); if to_send_vec.is_empty() { panic!( @@ -312,10 +358,16 @@ where } let (surbs_for_reply, _) = self - .received_reply_surbs + .full_reply_storage + .surbs_storage_ref() .get_reply_surbs(&target, to_send_vec.len()); - let surbs_for_reply = - surbs_for_reply.expect("is is possible for this to ever show up?"); + + let Some(surbs_for_reply) = surbs_for_reply else { + // probably retransmission + debug!("somehow different task has stolen our reply surbs!"); + self.insert_pending_replies(&target, to_send); + return + }; if let Err(err) = self .message_handler @@ -327,9 +379,10 @@ where ) .await { - let err = err.return_unused_surbs(&self.received_reply_surbs, &target); + let err = + err.return_unused_surbs(self.full_reply_storage.surbs_storage_ref(), &target); + self.insert_pending_replies(&target, to_send); warn!("failed to clear pending queue for {:?} - {err}", target); - // TODO: perhaps there should be some timer here to repeat the request once topology recovers } } else { trace!("the pending queue is empty"); @@ -345,15 +398,19 @@ where trace!("handling received surbs"); // clear the requesting flag since we should have been asking for surbs - self.received_reply_surbs + self.full_reply_storage + .surbs_storage_ref() .reset_surbs_last_received_at(&from); if from_surb_request { - self.received_reply_surbs + self.full_reply_storage + .surbs_storage_ref() .decrement_pending_reception(&from, reply_surbs.len() as u32); } // store received surbs - self.received_reply_surbs.insert_surbs(&from, reply_surbs); + self.full_reply_storage + .surbs_storage_ref() + .insert_surbs(&from, reply_surbs); // use as many as we can for clearing pending queue self.try_clear_pending_queue(from).await; @@ -367,15 +424,19 @@ where async fn handle_surb_request(&mut self, recipient: Recipient, mut amount: u32) { // 1. check whether we sent any surbs in the past to this recipient, otherwise // they have no business in asking for more - if !self.tag_storage.exists(&recipient) { + if !self + .full_reply_storage + .tags_storage_ref() + .exists(&recipient) + { warn!("{recipient} asked us for reply SURBs even though we never sent them any anonymous messages before!"); return; } // 2. check whether the requested amount is within sane range - if amount > self.maximum_allowed_reply_surb_request_size { - warn!("The requested reply surb amount is larger than our maximum allowed ({amount} > {}). Lowering it to a more sane value...", self.maximum_allowed_reply_surb_request_size); - amount = self.maximum_allowed_reply_surb_request_size; + if amount > self.config.maximum_allowed_reply_surb_request_size { + warn!("The requested reply surb amount is larger than our maximum allowed ({amount} > {}). Lowering it to a more sane value...", self.config.maximum_allowed_reply_surb_request_size); + amount = self.config.maximum_allowed_reply_surb_request_size; } // 3. construct and send the surbs away @@ -435,8 +496,8 @@ where } let request_size = min( - self.max_surb_request_size, - max(queue_size, self.min_surb_request_size), + self.config.max_surb_request_size, + max(queue_size, self.config.min_surb_request_size), ); if let Err(err) = self @@ -449,23 +510,33 @@ where async fn inspect_stale_entries(&mut self) { let mut to_request = Vec::new(); + let mut to_remove = Vec::new(); - let now = Instant::now(); + let now = OffsetDateTime::now_utc(); for (pending_reply_target, vals) in &self.pending_replies { if vals.is_empty() { - // TODO: remove it from the map before getting here continue; } - let last_received = self - .received_reply_surbs - .surbs_last_received_at(pending_reply_target) - .expect("I think this shouldnt fail? to be verified."); + let Some(last_received) = self + .full_reply_storage.surbs_storage_ref() + .surbs_last_received_at(pending_reply_target) else { + error!("we have {} pending replies for {pending_reply_target}, but we somehow never received any reply surbs from them!", vals.len()); + to_remove.push(*pending_reply_target); + continue + }; - let diff = now - last_received; + // this should never ever happen (famous last words, eh?), but in case it DOES happen eventually + // purge that malformed data + let Ok(last_received_time) = OffsetDateTime::from_unix_timestamp(last_received) else { + error!("somehow our stored timestamp ({last_received}) for surbs from {pending_reply_target} is corrupted!. Going to remove all the associated entries"); + to_remove.push(*pending_reply_target); + continue + }; + let diff = now - last_received_time; - if diff > self.max_surb_waiting_period { - warn!("We haven't received any surbs in {:?} from {:?}. Going to explicitly ask for more", diff, pending_reply_target); + if diff > self.config.max_surb_waiting_period { + warn!("We haven't received any surbs in {:?} from {pending_reply_target}. Going to explicitly ask for more", diff); to_request.push(*pending_reply_target); } } @@ -473,23 +544,96 @@ where for pending_reply_target in to_request { self.request_reply_surbs_for_queue_clearing(pending_reply_target) .await; - self.received_reply_surbs + self.full_reply_storage + .surbs_storage_ref() .reset_pending_reception(&pending_reply_target) } + + for to_remove in to_remove { + self.pending_replies.remove(&to_remove); + } + } + + async fn invalidate_old_data(&self) { + let now = OffsetDateTime::now_utc(); + + let mut to_remove_surbs = Vec::new(); + let mut to_remove_keys = Vec::new(); + for map_ref in self.full_reply_storage.surbs_storage_ref().as_raw_iter() { + let (sender, received) = map_ref.pair(); + // TODO: handle the following edge case: + // there's a malicious client sending us exactly one reply surb just before we should have invalidated + // the data thus making us keep everything in memory + // possible solution: keep timestamp PER reply surb (but that seems like an overkill) + // but I doubt this is ever going to be a problem... + // ... + // However, if you're reading this message, it probably became a legit problem, + // so I guess add timestamp per surb then? chop-chop. + + let last_received = received.surbs_last_received_at(); + // this should never ever happen (famous last words, eh?), but in case it DOES happen eventually + // purge that malformed data + let Ok(last_received_time) = OffsetDateTime::from_unix_timestamp(last_received) else { + error!("somehow our stored timestamp ({last_received}) for surbs from {sender} is corrupted!. Going to remove all the associated entries"); + to_remove_surbs.push(*sender); + continue + }; + let diff = now - last_received_time; + + if diff > self.config.max_reply_surb_age { + info!("it's been {diff:?} since we last received any reply surb from {sender}. Going to remove all stored entries..."); + + to_remove_surbs.push(*sender); + } + } + + for map_ref in self.full_reply_storage.key_storage_ref().as_raw_iter() { + let (digest, reply_key) = map_ref.pair(); + + // this should never ever happen (famous last words, eh?), but in case it DOES happen eventually + // purge that malformed data + let Ok(sent_at) = OffsetDateTime::from_unix_timestamp(reply_key.sent_at_timestamp) else { + error!("somehow our stored timestamp ({}) for one of our reply key is corrupted!. Going to remove all the entry", reply_key.sent_at_timestamp); + to_remove_keys.push(*digest); + continue + }; + + let diff = now - sent_at; + + if diff > self.config.max_reply_key_age { + debug!("it's been {diff:?} since we created this reply key. it's probably never going to get used, so we're going to purge it..."); + to_remove_keys.push(*digest); + } + } + + for to_remove in to_remove_surbs { + self.full_reply_storage + .surbs_storage_ref() + .remove(&to_remove); + } + + for to_remove in to_remove_keys { + self.full_reply_storage.key_storage().remove(to_remove) + } + } + + fn create_interval_stream(polling_rate: Duration) -> IntervalStream { + #[cfg(not(target_arch = "wasm32"))] + return tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(polling_rate)); + + #[cfg(target_arch = "wasm32")] + return gloo_timers::future::IntervalStream::new(polling_rate.as_millis() as u32); } pub(crate) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) { debug!("Started ReplyController with graceful shutdown support"); let polling_rate = Duration::from_secs(5); + let mut stale_inspection = Self::create_interval_stream(polling_rate); - #[cfg(not(target_arch = "wasm32"))] - let mut interval = - tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(polling_rate)); - - #[cfg(target_arch = "wasm32")] - let mut interval = - gloo_timers::future::IntervalStream::new(polling_rate.as_millis() as u32); + // this is in the order of hours/days so we don't have to poll it that often + let polling_rate = Duration::from_secs(self.config.max_reply_surb_age.as_secs() / 10); + let mut invalidation_inspection = Self::create_interval_stream(polling_rate); while !shutdown.is_shutdown() { tokio::select! { @@ -504,9 +648,12 @@ where break; } }, - _ = interval.next() => { + _ = stale_inspection.next() => { self.inspect_stale_entries().await }, + _ = invalidation_inspection.next() => { + self.invalidate_old_data().await + } } } assert!(shutdown.is_shutdown_poll()); @@ -518,12 +665,12 @@ where debug!("Started ReplyController without graceful shutdown support"); let polling_rate = Duration::from_secs(5); - let mut interval_timer = tokio::time::interval(polling_rate); + let mut interval = Self::create_interval_stream(polling_rate); loop { tokio::select! { req = self.request_receiver.next() => self.handle_request(req.unwrap()).await, - _ = interval_timer.tick() => self.inspect_stale_entries().await + _ = interval.next() => self.inspect_stale_entries().await } } } diff --git a/clients/client-core/src/client/replies/reply_storage/backend/fs_backend/mod.rs b/clients/client-core/src/client/replies/reply_storage/backend/fs_backend/mod.rs index be542e131d..50916e00c3 100644 --- a/clients/client-core/src/client/replies/reply_storage/backend/fs_backend/mod.rs +++ b/clients/client-core/src/client/replies/reply_storage/backend/fs_backend/mod.rs @@ -16,7 +16,6 @@ use nymsphinx::anonymous_replies::requests::AnonymousSenderTag; use std::fs; use std::path::{Path, PathBuf}; use time::OffsetDateTime; -use tokio::time::Instant; mod error; mod manager; @@ -221,7 +220,7 @@ impl Backend { let mut received_surbs = Vec::with_capacity(surb_senders.len()); for sender in surb_senders { let sender_id = sender.id; - let (sender_tag, surbs_last_received_at): (AnonymousSenderTag, Instant) = + let (sender_tag, surbs_last_received_at_timestamp): (AnonymousSenderTag, i64) = sender.try_into()?; let stored_surbs = self .manager @@ -233,7 +232,7 @@ impl Backend { received_surbs.push(( sender_tag, - ReceivedReplySurbs::new_retrieved(stored_surbs, surbs_last_received_at), + ReceivedReplySurbs::new_retrieved(stored_surbs, surbs_last_received_at_timestamp), )) } diff --git a/clients/client-core/src/client/replies/reply_storage/backend/fs_backend/models.rs b/clients/client-core/src/client/replies/reply_storage/backend/fs_backend/models.rs index 19f0c69037..65088ad05e 100644 --- a/clients/client-core/src/client/replies/reply_storage/backend/fs_backend/models.rs +++ b/clients/client-core/src/client/replies/reply_storage/backend/fs_backend/models.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::client::replies::reply_storage::backend::fs_backend::error::StorageError; +use crate::client::replies::reply_storage::key_storage::UsedReplyKey; use crypto::generic_array::typenum::Unsigned; use crypto::Digest; use nymsphinx::addressing::clients::{Recipient, RecipientBytes}; @@ -9,9 +10,6 @@ use nymsphinx::anonymous_replies::encryption_key::EncryptionKeyDigest; use nymsphinx::anonymous_replies::requests::{AnonymousSenderTag, SENDER_TAG_SIZE}; use nymsphinx::anonymous_replies::{ReplySurb, SurbEncryptionKey, SurbEncryptionKeySize}; use nymsphinx::params::ReplySurbKeyDigestAlgorithm; -use std::time::Duration; -use time::OffsetDateTime; -use tokio::time::Instant; #[derive(Debug, Clone)] pub(crate) struct StoredSenderTag { @@ -63,21 +61,20 @@ impl TryFrom for (RecipientBytes, AnonymousSenderTag) { pub(crate) struct StoredReplyKey { pub(crate) key_digest: Vec, pub(crate) reply_key: Vec, + pub(crate) sent_at_timestamp: i64, } impl StoredReplyKey { - pub(crate) fn new( - key_digest: EncryptionKeyDigest, - reply_key: SurbEncryptionKey, - ) -> StoredReplyKey { + pub(crate) fn new(key_digest: EncryptionKeyDigest, reply_key: UsedReplyKey) -> StoredReplyKey { StoredReplyKey { key_digest: key_digest.to_vec(), - reply_key: reply_key.to_bytes(), + reply_key: (*reply_key).to_bytes(), + sent_at_timestamp: reply_key.sent_at_timestamp, } } } -impl TryFrom for (EncryptionKeyDigest, SurbEncryptionKey) { +impl TryFrom for (EncryptionKeyDigest, UsedReplyKey) { type Error = StorageError; fn try_from(value: StoredReplyKey) -> Result { @@ -102,7 +99,10 @@ impl TryFrom for (EncryptionKeyDigest, SurbEncryptionKey) { }); }; - Ok((digest, reply_key)) + Ok(( + digest, + UsedReplyKey::new(reply_key, value.sent_at_timestamp), + )) } } @@ -113,25 +113,18 @@ pub(crate) struct StoredSurbSender { } impl StoredSurbSender { - pub(crate) fn new(tag: AnonymousSenderTag, last_sent: Instant) -> Self { - // this doesn't have to be sub-second accurate - // as a matter of fact even if it's off by few minutes or even hours, - // it would still be good enough - let elapsed = last_sent.elapsed(); - let now = OffsetDateTime::now_utc(); - let last_sent = now - elapsed; - + pub(crate) fn new(tag: AnonymousSenderTag, last_sent_timestamp: i64) -> Self { StoredSurbSender { // for the purposes of STORING data, // we ignore that field anyway id: 0, tag: tag.to_bytes().to_vec(), - last_sent_timestamp: last_sent.unix_timestamp(), + last_sent_timestamp, } } } -impl TryFrom for (AnonymousSenderTag, Instant) { +impl TryFrom for (AnonymousSenderTag, i64) { type Error = StorageError; fn try_from(value: StoredSurbSender) -> Result { @@ -145,26 +138,10 @@ impl TryFrom for (AnonymousSenderTag, Instant) { }); }; - let datetime = - OffsetDateTime::from_unix_timestamp(value.last_sent_timestamp).map_err(|err| { - StorageError::CorruptedData { - details: format!("failed to parse stored timestamp - {err}"), - } - })?; - - let duration_since: Duration = - (OffsetDateTime::now_utc() - datetime) - .try_into() - .map_err(|err| StorageError::CorruptedData { - details: format!( - "failed to extract valid duration from the stored timestamp - {err}" - ), - })?; - - let now = Instant::now(); - let instant = now.checked_sub(duration_since).unwrap_or(now); - - Ok((AnonymousSenderTag::from_bytes(sender_tag_bytes), instant)) + Ok(( + AnonymousSenderTag::from_bytes(sender_tag_bytes), + value.last_sent_timestamp, + )) } } diff --git a/clients/client-core/src/client/replies/reply_storage/combined.rs b/clients/client-core/src/client/replies/reply_storage/combined.rs index 8af328ccdb..c9fb9e4a5c 100644 --- a/clients/client-core/src/client/replies/reply_storage/combined.rs +++ b/clients/client-core/src/client/replies/reply_storage/combined.rs @@ -46,17 +46,14 @@ impl CombinedReplyStorage { self.used_tags.clone() } - #[cfg(not(target_arch = "wasm32"))] pub fn key_storage_ref(&self) -> &SentReplyKeys { &self.sent_reply_keys } - #[cfg(not(target_arch = "wasm32"))] pub fn surbs_storage_ref(&self) -> &ReceivedReplySurbsMap { &self.received_reply_surbs } - #[cfg(not(target_arch = "wasm32"))] pub fn tags_storage_ref(&self) -> &UsedSenderTags { &self.used_tags } diff --git a/clients/client-core/src/client/replies/reply_storage/key_storage.rs b/clients/client-core/src/client/replies/reply_storage/key_storage.rs index 05c6466bc5..9c68f40e06 100644 --- a/clients/client-core/src/client/replies/reply_storage/key_storage.rs +++ b/clients/client-core/src/client/replies/reply_storage/key_storage.rs @@ -1,24 +1,22 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use dashmap::iter::Iter; use dashmap::DashMap; use nymsphinx::anonymous_replies::encryption_key::EncryptionKeyDigest; use nymsphinx::anonymous_replies::SurbEncryptionKey; +use std::ops::Deref; use std::sync::Arc; - -#[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))] -use dashmap::iter::Iter; +use time::OffsetDateTime; #[derive(Debug, Clone)] -// TODO: we might have to also put the tag here -// TODO2: some timestamp to indicate when entries should get purged if we expect to never get the reply back pub struct SentReplyKeys { inner: Arc, } #[derive(Debug)] struct SentReplyKeysInner { - data: DashMap, + data: DashMap, } impl SentReplyKeys { @@ -31,7 +29,7 @@ impl SentReplyKeys { } #[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))] - pub(crate) fn from_raw(raw: Vec<(EncryptionKeyDigest, SurbEncryptionKey)>) -> SentReplyKeys { + pub(crate) fn from_raw(raw: Vec<(EncryptionKeyDigest, UsedReplyKey)>) -> SentReplyKeys { SentReplyKeys { inner: Arc::new(SentReplyKeysInner { data: raw.into_iter().collect(), @@ -39,22 +37,50 @@ impl SentReplyKeys { } } - #[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))] - pub(crate) fn as_raw_iter(&self) -> Iter<'_, EncryptionKeyDigest, SurbEncryptionKey> { + pub(crate) fn as_raw_iter(&self) -> Iter<'_, EncryptionKeyDigest, UsedReplyKey> { self.inner.data.iter() } pub(crate) fn insert_multiple(&self, keys: Vec) { + let now = OffsetDateTime::now_utc().unix_timestamp(); for key in keys { - self.insert(key) + self.insert(UsedReplyKey::new(key, now)) } } - pub(crate) fn insert(&self, key: SurbEncryptionKey) { + pub(crate) fn insert(&self, key: UsedReplyKey) { self.inner.data.insert(key.compute_digest(), key); } - pub(crate) fn try_pop(&self, digest: EncryptionKeyDigest) -> Option { + pub(crate) fn try_pop(&self, digest: EncryptionKeyDigest) -> Option { self.inner.data.remove(&digest).map(|(_k, v)| v) } + + pub(crate) fn remove(&self, digest: EncryptionKeyDigest) { + self.inner.data.remove(&digest); + } +} + +#[derive(Debug, Copy, Clone)] +pub(crate) struct UsedReplyKey { + key: SurbEncryptionKey, + // the purpose of this field is to perform invalidation at relatively very long intervals + pub(crate) sent_at_timestamp: i64, +} + +impl UsedReplyKey { + pub(crate) fn new(key: SurbEncryptionKey, sent_at_timestamp: i64) -> Self { + UsedReplyKey { + key, + sent_at_timestamp, + } + } +} + +impl Deref for UsedReplyKey { + type Target = SurbEncryptionKey; + + fn deref(&self) -> &Self::Target { + &self.key + } } diff --git a/clients/client-core/src/client/replies/reply_storage/surb_storage.rs b/clients/client-core/src/client/replies/reply_storage/surb_storage.rs index e511d3d450..ed2fd8d846 100644 --- a/clients/client-core/src/client/replies/reply_storage/surb_storage.rs +++ b/clients/client-core/src/client/replies/reply_storage/surb_storage.rs @@ -1,6 +1,7 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use dashmap::iter::Iter; use dashmap::DashMap; use log::trace; use nymsphinx::anonymous_replies::requests::AnonymousSenderTag; @@ -8,15 +9,7 @@ use nymsphinx::anonymous_replies::ReplySurb; use std::collections::VecDeque; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; - -#[cfg(not(target_arch = "wasm32"))] -use tokio::time::Instant; - -#[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))] -use dashmap::iter::Iter; - -#[cfg(target_arch = "wasm32")] -use wasm_timer::Instant; +use time::OffsetDateTime; #[derive(Debug, Clone)] pub struct ReceivedReplySurbsMap { @@ -63,18 +56,21 @@ impl ReceivedReplySurbsMap { } } - #[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))] pub(crate) fn as_raw_iter(&self) -> Iter<'_, AnonymousSenderTag, ReceivedReplySurbs> { self.inner.data.iter() } + pub(crate) fn remove(&self, target: &AnonymousSenderTag) { + self.inner.data.remove(target); + } + pub(crate) fn reset_surbs_last_received_at(&self, target: &AnonymousSenderTag) { if let Some(mut entry) = self.inner.data.get_mut(target) { - entry.surbs_last_received_at = Instant::now(); + entry.surbs_last_received_at_timestamp = OffsetDateTime::now_utc().unix_timestamp(); } } - pub(crate) fn surbs_last_received_at(&self, target: &AnonymousSenderTag) -> Option { + pub(crate) fn surbs_last_received_at(&self, target: &AnonymousSenderTag) -> Option { self.inner .data .get(target) @@ -199,7 +195,7 @@ pub(crate) struct ReceivedReplySurbs { data: VecDeque, pending_reception: u32, - surbs_last_received_at: Instant, + surbs_last_received_at_timestamp: i64, } impl ReceivedReplySurbs { @@ -207,19 +203,19 @@ impl ReceivedReplySurbs { ReceivedReplySurbs { data: initial_surbs, pending_reception: 0, - surbs_last_received_at: Instant::now(), + surbs_last_received_at_timestamp: OffsetDateTime::now_utc().unix_timestamp(), } } #[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))] pub(crate) fn new_retrieved( surbs: Vec, - surbs_last_received_at: Instant, + surbs_last_received_at_timestamp: i64, ) -> ReceivedReplySurbs { ReceivedReplySurbs { data: surbs.into(), pending_reception: 0, - surbs_last_received_at, + surbs_last_received_at_timestamp, } } @@ -228,8 +224,8 @@ impl ReceivedReplySurbs { &self.data } - pub(crate) fn surbs_last_received_at(&self) -> Instant { - self.surbs_last_received_at + pub(crate) fn surbs_last_received_at(&self) -> i64 { + self.surbs_last_received_at_timestamp } pub(crate) fn pending_reception(&self) -> u32 { @@ -276,7 +272,7 @@ impl ReceivedReplySurbs { let mut v = surbs.into_iter().collect::>(); trace!("storing {} surbs in the storage", v.len()); self.data.append(&mut v); - self.surbs_last_received_at = Instant::now(); + self.surbs_last_received_at_timestamp = OffsetDateTime::now_utc().unix_timestamp(); trace!("we now have {} surbs!", self.data.len()); } } diff --git a/clients/client-core/src/config/mod.rs b/clients/client-core/src/config/mod.rs index 23ce0ef54d..ea5ac339e1 100644 --- a/clients/client-core/src/config/mod.rs +++ b/clients/client-core/src/config/mod.rs @@ -47,6 +47,12 @@ const DEFAULT_MAXIMUM_ALLOWED_SURB_REQUEST_SIZE: u32 = 500; const DEFAULT_RETRANSMISSION_REPLY_SURB_REQUEST_SIZE: u32 = 10; const DEFAULT_MAXIMUM_REPLY_SURB_WAITING_PERIOD: Duration = Duration::from_secs(10); +// 12 hours +const DEFAULT_MAXIMUM_REPLY_SURB_AGE: Duration = Duration::from_secs(12 * 60 * 60); + +// 24 hours +const DEFAULT_MAXIMUM_REPLY_KEY_AGE: Duration = Duration::from_secs(24 * 60 * 60); + pub fn missing_string_value() -> String { MISSING_VALUE.to_string() } @@ -323,6 +329,14 @@ impl Config { pub fn get_maximum_reply_surb_waiting_period(&self) -> Duration { self.debug.maximum_reply_surb_waiting_period } + + pub fn get_maximum_reply_surb_age(&self) -> Duration { + self.debug.maximum_reply_surb_age + } + + pub fn get_maximum_reply_key_age(&self) -> Duration { + self.debug.maximum_reply_key_age + } } impl Default for Config { @@ -570,6 +584,16 @@ pub struct DebugConfig { /// for more even though in theory they wouldn't need to. #[serde(with = "humantime_serde")] pub maximum_reply_surb_waiting_period: Duration, + + /// Defines maximum amount of time given reply surb is going to be valid for. + /// This is going to be superseded by key rotation once implemented. + #[serde(with = "humantime_serde")] + pub maximum_reply_surb_age: Duration, + + /// Defines maximum amount of time given reply key is going to be valid for. + /// This is going to be superseded by key rotation once implemented. + #[serde(with = "humantime_serde")] + pub maximum_reply_key_age: Duration, } #[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)] @@ -602,6 +626,8 @@ impl Default for DebugConfig { maximum_allowed_reply_surb_request_size: DEFAULT_MAXIMUM_ALLOWED_SURB_REQUEST_SIZE, retransmission_reply_surb_request_size: DEFAULT_RETRANSMISSION_REPLY_SURB_REQUEST_SIZE, maximum_reply_surb_waiting_period: DEFAULT_MAXIMUM_REPLY_SURB_WAITING_PERIOD, + maximum_reply_surb_age: DEFAULT_MAXIMUM_REPLY_SURB_AGE, + maximum_reply_key_age: DEFAULT_MAXIMUM_REPLY_KEY_AGE, } } } diff --git a/clients/webassembly/src/client/config.rs b/clients/webassembly/src/client/config.rs index f16584c5a3..af03e1575a 100644 --- a/clients/webassembly/src/client/config.rs +++ b/clients/webassembly/src/client/config.rs @@ -129,6 +129,14 @@ pub struct Debug { /// Defines maximum amount of time the client is going to wait for reply surbs before explicitly asking /// for more even though in theory they wouldn't need to. pub maximum_reply_surb_waiting_period_ms: u64, + + /// Defines maximum amount of time given reply surb is going to be valid for. + /// This is going to be superseded by key rotation once implemented. + pub maximum_reply_surb_age_ms: u64, + + /// Defines maximum amount of time given reply key is going to be valid for. + /// This is going to be superseded by key rotation once implemented. + pub maximum_reply_key_age_ms: u64, } impl From for ConfigDebug { @@ -167,6 +175,8 @@ impl From for ConfigDebug { maximum_reply_surb_waiting_period: Duration::from_millis( debug.maximum_reply_surb_waiting_period_ms, ), + maximum_reply_surb_age: Duration::from_millis(debug.maximum_reply_surb_age_ms), + maximum_reply_key_age: Duration::from_millis(debug.maximum_reply_key_age_ms), } } } @@ -198,6 +208,8 @@ impl From for Debug { maximum_reply_surb_waiting_period_ms: debug .maximum_reply_surb_waiting_period .as_millis() as u64, + maximum_reply_surb_age_ms: debug.maximum_reply_surb_age.as_millis() as u64, + maximum_reply_key_age_ms: debug.maximum_reply_key_age.as_millis() as u64, } } }