fabi/Mangalord
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(admin): make force-reanalyze audit transactional; document storage exception

Browse Source 
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-07-01 07:16:08 +02:00
parent 5b76e0cc37
commit de510cc19a
8 changed files with 141 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/Cargo.lock generated
View File

@@ -1558,7 +1558,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
[[package]]
name = "mangalord"
version = "0.90.2"
version = "0.90.3"
dependencies = [
"anyhow",
"argon2",

2
backend/Cargo.toml
View File

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

42
backend/src/api/admin/analysis.rs
View File

@@ -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 }))
}

7
backend/src/api/admin/storage.rs
View File

@@ -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,

14
backend/src/crawler/jobs.rs
View File

@@ -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<EnqueueResult> {
pub async fn enqueue<'e, E>(executor: E, payload: &JobPayload) -> sqlx::Result<EnqueueResult>
where
E: sqlx::PgExecutor<'e>,
{
let json = serde_json::to_value(payload).expect("JobPayload is always serializable");
let id: Option<Uuid> = 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!(

38
backend/src/repo/page_analysis.rs
View File

@@ -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<EnqueueForPageOutcome> {
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<EnqueueForPageOutcome> {
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<u64> {
) -> AppResult<u64>
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

64
backend/tests/api_admin_analysis.rs
View File

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