From 7a49e2d507187e0341abc90950feed457a62d9a7 Mon Sep 17 00:00:00 2001 From: andrew Date: Wed, 15 Jul 2026 10:13:02 -0600 Subject: [PATCH] 0.6.4: Truthful maintenance digest + scheduled jobs queue for the lock instead of skipping The Telegram digest reported healthy weekly jobs as "never run yet": its maintenance section still globbed for cron-era per-script log filenames (clean-index-*.log etc.) that jobs run through pipeline_runner no longer write. It now reads job_runs, the scheduler's own record, so it reports the last success (age + summary snippet from that run's log), flags a newer failed or lock-skipped attempt on the same line, and distinguishes "never succeeded (last attempt: skipped, lock busy)" from genuinely never scheduled. Also matched the clear-bad-genres snippet grep to the script's current summary wording, and dropped the now-unused newest_log helper. The DB also showed WHY two jobs had never run: on Sunday the 08:30 strip-mb-tags run took 10m19s while holding the shared pipeline lock, so strip-watermark-art (08:35) and scrub-watermark-text (08:40) hit flock-style instant skip and lost their only slot of the week -- the 08:40 job missed by 19 seconds. Scheduled runs now wait up to 30 minutes for the lock (SCHEDULED_LOCK_WAIT_SECONDS) and only then record skipped_lock, so fixed-time blocks queue instead of starving; manual "Run now" keeps the instant skip since a person expects an immediate answer. Tests: scheduled-run queueing, lock-wait timeout, and manual instant-skip. Co-Authored-By: Claude Fable 5 --- README.md | 6 +- app/services/pipeline_runner.py | 41 +++++++-- pipeline/lib/pipeline-status.sh | 150 +++++++++++++++++++++----------- tests/test_pipeline_runner.py | 34 +++++++- 4 files changed, 167 insertions(+), 64 deletions(-) diff --git a/README.md b/README.md index 61a5d69..9166e8e 100644 --- a/README.md +++ b/README.md @@ -73,10 +73,10 @@ Optional, add these later if you want them: Pull the prebuilt image onto your Docker host: ```bash -docker pull git.kretzer.club/andrew/alembic:0.6.3 +docker pull git.kretzer.club/andrew/alembic:0.6.4 ``` -That is the whole install. You do not need to download the source or build anything. The `0.6.3` is the version; you can pin to it so nothing changes under you, or use `latest` to always get the newest. +That is the whole install. You do not need to download the source or build anything. The `0.6.4` is the version; you can pin to it so nothing changes under you, or use `latest` to always get the newest. (If you would rather build it yourself from source, you can, but you do not need to.) @@ -136,7 +136,7 @@ Create a file called `docker-compose.yml` on your server (put it wherever you ke ```yaml services: alembic: - image: git.kretzer.club/andrew/alembic:0.6.3 + image: git.kretzer.club/andrew/alembic:0.6.4 container_name: alembic ports: - "8420:8420" diff --git a/app/services/pipeline_runner.py b/app/services/pipeline_runner.py index f915f3b..74f0d50 100644 --- a/app/services/pipeline_runner.py +++ b/app/services/pipeline_runner.py @@ -12,6 +12,15 @@ from app.settings import settings # all-jobs-share-one-lock behavior exactly rather than over-engineering it. _lock = asyncio.Lock() +# How long a SCHEDULED run waits for the lock before giving up (recorded as +# skipped_lock). The old flock -n instant-skip starves fixed-time schedules: +# the Sunday 08:30 strip-mb-tags run took 10m19s on 2026-07-12, so the 08:35 +# and 08:40 jobs both hit the held lock and silently lost their only slot of +# the week. Scheduled jobs have nobody watching, so queueing (bounded) beats +# skipping; manual runs keep the instant skip because a person clicking "Run +# now" should get an immediate answer, not a silent 30-minute wait. +SCHEDULED_LOCK_WAIT_SECONDS = 1800.0 + _ENV_PASSTHROUGH_KEYS = ("PATH", "HOME", "LANG", "LC_ALL", "TZ") @@ -94,18 +103,30 @@ async def _execute( timeout: float | None, use_lock: bool, capture: bool, + lock_wait: float | None, ) -> tuple[JobRun, str]: """Shared core for run_job/run_job_capture: acquire the lock (unless use_lock=False), record the run, exec the subprocess, and finalize. When capture=True, stdout+stderr is captured as text and returned; otherwise it streams straight to the log file. Returns (JobRun, output_text) -- the text - is "" in the non-capture and skipped-lock cases.""" + is "" in the non-capture and skipped-lock cases. + + lock_wait: how long to wait for a held lock before recording + skipped_lock. None picks the policy default: SCHEDULED_LOCK_WAIT_SECONDS + for triggered_by="schedule", instant skip for everything else.""" started_at = time.time() + if lock_wait is None: + lock_wait = SCHEDULED_LOCK_WAIT_SECONDS if triggered_by == "schedule" else 0.0 acquired = False if use_lock: if not await _try_acquire_nowait(): - return _record_run(job_key, started_at, triggered_by, finished_at=started_at, status="skipped_lock"), "" + if lock_wait <= 0: + return _record_run(job_key, started_at, triggered_by, finished_at=started_at, status="skipped_lock"), "" + try: + await asyncio.wait_for(_lock.acquire(), timeout=lock_wait) + except asyncio.TimeoutError: + return _record_run(job_key, started_at, triggered_by, finished_at=time.time(), status="skipped_lock"), "" acquired = True try: @@ -170,18 +191,21 @@ async def run_job( triggered_by: str = "schedule", timeout: float | None = None, use_lock: bool = True, + lock_wait: float | None = None, ) -> JobRun: """Run a pipeline command under the shared mutual-exclusion lock, - recording one job_runs row start-to-finish. If the lock is already - held, records status='skipped_lock' immediately and returns without - running anything -- the old flock -n behavior, now visible in the UI - instead of silently skipping. + recording one job_runs row start-to-finish. If the lock is already held, + scheduled runs (triggered_by="schedule") wait up to + SCHEDULED_LOCK_WAIT_SECONDS for it before recording skipped_lock -- see + that constant for why -- while manual/other runs record skipped_lock + immediately (the old flock -n behavior, visible in the UI instead of + silently skipping). Pass lock_wait to override either way. use_lock=False runs the command WITHOUT taking the pipeline lock, for read-only jobs that never touch the library (e.g. the status report). Those must never be starved by a long-running write job, and running them concurrently is safe.""" - run, _ = await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=False) + run, _ = await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=False, lock_wait=lock_wait) return run @@ -191,6 +215,7 @@ async def run_job_capture( triggered_by: str = "manual", timeout: float | None = None, use_lock: bool = True, + lock_wait: float | None = None, ) -> tuple[JobRun, str]: """Like run_job(), but captures stdout+stderr as text and returns it alongside the JobRun, instead of only writing it to the log file -- @@ -198,7 +223,7 @@ async def run_job_capture( dedup_review_service's dry-run scan. The full output is still written to a log file afterward so job_runs.log_path works the same as any other job.""" - return await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=True) + return await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=True, lock_wait=lock_wait) async def run_playlist(playlist_name: str, no_m3u: bool = False, triggered_by: str = "schedule") -> JobRun: diff --git a/pipeline/lib/pipeline-status.sh b/pipeline/lib/pipeline-status.sh index 17415f3..d32e880 100755 --- a/pipeline/lib/pipeline-status.sh +++ b/pipeline/lib/pipeline-status.sh @@ -70,17 +70,6 @@ human_age() { fi } -# Newest matching log → "|", empty if no match. -newest_log() { - local glob="$1" - local newest - newest=$(ls -1t $glob 2>/dev/null | head -1) - [[ -z "$newest" ]] && return - local mtime - mtime=$(stat -c %Y "$newest" 2>/dev/null) || return - echo "$((NOW_TS - mtime))|$newest" -} - # Per-playlist log status. playlist_status() { local log="$1" @@ -226,20 +215,15 @@ dedup_today_status() { fi # 3. Maintenance freshness - # Each weekly/monthly task: find its newest log and check age. + # Read from the app's job_runs table (the scheduler's own record) instead of + # globbing for log files. Jobs run through pipeline_runner write their logs + # to logs//.log, so the old cron-era filename globs + # matched nothing for jobs that don't also write their own log, and healthy + # jobs were reported as "never run yet". The DB also distinguishes lock + # skips and failures, which a missing log file can't. # Limits: weekly tasks should be <9 days old; daily <2; monthly <35. section "Maintenance (last run)" - declare -A MAINT_GLOB=( - [strip-mb-tags]="${ALEMBIC_CONFIG_DIR:-/config}/logs/mb-strip-*.log" - [strip-watermark-art]="${ALEMBIC_CONFIG_DIR:-/config}/logs/strip-watermark-*.log" - [scrub-watermark-text]="${ALEMBIC_CONFIG_DIR:-/config}/logs/scrub-text-*.log" - [clean-sldl-index]="${ALEMBIC_CONFIG_DIR:-/config}/logs/clean-index-*.log" - [clear-bad-genres]="${ALEMBIC_CONFIG_DIR:-/config}/logs/clear-bad-genres-*.log" - [spotify-genre]="${ALEMBIC_CONFIG_DIR:-/config}/logs/spotify-genre-*.log" - [dedup-library]="${ALEMBIC_CONFIG_DIR:-/config}/logs/dedup-*.log" - [upgrade-mp3-to-flac]="${ALEMBIC_CONFIG_DIR:-/config}/logs/upgrade-mp3-*.log" - ) declare -A MAINT_LIMIT=( [strip-mb-tags]=$((9*86400)) [strip-watermark-art]=$((9*86400)) @@ -250,42 +234,106 @@ dedup_today_status() { [dedup-library]=$((2*86400)) [upgrade-mp3-to-flac]=$((35*86400)) ) + # Emits one line per task: task|latest_status|latest_age_s|success_age_s|success_log + # (ages are -1 when there is no such run). Read-only DB open; prints nothing + # if the DB is missing so every task falls through to "never run yet". + maint_rows=$(python3 - "${ALEMBIC_CONFIG_DIR:-/config}/alembic.db" <<'PYEOF' +import sqlite3 +import sys +import time + +# Digest label -> job_runs.job_key. The scheduled genre refresh records under +# genre:run (via genre_review_service) and dedup under dedup:scan (via +# dedup_review_service); everything else is its MAINTENANCE_JOBS key. +KEYS = [ + ("strip-mb-tags", "maintenance:strip_mb_tags"), + ("strip-watermark-art", "maintenance:strip_watermark_art"), + ("scrub-watermark-text", "maintenance:scrub_watermark_text"), + ("clean-sldl-index", "maintenance:clean_sldl_index"), + ("clear-bad-genres", "maintenance:clear_bad_genres"), + ("spotify-genre", "genre:run"), + ("dedup-library", "dedup:scan"), + ("upgrade-mp3-to-flac", "maintenance:upgrade_mp3_to_flac"), +] +now = time.time() +try: + conn = sqlite3.connect(f"file:{sys.argv[1]}?mode=ro", uri=True) + conn.execute("SELECT 1 FROM job_runs LIMIT 1") +except sqlite3.Error: + sys.exit(0) +for task, key in KEYS: + latest = conn.execute( + "SELECT status, started_at FROM job_runs WHERE job_key = ? ORDER BY started_at DESC LIMIT 1", + (key,), + ).fetchone() + if latest is None: + print(f"{task}|none|-1|-1|") + continue + success = conn.execute( + "SELECT started_at, log_path FROM job_runs WHERE job_key = ? AND status = 'success' " + "ORDER BY started_at DESC LIMIT 1", + (key,), + ).fetchone() + s_age = int(now - success[0]) if success else -1 + s_log = (success[1] or "") if success else "" + print(f"{task}|{latest[0]}|{int(now - latest[1])}|{s_age}|{s_log}") +PYEOF +) + for task in strip-mb-tags strip-watermark-art scrub-watermark-text clean-sldl-index clear-bad-genres spotify-genre dedup-library upgrade-mp3-to-flac; do - res=$(newest_log "${MAINT_GLOB[$task]}") - if [[ -z "$res" ]]; then + row=$(grep "^${task}|" <<< "$maint_rows" || true) + IFS='|' read -r _ latest_status latest_age age_s log <<< "$row" + if [[ -z "$row" || "$latest_status" == "none" ]]; then mark_skip "$(printf '%-22s never run yet' "$task")" continue fi - age_s=${res%%|*} - log=${res##*|} + if [[ "$age_s" -lt 0 ]]; then + # attempted, but never once succeeded -- say what the last attempt did + case "$latest_status" in + skipped_lock) note="skipped, lock busy" ;; + running) note="running now" ;; + *) note="$latest_status" ;; + esac + mark_warn "$(printf '%-22s never succeeded (last attempt %s: %s)' "$task" "$(human_age "$latest_age")" "$note")" + continue + fi age_h=$(human_age "$age_s") - case "$task" in - spotify-genre) - snip=$(grep -oE "written: [0-9]+, unchanged: [0-9]+" "$log" | tail -1) ;; - clear-bad-genres) - snip=$(grep -oE "(would-clear|cleared): [0-9]+ tracks" "$log" | tail -1) ;; - clean-sldl-index) - snip=$(grep -oE "[0-9]+ m3us, total [0-9]+ kept, [0-9]+ dropped" "$log" | tail -1) ;; - strip-watermark-art) - snip=$(grep -oE "stripped from [0-9]+ tracks|no images shared" "$log" | tail -1) ;; - scrub-watermark-text) - snip=$(grep -oE "stripped [0-9]+ frame\(s\) across [0-9]+ file" "$log" | tail -1) ;; - strip-mb-tags) - snip=$(grep -oE "Done\. Log:" "$log" | head -1 | sed 's/Done\. Log:/done/') ;; - dedup-library) - snip=$(grep "=== Summary:" "$log" | tail -1 \ - | sed -E 's/.*=== Summary: //; s/ ===.*//; s/ \([^)]*\)//') ;; - upgrade-mp3-to-flac) - # The log's "to attempt" count is frozen at the last monthly run, so on - # its own it looks stale against the live "Library by format" section. - # Pair it with the current format:mp3 count so the drop (upgrades that - # have landed since) is visible instead of looking like a mismatch. - attempted=$(grep -oE "[0-9]+ MP3 tracks to attempt" "$log" | grep -oE "^[0-9]+" | tail -1) - mp3_now=$(beet ls 'format:mp3' 2>/dev/null | grep -c '^') - snip="${attempted:-?} attempted · ${mp3_now} MP3 now" ;; + snip="" + if [[ -n "$log" && -f "$log" ]]; then + case "$task" in + spotify-genre) + snip=$(grep -oE "written: [0-9]+, unchanged: [0-9]+" "$log" | tail -1) ;; + clear-bad-genres) + snip=$(grep -oE "rewrote: [0-9]+ tracks, blanked: [0-9]+ tracks|(would-clear|cleared): [0-9]+ tracks" "$log" | tail -1) ;; + clean-sldl-index) + snip=$(grep -oE "[0-9]+ m3us, total [0-9]+ kept, [0-9]+ dropped" "$log" | tail -1) ;; + strip-watermark-art) + snip=$(grep -oE "stripped from [0-9]+ tracks|no images shared" "$log" | tail -1) ;; + scrub-watermark-text) + snip=$(grep -oE "stripped [0-9]+ frame\(s\) across [0-9]+ file" "$log" | tail -1) ;; + strip-mb-tags) + snip=$(grep -oE "Done\. Log:" "$log" | head -1 | sed 's/Done\. Log:/done/') ;; + dedup-library) + snip=$(grep "=== Summary:" "$log" | tail -1 \ + | sed -E 's/.*=== Summary: //; s/ ===.*//; s/ \([^)]*\)//') ;; + upgrade-mp3-to-flac) + # The log's "to attempt" count is frozen at the last monthly run, so on + # its own it looks stale against the live "Library by format" section. + # Pair it with the current format:mp3 count so the drop (upgrades that + # have landed since) is visible instead of looking like a mismatch. + attempted=$(grep -oE "[0-9]+ MP3 tracks to attempt" "$log" | grep -oE "^[0-9]+" | tail -1) + mp3_now=$(beet ls 'format:mp3' 2>/dev/null | grep -c '^') + snip="${attempted:-?} attempted · ${mp3_now} MP3 now" ;; + esac + fi + # Age/snippet describe the last SUCCESS; if a newer attempt failed or was + # lock-skipped, say so rather than hiding it behind the healthy line. + case "$latest_status" in + failed) snip="${snip:+$snip, }latest attempt failed" ;; + skipped_lock) snip="${snip:+$snip, }latest attempt skipped, lock busy" ;; esac line=$(printf '%-22s %-9s %s' "$task" "$age_h" "${snip:-}") - if (( age_s > MAINT_LIMIT[$task] )); then + if (( age_s > MAINT_LIMIT[$task] )) || [[ "$latest_status" == "failed" ]]; then mark_warn "$line" else mark_ok "$line" diff --git a/tests/test_pipeline_runner.py b/tests/test_pipeline_runner.py index 0214199..58d3c94 100644 --- a/tests/test_pipeline_runner.py +++ b/tests/test_pipeline_runner.py @@ -25,11 +25,41 @@ def test_run_job_failure(): assert run.exit_code == 1 -def test_lock_skip_when_held(): +def test_manual_run_skips_immediately_when_lock_held(): async def scenario(): await pr._lock.acquire() try: - return await pr.run_job("test:locked", ["true"], use_lock=True) + return await pr.run_job("test:locked", ["true"], use_lock=True, triggered_by="manual") + finally: + pr._lock.release() + + run = _run(scenario()) + assert run.status == "skipped_lock" + + +def test_scheduled_run_waits_for_lock_by_default(monkeypatch): + # scheduled runs must QUEUE behind a held lock (bounded), not skip -- + # instant-skip starved the Sunday maintenance block (see + # SCHEDULED_LOCK_WAIT_SECONDS) + monkeypatch.setattr(pr, "SCHEDULED_LOCK_WAIT_SECONDS", 5.0) + + async def scenario(): + await pr._lock.acquire() + task = asyncio.ensure_future(pr.run_job("test:queued", ["true"], triggered_by="schedule")) + await asyncio.sleep(0.2) # let the job start waiting on the lock + pr._lock.release() + return await task + + run = _run(scenario()) + assert run.status == "success" + assert not pr._lock.locked() + + +def test_scheduled_run_skips_after_lock_wait_timeout(): + async def scenario(): + await pr._lock.acquire() + try: + return await pr.run_job("test:waited_out", ["true"], triggered_by="schedule", lock_wait=0.2) finally: pr._lock.release()