fix(crawler): reap dead jobs alongside done ones
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>
This commit is contained in:
2
backend/Cargo.lock
generated
2
backend/Cargo.lock
generated
@@ -1558,7 +1558,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mangalord"
|
name = "mangalord"
|
||||||
version = "0.93.7"
|
version = "0.93.8"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"argon2",
|
"argon2",
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mangalord"
|
name = "mangalord"
|
||||||
version = "0.93.7"
|
version = "0.93.8"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
default-run = "mangalord"
|
default-run = "mangalord"
|
||||||
|
|
||||||
|
|||||||
@@ -301,9 +301,9 @@ impl CronContext {
|
|||||||
}
|
}
|
||||||
Err(e) => tracing::error!(?e, "cron: enqueue_bookmarked_pending failed"),
|
Err(e) => tracing::error!(?e, "cron: enqueue_bookmarked_pending failed"),
|
||||||
}
|
}
|
||||||
match jobs::reap_done(pool, retention_days).await {
|
match jobs::reap_terminal(pool, retention_days).await {
|
||||||
Ok(n) => tracing::info!(reaped = n, "cron: done-job reaper finished"),
|
Ok(n) => tracing::info!(reaped = n, "cron: terminal-job reaper finished"),
|
||||||
Err(e) => tracing::error!(?e, "cron: done-job reaper failed"),
|
Err(e) => tracing::error!(?e, "cron: terminal-job reaper failed"),
|
||||||
}
|
}
|
||||||
match crate::repo::crawl_metrics::reap(pool, metrics_retention_days).await {
|
match crate::repo::crawl_metrics::reap(pool, metrics_retention_days).await {
|
||||||
Ok(n) => tracing::info!(reaped = n, "cron: crawl-metrics reaper finished"),
|
Ok(n) => tracing::info!(reaped = n, "cron: crawl-metrics reaper finished"),
|
||||||
|
|||||||
@@ -485,16 +485,19 @@ pub async fn reclaim_orphaned(pool: &PgPool) -> sqlx::Result<u64> {
|
|||||||
Ok(result.rows_affected())
|
Ok(result.rows_affected())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Delete `done` jobs whose `updated_at` is older than `retention_days`
|
/// Delete **terminal** jobs (`done` or `dead`) whose `updated_at` is older
|
||||||
/// days. `0` disables the reaper without touching the table. Returns the
|
/// than `retention_days` days. Both states are end-of-life — `done`
|
||||||
/// number of rows removed.
|
/// succeeded, `dead` exhausted its retries — and neither is ever leased
|
||||||
pub async fn reap_done(pool: &PgPool, retention_days: u32) -> sqlx::Result<u64> {
|
/// again, so reaping only `done` left `dead` rows to accumulate forever.
|
||||||
|
/// `pending` / `running` are active and never touched. `0` disables the
|
||||||
|
/// reaper without touching the table. Returns the number of rows removed.
|
||||||
|
pub async fn reap_terminal(pool: &PgPool, retention_days: u32) -> sqlx::Result<u64> {
|
||||||
if retention_days == 0 {
|
if retention_days == 0 {
|
||||||
return Ok(0);
|
return Ok(0);
|
||||||
}
|
}
|
||||||
let result = sqlx::query(
|
let result = sqlx::query(
|
||||||
"DELETE FROM crawler_jobs \
|
"DELETE FROM crawler_jobs \
|
||||||
WHERE state = 'done' \
|
WHERE state IN ('done', 'dead') \
|
||||||
AND updated_at < now() - ($1::bigint || ' days')::interval",
|
AND updated_at < now() - ($1::bigint || ' days')::interval",
|
||||||
)
|
)
|
||||||
.bind(retention_days as i64)
|
.bind(retention_days as i64)
|
||||||
|
|||||||
@@ -187,7 +187,7 @@ pub async fn list_ops(
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Delete metric rows older than `retention_days`. `0` disables the reaper
|
/// Delete metric rows older than `retention_days`. `0` disables the reaper
|
||||||
/// (returns 0 without touching the table). Mirrors `jobs::reap_done`.
|
/// (returns 0 without touching the table). Mirrors `jobs::reap_terminal`.
|
||||||
pub async fn reap(pool: &PgPool, retention_days: u32) -> sqlx::Result<u64> {
|
pub async fn reap(pool: &PgPool, retention_days: u32) -> sqlx::Result<u64> {
|
||||||
if retention_days == 0 {
|
if retention_days == 0 {
|
||||||
return Ok(0);
|
return Ok(0);
|
||||||
|
|||||||
@@ -903,9 +903,8 @@ pub struct JobHistoryFilter<'a> {
|
|||||||
/// manga/chapter/page context (best-effort) so the table can label rows.
|
/// manga/chapter/page context (best-effort) so the table can label rows.
|
||||||
/// Returns the page slice plus the filtered total for pagination.
|
/// Returns the page slice plus the filtered total for pagination.
|
||||||
///
|
///
|
||||||
/// History depth is bounded by the done-job reaper (`reap_done`): completed
|
/// History depth is bounded by the terminal-job reaper (`reap_terminal`):
|
||||||
/// jobs older than the retention window are gone. Terminal `dead` jobs
|
/// `done` and `dead` jobs older than the retention window are gone.
|
||||||
/// persist until requeued.
|
|
||||||
pub async fn list_job_history(
|
pub async fn list_job_history(
|
||||||
pool: &PgPool,
|
pool: &PgPool,
|
||||||
filter: JobHistoryFilter<'_>,
|
filter: JobHistoryFilter<'_>,
|
||||||
|
|||||||
@@ -718,42 +718,49 @@ async fn release_returns_to_pending_and_undoes_attempt_increment(pool: PgPool) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[sqlx::test(migrations = "./migrations")]
|
#[sqlx::test(migrations = "./migrations")]
|
||||||
async fn reap_done_deletes_old_rows_keeps_fresh(pool: PgPool) {
|
async fn reap_terminal_deletes_old_done_and_dead_keeps_fresh_and_active(pool: PgPool) {
|
||||||
// Two done rows: one old (updated_at 10 days ago), one fresh.
|
// Helper: enqueue a fresh pending job and return its id.
|
||||||
let old_id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
async fn enqueue_one(pool: &PgPool) -> Uuid {
|
||||||
.await
|
match jobs::enqueue(pool, &chapter_content_payload(Uuid::new_v4()))
|
||||||
.unwrap()
|
.await
|
||||||
{
|
.unwrap()
|
||||||
EnqueueResult::Inserted(id) => id,
|
{
|
||||||
_ => unreachable!(),
|
EnqueueResult::Inserted(id) => id,
|
||||||
};
|
_ => unreachable!(),
|
||||||
let fresh_id = 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")
|
sqlx::query("UPDATE crawler_jobs SET state='done', updated_at = now() - interval '10 days' WHERE id = $1")
|
||||||
.bind(old_id)
|
.bind(old_done).execute(&pool).await.unwrap();
|
||||||
.execute(&pool)
|
sqlx::query("UPDATE crawler_jobs SET state='dead', updated_at = now() - interval '10 days' WHERE id = $1")
|
||||||
.await
|
.bind(old_dead).execute(&pool).await.unwrap();
|
||||||
.unwrap();
|
// Fresh terminal rows — inside the retention window, kept.
|
||||||
sqlx::query("UPDATE crawler_jobs SET state='done' WHERE id = $1")
|
sqlx::query("UPDATE crawler_jobs SET state='done' WHERE id = $1")
|
||||||
.bind(fresh_id)
|
.bind(fresh_done).execute(&pool).await.unwrap();
|
||||||
.execute(&pool)
|
sqlx::query("UPDATE crawler_jobs SET state='dead' WHERE id = $1")
|
||||||
.await
|
.bind(fresh_dead).execute(&pool).await.unwrap();
|
||||||
.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_done(&pool, 7).await.unwrap();
|
let deleted = jobs::reap_terminal(&pool, 7).await.unwrap();
|
||||||
assert_eq!(deleted, 1);
|
assert_eq!(deleted, 2, "both old done and old dead rows are reaped");
|
||||||
|
|
||||||
let remaining: Vec<Uuid> = sqlx::query_scalar("SELECT id FROM crawler_jobs ORDER BY id")
|
let mut remaining: Vec<Uuid> = sqlx::query_scalar("SELECT id FROM crawler_jobs")
|
||||||
.fetch_all(&pool)
|
.fetch_all(&pool)
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
assert_eq!(remaining, vec![fresh_id], "only fresh row remains");
|
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")]
|
#[sqlx::test(migrations = "./migrations")]
|
||||||
@@ -840,7 +847,7 @@ async fn lease_ties_on_scheduled_at_break_by_created_at(pool: PgPool) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[sqlx::test(migrations = "./migrations")]
|
#[sqlx::test(migrations = "./migrations")]
|
||||||
async fn reap_done_zero_is_a_no_op(pool: PgPool) {
|
async fn reap_terminal_zero_is_a_no_op(pool: PgPool) {
|
||||||
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
|
||||||
.await
|
.await
|
||||||
.unwrap()
|
.unwrap()
|
||||||
@@ -854,7 +861,7 @@ async fn reap_done_zero_is_a_no_op(pool: PgPool) {
|
|||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
let deleted = jobs::reap_done(&pool, 0).await.unwrap();
|
let deleted = jobs::reap_terminal(&pool, 0).await.unwrap();
|
||||||
assert_eq!(deleted, 0);
|
assert_eq!(deleted, 0);
|
||||||
assert_eq!(job_count(&pool).await, 1);
|
assert_eq!(job_count(&pool).await, 1);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "mangalord-frontend",
|
"name": "mangalord-frontend",
|
||||||
"version": "0.93.7",
|
"version": "0.93.8",
|
||||||
"private": true,
|
"private": true,
|
||||||
"type": "module",
|
"type": "module",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
|
|||||||
Reference in New Issue
Block a user