fix: exponential idle backoff for the analysis worker
The analysis worker's flat 1s idle sleep meant an empty queue fired a lease query per worker per second forever (audit). Add idle_backoff (1s..30s, reset on lease), mirroring the crawler. Unit-tested. 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.128.20"
|
version = "0.128.21"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"argon2",
|
"argon2",
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mangalord"
|
name = "mangalord"
|
||||||
version = "0.128.20"
|
version = "0.128.21"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
default-run = "mangalord"
|
default-run = "mangalord"
|
||||||
|
|
||||||
|
|||||||
@@ -37,6 +37,22 @@ const LEASE_HEARTBEAT: Duration = Duration::from_secs(20);
|
|||||||
/// long enough not to hammer `/health` while vision is down.
|
/// long enough not to hammer `/health` while vision is down.
|
||||||
const READINESS_POLL: Duration = Duration::from_secs(2);
|
const READINESS_POLL: Duration = Duration::from_secs(2);
|
||||||
|
|
||||||
|
/// Longest an idle analysis worker waits between lease polls, bounding the
|
||||||
|
/// exponential [`idle_backoff`].
|
||||||
|
const IDLE_BACKOFF_CAP: Duration = Duration::from_secs(30);
|
||||||
|
|
||||||
|
/// Exponential idle backoff (1s, 2s, 4s … capped at [`IDLE_BACKOFF_CAP`]).
|
||||||
|
/// `consecutive_empty` is the number of empty lease polls so far (0 on the first
|
||||||
|
/// miss); reset to 0 the moment a job is leased. Replaces the old flat 1s sleep
|
||||||
|
/// so an idle worker isn't firing a `SELECT … FOR UPDATE SKIP LOCKED` lease
|
||||||
|
/// query every second (audit: analysis daemon idle poll). Mirrors the crawler
|
||||||
|
/// daemon's backoff.
|
||||||
|
fn idle_backoff(consecutive_empty: u32) -> Duration {
|
||||||
|
let cap = IDLE_BACKOFF_CAP.as_secs();
|
||||||
|
let secs = 1u64.checked_shl(consecutive_empty).unwrap_or(cap).min(cap);
|
||||||
|
Duration::from_secs(secs)
|
||||||
|
}
|
||||||
|
|
||||||
/// The unit of work: analyze one page. Implemented by
|
/// The unit of work: analyze one page. Implemented by
|
||||||
/// [`RealAnalyzeDispatcher`] in production and stubbed in tests.
|
/// [`RealAnalyzeDispatcher`] in production and stubbed in tests.
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
@@ -135,6 +151,9 @@ impl WorkerContext {
|
|||||||
// Last observed readiness, so we log only on transitions (not every
|
// Last observed readiness, so we log only on transitions (not every
|
||||||
// poll). `None` until the first probe.
|
// poll). `None` until the first probe.
|
||||||
let mut was_ready: Option<bool> = None;
|
let mut was_ready: Option<bool> = None;
|
||||||
|
// Empty lease polls seen in a row, driving the idle backoff. Reset to 0
|
||||||
|
// the moment a job is leased.
|
||||||
|
let mut consecutive_empty: u32 = 0;
|
||||||
loop {
|
loop {
|
||||||
if self.cancel.is_cancelled() {
|
if self.cancel.is_cancelled() {
|
||||||
tracing::info!(worker = self.id, "analysis worker: shutdown");
|
tracing::info!(worker = self.id, "analysis worker: shutdown");
|
||||||
@@ -178,11 +197,15 @@ impl WorkerContext {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
let Some(lease) = leases.into_iter().next() else {
|
let Some(lease) = leases.into_iter().next() else {
|
||||||
if self.sleep_or_cancel(Duration::from_secs(1)).await {
|
let wait = idle_backoff(consecutive_empty);
|
||||||
|
consecutive_empty = consecutive_empty.saturating_add(1);
|
||||||
|
if self.sleep_or_cancel(wait).await {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
|
// Leased work — drop back to a tight poll cadence.
|
||||||
|
consecutive_empty = 0;
|
||||||
self.process_lease(lease).await;
|
self.process_lease(lease).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -510,6 +533,17 @@ mod tests {
|
|||||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||||
use tokio::net::TcpListener;
|
use tokio::net::TcpListener;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn idle_backoff_grows_exponentially_and_caps() {
|
||||||
|
assert_eq!(idle_backoff(0), Duration::from_secs(1));
|
||||||
|
assert_eq!(idle_backoff(1), Duration::from_secs(2));
|
||||||
|
assert_eq!(idle_backoff(2), Duration::from_secs(4));
|
||||||
|
assert_eq!(idle_backoff(4), Duration::from_secs(16));
|
||||||
|
// Caps at IDLE_BACKOFF_CAP (30s) and never overflows on large counts.
|
||||||
|
assert_eq!(idle_backoff(5), IDLE_BACKOFF_CAP);
|
||||||
|
assert_eq!(idle_backoff(100), IDLE_BACKOFF_CAP);
|
||||||
|
}
|
||||||
|
|
||||||
/// Bind an ephemeral port that answers every request with `status_line`
|
/// Bind an ephemeral port that answers every request with `status_line`
|
||||||
/// (e.g. `"200 OK"`), and return its `/health` URL. The probe under test
|
/// (e.g. `"200 OK"`), and return its `/health` URL. The probe under test
|
||||||
/// only inspects the status code, so a zero-length body is enough.
|
/// only inspects the status code, so a zero-length body is enough.
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "mangalord-frontend",
|
"name": "mangalord-frontend",
|
||||||
"version": "0.128.20",
|
"version": "0.128.21",
|
||||||
"private": true,
|
"private": true,
|
||||||
"type": "module",
|
"type": "module",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
|
|||||||
Reference in New Issue
Block a user