From f5cb460aec23b68023f15555917a1ca52380ed5a Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sat, 11 Jul 2026 14:01:04 +0200 Subject: [PATCH] fix: abandon dispatch when heartbeat loses the lease MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Heartbeat renew failures were only logged, so under DB flakiness a job could be leased twice concurrently — the original worker kept crawling past a lapsed lease while another re-leased it. Track consecutive renew failures and, after MAX_HEARTBEAT_RENEW_FAILURES, signal the worker (via a CancellationToken) to abandon the in-flight dispatch instead of working a job it no longer owns. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/Cargo.lock | 2 +- backend/Cargo.toml | 2 +- backend/src/crawler/daemon.rs | 57 +++++++++++++++++++++++++++++++++-- frontend/package.json | 2 +- 4 files changed, 58 insertions(+), 5 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 00174bc..db05a69 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1558,7 +1558,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.124.18" +version = "0.124.19" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 1ebe6af..e637ccd 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.124.18" +version = "0.124.19" edition = "2021" default-run = "mangalord" diff --git a/backend/src/crawler/daemon.rs b/backend/src/crawler/daemon.rs index 487349c..6b77a4e 100644 --- a/backend/src/crawler/daemon.rs +++ b/backend/src/crawler/daemon.rs @@ -66,6 +66,19 @@ const LEASE_DURATION: Duration = Duration::from_secs(60); /// the lease window leaves two missed-beat's slack before expiry. const LEASE_HEARTBEAT: Duration = Duration::from_secs(20); +/// Consecutive failed lease renews the heartbeat tolerates before it gives up +/// and signals the worker to abandon the in-flight dispatch. At +/// [`LEASE_HEARTBEAT`] spacing this is ~1 lease window of DB flakiness — past +/// that the lease has very likely lapsed and another worker may re-lease the +/// job, so continuing to crawl it is wasted (and duplicated) work. +const MAX_HEARTBEAT_RENEW_FAILURES: u32 = 3; + +/// Whether the heartbeat should abandon the job after `consecutive_failures` +/// failed renews. Split out so the escalation threshold is unit-testable. +fn should_abort_after_renew_failures(consecutive_failures: u32) -> bool { + consecutive_failures >= MAX_HEARTBEAT_RENEW_FAILURES +} + /// How long a worker waits after a `BrowserUnavailable` outcome before looping /// back to lease again. The job was released (not failed) so it stays pending; /// this backoff keeps the worker from hot-looping lease→acquire→release while @@ -443,18 +456,34 @@ impl WorkerContext { // dispatch runs, so a slow-but-healthy job is never re-leased and // never inflates `attempts` toward `max_attempts`. Stops itself // once the job is no longer ours (renew returns false). + // Signalled by the heartbeat if it gives up after too many consecutive + // renew failures, so the worker can abandon a dispatch whose lease has + // very likely lapsed (rather than crawl a job another worker may now own). + let hb_lost = CancellationToken::new(); let heartbeat = { let hb_pool = self.pool.clone(); let hb_id = lease.id; let hb_gen = lease.lease_generation; + let hb_lost = hb_lost.clone(); tokio::spawn(async move { + let mut failures: u32 = 0; loop { tokio::time::sleep(LEASE_HEARTBEAT).await; match jobs::renew(&hb_pool, hb_id, hb_gen, LEASE_DURATION).await { - Ok(true) => {} + Ok(true) => failures = 0, Ok(false) => break, Err(e) => { - tracing::warn!(lease_id = %hb_id, ?e, "heartbeat renew failed"); + failures += 1; + tracing::warn!(lease_id = %hb_id, failures, ?e, "heartbeat renew failed"); + if should_abort_after_renew_failures(failures) { + tracing::error!( + lease_id = %hb_id, + failures, + "heartbeat lost the lease after repeated renew failures — signalling abandon" + ); + hb_lost.cancel(); + break; + } } } } @@ -492,6 +521,20 @@ impl WorkerContext { ); return; } + _ = hb_lost.cancelled() => { + // The heartbeat gave up renewing: the lease has very likely + // expired and may already be re-leased elsewhere. Abandon the + // dispatch rather than keep crawling a job we no longer own. Do + // NOT ack/release — a generation-guarded write would no-op, and + // another worker may now hold this lease. + heartbeat.abort(); + tracing::error!( + worker = self.id, + lease_id = %lease.id, + "worker: abandoning dispatch — lease lost (heartbeat renew failures)" + ); + return; + } o = tokio::time::timeout(self.job_timeout, dispatch) => o, }; heartbeat.abort(); @@ -798,6 +841,16 @@ mod tests { Utc.with_ymd_and_hms(y, mo, d, h, mi, 0).unwrap() } + #[test] + fn heartbeat_aborts_only_after_threshold_consecutive_failures() { + // A blip or two is tolerated; sustained failures escalate to abandon. + assert!(!should_abort_after_renew_failures(0)); + assert!(!should_abort_after_renew_failures(1)); + assert!(!should_abort_after_renew_failures(MAX_HEARTBEAT_RENEW_FAILURES - 1)); + assert!(should_abort_after_renew_failures(MAX_HEARTBEAT_RENEW_FAILURES)); + assert!(should_abort_after_renew_failures(MAX_HEARTBEAT_RENEW_FAILURES + 1)); + } + #[test] fn idle_backoff_grows_then_caps() { // First miss is a short 1s poll; the interval doubles each empty poll… diff --git a/frontend/package.json b/frontend/package.json index 3ae0676..da98562 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.124.18", + "version": "0.124.19", "private": true, "type": "module", "scripts": {