diff --git a/backend/Cargo.lock b/backend/Cargo.lock index ae91b23..2556483 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1517,7 +1517,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.85.1" +version = "0.86.0" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 10cf0b7..d7a4083 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.85.1" +version = "0.86.0" edition = "2021" default-run = "mangalord" diff --git a/backend/src/api/admin/crawler/control.rs b/backend/src/api/admin/crawler/control.rs index 32eece2..30076f4 100644 --- a/backend/src/api/admin/crawler/control.rs +++ b/backend/src/api/admin/crawler/control.rs @@ -24,6 +24,7 @@ use super::require_crawler; pub(super) fn routes() -> Router { Router::new() .route("/admin/crawler/run", post(run_now)) + .route("/admin/crawler/reconcile", post(reconcile_now)) .route("/admin/crawler/browser/restart", post(restart_browser)) .route("/admin/crawler/session", post(update_session)) .route( @@ -71,6 +72,45 @@ async fn run_now( Ok(Json(RunResponse { started: true })) } +async fn reconcile_now( + State(state): State, + admin: RequireAdmin, +) -> AppResult> { + let c = require_crawler(&state)?; + let rp = c.reconcile_pass.as_ref().ok_or_else(|| { + AppError::ServiceUnavailable("no source configured (CRAWLER_START_URL unset)".into()) + })?; + // Share `manual_pass_lock` with the metadata pass: reconcile's list walk + // and a metadata pass both contend for the single browser lease, so they + // must not run concurrently. A click while either is in flight gets 409. + let pass_guard = c + .manual_pass_lock + .clone() + .try_lock_owned() + .map_err(|_| AppError::Conflict("a metadata or reconcile pass is already running".into()))?; + let rp = std::sync::Arc::clone(rp); + // Fire-and-forget: the full list walk runs for minutes; progress streams + // over SSE (the Reconciling phase). The guard moves into the task so the + // lock releases only when the walk + enqueue finish. + tokio::spawn(async move { + let _pass_guard = pass_guard; + match rp.run().await { + Ok(stats) => tracing::info!(?stats, "manual reconcile pass complete"), + Err(e) => tracing::warn!(error = ?e, "manual reconcile pass failed"), + } + }); + repo::admin_audit::insert( + &state.db, + admin.0.id, + "crawler_reconcile", + "crawler", + None, + json!({}), + ) + .await?; + Ok(Json(RunResponse { started: true })) +} + #[derive(Debug, Serialize)] struct RestartResponse { ok: bool, diff --git a/backend/src/app.rs b/backend/src/app.rs index 038b24e..aadb207 100644 --- a/backend/src/app.rs +++ b/backend/src/app.rs @@ -170,6 +170,9 @@ pub struct CrawlerControl { /// 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 @@ -599,6 +602,19 @@ async fn spawn_crawler_daemon( 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(), @@ -679,6 +695,7 @@ async fn spawn_crawler_daemon( session: session_controller, status, metadata_pass, + reconcile_pass, drain_deadline: cfg.job_timeout, manual_pass_lock: Arc::new(tokio::sync::Mutex::new(())), }); @@ -770,6 +787,36 @@ impl MetadataPass for RealMetadataPass { } } +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, @@ -873,9 +920,80 @@ impl ChapterDispatcher for RealChapterDispatcher { } } } - // Other payload kinds aren't dispatched by this daemon yet — - // SyncManga / SyncChapterList are handled inline by the cron's - // metadata pass. + // 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), } } diff --git a/backend/src/crawler/daemon.rs b/backend/src/crawler/daemon.rs index 9d949b4..470ddc8 100644 --- a/backend/src/crawler/daemon.rs +++ b/backend/src/crawler/daemon.rs @@ -46,7 +46,7 @@ use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; use crate::crawler::content::SyncOutcome; -use crate::crawler::jobs::{self, JobPayload, Lease, KIND_SYNC_CHAPTER_CONTENT}; +use crate::crawler::jobs::{self, JobPayload, Lease, KIND_SYNC_CHAPTER_CONTENT, KIND_SYNC_MANGA}; use crate::crawler::pipeline; use crate::crawler::status::{Phase, StatusHandle}; @@ -76,6 +76,15 @@ pub trait ChapterDispatcher: Send + Sync { async fn dispatch(&self, payload: JobPayload) -> anyhow::Result; } +/// A full list-only walk that enqueues mangas missing from the DB as +/// `SyncManga` jobs. Triggered on demand from the admin `/reconcile` +/// endpoint (not on the cron). Mirrors [`MetadataPass`] so the endpoint can +/// hold a `dyn` and tests can stub it. +#[async_trait] +pub trait ReconcilePass: Send + Sync { + async fn run(&self) -> anyhow::Result; +} + /// Configuration for [`spawn`]. Use `None` for `metadata_pass` to disable /// the cron entirely (worker-pool-only mode — useful when only the /// bookmark-triggered enqueue path is wanted). @@ -342,9 +351,9 @@ impl WorkerContext { _ = self.cancel.cancelled() => return, } } - let leases = match jobs::lease( + let leases = match jobs::lease_kinds( &self.pool, - Some(KIND_SYNC_CHAPTER_CONTENT), + &[KIND_SYNC_CHAPTER_CONTENT, KIND_SYNC_MANGA], 1, LEASE_DURATION, ) diff --git a/backend/src/crawler/jobs.rs b/backend/src/crawler/jobs.rs index a99ff6e..eb0bbc0 100644 --- a/backend/src/crawler/jobs.rs +++ b/backend/src/crawler/jobs.rs @@ -15,11 +15,18 @@ use uuid::Uuid; #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum JobPayload { - /// Fetch one manga's detail page, upsert metadata, enqueue - /// `SyncChapterList`. + /// Fetch one manga's detail page, upsert metadata, sync its chapter + /// list. `url` and `title` are carried from the list ref so the worker + /// can reconstruct a `SourceMangaRef` without re-walking, and so the + /// dead-jobs/history UI can render a label even before any manga row + /// exists. `#[serde(default)]` keeps older/hand-inserted rows decodable. SyncManga { source_id: String, source_manga_key: String, + #[serde(default)] + url: String, + #[serde(default)] + title: String, }, /// Diff the chapter list, enqueue `SyncChapterContent` for new /// chapters, soft-drop vanished ones. @@ -61,6 +68,11 @@ pub enum JobState { /// without re-spelling the literal. pub const KIND_SYNC_CHAPTER_CONTENT: &str = "sync_chapter_content"; +/// Kind discriminator for manga-detail sync jobs (used by the reconcile +/// pass to enqueue missing mangas). The crawl worker leases this alongside +/// `KIND_SYNC_CHAPTER_CONTENT`; both serialize on the single browser. +pub const KIND_SYNC_MANGA: &str = "sync_manga"; + /// Kind discriminator for AI page-analysis jobs. The analysis daemon /// leases with this filter so it never contends with crawl jobs. pub const KIND_ANALYZE_PAGE: &str = "analyze_page"; @@ -174,6 +186,51 @@ pub async fn lease( .fetch_all(pool) .await?; + decode_leases(rows) +} + +/// Like [`lease`] but matches any of several `payload->>'kind'` values. The +/// crawl worker uses this to drain both `sync_chapter_content` and +/// `sync_manga` from one loop (both serialize on the single browser); the +/// analysis daemon keeps using single-kind [`lease`] so it never contends. +pub async fn lease_kinds( + pool: &PgPool, + kinds: &[&str], + max: i64, + lease_duration: Duration, +) -> sqlx::Result> { + let lease_ms: i64 = lease_duration.as_millis().min(i64::MAX as u128) as i64; + let kinds_vec: Vec = kinds.iter().map(|s| s.to_string()).collect(); + let rows: Vec<(Uuid, serde_json::Value, i32, i32)> = sqlx::query_as( + r#" + WITH leased AS ( + SELECT id FROM crawler_jobs + WHERE (state = 'pending' OR (state = 'running' AND leased_until < now())) + AND scheduled_at <= now() + AND payload->>'kind' = ANY($1) + ORDER BY scheduled_at, created_at + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + UPDATE crawler_jobs j + SET state = 'running', + attempts = j.attempts + 1, + leased_until = now() + ($3::bigint || ' milliseconds')::interval, + updated_at = now() + FROM leased l + WHERE j.id = l.id + RETURNING j.id, j.payload, j.attempts, j.max_attempts + "#, + ) + .bind(&kinds_vec) + .bind(max) + .bind(lease_ms) + .fetch_all(pool) + .await?; + decode_leases(rows) +} + +fn decode_leases(rows: Vec<(Uuid, serde_json::Value, i32, i32)>) -> sqlx::Result> { let mut leases = Vec::with_capacity(rows.len()); for (id, payload_json, attempts, max_attempts) in rows { let payload: JobPayload = serde_json::from_value(payload_json).map_err(|e| { @@ -393,6 +450,60 @@ mod tests { } } + #[test] + fn sync_manga_payload_round_trips_with_url_and_title() { + let payload = JobPayload::SyncManga { + source_id: "target".into(), + source_manga_key: "foo".into(), + url: "https://target.example/manga/foo".into(), + title: "Foo Title".into(), + }; + let json = serde_json::to_value(&payload).unwrap(); + assert_eq!(json["kind"], KIND_SYNC_MANGA); + assert_eq!(json["source_id"], "target"); + assert_eq!(json["source_manga_key"], "foo"); + assert_eq!(json["url"], "https://target.example/manga/foo"); + assert_eq!(json["title"], "Foo Title"); + + match serde_json::from_value::(json).unwrap() { + JobPayload::SyncManga { + source_id, + source_manga_key, + url, + title, + } => { + assert_eq!(source_id, "target"); + assert_eq!(source_manga_key, "foo"); + assert_eq!(url, "https://target.example/manga/foo"); + assert_eq!(title, "Foo Title"); + } + other => panic!("expected SyncManga, got {other:?}"), + } + } + + #[test] + fn sync_manga_payload_defaults_missing_url_and_title() { + // A row that predates the url/title fields must still decode. + let legacy = serde_json::json!({ + "kind": "sync_manga", + "source_id": "target", + "source_manga_key": "foo", + }); + match serde_json::from_value::(legacy).unwrap() { + JobPayload::SyncManga { + url, + title, + source_manga_key, + .. + } => { + assert_eq!(source_manga_key, "foo"); + assert_eq!(url, ""); + assert_eq!(title, ""); + } + other => panic!("expected SyncManga, got {other:?}"), + } + } + #[test] fn backoff_base_grows_exponentially_and_caps_at_one_hour() { // attempts == 1 → 60s, doubling each step. diff --git a/backend/src/crawler/mod.rs b/backend/src/crawler/mod.rs index a3465f6..81167ba 100644 --- a/backend/src/crawler/mod.rs +++ b/backend/src/crawler/mod.rs @@ -23,6 +23,7 @@ pub mod jobs; pub mod nav; pub mod pipeline; pub mod rate_limit; +pub mod reconcile; pub mod resync; pub mod safety; pub mod session; diff --git a/backend/src/crawler/pipeline.rs b/backend/src/crawler/pipeline.rs index 4d7fceb..abfe910 100644 --- a/backend/src/crawler/pipeline.rs +++ b/backend/src/crawler/pipeline.rs @@ -76,6 +76,213 @@ pub(crate) fn should_abort_pass(consecutive: u32, threshold: u32) -> bool { threshold > 0 && consecutive >= threshold } +/// Success outcome of [`process_manga_ref`]. +pub(crate) struct RefProcessed { + pub manga_id: Uuid, + pub status: UpsertStatus, + pub chapters_new: Option, + pub cover_fetched: bool, +} + +/// Why [`process_manga_ref`] produced no upsert. +pub(crate) enum RefError { + /// `fetch_manga` itself failed. Counts toward the consecutive-failure + /// breaker in the metadata pass; the `SyncManga` worker treats it as a + /// retryable job failure. + Fetch(anyhow::Error), + /// The detail fetched but the ref was skipped — partial-render guard or + /// an upsert error. The fetch *succeeded*, so the breaker resets, but no + /// manga was upserted. + Skip(anyhow::Error), +} + +/// Run fetch → partial-render guard → upsert → cover → chapter-sync for a +/// single discovered ref. Extracted from [`run_metadata_pass`] so the +/// `SyncManga` worker drives the identical per-manga work and the two paths +/// can't drift. Records the detail/cover crawl metrics internally; the +/// caller owns the discovered/seen/stop/breaker bookkeeping. +#[allow(clippy::too_many_arguments)] +pub(crate) async fn process_manga_ref( + ctx: &FetchContext<'_>, + source: &TargetSource, + db: &PgPool, + storage: &dyn Storage, + http: &reqwest::Client, + rate: &HostRateLimiters, + r: &SourceMangaRef, + skip_chapters: bool, + allowlist: &DownloadAllowlist, + max_image_bytes: usize, + status: Option<&crate::crawler::status::StatusHandle>, +) -> Result { + let source_id = source.id(); + let detail_started = std::time::Instant::now(); + let manga = match source.fetch_manga(ctx, r).await { + Ok(m) => m, + Err(e) => { + // Detail-fetch timing (failed). manga_id is unknown — the manga + // was never upserted. + let _ = repo::crawl_metrics::record( + db, + repo::crawl_metrics::OP_MANGA_DETAIL, + None, + None, + "failed", + detail_started.elapsed().as_millis() as i64, + None, + Some(&format!("{e:#}")), + ) + .await; + return Err(RefError::Fetch(e)); + } + }; + // Detail fetch succeeded; stash its duration to record once the + // manga_id is known (after upsert). + let detail_ms = detail_started.elapsed().as_millis() as i64; + + // Partial-render guard: an empty chapter list paired with a prior count + // > 0 is overwhelmingly a chromium snapshot taken between the wrapper + // render and its rows render. Treat as a transient skip (no upsert) so + // a later attempt retries. Skipped in `skip_chapters` mode because the + // parser returns an empty Vec there by design. + if !skip_chapters && manga.chapters.is_empty() { + match repo::crawler::live_chapter_count_for_source_manga(db, source_id, &r.source_manga_key) + .await + { + Ok(prior) if prior > 0 => { + return Err(RefError::Skip(anyhow::anyhow!( + "fetch_manga returned empty chapters but prior count {prior} > 0; \ + treating as partial-render transient" + ))); + } + Ok(_) => {} + Err(e) => { + // DB lookup failed — fail safe: skip rather than risk a + // soft-drop on a manga whose prior count we couldn't confirm. + return Err(RefError::Skip( + anyhow::Error::new(e).context("live_chapter_count_for_source_manga"), + )); + } + } + } + + let upsert = match repo::crawler::upsert_manga_from_source(db, source_id, &r.url, &manga).await { + Ok(u) => u, + Err(e) => { + return Err(RefError::Skip( + anyhow::Error::new(e).context("upsert_manga_from_source"), + )); + } + }; + // Detail-fetch timing (ok), now that we have the manga_id. + let _ = repo::crawl_metrics::record( + db, + repo::crawl_metrics::OP_MANGA_DETAIL, + Some(upsert.manga_id), + None, + "ok", + detail_ms, + None, + None, + ) + .await; + tracing::info!( + key = %manga.source_manga_key, + manga_id = %upsert.manga_id, + status = ?upsert.status, + title = %manga.title, + "manga upserted" + ); + + // Cover image: download when missing in storage or when metadata + // signaled an update (cover URL is part of metadata_hash, so Updated + // implies the URL may have moved). Failures are non-fatal. + let mut cover_fetched = false; + let needs_cover = + upsert.cover_image_path.is_none() || matches!(upsert.status, UpsertStatus::Updated); + if needs_cover { + if let Some(cover_url) = manga.cover_url.as_deref() { + // RAII: the guard clears `current_cover` on every exit path. + let _cover_guard = status.map(|s| { + s.begin_cover(crate::crawler::status::CoverTarget { + manga_id: upsert.manga_id, + manga_title: manga.title.clone(), + }) + }); + let cover_started = std::time::Instant::now(); + let cover_result = download_and_store_cover( + db, + storage, + http, + rate, + &r.url, + upsert.manga_id, + cover_url, + allowlist, + max_image_bytes, + ) + .await; + let cover_ms = cover_started.elapsed().as_millis() as i64; + let _ = repo::crawl_metrics::record( + db, + repo::crawl_metrics::OP_MANGA_COVER, + Some(upsert.manga_id), + None, + if cover_result.is_ok() { "ok" } else { "failed" }, + cover_ms, + None, + cover_result.as_ref().err().map(|e| format!("{e:#}")).as_deref(), + ) + .await; + match cover_result { + Ok(()) => cover_fetched = true, + Err(e) => tracing::warn!( + manga_id = %upsert.manga_id, + error = ?e, + "cover download failed" + ), + } + } + } + + // Chapter sync. `None` (skip_chapters mode, or a logged-and-swallowed + // sync error) refuses to stop on this manga because we can't confirm + // "no new chapters." + let chapters_new: Option = if skip_chapters { + None + } else { + match repo::crawler::sync_manga_chapters(db, source_id, upsert.manga_id, &manga.chapters) + .await + { + Ok(diff) => { + tracing::info!( + manga_id = %upsert.manga_id, + new = diff.new, + refreshed = diff.refreshed, + dropped = diff.dropped, + "chapters synced" + ); + Some(diff.new) + } + Err(e) => { + tracing::warn!( + manga_id = %upsert.manga_id, + error = ?e, + "chapter sync failed" + ); + None + } + } + }; + + Ok(RefProcessed { + manga_id: upsert.manga_id, + status: upsert.status, + chapters_new, + cover_fetched, + }) +} + /// Runs the discover → fetch → upsert → cover → chapter-list-diff pipeline /// for the target source. Pure metadata; chapter content is enqueued as /// separate `SyncChapterContent` jobs by the caller after this returns. @@ -244,32 +451,48 @@ pub async fn run_metadata_pass( key = %r.source_manga_key, "fetching metadata" ); - let detail_started = std::time::Instant::now(); - let manga = match source.fetch_manga(&ctx, &r).await { - Ok(m) => { + match process_manga_ref( + &ctx, + &source, + db, + storage, + http, + rate, + &r, + skip_chapters, + allowlist, + max_image_bytes, + status, + ) + .await + { + Ok(p) => { consecutive_failures = 0; - m + stats.upserted += 1; + if p.cover_fetched { + stats.covers_fetched += 1; + } + // Record success in the dedup set. Cover and chapter-sync + // failures inside process_manga_ref are non-fatal and + // don't roll this back — metadata is the durable source + // of truth for the dedup. + seen.insert(r.source_manga_key.clone()); + if should_stop(was_clean, p.status, p.chapters_new) { + hit_stop_condition = true; + tracing::info!( + key = %r.source_manga_key, + "stop condition met (Unchanged metadata + 0 new chapters); halting walk" + ); + break 'outer; + } } - Err(e) => { + Err(RefError::Fetch(e)) => { tracing::warn!( key = %r.source_manga_key, url = %r.url, error = ?e, "fetch_manga failed" ); - // Detail-fetch timing (failed). manga_id is unknown — the - // manga was never upserted. - let _ = repo::crawl_metrics::record( - db, - repo::crawl_metrics::OP_MANGA_DETAIL, - None, - None, - "failed", - detail_started.elapsed().as_millis() as i64, - None, - Some(&format!("{e:#}")), - ) - .await; stats.mangas_failed += 1; consecutive_failures += 1; if should_abort_pass(consecutive_failures, max_consecutive_failures) { @@ -282,191 +505,19 @@ pub async fn run_metadata_pass( ); break 'outer; } - continue; } - }; - // Detail fetch succeeded; stash its duration to record once the - // manga_id is known (after upsert). - let detail_ms = detail_started.elapsed().as_millis() as i64; - - // Partial-render guard: an empty chapter list paired with a - // prior count > 0 is overwhelmingly a chromium snapshot - // taken between the #chapter_table wrapper render and its - // rows render. The wait_for_selector wait in `navigate` - // narrows this window but cannot close it for slow renders - // beyond the selector budget. Treat as a transient failure - // here — skip upsert, skip seen.insert — so the next batch - // (or the next tick) retries. Skipped in `skip_chapters` - // mode because the parser is configured to return an empty - // Vec by design there. - if !skip_chapters && manga.chapters.is_empty() { - match repo::crawler::live_chapter_count_for_source_manga( - db, source_id, &r.source_manga_key, - ) - .await - { - Ok(prior) if prior > 0 => { - tracing::warn!( - key = %r.source_manga_key, - url = %r.url, - prior_chapter_count = prior, - "fetch_manga returned empty chapters but prior count > 0; treating as partial-render transient and skipping" - ); - stats.mangas_failed += 1; - continue; - } - Ok(_) => {} - Err(e) => { - // DB lookup failed — fail safe: skip rather - // than risk a soft-drop on a manga whose prior - // count we couldn't confirm. - tracing::warn!( - key = %r.source_manga_key, - error = ?e, - "live_chapter_count_for_source_manga failed; skipping cautiously" - ); - stats.mangas_failed += 1; - continue; - } - } - } - - let upsert = match repo::crawler::upsert_manga_from_source( - db, source_id, &r.url, &manga, - ) - .await - { - Ok(u) => u, - Err(e) => { - tracing::error!( + Err(RefError::Skip(e)) => { + // Fetch succeeded (so the breaker resets) but the ref was + // skipped — partial render or upsert error. Left out of + // `seen` so a reappearance in a later batch retries. + tracing::warn!( key = %r.source_manga_key, error = ?e, - "upsert_manga_from_source failed" + "manga ref skipped (partial-render or upsert error)" ); + consecutive_failures = 0; stats.mangas_failed += 1; - continue; } - }; - stats.upserted += 1; - // Detail-fetch timing (ok), now that we have the manga_id. - let _ = repo::crawl_metrics::record( - db, - repo::crawl_metrics::OP_MANGA_DETAIL, - Some(upsert.manga_id), - None, - "ok", - detail_ms, - None, - None, - ) - .await; - // Record success in the dedup set. Cover and chapter-sync - // failures below are non-fatal and don't roll this back — - // metadata is the durable source of truth for the dedup. - seen.insert(r.source_manga_key.clone()); - tracing::info!( - key = %manga.source_manga_key, - manga_id = %upsert.manga_id, - status = ?upsert.status, - title = %manga.title, - "manga upserted" - ); - - // Cover image: download when missing in storage or when metadata - // signaled an update (cover URL is part of metadata_hash, so - // Updated implies the URL may have moved). Failures are non-fatal. - let needs_cover = upsert.cover_image_path.is_none() - || matches!(upsert.status, repo::crawler::UpsertStatus::Updated); - if needs_cover { - if let Some(cover_url) = manga.cover_url.as_deref() { - // RAII: the guard clears `current_cover` on every - // exit path (success, panic, future early-return). - // Mirrors the chapter-side ChapterGuard. - let _cover_guard = status.map(|s| { - s.begin_cover(crate::crawler::status::CoverTarget { - manga_id: upsert.manga_id, - manga_title: manga.title.clone(), - }) - }); - let cover_started = std::time::Instant::now(); - let cover_result = download_and_store_cover( - db, - storage, - http, - rate, - &r.url, - upsert.manga_id, - cover_url, - allowlist, - max_image_bytes, - ) - .await; - let cover_ms = cover_started.elapsed().as_millis() as i64; - let _ = repo::crawl_metrics::record( - db, - repo::crawl_metrics::OP_MANGA_COVER, - Some(upsert.manga_id), - None, - if cover_result.is_ok() { "ok" } else { "failed" }, - cover_ms, - None, - cover_result.as_ref().err().map(|e| format!("{e:#}")).as_deref(), - ) - .await; - match cover_result { - Ok(()) => stats.covers_fetched += 1, - Err(e) => tracing::warn!( - manga_id = %upsert.manga_id, - error = ?e, - "cover download failed" - ), - } - } - } - - // Chapter sync. `chapters_new` feeds the stop check below: - // `None` (skip_chapters mode, or a logged-and-swallowed sync - // error) refuses to stop on this manga because we can't - // confirm "no new chapters." - let chapters_new: Option = if skip_chapters { - None - } else { - match repo::crawler::sync_manga_chapters( - db, - source_id, - upsert.manga_id, - &manga.chapters, - ) - .await - { - Ok(diff) => { - tracing::info!( - manga_id = %upsert.manga_id, - new = diff.new, - refreshed = diff.refreshed, - dropped = diff.dropped, - "chapters synced" - ); - Some(diff.new) - } - Err(e) => { - tracing::warn!( - manga_id = %upsert.manga_id, - error = ?e, - "chapter sync failed" - ); - None - } - } - }; - - if should_stop(was_clean, upsert.status, chapters_new) { - hit_stop_condition = true; - tracing::info!( - key = %manga.source_manga_key, - "stop condition met (Unchanged metadata + 0 new chapters); halting walk" - ); - break 'outer; } } } diff --git a/backend/src/crawler/reconcile.rs b/backend/src/crawler/reconcile.rs new file mode 100644 index 0000000..316ee1d --- /dev/null +++ b/backend/src/crawler/reconcile.rs @@ -0,0 +1,260 @@ +//! Reconcile pass — find mangas the metadata pass missed. +//! +//! The interleaved metadata pass ([`crate::crawler::pipeline`]) walks the +//! source list newest-first and visits each manga inline, which makes it +//! vulnerable to list drift: a manga can slip a pagination slot while the +//! walk spends minutes on detail work, and never get upserted. +//! +//! Reconcile fixes that with a cheap, full, unconditional list-only walk +//! (refs only — no `fetch_manga`, no early stop), set-diffs the walked keys +//! against `manga_sources`, and enqueues the strictly-missing ones as +//! `SyncManga` jobs for the crawl worker to visit. A set-diff is immune to +//! drift: a manga only has to appear *somewhere* in the list, not survive a +//! specific slot during a slow walk. +//! +//! Identity matches the DB exactly because both sides derive the key the +//! same way: the walker's refs carry `source_manga_key = +//! derive_key_from_url(list_href)`, which is precisely what +//! `manga_sources.source_manga_key` stores. + +use std::collections::HashSet; + +use anyhow::Context; +use sqlx::PgPool; + +use crate::crawler::browser_manager::BrowserManager; +use crate::crawler::jobs::{self, EnqueueResult, JobPayload}; +use crate::crawler::rate_limit::HostRateLimiters; +use crate::crawler::source::target::TargetSource; +use crate::crawler::source::{FetchContext, Source, SourceMangaRef}; +use crate::crawler::status::Phase; +use crate::repo; +use crate::repo::crawler::ensure_source; + +/// Counters surfaced when a reconcile pass finishes. +#[derive(Debug, Default, Clone, Copy)] +pub struct ReconcileStats { + /// Distinct refs collected from the full list walk. + pub walked: usize, + /// Walked keys with no `manga_sources` row (strict `NOT EXISTS`). + pub missing: usize, + /// Missing mangas newly enqueued as `SyncManga` jobs this pass. + pub enqueued: usize, + /// Missing mangas skipped because a blocking (pending/running/dead) + /// `SyncManga` job already exists, or the enqueue lost an insert race. + pub skipped: usize, +} + +/// Pure set-diff: which walked refs should be enqueued. A ref is selected +/// when its key is neither already in the DB (`existing`) nor already has a +/// blocking job (`blocked`). Repeated keys in `walked` (source index drift +/// can surface the same manga twice) are de-duplicated — the first ref wins. +pub(crate) fn select_missing<'a>( + walked: &'a [SourceMangaRef], + existing: &HashSet, + blocked: &HashSet, +) -> Vec<&'a SourceMangaRef> { + let mut seen = HashSet::new(); + let mut out = Vec::new(); + for r in walked { + if existing.contains(&r.source_manga_key) || blocked.contains(&r.source_manga_key) { + continue; + } + if seen.insert(r.source_manga_key.clone()) { + out.push(r); + } + } + out +} + +/// Run a full list-only walk, diff against the DB, and enqueue missing +/// mangas as `SyncManga` jobs. Holds the (exclusive) browser lease only for +/// the cheap walk; the enqueued jobs are drained later by the crawl worker. +#[allow(clippy::too_many_arguments)] +pub async fn reconcile_missing( + browser_manager: &BrowserManager, + db: &PgPool, + rate: &HostRateLimiters, + start_url: &str, + status: Option<&crate::crawler::status::StatusHandle>, + tor: Option<&crate::crawler::tor::TorController>, +) -> anyhow::Result { + let lease = browser_manager + .acquire() + .await + .context("acquire browser lease for reconcile pass")?; + let browser_ref: &chromiumoxide::Browser = &lease; + if let Some(s) = status { + s.set_phase(Phase::Reconciling { + walked: 0, + enqueued: 0, + }) + .await; + } + + // Chapter parsing is irrelevant here (we never call fetch_manga); the + // metadata-only constructor is intent-revealing and harmless. + let source = TargetSource::new(start_url.to_string()).without_chapter_parsing(); + let ctx = FetchContext { + browser: browser_ref, + rate, + tor, + }; + let source_id = source.id(); + ensure_source( + db, + source_id, + "Target Site", + &crate::crawler::url_utils::origin_of(start_url).unwrap_or_else(|| start_url.to_string()), + ) + .await + .context("ensure_source")?; + + tracing::info!("starting reconcile pass (full list-only walk)"); + let mut walker = source.discover(&ctx).await.context("discover failed")?; + + // Full unconditional walk: collect every ref, no early stop. Yield the + // lease if a coordinated browser restart is pending so the drain isn't + // stalled — the missing set we've gathered so far is still enqueued. + let mut walked: Vec = Vec::new(); + 'outer: loop { + if browser_manager.is_restart_pending() { + tracing::info!("reconcile pass: browser restart pending — yielding partial walk"); + break; + } + match walker.next_batch(&ctx).await? { + Some(batch) => { + for r in batch { + walked.push(r); + } + if let Some(s) = status { + s.set_phase(Phase::Reconciling { + walked: walked.len(), + enqueued: 0, + }) + .await; + } + if browser_manager.is_restart_pending() { + tracing::info!( + "reconcile pass: browser restart pending mid-batch — yielding partial walk" + ); + break 'outer; + } + } + None => break, + } + } + // Browser work is done — release before the (browser-free) diff+enqueue. + drop(lease); + + let existing = repo::crawler::existing_source_keys(db, source_id) + .await + .context("existing_source_keys")?; + let blocked = repo::crawler::sync_manga_keys_with_blocking_job(db, source_id) + .await + .context("sync_manga_keys_with_blocking_job")?; + + let missing = select_missing(&walked, &existing, &blocked); + + let mut stats = ReconcileStats { + walked: walked.len(), + missing: missing.len(), + ..Default::default() + }; + for r in missing { + let payload = JobPayload::SyncManga { + source_id: source_id.to_string(), + source_manga_key: r.source_manga_key.clone(), + url: r.url.clone(), + title: r.title.clone(), + }; + match jobs::enqueue(db, &payload).await { + Ok(EnqueueResult::Inserted(_)) => stats.enqueued += 1, + Ok(EnqueueResult::Skipped) => stats.skipped += 1, + Err(e) => { + tracing::warn!(key = %r.source_manga_key, error = ?e, "enqueue SyncManga failed"); + stats.skipped += 1; + } + } + if let Some(s) = status { + s.set_phase(Phase::Reconciling { + walked: stats.walked, + enqueued: stats.enqueued, + }) + .await; + } + } + + tracing::info!( + walked = stats.walked, + missing = stats.missing, + enqueued = stats.enqueued, + skipped = stats.skipped, + "reconcile pass complete" + ); + Ok(stats) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn r(key: &str) -> SourceMangaRef { + SourceMangaRef { + source_manga_key: key.to_string(), + title: format!("Title {key}"), + url: format!("https://target.example/manga/{key}"), + } + } + + fn set(keys: &[&str]) -> HashSet { + keys.iter().map(|s| s.to_string()).collect() + } + + #[test] + fn select_missing_returns_keys_not_in_manga_sources() { + let walked = vec![r("a"), r("b"), r("c")]; + let existing = set(&["b"]); + let blocked = set(&[]); + let got: Vec<&str> = select_missing(&walked, &existing, &blocked) + .iter() + .map(|r| r.source_manga_key.as_str()) + .collect(); + assert_eq!(got, vec!["a", "c"]); + } + + #[test] + fn select_missing_treats_dropped_rows_as_present() { + // The caller passes dropped keys in `existing` (existing_source_keys + // does not filter dropped_at), so a dropped manga is NOT re-enqueued. + let walked = vec![r("a"), r("dropped")]; + let existing = set(&["dropped"]); + let got: Vec<&str> = select_missing(&walked, &existing, &set(&[])) + .iter() + .map(|r| r.source_manga_key.as_str()) + .collect(); + assert_eq!(got, vec!["a"]); + } + + #[test] + fn select_missing_skips_keys_with_blocking_job() { + let walked = vec![r("a"), r("queued"), r("dead")]; + let blocked = set(&["queued", "dead"]); + let got: Vec<&str> = select_missing(&walked, &set(&[]), &blocked) + .iter() + .map(|r| r.source_manga_key.as_str()) + .collect(); + assert_eq!(got, vec!["a"]); + } + + #[test] + fn select_missing_dedups_repeated_walked_keys() { + // Index drift can surface the same manga twice in one walk. + let walked = vec![r("a"), r("a"), r("b")]; + let got: Vec<&str> = select_missing(&walked, &set(&[]), &set(&[])) + .iter() + .map(|r| r.source_manga_key.as_str()) + .collect(); + assert_eq!(got, vec!["a", "b"]); + } +} diff --git a/backend/src/crawler/status.rs b/backend/src/crawler/status.rs index 5875822..9c8fffe 100644 --- a/backend/src/crawler/status.rs +++ b/backend/src/crawler/status.rs @@ -41,6 +41,10 @@ pub enum Phase { /// Backfilling covers that failed on first attempt. `index`/`total` /// track progress through this tick's batch. CoverBackfill { index: usize, total: usize }, + /// Reconcile pass: walking the full source list (refs only, no detail + /// visit) to find mangas missing from the DB. `walked` is refs seen so + /// far this walk; `enqueued` is missing mangas queued as `SyncManga`. + Reconciling { walked: usize, enqueued: usize }, } /// A chapter being downloaded right now, with a live page count. Keyed in diff --git a/backend/src/repo/crawler.rs b/backend/src/repo/crawler.rs index 06c9993..5fb7261 100644 --- a/backend/src/repo/crawler.rs +++ b/backend/src/repo/crawler.rs @@ -619,6 +619,48 @@ pub async fn last_run_completed_cleanly( .unwrap_or(true)) } +// --------------------------------------------------------------------------- +// Reconcile: find source-list mangas missing from the DB. +// --------------------------------------------------------------------------- + +/// All `source_manga_key`s we have a `manga_sources` row for under +/// `source_id`. Intentionally **not** filtered by `dropped_at`: a +/// soft-dropped row still counts as "present" so reconcile only enqueues +/// mangas it has truly never seen (strict `NOT EXISTS`). +pub async fn existing_source_keys( + pool: &PgPool, + source_id: &str, +) -> sqlx::Result> { + let rows: Vec = + sqlx::query_scalar("SELECT source_manga_key FROM manga_sources WHERE source_id = $1") + .bind(source_id) + .fetch_all(pool) + .await?; + Ok(rows.into_iter().collect()) +} + +/// `source_manga_key`s that already have a `SyncManga` job in a state that +/// should block re-enqueue: `pending`/`running` (in flight) **and** `dead` +/// (leave-dead — a gone manga is not retried by a later reconcile). `done` +/// and `failed` (mid-backoff) do not block. +pub async fn sync_manga_keys_with_blocking_job( + pool: &PgPool, + source_id: &str, +) -> sqlx::Result> { + let rows: Vec = sqlx::query_scalar( + "SELECT DISTINCT payload->>'source_manga_key' \ + FROM crawler_jobs \ + WHERE payload->>'kind' = 'sync_manga' \ + AND payload->>'source_id' = $1 \ + AND state IN ('pending', 'running', 'dead') \ + AND payload->>'source_manga_key' IS NOT NULL", + ) + .bind(source_id) + .fetch_all(pool) + .await?; + Ok(rows.into_iter().collect()) +} + // --------------------------------------------------------------------------- // Dead-letter jobs: admin observability + requeue. // --------------------------------------------------------------------------- @@ -635,6 +677,14 @@ pub struct DeadJob { pub manga_id: Option, pub manga_title: Option, pub chapter_number: Option, + /// Title carried in the job payload. For `sync_manga` jobs whose manga + /// was never upserted there is no `manga_id`/`manga_title`, so the UI + /// falls back to this. + pub payload_title: Option, + /// Source detail URL carried in the payload (currently `sync_manga`). + pub source_url: Option, + /// Source-native key carried in the payload (currently `sync_manga`). + pub source_key: Option, pub attempts: i32, pub max_attempts: i32, pub last_error: Option, @@ -663,6 +713,9 @@ pub async fn list_dead_jobs( c.manga_id AS manga_id, m.title AS manga_title, c.number AS chapter_number, + cj.payload->>'title' AS payload_title, + cj.payload->>'url' AS source_url, + cj.payload->>'source_manga_key' AS source_key, cj.attempts, cj.max_attempts, cj.last_error, @@ -671,7 +724,7 @@ pub async fn list_dead_jobs( LEFT JOIN chapters c ON c.id = (cj.payload->>'chapter_id')::uuid LEFT JOIN mangas m ON m.id = c.manga_id WHERE cj.state = 'dead' - AND ($1::text IS NULL OR m.title ILIKE $1) + AND ($1::text IS NULL OR m.title ILIKE $1 OR cj.payload->>'title' ILIKE $1) ORDER BY cj.updated_at DESC LIMIT $2 OFFSET $3 "#, @@ -689,7 +742,7 @@ pub async fn list_dead_jobs( LEFT JOIN chapters c ON c.id = (cj.payload->>'chapter_id')::uuid LEFT JOIN mangas m ON m.id = c.manga_id WHERE cj.state = 'dead' - AND ($1::text IS NULL OR m.title ILIKE $1) + AND ($1::text IS NULL OR m.title ILIKE $1 OR cj.payload->>'title' ILIKE $1) "#, ) .bind(&search_pat) @@ -799,6 +852,11 @@ pub struct JobHistoryRow { pub page_number: Option, /// Source-side key, the only target a `sync_manga` job carries. pub source_key: Option, + /// Title carried in the payload — the display fallback for `sync_manga` + /// jobs whose manga was never upserted (no `manga_title`). + pub payload_title: Option, + /// Source detail URL carried in the payload (currently `sync_manga`). + pub source_url: Option, pub attempts: i32, pub max_attempts: i32, pub last_error: Option, @@ -855,6 +913,8 @@ pub async fn list_job_history( c.number AS chapter_number, pg.page_number AS page_number, cj.payload->>'source_manga_key' AS source_key, + cj.payload->>'title' AS payload_title, + cj.payload->>'url' AS source_url, cj.attempts, cj.max_attempts, cj.last_error, @@ -873,7 +933,7 @@ pub async fn list_job_history( ) cm ON true WHERE ($1::text IS NULL OR cj.state = $1) AND ($2::text IS NULL OR cj.payload->>'kind' = $2) - AND ($3::text IS NULL OR m.title ILIKE $3) + AND ($3::text IS NULL OR m.title ILIKE $3 OR cj.payload->>'title' ILIKE $3) ORDER BY cj.updated_at DESC LIMIT $4 OFFSET $5 "#, @@ -895,7 +955,7 @@ pub async fn list_job_history( LEFT JOIN mangas m ON m.id = COALESCE(c.manga_id, (cj.payload->>'manga_id')::uuid) WHERE ($1::text IS NULL OR cj.state = $1) AND ($2::text IS NULL OR cj.payload->>'kind' = $2) - AND ($3::text IS NULL OR m.title ILIKE $3) + AND ($3::text IS NULL OR m.title ILIKE $3 OR cj.payload->>'title' ILIKE $3) "#, ) .bind(filter.state) diff --git a/backend/tests/analysis_worker.rs b/backend/tests/analysis_worker.rs index ce425e5..cfe9af1 100644 --- a/backend/tests/analysis_worker.rs +++ b/backend/tests/analysis_worker.rs @@ -317,6 +317,8 @@ async fn worker_ignores_non_analyze_jobs(pool: PgPool) { &JobPayload::SyncManga { source_id: "s".into(), source_manga_key: "k".into(), + url: "https://target.example/manga/k".into(), + title: "K".into(), }, ) .await diff --git a/backend/tests/api_admin_crawler.rs b/backend/tests/api_admin_crawler.rs index 5e31d75..13ecd9c 100644 --- a/backend/tests/api_admin_crawler.rs +++ b/backend/tests/api_admin_crawler.rs @@ -206,6 +206,7 @@ async fn control_endpoints_return_503_when_daemon_disabled(pool: PgPool) { let cookie = seed_admin(&pool, &h.app).await; for uri in [ "/api/v1/admin/crawler/run", + "/api/v1/admin/crawler/reconcile", "/api/v1/admin/crawler/browser/restart", "/api/v1/admin/crawler/session/clear-expired", ] { @@ -779,6 +780,45 @@ async fn job_history_lists_filters_and_paginates(pool: PgPool) { assert_eq!(resp.status(), StatusCode::FORBIDDEN); } +#[sqlx::test(migrations = "./migrations")] +async fn job_history_surfaces_sync_manga_payload_title(pool: PgPool) { + // A reconcile-enqueued sync_manga job that died has no manga row, so the + // history row must fall back to the payload title/url and be searchable. + sqlx::query("INSERT INTO crawler_jobs (id, payload, state, last_error) VALUES ($1, $2, 'dead', 'broken-page body signature')") + .bind(Uuid::new_v4()) + .bind(json!({ + "kind": "sync_manga", + "source_id": "target", + "source_manga_key": "gone-1", + "url": "http://x/manga/gone-1", + "title": "Ghost Manga", + })) + .execute(&pool) + .await + .unwrap(); + let h = harness(pool.clone()); + let cookie = seed_admin(&pool, &h.app).await; + + let resp = h + .app + .clone() + .oneshot(get_with_cookie( + "/api/v1/admin/crawler/history?search=ghost", + &cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let body = body_json(resp).await; + assert_eq!(body["page"]["total"], 1); + let row = &body["items"][0]; + assert_eq!(row["kind"], "sync_manga"); + assert_eq!(row["manga_title"], serde_json::Value::Null); + assert_eq!(row["payload_title"], "Ghost Manga"); + assert_eq!(row["source_url"], "http://x/manga/gone-1"); + assert_eq!(row["source_key"], "gone-1"); +} + // --------------------------------------------------------------------------- // Operation metrics: durations, averages, recent-ops log. // --------------------------------------------------------------------------- diff --git a/backend/tests/crawler_daemon.rs b/backend/tests/crawler_daemon.rs index f969f94..e95ba31 100644 --- a/backend/tests/crawler_daemon.rs +++ b/backend/tests/crawler_daemon.rs @@ -91,6 +91,18 @@ impl ChapterDispatcher for PanickingDispatcher { } } +/// Always returns an error — used to drive a job to the dead-letter state. +struct FailingDispatcher { + seen: AtomicUsize, +} +#[async_trait::async_trait] +impl ChapterDispatcher for FailingDispatcher { + async fn dispatch(&self, _payload: JobPayload) -> anyhow::Result { + self.seen.fetch_add(1, Ordering::AcqRel); + anyhow::bail!("simulated detail-page failure") + } +} + /// Never completes — used to verify the worker's outer dispatch timeout. struct HangingDispatcher { seen: AtomicUsize, @@ -226,6 +238,79 @@ async fn workers_drain_jobs_through_dispatcher(pool: PgPool) { assert_eq!(count_state(&pool, "done").await, 3); } +async fn enqueue_sync_manga_job(pool: &PgPool, key: &str) -> Uuid { + let res = jobs::enqueue( + pool, + &JobPayload::SyncManga { + source_id: "target".into(), + source_manga_key: key.into(), + url: format!("http://x/{key}"), + title: format!("M {key}"), + }, + ) + .await + .unwrap(); + match res { + jobs::EnqueueResult::Inserted(id) => id, + jobs::EnqueueResult::Skipped => unreachable!("fresh key"), + } +} + +#[sqlx::test(migrations = "./migrations")] +async fn worker_leases_sync_manga_and_failure_dead_letters(pool: PgPool) { + // The crawl worker must pick up reconcile-enqueued SyncManga jobs (the + // lease_kinds change), and a failure must dead-letter (leave-dead). Set + // max_attempts=1 so a single failure is terminal — no waiting through + // exponential backoff. + let id = enqueue_sync_manga_job(&pool, "missing-1").await; + sqlx::query("UPDATE crawler_jobs SET max_attempts = 1 WHERE id = $1") + .bind(id) + .execute(&pool) + .await + .unwrap(); + + let dispatcher = Arc::new(FailingDispatcher { + 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(None, dispatcher.clone(), session_expired, 1), + ); + + for _ in 0..60 { + if count_state(&pool, "dead").await >= 1 { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + handle.shutdown().await; + + assert!( + dispatcher.seen.load(Ordering::Acquire) >= 1, + "worker must lease the sync_manga job" + ); + assert_eq!( + count_state(&pool, "dead").await, + 1, + "a failed sync_manga job dead-letters (leave-dead)" + ); + let last_error: Option = + sqlx::query_scalar("SELECT last_error FROM crawler_jobs WHERE id = $1") + .bind(id) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + last_error + .unwrap_or_default() + .contains("simulated detail-page failure"), + "dead job records the failure reason" + ); +} + #[sqlx::test(migrations = "./migrations")] async fn workers_idle_while_session_expired(pool: PgPool) { let id = enqueue_chapter_job(&pool).await; diff --git a/backend/tests/crawler_dead_jobs.rs b/backend/tests/crawler_dead_jobs.rs index 5972f4f..652eecf 100644 --- a/backend/tests/crawler_dead_jobs.rs +++ b/backend/tests/crawler_dead_jobs.rs @@ -116,6 +116,57 @@ async fn list_dead_jobs_filters_by_title_search(pool: PgPool) { assert_eq!(items[0].manga_title.as_deref(), Some("One Piece")); } +/// Insert a dead `sync_manga` job (no manga/chapter rows) — the reconcile +/// "detail page gone" case. +async fn insert_dead_sync_manga(pool: &PgPool, key: &str, title: &str) -> Uuid { + let id = Uuid::new_v4(); + let payload = json!({ + "kind": "sync_manga", + "source_id": "target", + "source_manga_key": key, + "url": format!("http://x/manga/{key}"), + "title": title, + }); + sqlx::query( + "INSERT INTO crawler_jobs (id, payload, state, attempts, last_error) \ + VALUES ($1, $2, 'dead', 5, 'broken-page body signature')", + ) + .bind(id) + .bind(payload) + .execute(pool) + .await + .unwrap(); + id +} + +#[sqlx::test(migrations = "./migrations")] +async fn dead_sync_manga_job_surfaces_payload_title_and_url(pool: PgPool) { + insert_dead_sync_manga(&pool, "gone-1", "Ghost Manga").await; + + let (items, total) = crawler::list_dead_jobs(&pool, None, 50, 0).await.unwrap(); + assert_eq!(total, 1); + let row = &items[0]; + // No manga row exists → manga_title is None; the payload fields fill in. + assert_eq!(row.manga_title, None); + assert_eq!(row.payload_title.as_deref(), Some("Ghost Manga")); + assert_eq!(row.source_url.as_deref(), Some("http://x/manga/gone-1")); + assert_eq!(row.source_key.as_deref(), Some("gone-1")); + assert_eq!(row.kind, "sync_manga"); + assert_eq!(row.last_error.as_deref(), Some("broken-page body signature")); +} + +#[sqlx::test(migrations = "./migrations")] +async fn dead_jobs_search_matches_payload_title(pool: PgPool) { + insert_dead_sync_manga(&pool, "gone-1", "Ghost Manga").await; + insert_dead_sync_manga(&pool, "gone-2", "Other Title").await; + + let (items, total) = crawler::list_dead_jobs(&pool, Some("ghost"), 50, 0) + .await + .unwrap(); + assert_eq!(total, 1); + assert_eq!(items[0].payload_title.as_deref(), Some("Ghost Manga")); +} + #[sqlx::test(migrations = "./migrations")] async fn requeue_all_resets_dead_jobs_to_pending(pool: PgPool) { let (_m, c1) = seed_chapter(&pool, "A", 1).await; diff --git a/backend/tests/crawler_jobs.rs b/backend/tests/crawler_jobs.rs index 0b7c4b9..ad4785c 100644 --- a/backend/tests/crawler_jobs.rs +++ b/backend/tests/crawler_jobs.rs @@ -7,7 +7,7 @@ use std::time::Duration; use mangalord::crawler::jobs::{ - self, EnqueueResult, JobPayload, KIND_SYNC_CHAPTER_CONTENT, + self, EnqueueResult, JobPayload, KIND_ANALYZE_PAGE, KIND_SYNC_CHAPTER_CONTENT, KIND_SYNC_MANGA, }; use sqlx::PgPool; use uuid::Uuid; @@ -27,6 +27,8 @@ fn sync_manga_payload(key: &str) -> JobPayload { JobPayload::SyncManga { source_id: "target".into(), source_manga_key: key.into(), + url: format!("https://target.example/manga/{key}"), + title: format!("Manga {key}"), } } @@ -278,6 +280,60 @@ async fn lease_with_kind_filter_only_matches_that_kind(pool: PgPool) { assert_eq!(job_state(&pool, manga_id).await, "pending"); } +#[sqlx::test(migrations = "./migrations")] +async fn lease_kinds_matches_listed_kinds_and_excludes_others(pool: PgPool) { + // The crawl worker drains both sync_chapter_content and sync_manga, but + // never analyze_page (owned by the analysis daemon). + let manga_id = match jobs::enqueue(&pool, &sync_manga_payload("foo")) + .await + .unwrap() + { + EnqueueResult::Inserted(id) => id, + _ => unreachable!(), + }; + let chapter_id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4())) + .await + .unwrap() + { + EnqueueResult::Inserted(id) => id, + _ => unreachable!(), + }; + let analyze_id = match jobs::enqueue( + &pool, + &JobPayload::AnalyzePage { + page_id: Uuid::new_v4(), + force: false, + }, + ) + .await + .unwrap() + { + EnqueueResult::Inserted(id) => id, + _ => unreachable!(), + }; + + let leases = jobs::lease_kinds( + &pool, + &[KIND_SYNC_CHAPTER_CONTENT, KIND_SYNC_MANGA], + 10, + Duration::from_secs(60), + ) + .await + .unwrap(); + let leased_ids: std::collections::HashSet = leases.iter().map(|l| l.id).collect(); + assert_eq!(leases.len(), 2, "both crawl kinds lease"); + assert!(leased_ids.contains(&manga_id)); + assert!(leased_ids.contains(&chapter_id)); + // analyze_page stays pending — not in the requested kinds. + assert_eq!(job_state(&pool, analyze_id).await, "pending"); + // Sanity: KIND_ANALYZE_PAGE alone leases only it. + let only_analyze = jobs::lease_kinds(&pool, &[KIND_ANALYZE_PAGE], 10, Duration::from_secs(60)) + .await + .unwrap(); + assert_eq!(only_analyze.len(), 1); + assert_eq!(only_analyze[0].id, analyze_id); +} + #[sqlx::test(migrations = "./migrations")] async fn concurrent_leases_under_skip_locked_return_disjoint_ids(pool: PgPool) { // 4 pending jobs, two concurrent calls each asking for up to 2. diff --git a/backend/tests/crawler_reconcile.rs b/backend/tests/crawler_reconcile.rs new file mode 100644 index 0000000..2eb2da2 --- /dev/null +++ b/backend/tests/crawler_reconcile.rs @@ -0,0 +1,145 @@ +//! Integration tests for the reconcile diff queries: +//! `existing_source_keys` (dropped rows count as present) and +//! `sync_manga_keys_with_blocking_job` (pending/running/dead block). +//! +//! The browser-driven full walk in `reconcile::reconcile_missing` needs a +//! real `chromiumoxide::Browser`, so it is covered by the manual/E2E path; +//! the pure set-diff (`select_missing`) is unit-tested in `crawler::reconcile`. + +use mangalord::crawler::jobs::{self, JobPayload}; +use mangalord::repo::crawler; +use sqlx::PgPool; +use uuid::Uuid; + +async fn ensure_target(pool: &PgPool) { + sqlx::query( + "INSERT INTO sources (id, name, base_url) VALUES ('target','T','http://x') \ + ON CONFLICT DO NOTHING", + ) + .execute(pool) + .await + .unwrap(); +} + +/// Insert a `manga_sources` row (with its backing manga) under `target`. +async fn seed_source_manga(pool: &PgPool, key: &str, dropped: bool) { + let manga_id = Uuid::new_v4(); + sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, $2)") + .bind(manga_id) + .bind(format!("M {key}")) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO manga_sources (source_id, source_manga_key, manga_id, source_url, dropped_at) \ + VALUES ('target', $1, $2, $3, CASE WHEN $4 THEN now() ELSE NULL END)", + ) + .bind(key) + .bind(manga_id) + .bind(format!("http://x/{key}")) + .bind(dropped) + .execute(pool) + .await + .unwrap(); +} + +async fn enqueue_sync_manga(pool: &PgPool, key: &str) -> Uuid { + let res = jobs::enqueue( + pool, + &JobPayload::SyncManga { + source_id: "target".into(), + source_manga_key: key.into(), + url: format!("http://x/{key}"), + title: format!("M {key}"), + }, + ) + .await + .unwrap(); + match res { + jobs::EnqueueResult::Inserted(id) => id, + jobs::EnqueueResult::Skipped => panic!("expected insert"), + } +} + +async fn set_job_state(pool: &PgPool, id: Uuid, state: &str) { + sqlx::query("UPDATE crawler_jobs SET state = $2::text WHERE id = $1") + .bind(id) + .bind(state) + .execute(pool) + .await + .unwrap(); +} + +#[sqlx::test(migrations = "./migrations")] +async fn existing_source_keys_includes_dropped_rows(pool: PgPool) { + ensure_target(&pool).await; + seed_source_manga(&pool, "live", false).await; + seed_source_manga(&pool, "dropped", true).await; + + let keys = crawler::existing_source_keys(&pool, "target").await.unwrap(); + assert!(keys.contains("live")); + assert!( + keys.contains("dropped"), + "a dropped manga_sources row must still count as present" + ); + assert_eq!(keys.len(), 2); +} + +#[sqlx::test(migrations = "./migrations")] +async fn sync_manga_blocking_job_matches_pending_running_dead_only(pool: PgPool) { + ensure_target(&pool).await; + // One key per state. + enqueue_sync_manga(&pool, "pending").await; // stays pending + let running = enqueue_sync_manga(&pool, "running").await; + let dead = enqueue_sync_manga(&pool, "dead").await; + let done = enqueue_sync_manga(&pool, "done").await; + let failed = enqueue_sync_manga(&pool, "failed").await; + set_job_state(&pool, running, "running").await; + set_job_state(&pool, dead, "dead").await; + set_job_state(&pool, done, "done").await; + set_job_state(&pool, failed, "failed").await; + + let blocked = crawler::sync_manga_keys_with_blocking_job(&pool, "target") + .await + .unwrap(); + assert!(blocked.contains("pending")); + assert!(blocked.contains("running")); + assert!(blocked.contains("dead")); + assert!(!blocked.contains("done"), "done must not block re-enqueue"); + assert!( + !blocked.contains("failed"), + "failed (mid-backoff) must not block re-enqueue" + ); + assert_eq!(blocked.len(), 3); +} + +#[sqlx::test(migrations = "./migrations")] +async fn sync_manga_blocking_job_is_per_source(pool: PgPool) { + ensure_target(&pool).await; + sqlx::query( + "INSERT INTO sources (id, name, base_url) VALUES ('other','O','http://y') \ + ON CONFLICT DO NOTHING", + ) + .execute(&pool) + .await + .unwrap(); + enqueue_sync_manga(&pool, "shared").await; + jobs::enqueue( + &pool, + &JobPayload::SyncManga { + source_id: "other".into(), + source_manga_key: "elsewhere".into(), + url: "http://y/elsewhere".into(), + title: "Elsewhere".into(), + }, + ) + .await + .unwrap(); + + let blocked = crawler::sync_manga_keys_with_blocking_job(&pool, "target") + .await + .unwrap(); + assert!(blocked.contains("shared")); + assert!(!blocked.contains("elsewhere")); + assert_eq!(blocked.len(), 1); +} diff --git a/frontend/package.json b/frontend/package.json index 548eb87..a31f43c 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.85.1", + "version": "0.86.0", "private": true, "type": "module", "scripts": { diff --git a/frontend/src/lib/api/admin.test.ts b/frontend/src/lib/api/admin.test.ts index df34aa6..eb12041 100644 --- a/frontend/src/lib/api/admin.test.ts +++ b/frontend/src/lib/api/admin.test.ts @@ -23,6 +23,7 @@ import { getCrawlerStatus, crawlerStatusStreamUrl, runCrawlerPass, + reconcileCrawler, restartCrawlerBrowser, updateCrawlerSession, clearCrawlerSessionExpired, @@ -479,6 +480,15 @@ describe('admin crawler api client', () => { expect(fetchSpy.mock.calls[0][0]).toMatch(/\/v1\/admin\/crawler\/run$/); }); + it('reconcileCrawler POSTs /v1/admin/crawler/reconcile', async () => { + fetchSpy.mockResolvedValueOnce(ok({ started: true })); + const r = await reconcileCrawler(); + expect(r.started).toBe(true); + const init = fetchSpy.mock.calls[0][1] as RequestInit; + expect(init.method).toBe('POST'); + expect(fetchSpy.mock.calls[0][0]).toMatch(/\/v1\/admin\/crawler\/reconcile$/); + }); + it('restartCrawlerBrowser POSTs the restart endpoint', async () => { fetchSpy.mockResolvedValueOnce(ok({ ok: true, error: null })); const r = await restartCrawlerBrowser(); diff --git a/frontend/src/lib/api/admin.ts b/frontend/src/lib/api/admin.ts index 189d73c..38e8e58 100644 --- a/frontend/src/lib/api/admin.ts +++ b/frontend/src/lib/api/admin.ts @@ -332,7 +332,8 @@ export type CrawlerPhase = | { state: 'idle'; next_fire: string | null } | { state: 'walking_list' } | { state: 'fetching_metadata'; index: number; total: number | null; title: string } - | { state: 'cover_backfill'; index: number; total: number }; + | { state: 'cover_backfill'; index: number; total: number } + | { state: 'reconciling'; walked: number; enqueued: number }; /** A chapter being crawled right now, with a live page count. */ export type ActiveChapter = { @@ -382,6 +383,12 @@ export async function runCrawlerPass(): Promise<{ started: boolean }> { return request('/v1/admin/crawler/run', { method: 'POST' }); } +/** POST /v1/admin/crawler/reconcile — full list-only walk that enqueues + * mangas missing from the DB as sync_manga jobs. */ +export async function reconcileCrawler(): Promise<{ started: boolean }> { + return request('/v1/admin/crawler/reconcile', { method: 'POST' }); +} + /** POST /v1/admin/crawler/browser/restart — coordinated Chromium restart. */ export async function restartCrawlerBrowser(): Promise<{ ok: boolean; error: string | null }> { return request('/v1/admin/crawler/browser/restart', { method: 'POST' }); @@ -410,6 +417,12 @@ export type DeadJob = { manga_id: string | null; manga_title: string | null; chapter_number: number | null; + /** Payload title (fallback label for sync_manga jobs with no manga row). */ + payload_title: string | null; + /** Source detail URL from the payload (sync_manga). */ + source_url: string | null; + /** Source-native key from the payload (sync_manga). */ + source_key: string | null; attempts: number; max_attempts: number; last_error: string | null; @@ -513,6 +526,10 @@ export type CrawlerHistoryRow = { chapter_number: number | null; page_number: number | null; source_key: string | null; + /** Payload title (fallback label for sync_manga jobs with no manga row). */ + payload_title: string | null; + /** Source detail URL from the payload (sync_manga). */ + source_url: string | null; attempts: number; max_attempts: number; last_error: string | null; diff --git a/frontend/src/lib/components/crawler/CrawlerControls.svelte b/frontend/src/lib/components/crawler/CrawlerControls.svelte index a66b8af..3665f95 100644 --- a/frontend/src/lib/components/crawler/CrawlerControls.svelte +++ b/frontend/src/lib/components/crawler/CrawlerControls.svelte @@ -5,6 +5,7 @@ status, busy, onRunPass, + onReconcile, onOpenRestart, onOpenSession, onClearExpired @@ -12,6 +13,7 @@ status: CrawlerStatus; busy: boolean; onRunPass: () => void; + onReconcile: () => void; onOpenRestart: () => void; onOpenSession: () => void; onClearExpired: () => void; @@ -22,6 +24,12 @@ + diff --git a/frontend/src/lib/components/crawler/CrawlerHero.svelte b/frontend/src/lib/components/crawler/CrawlerHero.svelte index be27492..3b7b70d 100644 --- a/frontend/src/lib/components/crawler/CrawlerHero.svelte +++ b/frontend/src/lib/components/crawler/CrawlerHero.svelte @@ -17,6 +17,8 @@ return `Fetching metadata · ${p.index}/${p.total ?? '?'} · ${p.title}`; case 'cover_backfill': return `Backfilling covers · ${p.index + 1}/${p.total}`; + case 'reconciling': + return `Reconciling · walked ${p.walked} · enqueued ${p.enqueued}`; default: { // Exhaustive default: if a new phase state ships // without a label, TypeScript flags the missing case diff --git a/frontend/src/lib/components/crawler/CrawlerHistoryTable.svelte b/frontend/src/lib/components/crawler/CrawlerHistoryTable.svelte index f8e9291..501d78e 100644 --- a/frontend/src/lib/components/crawler/CrawlerHistoryTable.svelte +++ b/frontend/src/lib/components/crawler/CrawlerHistoryTable.svelte @@ -71,7 +71,8 @@ await load(); } - /** A human "Manga · Ch N · pP" target, or the source key / em-dash. */ + /** A human "Manga · Ch N · pP" target. Falls back to the payload title + * (sync_manga jobs with no manga row yet), then the source key. */ function target(r: CrawlerHistoryRow): string { if (r.manga_title) { let s = r.manga_title; @@ -79,7 +80,7 @@ if (r.page_number != null) s += ` · p${r.page_number}`; return s; } - return r.source_key ?? '—'; + return r.payload_title ?? r.source_key ?? '—'; } function fmtAgo(iso: string): string { @@ -172,7 +173,15 @@ {r.state} {r.kind ?? '—'} - {target(r)} + + {#if !r.manga_title && r.payload_title && r.source_url} + {target(r)} + {:else} + {target(r)} + {/if} + {fmtDuration(r.duration_ms)} {r.attempts}/{r.max_attempts} - {j.manga_title ?? '(unknown)'} + {#if j.manga_title} + {j.manga_title} + {:else if j.payload_title} + {#if j.source_url} + {j.payload_title} + {:else} + {j.payload_title} + {/if} + {:else} + {j.source_key ?? '(unknown)'} + {/if} {#if j.chapter_number != null}· ch.{j.chapter_number}{/if} {j.attempts}/{j.max_attempts} diff --git a/frontend/src/lib/components/crawler/DeadJobsTable.svelte.test.ts b/frontend/src/lib/components/crawler/DeadJobsTable.svelte.test.ts new file mode 100644 index 0000000..a939ee7 --- /dev/null +++ b/frontend/src/lib/components/crawler/DeadJobsTable.svelte.test.ts @@ -0,0 +1,69 @@ +import { describe, it, expect, afterEach } from 'vitest'; +import { render, screen, cleanup } from '@testing-library/svelte'; +import DeadJobsTable from './DeadJobsTable.svelte'; +import type { DeadJob } from '$lib/api/admin'; + +afterEach(() => cleanup()); + +function deadJob(overrides: Partial): DeadJob { + return { + id: 'job-1', + kind: 'sync_chapter_content', + chapter_id: null, + manga_id: null, + manga_title: null, + chapter_number: null, + payload_title: null, + source_url: null, + source_key: null, + attempts: 5, + max_attempts: 5, + last_error: 'boom', + updated_at: '2026-06-16T00:00:00Z', + ...overrides + }; +} + +function baseProps(jobs: DeadJob[]) { + return { + jobs, + total: jobs.length, + page: 1, + totalPages: 1, + search: '', + busy: false, + onSearch: () => {}, + onPageChange: () => {}, + onRequeue: () => {}, + onRequeueAll: () => {} + }; +} + +describe('DeadJobsTable', () => { + it('prefers manga_title when present', () => { + const jobs = [deadJob({ manga_title: 'Naruto', chapter_number: 700 })]; + render(DeadJobsTable, { props: baseProps(jobs) }); + expect(screen.getByText(/Naruto/)).toBeTruthy(); + }); + + it('falls back to payload_title (linked to source_url) for a sync_manga job with no manga row', () => { + const jobs = [ + deadJob({ + kind: 'sync_manga', + manga_title: null, + payload_title: 'Ghost Manga', + source_url: 'http://x/manga/gone-1', + source_key: 'gone-1' + }) + ]; + render(DeadJobsTable, { props: baseProps(jobs) }); + const link = screen.getByRole('link', { name: 'Ghost Manga' }) as HTMLAnchorElement; + expect(link.getAttribute('href')).toBe('http://x/manga/gone-1'); + }); + + it('falls back to source_key when there is no title at all', () => { + const jobs = [deadJob({ kind: 'sync_manga', source_key: 'only-key' })]; + render(DeadJobsTable, { props: baseProps(jobs) }); + expect(screen.getByText('only-key')).toBeTruthy(); + }); +}); diff --git a/frontend/src/routes/admin/crawler/+page.svelte b/frontend/src/routes/admin/crawler/+page.svelte index 9760b9b..bed9b8f 100644 --- a/frontend/src/routes/admin/crawler/+page.svelte +++ b/frontend/src/routes/admin/crawler/+page.svelte @@ -16,6 +16,7 @@ getCrawlerStatus, crawlerStatusStreamUrl, runCrawlerPass, + reconcileCrawler, restartCrawlerBrowser, updateCrawlerSession, clearCrawlerSessionExpired, @@ -308,6 +309,13 @@ }); } + async function onReconcile() { + await withBusy('reconcile', async () => { + await reconcileCrawler(); + notice = 'Reconcile started — walking the full list for missing mangas.'; + }); + } + async function onConfirmRestart() { restartModalOpen = false; await withBusy('restart browser', async () => { @@ -427,6 +435,7 @@ {status} {busy} {onRunPass} + {onReconcile} onOpenRestart={() => (restartModalOpen = true)} onOpenSession={() => { sessionModalOpen = true;