peter
e2362e1ed1
The supervisor kills a job that writes nothing for download_stall_timeout_seconds (150s) and labels it a "network" stall. But the ffmpeg progress writer only wrote on an integer-percent CHANGE, so a re-encode that sits on one percent for minutes (AV1/VP9 -> H.264 or a crop on the CPU-only prod host) produced no row write and was killed mid-encode as "stalled" — and the retry stalled the same way. Fix: the writer now heartbeats (`_touch_job`, bumping updated_at without touching the progress bar) once per interval even when the whole percent is unchanged, so a live-but-quiet re-encode keeps a fresh updated_at. Reorders the rate-limit gate first so a burst still costs one DB op — either a progress write or a heartbeat. Pure-logic test with an injected clock: three stuck-at-10% intervals write no new progress but heartbeat each time. Full lane green, ruff clean.
97 lines
4.4 KiB
Python
97 lines
4.4 KiB
Python
"""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))
|
|
write = worker._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
|
|
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, "_touch_job", lambda job_id: None)
|
|
write = worker._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
|
|
write(10.1)
|
|
clock["t"] += 5
|
|
write(10.9) # still 10% — the bar shows integers
|
|
assert [w["progress"] for w in writes] == [10]
|
|
|
|
def test_a_slow_same_percent_stretch_heartbeats_the_watchdog(self, monkeypatch):
|
|
"""B-sub-4: a re-encode that sits on one integer percent for minutes must NOT be killed as
|
|
stalled — the writer touches the row (heartbeat) each interval even though it writes no new
|
|
progress, so the stall watchdog keeps seeing a fresh updated_at."""
|
|
writes, touches = [], []
|
|
clock = {"t": 1000.0}
|
|
monkeypatch.setattr(worker, "_set_job", lambda job_id, **f: writes.append(f))
|
|
monkeypatch.setattr(worker, "_touch_job", lambda job_id: touches.append(job_id))
|
|
write = worker._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
|
|
write(10.1) # first real write → progress 10
|
|
for _ in range(3): # three more intervals stuck on 10%
|
|
clock["t"] += 1.5
|
|
write(10.4)
|
|
assert [w["progress"] for w in writes] == [10] # progress never rewritten
|
|
assert touches == [7, 7, 7] # …but each interval heartbeats the watchdog
|
|
|
|
|
|
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._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)
|
|
cancelled = worker._cancel_poll(7, now=lambda: clock["t"])
|
|
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"])
|
|
cancelled = worker._cancel_poll(7, now=lambda: clock["t"])
|
|
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
|