Merge pull request 'DB maintenance + health observability + Postgres integration CI lane' (#72) from dev into main
CI & Build / Python lint (push) Successful in 3s
CI & Build / TypeScript typecheck (push) Successful in 34s
CI & Build / integration (push) Successful in 35s
CI & Build / Python tests (push) Successful in 51s
CI & Build / Build & push image (push) Successful in 14s

This commit was merged in pull request #72.
This commit is contained in:
2026-06-14 20:54:34 -04:00
12 changed files with 915 additions and 4 deletions
+69 -1
View File
@@ -142,7 +142,75 @@ jobs:
uv pip install --python /opt/venv/bin/python -e ".[dev]"
- name: Run tests
run: /opt/venv/bin/python -m pytest tests/ -q
# Integration tests (real Postgres) run in the `integration` job below.
run: /opt/venv/bin/python -m pytest tests/ -q -m "not integration"
# Real-Postgres lane (family rule 6). Exercises the async SQLAlchemy connection
# path the unit stubs can't reach — the un-awaited execution_options regression
# that made every VACUUM report 0/6 lived here. Like `test`, it runs for
# visibility and does NOT gate the build.
#
# Job key stays separator-free ("integration"): act_runner derives the service-
# container name from the (truncated) job display name and the discovery step
# filters `docker ps` by it. Service hostnames aren't routable on this runner,
# so the step resolves the Postgres container's bridge IP. No `name:` on purpose.
integration:
if: github.ref == 'refs/heads/dev' || github.ref == 'refs/heads/main' || startsWith(github.ref, 'refs/tags/v')
runs-on: python-ci
container:
image: git.fabledsword.com/bvandeusen/ci-python:3.14
env:
# Config + the module engine read these at import time. DATABASE_URL itself
# is built from the discovered service IP in the run step.
SECRET_KEY: ci_integration_placeholder
services:
postgres:
image: postgres:16-alpine
env:
POSTGRES_USER: scribe
POSTGRES_PASSWORD: ci_integration
POSTGRES_DB: scribe_test
options: >-
--health-cmd "pg_isready -U scribe"
--health-interval 10s
--health-timeout 5s
--health-retries 10
steps:
- uses: actions/checkout@v6
- name: Create virtual environment
run: uv venv /opt/venv
- name: Install package with dev deps
run: |
uv pip install --python /opt/venv/bin/python setuptools wheel
uv pip install --python /opt/venv/bin/python --no-build-isolation http-ece
uv pip install --python /opt/venv/bin/python -e ".[dev]"
- name: Integration suite (resolve service IP, migrate, test)
run: |
set -eux
echo "=== container landscape (diagnostic for the name filter) ==="
docker ps -a --format '{{.ID}} {{.Image}} -> {{.Names}}'
PG=$(docker ps --filter "name=integration" --filter "ancestor=postgres:16-alpine" -q | head -n1)
test -n "$PG"
PG_IP=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$PG")
test -n "$PG_IP"
export DATABASE_URL="postgresql+asyncpg://scribe:ci_integration@${PG_IP}:5432/scribe_test"
# Wait for Postgres to accept connections (busybox sh — the runner
# default — has no bash /dev/tcp, so use Python).
/opt/venv/bin/python - "$PG_IP" <<'PY'
import socket, sys, time
for _ in range(30):
try:
socket.create_connection((sys.argv[1], 5432), timeout=2).close()
break
except OSError:
time.sleep(1)
else:
sys.exit("postgres did not become reachable")
PY
# Real migrations build the schema; the maintenance tests then run
# VACUUM (ANALYZE) and read pg_stat_user_tables against it.
/opt/venv/bin/alembic upgrade head
/opt/venv/bin/python -m pytest tests/ -v -m integration
build:
name: Build & push image
+220
View File
@@ -150,6 +150,33 @@ const serverMarketplaceUrl = ref('');
const adminMarketplaceUrl = ref('');
const savingMarketplaceUrl = ref(false);
const marketplaceUrlSaved = ref(false);
// DB maintenance (admin) — daily targeted VACUUM (ANALYZE).
interface DbMaintTableResult { table: string; ok: boolean; elapsed_ms: number; error: string | null }
interface DbMaintRun { started_at: string; elapsed_ms: number; tables: DbMaintTableResult[] }
const dbMaintEnabled = ref(true);
const dbMaintHour = ref(4);
const dbMaintLastRun = ref<DbMaintRun | null>(null);
const savingDbMaint = ref(false);
const dbMaintSaved = ref(false);
const runningDbMaint = ref(false);
interface DbTableHealth {
table: string; live: number; dead: number; dead_pct: number;
total_bytes: number; mod_since_analyze: number;
last_vacuum: string | null; last_analyze: string | null;
}
interface DbHealth { db_bytes: number; tables: DbTableHealth[] }
const dbHealth = ref<DbHealth | null>(null);
const loadingHealth = ref(false);
const DEAD_PCT_WARN = 20; // dead-tuple ratio above this = autovacuum falling behind
function formatBytes(n: number): string {
if (n < 1024) return `${n} B`;
const units = ["KB", "MB", "GB", "TB"];
let v = n / 1024, i = 0;
while (v >= 1024 && i < units.length - 1) { v /= 1024; i++; }
return `${v.toFixed(v >= 10 || i === 0 ? 0 : 1)} ${units[i]}`;
}
const pluginInstallCommands = computed(() => {
const mkt = (pluginMarketplaceUrl.value || '').trim()
|| serverMarketplaceUrl.value
@@ -448,6 +475,19 @@ onMounted(async () => {
// not configured yet
}
// DB maintenance config (admin only — endpoint is admin-gated).
if (authStore.isAdmin) {
try {
const dm = await apiGet<{ enabled: boolean; hour: number; last_run: DbMaintRun | null }>("/api/admin/db-maintenance");
dbMaintEnabled.value = dm.enabled;
dbMaintHour.value = dm.hour;
dbMaintLastRun.value = dm.last_run;
} catch {
// leave defaults
}
await loadDbHealth();
}
// Load admin settings
if (authStore.isAdmin) {
try {
@@ -697,6 +737,54 @@ async function saveMarketplaceUrl() {
}
}
async function saveDbMaintenance() {
savingDbMaint.value = true;
dbMaintSaved.value = false;
try {
await apiPut("/api/admin/db-maintenance", {
enabled: dbMaintEnabled.value,
hour: dbMaintHour.value,
});
dbMaintSaved.value = true;
setTimeout(() => (dbMaintSaved.value = false), 2000);
} catch (e) {
const body = (e as { body?: { error?: string } }).body;
toastStore.show(body?.error || "Failed to save maintenance settings", "error");
} finally {
savingDbMaint.value = false;
}
}
async function loadDbHealth() {
loadingHealth.value = true;
try {
dbHealth.value = await apiGet<DbHealth>("/api/admin/db-maintenance/health");
} catch {
// leave previous value
} finally {
loadingHealth.value = false;
}
}
async function runDbMaintenanceNow() {
runningDbMaint.value = true;
try {
const summary = await apiPost<DbMaintRun>("/api/admin/db-maintenance/run", {});
dbMaintLastRun.value = summary;
const failed = summary.tables.filter((t) => !t.ok).length;
toastStore.show(
failed ? `Maintenance ran with ${failed} error(s)` : "Maintenance complete",
failed ? "error" : "success",
);
await loadDbHealth(); // reflect the dead-tuple drop
} catch (e) {
const body = (e as { body?: { error?: string } }).body;
toastStore.show(body?.error || "Maintenance run failed", "error");
} finally {
runningDbMaint.value = false;
}
}
async function handleRestoreFile(event: Event) {
const file = (event.target as HTMLInputElement).files?.[0];
if (!file) return;
@@ -1724,6 +1812,86 @@ function formatUserDate(iso: string): string {
</div>
</section>
<section class="settings-section full-width">
<h2>Database maintenance</h2>
<p class="section-desc">
A daily <code>VACUUM (ANALYZE)</code> over the high-churn tables (logs, notifications,
tokens, notes, version history) on top of Postgres autovacuum to reclaim space left
by the nightly cleanup sweeps and keep query plans fresh. Runs at the hour below (UTC),
just after trash purge.
</p>
<div class="checkbox-field">
<label>
<input type="checkbox" v-model="dbMaintEnabled" />
Run scheduled maintenance daily
</label>
</div>
<div class="field url-field">
<label for="db-maint-hour">Run hour (UTC)</label>
<select id="db-maint-hour" v-model.number="dbMaintHour" class="input">
<option v-for="h in 24" :key="h - 1" :value="h - 1">
{{ String(h - 1).padStart(2, '0') }}:00
</option>
</select>
</div>
<div class="actions">
<button class="btn-save" @click="saveDbMaintenance" :disabled="savingDbMaint">
{{ savingDbMaint ? "Saving..." : "Save" }}
</button>
<button class="btn-save btn-secondary" @click="runDbMaintenanceNow" :disabled="runningDbMaint">
{{ runningDbMaint ? "Running..." : "Run now" }}
</button>
<span v-if="dbMaintSaved" class="saved-msg">Saved!</span>
</div>
<div v-if="dbMaintLastRun" class="db-maint-last">
<span class="db-maint-last-label">
Last run {{ new Date(dbMaintLastRun.started_at).toLocaleString() }}
· {{ dbMaintLastRun.elapsed_ms }}ms
</span>
<ul class="db-maint-table-list">
<li v-for="t in dbMaintLastRun.tables" :key="t.table" :class="{ 'dm-failed': !t.ok }">
<code>{{ t.table }}</code>
<span class="dm-status">{{ t.ok ? `${t.elapsed_ms}ms` : `${t.error}` }}</span>
</li>
</ul>
</div>
<div class="db-health">
<h3 class="subsection-label">
Table health
<span v-if="dbHealth" class="db-health-total">· database {{ formatBytes(dbHealth.db_bytes) }}</span>
</h3>
<p class="field-hint">
Dead-tuple ratio is bloat rows left by updates/deletes not yet reclaimed.
Above {{ DEAD_PCT_WARN }}% on a large table means autovacuum is falling behind;
consider adding it to the maintenance set.
</p>
<p v-if="loadingHealth && !dbHealth" class="field-hint">Loading</p>
<div v-else-if="dbHealth" class="db-health-scroll">
<table class="db-health-table">
<thead>
<tr>
<th>Table</th><th class="num">Size</th><th class="num">Live</th>
<th class="num">Dead</th><th class="num">Dead %</th>
<th>Last vacuum</th><th>Last analyze</th>
</tr>
</thead>
<tbody>
<tr v-for="t in dbHealth.tables" :key="t.table" :class="{ 'dh-warn': t.dead_pct >= DEAD_PCT_WARN }">
<td><code>{{ t.table }}</code></td>
<td class="num">{{ formatBytes(t.total_bytes) }}</td>
<td class="num">{{ t.live.toLocaleString() }}</td>
<td class="num">{{ t.dead.toLocaleString() }}</td>
<td class="num">{{ t.dead_pct }}%</td>
<td>{{ t.last_vacuum ? new Date(t.last_vacuum).toLocaleString() : "—" }}</td>
<td>{{ t.last_analyze ? new Date(t.last_analyze).toLocaleString() : "—" }}</td>
</tr>
</tbody>
</table>
</div>
</div>
</section>
<section class="settings-section full-width">
<h2>Email / SMTP</h2>
<p class="section-desc">Configure SMTP to enable email notifications for all users.</p>
@@ -2352,6 +2520,58 @@ function formatUserDate(iso: string): string {
}
.btn-secondary:hover:not(:disabled) { background: var(--color-action-secondary-hover); }
.btn-secondary:disabled { opacity: 0.6; cursor: default; }
/* DB maintenance last-run summary */
.db-maint-last { margin-top: 1rem; }
.db-maint-last-label {
display: block;
font-size: 0.8rem;
color: var(--color-text-muted);
margin-bottom: 0.4rem;
}
.db-maint-table-list {
list-style: none;
margin: 0;
padding: 0;
display: flex;
flex-direction: column;
gap: 0.2rem;
}
.db-maint-table-list li {
display: flex;
justify-content: space-between;
gap: 1rem;
font-size: 0.82rem;
padding: 0.2rem 0;
}
.db-maint-table-list .dm-status { color: var(--color-text-muted); }
.db-maint-table-list li.dm-failed .dm-status { color: var(--color-danger); }
/* DB table-health readout */
.db-health { margin-top: 1.5rem; }
.db-health-total { color: var(--color-text-muted); font-weight: 400; }
.db-health-scroll { overflow-x: auto; margin-top: 0.5rem; }
.db-health-table {
width: 100%;
border-collapse: collapse;
font-size: 0.82rem;
}
.db-health-table th, .db-health-table td {
text-align: left;
padding: 0.35rem 0.6rem;
border-bottom: 1px solid var(--color-border);
white-space: nowrap;
}
.db-health-table th {
font-size: 0.72rem;
text-transform: uppercase;
letter-spacing: 0.04em;
color: var(--color-text-muted);
font-weight: 500;
}
.db-health-table td.num, .db-health-table th.num { text-align: right; font-variant-numeric: tabular-nums; }
.db-health-table tr.dh-warn td { color: var(--color-warning); }
.db-health-table tr.dh-warn td:first-child code { color: var(--color-warning); }
.btn-warn:hover:not(:disabled) {
background: var(--color-warning);
color: #fff;
+3
View File
@@ -36,6 +36,9 @@ where = ["src"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
markers = [
"integration: requires a real Postgres database (runs only in the CI integration lane)",
]
[tool.ruff]
line-length = 120
+13
View File
@@ -187,6 +187,15 @@ def create_app() -> Quart:
from scribe.services.trash_scheduler import start_trash_scheduler
start_trash_scheduler(asyncio.get_running_loop())
# DB maintenance scheduler (daily targeted VACUUM ANALYZE, default 04:00 UTC)
from scribe.services.db_maintenance_scheduler import (
get_maintenance_hour,
start_db_maintenance_scheduler,
)
start_db_maintenance_scheduler(
asyncio.get_running_loop(), await get_maintenance_hour()
)
# Diagnostic instrumentation — heartbeat, signal handlers, asyncio
# exception hook. Cheap (~1 log line/min), high diagnostic value when
# the app crashes mysteriously. See services/diagnostics.py.
@@ -203,6 +212,10 @@ def create_app() -> Quart:
stop_version_pinning_scheduler()
from scribe.services.trash_scheduler import stop_trash_scheduler
stop_trash_scheduler()
from scribe.services.db_maintenance_scheduler import (
stop_db_maintenance_scheduler,
)
stop_db_maintenance_scheduler()
from scribe.services.diagnostics import stop_diagnostics
stop_diagnostics()
+69 -1
View File
@@ -21,7 +21,12 @@ from scribe.services.backup import (
from scribe.services.email import SMTP_SETTING_KEYS, get_base_url, get_smtp_config, is_smtp_configured, send_test_email
from scribe.services.logging import get_logs, get_log_stats, log_audit
from scribe.services.notifications import send_invitation_email
from scribe.services.settings import set_setting, set_settings_batch
from scribe.services.settings import (
get_admin_setting,
set_admin_setting,
set_setting,
set_settings_batch,
)
admin_bp = Blueprint("admin", __name__, url_prefix="/api/admin")
@@ -204,6 +209,69 @@ async def update_base_url():
return jsonify({"status": "ok"})
@admin_bp.route("/db-maintenance", methods=["GET"])
@admin_required
async def get_db_maintenance():
"""Current DB-maintenance config + the last run's summary."""
from scribe.services.db_maintenance import get_last_run
from scribe.services.db_maintenance_scheduler import (
get_maintenance_hour,
is_maintenance_enabled,
)
return jsonify({
"enabled": await is_maintenance_enabled(),
"hour": await get_maintenance_hour(),
"last_run": await get_last_run(),
})
@admin_bp.route("/db-maintenance/health", methods=["GET"])
@admin_required
async def get_db_maintenance_health():
"""Read-only per-table bloat/health stats from Postgres + total DB size."""
from scribe.services.db_maintenance import get_table_health
return jsonify(await get_table_health())
@admin_bp.route("/db-maintenance", methods=["PUT"])
@admin_required
async def update_db_maintenance():
"""Set whether scheduled maintenance runs and at what UTC hour."""
from scribe.services.db_maintenance_scheduler import reschedule_db_maintenance
data = await request.get_json() or {}
enabled = bool(data.get("enabled", True))
try:
hour = int(data.get("hour", 4))
except (TypeError, ValueError):
return jsonify({"error": "hour must be an integer 023"}), 400
if not 0 <= hour <= 23:
return jsonify({"error": "hour must be between 0 and 23"}), 400
await set_admin_setting("db_maintenance_enabled", "true" if enabled else "false")
await set_admin_setting("db_maintenance_hour", str(hour))
reschedule_db_maintenance(hour)
uid = get_current_user_id()
await log_audit(
"db_maintenance_config", user_id=uid, username=g.user.username,
ip_address=request.remote_addr, details={"enabled": enabled, "hour": hour},
)
return jsonify({"status": "ok", "enabled": enabled, "hour": hour})
@admin_bp.route("/db-maintenance/run", methods=["POST"])
@admin_required
async def run_db_maintenance_now():
"""Run a VACUUM (ANALYZE) sweep immediately and return its summary."""
from scribe.services.db_maintenance import run_maintenance
uid = get_current_user_id()
await log_audit(
"db_maintenance_run", user_id=uid, username=g.user.username,
ip_address=request.remote_addr,
)
summary = await run_maintenance()
return jsonify(summary)
@admin_bp.route("/invitations", methods=["POST"])
@admin_required
async def create_invite():
+175
View File
@@ -0,0 +1,175 @@
"""Basic Postgres maintenance — targeted VACUUM (ANALYZE).
Postgres autovacuum already reclaims dead tuples and refreshes planner stats
in the background. This adds a daily, off-hours top-up over the handful of
tables the retention/purge sweeps churn hardest (log retention, notification
purge, auth-token purge, trash purge, version pruning) — so the bloat those
bulk DELETEs leave behind is collected promptly and the planner keeps good
stats. It is deliberately narrow; the rest of the schema is left to autovacuum.
Security: table names cannot be parameterized in SQL, so VACUUM is only ever
issued against names from the hardcoded MAINTENANCE_TABLES allowlist — there is
no path from user input to the table name.
"""
from __future__ import annotations
import json
import logging
import time
from datetime import datetime, timezone
from sqlalchemy import text
from scribe.models import async_session, engine
from scribe.services.settings import get_admin_setting, set_admin_setting
logger = logging.getLogger(__name__)
# High-churn tables — exactly the ones the scheduled delete sweeps hit:
# app_logs ← log retention (hourly)
# notifications ← read-notification purge (hourly)
# password_reset_tokens ← auth-token purge (daily)
# invitation_tokens ← auth-token purge (daily)
# notes ← trash purge (daily) + general churn
# note_versions ← version-pinning prune (daily)
MAINTENANCE_TABLES: tuple[str, ...] = (
"app_logs",
"notifications",
"password_reset_tokens",
"invitation_tokens",
"notes",
"note_versions",
)
_LAST_RUN_KEY = "db_maintenance_last_run"
async def run_maintenance(tables: tuple[str, ...] | list[str] | None = None) -> dict:
"""Run VACUUM (ANALYZE) over the allowlisted high-churn tables.
VACUUM cannot run inside a transaction block, so this uses a dedicated
AUTOCOMMIT connection rather than the usual session. Each table is
vacuumed independently; one failure is logged and does not abort the rest.
Returns a summary dict (also persisted as the db_maintenance_last_run
admin setting) of shape:
{"started_at": iso, "elapsed_ms": int, "tables": [
{"table": str, "ok": bool, "elapsed_ms": int, "error": str|None}
]}
"""
# Only ever operate on the closed allowlist — never an arbitrary name.
requested = tuple(tables) if tables is not None else MAINTENANCE_TABLES
targets = [t for t in requested if t in MAINTENANCE_TABLES]
started = time.monotonic()
started_at = datetime.now(timezone.utc).isoformat()
results: list[dict] = []
async with engine.connect() as conn:
# On AsyncConnection, execution_options() is a coroutine and MUST be
# awaited (it returns the connection with AUTOCOMMIT set — VACUUM can't
# run inside a transaction block).
autocommit = await conn.execution_options(isolation_level="AUTOCOMMIT")
for table in targets:
t0 = time.monotonic()
try:
# table is from the allowlist above — safe to interpolate.
await autocommit.exec_driver_sql(f"VACUUM (ANALYZE) {table}")
results.append({
"table": table,
"ok": True,
"elapsed_ms": int((time.monotonic() - t0) * 1000),
"error": None,
})
except Exception as exc: # noqa: BLE001 — record per-table, keep going
logger.exception("VACUUM (ANALYZE) failed for %s", table)
results.append({
"table": table,
"ok": False,
"elapsed_ms": int((time.monotonic() - t0) * 1000),
"error": str(exc),
})
summary = {
"started_at": started_at,
"elapsed_ms": int((time.monotonic() - started) * 1000),
"tables": results,
}
ok = sum(1 for r in results if r["ok"])
logger.info(
"DB maintenance: vacuumed %d/%d table(s) in %dms",
ok, len(results), summary["elapsed_ms"],
)
try:
await set_admin_setting(_LAST_RUN_KEY, json.dumps(summary))
except Exception: # noqa: BLE001 — persisting the summary is best-effort
logger.warning("Failed to persist db_maintenance last-run summary", exc_info=True)
return summary
# Read-only table-health query against Postgres' own statistics views. The key
# signal is the dead-tuple ratio (dead / (live + dead)) — that IS bloat; a high
# ratio on a large table means autovacuum isn't keeping up. Timestamps confirm
# maintenance is actually running on each table.
_HEALTH_SQL = text("""
SELECT
relname AS table_name,
n_live_tup AS live,
n_dead_tup AS dead,
CASE WHEN n_live_tup + n_dead_tup > 0
THEN round(100.0 * n_dead_tup / (n_live_tup + n_dead_tup), 1)
ELSE 0 END AS dead_pct,
pg_total_relation_size(relid) AS total_bytes,
n_mod_since_analyze AS mod_since_analyze,
GREATEST(last_vacuum, last_autovacuum) AS last_vacuum,
GREATEST(last_analyze, last_autoanalyze) AS last_analyze
FROM pg_stat_user_tables
ORDER BY n_dead_tup DESC, total_bytes DESC
""")
def _iso(value) -> str | None:
return value.isoformat() if value is not None else None
async def get_table_health() -> dict:
"""Per-table health from Postgres statistics + the total database size.
Read-only (system catalogs only). Returns:
{"db_bytes": int, "tables": [
{"table": str, "live": int, "dead": int, "dead_pct": float,
"total_bytes": int, "mod_since_analyze": int,
"last_vacuum": iso|None, "last_analyze": iso|None}
]}
"""
async with async_session() as session:
db_bytes = (
await session.execute(text("SELECT pg_database_size(current_database())"))
).scalar() or 0
rows = (await session.execute(_HEALTH_SQL)).mappings().all()
tables = [
{
"table": r["table_name"],
"live": int(r["live"] or 0),
"dead": int(r["dead"] or 0),
"dead_pct": float(r["dead_pct"] or 0),
"total_bytes": int(r["total_bytes"] or 0),
"mod_since_analyze": int(r["mod_since_analyze"] or 0),
"last_vacuum": _iso(r["last_vacuum"]),
"last_analyze": _iso(r["last_analyze"]),
}
for r in rows
]
return {"db_bytes": int(db_bytes), "tables": tables}
async def get_last_run() -> dict | None:
"""Return the most recent run summary, or None if it has never run."""
raw = await get_admin_setting(_LAST_RUN_KEY, "")
if not raw:
return None
try:
return json.loads(raw)
except (ValueError, TypeError):
return None
@@ -0,0 +1,107 @@
"""Daily APScheduler cron for basic DB maintenance (targeted VACUUM ANALYZE).
Mirrors trash_scheduler.py: a single global BackgroundScheduler job bridges
into the asyncio loop to run the async maintenance. Scheduled for 04:00 UTC by
default — after the 03:30 trash purge — so it collects the dead tuples that
night's delete sweeps leave behind.
Two things are operator-tunable from the admin Settings card:
- db_maintenance_enabled ("true"/"false") — checked at fire time, so toggling
it needs no reschedule.
- db_maintenance_hour ("0".."23", UTC) — the cron hour. Changing it calls
reschedule_db_maintenance() so the live job moves without a restart.
"""
from __future__ import annotations
import asyncio
import logging
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from scribe.services.settings import get_admin_setting
logger = logging.getLogger(__name__)
_JOB_ID = "db_maintenance_vacuum"
_DEFAULT_HOUR = 4
_scheduler: BackgroundScheduler | None = None
_loop: asyncio.AbstractEventLoop | None = None
async def get_maintenance_hour() -> int:
"""The configured run-hour (UTC, 023), clamped; default 04:00."""
raw = await get_admin_setting("db_maintenance_hour", str(_DEFAULT_HOUR))
try:
hour = int(raw)
except (TypeError, ValueError):
return _DEFAULT_HOUR
return hour if 0 <= hour <= 23 else _DEFAULT_HOUR
async def is_maintenance_enabled() -> bool:
"""Whether the scheduled run is enabled (default on)."""
return (await get_admin_setting("db_maintenance_enabled", "true")) != "false"
def _run_maintenance_threadsafe() -> None:
"""APScheduler invokes this from a worker thread; bridge into the loop."""
if _loop is None:
logger.warning("db maintenance scheduler: no loop registered")
return
async def _runner():
try:
if not await is_maintenance_enabled():
logger.debug("db maintenance: disabled, skipping scheduled run")
return
from scribe.services.db_maintenance import run_maintenance
await run_maintenance()
except Exception:
logger.exception("db maintenance run failed")
asyncio.run_coroutine_threadsafe(_runner(), _loop)
def start_db_maintenance_scheduler(
loop: asyncio.AbstractEventLoop, hour: int = _DEFAULT_HOUR
) -> None:
"""Start the daily job. `hour` is the configured UTC run-hour, resolved by
the caller (which has an async context) via get_maintenance_hour() — passed
in rather than read here so we never block the event loop at startup. The
job's enabled-gate is re-checked at every fire, so only the hour is needed
up front."""
global _scheduler, _loop
if _scheduler is not None:
return
_loop = loop
hour = hour if 0 <= hour <= 23 else _DEFAULT_HOUR
_scheduler = BackgroundScheduler()
_scheduler.add_job(
_run_maintenance_threadsafe,
trigger=CronTrigger(hour=hour, minute=0, timezone="UTC"),
id=_JOB_ID,
replace_existing=True,
)
_scheduler.start()
logger.info("DB maintenance scheduler started (daily %02d:00 UTC)", hour)
def reschedule_db_maintenance(hour: int) -> None:
"""Move the live job to a new UTC hour (called when the admin changes it)."""
if _scheduler is None:
return
hour = hour if 0 <= hour <= 23 else _DEFAULT_HOUR
_scheduler.reschedule_job(
_JOB_ID, trigger=CronTrigger(hour=hour, minute=0, timezone="UTC")
)
logger.info("DB maintenance scheduler rescheduled to %02d:00 UTC", hour)
def stop_db_maintenance_scheduler() -> None:
global _scheduler
if _scheduler is not None:
_scheduler.shutdown(wait=False)
_scheduler = None
logger.info("DB maintenance scheduler stopped")
+17
View File
@@ -26,6 +26,23 @@ async def get_admin_setting(key: str, default: str = "") -> str:
return setting.value if setting and setting.value else default
async def set_admin_setting(key: str, value: str) -> None:
"""Write an instance-global setting onto the first admin account.
The write-side counterpart to get_admin_setting, for non-per-user settings
written outside a request context (e.g. a scheduler persisting its last-run
summary) where get_current_user_id() isn't available.
"""
async with async_session() as session:
admin_id = (
await session.execute(
select(User.id).where(User.role == "admin").order_by(User.id)
)
).scalars().first()
if admin_id is not None:
await set_setting(admin_id, key, value)
async def get_setting(user_id: int, key: str, default: str = "") -> str:
async with async_session() as session:
result = await session.execute(
+8 -2
View File
@@ -12,8 +12,14 @@ import pytest
@pytest.fixture(autouse=True)
def _isolate_env(monkeypatch):
"""Prevent tests from accidentally reading production env vars."""
def _isolate_env(request, monkeypatch):
"""Prevent unit tests from accidentally reading production env vars.
Integration tests (marked `integration`) are skipped here: they must use the
real DATABASE_URL injected by the CI integration lane, not the fake one.
"""
if request.node.get_closest_marker("integration"):
return
monkeypatch.setenv("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
monkeypatch.setenv("SECRET_KEY", "test-secret-key")
monkeypatch.setenv("OLLAMA_URL", "http://localhost:11434")
+56
View File
@@ -0,0 +1,56 @@
"""Real-Postgres integration tests for DB maintenance + health.
These run only in the CI integration lane (a real Postgres service + schema
built by `alembic upgrade head`). They exercise the actual async SQLAlchemy
connection path that unit mocks cannot: the un-awaited
`AsyncConnection.execution_options` regression (which made every VACUUM raise
AttributeError, reporting 0/6) passes the unit suite but fails here.
"""
import pytest
import pytest_asyncio
from scribe.models import engine
from scribe.services.db_maintenance import (
MAINTENANCE_TABLES,
get_table_health,
run_maintenance,
)
pytestmark = pytest.mark.integration
@pytest_asyncio.fixture(autouse=True)
async def _dispose_engine():
"""Dispose the app's module-level engine after each test.
The engine pools asyncpg connections per event loop, but pytest-asyncio runs
each test on a fresh loop — so without this, test 2 gets handed test 1's
connection bound to a now-dead loop ("Future attached to a different loop").
Disposing in the test's own loop teardown clears the pool cleanly.
"""
yield
await engine.dispose()
@pytest.mark.asyncio
async def test_run_maintenance_vacuums_real_tables():
summary = await run_maintenance()
assert summary["tables"], "no tables were vacuumed"
# Every allowlisted table exists after `alembic upgrade head`, so every
# VACUUM (ANALYZE) must succeed. With the un-awaited execution_options bug
# they would ALL fail with AttributeError — this is the guard.
failed = [t for t in summary["tables"] if not t["ok"]]
assert not failed, f"VACUUM failed for: {failed}"
assert {t["table"] for t in summary["tables"]} <= set(MAINTENANCE_TABLES)
@pytest.mark.asyncio
async def test_table_health_reports_real_stats():
health = await get_table_health()
assert health["db_bytes"] > 0
names = {t["table"] for t in health["tables"]}
# Core table built by migrations must show up in pg_stat_user_tables.
assert "notes" in names
for t in health["tables"]:
assert t["dead_pct"] >= 0
assert t["total_bytes"] >= 0
+54
View File
@@ -0,0 +1,54 @@
"""Structural tests for the DB-maintenance admin routes + scheduler surface."""
def test_admin_handlers_callable():
from scribe.routes import admin as admin_routes
for name in (
"get_db_maintenance",
"update_db_maintenance",
"run_db_maintenance_now",
"get_db_maintenance_health",
):
assert callable(getattr(admin_routes, name))
def test_db_maintenance_routes_registered():
from scribe.app import create_app
app = create_app()
rules = {r.rule for r in app.url_map.iter_rules()}
assert "/api/admin/db-maintenance" in rules
assert "/api/admin/db-maintenance/run" in rules
assert "/api/admin/db-maintenance/health" in rules
def test_scheduler_surface():
from scribe.services import db_maintenance_scheduler as sched
for fn in (
"start_db_maintenance_scheduler",
"stop_db_maintenance_scheduler",
"reschedule_db_maintenance",
"get_maintenance_hour",
"is_maintenance_enabled",
):
assert callable(getattr(sched, fn)), f"scheduler missing {fn}"
def test_service_surface_and_allowlist():
from scribe.services.db_maintenance import (
MAINTENANCE_TABLES,
get_last_run,
get_table_health,
run_maintenance,
)
assert callable(run_maintenance) and callable(get_last_run) and callable(get_table_health)
# The allowlist is a closed tuple covering exactly the high-churn tables.
assert isinstance(MAINTENANCE_TABLES, tuple)
assert set(MAINTENANCE_TABLES) == {
"app_logs", "notifications", "password_reset_tokens",
"invitation_tokens", "notes", "note_versions",
}
def test_set_admin_setting_exists():
from scribe.services.settings import set_admin_setting
assert callable(set_admin_setting)
+124
View File
@@ -0,0 +1,124 @@
"""Tests for services/db_maintenance.py — mocks the engine (no real DB).
Verifies that VACUUM (ANALYZE) is issued once per allowlisted table, that the
allowlist is closed (an arbitrary name can't be vacuumed), and that a single
table failure doesn't abort the rest of the sweep.
"""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from scribe.services.db_maintenance import MAINTENANCE_TABLES, run_maintenance
def _mock_engine(exec_side_effect=None):
"""An engine whose connect() yields a conn with a recording exec_driver_sql."""
conn = MagicMock()
autocommit = MagicMock()
autocommit.exec_driver_sql = AsyncMock(side_effect=exec_side_effect)
# On AsyncConnection, execution_options() is a coroutine — mock it as async
# so the test exercises the real (awaited) call shape.
conn.execution_options = AsyncMock(return_value=autocommit)
cm = AsyncMock()
cm.__aenter__ = AsyncMock(return_value=conn)
cm.__aexit__ = AsyncMock(return_value=False)
engine = MagicMock()
engine.connect.return_value = cm
return engine, autocommit
@pytest.mark.asyncio
async def test_vacuums_each_allowlisted_table_once():
engine, autocommit = _mock_engine()
with patch("scribe.services.db_maintenance.engine", engine), \
patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
summary = await run_maintenance()
issued = [c.args[0] for c in autocommit.exec_driver_sql.await_args_list]
assert issued == [f"VACUUM (ANALYZE) {t}" for t in MAINTENANCE_TABLES]
assert all(r["ok"] for r in summary["tables"])
assert len(summary["tables"]) == len(MAINTENANCE_TABLES)
# AUTOCOMMIT was requested (and awaited) — VACUUM can't run in a transaction.
conn = engine.connect.return_value.__aenter__.return_value
conn.execution_options.assert_awaited_once_with(isolation_level="AUTOCOMMIT")
@pytest.mark.asyncio
async def test_allowlist_is_closed():
"""A caller-supplied name not on the allowlist is silently ignored."""
engine, autocommit = _mock_engine()
with patch("scribe.services.db_maintenance.engine", engine), \
patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
summary = await run_maintenance(tables=["app_logs", "users; DROP TABLE notes"])
issued = [c.args[0] for c in autocommit.exec_driver_sql.await_args_list]
assert issued == ["VACUUM (ANALYZE) app_logs"]
assert [r["table"] for r in summary["tables"]] == ["app_logs"]
@pytest.mark.asyncio
async def test_one_table_failure_does_not_abort_the_rest():
# First table raises, the remaining tables still get vacuumed.
calls = {"n": 0}
def flaky(_sql):
calls["n"] += 1
if calls["n"] == 1:
raise RuntimeError("boom")
return MagicMock()
engine, autocommit = _mock_engine(exec_side_effect=flaky)
with patch("scribe.services.db_maintenance.engine", engine), \
patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
summary = await run_maintenance()
assert autocommit.exec_driver_sql.await_count == len(MAINTENANCE_TABLES)
assert summary["tables"][0]["ok"] is False
assert summary["tables"][0]["error"] == "boom"
assert all(r["ok"] for r in summary["tables"][1:])
def _health_session(db_bytes, rows):
s = AsyncMock()
s.__aenter__ = AsyncMock(return_value=s)
s.__aexit__ = AsyncMock(return_value=False)
size_res = MagicMock()
size_res.scalar.return_value = db_bytes
rows_res = MagicMock()
rows_res.mappings.return_value.all.return_value = rows
s.execute = AsyncMock(side_effect=[size_res, rows_res])
return s
@pytest.mark.asyncio
async def test_table_health_shapes_rows_and_db_size():
from datetime import datetime, timezone
from scribe.services.db_maintenance import get_table_health
vac = datetime(2026, 6, 14, 4, 0, tzinfo=timezone.utc)
rows = [
{"table_name": "notes", "live": 1000, "dead": 300, "dead_pct": 23.1,
"total_bytes": 5_000_000, "mod_since_analyze": 50,
"last_vacuum": vac, "last_analyze": None},
]
with patch("scribe.services.db_maintenance.async_session",
return_value=_health_session(42_000_000, rows)):
health = await get_table_health()
assert health["db_bytes"] == 42_000_000
t = health["tables"][0]
assert t["table"] == "notes"
assert t["dead_pct"] == 23.1
assert t["last_vacuum"] == vac.isoformat()
assert t["last_analyze"] is None # null timestamp passes through as None
@pytest.mark.asyncio
async def test_summary_is_persisted_as_admin_setting():
engine, _ = _mock_engine()
setter = AsyncMock()
with patch("scribe.services.db_maintenance.engine", engine), \
patch("scribe.services.db_maintenance.set_admin_setting", setter):
await run_maintenance()
setter.assert_awaited_once()
assert setter.await_args.args[0] == "db_maintenance_last_run"