fix(force-analyze): release running lease so worker drops stale-payload local (0.87.20)
0.87.11 followup. The UPDATE matched pending+running but the worker holds the pre-update `force=false` in an in-memory local — so on the running branch the row flipped but the worker's skip-if-done still acked done without re-analyzing. Now `jobs::release` any running row the UPDATE matched, so the worker's ack_done no-ops (state guard) and a fresh lease picks up the updated payload. New test mints the running-state shape, asserts state=pending + force=true + attempts=0 post-click. Mutation-confirmed. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -56,8 +56,11 @@ pub enum EnqueueForPageOutcome {
|
||||
/// requests.
|
||||
UpgradedToForce,
|
||||
/// A matching `(pending|running)` job already encodes the request;
|
||||
/// nothing changed. For force requests this means the existing row
|
||||
/// is already `force=true`.
|
||||
/// nothing changed. For force requests this is the retry-race tail:
|
||||
/// the INSERT was Skipped, the UPDATE matched zero rows (the sibling
|
||||
/// drained between the two), and the second INSERT was Skipped too
|
||||
/// (a fresh sibling landed). The next ack of either job will leave
|
||||
/// the page in the admin's intended state without further action.
|
||||
AlreadyEnqueued,
|
||||
}
|
||||
|
||||
@@ -89,8 +92,10 @@ pub async fn enqueue_for_page(
|
||||
// partial index predicate (`pending|running` + `analyze_page`
|
||||
// + this page_id) and adds `force=false` so we never overwrite
|
||||
// a row that's already what the admin wants. RETURNING tells
|
||||
// us whether the upgrade actually moved anything.
|
||||
let upgraded: Option<Uuid> = sqlx::query_scalar(
|
||||
// us whether the upgrade actually moved anything AND its
|
||||
// pre-update state so the running-state race fix below can
|
||||
// release the in-flight lease.
|
||||
let upgraded: Vec<(Uuid, String)> = sqlx::query_as(
|
||||
"UPDATE crawler_jobs \
|
||||
SET payload = jsonb_set(payload, '{force}', 'true'::jsonb), \
|
||||
updated_at = now() \
|
||||
@@ -98,23 +103,44 @@ pub async fn enqueue_for_page(
|
||||
AND payload->>'kind' = 'analyze_page' \
|
||||
AND payload->>'page_id' = $1 \
|
||||
AND (payload->>'force')::boolean = false \
|
||||
RETURNING id",
|
||||
RETURNING id, state",
|
||||
)
|
||||
.bind(page_id.to_string())
|
||||
.fetch_optional(pool)
|
||||
.fetch_all(pool)
|
||||
.await?;
|
||||
Ok(match upgraded {
|
||||
Some(_) => EnqueueForPageOutcome::UpgradedToForce,
|
||||
if upgraded.is_empty() {
|
||||
// Race: between the skipped INSERT and the UPDATE the
|
||||
// sibling row could have been completed (job state moves
|
||||
// out of pending/running). At that point a re-INSERT
|
||||
// would succeed. Retry once to close the race without
|
||||
// unbounded loops.
|
||||
None => match jobs::enqueue(pool, &JobPayload::AnalyzePage { page_id, force }).await? {
|
||||
EnqueueResult::Inserted(_) => EnqueueForPageOutcome::Inserted,
|
||||
EnqueueResult::Skipped => EnqueueForPageOutcome::AlreadyEnqueued,
|
||||
},
|
||||
})
|
||||
return Ok(
|
||||
match jobs::enqueue(pool, &JobPayload::AnalyzePage { page_id, force }).await? {
|
||||
EnqueueResult::Inserted(_) => EnqueueForPageOutcome::Inserted,
|
||||
EnqueueResult::Skipped => EnqueueForPageOutcome::AlreadyEnqueued,
|
||||
},
|
||||
);
|
||||
}
|
||||
// Race fix for the running-state branch (0.87.20 followup):
|
||||
// a worker that has already leased the row holds the
|
||||
// pre-upgrade `force=false` in its in-memory `lease.payload`
|
||||
// (see `analysis::daemon::WorkerContext::process_lease`).
|
||||
// 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.
|
||||
for (id, state) in &upgraded {
|
||||
if state == "running" {
|
||||
let _ = jobs::release(pool, *id).await;
|
||||
}
|
||||
}
|
||||
Ok(EnqueueForPageOutcome::UpgradedToForce)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user