use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::RwLock as StdRwLock; use anyhow::Context; use async_trait::async_trait; use axum::extract::{DefaultBodyLimit, FromRequestParts, Request, State}; use axum::http::{HeaderName, HeaderValue, Method}; use axum::middleware::{self, Next}; use axum::response::Response; use axum::Router; use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; use tokio_util::sync::CancellationToken; use tower_http::cors::{AllowOrigin, CorsLayer}; use tower_http::trace::TraceLayer; use crate::auth::extractor::CurrentUser; use crate::auth::rate_limit::AuthRateLimiter; use crate::error::AppError; use crate::config::{AnalysisConfig, AuthConfig, Config, CrawlerConfig, UploadConfig}; use crate::crawler::browser_manager::{self, BrowserManager}; use crate::crawler::content::{self, SyncOutcome}; use crate::crawler::daemon::{self, ChapterDispatcher, DaemonConfig, MetadataPass}; use crate::crawler::jobs::JobPayload; use crate::crawler::pipeline::{self, MetadataStats}; use crate::crawler::rate_limit::HostRateLimiters; use crate::crawler::resync::{RealResyncService, ResyncService}; use crate::crawler::safety::DownloadAllowlist; use crate::crawler::session; use crate::repo; use crate::storage::{LocalStorage, Storage}; #[derive(Clone)] pub struct AppState { pub db: PgPool, pub storage: Arc, pub auth: AuthConfig, pub upload: UploadConfig, /// Shared rate limiter guarding the `/auth/*` mutation endpoints. /// One instance per AppState so tests stay isolated across the /// same process. pub auth_limiter: Arc, /// Runtime-swappable controls (crawler resync/control handles + the /// analysis enable gate). Shared with the [`Supervisors`] so a config /// reload that respawns a daemon updates what handlers see, live — /// without a restart. In tests this starts empty (no daemons). pub runtime: Arc, /// Applies a persisted settings change to the running daemons /// (graceful stop + respawn). `Some` in production ([`build`]); `None` /// in the test harness, where settings still persist but no daemon is /// spawned. See [`DaemonReloader`]. pub reloader: Option>, /// Env-derived crawler config — the **base** the stored settings DTO is /// overlaid onto (carries env-only fields: browser, proxy, TOR, session). pub crawler_base: CrawlerConfig, /// Env-derived analysis config base (carries the env-only vision API key). pub analysis_base: AnalysisConfig, /// Browser origins permitted to issue mutating requests to /// `/api/v1/admin/*`. See [`crate::config::Config::admin_allowed_origins`] /// for the policy. Cloned per-request into the CSRF middleware; the /// `Arc` keeps the clone cheap. Empty list → check is skipped /// (operator opt-out documented in `.env.example`). pub admin_allowed_origins: Arc>, /// Live analysis-event broadcaster. Always present (the worker and the /// admin enqueue path publish here; the SSE endpoint subscribes). When /// the worker is disabled the channel simply stays quiet. pub analysis_events: Arc, } impl AppState { /// Current crawler control handle, or `None` when the daemon is stopped. pub fn crawler(&self) -> Option> { self.runtime.crawler_control() } /// Current force-resync service, or `None` when the daemon is stopped. pub fn resync(&self) -> Option> { self.runtime.resync() } /// Whether AI page analysis is currently enabled (read live). pub fn analysis_enabled(&self) -> bool { self.runtime.analysis_enabled() } } /// The crawler control handles, swapped atomically when the daemon is /// (re)spawned or stopped. Cloned cheaply on every read. #[derive(Clone, Default)] struct CrawlerControls { resync: Option>, control: Option>, } /// Shared, runtime-mutable surface read by handlers and mutated by the /// [`Supervisors`] on a config reload. The analysis gate is an `Arc` /// so the crawler's chapter dispatcher (which enqueues `analyze_page` jobs) /// sees toggles live without being respawned. pub struct RuntimeControls { crawler: StdRwLock, analysis_enabled: Arc, } impl RuntimeControls { /// Empty controls (no crawler daemon) with the given initial analysis gate. pub fn new(analysis_enabled: bool) -> Self { Self { crawler: StdRwLock::new(CrawlerControls::default()), analysis_enabled: Arc::new(AtomicBool::new(analysis_enabled)), } } pub fn analysis_enabled(&self) -> bool { self.analysis_enabled.load(Ordering::Relaxed) } /// Flip the analysis gate. Used by the [`Supervisors`] on reload and by /// the test harness's stub reloader. pub fn set_analysis_enabled(&self, v: bool) { self.analysis_enabled.store(v, Ordering::Relaxed); } /// The shared gate handle, passed to the crawler chapter dispatcher. fn analysis_gate(&self) -> Arc { Arc::clone(&self.analysis_enabled) } pub fn crawler_control(&self) -> Option> { self.crawler.read().unwrap().control.clone() } pub fn resync(&self) -> Option> { self.crawler.read().unwrap().resync.clone() } /// Test helper: install a resync service without a running daemon. pub fn set_resync(&self, resync: Option>) { self.crawler.write().unwrap().resync = resync; } fn set_crawler( &self, resync: Option>, control: Option>, ) { let mut g = self.crawler.write().unwrap(); g.resync = resync; g.control = control; } } /// Applies a validated settings change to the running daemons by gracefully /// stopping and respawning them with the new effective config (and updating /// the shared [`RuntimeControls`]). Implemented by [`Supervisors`] in /// production; the settings handlers call it after persisting. #[async_trait] pub trait DaemonReloader: Send + Sync { async fn reload_crawler(&self, cfg: CrawlerConfig) -> anyhow::Result<()>; async fn reload_analysis(&self, cfg: AnalysisConfig) -> anyhow::Result<()>; } /// Shared handle the admin crawler endpoints use to observe and control /// the running daemon. Bundled so the handlers take one optional field on /// `AppState` rather than many. pub struct CrawlerControl { pub browser_manager: Arc, pub session: Arc, pub status: crate::crawler::status::StatusHandle, /// Used by the "run metadata pass now" endpoint; `None` when no /// `CRAWLER_START_URL` is configured (cron disabled). pub metadata_pass: Option>, /// Used by the "reconcile missing" endpoint; `None` when no /// `CRAWLER_START_URL` is configured. pub reconcile_pass: Option>, /// Drain budget for a manually-triggered coordinated browser restart. pub drain_deadline: std::time::Duration, /// Held for the duration of a `/admin/crawler/run` pass so a second /// click returns 409 instead of fanning N overlapping metadata passes /// onto the single browser lease. The daemon's daily cron does NOT /// take this lock — cron and operator-triggered are different /// trust modes (cron is single-fire by definition; the lock only /// dedups operator clicks). `Arc` so the spawned task can hold an /// owned guard past the request boundary. pub manual_pass_lock: Arc>, } /// Bundle returned by [`build`]. The router is what `axum::serve` consumes; /// the [`Supervisors`] own the background daemons (when enabled), outlive the /// HTTP server, and are awaited via [`AppHandle::shutdown`] after the listener /// has finished gracefully. pub struct AppHandle { pub router: Router, pub supervisors: Arc, } impl AppHandle { pub async fn shutdown(self) { self.supervisors.shutdown().await; } } /// Owns the crawler + analysis daemon handles and respawns them on a config /// reload. Holds the build-time inputs (db, storage, env-base configs, the /// shared event bus and [`RuntimeControls`]) needed to spawn from scratch. pub struct Supervisors { db: PgPool, storage: Arc, runtime: Arc, analysis_events: Arc, /// Serializes reloads + owns the live crawler daemon handle. crawler_handle: tokio::sync::Mutex>, /// Serializes reloads + owns the live analysis daemon handle. analysis_handle: tokio::sync::Mutex>, } impl Supervisors { pub async fn shutdown(&self) { if let Some(h) = self.crawler_handle.lock().await.take() { h.shutdown().await; } if let Some(h) = self.analysis_handle.lock().await.take() { h.shutdown().await; } } } #[async_trait] impl DaemonReloader for Supervisors { async fn reload_crawler(&self, cfg: CrawlerConfig) -> anyhow::Result<()> { // Serialize reloads; do the (possibly slow) graceful shutdown under // the handle lock but outside the read-path RwLock so status reads // never block on a draining browser. let mut guard = self.crawler_handle.lock().await; if let Some(h) = guard.take() { h.shutdown().await; } self.runtime.set_crawler(None, None); if cfg.daemon_enabled { let spawned = spawn_crawler_daemon( self.db.clone(), Arc::clone(&self.storage), &cfg, self.runtime.analysis_gate(), ) .await?; self.runtime .set_crawler(Some(spawned.resync), Some(spawned.crawler)); *guard = Some(spawned.handle); tracing::info!("crawler daemon (re)started from settings"); } else { tracing::info!("crawler daemon stopped (disabled in settings)"); } Ok(()) } async fn reload_analysis(&self, cfg: AnalysisConfig) -> anyhow::Result<()> { let mut guard = self.analysis_handle.lock().await; if let Some(h) = guard.take() { h.shutdown().await; } self.runtime.set_analysis_enabled(cfg.enabled); if cfg.enabled { let handle = spawn_analysis_daemon( self.db.clone(), Arc::clone(&self.storage), &cfg, Arc::clone(&self.analysis_events), ) .await?; *guard = Some(handle); tracing::info!(model = %cfg.model, "analysis daemon (re)started from settings"); } else { tracing::info!("analysis daemon stopped (disabled in settings)"); } Ok(()) } } pub async fn build(config: Config) -> anyhow::Result { let db = PgPoolOptions::new() .max_connections(10) .connect(&config.database_url) .await?; sqlx::migrate!("./migrations").run(&db).await?; if let Some((username, password)) = config.admin_bootstrap.as_ref() { repo::user::bootstrap_admin(&db, username, password) .await .context("bootstrap_admin from ADMIN_USERNAME/ADMIN_PASSWORD env")?; tracing::info!(admin_username = %username, "admin bootstrap ensured"); } let storage: Arc = Arc::new(LocalStorage::new(config.storage_dir.clone())); // Seed the settings rows from env on first boot, then load the effective // config (DB overlaid on the env base). The DB is the source of truth // after the first boot; env only fills a missing row. let crawler_cfg = load_effective_crawler(&db, &config.crawler).await?; let analysis_cfg = load_effective_analysis(&db, &config.analysis).await?; // Live-event bus shared by the worker (progress) and the admin enqueue // path; the SSE endpoint subscribes. Created unconditionally so the // stream exists even with the worker disabled. let analysis_events = Arc::new(crate::analysis::events::AnalysisEvents::new()); let runtime = Arc::new(RuntimeControls::new(analysis_cfg.enabled)); let supervisors = Arc::new(Supervisors { db: db.clone(), storage: Arc::clone(&storage), runtime: Arc::clone(&runtime), analysis_events: Arc::clone(&analysis_events), crawler_handle: tokio::sync::Mutex::new(None), analysis_handle: tokio::sync::Mutex::new(None), }); // Initial spawn goes through the same reload path so there's one code // path for "bring the daemon up with this config". A spawn failure is // logged but does NOT abort startup: the persisted settings are the // source of truth and a bad value (e.g. a config that won't launch the // browser) must not wedge the whole server into a boot loop — the // server comes up with that daemon stopped so an admin can fix it via // the settings UI. if crawler_cfg.daemon_enabled { if let Err(e) = supervisors.reload_crawler(crawler_cfg).await { tracing::error!(?e, "crawler daemon failed to start; continuing with it stopped"); } } else { tracing::info!("crawler daemon disabled"); } if analysis_cfg.enabled { if let Err(e) = supervisors.reload_analysis(analysis_cfg).await { tracing::error!(?e, "analysis worker failed to start; continuing with it stopped"); } } else { tracing::info!("analysis worker disabled"); } let auth_limiter = Arc::new(AuthRateLimiter::new(config.auth.rate_limit)); let state = AppState { db, storage, auth: config.auth.clone(), upload: config.upload.clone(), auth_limiter, runtime, reloader: Some(Arc::clone(&supervisors) as Arc), crawler_base: config.crawler.clone(), analysis_base: config.analysis.clone(), admin_allowed_origins: Arc::new(config.admin_allowed_origins.clone()), analysis_events, }; let router = router(state).layer(cors_layer(&config.cors_allowed_origins)); Ok(AppHandle { router, supervisors }) } /// Seed the crawler settings row from env when absent, then return the /// effective [`CrawlerConfig`] (DB DTO overlaid on the env base). A stored /// DTO that fails to convert (corruption / drift) falls back to the env base /// rather than blocking startup. async fn load_effective_crawler( db: &PgPool, base: &CrawlerConfig, ) -> anyhow::Result { use crate::settings::{CrawlerSettings, KEY_CRAWLER}; let dto = match repo::app_settings::get(db, KEY_CRAWLER).await? { Some(v) => serde_json::from_value::(v) .unwrap_or_else(|_| CrawlerSettings::from_config(base)), None => { let dto = CrawlerSettings::from_config(base); repo::app_settings::seed_if_absent(db, KEY_CRAWLER, &serde_json::to_value(&dto)?) .await?; tracing::info!("seeded crawler settings from env"); dto } }; Ok(dto.to_config(base).unwrap_or_else(|e| { tracing::warn!(?e, "stored crawler settings invalid; using env base"); base.clone() })) } /// Analysis counterpart of [`load_effective_crawler`]. async fn load_effective_analysis( db: &PgPool, base: &AnalysisConfig, ) -> anyhow::Result { use crate::settings::{AnalysisSettings, KEY_ANALYSIS}; let dto = match repo::app_settings::get(db, KEY_ANALYSIS).await? { Some(v) => serde_json::from_value::(v) .unwrap_or_else(|_| AnalysisSettings::from_config(base)), None => { let dto = AnalysisSettings::from_config(base); repo::app_settings::seed_if_absent(db, KEY_ANALYSIS, &serde_json::to_value(&dto)?) .await?; tracing::info!("seeded analysis settings from env"); dto } }; Ok(dto.to_config(base).unwrap_or_else(|e| { tracing::warn!(?e, "stored analysis settings invalid; using env base"); base.clone() })) } /// Spawn the AI content-analysis worker daemon with the given config. Returns /// its handle. Independent of the crawler daemon (works for uploads with the /// crawler off). Uses a plain reqwest client — no cookie jar / proxy. async fn spawn_analysis_daemon( db: PgPool, storage: Arc, cfg: &AnalysisConfig, events: Arc, ) -> anyhow::Result { // Reclaim jobs orphaned by a previous crash/kill. `crawler::jobs::reclaim_orphaned` // is keyed only on `state='running' AND leased_until < now()`, so it // covers any kind that uses the table — including this daemon's // `analyze_page` jobs. Previously this ran only inside // `spawn_crawler_daemon`, which meant a deploy with the crawler off // (analysis-only) never refunded orphaned analysis leases until the // lease expiry path lazily picked them up. Safe under multi-replica // (only expired leases are touched) and idempotent (the crawler-side // call happens-before this if both are running). match crate::crawler::jobs::reclaim_orphaned(&db).await { Ok(0) => {} Ok(n) => tracing::info!( reclaimed = n, "analysis: reclaimed orphaned in-flight jobs at startup" ), Err(e) => tracing::warn!(?e, "analysis: reclaim_orphaned at startup failed"), } // Pick the engine. The OCR backend runs in-process (no network, no // readiness gate); the vision backend talks to a local LLM server. let (dispatcher, readiness): ( Arc, Option>, ) = match cfg.effective_backend() { crate::config::AnalysisBackend::Ocr => { // Load the `.rten` models once; a bad path is a loud boot error. let engine = crate::analysis::ocr::OcrsEngine::from_model_paths( &cfg.ocr_detection_model, &cfg.ocr_recognition_model, cfg.ocr_max_decode_pixels, ) .context("build ocrs engine")?; // Cap concurrent CPU-bound OCR runs across all workers so a high // ANALYSIS_WORKERS can't oversubscribe the blocking pool. let cores = std::thread::available_parallelism() .map(|n| n.get()) .unwrap_or(1); let permits = crate::analysis::ocr::ocr_concurrency_limit(cfg.workers, cores); let dispatcher = Arc::new(crate::analysis::ocr::OcrAnalyzeDispatcher { db: db.clone(), storage, engine: Arc::new(engine), max_image_bytes: cfg.max_image_bytes, ocr_permits: Arc::new(tokio::sync::Semaphore::new(permits)), }); // In-process engine is always ready — no gate. (dispatcher, None) } crate::config::AnalysisBackend::Vision => { let http = reqwest::Client::builder() .timeout(cfg.request_timeout) // Refuse to honour ambient HTTP_PROXY / HTTPS_PROXY container // env. The vision call carries an env-managed bearer token + // page image bytes; a stray upstream proxy would exfiltrate // both. Mirrors the crawler client's `.no_proxy()` (see // `spawn_crawler_daemon`). .no_proxy() // Re-validate redirect hops so a hostile/compromised vision // endpoint can't 302 the bearer token + page bytes into the // deployment's internal network. No allowlist here (the // endpoint is a single admin-configured URL), so the policy // enforces scheme + private-IP only. .redirect(crate::crawler::safety::public_redirect_policy()) .build() .context("build analysis http client")?; let vision = crate::analysis::vision::VisionClient::new(http, cfg); let dispatcher = Arc::new(crate::analysis::daemon::RealAnalyzeDispatcher { db: db.clone(), storage, vision, model: cfg.model.clone(), max_image_bytes: cfg.max_image_bytes, }); // When a readiness URL is configured, gate leasing on it so an // autoscaler that idle-stops the vision container never lets a job // burn its retries. A dedicated short-timeout client keeps the // probe snappy and independent of the (long) per-request timeout. let readiness: Option> = match &cfg.vision_health_url { Some(url) if !url.is_empty() => { let probe = reqwest::Client::builder() .timeout(std::time::Duration::from_secs(5)) // Same reasoning as the main analysis client: do // not honour ambient HTTP_PROXY env. The readiness // probe is unauthenticated but a hostile upstream // still gets a useful side-channel on backend // uptime + vision health. .no_proxy() .redirect(crate::crawler::safety::public_redirect_policy()) .build() .context("build vision readiness http client")?; Some(Arc::new(crate::analysis::daemon::HttpVisionReadiness { http: probe, health_url: url.clone(), })) } _ => None, }; (dispatcher, readiness) } }; let handle = crate::analysis::daemon::spawn( db, CancellationToken::new(), crate::analysis::daemon::AnalysisDaemonConfig { dispatcher, workers: cfg.workers, job_timeout: cfg.job_timeout, events, readiness, }, ); // Log the *effective* backend (what the worker actually dispatches // through), not the raw `cfg.backend` — with vision dormant they diverge // when an operator requests `vision`, and a misleading line here was a // real observability footgun. The model label tracks the effective engine. let effective_backend = cfg.effective_backend(); let effective_model: &str = match effective_backend { crate::config::AnalysisBackend::Ocr => crate::analysis::ocr::OCR_MODEL_LABEL, crate::config::AnalysisBackend::Vision => &cfg.model, }; tracing::info!( workers = cfg.workers, backend = ?effective_backend, model = %effective_model, "analysis worker daemon started" ); Ok(handle) } /// Bundle returned by [`spawn_crawler_daemon`]. The handle owns the /// daemon's tasks; `resync` is the operator-trigger service shared with /// `AppState` so admin endpoints can call into the same browser / /// rate-limit machinery. struct SpawnedDaemon { handle: daemon::DaemonHandle, resync: Arc, crawler: Arc, } async fn spawn_crawler_daemon( db: PgPool, storage: Arc, cfg: &CrawlerConfig, analysis_enabled: Arc, ) -> anyhow::Result { // Reqwest client with a shared cookie jar so CDN image fetches include // PHPSESSID. The same `Arc` is held by the SessionController, so a // runtime session refresh rewrites it in place. Initial value: a // persisted runtime session (survives restart) takes precedence over // CRAWLER_PHPSESSID env. let cookie_jar = Arc::new(reqwest::cookie::Jar::default()); let initial_sid = crate::crawler::session_control::SessionController::load_persisted(&db) .await .or_else(|| cfg.phpsessid.clone()); if let (Some(sid), Some(domain), Some(start_url)) = (&initial_sid, &cfg.cookie_domain, &cfg.start_url) { let cookie_str = format!("PHPSESSID={sid}; Domain={domain}; Path=/"); let seed_url = reqwest::Url::parse(start_url) .context("parse CRAWLER_START_URL for cookie seed")?; cookie_jar.add_cookie_str(&cookie_str, &seed_url); } let mut http_builder = reqwest::Client::builder() .timeout(std::time::Duration::from_secs(30)) .no_proxy() // Re-validate every redirect hop against the download allowlist: // reqwest's default policy follows up to 10 redirects, and // `is_safe_url` only guards the initial URL, so an allowlisted CDN // 302ing to a private IP would otherwise be followed (SSRF). .redirect(crate::crawler::safety::safe_redirect_policy( cfg.download_allowlist.clone(), )) .cookie_provider(Arc::clone(&cookie_jar)); if let Some(ua) = &cfg.user_agent { http_builder = http_builder.user_agent(ua); } if let Some(proxy) = &cfg.proxy { http_builder = http_builder .proxy(reqwest::Proxy::all(proxy).with_context(|| format!("parse proxy: {proxy}"))?); } let http = http_builder.build().context("build crawler reqwest")?; let mut rate = HostRateLimiters::new(std::time::Duration::from_millis(cfg.rate_ms)); if let Some(host) = &cfg.cdn_host { rate = rate.with_override(host, std::time::Duration::from_millis(cfg.cdn_rate_ms)); } let rate = Arc::new(rate); let tor = crate::crawler::tor::TorController::from_parts( cfg.tor_control_url.as_deref(), cfg.tor_control_password.as_deref(), cfg.tor_control_cookie_path.as_deref(), ) .context("build TorController from CRAWLER_TOR_CONTROL_* env")? .map(Arc::new); if let Some(t) = &tor { tracing::info!(?t, "TOR control configured; transient pages will trigger NEWNYM"); } let tor_recircuit_max = cfg.tor_recircuit_max_attempts; // Session controller + sticky session-expired flag. Created before the // browser so the on_launch hook can read the *current* session value // (rather than a value captured at startup), and so a runtime refresh // updates the cookie everywhere. let session_expired = Arc::new(AtomicBool::new(false)); let session_controller = crate::crawler::session_control::SessionController::new( initial_sid, Arc::clone(&cookie_jar), cfg.cookie_domain.clone(), cfg.start_url.clone(), db.clone(), Arc::clone(&session_expired), ); // Live status surface, sized to the worker count. let status = crate::crawler::status::StatusHandle::new(cfg.chapter_workers); // Browser manager. on_launch re-injects PHPSESSID on every fresh // chromium spawn so an idle teardown followed by re-launch stays // authenticated without operator action. let mut launch_opts = cfg.browser.clone(); if let Some(proxy) = &cfg.proxy { let chromium_proxy = crate::crawler::url_utils::chromium_proxy_arg(proxy); launch_opts.extra_args.push(format!("--proxy-server={chromium_proxy}")); } let on_launch = match (&cfg.cookie_domain, &cfg.start_url) { (Some(domain), Some(start_url)) => { let domain = domain.clone(); let start_url = start_url.clone(); let tor_for_launch = tor.as_ref().map(Arc::clone); let sc = Arc::clone(&session_controller); let on_launch: browser_manager::OnLaunch = Arc::new(move |browser| { let domain = domain.clone(); let start_url = start_url.clone(); let tor_for_launch = tor_for_launch.as_ref().map(Arc::clone); let sc = Arc::clone(&sc); Box::pin(async move { // Read the *current* session each launch so a runtime // refresh is picked up on the next (re)launch. No session // configured → run unauthenticated (metadata needs no auth). let Some(sid) = sc.current().await else { tracing::info!("on_launch: no session set — skipping inject + probe"); return Ok(()); }; session::inject_phpsessid(&browser, &sid, &domain) .await .context("on_launch: inject_phpsessid")?; session::verify_session_with_recircuit( &browser, &start_url, tor_for_launch.as_deref(), tor_recircuit_max, ) .await .context("on_launch: verify_session")?; Ok(()) }) }); on_launch } _ => browser_manager::noop_on_launch(), }; let browser_manager = BrowserManager::new(launch_opts, cfg.idle_timeout, on_launch); let metadata_pass: Option> = cfg.start_url.as_ref().map(|url| { let m: Arc = Arc::new(RealMetadataPass { browser_manager: Arc::clone(&browser_manager), db: db.clone(), storage: Arc::clone(&storage), http: http.clone(), rate: Arc::clone(&rate), start_url: url.clone(), manga_limit: cfg.manga_limit, download_allowlist: cfg.download_allowlist.clone(), max_image_bytes: cfg.max_image_bytes, metadata_max_consecutive_failures: cfg.metadata_max_consecutive_failures, status: status.clone(), tor: tor.as_ref().map(Arc::clone), }); m }); let reconcile_pass: Option> = cfg.start_url.as_ref().map(|url| { let m: Arc = Arc::new(RealReconcilePass { browser_manager: Arc::clone(&browser_manager), db: db.clone(), rate: Arc::clone(&rate), start_url: url.clone(), status: status.clone(), tor: tor.as_ref().map(Arc::clone), }); m }); let dispatcher: Arc = Arc::new(RealChapterDispatcher { browser_manager: Arc::clone(&browser_manager), db: db.clone(), storage: Arc::clone(&storage), http: http.clone(), rate: Arc::clone(&rate), download_allowlist: cfg.download_allowlist.clone(), max_image_bytes: cfg.max_image_bytes, max_images_per_chapter: cfg.max_images_per_chapter, analysis_enabled, transient_failures: Arc::new(AtomicU32::new(0)), restart_threshold: cfg.browser_restart_threshold, drain_deadline: cfg.job_timeout, status: status.clone(), tor: tor.as_ref().map(Arc::clone), }); let resync: Arc = Arc::new(RealResyncService { browser_manager: Arc::clone(&browser_manager), db: db.clone(), storage: Arc::clone(&storage), http, rate: Arc::clone(&rate), download_allowlist: cfg.download_allowlist.clone(), max_image_bytes: cfg.max_image_bytes, max_images_per_chapter: cfg.max_images_per_chapter, tor: tor.as_ref().map(Arc::clone), }); // Shared cancellation: daemon shutdown cancels the BrowserManager's // idle reaper too. Reaper itself is added to the daemon's extra_tasks // so DaemonHandle::shutdown awaits its completion. let cancel = CancellationToken::new(); let reaper_task = browser_manager::spawn_idle_reaper( Arc::clone(&browser_manager), cancel.clone(), ); // Also close the browser explicitly on shutdown so we don't rely on // kill-on-drop when other Arc holders may still exist. let shutdown_task = { let cancel = cancel.clone(); let mgr = Arc::clone(&browser_manager); tokio::spawn(async move { cancel.cancelled().await; mgr.shutdown().await; }) }; // 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, DaemonConfig { metadata_pass: metadata_pass.clone(), dispatcher, chapter_workers: cfg.chapter_workers, daily_at: cfg.daily_at, tz: cfg.tz, retention_days: cfg.retention_days, metrics_retention_days: cfg.metrics_retention_days, session_expired, status: status.clone(), job_timeout: cfg.job_timeout, extra_tasks: vec![reaper_task, shutdown_task], }, ); let crawler = Arc::new(CrawlerControl { browser_manager: Arc::clone(&browser_manager), session: session_controller, status, metadata_pass, reconcile_pass, drain_deadline: cfg.job_timeout, manual_pass_lock: Arc::new(tokio::sync::Mutex::new(())), }); Ok(SpawnedDaemon { handle: daemon_handle, resync, crawler, }) } // Real impls of the daemon traits, owning the browser manager + I/O. Kept // in app.rs because they need the same builder-side env wiring that // AppState gets — the daemon module itself stays free of reqwest / storage // details so its tests don't pull them in. struct RealMetadataPass { browser_manager: Arc, db: PgPool, storage: Arc, http: reqwest::Client, rate: Arc, start_url: String, manga_limit: usize, download_allowlist: DownloadAllowlist, max_image_bytes: usize, metadata_max_consecutive_failures: u32, status: crate::crawler::status::StatusHandle, tor: Option>, } #[async_trait] impl MetadataPass for RealMetadataPass { async fn run(&self) -> anyhow::Result { let result = pipeline::run_metadata_pass( &self.browser_manager, &self.db, self.storage.as_ref(), &self.http, &self.rate, &self.start_url, self.manga_limit, false, &self.download_allowlist, self.max_image_bytes, self.metadata_max_consecutive_failures, Some(&self.status), self.tor.as_deref(), ) .await; if let Err(e) = &result { if crate::crawler::nav::anyhow_looks_browser_dead(e) { self.browser_manager.invalidate().await; } } // Cover backfill follows the metadata pass even when the pass // errored — the early-stop walk can complete its work and bail // late, and a transient browser failure shouldn't cancel the // residual cover backlog. The backfill has its own per-call cap // so a runaway error stream can't monopolise the tick. It sets the // CoverBackfill{index,total} phase + current_cover per entry. match pipeline::backfill_missing_covers( &self.browser_manager, &self.db, self.storage.as_ref(), &self.http, &self.rate, pipeline::COVER_BACKFILL_DEFAULT_MAX, &self.download_allowlist, self.max_image_bytes, Some(&self.status), self.tor.as_deref(), ) .await { Ok(stats) => { if stats.considered > 0 { tracing::info!(?stats, "cover backfill complete"); } } Err(e) => { tracing::warn!(error = ?e, "cover backfill failed"); if crate::crawler::nav::anyhow_looks_browser_dead(&e) { self.browser_manager.invalidate().await; } } } result } } struct RealReconcilePass { browser_manager: Arc, db: PgPool, rate: Arc, start_url: String, status: crate::crawler::status::StatusHandle, tor: Option>, } #[async_trait] impl crate::crawler::daemon::ReconcilePass for RealReconcilePass { async fn run(&self) -> anyhow::Result { let result = crate::crawler::reconcile::reconcile_missing( &self.browser_manager, &self.db, &self.rate, &self.start_url, Some(&self.status), self.tor.as_deref(), ) .await; if let Err(e) = &result { if crate::crawler::nav::anyhow_looks_browser_dead(e) { self.browser_manager.invalidate().await; } } result } } struct RealChapterDispatcher { browser_manager: Arc, db: PgPool, storage: Arc, http: reqwest::Client, rate: Arc, download_allowlist: DownloadAllowlist, max_image_bytes: usize, /// Per-chapter image-count cap (see `CrawlerConfig::max_images_per_chapter`). max_images_per_chapter: usize, /// Enqueue `analyze_page` jobs for freshly-crawled pages. Shared gate /// (read live) so toggling analysis at runtime takes effect without a /// crawler respawn. Mirrors the analysis enable setting. analysis_enabled: Arc, /// Consecutive transient chapter failures; resets on any success. /// Drives the automatic coordinated browser restart. transient_failures: Arc, /// Consecutive-failure count that triggers an auto restart. restart_threshold: u32, /// How long a coordinated restart waits for in-flight leases to drain. drain_deadline: std::time::Duration, /// Live status surface — the dispatcher registers each chapter it /// crawls (with a realtime page count) here. status: crate::crawler::status::StatusHandle, tor: Option>, } #[async_trait] impl ChapterDispatcher for RealChapterDispatcher { async fn dispatch(&self, payload: JobPayload) -> anyhow::Result { match payload { JobPayload::SyncChapterContent { source_id: _, chapter_id, source_chapter_key: _, } => { let row = repo::chapter::dispatch_target(&self.db, chapter_id) .await .context("look up chapter for dispatch")?; let Some((manga_id, source_url, manga_title, chapter_number)) = row else { // Chapter (or its source row) is gone — ack done. return Ok(SyncOutcome::Skipped); }; // Register the chapter as crawling now (live status). The // guard removes it on every exit path — success, panic, or // the worker's outer-timeout drop. let _active = self.status.begin_chapter(crate::crawler::status::ActiveChapter { manga_id, manga_title, chapter_id, chapter_number, pages_done: 0, pages_total: None, }); let lease = self.browser_manager.acquire().await?; let result = content::sync_chapter_content( &lease, &self.db, self.storage.as_ref(), &self.http, &self.rate, chapter_id, manga_id, &source_url, false, &self.download_allowlist, self.max_image_bytes, self.max_images_per_chapter, self.tor.as_deref(), Some(&self.status), self.analysis_enabled.load(Ordering::Relaxed), ) .await; drop(lease); match result { Ok(outcome) => { // Any successful dispatch (including a clean Skipped) // means the browser is healthy — reset the streak. self.transient_failures.store(0, Ordering::Release); Ok(outcome) } Err(e) => { let streak = self.transient_failures.fetch_add(1, Ordering::AcqRel) + 1; if crate::crawler::nav::anyhow_looks_browser_dead(&e) { // Hard browser-dead: lazy invalidate (next acquire // relaunches). Reset the streak — we're recovering. self.browser_manager.invalidate().await; self.transient_failures.store(0, Ordering::Release); } else if self.restart_threshold > 0 && streak >= self.restart_threshold { // Persistent transients that TOR recircuit couldn't // fix — proactively restart Chromium. tracing::warn!( streak, threshold = self.restart_threshold, "auto browser restart: consecutive transient chapter failures" ); let _ = self .browser_manager .coordinated_restart(self.drain_deadline) .await; self.transient_failures.store(0, Ordering::Release); } Err(e) } } } // Reconcile-enqueued manga-detail sync: fetch the detail page, // upsert metadata, sync chapters — the identical per-ref work the // cron metadata pass runs inline, via the shared // `pipeline::process_manga_ref`. JobPayload::SyncManga { source_id: _, source_manga_key, url, title, } => { let source = crate::crawler::source::target::TargetSource::new(url.clone()); let r = crate::crawler::source::SourceMangaRef { source_manga_key, title, url, }; // Scope the lease so it (and the borrowing FetchContext) drop // before any browser-restart handling in the match below. let result = { let lease = self.browser_manager.acquire().await?; let ctx = crate::crawler::source::FetchContext { browser: &lease, rate: &self.rate, tor: self.tor.as_deref(), }; pipeline::process_manga_ref( &ctx, &source, &self.db, self.storage.as_ref(), &self.http, &self.rate, &r, false, // chapters ON — we want chapter rows synced &self.download_allowlist, self.max_image_bytes, Some(&self.status), ) .await }; match result { Ok(p) => { self.transient_failures.store(0, Ordering::Release); tracing::info!( manga_id = %p.manga_id, key = %r.source_manga_key, "SyncManga: manga synced" ); Ok(SyncOutcome::Fetched { pages: 0 }) } Err(pipeline::RefError::Fetch(e)) | Err(pipeline::RefError::Skip(e)) => { let streak = self.transient_failures.fetch_add(1, Ordering::AcqRel) + 1; if crate::crawler::nav::anyhow_looks_browser_dead(&e) { self.browser_manager.invalidate().await; self.transient_failures.store(0, Ordering::Release); } else if self.restart_threshold > 0 && streak >= self.restart_threshold { tracing::warn!( streak, threshold = self.restart_threshold, "auto browser restart: consecutive transient sync_manga failures" ); let _ = self .browser_manager .coordinated_restart(self.drain_deadline) .await; self.transient_failures.store(0, Ordering::Release); } Err(e) } } } // Other payload kinds aren't dispatched by this daemon — // SyncChapterList is handled inline by the cron's metadata pass; // analyze_page is owned by the analysis daemon. _ => Ok(SyncOutcome::Skipped), } } } /// Build a router from a pre-assembled state. Used by integration tests /// so they can swap in a test DB pool and a `tempfile`-backed storage. pub fn router(state: AppState) -> Router { let max_request_bytes = state.upload.max_request_bytes; Router::new() .nest("/api/v1", crate::api::routes()) .layer(middleware::from_fn(admin_no_store_guard)) .layer(middleware::from_fn_with_state( state.clone(), admin_csrf_guard, )) .layer(middleware::from_fn_with_state( state.clone(), private_mode_guard, )) .layer(DefaultBodyLimit::max(max_request_bytes)) .with_state(state) .layer(TraceLayer::new_for_http()) } /// Path prefix the admin-only middlewares scope themselves to. The router /// already nests `/api/v1`, so callers see `/api/v1/admin/...`. const ADMIN_PATH_PREFIX: &str = "/api/v1/admin/"; /// CSRF defence for cookie-authenticated admin mutations. The session /// cookie is `SameSite=Lax`, which still permits top-level form-POSTs /// from a malicious page — this middleware rejects such requests by /// comparing the request's `Origin` (with `Referer` as fallback) against /// the configured allowlist. Safe methods (`GET`/`HEAD`/`OPTIONS`) are /// always allowed. /// /// **Bearer-token-only requests** (`Authorization: Bearer …` with NO /// session cookie) are bot API callers — they can't be a CSRF vector /// because the browser never attaches the Authorization header /// automatically. Skip the check for them. /// /// **Bearer + cookie (the "cookie-ride")** is treated as cookie-auth. /// An attacker page can mint `Authorization: Bearer junk` on a /// credentialed cross-site POST; the cookie carries the actual /// authority. Closing this hole means cookie precedence: as soon as a /// session cookie is present, the CSRF gate fires regardless of the /// Authorization header. (0.87.10 closed a 0.87.2 regression here.) /// /// **Cookie-auth requests** must: /// * be from an allowed origin (`Origin` then `Referer`), AND /// * have at least one of those headers present (a browser always /// sends one on a cross-site POST; a missing pair on a cookie-auth /// request is the exact niche an extension / no-referrer-policy /// CSRF would try to exploit). /// /// When the allowlist is empty we fail-closed for cookie-auth requests /// — this used to silently let everything through, so an operator who /// forgot to set `ADMIN_ALLOWED_ORIGINS` shipped an unguarded admin /// surface. Operators on a pure-bot-token deploy aren't impacted (their /// requests carry only `Authorization: Bearer …` with no session cookie /// and skip the gate via the bearer-only branch above). /// /// **No auth at all** (no cookie, no bearer): bypass the CSRF gate so /// the auth extractor returns a clean 401 instead of a confusing 403. async fn admin_csrf_guard( State(state): State, req: Request, next: Next, ) -> Result { if !req.uri().path().starts_with(ADMIN_PATH_PREFIX) { return Ok(next.run(req).await); } if matches!( *req.method(), Method::GET | Method::HEAD | Method::OPTIONS ) { return Ok(next.run(req).await); } let headers = req.headers(); // Detect both auth modes BEFORE choosing a bypass. The previous // version short-circuited on "is_bearer" regardless of cookie state, // which let an attacker page do a credentialed cross-site POST with // a forged `Authorization: Bearer junk` header: the header existed, // CSRF bypassed, then the auth extractor authenticated the victim // via the session cookie. Cookie precedence here re-anchors the // gate to the actual authority being ridden. // Drive the parse off the auth-module constant so a rename of // `SESSION_COOKIE_NAME` propagates here instead of silently // reopening the cookie-ride. Split on `=` (not a prefix match) so // an attacker can't shadow with `mangalord_session_x=` etc. let has_session_cookie = headers .get(axum::http::header::COOKIE) .and_then(|v| v.to_str().ok()) .is_some_and(|raw| { raw.split(';').any(|p| { p.trim_start().split('=').next() == Some(crate::auth::extractor::SESSION_COOKIE_NAME) }) }); // Bearer-token-only requests are bot callers and bypass the gate — // a browser can't set Authorization on a cross-site POST. But ONLY // when no session cookie is also attached; with both present we // must treat the request as cookie-auth to defeat the cookie-ride. let is_bearer = headers .get(axum::http::header::AUTHORIZATION) .and_then(|v| v.to_str().ok()) .is_some_and(|v| v.trim_start().to_ascii_lowercase().starts_with("bearer ")); if is_bearer && !has_session_cookie { return Ok(next.run(req).await); } // No auth context at all → let the auth extractor return a clean 401 // ("log in first"). A no-auth request can't be a CSRF vector — there's // no authority to ride. Without this, an anonymous curl POST to an // admin endpoint surfaces 403 with our CSRF message instead of the // expected 401, which is confusing for both operators and tests. if !has_session_cookie { return Ok(next.run(req).await); } let origin = headers.get("origin").and_then(|v| v.to_str().ok()); let referer = headers.get("referer").and_then(|v| v.to_str().ok()); let candidate = origin.or(referer); if state.admin_allowed_origins.is_empty() { // Fail-closed: cookie-auth admin mutations without an allowlist // configuration are refused. Set `ADMIN_ALLOWED_ORIGINS` for the // browser deployment, or call with `Authorization: Bearer …`. tracing::warn!( path = %req.uri().path(), "admin CSRF: ADMIN_ALLOWED_ORIGINS is empty — cookie-auth admin mutations are refused (set the env var for browser-exposed deploys)" ); return Err(AppError::Forbidden); } // Cookie-auth path with an allowlist configured: a browser always // sends Origin or Referer on a cross-site POST. A missing pair is // either a no-referrer-policy edge case or a tool deliberately // hiding origin — refuse rather than risk it. let Some(candidate) = candidate else { tracing::warn!( path = %req.uri().path(), "admin CSRF: cookie-auth request with neither Origin nor Referer — refusing" ); return Err(AppError::Forbidden); }; if origin_in_allowlist(candidate, &state.admin_allowed_origins) { return Ok(next.run(req).await); } tracing::warn!( candidate = %truncate_for_log(candidate, 64), path = %req.uri().path(), "admin CSRF: rejecting mutation with disallowed origin" ); Err(AppError::Forbidden) } /// Match `candidate` (an `Origin` value, or a `Referer` URL whose /// origin we'll extract) against `allowed`. `Origin` is `scheme://host[:port]` /// with no path; `Referer` is a full URL — compare by parsing both and /// matching scheme + host + port. fn origin_in_allowlist(candidate: &str, allowed: &[String]) -> bool { let cand_origin = parse_origin(candidate); let Some(cand) = cand_origin else { return false }; allowed .iter() .filter_map(|a| parse_origin(a)) .any(|a| a == cand) } /// Extract the origin (`scheme://host[:port]`) from an `Origin` header /// value or a `Referer` URL. Returns `None` when the input doesn't parse /// as a URL with a host. fn parse_origin(raw: &str) -> Option { let url = reqwest::Url::parse(raw.trim()).ok()?; let host = url.host_str()?; let scheme = url.scheme(); let port_str = match (url.port(), scheme) { (Some(80), "http") => String::new(), (Some(443), "https") => String::new(), (Some(p), _) => format!(":{p}"), (None, _) => String::new(), }; Some(format!("{scheme}://{host}{port_str}")) } fn truncate_for_log(s: &str, max: usize) -> &str { let end = s .char_indices() .take(max) .last() .map(|(i, c)| i + c.len_utf8()) .unwrap_or(0); &s[..end.min(s.len())] } /// Forbids intermediaries (CDN, browser bfcache, reverse proxy with a /// permissive default) from caching admin responses. Defence-in-depth /// for cookie-authenticated reads — even though the responses already /// vary on cookie, a misconfigured cache layer in front of the /// SvelteKit container could leak a logged-in admin's view to another /// session. Headers added on response so the rest of the API is /// unaffected. async fn admin_no_store_guard(req: Request, next: Next) -> Response { let is_admin_path = req.uri().path().starts_with(ADMIN_PATH_PREFIX); let mut resp = next.run(req).await; if is_admin_path { resp.headers_mut().insert( axum::http::header::CACHE_CONTROL, HeaderValue::from_static("no-store"), ); } resp } /// Paths reachable anonymously even when `PRIVATE_MODE=true`. Login and /// logout are needed for the auth flow itself; `/health` is reserved /// for load-balancer probes; `/auth/config` lets the frontend decide /// whether to render the login form or its anonymous alternatives; /// `/auth/register` is exempted from the gate so the handler can /// return its informative `registration_disabled` 403 (the same code /// public-mode deployments use when `ALLOW_SELF_REGISTER=false`) — /// the handler itself force-blocks the request body in private mode, /// so no account ever gets created here. Everything else demands a /// valid session cookie or bearer token. fn is_public_in_private_mode(path: &str) -> bool { matches!( path, "/api/v1/health" | "/api/v1/auth/config" | "/api/v1/auth/login" | "/api/v1/auth/logout" | "/api/v1/auth/register" ) } /// Site-wide auth gate for `PRIVATE_MODE=true`. With the flag off this /// is a no-op pass-through, so public deployments take no extra DB /// hit. With it on, the guard reuses [`CurrentUser`] — the same /// session-cookie-then-bearer-token logic the per-handler extractor /// uses — so the two paths can never drift. async fn private_mode_guard( State(state): State, req: Request, next: Next, ) -> Result { if !state.auth.private_mode { return Ok(next.run(req).await); } if is_public_in_private_mode(req.uri().path()) { return Ok(next.run(req).await); } let (mut parts, body) = req.into_parts(); match CurrentUser::from_request_parts(&mut parts, &state).await { Ok(_) => { let req = Request::from_parts(parts, body); Ok(next.run(req).await) } Err(_) => Err(AppError::Unauthenticated), } } pub(crate) fn cors_layer(allowed_origins: &[String]) -> CorsLayer { if allowed_origins.is_empty() { // Same-origin only — no CORS headers emitted. return CorsLayer::new(); } let origins: Vec = allowed_origins .iter() .filter_map(|o| HeaderValue::from_str(o).ok()) .collect(); CorsLayer::new() .allow_origin(AllowOrigin::list(origins)) .allow_credentials(true) .allow_methods([Method::GET, Method::POST, Method::PUT, Method::DELETE]) .allow_headers([ HeaderName::from_static("content-type"), HeaderName::from_static("authorization"), ]) } #[cfg(test)] mod tests { use super::*; use axum::body::Body; use axum::http::Request; use axum::routing::get; use tower::ServiceExt; fn test_router() -> Router { Router::new().route("/", get(|| async { "ok" })) } #[tokio::test] async fn allowlist_preflight_emits_credentialed_headers() { let app = test_router().layer(cors_layer(&["https://app.example.com".to_string()])); let resp = app .oneshot( Request::builder() .method(Method::OPTIONS) .uri("/") .header("origin", "https://app.example.com") .header("access-control-request-method", "POST") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!( resp.headers().get("access-control-allow-origin").unwrap(), "https://app.example.com" ); assert_eq!( resp.headers().get("access-control-allow-credentials").unwrap(), "true" ); } #[tokio::test] async fn allowlist_rejects_unlisted_origin() { let app = test_router().layer(cors_layer(&["https://app.example.com".to_string()])); let resp = app .oneshot( Request::builder() .method(Method::OPTIONS) .uri("/") .header("origin", "https://evil.example.org") .header("access-control-request-method", "POST") .body(Body::empty()) .unwrap(), ) .await .unwrap(); // Browsers will refuse the response when the allow-origin header // is absent (or doesn't echo the requesting origin). assert!(resp.headers().get("access-control-allow-origin").is_none()); } #[tokio::test] async fn empty_allowlist_is_same_origin_only() { let app = test_router().layer(cors_layer(&[])); let resp = app .oneshot( Request::builder() .method(Method::OPTIONS) .uri("/") .header("origin", "https://app.example.com") .header("access-control-request-method", "POST") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert!(resp.headers().get("access-control-allow-origin").is_none()); assert!(resp.headers().get("access-control-allow-credentials").is_none()); } /// `spawn_analysis_daemon` runs `crawler::jobs::reclaim_orphaned` at /// startup so an analysis-only deploy refunds expired /// `analyze_page` leases that a crashed previous run left running. /// Without this, the call could be silently removed and the /// reclaim test in `tests/crawler_jobs.rs` would still pass /// (it covers the helper, not the wiring). /// /// Seed an expired-lease `analyze_page` row in `crawler_jobs`, call /// `spawn_analysis_daemon`, then assert the row went back to /// `pending` with the attempt refunded. #[sqlx::test(migrations = "./migrations")] async fn spawn_analysis_daemon_reclaims_orphaned_analyze_leases(pool: PgPool) { use crate::storage::LocalStorage; use std::time::Duration; use uuid::Uuid; // Seed a chapter + page so the worker has something it COULD lease. // We don't need the worker to actually run — we're testing the // reclaim that happens before workers start. let manga_id: Uuid = sqlx::query_scalar("INSERT INTO mangas (title) VALUES ('M') RETURNING id") .fetch_one(&pool) .await .unwrap(); let chapter_id: Uuid = sqlx::query_scalar( "INSERT INTO chapters (manga_id, number) VALUES ($1, 1) RETURNING id", ) .bind(manga_id) .fetch_one(&pool) .await .unwrap(); let page_id: Uuid = sqlx::query_scalar( "INSERT INTO pages (chapter_id, page_number, storage_key, content_type) \ VALUES ($1, 1, 'k/1.png', 'image/png') RETURNING id", ) .bind(chapter_id) .fetch_one(&pool) .await .unwrap(); // Plant an expired-lease analyze_page row that mimics a previous // worker that crashed mid-dispatch (attempts=1, leased_until in // the past, state=running). let payload = serde_json::json!({ "kind": "analyze_page", "page_id": page_id, "force": false }); sqlx::query( "INSERT INTO crawler_jobs (payload, state, attempts, leased_until) \ VALUES ($1, 'running', 1, now() - interval '1 hour')", ) .bind(payload) .execute(&pool) .await .unwrap(); // Bind to a local so the TempDir lives for the rest of the test. // `tempfile::tempdir().unwrap().path()` would drop the TempDir // at end-of-expression and `LocalStorage` would hold a path to // a deleted directory. let storage_dir = tempfile::tempdir().unwrap(); let storage: Arc = Arc::new(LocalStorage::new(storage_dir.path())); // The worker always runs OCR now (vision is dormant — see // `effective_backend`), and the `.rten` models aren't shipped to unit // CI. Point the engine at a path that can't exist so the *engine build* // fails deterministically. Reclaim runs at the very top of // `spawn_analysis_daemon`, before — and independently of — engine // readiness, so the row must still be reclaimed even though spawn // returns Err. That's exactly the regression this test guards (an // analysis-only deploy must reclaim orphaned leases at startup). let cfg = crate::config::AnalysisConfig { ocr_detection_model: "/nonexistent/text-detection.rten".to_string(), ocr_recognition_model: "/nonexistent/text-recognition.rten".to_string(), workers: 1, job_timeout: Duration::from_secs(1), ..Default::default() }; let events = Arc::new(crate::analysis::events::AnalysisEvents::new()); let spawned = spawn_analysis_daemon(pool.clone(), storage, &cfg, events).await; assert!( spawned.is_err(), "engine build must fail with a missing model path" ); // The reclaim must have moved the row back to pending with the // attempt refunded (attempts goes from 1 → 0). Reaching it on the // initial running state would mean reclaim never ran. let (state, attempts): (String, i32) = sqlx::query_as( "SELECT state, attempts FROM crawler_jobs WHERE payload->>'page_id' = $1", ) .bind(page_id.to_string()) .fetch_one(&pool) .await .unwrap(); assert_eq!( state, "pending", "expired-lease analyze_page row must be reclaimed to pending" ); assert_eq!(attempts, 0, "reclaim_orphaned refunds the attempt"); } }