From 660184a048582fc4b89119f3edfb8e3e9affd596 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Mon, 22 Jun 2026 21:18:18 +0200 Subject: [PATCH] fix(daemons): race in-flight work against cancellation on shutdown (0.87.4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two high findings sharing one root cause: long-running daemon work that ignored the shutdown signal until it finished naturally. **Cron tick (crawler).** `CronContext::run_tick` ran the metadata-pass body inside `AssertUnwindSafe(...).catch_unwind().await` with no cancellation. A fresh-catalog walk (many minutes) wedged `DaemonHandle::shutdown`, `Supervisors::reload_crawler` (under the supervisor lock), and SIGTERM responsiveness for the whole pass. Wrap the body in `tokio::select! { biased; _ = self.cancel.cancelled() => ...; r = body => ... }`. On cancel the body future drops, which cascades to dropping `metadata.run()` — exactly what gives Chromium/lease teardown a chance via the BrowserManager idle reaper. The advisory unlock + conn drop still run unconditionally. **Analysis dispatch.** `process_lease` wrapped the vision call only in `tokio::time::timeout(self.job_timeout, …)` — default 600s. Toggling analysis off in the dashboard (under the supervisor lock) or stopping the container blocked on whichever call was in flight. Same fix: `tokio::select!` against `self.cancel.cancelled()`. On cancel, `jobs::release` returns the lease to the queue without burning an attempt — next tick picks it up. Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/Cargo.lock | 2 +- backend/Cargo.toml | 2 +- backend/src/analysis/daemon.rs | 49 +++++++++++++++++++++- backend/src/crawler/daemon.rs | 51 +++++++++++++++++++++-- backend/tests/analysis_worker.rs | 71 +++++++++++++++++++++++++++++++- backend/tests/crawler_daemon.rs | 65 ++++++++++++++++++++++++++++- frontend/package.json | 2 +- 7 files changed, 232 insertions(+), 10 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 4b39d05..1c5dce5 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1517,7 +1517,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.87.3" +version = "0.87.4" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index caca5e2..9c62fad 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.87.3" +version = "0.87.4" edition = "2021" default-run = "mangalord" diff --git a/backend/src/analysis/daemon.rs b/backend/src/analysis/daemon.rs index 07ebf9c..9cdc0ef 100644 --- a/backend/src/analysis/daemon.rs +++ b/backend/src/analysis/daemon.rs @@ -255,9 +255,30 @@ impl WorkerContext { let started = std::time::Instant::now(); let dispatch = AssertUnwindSafe(self.dispatcher.dispatch(page_id)).catch_unwind(); - let outcome = tokio::time::timeout(self.job_timeout, dispatch).await; + // Race the vision call against shutdown. Without this, shutdown + // (SIGTERM / `Supervisors::reload_analysis` toggling analysis off) + // blocks for up to `job_timeout` (default 600s) on whichever call + // happens to be in flight. On cancel we drop the dispatch future + // and `jobs::release` the lease — next tick picks it up without + // burning an attempt. + let cancel_result = tokio::select! { + biased; + _ = self.cancel.cancelled() => None, + o = tokio::time::timeout(self.job_timeout, dispatch) => Some(o), + }; let elapsed_ms = started.elapsed().as_millis() as i64; heartbeat.abort(); + let Some(outcome) = cancel_result else { + tracing::info!( + worker = self.id, + lease_id = %lease.id, + "analysis worker: cancelled mid-dispatch — releasing lease" + ); + let _ = jobs::release(&self.pool, lease.id).await; + // Don't publish Failed (the page didn't actually fail) and + // don't write a duration row — neither outcome is true. + return; + }; // Flatten timeout / panic / dispatch error into one Result + message. let result: Result<(), String> = match outcome { @@ -413,6 +434,32 @@ pub mod test_support { } } + /// A dispatcher that sleeps for `delay` before returning `Ok`. Lets a + /// test fire shutdown WHILE a dispatch is in flight, to exercise the + /// cancellation race in `process_lease`. + pub struct SlowDispatcher { + pub calls: AtomicUsize, + pub delay: Duration, + } + + impl SlowDispatcher { + pub fn new(delay: Duration) -> Arc { + Arc::new(Self { calls: AtomicUsize::new(0), delay }) + } + pub fn call_count(&self) -> usize { + self.calls.load(Ordering::Acquire) + } + } + + #[async_trait] + impl AnalyzeDispatcher for SlowDispatcher { + async fn dispatch(&self, _page_id: Uuid) -> anyhow::Result<()> { + self.calls.fetch_add(1, Ordering::AcqRel); + tokio::time::sleep(self.delay).await; + Ok(()) + } + } + #[async_trait] impl AnalyzeDispatcher for CountingDispatcher { async fn dispatch(&self, _page_id: Uuid) -> anyhow::Result<()> { diff --git a/backend/src/crawler/daemon.rs b/backend/src/crawler/daemon.rs index 470ddc8..f756cee 100644 --- a/backend/src/crawler/daemon.rs +++ b/backend/src/crawler/daemon.rs @@ -280,7 +280,8 @@ impl CronContext { // no future tick would ever run — workers would keep going but // no new metadata work would be scheduled until daemon restart. // The advisory unlock below runs unconditionally so a panicked - // tick doesn't leave the lock held for another replica. + // (or cancelled) tick doesn't leave the lock held for another + // replica. let metadata = &self.metadata; let pool = &self.pool; let retention_days = self.retention_days; @@ -312,8 +313,23 @@ impl CronContext { tracing::warn!(?e, "cron: persist last_metadata_tick_at failed"); } }; - if let Err(_panic) = AssertUnwindSafe(body).catch_unwind().await { - tracing::error!("cron: tick body panicked — continuing"); + // Race the tick body against shutdown. Without this, a long + // metadata pass on a fresh catalog (minutes) wedges + // `DaemonHandle::shutdown`, `Supervisors::reload_crawler` (under + // the supervisor lock), and SIGTERM responsiveness for the + // entire pass. On cancel, dropping `body` cascades to dropping + // `metadata.run()`, which is what gives Chromium/lease cleanup + // their chance via the Browse Manager's idle reaper. + tokio::select! { + biased; + _ = self.cancel.cancelled() => { + tracing::info!("cron: cancelled mid-tick — abandoning to release advisory lock"); + } + r = AssertUnwindSafe(body).catch_unwind() => { + if r.is_err() { + tracing::error!("cron: tick body panicked — continuing"); + } + } } let _ = sqlx::query("SELECT pg_advisory_unlock($1)") @@ -662,6 +678,35 @@ pub mod test_support { } } + /// `MetadataPass` that sleeps for `delay` before returning. Lets a + /// test trigger shutdown WHILE a tick body is in flight, exercising + /// the cancellation race in `CronContext::run_tick`. + pub struct SlowMetadataPass { + pub delay: std::time::Duration, + pub started: AtomicUsize, + } + + impl SlowMetadataPass { + pub fn new(delay: std::time::Duration) -> Arc { + Arc::new(Self { + delay, + started: AtomicUsize::new(0), + }) + } + pub fn start_count(&self) -> usize { + self.started.load(Ordering::Acquire) + } + } + + #[async_trait] + impl MetadataPass for SlowMetadataPass { + async fn run(&self) -> anyhow::Result { + self.started.fetch_add(1, Ordering::AcqRel); + tokio::time::sleep(self.delay).await; + Ok(pipeline::MetadataStats::default()) + } + } + pub type DispatchFn = Arc< dyn Fn(JobPayload) -> futures_util::future::BoxFuture<'static, anyhow::Result> + Send diff --git a/backend/tests/analysis_worker.rs b/backend/tests/analysis_worker.rs index cfe9af1..0d13c40 100644 --- a/backend/tests/analysis_worker.rs +++ b/backend/tests/analysis_worker.rs @@ -5,7 +5,11 @@ use std::sync::Arc; use std::time::Duration; -use mangalord::analysis::daemon::{self, test_support::CountingDispatcher, AnalysisDaemonConfig}; +use mangalord::analysis::daemon::{ + self, + test_support::{CountingDispatcher, SlowDispatcher}, + AnalysisDaemonConfig, +}; use mangalord::domain::page_analysis::{ AnalysisStatus, SafetyFlag, VisionAnalysis, }; @@ -346,3 +350,68 @@ async fn worker_ignores_non_analyze_jobs(pool: PgPool) { .unwrap(); assert_eq!(crawl_state, "pending", "crawl job must be left for the crawler"); } + +/// Shutdown must race in-flight dispatches against cancellation, not wait +/// for them to finish. The previous behaviour blocked `handle.shutdown()` +/// for up to `job_timeout` (default 600s) on one in-flight vision call, +/// which also stranded SIGTERM and any `Supervisors::reload_analysis` +/// caller under the supervisor lock. +#[sqlx::test(migrations = "./migrations")] +async fn worker_shutdown_does_not_block_on_in_flight_dispatch(pool: PgPool) { + let page_id = seed_page(&pool).await; + page_analysis::enqueue_for_page(&pool, page_id, false) + .await + .unwrap(); + + // 30s dispatch + 30s job_timeout. Shutdown must observe the cancel + // signal and return well before either deadline; if it waits for + // dispatch to finish the test would take 30s. + let dispatcher = SlowDispatcher::new(Duration::from_secs(30)); + let cancel = CancellationToken::new(); + let handle = daemon::spawn( + pool.clone(), + cancel.clone(), + AnalysisDaemonConfig { + dispatcher: dispatcher.clone(), + workers: 1, + job_timeout: Duration::from_secs(30), + events: std::sync::Arc::new( + mangalord::analysis::events::AnalysisEvents::new(), + ), + readiness: None, + }, + ); + + // Wait for the worker to lease the job and enter the dispatch. + wait_for_state(&pool, page_id, "running").await; + assert_eq!( + dispatcher.call_count(), + 1, + "dispatcher should be in flight when we cancel" + ); + + // Now cancel. Shutdown must return within a few seconds — comfortably + // under both the slow-dispatch and the job_timeout deadlines. + let started = std::time::Instant::now(); + tokio::time::timeout(Duration::from_secs(5), handle.shutdown()) + .await + .expect("shutdown must observe cancel and return promptly"); + assert!( + started.elapsed() < Duration::from_secs(5), + "shutdown took too long; cancel didn't propagate into dispatch" + ); + + // The lease was released, not failed — attempts must not have grown. + let attempts: i32 = sqlx::query_scalar( + "SELECT attempts FROM crawler_jobs \ + WHERE payload->>'kind' = 'analyze_page' AND payload->>'page_id' = $1", + ) + .bind(page_id.to_string()) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + attempts, 0, + "cancellation must not burn a job attempt (release, not ack_failed)" + ); +} diff --git a/backend/tests/crawler_daemon.rs b/backend/tests/crawler_daemon.rs index e95ba31..e5e4c31 100644 --- a/backend/tests/crawler_daemon.rs +++ b/backend/tests/crawler_daemon.rs @@ -11,8 +11,9 @@ use chrono::NaiveTime; use chrono_tz::Tz; use mangalord::crawler::content::SyncOutcome; use mangalord::crawler::daemon::{ - self, test_support::CountingMetadataPass, ChapterDispatcher, DaemonConfig, MetadataPass, - CRON_LOCK_KEY, + self, + test_support::{CountingMetadataPass, SlowMetadataPass}, + ChapterDispatcher, DaemonConfig, MetadataPass, CRON_LOCK_KEY, }; use mangalord::crawler::jobs::{self, JobPayload}; use mangalord::crawler::pipeline; @@ -834,3 +835,63 @@ async fn enqueue_pending_for_manga_queues_chapters_in_ascending_number_order(poo assert_eq!(leased_chapter_ids, vec![c1, c2, c3]); } + +/// Shutdown must race the in-flight metadata pass against cancellation, +/// not wait for the pass to drain. Without this, a long catalog walk +/// (minutes) wedges `DaemonHandle::shutdown`, `Supervisors::reload_crawler` +/// (under the supervisor lock), and SIGTERM responsiveness. +#[sqlx::test(migrations = "./migrations")] +async fn cron_shutdown_does_not_block_on_in_flight_metadata_pass(pool: PgPool) { + // Pre-seed a stale tick so the daemon does its catch-up tick at spawn + // — that immediately drops us into `metadata.run().await`, which our + // SlowMetadataPass will then sit on for 30s. + sqlx::query( + "INSERT INTO crawler_state (key, value) VALUES ($1, $2) + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value", + ) + .bind("last_metadata_tick_at") + .bind(json!({"at": "2020-01-01T00:00:00Z"})) + .execute(&pool) + .await + .unwrap(); + + let slow = SlowMetadataPass::new(Duration::from_secs(30)); + let dispatcher = Arc::new(AlwaysDoneDispatcher { + seen: AtomicUsize::new(0), + }); + let session_expired = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let cancel = CancellationToken::new(); + let handle = daemon::spawn( + pool.clone(), + cancel.clone(), + make_cfg( + Some(slow.clone() as Arc), + dispatcher, + session_expired, + 1, + ), + ); + + // Wait for the catch-up to enter `metadata.run` so we're sure to be + // observing cancellation mid-pass. + let deadline = std::time::Instant::now() + Duration::from_secs(3); + while slow.start_count() == 0 && std::time::Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert_eq!( + slow.start_count(), + 1, + "metadata pass should be in flight when we cancel" + ); + + // Cancellation must drop the pass future and return promptly — under + // the 30s slow-pass deadline by a wide margin. + let started = std::time::Instant::now(); + tokio::time::timeout(Duration::from_secs(5), handle.shutdown()) + .await + .expect("shutdown must observe cancel and return promptly"); + assert!( + started.elapsed() < Duration::from_secs(5), + "shutdown took too long; cancel didn't propagate into the cron tick body" + ); +} diff --git a/frontend/package.json b/frontend/package.json index f06abe1..6f6893b 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.87.3", + "version": "0.87.4", "private": true, "type": "module", "scripts": {