FC-6 series authoring + backup/NFS hardening + UX fixes #83

Merged
bvandeusen merged 10 commits from dev into main 2026-06-07 19:31:05 -04:00
5 changed files with 188 additions and 27 deletions
Showing only changes of commit daaa7543a8 - Show all commits
+56 -11
View File
@@ -15,7 +15,10 @@ lifecycle + soft/hard time limits + retention bookkeeping.
from __future__ import annotations from __future__ import annotations
import json import json
import os
import shutil
import subprocess import subprocess
import tempfile
from datetime import UTC, datetime from datetime import UTC, datetime
from pathlib import Path from pathlib import Path
@@ -26,6 +29,35 @@ _BACKUPS_DIRNAME = "_backups"
# blocking syscall ignores that signal. These bound the worst case. # blocking syscall ignores that signal. These bound the worst case.
_DB_SUBPROCESS_TIMEOUT_S = 12 * 60 # 12 min (Celery soft is 10 min) _DB_SUBPROCESS_TIMEOUT_S = 12 * 60 # 12 min (Celery soft is 10 min)
_IMAGES_SUBPROCESS_TIMEOUT_S = 7 * 60 * 60 # 7 hr (Celery soft is 6 hr) _IMAGES_SUBPROCESS_TIMEOUT_S = 7 * 60 * 60 # 7 hr (Celery soft is 6 hr)
# Grace after SIGKILL to reap the child. If it can't be reaped in this window
# (an uninterruptible NFS D-state — the failure mode that wedged the
# concurrency-1 maintenance lane for hours, operator-flagged 2026-06-07), we
# STOP waiting and fail fast, freeing the worker slot. The orphan is reaped by
# the OS once its blocking syscall clears.
_KILL_REAP_GRACE_S = 10
def _run_bounded(cmd: list[str], timeout: int) -> None:
"""subprocess.run(check=True, timeout) whose reaper can't itself hang.
subprocess.run's timeout path SIGKILLs the child then blocks in wait() to
reap it — but a process stuck in uninterruptible I/O (NFS) can't be reaped,
so wait() blocks for hours. Here we bound the post-kill reap and re-raise
TimeoutExpired regardless, so the caller fails fast instead of wedging."""
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
try:
out, err = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
proc.kill()
try:
proc.communicate(timeout=_KILL_REAP_GRACE_S)
except subprocess.TimeoutExpired:
pass # unkillable (D-state) — abandon the reap, fail fast
raise
if proc.returncode != 0:
raise subprocess.CalledProcessError(
proc.returncode, cmd, output=out, stderr=err
)
def _libpq_url(sa_url: str) -> str: def _libpq_url(sa_url: str) -> str:
@@ -84,14 +116,25 @@ def backup_db(
ts = _now_ts() ts = _now_ts()
out_dir = _backups_dir(images_root) out_dir = _backups_dir(images_root)
sql_path = out_dir / f"fc_db_{ts}.sql" sql_path = out_dir / f"fc_db_{ts}.sql"
subprocess.run( # Dump to LOCAL disk first, then move the finished file to the (NFS) backups
[ # dir. pg_dump's long phase is then a DB-socket wait + local writes — both
"pg_dump", "--no-owner", "--no-acl", # killable — instead of an NFS write that can hang uninterruptibly. Only the
"-f", str(sql_path), _libpq_url(db_url), # final move touches NFS, and it's a bounded single-file step.
], fd, tmp_name = tempfile.mkstemp(prefix="fc_db_", suffix=".sql")
capture_output=True, check=True, os.close(fd)
timeout=_DB_SUBPROCESS_TIMEOUT_S, tmp_path = Path(tmp_name)
) try:
_run_bounded(
[
"pg_dump", "--no-owner", "--no-acl",
"-f", str(tmp_path), _libpq_url(db_url),
],
_DB_SUBPROCESS_TIMEOUT_S,
)
shutil.move(str(tmp_path), str(sql_path))
finally:
if tmp_path.exists():
tmp_path.unlink(missing_ok=True)
manifest_path = _write_manifest( manifest_path = _write_manifest(
out_dir, kind="db", ts=ts, tag=tag, triggered_by=triggered_by, out_dir, kind="db", ts=ts, tag=tag, triggered_by=triggered_by,
artifact_path=sql_path, artifact_path=sql_path,
@@ -114,15 +157,17 @@ def backup_images(
ts = _now_ts() ts = _now_ts()
out_dir = _backups_dir(images_root) out_dir = _backups_dir(images_root)
tar_path = out_dir / f"fc_images_{ts}.tar.zst" tar_path = out_dir / f"fc_images_{ts}.tar.zst"
subprocess.run( # No local-temp here (the archive is hundreds of GB — it can't stage in
# /tmp), but bounded-kill still applies so a tar wedged on NFS fails fast
# rather than holding the lane for hours.
_run_bounded(
[ [
"tar", "--zstd", "-cf", str(tar_path), "tar", "--zstd", "-cf", str(tar_path),
"-C", str(images_root.parent), images_root.name, "-C", str(images_root.parent), images_root.name,
f"--exclude={images_root.name}/_backups", f"--exclude={images_root.name}/_backups",
f"--exclude={images_root.name}/_quarantine", f"--exclude={images_root.name}/_quarantine",
], ],
capture_output=True, check=True, _IMAGES_SUBPROCESS_TIMEOUT_S,
timeout=_IMAGES_SUBPROCESS_TIMEOUT_S,
) )
manifest_path = _write_manifest( manifest_path = _write_manifest(
out_dir, kind="images", ts=ts, tag=tag, triggered_by=triggered_by, out_dir, kind="images", ts=ts, tag=tag, triggered_by=triggered_by,
+16 -7
View File
@@ -83,8 +83,13 @@ TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days
# small buffer) so the sweep never flags in-flight work. # small buffer) so the sweep never flags in-flight work.
# #
# Backups: images backup has time_limit=23400s (6.5h). 7h covers it # Backups: images backup has time_limit=23400s (6.5h). 7h covers it
# with a 30-min buffer; db backup at 12 min hard limit fits trivially. # with a 30-min buffer.
BACKUP_STALL_THRESHOLD_MINUTES = 7 * 60 BACKUP_STALL_THRESHOLD_MINUTES = 7 * 60
# DB backup/restore is seconds-to-minutes (35-min hard limit). It must NOT share
# the images' 7h window — a DB backup wedged on NFS would otherwise sit "running"
# for 7 hours holding the concurrency-1 maintenance_long lane (operator-flagged
# 2026-06-07). 40 min gives a small buffer over the hard limit.
BACKUP_DB_STALL_THRESHOLD_MINUTES = 40
# Library audit: scan_library_for_rule has time_limit=7500s (2h5m). # Library audit: scan_library_for_rule has time_limit=7500s (2h5m).
# 2h15m gives a 10-min buffer. # 2h15m gives a 10-min buffer.
LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES = 135 LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES = 135
@@ -619,16 +624,20 @@ def recover_stalled_backup_runs() -> int:
""" """
SessionLocal = _sync_session_factory() SessionLocal = _sync_session_factory()
now = datetime.now(UTC) now = datetime.now(UTC)
cutoff = now - timedelta(minutes=BACKUP_STALL_THRESHOLD_MINUTES) db_cutoff = now - timedelta(minutes=BACKUP_DB_STALL_THRESHOLD_MINUTES)
msg = ( slow_cutoff = now - timedelta(minutes=BACKUP_STALL_THRESHOLD_MINUTES)
f"stranded by recovery sweep (no terminal status after " msg = "stranded by recovery sweep (no terminal status within the stall window)"
f"{BACKUP_STALL_THRESHOLD_MINUTES // 60}h)"
)
with SessionLocal() as session: with SessionLocal() as session:
result = session.execute( result = session.execute(
update(BackupRun) update(BackupRun)
.where(BackupRun.status.in_(["running", "restoring"])) .where(BackupRun.status.in_(["running", "restoring"]))
.where(BackupRun.started_at < cutoff) # db backups/restores are fast (40-min window); images run hours (7h).
.where(
or_(
and_(BackupRun.kind == "db", BackupRun.started_at < db_cutoff),
and_(BackupRun.kind != "db", BackupRun.started_at < slow_cutoff),
)
)
.values(status="error", finished_at=now, error=msg) .values(status="error", finished_at=now, error=msg)
) )
session.commit() session.commit()
@@ -181,7 +181,7 @@
<v-btn <v-btn
color="error" variant="flat" rounded="pill" color="error" variant="flat" rounded="pill"
prepend-icon="mdi-format-letter-case" prepend-icon="mdi-format-letter-case"
:disabled="!normPreview.total_changes" :disabled="!normPreview.total_changes || normResult === 'queued'"
:loading="normCommitting" :loading="normCommitting"
@click="onNormCommit" @click="onNormCommit"
>Standardize {{ normPreview.total_changes }} tag group(s)</v-btn> >Standardize {{ normPreview.total_changes }} tag group(s)</v-btn>
@@ -291,7 +291,11 @@ async function onNormCommit() {
// confirm it's queued; the operator can re-run Preview later to verify. // confirm it's queued; the operator can re-run Preview later to verify.
await store.normalizeTags({ dryRun: false }) await store.normalizeTags({ dryRun: false })
normResult.value = 'queued' normResult.value = 'queued'
normPreview.value = { total_changes: 0, tags_to_rename: 0, collisions: 0, tags_to_merge: 0, sample: [] } // Keep the preview showing what was QUEUED — do NOT zero it out. Overwriting
// it with a zeroed object made the card read "0 tag groups to change" the
// instant Standardize was clicked, which looked like nothing happened
// (operator-flagged 2026-06-07). The button disables on `queued`; re-running
// Preview later reflects the real remaining count as the task applies.
} finally { } finally {
normCommitting.value = false normCommitting.value = false
} }
+60 -5
View File
@@ -5,6 +5,7 @@ without external binaries. The real subprocess behavior is exercised
implicitly via the Celery task tests in test_tasks_backup.py. implicitly via the Celery task tests in test_tasks_backup.py.
""" """
import json import json
import subprocess
from pathlib import Path from pathlib import Path
import pytest import pytest
@@ -16,9 +17,9 @@ pytestmark = pytest.mark.integration
@pytest.fixture @pytest.fixture
def fake_subprocess(monkeypatch): def fake_subprocess(monkeypatch):
"""Replace subprocess.run with a fake that writes a sentinel to """Fake both subprocess.run (restore path) AND subprocess.Popen (the
the target path (for pg_dump's -f, for tar's -cf). Captures all bounded-kill backup path) so tests run without external binaries. Each
calls in a list.""" writes the target sentinel and records the cmd in a shared list."""
calls = [] calls = []
class _FakeProc: class _FakeProc:
@@ -26,17 +27,33 @@ def fake_subprocess(monkeypatch):
stdout = b"" stdout = b""
stderr = b"" stderr = b""
def _fake_run(cmd, **kwargs): def _write_sentinel(cmd):
calls.append(list(cmd))
if cmd[0] == "pg_dump": if cmd[0] == "pg_dump":
i = cmd.index("-f") i = cmd.index("-f")
Path(cmd[i + 1]).write_bytes(b"-- fake pg_dump\n") Path(cmd[i + 1]).write_bytes(b"-- fake pg_dump\n")
elif cmd[0] == "tar" and "-cf" in cmd: elif cmd[0] == "tar" and "-cf" in cmd:
i = cmd.index("-cf") i = cmd.index("-cf")
Path(cmd[i + 1]).write_bytes(b"fake tar payload") Path(cmd[i + 1]).write_bytes(b"fake tar payload")
def _fake_run(cmd, **kwargs):
calls.append(list(cmd))
_write_sentinel(cmd)
return _FakeProc() return _FakeProc()
class _FakePopen:
def __init__(self, cmd, **kwargs):
calls.append(list(cmd))
self.returncode = 0
_write_sentinel(cmd)
def communicate(self, timeout=None):
return (b"", b"")
def kill(self):
self.returncode = -9
monkeypatch.setattr("subprocess.run", _fake_run) monkeypatch.setattr("subprocess.run", _fake_run)
monkeypatch.setattr("subprocess.Popen", _FakePopen)
return calls return calls
@@ -198,3 +215,41 @@ def test_backups_dir_created_on_first_use(tmp_path):
d = backup_service._backups_dir(tmp_path) d = backup_service._backups_dir(tmp_path)
assert d.is_dir() assert d.is_dir()
assert d.name == "_backups" assert d.name == "_backups"
# --- bounded-kill + local-temp (FC #739) -----------------------------
def test_backup_db_dumps_to_local_temp_not_nfs_backups_dir(tmp_path, fake_subprocess):
"""pg_dump must target a LOCAL temp path, not the (NFS) _backups dir —
so its long phase can't hang uninterruptibly on an NFS write."""
backup_service.backup_db(db_url="postgresql://u@h/d", images_root=tmp_path)
cmd = fake_subprocess[0]
dump_target = cmd[cmd.index("-f") + 1]
assert "_backups" not in dump_target
# The finished file still ends up in _backups (moved there).
result = backup_service.backup_db(
db_url="postgresql://u@h/d", images_root=tmp_path,
)
assert "_backups" in result["sql_path"]
def test_run_bounded_fails_fast_when_unkillable(monkeypatch):
"""A child stuck in D-state (communicate keeps timing out even after kill)
must NOT block the reaper — _run_bounded kills then re-raises promptly."""
killed = {"n": 0}
class _Hang:
def __init__(self, cmd, **kwargs):
pass
def communicate(self, timeout=None):
raise subprocess.TimeoutExpired(cmd="x", timeout=timeout or 0)
def kill(self):
killed["n"] += 1
monkeypatch.setattr("subprocess.Popen", _Hang)
with pytest.raises(subprocess.TimeoutExpired):
backup_service._run_bounded(["pg_dump"], 1)
assert killed["n"] == 1
+50 -2
View File
@@ -34,16 +34,32 @@ def fake_subprocess_and_images_root(monkeypatch, tmp_path):
stdout = b"" stdout = b""
stderr = b"" stderr = b""
def _fake_run(cmd, **kwargs): def _sentinel(cmd):
if cmd[0] == "pg_dump": if cmd[0] == "pg_dump":
i = cmd.index("-f") i = cmd.index("-f")
Path(cmd[i + 1]).write_bytes(b"-- fake pg_dump\n") Path(cmd[i + 1]).write_bytes(b"-- fake pg_dump\n")
elif cmd[0] == "tar" and "-cf" in cmd: elif cmd[0] == "tar" and "-cf" in cmd:
i = cmd.index("-cf") i = cmd.index("-cf")
Path(cmd[i + 1]).write_bytes(b"fake tar payload") Path(cmd[i + 1]).write_bytes(b"fake tar payload")
def _fake_run(cmd, **kwargs):
_sentinel(cmd)
return _FakeProc() return _FakeProc()
class _FakePopen:
def __init__(self, cmd, **kwargs):
self.returncode = 0
_sentinel(cmd)
def communicate(self, timeout=None):
return (b"", b"")
def kill(self):
self.returncode = -9
# backup_db/backup_images go through _run_bounded (Popen); restore via run.
monkeypatch.setattr("subprocess.run", _fake_run) monkeypatch.setattr("subprocess.run", _fake_run)
monkeypatch.setattr("subprocess.Popen", _FakePopen)
def _seed_backup(db_sync, *, kind, status, started_at, tag=None, def _seed_backup(db_sync, *, kind, status, started_at, tag=None,
@@ -89,7 +105,8 @@ async def test_backup_db_task_records_failure_on_subprocess_error(db_sync, monke
def _boom(*a, **kw): def _boom(*a, **kw):
raise RuntimeError("synthetic pg_dump fail") raise RuntimeError("synthetic pg_dump fail")
monkeypatch.setattr("subprocess.run", _boom) # backup_db dumps via _run_bounded → subprocess.Popen.
monkeypatch.setattr("subprocess.Popen", _boom)
with pytest.raises(RuntimeError): with pytest.raises(RuntimeError):
backup_db_task.delay().get() backup_db_task.delay().get()
@@ -237,6 +254,37 @@ def test_prune_backups_never_deletes_running_or_restoring(db_sync):
assert set(statuses) == {"running", "restoring"} assert set(statuses) == {"running", "restoring"}
# --- recover_stalled_backup_runs (per-kind threshold, FC #739) -------
def test_stall_sweep_flips_db_fast_but_spares_running_images(db_sync):
from backend.app.tasks.maintenance import recover_stalled_backup_runs
now = datetime.now(UTC)
# A db backup stuck 50 min → past the 40-min db window → flipped to error.
db_id = _seed_backup(
db_sync, kind="db", status="running",
started_at=now - timedelta(minutes=50),
)
# An images backup running 50 min is still well under the 7h window → spared.
img_id = _seed_backup(
db_sync, kind="images", status="running",
started_at=now - timedelta(minutes=50),
)
db_sync.commit()
recover_stalled_backup_runs.apply().get()
db_status = db_sync.execute(
select(BackupRun.status).where(BackupRun.id == db_id)
).scalar_one()
img_status = db_sync.execute(
select(BackupRun.status).where(BackupRun.id == img_id)
).scalar_one()
assert db_status == "error"
assert img_status == "running"
# --- backup_db_nightly ---------------------------------------------- # --- backup_db_nightly ----------------------------------------------