diff --git a/backend/Cargo.lock b/backend/Cargo.lock index fd6e23c..a634053 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1558,7 +1558,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.90.2" +version = "0.90.3" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index a791cb9..32c27a4 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.90.2" +version = "0.90.3" edition = "2021" default-run = "mangalord" diff --git a/backend/src/api/admin/analysis.rs b/backend/src/api/admin/analysis.rs index 32d8d99..5776965 100644 --- a/backend/src/api/admin/analysis.rs +++ b/backend/src/api/admin/analysis.rs @@ -368,21 +368,15 @@ async fn reenqueue( (repo::page_analysis::ReenqueueScope::All, "analysis", None) }; + // Enqueue + audit in one transaction so a failed audit insert rolls the + // enqueue back too — the audit trail can't silently miss a re-enqueue that + // actually landed. Both are plain DB writes, so they share the tx cleanly + // (the live event + response are emitted only after commit). + let mut tx = state.db.begin().await?; let enqueued = - repo::page_analysis::enqueue_pages(&state.db, scope, body.only_unanalyzed).await?; - - // Push a live event so connected dashboards mark the in-scope pages as - // queued. Skip the no-op (nothing actually enqueued). - if enqueued > 0 { - state.analysis_events.publish(AnalysisEvent::Enqueued { - count: enqueued, - manga_id: body.manga_id, - chapter_id: body.chapter_id, - }); - } - + repo::page_analysis::enqueue_pages(&mut *tx, scope, body.only_unanalyzed).await?; repo::admin_audit::insert( - &state.db, + &mut *tx, admin.0.id, "analysis_reenqueue", target_type, @@ -395,6 +389,18 @@ async fn reenqueue( }), ) .await?; + tx.commit().await?; + + // Push a live event so connected dashboards mark the in-scope pages as + // queued. Skip the no-op (nothing actually enqueued). After commit so a + // rolled-back enqueue never emits a phantom event. + if enqueued > 0 { + state.analysis_events.publish(AnalysisEvent::Enqueued { + count: enqueued, + manga_id: body.manga_id, + chapter_id: body.chapter_id, + }); + } Ok(Json(ReenqueueResponse { enqueued })) } @@ -421,8 +427,13 @@ async fn analyze_page( // 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. + // + // Enqueue + audit in one transaction (via the `_conn` form) so a failed + // audit insert rolls the enqueue/upgrade back too — the audit trail can't + // miss a force-reanalyze that landed. + let mut tx = state.db.begin().await?; let outcome = - repo::page_analysis::enqueue_for_page(&state.db, page_id, true).await?; + repo::page_analysis::enqueue_for_page_conn(&mut tx, page_id, true).await?; let outcome_label = match outcome { repo::page_analysis::EnqueueForPageOutcome::Inserted => "inserted", @@ -430,7 +441,7 @@ async fn analyze_page( repo::page_analysis::EnqueueForPageOutcome::AlreadyEnqueued => "already_force_enqueued", }; repo::admin_audit::insert( - &state.db, + &mut *tx, admin.0.id, "analysis_force_page", "page", @@ -438,6 +449,7 @@ async fn analyze_page( json!({ "force": true, "outcome": outcome_label }), ) .await?; + tx.commit().await?; Ok(Json(AnalyzePageResponse { enqueued: true })) } diff --git a/backend/src/api/admin/storage.rs b/backend/src/api/admin/storage.rs index e227309..ccc8cc4 100644 --- a/backend/src/api/admin/storage.rs +++ b/backend/src/api/admin/storage.rs @@ -205,6 +205,13 @@ async fn backfill( resp.more_remaining = more_pages || more_covers; } + // Audit is written after the action here *by necessity*, not oversight + // (unlike reenqueue/analyze_page, which wrap their single DB mutation + + // audit in one tx). The backfill is a budgeted scan that interleaves + // filesystem `storage.size()` reads with per-batch DB writes across many + // iterations; wrapping it in a transaction would hold one open across all + // that I/O (a long-running tx). The batch size-writes are also idempotent + // recomputations, so a lost audit row has negligible impact. repo::admin_audit::insert( &state.db, admin.0.id, diff --git a/backend/src/crawler/jobs.rs b/backend/src/crawler/jobs.rs index 87c5e09..9ab0224 100644 --- a/backend/src/crawler/jobs.rs +++ b/backend/src/crawler/jobs.rs @@ -132,14 +132,17 @@ fn backoff_for(attempts: i32) -> Duration { /// `Skipped`. The slot frees again once the previous job leaves the /// in-flight states (done/failed/dead), so a re-enqueue after a force /// refetch succeeds. -pub async fn enqueue(pool: &PgPool, payload: &JobPayload) -> sqlx::Result { +pub async fn enqueue<'e, E>(executor: E, payload: &JobPayload) -> sqlx::Result +where + E: sqlx::PgExecutor<'e>, +{ let json = serde_json::to_value(payload).expect("JobPayload is always serializable"); let id: Option = sqlx::query_scalar( "INSERT INTO crawler_jobs (payload) VALUES ($1) \ ON CONFLICT DO NOTHING RETURNING id", ) .bind(json) - .fetch_optional(pool) + .fetch_optional(executor) .await?; Ok(match id { Some(id) => EnqueueResult::Inserted(id), @@ -422,7 +425,10 @@ pub async fn release( /// 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<()> { +pub async fn release_unowned<'e, E>(executor: E, lease_id: Uuid) -> sqlx::Result<()> +where + E: sqlx::PgExecutor<'e>, +{ let res = sqlx::query( "UPDATE crawler_jobs \ SET state = 'pending', leased_until = NULL, \ @@ -432,7 +438,7 @@ pub async fn release_unowned(pool: &PgPool, lease_id: Uuid) -> sqlx::Result<()> WHERE id = $1 AND state = 'running'", ) .bind(lease_id) - .execute(pool) + .execute(executor) .await?; if res.rows_affected() == 0 { tracing::warn!( diff --git a/backend/src/repo/page_analysis.rs b/backend/src/repo/page_analysis.rs index 58edf53..3258e46 100644 --- a/backend/src/repo/page_analysis.rs +++ b/backend/src/repo/page_analysis.rs @@ -78,13 +78,30 @@ pub enum EnqueueForPageOutcome { /// 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. +/// Pool convenience wrapper for [`enqueue_for_page_conn`]. Use this from +/// callers that don't need to share a transaction (chapter upload, the +/// crawler). The admin force-reanalyze handler uses the `_conn` form so the +/// enqueue and its `admin_audit` row commit together. pub async fn enqueue_for_page( pool: &PgPool, page_id: Uuid, force: bool, +) -> AppResult { + let mut conn = pool.acquire().await?; + enqueue_for_page_conn(&mut conn, page_id, force).await +} + +/// Enqueue (or force-upgrade) an `analyze_page` job on a caller-supplied +/// connection, so the admin handler can run it inside the same transaction as +/// its audit insert. The body is a sequence of single statements, each +/// reborrowing `&mut *conn`. +pub async fn enqueue_for_page_conn( + conn: &mut sqlx::PgConnection, + page_id: Uuid, + force: bool, ) -> AppResult { use crate::crawler::jobs::EnqueueResult; - match jobs::enqueue(pool, &JobPayload::AnalyzePage { page_id, force }).await? { + match jobs::enqueue(&mut *conn, &JobPayload::AnalyzePage { page_id, force }).await? { EnqueueResult::Inserted(_) => Ok(EnqueueForPageOutcome::Inserted), EnqueueResult::Skipped if !force => Ok(EnqueueForPageOutcome::AlreadyEnqueued), EnqueueResult::Skipped => { @@ -106,7 +123,7 @@ pub async fn enqueue_for_page( RETURNING id, state", ) .bind(page_id.to_string()) - .fetch_all(pool) + .fetch_all(&mut *conn) .await?; if upgraded.is_empty() { // Race: between the skipped INSERT and the UPDATE the @@ -115,7 +132,9 @@ pub async fn enqueue_for_page( // would succeed. Retry once to close the race without // unbounded loops. return Ok( - match jobs::enqueue(pool, &JobPayload::AnalyzePage { page_id, force }).await? { + match jobs::enqueue(&mut *conn, &JobPayload::AnalyzePage { page_id, force }) + .await? + { EnqueueResult::Inserted(_) => EnqueueForPageOutcome::Inserted, EnqueueResult::Skipped => EnqueueForPageOutcome::AlreadyEnqueued, }, @@ -139,7 +158,7 @@ pub async fn enqueue_for_page( // the original's ack would clobber it. for (id, state) in &upgraded { if state == "running" { - let _ = jobs::release_unowned(pool, *id).await; + let _ = jobs::release_unowned(&mut *conn, *id).await; } } Ok(EnqueueForPageOutcome::UpgradedToForce) @@ -168,11 +187,14 @@ pub enum ReenqueueScope { /// skip-if-done net would no-op them). Pages with a pending/running /// `analyze_page` job are always skipped so repeated calls don't pile up /// duplicates. Returns the number of jobs enqueued. -pub async fn enqueue_pages( - pool: &PgPool, +pub async fn enqueue_pages<'e, E>( + executor: E, scope: ReenqueueScope, only_unanalyzed: bool, -) -> AppResult { +) -> AppResult +where + E: sqlx::PgExecutor<'e>, +{ // Scope predicate; the bound uuid (when present) is always $2. let scope_clause = match scope { ReenqueueScope::All => "", @@ -206,7 +228,7 @@ pub async fn enqueue_pages( ReenqueueScope::All => query, ReenqueueScope::Manga(id) | ReenqueueScope::Chapter(id) => query.bind(id), }; - Ok(query.execute(pool).await?.rows_affected()) + Ok(query.execute(executor).await?.rows_affected()) } /// Per-manga analysis coverage for the admin overview. Only mangas that diff --git a/backend/tests/api_admin_analysis.rs b/backend/tests/api_admin_analysis.rs index 3b4ec78..a58d3a2 100644 --- a/backend/tests/api_admin_analysis.rs +++ b/backend/tests/api_admin_analysis.rs @@ -74,6 +74,38 @@ async fn analyze_job_count(pool: &PgPool) -> i64 { .unwrap() } +#[sqlx::test(migrations = "./migrations")] +async fn force_analyze_enqueue_rolls_back_with_its_transaction(pool: PgPool) { + // The admin force-reanalyze handler runs enqueue_for_page_conn + the audit + // insert in one transaction. Prove the enqueue genuinely participates in + // that tx: when the tx is abandoned (the path taken if the audit insert + // fails and the `?` propagates), no job survives — so the audit trail can + // never miss a force-reanalyze that actually landed. + let page_id = seed_pages(&pool, 1).await[0]; + + let mut tx = pool.begin().await.unwrap(); + let outcome = + mangalord::repo::page_analysis::enqueue_for_page_conn(&mut tx, page_id, true) + .await + .unwrap(); + assert!(matches!( + outcome, + mangalord::repo::page_analysis::EnqueueForPageOutcome::Inserted + )); + // Visible inside the open transaction… + let in_tx: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'", + ) + .fetch_one(&mut *tx) + .await + .unwrap(); + assert_eq!(in_tx, 1); + + // …but gone once the tx rolls back instead of committing. + tx.rollback().await.unwrap(); + assert_eq!(analyze_job_count(&pool).await, 0); +} + #[sqlx::test(migrations = "./migrations")] async fn chapter_upload_enqueues_one_analysis_job_per_page(pool: PgPool) { let h = common::harness_with_analysis(pool.clone()); @@ -194,6 +226,38 @@ async fn reenqueue_backfills_existing_pages(pool: PgPool) { assert_eq!(analyze_job_count(&pool).await, 3); } +#[sqlx::test(migrations = "./migrations")] +async fn reenqueue_writes_audit_row_atomically_with_jobs(pool: PgPool) { + // The enqueue and its admin_audit row are committed in one transaction, + // so a successful re-enqueue always leaves both the jobs AND exactly one + // matching audit row — the audit trail can't miss an enqueue that landed. + let h = common::harness_with_analysis(pool.clone()); + let cookie = seed_admin(&pool, &h.app).await; + seed_pages(&pool, 2).await; + + let resp = h + .app + .clone() + .oneshot(common::post_json_with_cookie( + "/api/v1/admin/analysis/reenqueue", + json!({ "only_unanalyzed": true }), + &cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(analyze_job_count(&pool).await, 2); + + let audit: serde_json::Value = sqlx::query_scalar( + "SELECT payload FROM admin_audit WHERE action = 'analysis_reenqueue'", + ) + .fetch_one(&pool) + .await + .expect("exactly one analysis_reenqueue audit row committed with the jobs"); + assert_eq!(audit["enqueued"], 2); + assert_eq!(audit["only_unanalyzed"], true); +} + #[sqlx::test(migrations = "./migrations")] async fn reenqueue_scoped_to_manga(pool: PgPool) { let h = common::harness_with_analysis(pool.clone()); diff --git a/frontend/package.json b/frontend/package.json index ba0db7f..5fa7fce 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.90.2", + "version": "0.90.3", "private": true, "type": "module", "scripts": {