83a9ab40cd
The retention reaper deleted only `state = 'done'` rows, so terminal
`dead` jobs (retries exhausted) accumulated in crawler_jobs forever —
slow unbounded table growth. Both states are end-of-life and never
leased again.
Rename `reap_done` → `reap_terminal` and broaden the filter to
`state IN ('done','dead')`; `pending`/`running` stay untouched. Test
extended to assert old done + old dead both reap while fresh terminal and
old-but-active rows survive.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1216 lines
42 KiB
Rust
1216 lines
42 KiB
Rust
//! Integration tests for `crawler::jobs` queue operations.
|
|
//!
|
|
//! Uses `#[sqlx::test(migrations = "./migrations")]` which provisions a fresh
|
|
//! migrated DB per test. No browser, no axum router — these exercise the SQL
|
|
//! shape and dedup-index semantics directly against Postgres.
|
|
|
|
use std::time::Duration;
|
|
|
|
use mangalord::crawler::jobs::{
|
|
self, EnqueueResult, JobPayload, KIND_ANALYZE_PAGE, KIND_SYNC_CHAPTER_CONTENT, KIND_SYNC_MANGA,
|
|
};
|
|
use sqlx::PgPool;
|
|
use uuid::Uuid;
|
|
|
|
fn chapter_content_payload(chapter_id: Uuid) -> JobPayload {
|
|
JobPayload::SyncChapterContent {
|
|
source_id: "target".into(),
|
|
chapter_id,
|
|
source_chapter_key: format!("ch-{chapter_id}"),
|
|
}
|
|
}
|
|
|
|
/// A non-`SyncChapterContent` payload, used to assert that only the
|
|
/// chapter-content kind is deduplicated by the partial index and that
|
|
/// `lease`'s kind filter correctly excludes other kinds.
|
|
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}"),
|
|
}
|
|
}
|
|
|
|
async fn job_state(pool: &PgPool, id: Uuid) -> String {
|
|
sqlx::query_scalar::<_, String>("SELECT state FROM crawler_jobs WHERE id = $1")
|
|
.bind(id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap()
|
|
}
|
|
|
|
async fn job_attempts(pool: &PgPool, id: Uuid) -> i32 {
|
|
sqlx::query_scalar::<_, i32>("SELECT attempts FROM crawler_jobs WHERE id = $1")
|
|
.bind(id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap()
|
|
}
|
|
|
|
async fn job_count(pool: &PgPool) -> i64 {
|
|
sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM crawler_jobs")
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap()
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn enqueue_inserts_pending_row_with_round_trip_payload(pool: PgPool) {
|
|
let chapter_id = Uuid::new_v4();
|
|
let payload = chapter_content_payload(chapter_id);
|
|
|
|
let result = jobs::enqueue(&pool, &payload).await.unwrap();
|
|
let id = match result {
|
|
EnqueueResult::Inserted(id) => id,
|
|
EnqueueResult::Skipped => panic!("expected Inserted on first enqueue"),
|
|
};
|
|
|
|
assert_eq!(job_state(&pool, id).await, "pending");
|
|
assert_eq!(job_attempts(&pool, id).await, 0);
|
|
|
|
let raw_payload: serde_json::Value =
|
|
sqlx::query_scalar("SELECT payload FROM crawler_jobs WHERE id = $1")
|
|
.bind(id)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
let decoded: JobPayload = serde_json::from_value(raw_payload).unwrap();
|
|
match decoded {
|
|
JobPayload::SyncChapterContent {
|
|
source_id,
|
|
chapter_id: c,
|
|
source_chapter_key,
|
|
} => {
|
|
assert_eq!(source_id, "target");
|
|
assert_eq!(c, chapter_id);
|
|
assert_eq!(source_chapter_key, format!("ch-{chapter_id}"));
|
|
}
|
|
_ => panic!("payload variant mismatch"),
|
|
}
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn duplicate_chapter_content_while_pending_is_skipped(pool: PgPool) {
|
|
let chapter_id = Uuid::new_v4();
|
|
let p = chapter_content_payload(chapter_id);
|
|
|
|
let first = jobs::enqueue(&pool, &p).await.unwrap();
|
|
assert!(matches!(first, EnqueueResult::Inserted(_)));
|
|
|
|
let second = jobs::enqueue(&pool, &p).await.unwrap();
|
|
assert!(matches!(second, EnqueueResult::Skipped));
|
|
|
|
assert_eq!(job_count(&pool).await, 1);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn duplicate_after_done_releases_dedup_slot(pool: PgPool) {
|
|
let chapter_id = Uuid::new_v4();
|
|
let p = chapter_content_payload(chapter_id);
|
|
|
|
let first_id = match jobs::enqueue(&pool, &p).await.unwrap() {
|
|
EnqueueResult::Inserted(id) => id,
|
|
EnqueueResult::Skipped => panic!("first enqueue should insert"),
|
|
};
|
|
// Move the first job out of (pending|running) so the partial index drops it.
|
|
sqlx::query("UPDATE crawler_jobs SET state = 'done' WHERE id = $1")
|
|
.bind(first_id)
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
|
|
let second = jobs::enqueue(&pool, &p).await.unwrap();
|
|
assert!(
|
|
matches!(second, EnqueueResult::Inserted(_)),
|
|
"after done the chapter_id slot is free again"
|
|
);
|
|
assert_eq!(job_count(&pool).await, 2);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn different_chapter_ids_can_coexist(pool: PgPool) {
|
|
let p1 = chapter_content_payload(Uuid::new_v4());
|
|
let p2 = chapter_content_payload(Uuid::new_v4());
|
|
assert!(matches!(
|
|
jobs::enqueue(&pool, &p1).await.unwrap(),
|
|
EnqueueResult::Inserted(_)
|
|
));
|
|
assert!(matches!(
|
|
jobs::enqueue(&pool, &p2).await.unwrap(),
|
|
EnqueueResult::Inserted(_)
|
|
));
|
|
assert_eq!(job_count(&pool).await, 2);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn non_chapter_content_payloads_are_never_deduped(pool: PgPool) {
|
|
let p = sync_manga_payload("foo");
|
|
assert!(matches!(
|
|
jobs::enqueue(&pool, &p).await.unwrap(),
|
|
EnqueueResult::Inserted(_)
|
|
));
|
|
assert!(matches!(
|
|
jobs::enqueue(&pool, &p).await.unwrap(),
|
|
EnqueueResult::Inserted(_)
|
|
));
|
|
assert_eq!(job_count(&pool).await, 2);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn lease_marks_running_and_bumps_attempts_and_sets_leased_until(pool: PgPool) {
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
EnqueueResult::Skipped => unreachable!(),
|
|
};
|
|
|
|
let leases = jobs::lease(&pool, None, 10, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(leases.len(), 1);
|
|
let lease = &leases[0];
|
|
assert_eq!(lease.id, id);
|
|
assert_eq!(lease.attempts, 1);
|
|
|
|
assert_eq!(job_state(&pool, id).await, "running");
|
|
|
|
let leased_until: Option<chrono::DateTime<chrono::Utc>> =
|
|
sqlx::query_scalar("SELECT leased_until FROM crawler_jobs WHERE id = $1")
|
|
.bind(id)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
let leased_until = leased_until.expect("leased_until set");
|
|
assert!(leased_until > chrono::Utc::now());
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn renew_extends_leased_until_while_running(pool: PgPool) {
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
EnqueueResult::Skipped => unreachable!(),
|
|
};
|
|
|
|
// Lease with a short window, then collapse leased_until to the recent
|
|
// past so the renew is unambiguously an extension.
|
|
let leases = jobs::lease(&pool, None, 1, Duration::from_secs(5))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(leases.len(), 1);
|
|
sqlx::query("UPDATE crawler_jobs SET leased_until = now() - interval '1 second' WHERE id = $1")
|
|
.bind(id)
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
|
|
let still_owned = jobs::renew(&pool, id, leases[0].lease_generation, Duration::from_secs(120))
|
|
.await
|
|
.unwrap();
|
|
assert!(still_owned, "renew on a running job returns true");
|
|
|
|
let leased_until: chrono::DateTime<chrono::Utc> =
|
|
sqlx::query_scalar("SELECT leased_until FROM crawler_jobs WHERE id = $1")
|
|
.bind(id)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert!(
|
|
leased_until > chrono::Utc::now() + chrono::Duration::seconds(60),
|
|
"leased_until pushed ~120s into the future"
|
|
);
|
|
assert_eq!(job_state(&pool, id).await, "running");
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn renew_is_noop_once_job_no_longer_running(pool: PgPool) {
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
EnqueueResult::Skipped => unreachable!(),
|
|
};
|
|
let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
// Job completes — heartbeat should now see it's no longer ours.
|
|
jobs::ack_done(&pool, leases[0].id, leases[0].lease_generation)
|
|
.await
|
|
.unwrap();
|
|
|
|
let still_owned = jobs::renew(&pool, id, leases[0].lease_generation, Duration::from_secs(120))
|
|
.await
|
|
.unwrap();
|
|
assert!(!still_owned, "renew on a non-running job returns false");
|
|
assert_eq!(job_state(&pool, id).await, "done");
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn lease_with_kind_filter_only_matches_that_kind(pool: PgPool) {
|
|
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 leases = jobs::lease(
|
|
&pool,
|
|
Some(KIND_SYNC_CHAPTER_CONTENT),
|
|
10,
|
|
Duration::from_secs(60),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(leases.len(), 1, "only chapter content payload leases");
|
|
assert_eq!(leases[0].id, chapter_id);
|
|
// sync_manga is still pending
|
|
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.
|
|
let mut ids = Vec::new();
|
|
for _ in 0..4 {
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
ids.push(id);
|
|
}
|
|
|
|
let (a, b) = tokio::join!(
|
|
jobs::lease(&pool, None, 2, Duration::from_secs(60)),
|
|
jobs::lease(&pool, None, 2, Duration::from_secs(60)),
|
|
);
|
|
let a = a.unwrap();
|
|
let b = b.unwrap();
|
|
let mut seen: Vec<Uuid> = a.iter().chain(b.iter()).map(|l| l.id).collect();
|
|
seen.sort();
|
|
seen.dedup();
|
|
let count = a.len() + b.len();
|
|
assert_eq!(
|
|
seen.len(),
|
|
count,
|
|
"no id appears in both lease results (SKIP LOCKED)"
|
|
);
|
|
assert!(count >= 2, "at least one lease saw work");
|
|
assert!(count <= 4);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn stale_running_lease_can_be_reclaimed(pool: PgPool) {
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
|
|
let first = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(first.len(), 1);
|
|
// Pretend the worker crashed: rewind leased_until into the past.
|
|
sqlx::query("UPDATE crawler_jobs SET leased_until = now() - interval '1 minute' WHERE id = $1")
|
|
.bind(id)
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
|
|
let second = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(second.len(), 1, "stale running row was re-leased");
|
|
assert_eq!(second[0].id, id);
|
|
assert_eq!(second[0].attempts, 2, "attempts bumped again");
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reclaim_orphaned_resets_only_expired_running_jobs(pool: PgPool) {
|
|
// Three jobs: (a) running with an expired lease — a crash orphan that
|
|
// must be reclaimed; (b) running with a live lease — a healthy peer's
|
|
// in-flight job that must be left untouched; (c) pending — irrelevant,
|
|
// must be untouched. Reclaim must touch only (a), refund its attempt,
|
|
// and report exactly 1.
|
|
let orphan = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let live = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let pending = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
|
|
// Lease the two that should become running (attempts -> 1 each).
|
|
jobs::lease(&pool, None, 10, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
// Leave `pending` pending by returning it (release refunds its attempt).
|
|
// Test only has the id, not a Lease, so use the unowned variant.
|
|
jobs::release_unowned(&pool, pending).await.unwrap();
|
|
assert_eq!(job_state(&pool, pending).await, "pending");
|
|
|
|
// Orphan: rewind its lease into the past (worker crashed). Live: keep
|
|
// its lease in the future (peer still heartbeating).
|
|
sqlx::query("UPDATE crawler_jobs SET leased_until = now() - interval '1 minute' WHERE id = $1")
|
|
.bind(orphan)
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(job_state(&pool, live).await, "running");
|
|
assert_eq!(job_attempts(&pool, live).await, 1);
|
|
|
|
let reclaimed = jobs::reclaim_orphaned(&pool).await.unwrap();
|
|
assert_eq!(reclaimed, 1, "only the expired-lease running job is reclaimed");
|
|
|
|
// Orphan: back to pending, attempt refunded, lease cleared.
|
|
assert_eq!(job_state(&pool, orphan).await, "pending");
|
|
assert_eq!(
|
|
job_attempts(&pool, orphan).await,
|
|
0,
|
|
"reclaim refunds the crashed attempt"
|
|
);
|
|
let orphan_leased: Option<chrono::DateTime<chrono::Utc>> =
|
|
sqlx::query_scalar("SELECT leased_until FROM crawler_jobs WHERE id = $1")
|
|
.bind(orphan)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert!(orphan_leased.is_none(), "reclaim clears the lease");
|
|
|
|
// Live peer job: untouched.
|
|
assert_eq!(job_state(&pool, live).await, "running");
|
|
assert_eq!(job_attempts(&pool, live).await, 1);
|
|
|
|
// Pending job: untouched.
|
|
assert_eq!(job_state(&pool, pending).await, "pending");
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn ack_done_transitions_state_and_clears_lease(pool: PgPool) {
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
jobs::ack_done(&pool, leases[0].id, leases[0].lease_generation)
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(job_state(&pool, id).await, "done");
|
|
let leased_until: Option<chrono::DateTime<chrono::Utc>> =
|
|
sqlx::query_scalar("SELECT leased_until FROM crawler_jobs WHERE id = $1")
|
|
.bind(id)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert!(leased_until.is_none());
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn ack_failed_under_max_returns_to_pending_with_future_schedule(pool: PgPool) {
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
let lease = &leases[0];
|
|
jobs::ack_failed(
|
|
&pool,
|
|
lease.id,
|
|
"boom",
|
|
lease.attempts,
|
|
lease.max_attempts,
|
|
lease.lease_generation,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(job_state(&pool, id).await, "pending");
|
|
|
|
let (scheduled_at, last_error): (chrono::DateTime<chrono::Utc>, Option<String>) =
|
|
sqlx::query_as("SELECT scheduled_at, last_error FROM crawler_jobs WHERE id = $1")
|
|
.bind(id)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert!(scheduled_at > chrono::Utc::now());
|
|
assert_eq!(last_error.as_deref(), Some("boom"));
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn ack_failed_at_max_marks_dead(pool: PgPool) {
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
// Force a single lease then mark "this was attempt N where N == max_attempts".
|
|
let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
let lease = &leases[0];
|
|
jobs::ack_failed(
|
|
&pool,
|
|
lease.id,
|
|
"final boom",
|
|
lease.max_attempts,
|
|
lease.max_attempts,
|
|
lease.lease_generation,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(job_state(&pool, id).await, "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_eq!(last_error.as_deref(), Some("final boom"));
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn ack_done_no_ops_when_lease_was_stolen(pool: PgPool) {
|
|
// Worker A's lease expires, worker B re-leases the job (state stays
|
|
// 'running' but attempts++ and leased_until refreshed). A late
|
|
// ack_done from worker A must not clobber B's progress.
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
// Worker A grabs the lease, but its lease expires immediately.
|
|
let a_leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
sqlx::query("UPDATE crawler_jobs SET leased_until = now() - interval '1 minute' WHERE id = $1")
|
|
.bind(id)
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
// Worker B re-leases the expired-but-still-running job.
|
|
let b_leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(b_leases.len(), 1);
|
|
assert_eq!(b_leases[0].attempts, 2, "re-lease bumps attempts");
|
|
assert_ne!(
|
|
a_leases[0].lease_generation, b_leases[0].lease_generation,
|
|
"re-lease must bump lease_generation so A's late ack can be distinguished"
|
|
);
|
|
|
|
// Worker A's late ack_done — now scoped to A's lease_generation by
|
|
// migration 0032. The row stays `running` with B's progress
|
|
// intact; A's stale ack matches nothing and is a no-op.
|
|
jobs::ack_done(&pool, a_leases[0].id, a_leases[0].lease_generation)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
job_state(&pool, id).await,
|
|
"running",
|
|
"A's stale ack_done must NOT clobber B's running lease"
|
|
);
|
|
// Finalize: worker B acks done with its own generation.
|
|
jobs::ack_done(&pool, b_leases[0].id, b_leases[0].lease_generation)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(job_state(&pool, id).await, "done");
|
|
assert_eq!(job_attempts(&pool, id).await, 2);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn ack_failed_no_ops_when_state_is_not_running(pool: PgPool) {
|
|
// After a job transitions to 'done', a stale ack_failed (e.g. a
|
|
// worker that finished work and queued its ack but then handed off
|
|
// before the SQL ran) must not flip the state back to 'pending' or
|
|
// 'dead'. The `state = 'running'` predicate enforces this.
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
jobs::ack_done(&pool, leases[0].id, leases[0].lease_generation)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(job_state(&pool, id).await, "done");
|
|
|
|
// Late ack_failed arrives. Must be a no-op.
|
|
jobs::ack_failed(&pool, leases[0].id, "late", 1, 5, leases[0].lease_generation)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
job_state(&pool, id).await,
|
|
"done",
|
|
"late ack_failed must not resurrect a done job"
|
|
);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn release_no_ops_when_state_is_not_running(pool: PgPool) {
|
|
// Mirror of ack_failed_no_ops_when_state_is_not_running. release also
|
|
// decrements `attempts`, which would corrupt a re-leased job's
|
|
// attempt count if the guard were missing.
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
jobs::ack_done(&pool, leases[0].id, leases[0].lease_generation)
|
|
.await
|
|
.unwrap();
|
|
let attempts_before = job_attempts(&pool, id).await;
|
|
|
|
// Late release arrives.
|
|
jobs::release(&pool, leases[0].id, leases[0].lease_generation)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
job_state(&pool, id).await,
|
|
"done",
|
|
"late release must not flip a done job back to pending"
|
|
);
|
|
assert_eq!(
|
|
job_attempts(&pool, id).await,
|
|
attempts_before,
|
|
"late release must not decrement attempts of a non-running job"
|
|
);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn release_returns_to_pending_and_undoes_attempt_increment(pool: PgPool) {
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(leases[0].attempts, 1);
|
|
jobs::release(&pool, leases[0].id, leases[0].lease_generation)
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(job_state(&pool, id).await, "pending");
|
|
assert_eq!(job_attempts(&pool, id).await, 0);
|
|
let leased_until: Option<chrono::DateTime<chrono::Utc>> =
|
|
sqlx::query_scalar("SELECT leased_until FROM crawler_jobs WHERE id = $1")
|
|
.bind(id)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert!(leased_until.is_none());
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reap_terminal_deletes_old_done_and_dead_keeps_fresh_and_active(pool: PgPool) {
|
|
// Helper: enqueue a fresh pending job and return its id.
|
|
async fn enqueue_one(pool: &PgPool) -> Uuid {
|
|
match jobs::enqueue(pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
}
|
|
}
|
|
|
|
let old_done = enqueue_one(&pool).await;
|
|
let old_dead = enqueue_one(&pool).await;
|
|
let fresh_done = enqueue_one(&pool).await;
|
|
let fresh_dead = enqueue_one(&pool).await;
|
|
let old_pending = enqueue_one(&pool).await;
|
|
|
|
// Old terminal rows (10 days) in both terminal states — both must reap.
|
|
sqlx::query("UPDATE crawler_jobs SET state='done', updated_at = now() - interval '10 days' WHERE id = $1")
|
|
.bind(old_done).execute(&pool).await.unwrap();
|
|
sqlx::query("UPDATE crawler_jobs SET state='dead', updated_at = now() - interval '10 days' WHERE id = $1")
|
|
.bind(old_dead).execute(&pool).await.unwrap();
|
|
// Fresh terminal rows — inside the retention window, kept.
|
|
sqlx::query("UPDATE crawler_jobs SET state='done' WHERE id = $1")
|
|
.bind(fresh_done).execute(&pool).await.unwrap();
|
|
sqlx::query("UPDATE crawler_jobs SET state='dead' WHERE id = $1")
|
|
.bind(fresh_dead).execute(&pool).await.unwrap();
|
|
// Old but still active (pending) — never reaped regardless of age.
|
|
sqlx::query("UPDATE crawler_jobs SET updated_at = now() - interval '10 days' WHERE id = $1")
|
|
.bind(old_pending).execute(&pool).await.unwrap();
|
|
|
|
let deleted = jobs::reap_terminal(&pool, 7).await.unwrap();
|
|
assert_eq!(deleted, 2, "both old done and old dead rows are reaped");
|
|
|
|
let mut remaining: Vec<Uuid> = sqlx::query_scalar("SELECT id FROM crawler_jobs")
|
|
.fetch_all(&pool)
|
|
.await
|
|
.unwrap();
|
|
remaining.sort();
|
|
let mut expected = vec![fresh_done, fresh_dead, old_pending];
|
|
expected.sort();
|
|
assert_eq!(remaining, expected, "fresh terminal + active-pending rows survive");
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn lease_ties_on_scheduled_at_break_by_created_at(pool: PgPool) {
|
|
// Locks in the tiebreaker that lets enqueue order survive the lease
|
|
// step: when many jobs share `scheduled_at` (the common cron-batch
|
|
// case), the worker must pick the earliest-inserted row, not whatever
|
|
// Postgres returns in heap order. The enqueue path inserts chapters
|
|
// in chapter-number order, so this tiebreaker is what makes "queue
|
|
// in rising order" observable at the dequeue side too.
|
|
let a = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let b = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let c = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
|
|
// Pin `scheduled_at` to a single literal instant (shared across all
|
|
// three rows — `now()` would yield a different microsecond per UPDATE
|
|
// and make scheduled_at the actual sort key). Reverse `created_at`
|
|
// against insertion order so heap order would give the wrong answer.
|
|
let shared_scheduled = chrono::Utc::now() - chrono::Duration::hours(1);
|
|
sqlx::query(
|
|
"UPDATE crawler_jobs \
|
|
SET scheduled_at = $2, \
|
|
created_at = $3 \
|
|
WHERE id = $1",
|
|
)
|
|
.bind(a)
|
|
.bind(shared_scheduled)
|
|
.bind(chrono::Utc::now() - chrono::Duration::seconds(10))
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
sqlx::query(
|
|
"UPDATE crawler_jobs \
|
|
SET scheduled_at = $2, \
|
|
created_at = $3 \
|
|
WHERE id = $1",
|
|
)
|
|
.bind(b)
|
|
.bind(shared_scheduled)
|
|
.bind(chrono::Utc::now() - chrono::Duration::seconds(20))
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
sqlx::query(
|
|
"UPDATE crawler_jobs \
|
|
SET scheduled_at = $2, \
|
|
created_at = $3 \
|
|
WHERE id = $1",
|
|
)
|
|
.bind(c)
|
|
.bind(shared_scheduled)
|
|
.bind(chrono::Utc::now() - chrono::Duration::seconds(30))
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
|
|
let leases = jobs::lease(&pool, None, 10, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
let order: Vec<Uuid> = leases.iter().map(|l| l.id).collect();
|
|
assert_eq!(
|
|
order,
|
|
vec![c, b, a],
|
|
"lease must return jobs in created_at order when scheduled_at ties"
|
|
);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reap_terminal_zero_is_a_no_op(pool: PgPool) {
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
sqlx::query("UPDATE crawler_jobs SET state='done', updated_at = now() - interval '999 days' WHERE id = $1")
|
|
.bind(id)
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
|
|
let deleted = jobs::reap_terminal(&pool, 0).await.unwrap();
|
|
assert_eq!(deleted, 0);
|
|
assert_eq!(job_count(&pool).await, 1);
|
|
}
|
|
|
|
/// Migration 0031 ships a pre-dedup step (demote all-but-lowest-id
|
|
/// duplicates to `dead`) so it can apply cleanly on dirty production
|
|
/// databases where the prior buggy `enqueue_pages` left duplicate
|
|
/// `analyze_page` rows in `pending`/`running`. This test simulates the
|
|
/// dirty state by dropping the unique index, inserting duplicates, then
|
|
/// running the migration's pre-dedup SQL — the same SQL the migration
|
|
/// file runs — and confirming the duplicates land in `dead` so the
|
|
/// subsequent index creation would succeed.
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn migration_0031_prededup_demotes_duplicate_analyze_page_rows(pool: PgPool) {
|
|
// 1) Simulate pre-0031 dirty state: drop the dedup index so a second
|
|
// pending insert for the same page_id is accepted.
|
|
sqlx::query("DROP INDEX IF EXISTS crawler_jobs_analyze_page_dedup_idx")
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
|
|
let page_a = Uuid::new_v4();
|
|
let page_b = Uuid::new_v4();
|
|
let make_row = |page_id: Uuid| {
|
|
serde_json::json!({
|
|
"kind": "analyze_page",
|
|
"page_id": page_id,
|
|
"force": false
|
|
})
|
|
};
|
|
// Two pending duplicates for page A, plus one running + one pending
|
|
// for page B (the migration handles all in-flight states).
|
|
let a1: Uuid = sqlx::query_scalar(
|
|
"INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'pending') RETURNING id",
|
|
)
|
|
.bind(make_row(page_a))
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
let a2: Uuid = sqlx::query_scalar(
|
|
"INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'pending') RETURNING id",
|
|
)
|
|
.bind(make_row(page_a))
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
let b1: Uuid = sqlx::query_scalar(
|
|
"INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'running') RETURNING id",
|
|
)
|
|
.bind(make_row(page_b))
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
let b2: Uuid = sqlx::query_scalar(
|
|
"INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'pending') RETURNING id",
|
|
)
|
|
.bind(make_row(page_b))
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
|
|
// 2) Run the migration's pre-dedup statement verbatim.
|
|
sqlx::query(
|
|
"UPDATE crawler_jobs \
|
|
SET state = 'dead', \
|
|
last_error = COALESCE(last_error, '') || \
|
|
CASE WHEN last_error IS NULL OR last_error = '' THEN '' \
|
|
ELSE ' | ' END || \
|
|
'pre-0031 duplicate analyze_page; superseded by earlier sibling', \
|
|
updated_at = now() \
|
|
WHERE id IN ( \
|
|
SELECT id FROM ( \
|
|
SELECT id, \
|
|
row_number() OVER ( \
|
|
PARTITION BY payload->>'page_id' \
|
|
ORDER BY id \
|
|
) AS rn \
|
|
FROM crawler_jobs \
|
|
WHERE payload->>'kind' = 'analyze_page' \
|
|
AND state IN ('pending', 'running') \
|
|
) ranked \
|
|
WHERE rn > 1 \
|
|
)",
|
|
)
|
|
.execute(&pool)
|
|
.await
|
|
.unwrap();
|
|
|
|
// 3) The earliest-id row per page_id stays in-flight; later
|
|
// duplicates moved to `dead` with the curator-recoverable
|
|
// last_error marker. Compute the expected winner per page by
|
|
// Postgres' UUID ordering (sql min, not Rust Ord — they differ).
|
|
let winner_a: Uuid =
|
|
sqlx::query_scalar("SELECT LEAST($1::uuid, $2::uuid)")
|
|
.bind(a1)
|
|
.bind(a2)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
let loser_a = if winner_a == a1 { a2 } else { a1 };
|
|
let winner_b: Uuid =
|
|
sqlx::query_scalar("SELECT LEAST($1::uuid, $2::uuid)")
|
|
.bind(b1)
|
|
.bind(b2)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
let loser_b = if winner_b == b1 { b2 } else { b1 };
|
|
|
|
let state_a_winner: String =
|
|
sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1")
|
|
.bind(winner_a)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
let state_a_loser: String =
|
|
sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1")
|
|
.bind(loser_a)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert!(
|
|
state_a_winner == "pending" || state_a_winner == "running",
|
|
"winner kept in-flight"
|
|
);
|
|
assert_eq!(state_a_loser, "dead", "loser demoted to dead");
|
|
|
|
let state_b_winner: String =
|
|
sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1")
|
|
.bind(winner_b)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
let state_b_loser: String =
|
|
sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1")
|
|
.bind(loser_b)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert!(state_b_winner == "pending" || state_b_winner == "running");
|
|
assert_eq!(state_b_loser, "dead");
|
|
|
|
// 4) After dedup, the index is creatable. Migration would now succeed.
|
|
sqlx::query(
|
|
"CREATE UNIQUE INDEX crawler_jobs_analyze_page_dedup_idx \
|
|
ON crawler_jobs ((payload->>'page_id')) \
|
|
WHERE state IN ('pending', 'running') \
|
|
AND payload->>'kind' = 'analyze_page'",
|
|
)
|
|
.execute(&pool)
|
|
.await
|
|
.expect("post-dedup index creation must succeed");
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Lease-generation race (0.87.26). Closes the gap that 0.87.20 narrowed:
|
|
// `ack_done` / `ack_failed` matched on `(id, state='running')` only, so a
|
|
// late ack from a dead lease (id same, generation different) could clobber
|
|
// the successor's running row. The new `lease_generation` column scopes
|
|
// the guard to the *specific* lease the worker was holding.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn lease_assigns_strictly_increasing_generation_per_release(pool: PgPool) {
|
|
// Each lease (the very first one, AND every re-lease after release /
|
|
// expiry / reclaim) must carry a strictly-increasing generation so
|
|
// downstream guards can distinguish "the lease that was leased" from
|
|
// "the lease that's currently leased."
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let a = jobs::lease(&pool, None, 1, Duration::from_secs(60)).await.unwrap();
|
|
assert_eq!(a.len(), 1);
|
|
assert!(a[0].lease_generation >= 1, "first lease must carry a positive generation");
|
|
|
|
jobs::release_unowned(&pool, id).await.unwrap();
|
|
let b = jobs::lease(&pool, None, 1, Duration::from_secs(60)).await.unwrap();
|
|
assert_eq!(b.len(), 1);
|
|
assert!(
|
|
b[0].lease_generation > a[0].lease_generation,
|
|
"successor lease must carry a strictly-higher generation: a={}, b={}",
|
|
a[0].lease_generation,
|
|
b[0].lease_generation
|
|
);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn ack_done_from_dead_lease_after_release_unowned_is_a_no_op(pool: PgPool) {
|
|
// The race that motivated the lease_generation column. A worker A
|
|
// holds a lease. An external actor (force-analyze, reclaim) calls
|
|
// `release_unowned` on the row, putting it back to `pending`. A
|
|
// worker B re-leases the same id. THEN A's dispatch finally returns
|
|
// and acks done. Before this fix, A's `ack_done` matched
|
|
// `state='running'` (set by B) and clobbered B's lease — A's stale
|
|
// result became the row's `done` payload, B's still-in-flight work
|
|
// was lost. Now A's `ack_done(id, generation=A)` finds nothing
|
|
// because B's lease bumped the generation, so the row stays
|
|
// `running` for B.
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let a = jobs::lease(&pool, None, 1, Duration::from_secs(300))
|
|
.await
|
|
.unwrap()[0]
|
|
.clone();
|
|
assert_eq!(a.id, id);
|
|
|
|
// External release: e.g. force-analyze flipping payload.force = true
|
|
// and dropping the in-flight lease so the new payload is picked up.
|
|
jobs::release_unowned(&pool, id).await.unwrap();
|
|
assert_eq!(job_state(&pool, id).await, "pending");
|
|
|
|
let b = jobs::lease(&pool, None, 1, Duration::from_secs(300))
|
|
.await
|
|
.unwrap()[0]
|
|
.clone();
|
|
assert_eq!(b.id, id);
|
|
assert_ne!(
|
|
a.lease_generation, b.lease_generation,
|
|
"successor must hold a different lease_generation"
|
|
);
|
|
assert_eq!(job_state(&pool, id).await, "running");
|
|
|
|
// A's stale ack arrives. MUST be a no-op — the row stays running
|
|
// with B's progress intact.
|
|
jobs::ack_done(&pool, a.id, a.lease_generation).await.unwrap();
|
|
assert_eq!(
|
|
job_state(&pool, id).await,
|
|
"running",
|
|
"stale ack_done from a dead lease must not clobber the successor's lease"
|
|
);
|
|
|
|
// B finishes normally with its own generation.
|
|
jobs::ack_done(&pool, b.id, b.lease_generation).await.unwrap();
|
|
assert_eq!(job_state(&pool, id).await, "done");
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn ack_failed_from_dead_lease_after_release_unowned_is_a_no_op(pool: PgPool) {
|
|
// Mirror of the ack_done case for ack_failed. A late ack_failed
|
|
// from A would otherwise either (a) flip B's running row back to
|
|
// `pending` with a stale error and a stale backoff, or (b) at
|
|
// max_attempts, dead-letter B's in-flight work.
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let a = jobs::lease(&pool, None, 1, Duration::from_secs(300))
|
|
.await
|
|
.unwrap()[0]
|
|
.clone();
|
|
jobs::release_unowned(&pool, id).await.unwrap();
|
|
let b = jobs::lease(&pool, None, 1, Duration::from_secs(300))
|
|
.await
|
|
.unwrap()[0]
|
|
.clone();
|
|
assert_eq!(b.id, id);
|
|
|
|
// A's stale ack_failed arrives.
|
|
jobs::ack_failed(&pool, a.id, "A late error", a.attempts, a.max_attempts, a.lease_generation)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
job_state(&pool, id).await,
|
|
"running",
|
|
"stale ack_failed from a dead lease must not clobber the successor"
|
|
);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn renew_from_dead_lease_is_a_no_op(pool: PgPool) {
|
|
// A's heartbeat must not extend B's lease window after A was
|
|
// released and B re-leased. Without the generation check, an A-side
|
|
// renew would push B's `leased_until` further into the future,
|
|
// delaying reclaim of a B-side crash.
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let a = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap()[0]
|
|
.clone();
|
|
jobs::release_unowned(&pool, id).await.unwrap();
|
|
let _b = jobs::lease(&pool, None, 1, Duration::from_secs(60))
|
|
.await
|
|
.unwrap();
|
|
|
|
// A's heartbeat arrives. Should report not-owned and skip.
|
|
let still_owned = jobs::renew(&pool, a.id, a.lease_generation, Duration::from_secs(900))
|
|
.await
|
|
.unwrap();
|
|
assert!(
|
|
!still_owned,
|
|
"renew must report not-owned when the lease_generation no longer matches"
|
|
);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn release_unowned_bumps_generation_even_when_lease_is_held(pool: PgPool) {
|
|
// Belt-and-braces: release_unowned exists precisely for cases
|
|
// where the caller has no `Lease` in hand (force-analyze, ops
|
|
// tools). It must bump the generation unconditionally so any
|
|
// ack from the in-flight lease is invalidated.
|
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
EnqueueResult::Inserted(id) => id,
|
|
_ => unreachable!(),
|
|
};
|
|
let a = jobs::lease(&pool, None, 1, Duration::from_secs(300))
|
|
.await
|
|
.unwrap()[0]
|
|
.clone();
|
|
let gen_before: i64 = sqlx::query_scalar(
|
|
"SELECT lease_generation FROM crawler_jobs WHERE id = $1",
|
|
)
|
|
.bind(id)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(gen_before, a.lease_generation);
|
|
|
|
jobs::release_unowned(&pool, id).await.unwrap();
|
|
let gen_after: i64 = sqlx::query_scalar(
|
|
"SELECT lease_generation FROM crawler_jobs WHERE id = $1",
|
|
)
|
|
.bind(id)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert!(
|
|
gen_after > gen_before,
|
|
"release_unowned must bump lease_generation: before={gen_before}, after={gen_after}"
|
|
);
|
|
}
|