fix(r5): address the fifth /code-review high round — 6 findings
- run_ffmpeg stops the process from ONE exit path (`except BaseException`), so an exception out of on_progress (a DB write) can no longer orphan a live ffmpeg after a 10s join on its pipes; _stop is now a no-op on a dead process - both run_ffmpeg callbacks are throttled to ~1s (they were a DB round-trip per output line: ~10^5 queries on a long re-encode), mirroring the download hook - volume arithmetic extracted to lib/playerVolume with unit tests; a corrupt level now falls back to the LAST GOOD one instead of full blast, and the sanitising happens where the stored value enters - RSS is back to two except clauses (the shared tail already sits after the try) - new tests: callback-exception cleanup, the two throttles; the fake Popen grew poll() and its wait count is now asserted Mutation-checked: each new group goes red when its fix is reverted.
This commit is contained in:
@@ -288,7 +288,10 @@ def _drain_tail(stream, sink: list[str]) -> None:
|
||||
|
||||
|
||||
def _stop(proc: subprocess.Popen) -> None:
|
||||
"""Terminate, then kill — never wait unbounded on a process we've given up on."""
|
||||
"""Terminate, then kill — never wait unbounded on a process we've given up on. Safe to call on
|
||||
a process that already exited (the single exit path below may reach it after a clean wait)."""
|
||||
if proc.poll() is not None:
|
||||
return
|
||||
proc.terminate()
|
||||
try:
|
||||
proc.wait(timeout=5)
|
||||
@@ -335,16 +338,20 @@ def run_ffmpeg(cmd, total_s, on_progress, should_cancel, stall_timeout_s=_FFMPEG
|
||||
if secs is not None and total_s:
|
||||
on_progress(max(0.0, min(99.0, secs * 100.0 / total_s)))
|
||||
if should_cancel():
|
||||
_stop(proc)
|
||||
raise EditAborted
|
||||
if time.monotonic() - last_output > stall_timeout_s:
|
||||
_stop(proc)
|
||||
raise RuntimeError(f"ffmpeg stalled: no progress for {int(stall_timeout_s)}s")
|
||||
try:
|
||||
ret = proc.wait(timeout=_FFMPEG_EXIT_GRACE_S)
|
||||
except subprocess.TimeoutExpired: # pipes closed but the process lingers
|
||||
_stop(proc)
|
||||
raise RuntimeError("ffmpeg did not exit after closing its output")
|
||||
except BaseException:
|
||||
# ONE exit path stops the process, so the invariant is unconditional: nothing leaves this
|
||||
# function with ffmpeg still running. Not just our own EditAborted/stall — `on_progress`
|
||||
# writes to the database, and an exception from THERE used to unwind past a live process,
|
||||
# orphaning it (still encoding, no one left to reap it) after a 10s join on its pipes.
|
||||
_stop(proc)
|
||||
raise
|
||||
finally:
|
||||
for t in readers:
|
||||
t.join(timeout=_FFMPEG_EXIT_GRACE_S)
|
||||
|
||||
@@ -74,15 +74,16 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str
|
||||
channel = futures[fut]
|
||||
try:
|
||||
entries = fut.result()
|
||||
except Exception as exc:
|
||||
# A failed poll is NOT an empty feed: skip the apply so `last_rss_at` stays where
|
||||
# it was. Stamping it here made a permanently-broken feed look freshly polled.
|
||||
# An RssError is an EXPECTED outcome (404, timeout) → one warning line; anything
|
||||
# else is a bug on our side and earns a traceback.
|
||||
if isinstance(exc, RssError):
|
||||
log.warning("RSS poll skipped for channel %s: %s", channel.id, exc)
|
||||
else:
|
||||
log.exception("RSS fetch failed for channel %s", channel.id)
|
||||
# A failed poll is NOT an empty feed: skip the apply so `last_rss_at` stays where it
|
||||
# was (stamping it made a permanently-broken feed look freshly polled). The shared
|
||||
# counter/progress tail lives after the statement, so the two outcomes can stay two
|
||||
# clauses: an RssError is EXPECTED (404, timeout) and gets one warning line; anything
|
||||
# else is a bug on our side and earns a traceback.
|
||||
except RssError as exc:
|
||||
log.warning("RSS poll skipped for channel %s: %s", channel.id, exc)
|
||||
failed += 1
|
||||
except Exception:
|
||||
log.exception("RSS fetch failed for channel %s", channel.id)
|
||||
failed += 1
|
||||
else:
|
||||
try:
|
||||
|
||||
+44
-4
@@ -319,6 +319,46 @@ def _make_progress_hook(job_id: int, asset_id: int):
|
||||
return hook
|
||||
|
||||
|
||||
# ffmpeg's `-progress pipe:1` emits a block of ~12 lines every ~0.5s, and run_ffmpeg calls both of
|
||||
# its callbacks once per LINE. Each of these is a database round-trip (`_job_status` opens a session
|
||||
# and SELECTs; `_set_job` opens one and UPDATE+COMMITs), so an unthrottled pair costs ~10^5 queries
|
||||
# on a long re-encode. Both are therefore rate-limited to ~1s — the same 1s DB throttle the
|
||||
# download progress hook already applies (see `_make_progress_hook`).
|
||||
_FFMPEG_CALLBACK_INTERVAL_S = 1.0
|
||||
|
||||
|
||||
def _ffmpeg_progress_writer(job_id: int, phase: str):
|
||||
"""`on_progress` for run_ffmpeg: persist at most one row write per second, and only when the
|
||||
whole-percent value actually changed (the bar shows integers)."""
|
||||
state = {"at": 0.0, "pct": -1}
|
||||
|
||||
def write(pct: float) -> None:
|
||||
whole = int(pct)
|
||||
now = time.monotonic()
|
||||
if whole == state["pct"] or now - state["at"] < _FFMPEG_CALLBACK_INTERVAL_S:
|
||||
return
|
||||
state["at"] = now
|
||||
state["pct"] = whole
|
||||
_set_job(job_id, progress=whole, phase=phase)
|
||||
|
||||
return write
|
||||
|
||||
|
||||
def _ffmpeg_cancel_poll(job_id: int):
|
||||
"""`should_cancel` for run_ffmpeg: ask the DB at most once a second and reuse the answer in
|
||||
between. Cancellation is sticky, so once it says yes it never queries again."""
|
||||
state = {"at": 0.0, "cancelled": False}
|
||||
|
||||
def cancelled() -> bool:
|
||||
now = time.monotonic()
|
||||
if not state["cancelled"] and now - state["at"] >= _FFMPEG_CALLBACK_INTERVAL_S:
|
||||
state["at"] = now
|
||||
state["cancelled"] = _job_status(job_id) not in ("running", None)
|
||||
return state["cancelled"]
|
||||
|
||||
return cancelled
|
||||
|
||||
|
||||
def _pp_phase(pp: str) -> str:
|
||||
"""Map a yt-dlp postprocessor class name to a user-facing phase label. ffmpeg post-steps
|
||||
aren't byte-progress operations (no %), so the UI shows the step name + an indeterminate
|
||||
@@ -492,8 +532,8 @@ def _ensure_browser_playable(job_id: int, produced: Path, spec: dict, info: dict
|
||||
editmod.run_ffmpeg(
|
||||
editmod.build_browser_safe_cmd(produced, dest, reenc_v, reenc_a),
|
||||
total_s=total_s,
|
||||
on_progress=lambda pct: _set_job(job_id, progress=int(pct), phase="processing"),
|
||||
should_cancel=lambda: _job_status(job_id) not in ("running", None),
|
||||
on_progress=_ffmpeg_progress_writer(job_id, "processing"),
|
||||
should_cancel=_ffmpeg_cancel_poll(job_id),
|
||||
)
|
||||
produced.unlink(missing_ok=True)
|
||||
# Keep the stored asset metadata truthful about what's now on disk.
|
||||
@@ -671,8 +711,8 @@ def _process_edit(job_id: int, asset_id: int) -> None:
|
||||
editmod.run_ffmpeg(
|
||||
cmd,
|
||||
total_s,
|
||||
on_progress=lambda pct: _set_job(job_id, progress=int(pct), phase="editing"),
|
||||
should_cancel=lambda: _job_status(job_id) not in ("running", None),
|
||||
on_progress=_ffmpeg_progress_writer(job_id, "editing"),
|
||||
should_cancel=_ffmpeg_cancel_poll(job_id),
|
||||
)
|
||||
if not dest_staging.exists():
|
||||
raise RuntimeError("ffmpeg produced no output file")
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
"""The rate limit in front of run_ffmpeg's callbacks.
|
||||
|
||||
run_ffmpeg calls `on_progress`/`should_cancel` once per ffmpeg output LINE (`-progress pipe:1`
|
||||
emits a block of ~12 every half second), and both callbacks are database round-trips: `_set_job`
|
||||
opens a session and UPDATE+COMMITs, `_job_status` opens one and SELECTs. Unthrottled, a long
|
||||
re-encode is ~10^5 of each."""
|
||||
from app import worker
|
||||
|
||||
|
||||
class TestProgressWriter:
|
||||
def test_writes_the_first_value_immediately(self, monkeypatch):
|
||||
writes = []
|
||||
monkeypatch.setattr(worker, "_set_job", lambda job_id, **f: writes.append(f))
|
||||
worker._ffmpeg_progress_writer(7, "processing")(12.4)
|
||||
assert writes == [{"progress": 12, "phase": "processing"}]
|
||||
|
||||
def test_a_burst_of_lines_costs_one_write(self, monkeypatch):
|
||||
writes = []
|
||||
monkeypatch.setattr(worker, "_set_job", lambda job_id, **f: writes.append(f))
|
||||
write = worker._ffmpeg_progress_writer(7, "processing")
|
||||
for pct in range(0, 40): # what one second of ffmpeg output looks like
|
||||
write(pct)
|
||||
assert len(writes) == 1
|
||||
|
||||
def test_the_next_second_writes_again(self, monkeypatch):
|
||||
writes = []
|
||||
clock = {"t": 1000.0}
|
||||
monkeypatch.setattr(worker, "_set_job", lambda job_id, **f: writes.append(f))
|
||||
monkeypatch.setattr(worker.time, "monotonic", lambda: clock["t"])
|
||||
write = worker._ffmpeg_progress_writer(7, "processing")
|
||||
write(10)
|
||||
write(11) # same second → dropped
|
||||
clock["t"] += 1.5
|
||||
write(12)
|
||||
assert [w["progress"] for w in writes] == [10, 12]
|
||||
|
||||
def test_an_unchanged_whole_percent_is_never_rewritten(self, monkeypatch):
|
||||
writes = []
|
||||
clock = {"t": 1000.0}
|
||||
monkeypatch.setattr(worker, "_set_job", lambda job_id, **f: writes.append(f))
|
||||
monkeypatch.setattr(worker.time, "monotonic", lambda: clock["t"])
|
||||
write = worker._ffmpeg_progress_writer(7, "processing")
|
||||
write(10.1)
|
||||
clock["t"] += 5
|
||||
write(10.9) # still 10% — the bar shows integers
|
||||
assert [w["progress"] for w in writes] == [10]
|
||||
|
||||
|
||||
class TestCancelPoll:
|
||||
def test_asks_once_then_reuses_the_answer(self, monkeypatch):
|
||||
calls = []
|
||||
monkeypatch.setattr(worker, "_job_status", lambda job_id: calls.append(job_id) or "running")
|
||||
cancelled = worker._ffmpeg_cancel_poll(7)
|
||||
for _ in range(40):
|
||||
assert cancelled() is False
|
||||
assert len(calls) == 1
|
||||
|
||||
def test_a_cancelled_job_is_reported_and_never_queried_again(self, monkeypatch):
|
||||
calls = []
|
||||
clock = {"t": 1000.0}
|
||||
|
||||
def status(job_id):
|
||||
calls.append(job_id)
|
||||
return "cancelled"
|
||||
|
||||
monkeypatch.setattr(worker, "_job_status", status)
|
||||
monkeypatch.setattr(worker.time, "monotonic", lambda: clock["t"])
|
||||
cancelled = worker._ffmpeg_cancel_poll(7)
|
||||
assert cancelled() is True
|
||||
clock["t"] += 60 # cancellation is sticky: no re-query, and no un-cancelling
|
||||
assert cancelled() is True
|
||||
assert len(calls) == 1
|
||||
|
||||
def test_a_cancel_arriving_later_is_seen_on_the_next_tick(self, monkeypatch):
|
||||
clock = {"t": 1000.0}
|
||||
state = {"status": "running"}
|
||||
monkeypatch.setattr(worker, "_job_status", lambda job_id: state["status"])
|
||||
monkeypatch.setattr(worker.time, "monotonic", lambda: clock["t"])
|
||||
cancelled = worker._ffmpeg_cancel_poll(7)
|
||||
assert cancelled() is False
|
||||
state["status"] = "paused"
|
||||
clock["t"] += 1.0 # the throttle must not delay a cancel by more than its interval
|
||||
assert cancelled() is True
|
||||
@@ -51,6 +51,11 @@ class _FakeProc:
|
||||
self.killed = False
|
||||
self.waits = 0
|
||||
|
||||
def poll(self):
|
||||
# Popen.poll(): None while the process is alive. Our fake stays alive until it is stopped —
|
||||
# `_stop` uses this to skip a process that has already gone.
|
||||
return self._rc if self.terminated else None
|
||||
|
||||
def terminate(self):
|
||||
self.terminated = True
|
||||
# A terminated process's pipes hit EOF — that is how the reader threads finish.
|
||||
@@ -132,6 +137,19 @@ def test_a_wedged_process_is_killed_by_the_stall_watchdog(fake_popen):
|
||||
assert proc.terminated # the worker thread is freed, not parked forever
|
||||
|
||||
|
||||
def test_a_callback_exception_still_stops_the_process(fake_popen):
|
||||
# `on_progress` is a database write in production. A failure THERE used to unwind past a live
|
||||
# ffmpeg — orphaning it, still encoding, after a 10s join on its pipes.
|
||||
proc = fake_popen(_FakeProc(_FakeStream(["out_time_us=1000000\n"]), _FakeStream(block=True)))
|
||||
|
||||
def boom(_pct):
|
||||
raise RuntimeError("database is down")
|
||||
|
||||
with pytest.raises(RuntimeError, match="database is down"):
|
||||
edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=boom, should_cancel=lambda: False)
|
||||
assert proc.terminated
|
||||
|
||||
|
||||
def test_a_process_that_ignores_sigterm_is_killed(fake_popen):
|
||||
# `_stop`'s escalation: terminate, wait 5s, then SIGKILL. Without it a cancel could park the
|
||||
# worker thread on a process that simply refuses to go.
|
||||
@@ -150,6 +168,8 @@ def test_a_process_lingering_after_eof_is_stopped_and_reported(fake_popen):
|
||||
with pytest.raises(RuntimeError, match="did not exit"):
|
||||
edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=lambda _p: None, should_cancel=lambda: False)
|
||||
assert proc.terminated
|
||||
# Two waits: the bounded grace wait that timed out, then _stop's own — never an unbounded one.
|
||||
assert proc.waits == 2
|
||||
|
||||
|
||||
def test_stderr_is_drained_while_the_process_runs(fake_popen):
|
||||
|
||||
Reference in New Issue
Block a user