R7: factor run_job/run_job_capture onto one shared core
The two were ~90% identical (lock/skip, job_runs create, subprocess exec, timeout/kill, finalize) and had already drifted once during the use_lock change. Extract _execute(capture=bool) plus _record_run/_finalize_run helpers; run_job and run_job_capture become thin wrappers. run_job_capture also gains the use_lock parameter for free. Behavior is unchanged. Covered by tests/test_pipeline_runner.py, now including run_job_capture output capture + persistence and its skipped-lock case (42 tests green). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
andrew
+111
-153
@@ -58,6 +58,112 @@ def _summarize_log(log_path: Path, max_len: int = 200) -> str:
|
||||
return ""
|
||||
|
||||
|
||||
def _record_run(job_key: str, started_at: float, triggered_by: str, **fields) -> JobRun:
|
||||
"""Insert a job_runs row and return the refreshed object."""
|
||||
db = SessionLocal()
|
||||
try:
|
||||
run = JobRun(job_key=job_key, started_at=started_at, triggered_by=triggered_by, **fields)
|
||||
db.add(run)
|
||||
db.commit()
|
||||
db.refresh(run)
|
||||
return run
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def _finalize_run(run_id: int, status: str, exit_code: int | None, log_path: Path) -> JobRun:
|
||||
"""Fill in the terminal fields on a previously-'running' job_runs row."""
|
||||
db = SessionLocal()
|
||||
try:
|
||||
run = db.get(JobRun, run_id)
|
||||
run.finished_at = time.time()
|
||||
run.status = status
|
||||
run.exit_code = exit_code
|
||||
run.summary = _summarize_log(log_path)
|
||||
db.commit()
|
||||
db.refresh(run)
|
||||
return run
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
async def _execute(
|
||||
job_key: str,
|
||||
argv: list[str],
|
||||
triggered_by: str,
|
||||
timeout: float | None,
|
||||
use_lock: bool,
|
||||
capture: bool,
|
||||
) -> 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."""
|
||||
started_at = time.time()
|
||||
|
||||
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"), ""
|
||||
acquired = True
|
||||
|
||||
try:
|
||||
log_dir = settings.logs_dir / job_key.replace(":", "_")
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
log_path = log_dir / f"{time.strftime('%Y%m%d-%H%M%S')}.log"
|
||||
run_id = _record_run(job_key, started_at, triggered_by, status="running", log_path=str(log_path)).id
|
||||
|
||||
output_text = ""
|
||||
exit_code: int | None
|
||||
try:
|
||||
if capture:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*argv, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, env=_subprocess_env(),
|
||||
)
|
||||
try:
|
||||
stdout_bytes, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
|
||||
exit_code = proc.returncode
|
||||
output_text = stdout_bytes.decode(errors="replace")
|
||||
except asyncio.TimeoutError:
|
||||
proc.kill()
|
||||
await proc.wait()
|
||||
exit_code = -1
|
||||
output_text += f"\n[pipeline_runner] TIMEOUT after {timeout}s -- process killed\n"
|
||||
status = "success" if exit_code == 0 else "failed"
|
||||
log_path.write_text(output_text)
|
||||
else:
|
||||
with open(log_path, "wb") as log_file:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*argv, stdout=log_file, stderr=asyncio.subprocess.STDOUT, env=_subprocess_env(),
|
||||
)
|
||||
try:
|
||||
exit_code = await asyncio.wait_for(proc.wait(), timeout=timeout)
|
||||
except asyncio.TimeoutError:
|
||||
proc.kill()
|
||||
await proc.wait()
|
||||
exit_code = -1
|
||||
with open(log_path, "a") as f:
|
||||
f.write(f"\n[pipeline_runner] TIMEOUT after {timeout}s -- process killed\n")
|
||||
status = "success" if exit_code == 0 else "failed"
|
||||
except Exception as exc:
|
||||
exit_code = None
|
||||
status = "failed"
|
||||
msg = f"\n[pipeline_runner] exception before/while running: {exc!r}\n"
|
||||
if capture:
|
||||
output_text += msg
|
||||
log_path.write_text(output_text)
|
||||
else:
|
||||
with open(log_path, "a") as f:
|
||||
f.write(msg)
|
||||
|
||||
run = _finalize_run(run_id, status, exit_code, log_path)
|
||||
return run, output_text
|
||||
finally:
|
||||
if acquired:
|
||||
_lock.release()
|
||||
|
||||
|
||||
async def run_job(
|
||||
job_key: str,
|
||||
argv: list[str],
|
||||
@@ -75,83 +181,8 @@ async def run_job(
|
||||
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()
|
||||
|
||||
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(":", "_")
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
log_path = log_dir / f"{time.strftime('%Y%m%d-%H%M%S')}.log"
|
||||
|
||||
db = SessionLocal()
|
||||
run = JobRun(
|
||||
job_key=job_key,
|
||||
started_at=started_at,
|
||||
status="running",
|
||||
triggered_by=triggered_by,
|
||||
log_path=str(log_path),
|
||||
)
|
||||
db.add(run)
|
||||
db.commit()
|
||||
db.refresh(run)
|
||||
run_id = run.id
|
||||
db.close()
|
||||
|
||||
exit_code: int | None
|
||||
try:
|
||||
with open(log_path, "wb") as log_file:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*argv,
|
||||
stdout=log_file,
|
||||
stderr=asyncio.subprocess.STDOUT,
|
||||
env=_subprocess_env(),
|
||||
)
|
||||
try:
|
||||
exit_code = await asyncio.wait_for(proc.wait(), timeout=timeout)
|
||||
except asyncio.TimeoutError:
|
||||
proc.kill()
|
||||
await proc.wait()
|
||||
exit_code = -1
|
||||
with open(log_path, "a") as f:
|
||||
f.write(f"\n[pipeline_runner] TIMEOUT after {timeout}s -- process killed\n")
|
||||
status = "success" if exit_code == 0 else "failed"
|
||||
except Exception as exc:
|
||||
exit_code = None
|
||||
status = "failed"
|
||||
with open(log_path, "a") as f:
|
||||
f.write(f"\n[pipeline_runner] exception before/while running: {exc!r}\n")
|
||||
|
||||
finished_at = time.time()
|
||||
db = SessionLocal()
|
||||
run = db.get(JobRun, run_id)
|
||||
run.finished_at = finished_at
|
||||
run.status = status
|
||||
run.exit_code = exit_code
|
||||
run.summary = _summarize_log(log_path)
|
||||
db.commit()
|
||||
db.refresh(run)
|
||||
db.close()
|
||||
return run
|
||||
finally:
|
||||
if acquired:
|
||||
_lock.release()
|
||||
run, _ = await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=False)
|
||||
return run
|
||||
|
||||
|
||||
async def run_job_capture(
|
||||
@@ -159,88 +190,15 @@ async def run_job_capture(
|
||||
argv: list[str],
|
||||
triggered_by: str = "manual",
|
||||
timeout: float | None = None,
|
||||
use_lock: bool = True,
|
||||
) -> 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 --
|
||||
for callers that need to parse structured (JSON-lines) output, e.g.
|
||||
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. Shares the same lock as run_job()."""
|
||||
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, ""
|
||||
|
||||
try:
|
||||
log_dir = settings.logs_dir / job_key.replace(":", "_")
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
log_path = log_dir / f"{time.strftime('%Y%m%d-%H%M%S')}.log"
|
||||
|
||||
db = SessionLocal()
|
||||
run = JobRun(
|
||||
job_key=job_key,
|
||||
started_at=started_at,
|
||||
status="running",
|
||||
triggered_by=triggered_by,
|
||||
log_path=str(log_path),
|
||||
)
|
||||
db.add(run)
|
||||
db.commit()
|
||||
db.refresh(run)
|
||||
run_id = run.id
|
||||
db.close()
|
||||
|
||||
output_text = ""
|
||||
exit_code: int | None
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*argv,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.STDOUT,
|
||||
env=_subprocess_env(),
|
||||
)
|
||||
try:
|
||||
stdout_bytes, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
|
||||
exit_code = proc.returncode
|
||||
output_text = stdout_bytes.decode(errors="replace")
|
||||
except asyncio.TimeoutError:
|
||||
proc.kill()
|
||||
await proc.wait()
|
||||
exit_code = -1
|
||||
output_text += f"\n[pipeline_runner] TIMEOUT after {timeout}s -- process killed\n"
|
||||
status = "success" if exit_code == 0 else "failed"
|
||||
except Exception as exc:
|
||||
exit_code = None
|
||||
status = "failed"
|
||||
output_text += f"\n[pipeline_runner] exception before/while running: {exc!r}\n"
|
||||
|
||||
log_path.write_text(output_text)
|
||||
|
||||
finished_at = time.time()
|
||||
db = SessionLocal()
|
||||
run = db.get(JobRun, run_id)
|
||||
run.finished_at = finished_at
|
||||
run.status = status
|
||||
run.exit_code = exit_code
|
||||
run.summary = _summarize_log(log_path)
|
||||
db.commit()
|
||||
db.refresh(run)
|
||||
db.close()
|
||||
return run, output_text
|
||||
finally:
|
||||
_lock.release()
|
||||
other job."""
|
||||
return await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=True)
|
||||
|
||||
|
||||
async def run_playlist(playlist_name: str, no_m3u: bool = False, triggered_by: str = "schedule") -> JobRun:
|
||||
|
||||
@@ -58,3 +58,25 @@ def test_timeout_kills_and_marks_failed():
|
||||
run = _run(pr.run_job("test:timeout", ["sleep", "5"], timeout=0.3))
|
||||
assert run.status == "failed"
|
||||
assert run.exit_code == -1
|
||||
|
||||
|
||||
def test_run_job_capture_returns_and_persists_output():
|
||||
from pathlib import Path
|
||||
|
||||
run, out = _run(pr.run_job_capture("test:cap", ["sh", "-c", "echo hello; echo world"]))
|
||||
assert run.status == "success"
|
||||
assert "hello" in out and "world" in out
|
||||
assert "hello" in Path(run.log_path).read_text() # also written to the log file
|
||||
|
||||
|
||||
def test_run_job_capture_skips_when_locked():
|
||||
async def scenario():
|
||||
await pr._lock.acquire()
|
||||
try:
|
||||
return await pr.run_job_capture("test:caplock", ["true"])
|
||||
finally:
|
||||
pr._lock.release()
|
||||
|
||||
run, out = _run(scenario())
|
||||
assert run.status == "skipped_lock"
|
||||
assert out == ""
|
||||
|
||||
Reference in New Issue
Block a user