diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 8fe3158..c7f1a48 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1517,7 +1517,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.87.10" +version = "0.87.11" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 7552fdd..1b41aae 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.87.10" +version = "0.87.11" edition = "2021" default-run = "mangalord" diff --git a/backend/migrations/0031_crawler_jobs_analyze_page_dedup.sql b/backend/migrations/0031_crawler_jobs_analyze_page_dedup.sql index 0023338..9e9cc1f 100644 --- a/backend/migrations/0031_crawler_jobs_analyze_page_dedup.sql +++ b/backend/migrations/0031_crawler_jobs_analyze_page_dedup.sql @@ -11,6 +11,39 @@ -- and rely on `ON CONFLICT DO NOTHING` instead. At most one (pending|running) -- job per page_id can exist; the slot frees the moment the job transitions -- to done/failed/dead, so a force re-analyze still re-enqueues cleanly. + +-- **Pre-dedup step.** Production runs of the prior buggy producer can +-- have left duplicate rows tied on (kind='analyze_page', payload->>'page_id', +-- state IN ('pending','running')). Adding the UNIQUE INDEX onto a dirty +-- table would fail to create and crash `sqlx::migrate!` at boot. +-- +-- Demote all-but-the-lowest-id duplicates to `dead` first. Why dead and +-- not done: the dropped sibling never actually ran, and `dead` is the +-- terminal state operators inspect via the dead-jobs requeue endpoint — +-- so a curator can recover any genuinely-needed page. The lowest-id row +-- is the one a worker most likely already leased / has heartbeat'd, so +-- keeping it minimises lease-state thrash. +UPDATE crawler_jobs + SET state = 'dead', + last_error = COALESCE(last_error, '') || + CASE WHEN last_error IS NULL OR last_error = '' THEN '' + ELSE ' | ' END || + 'pre-0031 duplicate analyze_page; superseded by earlier sibling', + updated_at = now() + WHERE id IN ( + SELECT id FROM ( + SELECT id, + row_number() OVER ( + PARTITION BY payload->>'page_id' + ORDER BY id + ) AS rn + FROM crawler_jobs + WHERE payload->>'kind' = 'analyze_page' + AND state IN ('pending', 'running') + ) ranked + WHERE rn > 1 + ); + CREATE UNIQUE INDEX crawler_jobs_analyze_page_dedup_idx ON crawler_jobs ((payload->>'page_id')) WHERE state IN ('pending', 'running') diff --git a/backend/src/api/admin/analysis.rs b/backend/src/api/admin/analysis.rs index 96d5ff5..32d8d99 100644 --- a/backend/src/api/admin/analysis.rs +++ b/backend/src/api/admin/analysis.rs @@ -415,15 +415,27 @@ async fn analyze_page( return Err(AppError::NotFound); } - repo::page_analysis::enqueue_for_page(&state.db, page_id, true).await?; + // Surface the actual outcome so the admin (and audit log) sees + // whether the click actually moved anything. Before 0.87.11 the + // partial unique index could silently swallow a force request + // when a `force=false` job was already pending — the worker would + // then pick up the non-force row, hit skip-if-done, ack done, and + // the admin saw "queued for re-analysis" with no re-analysis. + let outcome = + repo::page_analysis::enqueue_for_page(&state.db, page_id, true).await?; + let outcome_label = match outcome { + repo::page_analysis::EnqueueForPageOutcome::Inserted => "inserted", + repo::page_analysis::EnqueueForPageOutcome::UpgradedToForce => "upgraded_pending_to_force", + repo::page_analysis::EnqueueForPageOutcome::AlreadyEnqueued => "already_force_enqueued", + }; repo::admin_audit::insert( &state.db, admin.0.id, "analysis_force_page", "page", Some(page_id), - json!({ "force": true }), + json!({ "force": true, "outcome": outcome_label }), ) .await?; diff --git a/backend/src/repo/bookmark.rs b/backend/src/repo/bookmark.rs index 7e242fc..21dbb0b 100644 --- a/backend/src/repo/bookmark.rs +++ b/backend/src/repo/bookmark.rs @@ -63,11 +63,13 @@ pub async fn list_for_user( INNER JOIN mangas m ON m.id = b.manga_id LEFT JOIN chapters c ON c.id = b.chapter_id WHERE b.user_id = $1 - -- `manga_id` tiebreaker: the (user_id, manga_id) PK guarantees no - -- two bookmarks share both, so this is a deterministic ordering. - -- Without it, multiple bookmarks added in the same `created_at` - -- tick (timestamptz resolution: 1µs) can skip/repeat across pages. - ORDER BY b.created_at DESC, b.manga_id DESC + -- `id` tiebreaker for deterministic pagination. The bookmarks PK + -- is `id` (see migration 0001); the (user_id, manga_id) pair + -- alone is not unique — migration 0004 lets a user keep both a + -- chapter-level bookmark and a manga-level bookmark on the same + -- manga, so two rows can tie on (created_at, manga_id). A 0.87.6 + -- earlier revision mistakenly used `manga_id` here. + ORDER BY b.created_at DESC, b.id DESC LIMIT $2 OFFSET $3 "#, ) diff --git a/backend/src/repo/page_analysis.rs b/backend/src/repo/page_analysis.rs index 78077d4..43975af 100644 --- a/backend/src/repo/page_analysis.rs +++ b/backend/src/repo/page_analysis.rs @@ -44,12 +44,79 @@ pub struct PageSearchQuery { /// aborting the whole page's analysis on one bad model output. const MAX_TAG_CHARS: usize = 64; +/// Outcome of [`enqueue_for_page`] — surfaces whether the admin's intent +/// actually landed as a worker-visible change. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EnqueueForPageOutcome { + /// A fresh `analyze_page` job row was inserted. + Inserted, + /// A pending `force=false` row was upgraded in-place to `force=true`, + /// because the partial unique index `crawler_jobs_analyze_page_dedup_idx` + /// blocks a second pending insert per page. Returned only for force + /// 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`. + AlreadyEnqueued, +} + /// Enqueue an `analyze_page` job for `page_id`. `force` re-analyzes a page -/// that is already `done`. Enqueue is idempotent at the job level only in -/// that duplicate pending jobs are harmless — processing is idempotent. -pub async fn enqueue_for_page(pool: &PgPool, page_id: Uuid, force: bool) -> AppResult<()> { - jobs::enqueue(pool, &JobPayload::AnalyzePage { page_id, force }).await?; - Ok(()) +/// that is already `done`. +/// +/// **Force-vs-dedup correctness.** Migration 0031's partial unique index +/// is keyed on `page_id` (not on `(page_id, force)`), so a plain +/// `jobs::enqueue` of a `force=true` payload when a `force=false` job is +/// already pending silently goes to `Skipped` — the worker would then +/// pick up the non-force row, hit the skip-if-done net, and ack done +/// without re-analyzing. The admin's force request was lost. +/// +/// This function fixes that by: (1) trying the insert; (2) on Skipped +/// with `force=true`, running an UPDATE that flips the existing pending +/// row's `force` flag to `true`; (3) reporting which path ran so the +/// caller can audit accurately. +pub async fn enqueue_for_page( + pool: &PgPool, + page_id: Uuid, + force: bool, +) -> AppResult { + use crate::crawler::jobs::EnqueueResult; + match jobs::enqueue(pool, &JobPayload::AnalyzePage { page_id, force }).await? { + EnqueueResult::Inserted(_) => Ok(EnqueueForPageOutcome::Inserted), + EnqueueResult::Skipped if !force => Ok(EnqueueForPageOutcome::AlreadyEnqueued), + EnqueueResult::Skipped => { + // Force-specific upgrade. The WHERE clause matches the same + // 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( + "UPDATE crawler_jobs \ + SET payload = jsonb_set(payload, '{force}', 'true'::jsonb), \ + updated_at = now() \ + WHERE state IN ('pending', 'running') \ + AND payload->>'kind' = 'analyze_page' \ + AND payload->>'page_id' = $1 \ + AND (payload->>'force')::boolean = false \ + RETURNING id", + ) + .bind(page_id.to_string()) + .fetch_optional(pool) + .await?; + Ok(match upgraded { + Some(_) => EnqueueForPageOutcome::UpgradedToForce, + // 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, + }, + }) + } + } } /// How wide a bulk re-enqueue reaches. diff --git a/backend/tests/analysis_worker.rs b/backend/tests/analysis_worker.rs index 0d13c40..ec3d07a 100644 --- a/backend/tests/analysis_worker.rs +++ b/backend/tests/analysis_worker.rs @@ -200,6 +200,80 @@ async fn worker_marks_failed_row_on_terminal_failure(pool: PgPool) { assert!(row.error.is_some()); } +/// Non-terminal failure of a force-re-analyze must NOT overwrite the +/// prior `done` row's `duration_ms`. Before the 0.87.6 gate change, +/// `record_duration` ran unconditionally — a transient failure-retry +/// of a force-re-analyze would clobber the legitimately-measured +/// previous duration with the duration of an attempt that wrote +/// nothing. Test pins the gate. +#[sqlx::test(migrations = "./migrations")] +async fn worker_non_terminal_force_failure_does_not_overwrite_done_duration(pool: PgPool) { + let page_id = seed_page(&pool).await; + // 1) Pre-seed a done analysis row with a meaningful duration so we + // have something to be "overwritten". + page_analysis::persist_analysis( + &pool, + page_id, + &VisionAnalysis { + ocr_results: vec![], + tagging_results: vec![], + scene_description: String::new(), + safety_flag: SafetyFlag::default(), + }, + "m", + ) + .await + .unwrap(); + sqlx::query("UPDATE page_analysis SET duration_ms = $1 WHERE page_id = $2") + .bind(12345_i64) + .bind(page_id) + .execute(&pool) + .await + .unwrap(); + + // 2) Force re-analyze (force=true) — but with a failing dispatcher + // AND max_attempts=3 so the first failure is NON-terminal. + page_analysis::enqueue_for_page(&pool, page_id, true) + .await + .unwrap(); + sqlx::query("UPDATE crawler_jobs SET max_attempts = 3 WHERE payload->>'page_id' = $1") + .bind(page_id.to_string()) + .execute(&pool) + .await + .unwrap(); + + let dispatcher = CountingDispatcher::failing(); + let (handle, cancel) = spawn_with(&pool, dispatcher.clone()); + + // 3) Wait until the worker has dispatched once — the job goes back + // to `pending` because the retry is non-terminal (attempts < max). + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while dispatcher.call_count() == 0 && std::time::Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!(dispatcher.call_count() >= 1, "dispatcher must have run"); + + // Wait briefly for `record_duration` to either run or get gated. + // Then cancel before the next backoff fires (it'd burn 60s otherwise). + tokio::time::sleep(Duration::from_millis(200)).await; + cancel.cancel(); + handle.shutdown().await; + + // 4) The prior `done` row's `duration_ms` must NOT have been + // overwritten by the failed retry's duration. + let dur: Option = + sqlx::query_scalar("SELECT duration_ms FROM page_analysis WHERE page_id = $1") + .bind(page_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + dur, + Some(12345), + "non-terminal failure must not overwrite the prior done row's duration" + ); +} + #[sqlx::test(migrations = "./migrations")] async fn worker_isolates_dispatcher_panics(pool: PgPool) { let page_id = seed_page(&pool).await; diff --git a/backend/tests/api_admin_analysis.rs b/backend/tests/api_admin_analysis.rs index 7feb571..85cd4ff 100644 --- a/backend/tests/api_admin_analysis.rs +++ b/backend/tests/api_admin_analysis.rs @@ -392,6 +392,78 @@ async fn force_analyze_page_enqueues_with_force_flag(pool: PgPool) { assert_eq!(payload["force"].as_bool(), Some(true)); } +#[sqlx::test(migrations = "./migrations")] +async fn force_analyze_upgrades_pending_non_force_to_force(pool: PgPool) { + // The cover-up case the partial unique index used to silently + // create: when a `force=false` job is already pending and admin + // clicks force, the INSERT used to dedup-skip silently. The worker + // then picked up the non-force row, hit skip-if-done on a `done` + // page, and admin saw "queued" with no re-analysis. 0.87.11 + // upgrades the pending row in-place instead. + 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 (simulates the crawler's + // post-upload enqueue path). + mangalord::repo::page_analysis::enqueue_for_page(&pool, page_id, false) + .await + .unwrap(); + let payload_before: serde_json::Value = sqlx::query_scalar( + "SELECT payload FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(payload_before["force"].as_bool(), Some(false)); + + // 2) Admin force-click. + 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); + + // 3) STILL one row (no dedup duplicate), but its `force` flag + // flipped to true — the admin intent landed where the worker + // can see it. + let rows: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(rows, 1, "upgrade in place; no duplicate row"); + + let payload_after: serde_json::Value = sqlx::query_scalar( + "SELECT payload FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(payload_after["force"].as_bool(), Some(true)); + + // 4) Audit row carries the outcome so a moderator can tell + // upgrade-paths from fresh-inserts. + let audit_payload: serde_json::Value = sqlx::query_scalar( + "SELECT payload FROM admin_audit WHERE action = 'analysis_force_page'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + audit_payload["outcome"].as_str(), + Some("upgraded_pending_to_force") + ); +} + #[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/backend/tests/crawler_jobs.rs b/backend/tests/crawler_jobs.rs index ad4785c..ecc246c 100644 --- a/backend/tests/crawler_jobs.rs +++ b/backend/tests/crawler_jobs.rs @@ -824,3 +824,153 @@ async fn reap_done_zero_is_a_no_op(pool: PgPool) { assert_eq!(deleted, 0); assert_eq!(job_count(&pool).await, 1); } + +/// Migration 0031 ships a pre-dedup step (demote all-but-lowest-id +/// duplicates to `dead`) so it can apply cleanly on dirty production +/// databases where the prior buggy `enqueue_pages` left duplicate +/// `analyze_page` rows in `pending`/`running`. This test simulates the +/// dirty state by dropping the unique index, inserting duplicates, then +/// running the migration's pre-dedup SQL — the same SQL the migration +/// file runs — and confirming the duplicates land in `dead` so the +/// subsequent index creation would succeed. +#[sqlx::test(migrations = "./migrations")] +async fn migration_0031_prededup_demotes_duplicate_analyze_page_rows(pool: PgPool) { + // 1) Simulate pre-0031 dirty state: drop the dedup index so a second + // pending insert for the same page_id is accepted. + sqlx::query("DROP INDEX IF EXISTS crawler_jobs_analyze_page_dedup_idx") + .execute(&pool) + .await + .unwrap(); + + let page_a = Uuid::new_v4(); + let page_b = Uuid::new_v4(); + let make_row = |page_id: Uuid| { + serde_json::json!({ + "kind": "analyze_page", + "page_id": page_id, + "force": false + }) + }; + // Two pending duplicates for page A, plus one running + one pending + // for page B (the migration handles all in-flight states). + let a1: Uuid = sqlx::query_scalar( + "INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'pending') RETURNING id", + ) + .bind(make_row(page_a)) + .fetch_one(&pool) + .await + .unwrap(); + let a2: Uuid = sqlx::query_scalar( + "INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'pending') RETURNING id", + ) + .bind(make_row(page_a)) + .fetch_one(&pool) + .await + .unwrap(); + let b1: Uuid = sqlx::query_scalar( + "INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'running') RETURNING id", + ) + .bind(make_row(page_b)) + .fetch_one(&pool) + .await + .unwrap(); + let b2: Uuid = sqlx::query_scalar( + "INSERT INTO crawler_jobs (payload, state) VALUES ($1, 'pending') RETURNING id", + ) + .bind(make_row(page_b)) + .fetch_one(&pool) + .await + .unwrap(); + + // 2) Run the migration's pre-dedup statement verbatim. + sqlx::query( + "UPDATE crawler_jobs \ + SET state = 'dead', \ + last_error = COALESCE(last_error, '') || \ + CASE WHEN last_error IS NULL OR last_error = '' THEN '' \ + ELSE ' | ' END || \ + 'pre-0031 duplicate analyze_page; superseded by earlier sibling', \ + updated_at = now() \ + WHERE id IN ( \ + SELECT id FROM ( \ + SELECT id, \ + row_number() OVER ( \ + PARTITION BY payload->>'page_id' \ + ORDER BY id \ + ) AS rn \ + FROM crawler_jobs \ + WHERE payload->>'kind' = 'analyze_page' \ + AND state IN ('pending', 'running') \ + ) ranked \ + WHERE rn > 1 \ + )", + ) + .execute(&pool) + .await + .unwrap(); + + // 3) The earliest-id row per page_id stays in-flight; later + // duplicates moved to `dead` with the curator-recoverable + // last_error marker. Compute the expected winner per page by + // Postgres' UUID ordering (sql min, not Rust Ord — they differ). + let winner_a: Uuid = + sqlx::query_scalar("SELECT LEAST($1::uuid, $2::uuid)") + .bind(a1) + .bind(a2) + .fetch_one(&pool) + .await + .unwrap(); + let loser_a = if winner_a == a1 { a2 } else { a1 }; + let winner_b: Uuid = + sqlx::query_scalar("SELECT LEAST($1::uuid, $2::uuid)") + .bind(b1) + .bind(b2) + .fetch_one(&pool) + .await + .unwrap(); + let loser_b = if winner_b == b1 { b2 } else { b1 }; + + let state_a_winner: String = + sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1") + .bind(winner_a) + .fetch_one(&pool) + .await + .unwrap(); + let state_a_loser: String = + sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1") + .bind(loser_a) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + state_a_winner == "pending" || state_a_winner == "running", + "winner kept in-flight" + ); + assert_eq!(state_a_loser, "dead", "loser demoted to dead"); + + let state_b_winner: String = + sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1") + .bind(winner_b) + .fetch_one(&pool) + .await + .unwrap(); + let state_b_loser: String = + sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1") + .bind(loser_b) + .fetch_one(&pool) + .await + .unwrap(); + assert!(state_b_winner == "pending" || state_b_winner == "running"); + assert_eq!(state_b_loser, "dead"); + + // 4) After dedup, the index is creatable. Migration would now succeed. + sqlx::query( + "CREATE UNIQUE INDEX crawler_jobs_analyze_page_dedup_idx \ + ON crawler_jobs ((payload->>'page_id')) \ + WHERE state IN ('pending', 'running') \ + AND payload->>'kind' = 'analyze_page'", + ) + .execute(&pool) + .await + .expect("post-dedup index creation must succeed"); +} diff --git a/frontend/package.json b/frontend/package.json index b8be20f..a19fcbd 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.87.10", + "version": "0.87.11", "private": true, "type": "module", "scripts": {