fabi/Mangalord
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(jobs): per-lease generation token closes the ack-from-dead-lease race (0.87.26)

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-24 20:05:55 +02:00
parent ae708a72d0
commit 93b7e451bf
10 changed files with 419 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
backend/src/analysis/daemon.rs
View File

@@ -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`

14
backend/src/crawler/daemon.rs
View File

@@ -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;
}

134
backend/src/crawler/jobs.rs
View File

@@ -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<Vec<Lease>> {
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<Vec<Lease>> {
let lease_ms: i64 = lease_duration.as_millis().min(i64::MAX as u128) as i64;
let kinds_vec: Vec<String> = 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<Vec<Lease>> {
fn decode_leases(
rows: Vec<(Uuid, serde_json::Value, i32, i32, i64)>,
) -> sqlx::Result<Vec<Lease>> {
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<bool> {
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<u64> {
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)

18
backend/src/repo/page_analysis.rs
View File

@@ -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)