perf(admin): fix O(mangas x jobs) overview query that pinned Postgres (0.85.1)
The admin overview dashboard polls /admin/overview every 30s. Its `manga_stats` aggregate evaluated MANGA_SYNC_STATE_CASE over the WHOLE library (14.5k mangas) with no LIMIT, and that CASE runs a correlated EXISTS over crawler_jobs. With no index on the `sync_chapter_list` manga_id path, Postgres seqscanned all in-flight jobs *per manga* — O(mangas x jobs). The disabled-analysis backlog (9.5k pending `analyze_page` rows that can never match the sync-kind filter) inflated every scan, so each call took 6-7 min. The 30s poll stacked ~10 of them concurrently → load 11, Postgres at ~100%. EXPLAIN ANALYZE on the live DB: the rewrite drops the query from 6-7 min to ~1.5s (per-manga crawler_jobs seqscan → single index-backed pass). Fixes: - Rewrite `manga_stats` to collect the in-flight manga-id set in ONE pass over the sync jobs (index-backed), then classify via a hash semi-join — O(mangas + jobs), immune to the analyze_page backlog size. - Add partial index crawler_jobs_sync_chapter_list_manga_idx (migration 0029) covering the sync_chapter_list manga_id path; mirrors the existing sync_manga index (0020). Also speeds the per-row Mangas tab listing. - statement_timeout backstop (5s) on the aggregate so a pathological plan can never pin a backend for minutes again. - Single-flight + 10s TTL cache on the /admin/overview handler so concurrent pollers coalesce instead of stacking. - Slow the dashboard poll 30s -> 60s (server-cached now anyway). The in_progress rule in the rewrite is kept in lockstep with the first arm of MANGA_SYNC_STATE_CASE (documented in both places). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2
backend/Cargo.lock
generated
2
backend/Cargo.lock
generated
@@ -1517,7 +1517,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mangalord"
|
name = "mangalord"
|
||||||
version = "0.85.0"
|
version = "0.85.1"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"argon2",
|
"argon2",
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mangalord"
|
name = "mangalord"
|
||||||
version = "0.85.0"
|
version = "0.85.1"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
default-run = "mangalord"
|
default-run = "mangalord"
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,24 @@
|
|||||||
|
-- Index the `sync_chapter_list` in-flight path used by the admin sync-state
|
||||||
|
-- derivation (overview `manga_stats` + the Mangas tab listing).
|
||||||
|
--
|
||||||
|
-- A manga is "in_progress" if a pending/running job targets it. For the
|
||||||
|
-- `sync_manga` kind that join is already covered by
|
||||||
|
-- crawler_jobs_sync_manga_key_idx (0020). The OTHER kind, `sync_chapter_list`,
|
||||||
|
-- carries the target `manga_id` directly in its payload and had NO index — so
|
||||||
|
-- the EXISTS fell back to a full seqscan of crawler_jobs. Per manga that is
|
||||||
|
-- cheap; across the whole library (overview scans every manga) it is O(mangas
|
||||||
|
-- x jobs), and the disabled-analysis backlog (thousands of pending
|
||||||
|
-- `analyze_page` rows that can never match this filter) inflates every scan.
|
||||||
|
--
|
||||||
|
-- Partial on the same `state IN ('pending','running') AND kind = ...`
|
||||||
|
-- predicate as the sibling indexes so it stays tiny (only in-flight list
|
||||||
|
-- jobs) and Postgres can probe it instead of scanning. Mirrors 0020.
|
||||||
|
--
|
||||||
|
-- Not CONCURRENTLY: sqlx::migrate! wraps each migration in a transaction;
|
||||||
|
-- CREATE INDEX CONCURRENTLY can't run inside one. The table is small at our
|
||||||
|
-- scale so a brief build lock on deploy is safe. IF NOT EXISTS keeps it
|
||||||
|
-- idempotent with any operator who pre-created it on the live DB.
|
||||||
|
CREATE INDEX IF NOT EXISTS crawler_jobs_sync_chapter_list_manga_idx
|
||||||
|
ON crawler_jobs ((payload->>'manga_id'))
|
||||||
|
WHERE state IN ('pending', 'running')
|
||||||
|
AND payload->>'kind' = 'sync_chapter_list';
|
||||||
@@ -9,11 +9,15 @@
|
|||||||
//!
|
//!
|
||||||
//! Admin-only (`RequireAdmin`, cookie-only).
|
//! Admin-only (`RequireAdmin`, cookie-only).
|
||||||
|
|
||||||
|
use std::sync::OnceLock;
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
use axum::extract::State;
|
use axum::extract::State;
|
||||||
use axum::routing::get;
|
use axum::routing::get;
|
||||||
use axum::{Json, Router};
|
use axum::{Json, Router};
|
||||||
use chrono::{DateTime, Utc};
|
use chrono::{DateTime, Utc};
|
||||||
use serde::Serialize;
|
use serde::Serialize;
|
||||||
|
use tokio::sync::Mutex;
|
||||||
|
|
||||||
use crate::app::AppState;
|
use crate::app::AppState;
|
||||||
use crate::auth::extractor::RequireAdmin;
|
use crate::auth::extractor::RequireAdmin;
|
||||||
@@ -24,14 +28,14 @@ pub fn routes() -> Router<AppState> {
|
|||||||
Router::new().route("/admin/overview", get(overview))
|
Router::new().route("/admin/overview", get(overview))
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Serialize)]
|
#[derive(Debug, Clone, Serialize)]
|
||||||
pub struct OverviewStats {
|
pub struct OverviewStats {
|
||||||
pub users: UsersOverview,
|
pub users: UsersOverview,
|
||||||
pub mangas: repo::admin_view::MangaStats,
|
pub mangas: repo::admin_view::MangaStats,
|
||||||
pub analysis: AnalysisOverview,
|
pub analysis: AnalysisOverview,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Serialize)]
|
#[derive(Debug, Clone, Serialize)]
|
||||||
pub struct UsersOverview {
|
pub struct UsersOverview {
|
||||||
pub total: i64,
|
pub total: i64,
|
||||||
pub admins: i64,
|
pub admins: i64,
|
||||||
@@ -39,16 +43,42 @@ pub struct UsersOverview {
|
|||||||
pub newest_created_at: Option<DateTime<Utc>>,
|
pub newest_created_at: Option<DateTime<Utc>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Serialize)]
|
#[derive(Debug, Clone, Serialize)]
|
||||||
pub struct AnalysisOverview {
|
pub struct AnalysisOverview {
|
||||||
pub analyzed_pages: i64,
|
pub analyzed_pages: i64,
|
||||||
pub total_pages: i64,
|
pub total_pages: i64,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Process-wide single-flight cache for the overview aggregate.
|
||||||
|
///
|
||||||
|
/// The admin dashboard polls this endpoint on a timer; without coalescing, a
|
||||||
|
/// slow DB (e.g. a deep crawl/analysis backlog) lets identical requests stack
|
||||||
|
/// and pin Postgres. Holding the async mutex across the recompute gives BOTH a
|
||||||
|
/// short-TTL cache and single-flight: concurrent pollers block on the lock,
|
||||||
|
/// then read the freshly cached value instead of each launching their own
|
||||||
|
/// aggregate. Resets on process restart, which is fine for a dashboard stat.
|
||||||
|
///
|
||||||
|
/// Unlike per-`AppState` shared state (e.g. `auth_limiter`), this is a single
|
||||||
|
/// production-instance cache, so a process global is adequate. It is
|
||||||
|
/// test-safe because the cache lives in the handler *body*: the two auth tests
|
||||||
|
/// are rejected at the `RequireAdmin` extractor before reaching it, and the
|
||||||
|
/// one aggregate test issues a single request. A future test that mutates data
|
||||||
|
/// and re-reads within the TTL would need to account for this.
|
||||||
|
static OVERVIEW_CACHE: OnceLock<Mutex<Option<(Instant, OverviewStats)>>> = OnceLock::new();
|
||||||
|
const OVERVIEW_TTL: Duration = Duration::from_secs(10);
|
||||||
|
|
||||||
async fn overview(
|
async fn overview(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
_admin: RequireAdmin,
|
_admin: RequireAdmin,
|
||||||
) -> AppResult<Json<OverviewStats>> {
|
) -> AppResult<Json<OverviewStats>> {
|
||||||
|
let cache = OVERVIEW_CACHE.get_or_init(|| Mutex::new(None));
|
||||||
|
let mut guard = cache.lock().await;
|
||||||
|
if let Some((at, cached)) = guard.as_ref() {
|
||||||
|
if at.elapsed() < OVERVIEW_TTL {
|
||||||
|
return Ok(Json(cached.clone()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Independent reads — run concurrently so latency is the slowest
|
// Independent reads — run concurrently so latency is the slowest
|
||||||
// query, not their sum (mirrors the storage handler).
|
// query, not their sum (mirrors the storage handler).
|
||||||
let (user_counts, user_newest, mangas, coverage) = tokio::try_join!(
|
let (user_counts, user_newest, mangas, coverage) = tokio::try_join!(
|
||||||
@@ -65,7 +95,7 @@ async fn overview(
|
|||||||
};
|
};
|
||||||
let (analyzed_pages, total_pages) = coverage;
|
let (analyzed_pages, total_pages) = coverage;
|
||||||
|
|
||||||
Ok(Json(OverviewStats {
|
let stats = OverviewStats {
|
||||||
users: UsersOverview {
|
users: UsersOverview {
|
||||||
total,
|
total,
|
||||||
admins,
|
admins,
|
||||||
@@ -77,5 +107,7 @@ async fn overview(
|
|||||||
analyzed_pages,
|
analyzed_pages,
|
||||||
total_pages,
|
total_pages,
|
||||||
},
|
},
|
||||||
}))
|
};
|
||||||
|
*guard = Some((Instant::now(), stats.clone()));
|
||||||
|
Ok(Json(stats))
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -85,7 +85,7 @@ const MANGA_SYNC_STATE_CASE: &str = r#"
|
|||||||
|
|
||||||
/// Library shape for the admin overview: total mangas split by derived
|
/// Library shape for the admin overview: total mangas split by derived
|
||||||
/// sync state, plus library-wide chapter/page totals and the newest manga.
|
/// sync state, plus library-wide chapter/page totals and the newest manga.
|
||||||
#[derive(Debug, Serialize)]
|
#[derive(Debug, Clone, Serialize)]
|
||||||
pub struct MangaStats {
|
pub struct MangaStats {
|
||||||
pub total: i64,
|
pub total: i64,
|
||||||
pub synced: i64,
|
pub synced: i64,
|
||||||
@@ -97,23 +97,71 @@ pub struct MangaStats {
|
|||||||
pub newest_seen_at: Option<DateTime<Utc>>,
|
pub newest_seen_at: Option<DateTime<Utc>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Aggregate the manga library for the overview dashboard. Sync-state
|
/// Aggregate the manga library for the overview dashboard.
|
||||||
/// counts reuse `MANGA_SYNC_STATE_CASE` so they can't drift from the
|
///
|
||||||
/// per-row classification on the Mangas tab.
|
/// Unlike the paginated Mangas tab (`list_mangas_with_sync_state`, which
|
||||||
|
/// evaluates `MANGA_SYNC_STATE_CASE` for only the page slice), this scans the
|
||||||
|
/// WHOLE library. The per-row correlated form is therefore O(mangas x jobs):
|
||||||
|
/// each manga seqscans `crawler_jobs`, and the disabled-analysis backlog
|
||||||
|
/// inflates every scan. Instead we collect the set of "in-flight" manga ids in
|
||||||
|
/// ONE pass over the in-flight sync jobs (index-backed: 0020 + 0029), then
|
||||||
|
/// classify each manga with a hash semi-join — O(mangas + jobs).
|
||||||
|
///
|
||||||
|
/// The `in_progress` rule here MUST stay in lockstep with the first arm of
|
||||||
|
/// `MANGA_SYNC_STATE_CASE`: pending/running `sync_chapter_list` (by manga_id)
|
||||||
|
/// OR `sync_manga` (resolved through `manga_sources`). The `dropped`/`synced`
|
||||||
|
/// arms are identical to the shared case.
|
||||||
pub async fn manga_stats(pool: &PgPool) -> AppResult<MangaStats> {
|
pub async fn manga_stats(pool: &PgPool) -> AppResult<MangaStats> {
|
||||||
let counts_sql = format!(
|
let counts_sql = r#"
|
||||||
r#"
|
WITH in_progress_mangas AS (
|
||||||
|
-- sync_chapter_list jobs carry the target manga_id directly
|
||||||
|
SELECT (cj.payload->>'manga_id')::uuid AS manga_id
|
||||||
|
FROM crawler_jobs cj
|
||||||
|
WHERE cj.state IN ('pending', 'running')
|
||||||
|
AND cj.payload->>'kind' = 'sync_chapter_list'
|
||||||
|
AND cj.payload->>'manga_id' IS NOT NULL
|
||||||
|
UNION
|
||||||
|
-- sync_manga jobs resolve to a manga via manga_sources
|
||||||
|
SELECT ms.manga_id
|
||||||
|
FROM crawler_jobs cj
|
||||||
|
JOIN manga_sources ms
|
||||||
|
ON ms.source_id = cj.payload->>'source_id'
|
||||||
|
AND ms.source_manga_key = cj.payload->>'source_manga_key'
|
||||||
|
WHERE cj.state IN ('pending', 'running')
|
||||||
|
AND cj.payload->>'kind' = 'sync_manga'
|
||||||
|
)
|
||||||
SELECT
|
SELECT
|
||||||
COUNT(*)::bigint AS total,
|
COUNT(*)::bigint AS total,
|
||||||
COUNT(*) FILTER (WHERE s = 'synced')::bigint AS synced,
|
COUNT(*) FILTER (WHERE s = 'synced')::bigint AS synced,
|
||||||
COUNT(*) FILTER (WHERE s = 'in_progress')::bigint AS in_progress,
|
COUNT(*) FILTER (WHERE s = 'in_progress')::bigint AS in_progress,
|
||||||
COUNT(*) FILTER (WHERE s = 'dropped')::bigint AS dropped
|
COUNT(*) FILTER (WHERE s = 'dropped')::bigint AS dropped
|
||||||
FROM (SELECT {case} AS s FROM mangas m) q
|
FROM (
|
||||||
"#,
|
SELECT
|
||||||
case = MANGA_SYNC_STATE_CASE
|
CASE
|
||||||
);
|
WHEN m.id IN (SELECT manga_id FROM in_progress_mangas)
|
||||||
|
THEN 'in_progress'
|
||||||
|
WHEN EXISTS (SELECT 1 FROM manga_sources ms WHERE ms.manga_id = m.id)
|
||||||
|
AND NOT EXISTS (
|
||||||
|
SELECT 1 FROM manga_sources ms
|
||||||
|
WHERE ms.manga_id = m.id AND ms.dropped_at IS NULL
|
||||||
|
)
|
||||||
|
THEN 'dropped'
|
||||||
|
ELSE 'synced'
|
||||||
|
END AS s
|
||||||
|
FROM mangas m
|
||||||
|
) q
|
||||||
|
"#;
|
||||||
|
// Backstop: even with the single-pass rewrite + indexes this is cheap, but
|
||||||
|
// a `statement_timeout` guarantees a pathological plan can never pin a
|
||||||
|
// backend for minutes and let pollers stack again. SET LOCAL is scoped to
|
||||||
|
// the transaction.
|
||||||
|
let mut tx = pool.begin().await?;
|
||||||
|
sqlx::query("SET LOCAL statement_timeout = '5s'")
|
||||||
|
.execute(&mut *tx)
|
||||||
|
.await?;
|
||||||
let (total, synced, in_progress, dropped): (i64, i64, i64, i64) =
|
let (total, synced, in_progress, dropped): (i64, i64, i64, i64) =
|
||||||
sqlx::query_as(&counts_sql).fetch_one(pool).await?;
|
sqlx::query_as(counts_sql).fetch_one(&mut *tx).await?;
|
||||||
|
tx.commit().await?;
|
||||||
|
|
||||||
let (total_chapters,): (i64,) =
|
let (total_chapters,): (i64,) =
|
||||||
sqlx::query_as("SELECT COUNT(*)::bigint FROM chapters")
|
sqlx::query_as("SELECT COUNT(*)::bigint FROM chapters")
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "mangalord-frontend",
|
"name": "mangalord-frontend",
|
||||||
"version": "0.85.0",
|
"version": "0.85.1",
|
||||||
"private": true,
|
"private": true,
|
||||||
"type": "module",
|
"type": "module",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
|
|||||||
@@ -190,10 +190,13 @@
|
|||||||
openCrawlerStream();
|
openCrawlerStream();
|
||||||
openAnalysisStream();
|
openAnalysisStream();
|
||||||
sysTimer = setInterval(refreshSys, 5000);
|
sysTimer = setInterval(refreshSys, 5000);
|
||||||
|
// 60s: the overview aggregate is server-cached (~10s TTL) and these
|
||||||
|
// are background freshness updates, so a slower poll halves idle DB
|
||||||
|
// load without the dashboard feeling stale.
|
||||||
overviewTimer = setInterval(() => {
|
overviewTimer = setInterval(() => {
|
||||||
refreshOverview();
|
refreshOverview();
|
||||||
refreshMetrics();
|
refreshMetrics();
|
||||||
}, 30000);
|
}, 60000);
|
||||||
if (typeof document !== 'undefined') {
|
if (typeof document !== 'undefined') {
|
||||||
document.addEventListener('visibilitychange', onVisibilityChange);
|
document.addEventListener('visibilitychange', onVisibilityChange);
|
||||||
window.addEventListener('pagehide', onPageHide);
|
window.addEventListener('pagehide', onPageHide);
|
||||||
|
|||||||
Reference in New Issue
Block a user