Files
Mangalord/backend/tests/crawler_dead_jobs.rs
MechaCat02 d6ac648ac9 feat(admin): crawler observability dashboard + reliability hardening (0.55.0)
Admin-only crawler dashboard backed by an SSE live-status stream,
coordinated browser restart, runtime PHPSESSID refresh, dead-letter
requeue, and a batch of reliability fixes. Closes everything from
the two-pass audit (10 commits' worth) and bumps 0.52.0 -> 0.55.0.

Backend:

- New /admin/crawler/* surface (cookie-auth, RequireAdmin) split
  into status / control / dead_jobs / backlog modules. SSE stream
  composes in-memory status with DB-derived queue counts, memoizes
  the counts for 1s and debounces watch pokes for 250ms (~10x QPS
  reduction per subscriber). One-shot GET /admin/crawler shares the
  same compose path.
- POST /admin/crawler/run gated by manual_pass_lock try_lock_owned
  (409 Conflict on overlapping click); browser restart goes through
  the coordinated_restart gate (drain + relaunch + auto-clear of the
  sticky session_expired flag on Ok).
- Runtime PHPSESSID refresh via SessionController (allow-list
  validation, never logged, audit row carries SHA-256 fingerprint).
  Storage layer is repo::crawler::runtime_session_{load,persist}.
- Dead-letter requeue with four scopes (all/manga/chapter/job);
  scope=all requires confirm:true; DISTINCT ON dedup keeps the
  partial unique index from rejecting requeues for chapters with
  multiple dead rows. SQL is four &'static str constants per scope.
- StatusHandle + ChapterGuard / CoverGuard RAII model survives
  panics; last-writer-wins on cover so concurrent dispatches don't
  clobber each other's slot. Pure functions (should_stop /
  should_mark_clean_exit / should_abort_pass) with named regression
  tests.
- Reliability bundle: per-lease heartbeat, jitter on retries,
  per-job timeout, circuit breaker on consecutive failures, BrowserManager
  coordinated restart gate, request fingerprint changes.
- Streaming page download: Storage::put_stream trait method,
  LocalStorage impl atomic via temp + fsync + UUID-suffixed rename.
  Pages stream through with peak memory ~one HTTP chunk + 64-byte
  sniff prefix instead of one full image per dispatch.
- New partial indexes (migration 0022): mangas_missing_cover_idx
  and crawler_jobs_dead_idx, both ordered by updated_at DESC to
  match the dashboard's LIMIT/OFFSET reads.
- Security hardening: admin_csrf_guard (Origin/Referer allowlist
  on /admin/* mutations, opt-in via ADMIN_ALLOWED_ORIGINS),
  admin_no_store_guard (Cache-Control: no-store on admin
  responses), audit rows carry per-scope target_id.

Frontend:

- /admin/crawler page decomposed into lib/components/crawler/
  (11 components: ProgressBar, SearchBar, CrawlerHero,
  CrawlerControls, ActiveChaptersCard, ActiveJobsTable,
  MissingCoversTable, DeadJobsTable, RestartConfirmModal,
  RequeueAllConfirmModal, SessionModal). Page is 532 LOC of
  orchestration; each component 22-148 LOC.
- EventSource lifecycle wired to visibilitychange / pagehide /
  pageshow (BFCache); after 5 consecutive errors probes the status
  endpoint so a 401 routes through the global on401Hook instead of
  infinite silent reconnects.
- Backlog $effect refetches debounced 500ms with per-loader
  AbortControllers; refresh after a control action only runs when
  the SSE stream is dead.
- Inline requeue button on /admin/mangas patches the affected row's
  sync_state locally (no full chapter-list refetch); proper
  aria-label. Requeue-all gets its own confirm modal; both confirm
  modals autofocus Cancel.
- SvelteKit reverse proxy bypasses its 5-minute AbortController
  for Accept: text/event-stream; pure shouldBypassProxyTimeout
  helper covered by unit tests.

Config / docs:

- New env vars (.env.example): ADMIN_ALLOWED_ORIGINS,
  CRAWLER_JOB_TIMEOUT_SECS, CRAWLER_METADATA_MAX_CONSECUTIVE_FAILURES,
  CRAWLER_BROWSER_RESTART_THRESHOLD.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-06-06 18:49:56 +02:00

305 lines
11 KiB
Rust

//! Integration tests for the dead-letter admin queries in
//! `repo::crawler`: listing dead jobs with manga/chapter context and the
//! scoped requeue (all / per-manga / single) used by the admin dashboard.
use mangalord::repo::crawler::{self, RequeueScope};
use serde_json::json;
use sqlx::PgPool;
use uuid::Uuid;
/// Seed a manga with no cover + a live source row (so it's "queued for a
/// cover fetch"). Returns the manga id.
async fn seed_missing_cover(pool: &PgPool, title: &str) -> Uuid {
let manga_id = Uuid::new_v4();
sqlx::query("INSERT INTO mangas (id, title, cover_image_path) VALUES ($1, $2, NULL)")
.bind(manga_id)
.bind(title)
.execute(pool)
.await
.unwrap();
sqlx::query("INSERT INTO sources (id, name, base_url) VALUES ('target', 'T', 'http://x') ON CONFLICT DO NOTHING")
.execute(pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO manga_sources (source_id, source_manga_key, manga_id, source_url) \
VALUES ('target', $1, $2, 'http://x/m')",
)
.bind(format!("k-{manga_id}"))
.bind(manga_id)
.execute(pool)
.await
.unwrap();
manga_id
}
/// Seed a manga + chapter and return their ids.
async fn seed_chapter(pool: &PgPool, title: &str, number: i32) -> (Uuid, Uuid) {
let manga_id = Uuid::new_v4();
let chapter_id = Uuid::new_v4();
sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, $2)")
.bind(manga_id)
.bind(title)
.execute(pool)
.await
.unwrap();
sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, $3)")
.bind(chapter_id)
.bind(manga_id)
.bind(number)
.execute(pool)
.await
.unwrap();
(manga_id, chapter_id)
}
/// Insert a crawler_jobs row in a given state for a chapter-content job.
async fn insert_job(pool: &PgPool, chapter_id: Uuid, state: &str, attempts: i32) -> Uuid {
let id = Uuid::new_v4();
let payload = json!({
"kind": "sync_chapter_content",
"source_id": "target",
"chapter_id": chapter_id,
"source_chapter_key": "k",
});
sqlx::query(
"INSERT INTO crawler_jobs (id, payload, state, attempts, last_error) \
VALUES ($1, $2, $3, $4, 'boom')",
)
.bind(id)
.bind(payload)
.bind(state)
.bind(attempts)
.execute(pool)
.await
.unwrap();
id
}
async fn state_of(pool: &PgPool, id: Uuid) -> String {
sqlx::query_scalar::<_, String>("SELECT state FROM crawler_jobs WHERE id = $1")
.bind(id)
.fetch_one(pool)
.await
.unwrap()
}
#[sqlx::test(migrations = "./migrations")]
async fn list_dead_jobs_returns_context_and_total(pool: PgPool) {
let (_m, c1) = seed_chapter(&pool, "Naruto", 700).await;
insert_job(&pool, c1, "dead", 5).await;
// A non-dead job must not appear.
let (_m2, c2) = seed_chapter(&pool, "Bleach", 1).await;
insert_job(&pool, c2, "pending", 0).await;
let (items, total) = crawler::list_dead_jobs(&pool, None, 50, 0).await.unwrap();
assert_eq!(total, 1);
assert_eq!(items.len(), 1);
let row = &items[0];
assert_eq!(row.manga_title.as_deref(), Some("Naruto"));
assert_eq!(row.chapter_number, Some(700));
assert_eq!(row.attempts, 5);
assert_eq!(row.last_error.as_deref(), Some("boom"));
}
#[sqlx::test(migrations = "./migrations")]
async fn list_dead_jobs_filters_by_title_search(pool: PgPool) {
let (_m, c1) = seed_chapter(&pool, "Naruto", 700).await;
insert_job(&pool, c1, "dead", 5).await;
let (_m2, c2) = seed_chapter(&pool, "One Piece", 1).await;
insert_job(&pool, c2, "dead", 5).await;
let (items, total) = crawler::list_dead_jobs(&pool, Some("piece"), 50, 0)
.await
.unwrap();
assert_eq!(total, 1);
assert_eq!(items[0].manga_title.as_deref(), Some("One Piece"));
}
#[sqlx::test(migrations = "./migrations")]
async fn requeue_all_resets_dead_jobs_to_pending(pool: PgPool) {
let (_m, c1) = seed_chapter(&pool, "A", 1).await;
let (_m2, c2) = seed_chapter(&pool, "B", 1).await;
let j1 = insert_job(&pool, c1, "dead", 5).await;
let j2 = insert_job(&pool, c2, "dead", 5).await;
let n = crawler::requeue_dead_jobs(&pool, RequeueScope::All)
.await
.unwrap();
assert_eq!(n, 2);
assert_eq!(state_of(&pool, j1).await, "pending");
assert_eq!(state_of(&pool, j2).await, "pending");
let attempts: i32 = sqlx::query_scalar("SELECT attempts FROM crawler_jobs WHERE id = $1")
.bind(j1)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(attempts, 0, "attempts reset on requeue");
}
#[sqlx::test(migrations = "./migrations")]
async fn requeue_by_manga_scopes_to_that_manga(pool: PgPool) {
let (m1, c1) = seed_chapter(&pool, "A", 1).await;
let (_m2, c2) = seed_chapter(&pool, "B", 1).await;
let j1 = insert_job(&pool, c1, "dead", 5).await;
let j2 = insert_job(&pool, c2, "dead", 5).await;
let n = crawler::requeue_dead_jobs(&pool, RequeueScope::Manga(m1))
.await
.unwrap();
assert_eq!(n, 1);
assert_eq!(state_of(&pool, j1).await, "pending");
assert_eq!(state_of(&pool, j2).await, "dead", "other manga untouched");
}
#[sqlx::test(migrations = "./migrations")]
async fn requeue_by_chapter_scopes_to_that_chapter(pool: PgPool) {
let (_m, c1) = seed_chapter(&pool, "A", 1).await;
let (_m2, c2) = seed_chapter(&pool, "A", 2).await;
let j1 = insert_job(&pool, c1, "dead", 5).await;
let j2 = insert_job(&pool, c2, "dead", 5).await;
let n = crawler::requeue_dead_jobs(&pool, RequeueScope::Chapter(c1))
.await
.unwrap();
assert_eq!(n, 1);
assert_eq!(state_of(&pool, j1).await, "pending");
assert_eq!(state_of(&pool, j2).await, "dead", "other chapter untouched");
}
#[sqlx::test(migrations = "./migrations")]
async fn requeue_single_job(pool: PgPool) {
let (_m, c1) = seed_chapter(&pool, "A", 1).await;
let (_m2, c2) = seed_chapter(&pool, "B", 1).await;
let j1 = insert_job(&pool, c1, "dead", 5).await;
let j2 = insert_job(&pool, c2, "dead", 5).await;
let n = crawler::requeue_dead_jobs(&pool, RequeueScope::Job(j1))
.await
.unwrap();
assert_eq!(n, 1);
assert_eq!(state_of(&pool, j1).await, "pending");
assert_eq!(state_of(&pool, j2).await, "dead");
}
#[sqlx::test(migrations = "./migrations")]
async fn requeue_skips_dead_when_live_job_exists_for_same_chapter(pool: PgPool) {
let (_m, c1) = seed_chapter(&pool, "A", 1).await;
let dead = insert_job(&pool, c1, "dead", 5).await;
// A live pending job for the SAME chapter already exists.
insert_job(&pool, c1, "pending", 0).await;
let n = crawler::requeue_dead_jobs(&pool, RequeueScope::All)
.await
.unwrap();
assert_eq!(n, 0, "must not resurrect a dead job that has a live counterpart");
assert_eq!(state_of(&pool, dead).await, "dead");
}
#[sqlx::test(migrations = "./migrations")]
async fn requeue_with_two_dead_jobs_for_one_chapter_revives_one_not_500(pool: PgPool) {
// Regression: two dead jobs for the SAME chapter must not both flip to
// pending in one statement — that would violate the partial unique
// dedup index and abort the whole requeue.
let (manga_id, c1) = seed_chapter(&pool, "A", 1).await;
let older = insert_job(&pool, c1, "dead", 5).await;
let newer = insert_job(&pool, c1, "dead", 5).await;
// Make `newer` unambiguously newer.
sqlx::query("UPDATE crawler_jobs SET updated_at = now() - interval '1 hour' WHERE id = $1")
.bind(older)
.execute(&pool)
.await
.unwrap();
for scope in [RequeueScope::All, RequeueScope::Manga(manga_id), RequeueScope::Chapter(c1)] {
// Reset to two-dead before each scope variant.
sqlx::query("UPDATE crawler_jobs SET state = 'dead' WHERE id = ANY($1)")
.bind(vec![older, newer])
.execute(&pool)
.await
.unwrap();
let n = crawler::requeue_dead_jobs(&pool, scope)
.await
.expect("requeue must not error on duplicate dead jobs");
assert_eq!(n, 1, "exactly one dead job per chapter is revived");
// The newest one is the survivor; the other stays dead.
assert_eq!(state_of(&pool, newer).await, "pending");
assert_eq!(state_of(&pool, older).await, "dead");
}
}
#[sqlx::test(migrations = "./migrations")]
async fn list_active_jobs_returns_pending_and_running_running_first(pool: PgPool) {
let (_m, c1) = seed_chapter(&pool, "Naruto", 700).await;
let (_m2, c2) = seed_chapter(&pool, "Bleach", 10).await;
insert_job(&pool, c1, "pending", 0).await;
insert_job(&pool, c2, "running", 1).await;
// A dead + a done job must NOT appear.
let (_m3, c3) = seed_chapter(&pool, "Gone", 1).await;
insert_job(&pool, c3, "dead", 5).await;
let (items, total) = crawler::list_active_jobs(&pool, None, 50, 0).await.unwrap();
assert_eq!(total, 2);
assert_eq!(items.len(), 2);
// Running first.
assert_eq!(items[0].state, "running");
assert_eq!(items[0].manga_title.as_deref(), Some("Bleach"));
assert_eq!(items[1].state, "pending");
assert_eq!(items[1].chapter_number, Some(700));
}
#[sqlx::test(migrations = "./migrations")]
async fn list_active_jobs_filters_by_title(pool: PgPool) {
let (_m, c1) = seed_chapter(&pool, "Naruto", 1).await;
let (_m2, c2) = seed_chapter(&pool, "One Piece", 1).await;
insert_job(&pool, c1, "pending", 0).await;
insert_job(&pool, c2, "pending", 0).await;
let (items, total) = crawler::list_active_jobs(&pool, Some("piece"), 50, 0)
.await
.unwrap();
assert_eq!(total, 1);
assert_eq!(items[0].manga_title.as_deref(), Some("One Piece"));
}
#[sqlx::test(migrations = "./migrations")]
async fn missing_covers_count_and_list(pool: PgPool) {
seed_missing_cover(&pool, "Naruto").await;
seed_missing_cover(&pool, "Bleach").await;
// A manga WITH a cover must not be counted.
let with_cover = Uuid::new_v4();
sqlx::query("INSERT INTO mangas (id, title, cover_image_path) VALUES ($1, 'Done', 'k.jpg')")
.bind(with_cover)
.execute(&pool)
.await
.unwrap();
assert_eq!(crawler::count_missing_covers(&pool).await.unwrap(), 2);
let (items, total) = crawler::list_missing_cover_mangas(&pool, None, 50, 0)
.await
.unwrap();
assert_eq!(total, 2);
assert_eq!(items.len(), 2);
let (items, total) = crawler::list_missing_cover_mangas(&pool, Some("naru"), 50, 0)
.await
.unwrap();
assert_eq!(total, 1);
assert_eq!(items[0].manga_title, "Naruto");
}
#[sqlx::test(migrations = "./migrations")]
async fn job_state_counts_groups_by_state(pool: PgPool) {
let (_m, c1) = seed_chapter(&pool, "A", 1).await;
let (_m2, c2) = seed_chapter(&pool, "B", 1).await;
let (_m3, c3) = seed_chapter(&pool, "C", 1).await;
insert_job(&pool, c1, "pending", 0).await;
insert_job(&pool, c2, "dead", 5).await;
insert_job(&pool, c3, "dead", 5).await;
let (pending, running, dead) = crawler::job_state_counts(&pool).await.unwrap();
assert_eq!(pending, 1);
assert_eq!(running, 0);
assert_eq!(dead, 2);
}