fix: abandon dispatch when heartbeat loses the lease

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) <noreply@anthropic.com>

backend/Cargo.lock

@@ -1558,7 +1558,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
 
 [[package]]
 name = "mangalord"
-version = "0.124.18"
+version = "0.124.19"
 dependencies = [
  "anyhow",
  "argon2",

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mangalord"
-version = "0.124.18"
+version = "0.124.19"
 edition = "2021"
 default-run = "mangalord"
 

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…

frontend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "mangalord-frontend",
-  "version": "0.124.18",
+  "version": "0.124.19",
   "private": true,
   "type": "module",
   "scripts": {