fix(analyze-dedup): migration pre-dedup, force-collision upgrade, gate test (0.87.11)

Four 0.87.6 follow-ups from the adversarial review:

1. **Migration 0031 pre-dedup.** Demote all-but-lowest-id duplicate
   `analyze_page` rows in `(pending|running)` to `dead` before
   creating the unique index, with a curator-recoverable last_error
   marker. Without this, `sqlx::migrate!` would refuse to boot on
   any dirty production DB.

2. **`enqueue_for_page(force=true)` collision.** The partial unique
   index used to silently swallow force requests when a
   `force=false` job was already pending. Repo function now upgrades
   the pending row's `force` flag in place (or falls through to
   re-INSERT if the sibling drained mid-call), and reports an
   `EnqueueForPageOutcome` for accurate auditing.

3. **`record_duration` gate test.** New test seeds a `done` row with
   known duration, force-re-analyzes with failing dispatcher +
   max_attempts=3 (non-terminal), asserts duration_ms wasn't
   overwritten.

4. **`bookmark.rs` PK comment correction.** Use `b.id DESC` instead
   of the wrong `manga_id`; PK is actually `id`, and migration 0004
   allows both chapter-level and manga-level bookmarks on the same
   manga.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/Cargo.lock

@@ -1517,7 +1517,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
 
 [[package]]
 name = "mangalord"
-version = "0.87.10"
+version = "0.87.11"
 dependencies = [
  "anyhow",
  "argon2",

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mangalord"
-version = "0.87.10"
+version = "0.87.11"
 edition = "2021"
 default-run = "mangalord"
 

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

backend/src/api/admin/analysis.rs

@@ -415,15 +415,27 @@ async fn analyze_page(
         return Err(AppError::NotFound);
     }
 
+    // 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?;
 

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
         "#,
     )

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<EnqueueForPageOutcome> {
+    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<Uuid> = 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.

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<i64> =
+        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;

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());

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");
+}

frontend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "mangalord-frontend",
-  "version": "0.87.10",
+  "version": "0.87.11",
   "private": true,
   "type": "module",
   "scripts": {