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>
146 lines
4.6 KiB
Rust
146 lines
4.6 KiB
Rust
//! 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);
|
|
}
|