diff --git a/backend/app/api/admin.py b/backend/app/api/admin.py
index b82aacf..9785cd7 100644
--- a/backend/app/api/admin.py
+++ b/backend/app/api/admin.py
@@ -19,7 +19,7 @@ from __future__ import annotations
 import hashlib
 
 from quart import Blueprint, jsonify, request
-from sqlalchemy import select
+from sqlalchemy import select, text
 
 from ..extensions import get_session
 from ..models import Artist
@@ -222,3 +222,49 @@ async def tags_purge_legacy():
             lambda sync_sess: purge_legacy_tags(sync_sess, dry_run=dry_run)
         )
     return jsonify(result)
+
+
+@admin_bp.route("/maintenance/db-stats", methods=["GET"])
+async def db_stats():
+    """Per-table bloat readout (pg_stat_user_tables) for the high-churn tables
+    so the operator can see when a VACUUM is worth running."""
+    from ..tasks.maintenance import VACUUM_TABLES
+
+    wanted = set(VACUUM_TABLES)
+    async with get_session() as session:
+        rows = (await session.execute(text(
+            "SELECT relname, n_live_tup, n_dead_tup, last_vacuum, "
+            "last_autovacuum, last_analyze FROM pg_stat_user_tables"
+        ))).all()
+
+    def _iso(v):
+        return v.isoformat() if v is not None else None
+
+    out = []
+    for r in rows:
+        if r.relname not in wanted:
+            continue
+        live = r.n_live_tup or 0
+        dead = r.n_dead_tup or 0
+        total = live + dead
+        out.append({
+            "table": r.relname,
+            "live": live,
+            "dead": dead,
+            "dead_pct": round(100 * dead / total, 1) if total else 0.0,
+            "last_vacuum": _iso(r.last_vacuum),
+            "last_autovacuum": _iso(r.last_autovacuum),
+            "last_analyze": _iso(r.last_analyze),
+        })
+    out.sort(key=lambda t: t["dead"], reverse=True)
+    return jsonify({"tables": out})
+
+
+@admin_bp.route("/maintenance/vacuum", methods=["POST"])
+async def trigger_vacuum():
+    """Operator-triggered VACUUM (ANALYZE) over the high-churn tables — the
+    same maintenance-queue task the weekly Beat schedule runs."""
+    from ..tasks.maintenance import vacuum_analyze
+
+    vacuum_analyze.delay()
+    return jsonify({"status": "queued"}), 202
diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py
index eca48e8..6620a6f 100644
--- a/backend/app/celery_app.py
+++ b/backend/app/celery_app.py
@@ -97,6 +97,10 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.maintenance.prune_task_runs",
                 "schedule": 86400.0,  # daily
             },
+            "vacuum-analyze": {
+                "task": "backend.app.tasks.maintenance.vacuum_analyze",
+                "schedule": 604800.0,  # weekly — reclaim dead-tuple bloat + refresh stats
+            },
             "fc3h-backup-db-nightly": {
                 "task": "backend.app.tasks.backup.backup_db_nightly",
                 "schedule": 3600.0,  # hourly tick; task self-gates on configured UTC hour
diff --git a/backend/app/tasks/_sync_engine.py b/backend/app/tasks/_sync_engine.py
index 508e490..eccfb84 100644
--- a/backend/app/tasks/_sync_engine.py
+++ b/backend/app/tasks/_sync_engine.py
@@ -34,3 +34,10 @@ def sync_session_factory():
         )
         _SESSIONMAKER = sessionmaker(_ENGINE, expire_on_commit=False)
     return _SESSIONMAKER
+
+
+def get_sync_engine():
+    """The process-wide sync Engine — for raw work that needs a connection
+    directly (e.g. AUTOCOMMIT VACUUM, which can't run inside a transaction)."""
+    sync_session_factory()  # ensure _ENGINE is initialized
+    return _ENGINE
diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py
index ada9f3e..999bff4 100644
--- a/backend/app/tasks/maintenance.py
+++ b/backend/app/tasks/maintenance.py
@@ -22,10 +22,29 @@ from ..models import (
     TaskRun,
 )
 from ..utils.phash import compute_phash
-from ._sync_engine import sync_session_factory as _sync_session_factory
+from ._sync_engine import (
+    get_sync_engine,
+    sync_session_factory as _sync_session_factory,
+)
 
 log = logging.getLogger(__name__)
 
+# High-churn tables whose dead-tuple bloat matters: the TABLESAMPLE showcase
+# reads physical blocks (bloat slows it directly), and the periodic
+# prune/backfill/recovery tasks generate dead tuples faster than autovacuum
+# always keeps up with. VACUUM reclaims them; ANALYZE refreshes planner stats.
+# Allowlist ONLY — names are interpolated into VACUUM, so they must never come
+# from request input.
+VACUUM_TABLES = (
+    "image_record",
+    "image_provenance",
+    "post_attachment",
+    "download_event",
+    "task_run",
+    "import_task",
+    "import_batch",
+)
+
 STUCK_THRESHOLD_MINUTES = 5
 # Archive ImportTasks run the per-member pipeline inline for every
 # member (import_archive_file: soft=30min/hard=35min). The ImportTask
@@ -761,3 +780,20 @@ def cleanup_old_download_events() -> int:
         )
         session.commit()
         return result.rowcount or 0
+
+
+@celery.task(name="backend.app.tasks.maintenance.vacuum_analyze")
+def vacuum_analyze() -> dict:
+    """Periodic VACUUM (ANALYZE) over the high-churn tables (VACUUM_TABLES) to
+    reclaim dead-tuple bloat and refresh planner statistics. VACUUM cannot run
+    inside a transaction block, so it runs on an AUTOCOMMIT connection.
+    Scheduled weekly; also operator-triggerable from Settings → Maintenance.
+    """
+    engine = get_sync_engine()
+    done = []
+    with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
+        for table in VACUUM_TABLES:
+            conn.exec_driver_sql(f"VACUUM (ANALYZE) {table}")
+            done.append(table)
+    log.info("vacuum_analyze complete: %s", done)
+    return {"vacuumed": done}
diff --git a/tests/test_api_admin.py b/tests/test_api_admin.py
index ec8edfe..08bf747 100644
--- a/tests/test_api_admin.py
+++ b/tests/test_api_admin.py
@@ -409,3 +409,29 @@ async def test_purge_legacy_commit_deletes_only_legacy(client, db):
         )
     )).scalar_one()
     assert gone == 0
+
+
+# --- DB maintenance: bloat readout + manual VACUUM trigger ----------
+
+
+@pytest.mark.asyncio
+async def test_db_stats_returns_table_bloat_shape(client):
+    resp = await client.get("/api/admin/maintenance/db-stats")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert "tables" in body
+    names = {t["table"] for t in body["tables"]}
+    assert "image_record" in names
+    for t in body["tables"]:
+        assert {"table", "live", "dead", "dead_pct", "last_vacuum"} <= set(t)
+
+
+@pytest.mark.asyncio
+async def test_trigger_vacuum_queues_the_task(client, monkeypatch):
+    from backend.app.tasks import maintenance
+
+    calls = []
+    monkeypatch.setattr(maintenance.vacuum_analyze, "delay", lambda: calls.append(1))
+    resp = await client.post("/api/admin/maintenance/vacuum")
+    assert resp.status_code == 202
+    assert calls == [1]
diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py
index 58ab626..9873d4d 100644
--- a/tests/test_maintenance.py
+++ b/tests/test_maintenance.py
@@ -638,3 +638,12 @@ def test_recover_stalled_download_dedupes_per_source(db_sync):
         select(Source.consecutive_failures).where(Source.id == sid)
     ).scalar_one()
     assert failures == 1
+
+
+def test_vacuum_analyze_runs_over_high_churn_tables():
+    """VACUUM (ANALYZE) runs (on its own AUTOCOMMIT connection) and reports the
+    tables it touched."""
+    from backend.app.tasks.maintenance import VACUUM_TABLES, vacuum_analyze
+
+    result = vacuum_analyze.apply().get()
+    assert result["vacuumed"] == list(VACUUM_TABLES)