From e5878d656de662344ab4e8ba5417fad87da48862 Mon Sep 17 00:00:00 2001 From: yahya <19204398+yhassanzadeh13@users.noreply.github.com> Date: Sun, 28 Jun 2026 17:35:01 -0700 Subject: [PATCH 1/3] [Chore] Clear stale rate-limit cache entries --- node/router/src/heartbeat.rs | 10 ++++++ node/router/src/helpers/cache.rs | 55 ++++++++++++++++++++++++++++++++ node/router/src/inbound.rs | 6 ++-- node/router/src/lib.rs | 4 +-- 4 files changed, 70 insertions(+), 5 deletions(-) diff --git a/node/router/src/heartbeat.rs b/node/router/src/heartbeat.rs index 9b53d286bc..081cfbd6bf 100644 --- a/node/router/src/heartbeat.rs +++ b/node/router/src/heartbeat.rs @@ -19,6 +19,7 @@ use crate::{ NodeType, Outbound, PeerPoolHandling, + Router, bootstrap_peers, messages::{DisconnectReason, Message, PeerRequest}, }; @@ -79,6 +80,8 @@ pub trait Heartbeat: Outbound { self.handle_puzzle_request(); // Unban any addresses whose ban time has expired. self.handle_banned_ips(); + // Clear stale rate-limit cache entries. + self.clear_stale_peers(); } /// TODO (howardwu): Consider checking minimum number of validators, to exclude clients and provers. @@ -373,4 +376,11 @@ pub trait Heartbeat: Outbound { fn handle_banned_ips(&self) { self.router().tcp().banned_peers().remove_old_bans(Self::IP_BAN_TIME_IN_SECS); } + + fn clear_stale_peers(&self) { + self.router().cache().clear_stale_entries( + Router::::CONNECTION_ATTEMPTS_SINCE_SECS, + Router::::MESSAGE_LIMIT_TIME_FRAME_IN_SECS, + ); + } } diff --git a/node/router/src/helpers/cache.rs b/node/router/src/helpers/cache.rs index d370dac7b7..be1ed69fe8 100644 --- a/node/router/src/helpers/cache.rs +++ b/node/router/src/helpers/cache.rs @@ -218,6 +218,25 @@ impl Cache { pub fn clear_peer_entries(&self, peer_ip: SocketAddr) { self.seen_outbound_block_requests.write().remove(&peer_ip); } + + /// Removes fully-expired entries from the inbound rate-limit caches, bounding their memory growth. + /// + /// An entry whose timestamps are *all* older than its interval carries no rate-limit weight, so + /// dropping it is observationally a no-op — a peer's live limit is never reset, and fresh data is + /// never evicted. Intended to be called periodically (e.g. from the heartbeat). + /// + /// `connection_interval_in_secs` and `message_interval_in_secs` are the windows for the connection + /// and message caches, which the caller owns rather than this cache. + pub fn clear_stale_entries(&self, connection_interval_in_secs: i64, message_interval_in_secs: i64) { + Self::clear_expired_entries(&self.seen_inbound_connections, connection_interval_in_secs); + Self::clear_expired_entries(&self.seen_inbound_messages, message_interval_in_secs); + Self::clear_expired_entries(&self.seen_inbound_puzzle_requests, Self::INBOUND_PUZZLE_REQUEST_INTERVAL); + Self::clear_expired_entries(&self.seen_inbound_block_requests, Self::INBOUND_BLOCK_REQUEST_INTERVAL); + Self::clear_expired_entries( + &self.seen_inbound_unconfirmed_solutions, + Self::INBOUND_UNCONFIRMED_SOLUTION_INTERVAL, + ); + } } impl Cache { @@ -310,6 +329,20 @@ impl Cache { // Return the previous timestamp. previous_timestamp } + + /// Clears expired entries from the map; pops the expired entries from the front of the deque and if the deque is empty, removes the key. + fn clear_expired_entries(map: &RwLock>>, interval_in_secs: i64) { + let mut map_write = map.write(); + let now = OffsetDateTime::now_utc(); + map_write.retain(|_, timestamps| { + while timestamps.front().is_some_and(|t| now - *t > Duration::seconds(interval_in_secs)) { + timestamps.pop_front(); + } + + // If the deque is empty, remove the key (returning false to remove the key) + !timestamps.is_empty() + }) + } } #[cfg(test)] @@ -469,4 +502,26 @@ mod tests { // Check the cache is empty. assert!(!cache.contains_outbound_peer_request(peer_ip)); } + + #[test] + fn test_seen_inbound_messages_clears_stale_entries() { + let cache = Cache::::default(); + // A timestamp well outside any rate-limit window. + let old = OffsetDateTime::now_utc() - Duration::seconds(120); + + // seed the cache with 1000 distinct peers, each with a single entry. + { + let mut map = cache.seen_inbound_messages.write(); + for port in 1..=1000u16 { + let peer_ip = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), port); + map.insert(peer_ip, VecDeque::from([old])); + } + } + + // realistic connection/message windows; 120s ≫ 5s so they're all expired + cache.clear_stale_entries(10, 5); + + // all entries should have been removed, not just emptied + assert!(cache.seen_inbound_messages.read().is_empty()); + } } diff --git a/node/router/src/inbound.rs b/node/router/src/inbound.rs index db4ae7e43e..fbf4eee567 100644 --- a/node/router/src/inbound.rs +++ b/node/router/src/inbound.rs @@ -16,6 +16,7 @@ use crate::{ Outbound, PeerPoolHandling, + Router, messages::{ BlockRequest, BlockResponse, @@ -53,8 +54,6 @@ pub trait Inbound: Reading + Outbound { const MAXIMUM_UNCONFIRMED_SOLUTIONS_PER_INTERVAL: usize = 64; /// The duration in seconds to sleep in between ping requests with a connected peer. const PING_SLEEP_IN_SECS: u64 = 20; // 20 seconds - /// The time frame to enforce the `MESSAGE_LIMIT`. - const MESSAGE_LIMIT_TIME_FRAME_IN_SECS: i64 = 5; /// The maximum number of messages accepted within `MESSAGE_LIMIT_TIME_FRAME_IN_SECS`. const MESSAGE_LIMIT: usize = 500; @@ -91,7 +90,8 @@ pub trait Inbound: Reading + Outbound { // Drop the peer, if they have sent more than `MESSAGE_LIMIT` messages // in the last `MESSAGE_LIMIT_TIME_FRAME_IN_SECS` seconds. - let num_messages = self.router().cache.insert_inbound_message(peer_ip, Self::MESSAGE_LIMIT_TIME_FRAME_IN_SECS); + let num_messages = + self.router().cache.insert_inbound_message(peer_ip, Router::::MESSAGE_LIMIT_TIME_FRAME_IN_SECS); if num_messages > Self::MESSAGE_LIMIT { bail!("Dropping '{peer_ip}' for spamming messages (num_messages = {num_messages})") } diff --git a/node/router/src/lib.rs b/node/router/src/lib.rs index 98c9942cc8..341afd925b 100644 --- a/node/router/src/lib.rs +++ b/node/router/src/lib.rs @@ -141,12 +141,12 @@ pub struct InnerRouter { } impl Router { - /// The minimum permitted interval between connection attempts for an IP; anything shorter is considered malicious. - #[cfg(not(feature = "test"))] const CONNECTION_ATTEMPTS_SINCE_SECS: i64 = 10; /// The maximum amount of connection attempts within a 10 second threshold. #[cfg(not(feature = "test"))] const MAX_CONNECTION_ATTEMPTS: usize = 10; + /// The time frame to enforce the `MESSAGE_LIMIT`. + const MESSAGE_LIMIT_TIME_FRAME_IN_SECS: i64 = 5; } impl Router { From 6d776c26ba08e3c15a792e274f545f3af1898777 Mon Sep 17 00:00:00 2001 From: yahya <19204398+yhassanzadeh13@users.noreply.github.com> Date: Sun, 28 Jun 2026 22:17:20 -0700 Subject: [PATCH 2/3] [Test] Cover clear_stale_entries guarantees; fix clear_peer_entries doc --- node/router/src/helpers/cache.rs | 86 +++++++++++++++++++++++++++++++- 1 file changed, 84 insertions(+), 2 deletions(-) diff --git a/node/router/src/helpers/cache.rs b/node/router/src/helpers/cache.rs index be1ed69fe8..d897da8245 100644 --- a/node/router/src/helpers/cache.rs +++ b/node/router/src/helpers/cache.rs @@ -214,7 +214,10 @@ impl Cache { Self::decrement_counter(&self.seen_outbound_peer_requests, peer_ip) } - /// Removes all cache entries applicable to the given key. + /// Removes the given peer's outbound block-request entries. + /// + /// Other per-peer caches are not pruned here; their stale entries are reclaimed + /// periodically by [`Cache::clear_stale_entries`]. pub fn clear_peer_entries(&self, peer_ip: SocketAddr) { self.seen_outbound_block_requests.write().remove(&peer_ip); } @@ -504,7 +507,7 @@ mod tests { } #[test] - fn test_seen_inbound_messages_clears_stale_entries() { + fn test_clear_stale_entries_removes_expired_keys() { let cache = Cache::::default(); // A timestamp well outside any rate-limit window. let old = OffsetDateTime::now_utc() - Duration::seconds(120); @@ -524,4 +527,83 @@ mod tests { // all entries should have been removed, not just emptied assert!(cache.seen_inbound_messages.read().is_empty()); } + + #[test] + fn test_clear_stale_entries_preserves_fresh_entries() { + let cache = Cache::::default(); + let peer_ip = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234); + + // Record three recent messages (count climbs to 3). + assert_eq!(cache.insert_inbound_message(peer_ip, 5), 1); + assert_eq!(cache.insert_inbound_message(peer_ip, 5), 2); + assert_eq!(cache.insert_inbound_message(peer_ip, 5), 3); + + // Sweep with realistic windows; these entries are sub-second old — well + // inside the 5s message window — so they must be retained. + cache.clear_stale_entries(10, 5); + // All three timestamps survive: nothing evicted, count not reset. + assert!(cache.seen_inbound_messages.read().get(&peer_ip).is_some_and(|v| v.len() == 3)); + // One more message, count climbs to 4. + assert_eq!(cache.insert_inbound_message(peer_ip, 5), 4); + } + + #[test] + fn test_clear_stale_entries_trims_expired_but_keeps_active_keys() { + let cache = Cache::::default(); + let peer_ip = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234); + + // two named timestamps, one expired one not + let old = OffsetDateTime::now_utc() - Duration::seconds(120); + let fresh = OffsetDateTime::now_utc(); + { + let mut map = cache.seen_inbound_messages.write(); + map.insert(peer_ip, VecDeque::from([old, fresh])); + } + + // with 5s message window, old (120s) is expired -> popped; the loop then hits fresh (not expired) -> stops, + // the key is non-empty and thus not removed + cache.clear_stale_entries(10, 5); + + // key kept, trimmed to one entry + assert!(cache.seen_inbound_messages.read().get(&peer_ip).is_some_and(|v| v.len() == 1)); + // and the only entry is the fresh one + assert!( + cache.seen_inbound_messages.read().get(&peer_ip).is_some_and(|v| v.front().is_some_and(|t| *t == fresh)) + ); + } + + #[test] + fn test_clear_stale_entries_sweeps_all_inbound_maps() { + let cache = Cache::::default(); + let peer_ip = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234); + let old = OffsetDateTime::now_utc() - Duration::seconds(120); + { + let mut map = cache.seen_inbound_messages.write(); + map.insert(peer_ip, VecDeque::from([old])); + let mut map = cache.seen_inbound_block_requests.write(); + map.insert(peer_ip, VecDeque::from([old])); + let mut map = cache.seen_inbound_puzzle_requests.write(); + map.insert(peer_ip, VecDeque::from([old])); + let mut map = cache.seen_inbound_unconfirmed_solutions.write(); + map.insert(peer_ip, VecDeque::from([old])); + let mut map = cache.seen_inbound_connections.write(); + map.insert(peer_ip.ip(), VecDeque::from([old])); + } + + assert_eq!(cache.seen_inbound_messages.read().len(), 1); + assert_eq!(cache.seen_inbound_block_requests.read().len(), 1); + assert_eq!(cache.seen_inbound_puzzle_requests.read().len(), 1); + assert_eq!(cache.seen_inbound_unconfirmed_solutions.read().len(), 1); + assert_eq!(cache.seen_inbound_connections.read().len(), 1); + + // 120s is older than every window of (10, 5) and the internal Self::INBOUND_*_INTERVAL constants (60). So all 5 must reap. + cache.clear_stale_entries(10, 5); + + // assert that all maps are empty + assert!(cache.seen_inbound_messages.read().is_empty()); + assert!(cache.seen_inbound_block_requests.read().is_empty()); + assert!(cache.seen_inbound_puzzle_requests.read().is_empty()); + assert!(cache.seen_inbound_unconfirmed_solutions.read().is_empty()); + assert!(cache.seen_inbound_connections.read().is_empty()); + } } From 21ab2613fd12e12d3b9a88373180b6965051a204 Mon Sep 17 00:00:00 2001 From: yahya <19204398+yhassanzadeh13@users.noreply.github.com> Date: Sun, 28 Jun 2026 22:23:44 -0700 Subject: [PATCH 3/3] [Doc] reverts back a comment --- node/router/src/lib.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/node/router/src/lib.rs b/node/router/src/lib.rs index 341afd925b..a30c2a47db 100644 --- a/node/router/src/lib.rs +++ b/node/router/src/lib.rs @@ -141,6 +141,7 @@ pub struct InnerRouter { } impl Router { + /// The minimum permitted interval between connection attempts for an IP; anything shorter is considered malicious. const CONNECTION_ATTEMPTS_SINCE_SECS: i64 = 10; /// The maximum amount of connection attempts within a 10 second threshold. #[cfg(not(feature = "test"))]