fix(crawler): evict idle hosts from the per-host rate-limiter map
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2
backend/Cargo.lock
generated
2
backend/Cargo.lock
generated
@@ -1558,7 +1558,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mangalord"
|
name = "mangalord"
|
||||||
version = "0.90.7"
|
version = "0.90.8"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"argon2",
|
"argon2",
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mangalord"
|
name = "mangalord"
|
||||||
version = "0.90.7"
|
version = "0.90.8"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
default-run = "mangalord"
|
default-run = "mangalord"
|
||||||
|
|
||||||
|
|||||||
@@ -44,6 +44,25 @@ impl RateLimiter {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Once the per-host map grows past this many entries, the next `wait_for`
|
||||||
|
/// sweeps out hosts idle longer than [`IDLE_EVICT_AFTER`]. A long crawl that
|
||||||
|
/// touches thousands of distinct CDN shards would otherwise retain one bucket
|
||||||
|
/// per host for the daemon's whole lifetime. Far above any single source's
|
||||||
|
/// real host count, so the sweep is rare.
|
||||||
|
const MAX_TRACKED_HOSTS: usize = 1024;
|
||||||
|
|
||||||
|
/// A host bucket untouched for at least this long is evicted on the next
|
||||||
|
/// over-cap sweep. Generous: a host still being crawled is touched every
|
||||||
|
/// `interval`, so only genuinely-finished hosts age out.
|
||||||
|
const IDLE_EVICT_AFTER: Duration = Duration::from_secs(3600);
|
||||||
|
|
||||||
|
/// A host's bucket plus when it was last used, so idle entries can be evicted.
|
||||||
|
#[derive(Debug)]
|
||||||
|
struct HostEntry {
|
||||||
|
limiter: Arc<Mutex<RateLimiter>>,
|
||||||
|
last_used: Instant,
|
||||||
|
}
|
||||||
|
|
||||||
/// Per-host rate limiter map. The outer `Mutex<HashMap>` is held only
|
/// Per-host rate limiter map. The outer `Mutex<HashMap>` is held only
|
||||||
/// during the entry-or-insert + Arc clone; the per-host `Mutex<RateLimiter>`
|
/// during the entry-or-insert + Arc clone; the per-host `Mutex<RateLimiter>`
|
||||||
/// is held during the actual `wait().await`. So N workers calling
|
/// is held during the actual `wait().await`. So N workers calling
|
||||||
@@ -54,7 +73,7 @@ impl RateLimiter {
|
|||||||
pub struct HostRateLimiters {
|
pub struct HostRateLimiters {
|
||||||
default_interval: Duration,
|
default_interval: Duration,
|
||||||
overrides: HashMap<String, Duration>,
|
overrides: HashMap<String, Duration>,
|
||||||
map: Mutex<HashMap<String, Arc<Mutex<RateLimiter>>>>,
|
map: Mutex<HashMap<String, HostEntry>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl HostRateLimiters {
|
impl HostRateLimiters {
|
||||||
@@ -82,20 +101,37 @@ impl HostRateLimiters {
|
|||||||
.ok_or_else(|| anyhow::anyhow!("no host in url: {url}"))?;
|
.ok_or_else(|| anyhow::anyhow!("no host in url: {url}"))?;
|
||||||
let limiter = {
|
let limiter = {
|
||||||
let mut map = self.map.lock().await;
|
let mut map = self.map.lock().await;
|
||||||
map.entry(host.clone())
|
let now = Instant::now();
|
||||||
.or_insert_with(|| {
|
// Bound the map: when it grows past the soft cap, drop hosts that
|
||||||
let interval = self
|
// have been idle past the TTL. Only fires over-cap, so the common
|
||||||
.overrides
|
// path is a plain lookup. A host still being crawled is touched
|
||||||
.get(&host)
|
// every `interval` and so never ages out.
|
||||||
.copied()
|
if map.len() >= MAX_TRACKED_HOSTS {
|
||||||
.unwrap_or(self.default_interval);
|
map.retain(|_, e| now.duration_since(e.last_used) < IDLE_EVICT_AFTER);
|
||||||
Arc::new(Mutex::new(RateLimiter::new(interval)))
|
}
|
||||||
})
|
let entry = map.entry(host.clone()).or_insert_with(|| {
|
||||||
.clone()
|
let interval = self
|
||||||
|
.overrides
|
||||||
|
.get(&host)
|
||||||
|
.copied()
|
||||||
|
.unwrap_or(self.default_interval);
|
||||||
|
HostEntry {
|
||||||
|
limiter: Arc::new(Mutex::new(RateLimiter::new(interval))),
|
||||||
|
last_used: now,
|
||||||
|
}
|
||||||
|
});
|
||||||
|
entry.last_used = now;
|
||||||
|
entry.limiter.clone()
|
||||||
};
|
};
|
||||||
limiter.lock().await.wait().await;
|
limiter.lock().await.wait().await;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Number of host buckets currently tracked. Test/observability hook.
|
||||||
|
#[cfg(test)]
|
||||||
|
pub(crate) async fn tracked_hosts(&self) -> usize {
|
||||||
|
self.map.lock().await.len()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// `host_of` was duplicated across session/rate_limit/pipeline; the
|
// `host_of` was duplicated across session/rate_limit/pipeline; the
|
||||||
@@ -165,6 +201,46 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test(start_paused = true)]
|
||||||
|
async fn host_rate_limiters_evict_idle_hosts_over_cap() {
|
||||||
|
// Fill the map to the soft cap with distinct hosts (first call to a
|
||||||
|
// fresh host never sleeps, so this is fast even under real time).
|
||||||
|
let rl = HostRateLimiters::new(Duration::from_millis(1));
|
||||||
|
for i in 0..MAX_TRACKED_HOSTS {
|
||||||
|
rl.wait_for(&format!("https://host{i}.example/x")).await.unwrap();
|
||||||
|
}
|
||||||
|
assert_eq!(rl.tracked_hosts().await, MAX_TRACKED_HOSTS);
|
||||||
|
|
||||||
|
// Let every tracked host age past the idle TTL, then touch one new
|
||||||
|
// host: the over-cap sweep should evict all the idle ones, leaving
|
||||||
|
// only the freshly-inserted entry.
|
||||||
|
tokio::time::sleep(IDLE_EVICT_AFTER + Duration::from_secs(1)).await;
|
||||||
|
rl.wait_for("https://newcomer.example/y").await.unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
rl.tracked_hosts().await,
|
||||||
|
1,
|
||||||
|
"idle hosts should be swept once the map is over the cap"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test(start_paused = true)]
|
||||||
|
async fn host_rate_limiters_keep_recently_used_hosts() {
|
||||||
|
// A host touched within the TTL must survive a sweep so active crawls
|
||||||
|
// aren't reset. Fill to the cap, then re-touch one host right before
|
||||||
|
// adding a newcomer that triggers the sweep.
|
||||||
|
let rl = HostRateLimiters::new(Duration::from_millis(1));
|
||||||
|
for i in 0..MAX_TRACKED_HOSTS {
|
||||||
|
rl.wait_for(&format!("https://host{i}.example/x")).await.unwrap();
|
||||||
|
}
|
||||||
|
tokio::time::sleep(IDLE_EVICT_AFTER + Duration::from_secs(1)).await;
|
||||||
|
// Re-touch host0 so it's recent again.
|
||||||
|
rl.wait_for("https://host0.example/x").await.unwrap();
|
||||||
|
// Newcomer triggers the sweep (map is at cap+1 conceptually).
|
||||||
|
rl.wait_for("https://newcomer.example/y").await.unwrap();
|
||||||
|
// host0 (recent) + newcomer survive; the rest aged out.
|
||||||
|
assert_eq!(rl.tracked_hosts().await, 2);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test(start_paused = true)]
|
#[tokio::test(start_paused = true)]
|
||||||
async fn host_rate_limiters_honor_overrides() {
|
async fn host_rate_limiters_honor_overrides() {
|
||||||
let rl = HostRateLimiters::new(Duration::from_millis(1000))
|
let rl = HostRateLimiters::new(Duration::from_millis(1000))
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "mangalord-frontend",
|
"name": "mangalord-frontend",
|
||||||
"version": "0.90.7",
|
"version": "0.90.8",
|
||||||
"private": true,
|
"private": true,
|
||||||
"type": "module",
|
"type": "module",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
|
|||||||
Reference in New Issue
Block a user