mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

node: Don't rate-limit persistent peers

This commit is contained in:
cloudhead 2024-04-30 14:40:05 +02:00
parent 9cd08a01db
commit afe13b9e47
No known key found for this signature in database
3 changed files with 85 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
radicle-node/src/service.rs
View File

@ -438,6 +438,7 @@ where
emitter: Emitter<Event>,
) -> Self {
let sessions = Sessions::new(rng.clone());
let limiter = RateLimiter::new(config.peers());
Self {
config,
@ -449,7 +450,7 @@ where
clock,
db,
outbox: Outbox::default(),
limiter: RateLimiter::default(),
limiter,
sessions,
fetching: HashMap::new(),
queue: VecDeque::new(),
@ -1155,10 +1156,12 @@ where
}
let host: HostName = addr.into();
if self
.limiter
.limit(host.clone(), &self.config.limits.rate.inbound, self.clock)
{
if self.limiter.limit(
host.clone(),
None,
&self.config.limits.rate.inbound,
self.clock,
) {
trace!(target: "service", "Rate limiting inbound connection from {host}..");
return false;
}
@ -1633,7 +1636,7 @@ where
};
if self
.limiter
.limit(peer.addr.clone().into(), limit, self.clock)
.limit(peer.addr.clone().into(), Some(remote), limit, self.clock)
{
debug!(target: "service", "Rate limiting message from {remote} ({})", peer.addr);
return Ok(());

114
radicle-node/src/service/limiter.rs
View File

@ -1,7 +1,7 @@
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use localtime::LocalTime;
use radicle::node::{address, config, HostName};
use radicle::node::{address, config, HostName, NodeId};
/// Peer rate limiter.
///
@ -12,15 +12,35 @@ use radicle::node::{address, config, HostName};
#[derive(Debug, Default)]
pub struct RateLimiter {
buckets: HashMap<HostName, TokenBucket>,
bypass: HashSet<NodeId>,
}
impl RateLimiter {
/// Create a new rate limiter with a bypass list. Nodes in the bypass list are not limited.
pub fn new(bypass: impl IntoIterator<Item = NodeId>) -> Self {
Self {
buckets: HashMap::default(),
bypass: bypass.into_iter().collect(),
}
}
/// Call this when the address has performed some rate-limited action.
/// Returns whether the action is rate-limited or not.
///
/// Supplying a different amount of tokens per address is useful if for eg. a peer
/// is outbound vs. inbound.
pub fn limit<T: AsTokens>(&mut self, addr: HostName, tokens: &T, now: LocalTime) -> bool {
pub fn limit<T: AsTokens>(
&mut self,
addr: HostName,
nid: Option<&NodeId>,
tokens: &T,
now: LocalTime,
) -> bool {
if let Some(nid) = nid {
if self.bypass.contains(nid) {
return false;
}
}
if let HostName::Ip(ip) = addr {
// Don't limit LAN addresses.
if !address::is_routable(&ip) {
@ -99,6 +119,8 @@ impl TokenBucket {
#[cfg(test)]
#[allow(clippy::bool_assert_comparison, clippy::redundant_clone)]
mod test {
use radicle::test::arbitrary;
use super::*;
impl AsTokens for (usize, f64) {
@ -116,64 +138,72 @@ mod test {
let mut r = RateLimiter::default();
let t = (3, 0.2); // Three tokens burst. One token every 5 seconds.
let a = HostName::Dns(String::from("seed.radicle.xyz"));
let n = arbitrary::gen::<NodeId>(1);
let n = Some(&n);
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(0)), false); // Burst capacity
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(1)), false); // Burst capacity
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(2)), false); // Burst capacity
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(3)), true); // Limited
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(4)), true); // Limited
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(5)), false); // Refilled (1)
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(6)), true); // Limited
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(7)), true); // Limited
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(8)), true); // Limited
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(9)), true); // Limited
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(10)), false); // Refilled (1)
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(11)), true); // Limited
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(12)), true); // Limited
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(13)), true); // Limited
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(14)), true); // Limited
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(15)), false); // Refilled (1)
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(16)), true); // Limited
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(60)), false); // Refilled (3)
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(60)), false); // Burst capacity
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(60)), false); // Burst capacity
assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(60)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(0)), false); // Burst capacity
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(1)), false); // Burst capacity
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(2)), false); // Burst capacity
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(3)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(4)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(5)), false); // Refilled (1)
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(6)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(7)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(8)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(9)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(10)), false); // Refilled (1)
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(11)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(12)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(13)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(14)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(15)), false); // Refilled (1)
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(16)), true); // Limited
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(60)), false); // Refilled (3)
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(60)), false); // Burst capacity
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(60)), false); // Burst capacity
assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(60)), true); // Limited
}
#[test]
#[rustfmt::skip]
fn test_limitter_multi() {
let t = (1, 1.0); // One token per second. One token burst.
let n = arbitrary::gen::<NodeId>(1);
let n = Some(&n);
let mut r = RateLimiter::default();
let addr1 = HostName::Dns(String::from("seed.radicle.xyz"));
let addr2 = HostName::Dns(String::from("seed.radicle.net"));
assert_eq!(r.limit(addr1.clone(), &t, LocalTime::from_secs(0)), false);
assert_eq!(r.limit(addr1.clone(), &t, LocalTime::from_secs(0)), true);
assert_eq!(r.limit(addr2.clone(), &t, LocalTime::from_secs(0)), false);
assert_eq!(r.limit(addr2.clone(), &t, LocalTime::from_secs(0)), true);
assert_eq!(r.limit(addr1.clone(), &t, LocalTime::from_secs(1)), false); // Refilled (1)
assert_eq!(r.limit(addr1.clone(), &t, LocalTime::from_secs(1)), true);
assert_eq!(r.limit(addr2.clone(), &t, LocalTime::from_secs(1)), false);
assert_eq!(r.limit(addr2.clone(), &t, LocalTime::from_secs(1)), true);
assert_eq!(r.limit(addr1.clone(), n, &t, LocalTime::from_secs(0)), false);
assert_eq!(r.limit(addr1.clone(), n, &t, LocalTime::from_secs(0)), true);
assert_eq!(r.limit(addr2.clone(), n, &t, LocalTime::from_secs(0)), false);
assert_eq!(r.limit(addr2.clone(), n, &t, LocalTime::from_secs(0)), true);
assert_eq!(r.limit(addr1.clone(), n, &t, LocalTime::from_secs(1)), false);
assert_eq!(r.limit(addr1.clone(), n, &t, LocalTime::from_secs(1)), true);
assert_eq!(r.limit(addr2.clone(), n, &t, LocalTime::from_secs(1)), false);
assert_eq!(r.limit(addr2.clone(), n, &t, LocalTime::from_secs(1)), true);
}
#[test]
#[rustfmt::skip]
fn test_limitter_different_rates() {
let t1 = (1, 1.0); // One token per second. One token burst.
let t2 = (2, 2.0); // Two tokens per second. Two token burst.
let n = arbitrary::gen::<NodeId>(1);
let n = Some(&n);
let mut r = RateLimiter::default();
let addr1 = HostName::Dns(String::from("seed.radicle.xyz"));
let addr2 = HostName::Dns(String::from("seed.radicle.net"));
assert_eq!(r.limit(addr1.clone(), &t1, LocalTime::from_secs(0)), false);
assert_eq!(r.limit(addr1.clone(), &t1, LocalTime::from_secs(0)), true);
assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(0)), false);
assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(0)), false);
assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(0)), true);
assert_eq!(r.limit(addr1.clone(), &t1, LocalTime::from_secs(1)), false); // Refilled (1)
assert_eq!(r.limit(addr1.clone(), &t1, LocalTime::from_secs(1)), true);
assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(1)), false); // Refilled (2)
assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(1)), false);
assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(1)), true);
assert_eq!(r.limit(addr1.clone(), n, &t1, LocalTime::from_secs(0)), false);
assert_eq!(r.limit(addr1.clone(), n, &t1, LocalTime::from_secs(0)), true);
assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(0)), false);
assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(0)), false);
assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(0)), true);
assert_eq!(r.limit(addr1.clone(), n, &t1, LocalTime::from_secs(1)), false); // Refilled (1)
assert_eq!(r.limit(addr1.clone(), n, &t1, LocalTime::from_secs(1)), true);
assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(1)), false); // Refilled (2)
assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(1)), false);
assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(1)), true);
}
}

4
radicle/src/node/config.rs
View File

@ -326,6 +326,10 @@ impl Config {
.map(|ca| &ca.addr)
}
pub fn peers(&self) -> impl Iterator<Item = NodeId> + '_ {
self.connect.iter().cloned().map(|p| p.id)
}
pub fn is_persistent(&self, id: &NodeId) -> bool {
self.peer(id).is_some()
}