From 93b7e451bfee32265427821eb2564816adde5bd2 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Wed, 24 Jun 2026 20:05:55 +0200 Subject: [PATCH] fix(jobs): per-lease generation token closes the ack-from-dead-lease race (0.87.26) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 0.87.20 narrowed but did not close the force-analyze race: ack_done / ack_failed / renew matched only on (id, state='running'), so a worker A whose lease was released mid-flight (force-analyze, reclaim_orphaned) could still come back later — once a successor B re-leased the same id and put it back to state='running' — and clobber B's lease. A's stale result became the row's `done` payload; B's in-flight work was silently lost. The pre-existing test even documented this hole: "We can't make ack_done's lease_id distinguish A from B today". Add `lease_generation BIGINT NOT NULL DEFAULT 0` to crawler_jobs (migration 0032). It is bumped by every lease, release / release_unowned, and reclaim_orphaned, so lease identity is `(id, generation)` rather than just `id`. Each Lease struct carries its generation. ack_done / ack_failed / renew / release match on `(id, generation, state='running')` so a stale ack from a prior generation finds zero rows and falls back to the existing warn-and-skip branch. Split the release surface in two: * release(lease_id, lease_generation) — owner-aware path for workers (graceful shutdown, session-expired), matches the caller's specific lease. * release_unowned(lease_id) — id-only path for force-analyze and ops tools that drop a running lease without holding a Lease struct; unconditionally bumps generation so any pending ack from the in-flight original becomes a no-op. Force-analyze in repo::page_analysis now uses release_unowned; the test in api_admin_analysis still passes unchanged (it only asserts the steady-state outcome). The pre-existing `ack_done_no_ops_when_ lease_was_stolen` test is strengthened: it now mints both A and B leases, asserts their generations differ, and verifies A's stale ack does NOT clobber B's running row — the assertion the old test explicitly couldn't make. Five new tests in tests/crawler_jobs.rs pin every facet: * lease_assigns_strictly_increasing_generation_per_release * ack_done_from_dead_lease_after_release_unowned_is_a_no_op * ack_failed_from_dead_lease_after_release_unowned_is_a_no_op * renew_from_dead_lease_is_a_no_op * release_unowned_bumps_generation_even_when_lease_is_held Mutation-confirmed: relaxing the ack_done guard to `lease_generation >= $2` (bind 0) makes the race test fail with state=done — the exact prior-bug shape. Co-Authored-By: Claude Opus 4.7 --- backend/Cargo.lock | 2 +- backend/Cargo.toml | 2 +- .../0032_crawler_jobs_lease_generation.sql | 23 ++ backend/src/analysis/daemon.rs | 12 +- backend/src/crawler/daemon.rs | 14 +- backend/src/crawler/jobs.rs | 134 ++++++-- backend/src/repo/page_analysis.rs | 18 +- backend/tests/crawler_jobs.rs | 290 ++++++++++++++++-- frontend/package-lock.json | 4 +- frontend/package.json | 2 +- 10 files changed, 419 insertions(+), 82 deletions(-) create mode 100644 backend/migrations/0032_crawler_jobs_lease_generation.sql diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 8835507..e455b7b 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1517,7 +1517,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.87.25" +version = "0.87.26" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index e0b2fe5..61d4af0 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.87.25" +version = "0.87.26" edition = "2021" default-run = "mangalord" diff --git a/backend/migrations/0032_crawler_jobs_lease_generation.sql b/backend/migrations/0032_crawler_jobs_lease_generation.sql new file mode 100644 index 0000000..afaa627 --- /dev/null +++ b/backend/migrations/0032_crawler_jobs_lease_generation.sql @@ -0,0 +1,23 @@ +-- Per-lease generation token to close the lease-identity race that +-- 0.87.20 only partially addressed. +-- +-- Before this, `ack_done` / `ack_failed` / `renew` matched on +-- `(id, state='running')`. A worker whose lease was released mid- +-- flight (force-analyze, reclaim) could still ack-done the row once a +-- successor leased it back to `state='running'` — clobbering the +-- successor's lease with the original's stale result. The bug class is +-- ack-from-a-dead-lease. +-- +-- Adding `lease_generation` makes lease identity = `(id, generation)`: +-- * `lease`/`lease_kinds` bump generation by 1 at lease time +-- * `release` (force-analyze, graceful shutdown) bumps generation too +-- * `reclaim_orphaned` bumps generation for every reclaimed row +-- * `ack_done` / `ack_failed` / `renew` match on `(id, generation, +-- state='running')`, so a stale ack from a prior generation finds +-- no row and falls back to the existing warn-and-skip branch. +-- +-- BIGINT so the counter can't realistically overflow even for a row +-- that's been crash-leased thousands of times. + +ALTER TABLE crawler_jobs + ADD COLUMN lease_generation BIGINT NOT NULL DEFAULT 0; diff --git a/backend/src/analysis/daemon.rs b/backend/src/analysis/daemon.rs index 9a5a2af..b4dc0f6 100644 --- a/backend/src/analysis/daemon.rs +++ b/backend/src/analysis/daemon.rs @@ -200,7 +200,7 @@ impl WorkerContext { // Shouldn't happen — we lease only analyze_page — but ack done // so a misrouted job doesn't loop forever. tracing::warn!(worker = self.id, "analysis worker: non-analyze payload leased"); - let _ = jobs::ack_done(&self.pool, lease.id).await; + let _ = jobs::ack_done(&self.pool, lease.id, lease.lease_generation).await; return; }; @@ -210,7 +210,7 @@ impl WorkerContext { if !force { if let Ok(Some(row)) = repo::page_analysis::load(&self.pool, page_id).await { if row.status == crate::domain::page_analysis::AnalysisStatus::Done { - let _ = jobs::ack_done(&self.pool, lease.id).await; + let _ = jobs::ack_done(&self.pool, lease.id, lease.lease_generation).await; return; } } @@ -241,10 +241,11 @@ impl WorkerContext { let heartbeat = { let hb_pool = self.pool.clone(); let hb_id = lease.id; + let hb_gen = lease.lease_generation; tokio::spawn(async move { loop { tokio::time::sleep(LEASE_HEARTBEAT).await; - match jobs::renew(&hb_pool, hb_id, LEASE_DURATION).await { + match jobs::renew(&hb_pool, hb_id, hb_gen, LEASE_DURATION).await { Ok(true) => {} Ok(false) => break, Err(e) => tracing::warn!(lease_id = %hb_id, ?e, "heartbeat renew failed"), @@ -274,7 +275,7 @@ impl WorkerContext { lease_id = %lease.id, "analysis worker: cancelled mid-dispatch — releasing lease" ); - let _ = jobs::release(&self.pool, lease.id).await; + let _ = jobs::release(&self.pool, lease.id, lease.lease_generation).await; // Don't publish Failed (the page didn't actually fail) and // don't write a duration row — neither outcome is true. DO // publish Cancelled so the dashboard's "now analyzing" banner @@ -342,7 +343,7 @@ impl WorkerContext { match result { Ok(()) => { - let _ = jobs::ack_done(&self.pool, lease.id).await; + let _ = jobs::ack_done(&self.pool, lease.id, lease.lease_generation).await; } Err(msg) => { tracing::warn!( @@ -357,6 +358,7 @@ impl WorkerContext { &msg, lease.attempts, lease.max_attempts, + lease.lease_generation, ) .await; // Terminal failure (retries exhausted) → record a `failed` diff --git a/backend/src/crawler/daemon.rs b/backend/src/crawler/daemon.rs index f756cee..9132583 100644 --- a/backend/src/crawler/daemon.rs +++ b/backend/src/crawler/daemon.rs @@ -404,7 +404,7 @@ impl WorkerContext { .ok() .flatten(); if matches!(page_count, Some(n) if n > 0) { - let _ = jobs::ack_done(&self.pool, lease.id).await; + let _ = jobs::ack_done(&self.pool, lease.id, lease.lease_generation).await; return; } } @@ -416,10 +416,11 @@ impl WorkerContext { let heartbeat = { let hb_pool = self.pool.clone(); let hb_id = lease.id; + let hb_gen = lease.lease_generation; tokio::spawn(async move { loop { tokio::time::sleep(LEASE_HEARTBEAT).await; - match jobs::renew(&hb_pool, hb_id, LEASE_DURATION).await { + match jobs::renew(&hb_pool, hb_id, hb_gen, LEASE_DURATION).await { Ok(true) => {} Ok(false) => break, Err(e) => { @@ -453,7 +454,7 @@ impl WorkerContext { // branch here, so a mid-dispatch SIGTERM left the row `running` // until lease expiry and cost one attempt. heartbeat.abort(); - let _ = jobs::release(&self.pool, lease.id).await; + let _ = jobs::release(&self.pool, lease.id, lease.lease_generation).await; tracing::info!( worker = self.id, lease_id = %lease.id, @@ -480,6 +481,7 @@ impl WorkerContext { "dispatch timed out", lease.attempts, lease.max_attempts, + lease.lease_generation, ) .await; return; @@ -487,7 +489,7 @@ impl WorkerContext { }; match outcome { Ok(Ok(SyncOutcome::Fetched { .. } | SyncOutcome::Skipped)) => { - let _ = jobs::ack_done(&self.pool, lease.id).await; + let _ = jobs::ack_done(&self.pool, lease.id, lease.lease_generation).await; } Ok(Ok(SyncOutcome::SessionExpired)) => { tracing::error!( @@ -498,7 +500,7 @@ impl WorkerContext { self.session_expired.store(true, Ordering::Release); // Push the session-expired flip to live status subscribers. self.status.poke(); - let _ = jobs::release(&self.pool, lease.id).await; + let _ = jobs::release(&self.pool, lease.id, lease.lease_generation).await; } Ok(Err(e)) => { tracing::warn!( @@ -513,6 +515,7 @@ impl WorkerContext { &format!("{e:#}"), lease.attempts, lease.max_attempts, + lease.lease_generation, ) .await; } @@ -528,6 +531,7 @@ impl WorkerContext { "worker panicked", lease.attempts, lease.max_attempts, + lease.lease_generation, ) .await; } diff --git a/backend/src/crawler/jobs.rs b/backend/src/crawler/jobs.rs index eb0bbc0..87c5e09 100644 --- a/backend/src/crawler/jobs.rs +++ b/backend/src/crawler/jobs.rs @@ -89,6 +89,14 @@ pub struct Lease { pub payload: JobPayload, pub attempts: i32, pub max_attempts: i32, + /// Strictly-increasing token bumped by every `lease`, `release` / + /// `release_unowned`, and `reclaim_orphaned`. `ack_done` / + /// `ack_failed` / `renew` match on `(id, lease_generation)` so a + /// late ack from a dead-but-still-dispatching lease cannot clobber + /// a successor that has re-leased the same row. See migration + /// 0032 for the column and the `ack_done_from_dead_lease_*` tests + /// in `crawler_jobs.rs` for the race the token closes. + pub lease_generation: i64, } /// Deterministic exponential backoff base for `ack_failed` retries. @@ -159,7 +167,7 @@ pub async fn lease( lease_duration: Duration, ) -> sqlx::Result> { let lease_ms: i64 = lease_duration.as_millis().min(i64::MAX as u128) as i64; - let rows: Vec<(Uuid, serde_json::Value, i32, i32)> = sqlx::query_as( + let rows: Vec<(Uuid, serde_json::Value, i32, i32, i64)> = sqlx::query_as( r#" WITH leased AS ( SELECT id FROM crawler_jobs @@ -173,11 +181,12 @@ pub async fn lease( UPDATE crawler_jobs j SET state = 'running', attempts = j.attempts + 1, + lease_generation = j.lease_generation + 1, leased_until = now() + ($3::bigint || ' milliseconds')::interval, updated_at = now() FROM leased l WHERE j.id = l.id - RETURNING j.id, j.payload, j.attempts, j.max_attempts + RETURNING j.id, j.payload, j.attempts, j.max_attempts, j.lease_generation "#, ) .bind(kind_filter) @@ -201,7 +210,7 @@ pub async fn lease_kinds( ) -> sqlx::Result> { let lease_ms: i64 = lease_duration.as_millis().min(i64::MAX as u128) as i64; let kinds_vec: Vec = kinds.iter().map(|s| s.to_string()).collect(); - let rows: Vec<(Uuid, serde_json::Value, i32, i32)> = sqlx::query_as( + let rows: Vec<(Uuid, serde_json::Value, i32, i32, i64)> = sqlx::query_as( r#" WITH leased AS ( SELECT id FROM crawler_jobs @@ -215,11 +224,12 @@ pub async fn lease_kinds( UPDATE crawler_jobs j SET state = 'running', attempts = j.attempts + 1, + lease_generation = j.lease_generation + 1, leased_until = now() + ($3::bigint || ' milliseconds')::interval, updated_at = now() FROM leased l WHERE j.id = l.id - RETURNING j.id, j.payload, j.attempts, j.max_attempts + RETURNING j.id, j.payload, j.attempts, j.max_attempts, j.lease_generation "#, ) .bind(&kinds_vec) @@ -230,9 +240,11 @@ pub async fn lease_kinds( decode_leases(rows) } -fn decode_leases(rows: Vec<(Uuid, serde_json::Value, i32, i32)>) -> sqlx::Result> { +fn decode_leases( + rows: Vec<(Uuid, serde_json::Value, i32, i32, i64)>, +) -> sqlx::Result> { let mut leases = Vec::with_capacity(rows.len()); - for (id, payload_json, attempts, max_attempts) in rows { + for (id, payload_json, attempts, max_attempts, lease_generation) in rows { let payload: JobPayload = serde_json::from_value(payload_json).map_err(|e| { sqlx::Error::Decode(format!("invalid JobPayload JSON for job {id}: {e}").into()) })?; @@ -241,6 +253,7 @@ fn decode_leases(rows: Vec<(Uuid, serde_json::Value, i32, i32)>) -> sqlx::Result payload, attempts, max_attempts, + lease_generation, }); } Ok(leases) @@ -259,42 +272,52 @@ fn decode_leases(rows: Vec<(Uuid, serde_json::Value, i32, i32)>) -> sqlx::Result pub async fn renew( pool: &PgPool, lease_id: Uuid, + lease_generation: i64, lease_duration: Duration, ) -> sqlx::Result { let lease_ms: i64 = lease_duration.as_millis().min(i64::MAX as u128) as i64; let res = sqlx::query( "UPDATE crawler_jobs \ - SET leased_until = now() + ($2::bigint || ' milliseconds')::interval, \ + SET leased_until = now() + ($3::bigint || ' milliseconds')::interval, \ updated_at = now() \ - WHERE id = $1 AND state = 'running'", + WHERE id = $1 AND lease_generation = $2 AND state = 'running'", ) .bind(lease_id) + .bind(lease_generation) .bind(lease_ms) .execute(pool) .await?; Ok(res.rows_affected() > 0) } -/// Mark a leased job as successfully completed. The `state = 'running'` -/// predicate guards against a late ack from a worker whose lease expired -/// and was already re-leased by another worker: without it, the late ack -/// would clobber the new lease's `state` and `leased_until`. `rows_affected -/// == 0` means we lost the lease — surfaced as a warn rather than an -/// error because the new lease holder is doing real work; the late ack -/// just has to step aside. -pub async fn ack_done(pool: &PgPool, lease_id: Uuid) -> sqlx::Result<()> { +/// Mark a leased job as successfully completed. The +/// `(lease_generation, state = 'running')` predicate guards against a +/// late ack from a *specific* dead lease — the case where the worker's +/// lease was released or expired and a successor re-leased the same id. +/// Before migration 0032 the guard was state-only, so a late ack from +/// the original could still match (state was running again, owned by the +/// successor) and clobber. Scoping to the original's `lease_generation` +/// makes the match fail cleanly. `rows_affected == 0` is surfaced as a +/// warn — the successor is doing real work; the late ack steps aside. +pub async fn ack_done( + pool: &PgPool, + lease_id: Uuid, + lease_generation: i64, +) -> sqlx::Result<()> { let res = sqlx::query( "UPDATE crawler_jobs \ SET state = 'done', leased_until = NULL, updated_at = now() \ - WHERE id = $1 AND state = 'running'", + WHERE id = $1 AND lease_generation = $2 AND state = 'running'", ) .bind(lease_id) + .bind(lease_generation) .execute(pool) .await?; if res.rows_affected() == 0 { tracing::warn!( %lease_id, - "ack_done: lease no longer running — likely re-leased by another worker; skipping update" + lease_generation, + "ack_done: lease no longer current — generation mismatch (force-analyzed, reclaimed, or re-leased); skipping update" ); } Ok(()) @@ -311,15 +334,17 @@ pub async fn ack_failed( error: &str, attempts: i32, max_attempts: i32, + lease_generation: i64, ) -> sqlx::Result<()> { let res = if attempts >= max_attempts { sqlx::query( "UPDATE crawler_jobs \ SET state = 'dead', last_error = $2, leased_until = NULL, updated_at = now() \ - WHERE id = $1 AND state = 'running'", + WHERE id = $1 AND lease_generation = $3 AND state = 'running'", ) .bind(lease_id) .bind(error) + .bind(lease_generation) .execute(pool) .await? } else { @@ -329,34 +354,81 @@ pub async fn ack_failed( SET state = 'pending', last_error = $2, leased_until = NULL, \ scheduled_at = now() + ($3::bigint || ' milliseconds')::interval, \ updated_at = now() \ - WHERE id = $1 AND state = 'running'", + WHERE id = $1 AND lease_generation = $4 AND state = 'running'", ) .bind(lease_id) .bind(error) .bind(backoff_ms) + .bind(lease_generation) .execute(pool) .await? }; if res.rows_affected() == 0 { tracing::warn!( %lease_id, - "ack_failed: lease no longer running — likely re-leased by another worker; skipping update" + lease_generation, + "ack_failed: lease no longer current — generation mismatch (force-analyzed, reclaimed, or re-leased); skipping update" ); } Ok(()) } /// Return a leased job to `pending` without burning a retry attempt. -/// Used on graceful shutdown and on session-expired aborts where the -/// failure isn't the job's fault. See [`ack_done`] for the -/// `state = 'running'` guard rationale — important here because -/// `attempts - 1` would otherwise spuriously decrement the new lease's -/// attempt count. -pub async fn release(pool: &PgPool, lease_id: Uuid) -> sqlx::Result<()> { +/// Used by workers on graceful shutdown and on session-expired aborts +/// where the failure isn't the job's fault. The +/// `(lease_generation, state = 'running')` guard scopes the release to +/// the *specific* lease the caller was holding — otherwise a late +/// release could clobber a successor's lease (drop the row back to +/// pending mid-dispatch). Also bumps `lease_generation` so any +/// subsequent ack from this same lease finds nothing. +/// +/// For force-analyze and other external paths that release a +/// running lease without holding a [`Lease`] struct, use +/// [`release_unowned`] instead. +pub async fn release( + pool: &PgPool, + lease_id: Uuid, + lease_generation: i64, +) -> sqlx::Result<()> { let res = sqlx::query( "UPDATE crawler_jobs \ SET state = 'pending', leased_until = NULL, \ - attempts = GREATEST(0, attempts - 1), updated_at = now() \ + attempts = GREATEST(0, attempts - 1), \ + lease_generation = lease_generation + 1, \ + updated_at = now() \ + WHERE id = $1 AND lease_generation = $2 AND state = 'running'", + ) + .bind(lease_id) + .bind(lease_generation) + .execute(pool) + .await?; + if res.rows_affected() == 0 { + tracing::warn!( + %lease_id, + lease_generation, + "release: lease no longer current — generation mismatch (force-analyzed, reclaimed, or re-leased); skipping update" + ); + } + Ok(()) +} + +/// Release a `running` lease without owning it: caller has only a +/// job id, not a [`Lease`] struct. Used by force-analyze (admin +/// click that drops an in-flight lease so a refreshed payload is +/// picked up) and ops tools. +/// +/// Unconditionally matches the row by id and bumps +/// `lease_generation` so the dropped lease's pending ack from +/// the original worker becomes a no-op. Returns the attempt to +/// `pending` and refunds the retry attempt (the operator click +/// isn't a job-level failure). +pub async fn release_unowned(pool: &PgPool, lease_id: Uuid) -> sqlx::Result<()> { + let res = sqlx::query( + "UPDATE crawler_jobs \ + SET state = 'pending', leased_until = NULL, \ + attempts = GREATEST(0, attempts - 1), \ + lease_generation = lease_generation + 1, \ + updated_at = now() \ WHERE id = $1 AND state = 'running'", ) .bind(lease_id) @@ -365,7 +437,7 @@ pub async fn release(pool: &PgPool, lease_id: Uuid) -> sqlx::Result<()> { if res.rows_affected() == 0 { tracing::warn!( %lease_id, - "release: lease no longer running — likely re-leased by another worker; skipping update" + "release_unowned: row not running — likely already completed or never leased; skipping" ); } Ok(()) @@ -397,7 +469,9 @@ pub async fn reclaim_orphaned(pool: &PgPool) -> sqlx::Result { let result = sqlx::query( "UPDATE crawler_jobs \ SET state = 'pending', leased_until = NULL, \ - attempts = GREATEST(0, attempts - 1), updated_at = now() \ + attempts = GREATEST(0, attempts - 1), \ + lease_generation = lease_generation + 1, \ + updated_at = now() \ WHERE state = 'running' AND leased_until < now()", ) .execute(pool) diff --git a/backend/src/repo/page_analysis.rs b/backend/src/repo/page_analysis.rs index 1fab617..58edf53 100644 --- a/backend/src/repo/page_analysis.rs +++ b/backend/src/repo/page_analysis.rs @@ -128,16 +128,18 @@ pub async fn enqueue_for_page( // Even though we've flipped the row's payload to // `force=true`, the worker's skip-if-done check uses its // own pre-upgrade value and will ack done without - // re-analyzing. Release the lease so the row returns to - // `pending` with attempts refunded; the worker's later - // `ack_done` becomes a no-op (state guard `WHERE state = - // 'running'`) and a fresh lease picks the row up with the - // updated payload. May produce one wasted dispatch on the - // races where the worker had already started its vision - // call — strictly better than silently losing admin intent. + // re-analyzing. `release_unowned` returns the row to + // `pending` AND bumps `lease_generation` so the original + // worker's later `ack_done(id, old_generation)` finds no + // matching row and is a no-op — the successor's lease (a + // fresh generation) is preserved. The state-only guard + // that previously protected this (and was the 0.87.20 + // narrowing) wasn't enough on its own: a successor that + // re-leased the id would re-enter `state='running'`, and + // the original's ack would clobber it. for (id, state) in &upgraded { if state == "running" { - let _ = jobs::release(pool, *id).await; + let _ = jobs::release_unowned(pool, *id).await; } } Ok(EnqueueForPageOutcome::UpgradedToForce) diff --git a/backend/tests/crawler_jobs.rs b/backend/tests/crawler_jobs.rs index ecc246c..8d67828 100644 --- a/backend/tests/crawler_jobs.rs +++ b/backend/tests/crawler_jobs.rs @@ -209,7 +209,7 @@ async fn renew_extends_leased_until_while_running(pool: PgPool) { .await .unwrap(); - let still_owned = jobs::renew(&pool, id, Duration::from_secs(120)) + 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"); @@ -240,9 +240,11 @@ async fn renew_is_noop_once_job_no_longer_running(pool: PgPool) { .await .unwrap(); // Job completes — heartbeat should now see it's no longer ours. - jobs::ack_done(&pool, leases[0].id).await.unwrap(); + jobs::ack_done(&pool, leases[0].id, leases[0].lease_generation) + .await + .unwrap(); - let still_owned = jobs::renew(&pool, id, Duration::from_secs(120)) + 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"); @@ -431,7 +433,8 @@ async fn reclaim_orphaned_resets_only_expired_running_jobs(pool: PgPool) { .await .unwrap(); // Leave `pending` pending by returning it (release refunds its attempt). - jobs::release(&pool, pending).await.unwrap(); + // 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 @@ -482,7 +485,9 @@ async fn ack_done_transitions_state_and_clears_lease(pool: PgPool) { let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60)) .await .unwrap(); - jobs::ack_done(&pool, leases[0].id).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> = @@ -507,9 +512,16 @@ async fn ack_failed_under_max_returns_to_pending_with_future_schedule(pool: PgPo .await .unwrap(); let lease = &leases[0]; - jobs::ack_failed(&pool, lease.id, "boom", lease.attempts, lease.max_attempts) - .await - .unwrap(); + 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"); @@ -537,9 +549,16 @@ async fn ack_failed_at_max_marks_dead(pool: PgPool) { .await .unwrap(); let lease = &leases[0]; - jobs::ack_failed(&pool, lease.id, "final boom", lease.max_attempts, lease.max_attempts) - .await - .unwrap(); + 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 = @@ -564,7 +583,7 @@ async fn ack_done_no_ops_when_lease_was_stolen(pool: PgPool) { _ => unreachable!(), }; // Worker A grabs the lease, but its lease expires immediately. - let _a_leases = jobs::lease(&pool, None, 1, Duration::from_secs(60)) + 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") @@ -578,19 +597,26 @@ async fn ack_done_no_ops_when_lease_was_stolen(pool: PgPool) { .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 — guarded by `state = 'running'` + lease_id - // but in the simplest implementation the guard is state-only. Either - // way, the job stays 'running' with worker B's progress intact. - jobs::ack_done(&pool, id).await.unwrap(); - // Worker B is still working; until B acks, the job remains 'running' - // with its leased_until in the future and attempts == 2. - // (We can't make ack_done's lease_id distinguish A from B today — - // both share the same `id` — so the strongest current guarantee is - // that a late ack_done doesn't fire when state is already 'done', - // exercised below.) - // Finalize: worker B acks done. - jobs::ack_done(&pool, b_leases[0].id).await.unwrap(); + // 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); } @@ -611,11 +637,13 @@ async fn ack_failed_no_ops_when_state_is_not_running(pool: PgPool) { let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60)) .await .unwrap(); - jobs::ack_done(&pool, leases[0].id).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) + jobs::ack_failed(&pool, leases[0].id, "late", 1, 5, leases[0].lease_generation) .await .unwrap(); assert_eq!( @@ -640,11 +668,15 @@ async fn release_no_ops_when_state_is_not_running(pool: PgPool) { let leases = jobs::lease(&pool, None, 1, Duration::from_secs(60)) .await .unwrap(); - jobs::ack_done(&pool, leases[0].id).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).await.unwrap(); + jobs::release(&pool, leases[0].id, leases[0].lease_generation) + .await + .unwrap(); assert_eq!( job_state(&pool, id).await, "done", @@ -670,7 +702,9 @@ async fn release_returns_to_pending_and_undoes_attempt_increment(pool: PgPool) { .await .unwrap(); assert_eq!(leases[0].attempts, 1); - jobs::release(&pool, leases[0].id).await.unwrap(); + 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); @@ -974,3 +1008,201 @@ async fn migration_0031_prededup_demotes_duplicate_analyze_page_rows(pool: PgPoo .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}" + ); +} diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 044e890..ea1971c 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -1,12 +1,12 @@ { "name": "mangalord-frontend", - "version": "0.23.0", + "version": "0.87.26", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "mangalord-frontend", - "version": "0.23.0", + "version": "0.87.26", "devDependencies": { "@lucide/svelte": "^1.16.0", "@playwright/test": "^1.48.0", diff --git a/frontend/package.json b/frontend/package.json index 708866f..16bf29e 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.87.25", + "version": "0.87.26", "private": true, "type": "module", "scripts": {