fix(crawler): harden crash & shutdown job recovery

Three failure-mode fixes surfaced by a crawler recovery audit:

- Graceful shutdown mid-dispatch now releases the in-flight job back to
  pending without burning a retry attempt. process_lease wraps the
  dispatch in a biased tokio::select! cancel arm that aborts the
  heartbeat and calls jobs::release; previously a mid-job SIGTERM left
  the row 'running' until lease expiry and cost one of max_attempts.

- Boot-time jobs::reclaim_orphaned resets 'running' jobs with an expired
  lease back to pending (attempt refunded), run once at daemon startup
  before workers start. Crash recovery is now immediate instead of
  waiting a full lease window for the lazy lease-expiry path. Safe under
  multi-replica: only already-expired leases are touched, which a
  healthy heartbeating peer never has.

- Manga-list parsing now warns when listing anchors are dropped for a
  missing/empty href or title (split into parse_manga_list_anchors so
  the drop count is testable), turning silent source markup drift into
  an observable signal. Returned refs are byte-identical to before.

Tests added: shutdown_mid_dispatch_releases_lease_without_burning_attempt,
reclaim_orphaned_resets_only_expired_running_jobs, and a drop-count unit
test. Patch bump 0.81.0 -> 0.81.1 (both manifests).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/Cargo.lock

@@ -1517,7 +1517,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
 
 [[package]]
 name = "mangalord"
-version = "0.81.0"
+version = "0.81.1"
 dependencies = [
  "anyhow",
  "argon2",

backend/Cargo.toml

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

backend/src/app.rs

@@ -645,6 +645,17 @@ async fn spawn_crawler_daemon(
         })
     };
 
+    // Reclaim jobs orphaned by a previous crash/kill (running with an
+    // expired lease) before workers start, so recovery is immediate instead
+    // of waiting a full lease window for `lease`'s expiry clause. Safe under
+    // multi-replica: only already-expired leases are touched. Best-effort —
+    // a failure here just defers recovery to the lazy lease path.
+    match crate::crawler::jobs::reclaim_orphaned(&db).await {
+        Ok(0) => {}
+        Ok(n) => tracing::info!(reclaimed = n, "crawler: reclaimed orphaned in-flight jobs at startup"),
+        Err(e) => tracing::warn!(?e, "crawler: reclaim_orphaned at startup failed"),
+    }
+
     let daemon_handle = daemon::spawn(
         db,
         cancel,

backend/src/crawler/daemon.rs

@@ -405,7 +405,30 @@ impl WorkerContext {
         // failed (exponential backoff) rather than wedging the worker.
         let dispatch = AssertUnwindSafe(self.dispatcher.dispatch(lease.payload.clone()))
             .catch_unwind();
-        let outcome = tokio::time::timeout(self.job_timeout, dispatch).await;
+        let outcome = tokio::select! {
+            // `biased` so a shutdown that arrives while the dispatch is also
+            // ready still takes the cancel arm and releases the lease.
+            biased;
+            _ = self.cancel.cancelled() => {
+                // Graceful shutdown mid-dispatch: drop the in-flight dispatch
+                // future (cancelling its browser work) and return the job to
+                // `pending` WITHOUT burning a retry attempt, so a clean
+                // restart doesn't march healthy jobs toward `max_attempts`.
+                // `release` refunds the lease's `attempts` increment, mirroring
+                // the session-expired path. Previously the worker had no cancel
+                // branch here, so a mid-dispatch SIGTERM left the row `running`
+                // until lease expiry and cost one attempt.
+                heartbeat.abort();
+                let _ = jobs::release(&self.pool, lease.id).await;
+                tracing::info!(
+                    worker = self.id,
+                    lease_id = %lease.id,
+                    "worker: shutdown mid-dispatch — released lease without burning an attempt"
+                );
+                return;
+            }
+            o = tokio::time::timeout(self.job_timeout, dispatch) => o,
+        };
         heartbeat.abort();
 
         let outcome = match outcome {

backend/src/crawler/jobs.rs

@@ -314,6 +314,40 @@ pub async fn release(pool: &PgPool, lease_id: Uuid) -> sqlx::Result<()> {
     Ok(())
 }
 
+/// Reclaim jobs orphaned by a crashed/killed worker: those still `running`
+/// whose `leased_until` has already lapsed. Each is returned to `pending`
+/// with its lease cleared and the lease's `attempts` increment refunded
+/// (`GREATEST(0, attempts - 1)`, mirroring [`release`]) — a crash that
+/// interrupted an attempt mid-flight shouldn't count against `max_attempts`.
+/// Returns the number reclaimed.
+///
+/// Intended to run once at daemon startup so crash recovery is immediate
+/// rather than waiting up to a full lease window for `lease`'s expiry clause
+/// to re-pick the row. It is safe under multi-replica deployment precisely
+/// because it only touches **already-expired** leases: a healthy peer
+/// heartbeats (`renew`) every ~20s, keeping its in-flight jobs' `leased_until`
+/// in the future, so this never steals live work — it only does eagerly what
+/// `lease` would do lazily, minus the attempt burn.
+///
+/// Trade-off: a job whose dispatch reliably hard-kills the process (e.g. OOM
+/// on a pathological payload) is refunded each boot and could loop without
+/// dead-lettering. That window is bounded in practice by the per-image size
+/// cap and the worker's `job_timeout` (a hang is acked-failed normally, which
+/// *does* burn an attempt); only a true process-killer evades it, which is an
+/// infrastructure signal worth surfacing rather than silently dead-lettering
+/// everyone's chapters during a crash loop.
+pub async fn reclaim_orphaned(pool: &PgPool) -> sqlx::Result<u64> {
+    let result = sqlx::query(
+        "UPDATE crawler_jobs \
+            SET state = 'pending', leased_until = NULL, \
+                attempts = GREATEST(0, attempts - 1), updated_at = now() \
+          WHERE state = 'running' AND leased_until < now()",
+    )
+    .execute(pool)
+    .await?;
+    Ok(result.rows_affected())
+}
+
 /// Delete `done` jobs whose `updated_at` is older than `retention_days`
 /// days. `0` disables the reaper without touching the table. Returns the
 /// number of rows removed.

backend/src/crawler/source/target.rs

@@ -332,25 +332,54 @@ fn parse_manga_list_from(doc: &scraper::Html) -> Result<Vec<SourceMangaRef>, Pag
     if !has_logo_sentinel(doc) {
         return Err(PageError::transient("manga list: #logo sentinel missing"));
     }
+    let (refs, dropped) = parse_manga_list_anchors(doc);
+    // A non-zero drop count means the listing selector matched anchors but
+    // some lacked a usable href/title. In steady state this is ~always zero;
+    // a sustained non-zero count is the early signal of a source markup
+    // drift silently eroding coverage (GAP-2), so surface it rather than
+    // swallowing it in the filter. `#logo` was already confirmed present,
+    // so this is genuine per-item loss, not a broken-page response.
+    if dropped > 0 {
+        tracing::warn!(
+            dropped,
+            kept = refs.len(),
+            "manga list: dropped listing anchors with missing/empty href or title \
+             (possible source markup drift)"
+        );
+    }
+    Ok(refs)
+}
+
+/// Extract `SourceMangaRef`s from a listing document, returning the kept
+/// refs alongside the count of anchors dropped for a missing/empty href or
+/// title. Splitting the count out of [`parse_manga_list_from`] keeps the
+/// drop accounting unit-testable and lets the caller log coverage erosion.
+fn parse_manga_list_anchors(doc: &scraper::Html) -> (Vec<SourceMangaRef>, usize) {
     let sel = scraper::Selector::parse("#left_side .pic_list .updatesli span a").unwrap();
-    Ok(doc
-        .select(&sel)
-        .filter_map(|a| {
-            let url = a.value().attr("href")?.trim().to_string();
-            if url.is_empty() {
-                return None;
-            }
-            let title = collapse_whitespace(&a.text().collect::<String>());
-            if title.is_empty() {
-                return None;
-            }
-            Some(SourceMangaRef {
-                source_manga_key: derive_key_from_url(&url),
-                title,
-                url,
-            })
-        })
-        .collect())
+    let mut refs = Vec::new();
+    let mut dropped = 0usize;
+    for a in doc.select(&sel) {
+        let url = a
+            .value()
+            .attr("href")
+            .map(|h| h.trim().to_string())
+            .unwrap_or_default();
+        if url.is_empty() {
+            dropped += 1;
+            continue;
+        }
+        let title = collapse_whitespace(&a.text().collect::<String>());
+        if title.is_empty() {
+            dropped += 1;
+            continue;
+        }
+        refs.push(SourceMangaRef {
+            source_manga_key: derive_key_from_url(&url),
+            title,
+            url,
+        });
+    }
+    (refs, dropped)
 }
 
 fn parse_manga_detail(
@@ -699,6 +728,35 @@ mod tests {
         assert_eq!(refs[1].source_manga_key, "bar-baz");
     }
 
+    #[test]
+    fn parse_manga_list_anchors_reports_dropped_count() {
+        // The fixture's third `.updatesli` anchor has an empty href. The
+        // kept list must exclude it AND the drop count must be 1 so the
+        // caller can warn on markup drift (GAP-2).
+        let doc = scraper::Html::parse_document(LISTING_HTML);
+        let (refs, dropped) = parse_manga_list_anchors(&doc);
+        assert_eq!(refs.len(), 2, "two well-formed anchors kept");
+        assert_eq!(dropped, 1, "the empty-href anchor is counted as dropped");
+
+        // Also cover the *missing*-href branch (no `href` attr at all),
+        // which is the path the refactor reroutes through
+        // `.map(..).unwrap_or_default()` — distinct from the empty-string
+        // href above. Plus a missing-title anchor.
+        let html = r#"<html><body>
+            <header><div id="logo">Target</div></header>
+            <div id="left_side"><div class="pic_list">
+              <div class="updatesli"><span><a href="/manga/keep">Keep</a></span></div>
+              <div class="updatesli"><span><a>no href at all</a></span></div>
+              <div class="updatesli"><span><a href="/manga/blank">   </a></span></div>
+            </div></div>
+            </body></html>"#;
+        let doc = scraper::Html::parse_document(html);
+        let (refs, dropped) = parse_manga_list_anchors(&doc);
+        assert_eq!(refs.len(), 1, "only the well-formed anchor is kept");
+        assert_eq!(refs[0].title, "Keep");
+        assert_eq!(dropped, 2, "missing-href and missing-title anchors both counted");
+    }
+
     #[test]
     fn parse_manga_list_returns_transient_when_logo_missing() {
         // Broken-page response: no #logo, no listing. Empty Vec would

backend/tests/crawler_daemon.rs

@@ -136,6 +136,60 @@ async fn worker_times_out_a_hung_dispatch_and_acks_failed(pool: PgPool) {
     assert!(dispatcher.seen.load(Ordering::Acquire) >= 1);
 }
 
+#[sqlx::test(migrations = "./migrations")]
+async fn shutdown_mid_dispatch_releases_lease_without_burning_attempt(pool: PgPool) {
+    // A worker that is mid-dispatch when shutdown is signalled must return
+    // its job to `pending` with the attempt refunded (attempts back to 0),
+    // rather than leaving it `running` until lease expiry with a burned
+    // attempt. Uses a never-completing dispatcher so the job is provably
+    // in-flight when we cancel.
+    enqueue_chapter_job(&pool).await;
+    let dispatcher = Arc::new(HangingDispatcher {
+        seen: AtomicUsize::new(0),
+    });
+    let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let cancel = CancellationToken::new();
+    let mut cfg = make_cfg(None, dispatcher.clone(), session_expired, 1);
+    // Long timeout so the dispatch can't be acked-failed by the outer
+    // timeout before we cancel.
+    cfg.job_timeout = Duration::from_secs(60);
+    let handle = daemon::spawn(pool.clone(), cancel.clone(), cfg);
+
+    // Wait until the job is actually leased (running) and the dispatch has
+    // entered the hanging future.
+    let mut entered = false;
+    for _ in 0..40 {
+        if dispatcher.seen.load(Ordering::Acquire) >= 1 && count_state(&pool, "running").await == 1
+        {
+            entered = true;
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    assert!(entered, "dispatch must be in-flight (running) before shutdown");
+
+    handle.shutdown().await;
+
+    assert_eq!(
+        count_state(&pool, "running").await,
+        0,
+        "no job left running after graceful shutdown"
+    );
+    assert_eq!(
+        count_state(&pool, "pending").await,
+        1,
+        "the in-flight job is released back to pending"
+    );
+    let attempts: i32 = sqlx::query_scalar("SELECT attempts FROM crawler_jobs")
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+    assert_eq!(
+        attempts, 0,
+        "graceful shutdown must refund the lease attempt (no burn)"
+    );
+}
+
 #[sqlx::test(migrations = "./migrations")]
 async fn workers_drain_jobs_through_dispatcher(pool: PgPool) {
     enqueue_chapter_job(&pool).await;

backend/tests/crawler_jobs.rs

@@ -341,6 +341,79 @@ async fn stale_running_lease_can_be_reclaimed(pool: PgPool) {
     assert_eq!(second[0].attempts, 2, "attempts bumped again");
 }
 
+#[sqlx::test(migrations = "./migrations")]
+async fn reclaim_orphaned_resets_only_expired_running_jobs(pool: PgPool) {
+    // Three jobs: (a) running with an expired lease — a crash orphan that
+    // must be reclaimed; (b) running with a live lease — a healthy peer's
+    // in-flight job that must be left untouched; (c) pending — irrelevant,
+    // must be untouched. Reclaim must touch only (a), refund its attempt,
+    // and report exactly 1.
+    let orphan = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
+        .await
+        .unwrap()
+    {
+        EnqueueResult::Inserted(id) => id,
+        _ => unreachable!(),
+    };
+    let live = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
+        .await
+        .unwrap()
+    {
+        EnqueueResult::Inserted(id) => id,
+        _ => unreachable!(),
+    };
+    let pending = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
+        .await
+        .unwrap()
+    {
+        EnqueueResult::Inserted(id) => id,
+        _ => unreachable!(),
+    };
+
+    // Lease the two that should become running (attempts -> 1 each).
+    jobs::lease(&pool, None, 10, Duration::from_secs(60))
+        .await
+        .unwrap();
+    // Leave `pending` pending by returning it (release refunds its attempt).
+    jobs::release(&pool, pending).await.unwrap();
+    assert_eq!(job_state(&pool, pending).await, "pending");
+
+    // Orphan: rewind its lease into the past (worker crashed). Live: keep
+    // its lease in the future (peer still heartbeating).
+    sqlx::query("UPDATE crawler_jobs SET leased_until = now() - interval '1 minute' WHERE id = $1")
+        .bind(orphan)
+        .execute(&pool)
+        .await
+        .unwrap();
+    assert_eq!(job_state(&pool, live).await, "running");
+    assert_eq!(job_attempts(&pool, live).await, 1);
+
+    let reclaimed = jobs::reclaim_orphaned(&pool).await.unwrap();
+    assert_eq!(reclaimed, 1, "only the expired-lease running job is reclaimed");
+
+    // Orphan: back to pending, attempt refunded, lease cleared.
+    assert_eq!(job_state(&pool, orphan).await, "pending");
+    assert_eq!(
+        job_attempts(&pool, orphan).await,
+        0,
+        "reclaim refunds the crashed attempt"
+    );
+    let orphan_leased: Option<chrono::DateTime<chrono::Utc>> =
+        sqlx::query_scalar("SELECT leased_until FROM crawler_jobs WHERE id = $1")
+            .bind(orphan)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+    assert!(orphan_leased.is_none(), "reclaim clears the lease");
+
+    // Live peer job: untouched.
+    assert_eq!(job_state(&pool, live).await, "running");
+    assert_eq!(job_attempts(&pool, live).await, 1);
+
+    // Pending job: untouched.
+    assert_eq!(job_state(&pool, pending).await, "pending");
+}
+
 #[sqlx::test(migrations = "./migrations")]
 async fn ack_done_transitions_state_and_clears_lease(pool: PgPool) {
     let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))

frontend/package.json

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