From e3e86843d6bf406f4eb53350dc2e80601967a7b4 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Tue, 23 Jun 2026 20:42:11 +0200 Subject: [PATCH] fix(force-analyze): release running lease so worker drops stale-payload local (0.87.20) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- backend/Cargo.lock | 2 +- backend/Cargo.toml | 2 +- backend/src/repo/page_analysis.rs | 52 ++++++++++++----- backend/tests/api_admin_analysis.rs | 89 +++++++++++++++++++++++++++++ frontend/package.json | 2 +- 5 files changed, 131 insertions(+), 16 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 6bf8ee8..5de3fae 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1517,7 +1517,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.87.19" +version = "0.87.20" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 06ee234..145f90d 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.87.19" +version = "0.87.20" edition = "2021" default-run = "mangalord" diff --git a/backend/src/repo/page_analysis.rs b/backend/src/repo/page_analysis.rs index 43975af..1fab617 100644 --- a/backend/src/repo/page_analysis.rs +++ b/backend/src/repo/page_analysis.rs @@ -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 = 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) } } } diff --git a/backend/tests/api_admin_analysis.rs b/backend/tests/api_admin_analysis.rs index 85cd4ff..3b4ec78 100644 --- a/backend/tests/api_admin_analysis.rs +++ b/backend/tests/api_admin_analysis.rs @@ -464,6 +464,95 @@ async fn force_analyze_upgrades_pending_non_force_to_force(pool: PgPool) { ); } +/// The running-state branch of `force=true` collision. A worker has +/// already leased the row (state=running, payload.force=false in its +/// in-memory `lease.payload`). Admin force-clicks → my code must NOT +/// only flip the row's payload to force=true (the worker holds the +/// stale value) but also release the in-flight lease so the worker +/// drops the stale payload and a fresh lease re-leases with the +/// updated value. +/// +/// Mutation-confirmed: removing the `jobs::release` loop in +/// `enqueue_for_page` (leaving only the UPDATE) makes this test fail +/// — the row's state stays `running` until the lease expires. +#[sqlx::test(migrations = "./migrations")] +async fn force_analyze_releases_running_lease_so_worker_re_picks_force(pool: PgPool) { + let h = common::harness_with_analysis(pool.clone()); + let cookie = seed_admin(&pool, &h.app).await; + let ids = seed_pages(&pool, 1).await; + let page_id = ids[0]; + + // 1) Pre-existing non-force pending row. + mangalord::repo::page_analysis::enqueue_for_page(&pool, page_id, false) + .await + .unwrap(); + + // 2) Simulate a worker leasing the job — state moves to `running` + // with leased_until in the future and attempts=1. The worker + // process (in real life) destructures its in-memory `lease.payload` + // here, capturing `force=false` as a Rust local. + let leases = mangalord::crawler::jobs::lease( + &pool, + Some(mangalord::crawler::jobs::KIND_ANALYZE_PAGE), + 1, + std::time::Duration::from_secs(300), + ) + .await + .unwrap(); + assert_eq!(leases.len(), 1, "test setup: should have leased the seeded job"); + let leased_id = leases[0].id; + // Sanity: row is now running with force=false in payload. + let (state, force): (String, bool) = sqlx::query_as( + "SELECT state, (payload->>'force')::boolean \ + FROM crawler_jobs WHERE id = $1", + ) + .bind(leased_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(state, "running"); + assert!(!force); + + // 3) Admin force-clicks. Without 0.87.20's release loop, the row's + // payload flips to force=true but state stays `running`; the + // worker (holding its stale `force=false` local) acks done and + // the page is never re-analyzed. With the release loop, the row + // goes back to `pending`. + let resp = h + .app + .clone() + .oneshot(common::post_json_with_cookie( + &format!("/api/v1/admin/pages/{page_id}/analyze"), + json!({}), + &cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + // 4) Row is back to pending, with the payload's force flag flipped + // to true. A subsequent worker lease would now see `force=true` + // and re-analyze. attempts is decremented by `jobs::release` so + // the page doesn't burn its retry budget on the admin click. + let (state, force, attempts): (String, bool, i32) = sqlx::query_as( + "SELECT state, (payload->>'force')::boolean, attempts \ + FROM crawler_jobs WHERE id = $1", + ) + .bind(leased_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + state, "pending", + "running lease must be released so the worker drops its stale-payload local" + ); + assert!(force, "payload's force flag must be flipped to true"); + assert_eq!( + attempts, 0, + "release refunds the attempt — the admin click doesn't burn the retry budget" + ); +} + #[sqlx::test(migrations = "./migrations")] async fn force_analyze_unknown_page_is_404(pool: PgPool) { let h = common::harness_with_analysis(pool.clone()); diff --git a/frontend/package.json b/frontend/package.json index d449c40..3867eb1 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.87.19", + "version": "0.87.20", "private": true, "type": "module", "scripts": {