347 lines
11 KiB
Rust
347 lines
11 KiB
Rust
//! Integration tests for the analysis worker daemon: it leases only
|
|
//! `analyze_page` jobs, acks done on success, skips already-done pages
|
|
//! (unless forced), and on terminal failure writes a `failed` row.
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use mangalord::analysis::daemon::{self, test_support::CountingDispatcher, AnalysisDaemonConfig};
|
|
use mangalord::domain::page_analysis::{
|
|
AnalysisStatus, SafetyFlag, VisionAnalysis,
|
|
};
|
|
use mangalord::repo::page_analysis;
|
|
use sqlx::PgPool;
|
|
use tokio_util::sync::CancellationToken;
|
|
use uuid::Uuid;
|
|
|
|
async fn seed_page(pool: &PgPool) -> Uuid {
|
|
let manga_id: Uuid =
|
|
sqlx::query_scalar("INSERT INTO mangas (title) VALUES ('M') RETURNING id")
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap();
|
|
let chapter_id: Uuid = sqlx::query_scalar(
|
|
"INSERT INTO chapters (manga_id, number) VALUES ($1, 1) RETURNING id",
|
|
)
|
|
.bind(manga_id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap();
|
|
sqlx::query_scalar(
|
|
"INSERT INTO pages (chapter_id, page_number, storage_key, content_type) \
|
|
VALUES ($1, 1, 'k/1.png', 'image/png') RETURNING id",
|
|
)
|
|
.bind(chapter_id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap()
|
|
}
|
|
|
|
async fn job_state(pool: &PgPool, page_id: Uuid) -> Option<String> {
|
|
sqlx::query_scalar(
|
|
"SELECT state FROM crawler_jobs \
|
|
WHERE payload->>'kind' = 'analyze_page' AND payload->>'page_id' = $1",
|
|
)
|
|
.bind(page_id.to_string())
|
|
.fetch_optional(pool)
|
|
.await
|
|
.unwrap()
|
|
}
|
|
|
|
/// Poll until the page's analyze_page job reaches `want`, or fail after 5s.
|
|
async fn wait_for_state(pool: &PgPool, page_id: Uuid, want: &str) {
|
|
let deadline = Duration::from_secs(5);
|
|
let result = tokio::time::timeout(deadline, async {
|
|
loop {
|
|
if job_state(pool, page_id).await.as_deref() == Some(want) {
|
|
return;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
}
|
|
})
|
|
.await;
|
|
assert!(result.is_ok(), "job for {page_id} never reached state {want}");
|
|
}
|
|
|
|
fn spawn_with(
|
|
pool: &PgPool,
|
|
dispatcher: Arc<CountingDispatcher>,
|
|
) -> (daemon::AnalysisDaemonHandle, CancellationToken) {
|
|
let cancel = CancellationToken::new();
|
|
let handle = daemon::spawn(
|
|
pool.clone(),
|
|
cancel.clone(),
|
|
AnalysisDaemonConfig {
|
|
dispatcher,
|
|
workers: 1,
|
|
job_timeout: Duration::from_secs(5),
|
|
events: std::sync::Arc::new(
|
|
mangalord::analysis::events::AnalysisEvents::new(),
|
|
),
|
|
readiness: None,
|
|
},
|
|
);
|
|
(handle, cancel)
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn worker_dispatches_and_acks_done(pool: PgPool) {
|
|
let page_id = seed_page(&pool).await;
|
|
page_analysis::enqueue_for_page(&pool, page_id, false)
|
|
.await
|
|
.unwrap();
|
|
|
|
let dispatcher = CountingDispatcher::ok();
|
|
let (handle, _c) = spawn_with(&pool, dispatcher.clone());
|
|
wait_for_state(&pool, page_id, "done").await;
|
|
handle.shutdown().await;
|
|
|
|
assert_eq!(dispatcher.call_count(), 1);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn worker_skips_already_done_page_unless_forced(pool: PgPool) {
|
|
let page_id = seed_page(&pool).await;
|
|
// Pre-mark the page analyzed.
|
|
page_analysis::persist_analysis(
|
|
&pool,
|
|
page_id,
|
|
&VisionAnalysis {
|
|
ocr_results: vec![],
|
|
tagging_results: vec![],
|
|
scene_description: String::new(),
|
|
safety_flag: SafetyFlag::default(),
|
|
},
|
|
"m",
|
|
)
|
|
.await
|
|
.unwrap();
|
|
page_analysis::enqueue_for_page(&pool, page_id, false)
|
|
.await
|
|
.unwrap();
|
|
|
|
let dispatcher = CountingDispatcher::ok();
|
|
let (handle, _c) = spawn_with(&pool, dispatcher.clone());
|
|
wait_for_state(&pool, page_id, "done").await;
|
|
handle.shutdown().await;
|
|
|
|
assert_eq!(
|
|
dispatcher.call_count(),
|
|
0,
|
|
"a non-forced job for a done page must skip dispatch"
|
|
);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn worker_reanalyzes_done_page_when_forced(pool: PgPool) {
|
|
let page_id = seed_page(&pool).await;
|
|
page_analysis::persist_analysis(
|
|
&pool,
|
|
page_id,
|
|
&VisionAnalysis {
|
|
ocr_results: vec![],
|
|
tagging_results: vec![],
|
|
scene_description: String::new(),
|
|
safety_flag: SafetyFlag::default(),
|
|
},
|
|
"m",
|
|
)
|
|
.await
|
|
.unwrap();
|
|
page_analysis::enqueue_for_page(&pool, page_id, true)
|
|
.await
|
|
.unwrap();
|
|
|
|
let dispatcher = CountingDispatcher::ok();
|
|
let (handle, _c) = spawn_with(&pool, dispatcher.clone());
|
|
wait_for_state(&pool, page_id, "done").await;
|
|
handle.shutdown().await;
|
|
|
|
assert_eq!(dispatcher.call_count(), 1, "force must re-dispatch");
|
|
|
|
// The worker stamps how long the dispatch took onto the existing row
|
|
// (recorded after ack; shutdown has awaited the in-flight job).
|
|
let dur: Option<i64> =
|
|
sqlx::query_scalar("SELECT duration_ms FROM page_analysis WHERE page_id = $1")
|
|
.bind(page_id)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert!(dur.is_some(), "worker should record analysis duration_ms");
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn worker_marks_failed_row_on_terminal_failure(pool: PgPool) {
|
|
let page_id = seed_page(&pool).await;
|
|
page_analysis::enqueue_for_page(&pool, page_id, false)
|
|
.await
|
|
.unwrap();
|
|
// Make the first failure terminal (no backoff wait in the test).
|
|
sqlx::query(
|
|
"UPDATE crawler_jobs SET max_attempts = 1 \
|
|
WHERE payload->>'page_id' = $1",
|
|
)
|
|
.bind(page_id.to_string())
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
|
|
let dispatcher = CountingDispatcher::failing();
|
|
let (handle, _c) = spawn_with(&pool, dispatcher.clone());
|
|
wait_for_state(&pool, page_id, "dead").await;
|
|
handle.shutdown().await;
|
|
|
|
let row = page_analysis::load(&pool, page_id).await.unwrap().unwrap();
|
|
assert_eq!(row.status, AnalysisStatus::Failed);
|
|
assert!(row.error.is_some());
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn worker_isolates_dispatcher_panics(pool: PgPool) {
|
|
let page_id = seed_page(&pool).await;
|
|
page_analysis::enqueue_for_page(&pool, page_id, false)
|
|
.await
|
|
.unwrap();
|
|
sqlx::query(
|
|
"UPDATE crawler_jobs SET max_attempts = 1 WHERE payload->>'page_id' = $1",
|
|
)
|
|
.bind(page_id.to_string())
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
|
|
let dispatcher = CountingDispatcher::panicking();
|
|
let (handle, _c) = spawn_with(&pool, dispatcher.clone());
|
|
wait_for_state(&pool, page_id, "dead").await;
|
|
handle.shutdown().await;
|
|
|
|
// The worker survived the panic and recorded a failed row.
|
|
let row = page_analysis::load(&pool, page_id).await.unwrap().unwrap();
|
|
assert_eq!(row.status, AnalysisStatus::Failed);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn worker_publishes_started_and_completed_events(pool: PgPool) {
|
|
let page_id = seed_page(&pool).await;
|
|
page_analysis::enqueue_for_page(&pool, page_id, false)
|
|
.await
|
|
.unwrap();
|
|
|
|
let events = Arc::new(mangalord::analysis::events::AnalysisEvents::new());
|
|
let mut rx = events.subscribe();
|
|
let cancel = CancellationToken::new();
|
|
let handle = daemon::spawn(
|
|
pool.clone(),
|
|
cancel,
|
|
AnalysisDaemonConfig {
|
|
dispatcher: CountingDispatcher::ok(),
|
|
workers: 1,
|
|
job_timeout: Duration::from_secs(5),
|
|
events: events.clone(),
|
|
readiness: None,
|
|
},
|
|
);
|
|
|
|
let mut kinds = Vec::new();
|
|
let _ = tokio::time::timeout(Duration::from_secs(5), async {
|
|
while let Ok(ev) = rx.recv().await {
|
|
let v = serde_json::to_value(&ev).unwrap();
|
|
let kind = v["kind"].as_str().unwrap().to_string();
|
|
// Each event must carry the page breadcrumb.
|
|
assert_eq!(v["page_id"].as_str().unwrap(), page_id.to_string());
|
|
let done = kind == "completed";
|
|
kinds.push(kind);
|
|
if done {
|
|
break;
|
|
}
|
|
}
|
|
})
|
|
.await;
|
|
handle.shutdown().await;
|
|
|
|
assert!(kinds.contains(&"started".to_string()), "expected a started event");
|
|
assert!(
|
|
kinds.contains(&"completed".to_string()),
|
|
"expected a completed event"
|
|
);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn worker_publishes_failed_event_on_dispatch_error(pool: PgPool) {
|
|
let page_id = seed_page(&pool).await;
|
|
page_analysis::enqueue_for_page(&pool, page_id, false)
|
|
.await
|
|
.unwrap();
|
|
sqlx::query("UPDATE crawler_jobs SET max_attempts = 1 WHERE payload->>'page_id' = $1")
|
|
.bind(page_id.to_string())
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
|
|
let events = Arc::new(mangalord::analysis::events::AnalysisEvents::new());
|
|
let mut rx = events.subscribe();
|
|
let cancel = CancellationToken::new();
|
|
let handle = daemon::spawn(
|
|
pool.clone(),
|
|
cancel,
|
|
AnalysisDaemonConfig {
|
|
dispatcher: CountingDispatcher::failing(),
|
|
workers: 1,
|
|
job_timeout: Duration::from_secs(5),
|
|
events: events.clone(),
|
|
readiness: None,
|
|
},
|
|
);
|
|
|
|
let mut saw_failed = false;
|
|
let _ = tokio::time::timeout(Duration::from_secs(5), async {
|
|
while let Ok(ev) = rx.recv().await {
|
|
let v = serde_json::to_value(&ev).unwrap();
|
|
if v["kind"] == "failed" {
|
|
saw_failed = true;
|
|
break;
|
|
}
|
|
}
|
|
})
|
|
.await;
|
|
handle.shutdown().await;
|
|
assert!(saw_failed, "expected a failed event");
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn worker_ignores_non_analyze_jobs(pool: PgPool) {
|
|
// A crawl job must be left untouched by the analysis worker.
|
|
use mangalord::crawler::jobs::{self, JobPayload};
|
|
let crawl = jobs::enqueue(
|
|
&pool,
|
|
&JobPayload::SyncManga {
|
|
source_id: "s".into(),
|
|
source_manga_key: "k".into(),
|
|
},
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let crawl_id = match crawl {
|
|
jobs::EnqueueResult::Inserted(id) => id,
|
|
jobs::EnqueueResult::Skipped => panic!("expected insert"),
|
|
};
|
|
|
|
let page_id = seed_page(&pool).await;
|
|
page_analysis::enqueue_for_page(&pool, page_id, false)
|
|
.await
|
|
.unwrap();
|
|
|
|
let dispatcher = CountingDispatcher::ok();
|
|
let (handle, _c) = spawn_with(&pool, dispatcher.clone());
|
|
wait_for_state(&pool, page_id, "done").await;
|
|
handle.shutdown().await;
|
|
|
|
let crawl_state: String =
|
|
sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1")
|
|
.bind(crawl_id)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(crawl_state, "pending", "crawl job must be left for the crawler");
|
|
}
|