From ce6d96c5e1aacce109d7b01c02ae6f4002a0ffff Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sun, 14 Jun 2026 16:59:46 +0200 Subject: [PATCH] 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) --- backend/Cargo.lock | 2 +- backend/Cargo.toml | 2 +- backend/src/app.rs | 11 ++++ backend/src/crawler/daemon.rs | 25 +++++++- backend/src/crawler/jobs.rs | 34 ++++++++++ backend/src/crawler/source/target.rs | 94 ++++++++++++++++++++++------ backend/tests/crawler_daemon.rs | 54 ++++++++++++++++ backend/tests/crawler_jobs.rs | 73 +++++++++++++++++++++ frontend/package.json | 2 +- 9 files changed, 275 insertions(+), 22 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index a68170e..957dbcf 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1517,7 +1517,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.81.0" +version = "0.81.1" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index b827ef0..dcf03bd 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.81.0" +version = "0.81.1" edition = "2021" default-run = "mangalord" diff --git a/backend/src/app.rs b/backend/src/app.rs index 35beccf..a8823ef 100644 --- a/backend/src/app.rs +++ b/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, diff --git a/backend/src/crawler/daemon.rs b/backend/src/crawler/daemon.rs index 2c808d9..b67ff4d 100644 --- a/backend/src/crawler/daemon.rs +++ b/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 { diff --git a/backend/src/crawler/jobs.rs b/backend/src/crawler/jobs.rs index f324c52..a99ff6e 100644 --- a/backend/src/crawler/jobs.rs +++ b/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 { + 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. diff --git a/backend/src/crawler/source/target.rs b/backend/src/crawler/source/target.rs index eb2e888..2f1dd06 100644 --- a/backend/src/crawler/source/target.rs +++ b/backend/src/crawler/source/target.rs @@ -332,25 +332,54 @@ fn parse_manga_list_from(doc: &scraper::Html) -> Result, 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, 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::()); - 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::()); + 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#" +
+ + "#; + 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 diff --git a/backend/tests/crawler_daemon.rs b/backend/tests/crawler_daemon.rs index f2d7a2e..d96e98c 100644 --- a/backend/tests/crawler_daemon.rs +++ b/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; diff --git a/backend/tests/crawler_jobs.rs b/backend/tests/crawler_jobs.rs index 14a33de..0b7c4b9 100644 --- a/backend/tests/crawler_jobs.rs +++ b/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> = + 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())) diff --git a/frontend/package.json b/frontend/package.json index 837ecdc..8960c35 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.81.0", + "version": "0.81.1", "private": true, "type": "module", "scripts": {