andrew/alembic
1
0
Fork 0
Code Pull Requests Actions Releases 12 Wiki Activity

Stop the daily status report (and its Telegram digest) being starved by the lock

Browse Source 
Root cause of a missed 9am notification: the read-only pipeline_status_report
runs at 09:30, but enrich_buy_url (09:10, --apply) held the single global
pipeline lock long past 09:30, so the status report recorded skipped_lock and
never sent its Telegram digest. build_fingerprint_index (09:25) was starved the
same way.

- run_job/run_lib_script gain use_lock; the status report runs with
  use_lock=False since it only reads logs, pings Navidrome, and sends Telegram.
  It can no longer be blocked by a long write job, and running concurrently is
  safe.
- _lib was silently dropping its timeout argument, so maintenance jobs had no
  timeout and a hung one held the lock until the next container restart. Thread
  timeout through _lib -> run_lib_script -> run_job, cap enrich_buy_url at 30m
  and build_fingerprint_index at 60m, and give the status report a 5m cap.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
andrew
2026-07-10 10:43:37 -06:00
parent ade32c4d0c
commit 852b3797d8
2 changed files with 60 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/pipeline_runner.py
+41 -18
View File
@@ -62,28 +62,37 @@ async def run_job(
argv: list[str],
triggered_by: str = "schedule",
timeout: float | None = None,
use_lock: bool = True,
) -> 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."""
instead of silently skipping.
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."""
started_at = time.time()
if not await _try_acquire_nowait():
db = SessionLocal()
run = JobRun(
job_key=job_key,
started_at=started_at,
finished_at=started_at,
status="skipped_lock",
triggered_by=triggered_by,
)
db.add(run)
db.commit()
db.refresh(run)
db.close()
return run
acquired = False
if use_lock:
if not await _try_acquire_nowait():
db = SessionLocal()
run = JobRun(
job_key=job_key,
started_at=started_at,
finished_at=started_at,
status="skipped_lock",
triggered_by=triggered_by,
)
db.add(run)
db.commit()
db.refresh(run)
db.close()
return run
acquired = True
try:
log_dir = settings.logs_dir / job_key.replace(":", "_")
@@ -140,7 +149,8 @@ async def run_job(
db.close()
return run
finally:
_lock.release()
if acquired:
_lock.release()
async def run_job_capture(
@@ -248,9 +258,22 @@ async def run_import_track(args: list[str], triggered_by: str = "manual") -> Job
return await run_job("manual:import", [str(script)] + args, triggered_by=triggered_by)
async def run_lib_script(job_key: str, script_name: str, args: list[str] | None = None, triggered_by: str = "schedule") -> JobRun:
async def run_lib_script(
job_key: str,
script_name: str,
args: list[str] | None = None,
triggered_by: str = "schedule",
timeout: float | None = None,
use_lock: bool = True,
) -> JobRun:
script = settings.pipeline_dir / "lib" / script_name
return await run_job(job_key, [str(script)] + (args or []), triggered_by=triggered_by)
return await run_job(
job_key,
[str(script)] + (args or []),
triggered_by=triggered_by,
timeout=timeout,
use_lock=use_lock,
)
async def run_beet(job_key: str, args: list[str], triggered_by: str = "schedule") -> JobRun:
app/services/scheduler_service.py
+19 -5
View File
@@ -47,9 +47,17 @@ TIMEZONE = _resolve_timezone(settings.timezone)
# ---------------------------------------------------------------------------
def _lib(job_key: str, script_name: str, args: list[str] | None = None, timeout: float | None = None):
def _lib(
job_key: str,
script_name: str,
args: list[str] | None = None,
timeout: float | None = None,
use_lock: bool = True,
):
async def _run(triggered_by: str = "schedule"):
return await pipeline_runner.run_lib_script(job_key, script_name, args, triggered_by=triggered_by)
return await pipeline_runner.run_lib_script(
job_key, script_name, args, triggered_by=triggered_by, timeout=timeout, use_lock=use_lock
)
return _run
@@ -120,15 +128,21 @@ MAINTENANCE_JOBS: dict[str, tuple[dict, callable]] = {
),
"maintenance:enrich_buy_url": (
dict(minute=10, hour=9),
_lib("maintenance:enrich_buy_url", "enrich-buy-url.py", ["--apply"]),
# 30-min cap: this hits external buy-link APIs per track and has hung
# holding the pipeline lock (2026-07-09/10), starving every job after
# it. A timeout releases the lock so the rest of the 9am chain runs.
_lib("maintenance:enrich_buy_url", "enrich-buy-url.py", ["--apply"], timeout=1800),
),
"maintenance:build_fingerprint_index": (
dict(minute=25, hour=9),
_lib("maintenance:build_fingerprint_index", "build-fingerprint-index.py", ["--workers", "8"]),
_lib("maintenance:build_fingerprint_index", "build-fingerprint-index.py", ["--workers", "8"], timeout=3600),
),
"maintenance:pipeline_status_report": (
dict(minute=30, hour=9),
_lib("maintenance:pipeline_status_report", "pipeline-status.sh"),
# Read-only (reads logs, pings Navidrome, sends Telegram). Runs WITHOUT
# the pipeline lock so the daily digest can never be starved by a long
# write job, and a short timeout so a stuck Telegram call can't wedge it.
_lib("maintenance:pipeline_status_report", "pipeline-status.sh", timeout=300, use_lock=False),
),
"maintenance:log_rotation": (
dict(minute=0, hour=0),