feat(crawler): reconcile pass to enqueue mangas missing from the DB (0.85.0)

The interleaved metadata pass misses mangas to list drift (a title slips a
pagination slot during the slow detail walk). Add a reconcile pass: a cheap,
full, list-only walk (refs only, no detail visit, no early stop) that
set-diffs the walked keys against manga_sources and enqueues the strictly-
missing ones as SyncManga jobs. A set-diff is immune to drift — a manga only
has to appear somewhere in the list.

- Build the previously-dead SyncManga worker by extending RealChapterDispatcher
  to run the shared pipeline::process_manga_ref (fetch → upsert → cover →
  chapters), refactored out of run_metadata_pass so both paths stay in lockstep.
  The crawl worker now leases both sync_chapter_content and sync_manga
  (jobs::lease_kinds); both serialize on the single exclusive browser.
- SyncManga payload carries url + title so the worker can rebuild the ref and
  the dead-jobs/history UI can label a missing manga that has no manga row yet.
- "Missing" = strict NOT EXISTS (dropped rows count as present). Re-enqueue is
  skipped when a pending/running/dead SyncManga job already exists, so a gone
  manga (detail 404 → retries → dead) is left dead and not retried.
- New POST /v1/admin/crawler/reconcile (fire-and-forget, shares
  manual_pass_lock) + "Reconcile missing" admin button + Reconciling status
  phase streamed over SSE.
- dead-jobs/history queries surface payload title/url/key; tables fall back to
  them for sync_manga rows; both searches match the payload title.

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

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

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.
 // ---------------------------------------------------------------------------

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<SyncOutcome> {
+        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<String> =
+        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;

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;

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<Uuid> = 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.

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);
+}