feat(maintenance): daily janitor for orphaned .part/.partial staging files

Downloads/imports stage into <name>.part / <name>.partial then os.replace() into
place, so a kill mid-write leaves a discardable temp — never a corrupt final.
cleanup_orphaned_temp_files sweeps ones left behind under the images root, only
older than 6h so an in-flight download's staging file is never removed. Daily beat.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-07 22:15:33 -04:00
parent 9eae636047
commit 0b78264d62
3 changed files with 63 additions and 0 deletions
+5
View File
@@ -115,6 +115,11 @@ def make_celery() -> Celery:
"task": "backend.app.tasks.maintenance.cleanup_old_tasks",
"schedule": 86400.0, # daily
},
"cleanup-orphaned-temp-files": {
"task": "backend.app.tasks.maintenance.cleanup_orphaned_temp_files",
"schedule": 86400.0, # daily — sweep .part/.partial left by a
# download/import killed mid-write (graceful-shutdown fallout)
},
"train-heads-nightly": {
"task": "backend.app.tasks.ml.scheduled_train_heads",
"schedule": 86400.0, # passive cadence; manual retrain stays available
+36
View File
@@ -79,6 +79,14 @@ VERIFY_PAGE = 200
FFPROBE_TIMEOUT_SECONDS = 10
TASK_RUN_KEEP_OK_SECONDS = 24 * 3600 # 24 h
TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days
# Orphaned staging files: downloads/imports stage into <name>.part / <name>.partial
# then os.replace() into place (importer / external_fetch / native_ingest_common /
# attachment_store), so a kill mid-write leaves a discardable temp, never a corrupt
# final. cleanup_orphaned_temp_files sweeps ones left behind; the min-age guard
# keeps it from deleting an in-flight download's staging file mid-write.
IMAGES_ROOT = Path("/images")
TEMP_STAGING_SUFFIXES = (".part", ".partial")
ORPHAN_TEMP_MIN_AGE_HOURS = 6
# Audit 2026-06-02: per-entity recovery sweep thresholds. Each must be
# > the entity's longest legitimate runtime (its task's time_limit + a
@@ -339,6 +347,34 @@ def cleanup_old_tasks() -> int:
return result.rowcount or 0
@celery.task(name="backend.app.tasks.maintenance.cleanup_orphaned_temp_files")
def cleanup_orphaned_temp_files() -> int:
"""Delete orphaned .part/.partial staging files under the images root, left by
a download/import killed mid-write. Only removes files older than
ORPHAN_TEMP_MIN_AGE_HOURS so an in-flight download's staging file is never
pulled out from under it. Returns the count removed."""
if not IMAGES_ROOT.is_dir():
return 0
cutoff = datetime.now(UTC).timestamp() - ORPHAN_TEMP_MIN_AGE_HOURS * 3600
removed = 0
for path in IMAGES_ROOT.rglob("*"):
if path.suffix not in TEMP_STAGING_SUFFIXES or not path.is_file():
continue
try:
if path.stat().st_mtime >= cutoff:
continue # too fresh — may be an active download
path.unlink()
removed += 1
except OSError as exc:
log.warning("cleanup_orphaned_temp_files: %s: %s", path, exc)
if removed:
log.info(
"cleanup_orphaned_temp_files: removed %d orphaned staging file(s)",
removed,
)
return removed
@celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs")
def recover_stalled_task_runs() -> int:
"""Flip task_run rows stuck in 'running' past their queue-specific
+22
View File
@@ -26,6 +26,28 @@ def _make_batch(session) -> int:
return batch.id
def test_cleanup_orphaned_temp_files_removes_stale_only(tmp_path, monkeypatch):
import os
from backend.app.tasks import maintenance as m
monkeypatch.setattr(m, "IMAGES_ROOT", tmp_path)
stale = tmp_path / "artist" / "img.jpg.part" # killed download → orphan
stale.parent.mkdir(parents=True)
stale.write_bytes(b"x")
old = datetime.now(UTC).timestamp() - 8 * 3600 # older than the 6h guard
os.utime(stale, (old, old))
fresh = tmp_path / "in_progress.jpg.partial" # active download → keep
fresh.write_bytes(b"x")
keep = tmp_path / "real.jpg" # a real image → keep
keep.write_bytes(b"x")
assert m.cleanup_orphaned_temp_files() == 1
assert not stale.exists()
assert fresh.exists()
assert keep.exists()
def test_recover_interrupted_only_old(db_sync, monkeypatch):
batch_id = _make_batch(db_sync)
now = datetime.now(UTC)