peter
cff8bed62b
Round-2 review of the R1 fixes found that the two-independent-limiter progress writer issued a redundant heartbeat every interval during an ACTIVE download: a percent change writes progress (which already refreshes updated_at, the stall watchdog's liveness source), then the sub-integer calls that follow fire the separate beat budget — ~2 DB writes/interval instead of the throttle's intended 1. Replace the second limiter with a shared `last_touch` clock (last DB write OR beat). Progress writes keep their own `allow_write` budget (so a beat still can't delay a real write — the R1 concern), but the heartbeat now fires only when nothing touched the row for a full interval. Active download → ~1 write/interval; a genuinely stuck re-encode → 1 heartbeat/interval. New regression test: an unchanged-percent call in the same interval as a write does NOT heartbeat. ruff clean, full DB-lane 365 pass.
112 lines
5.4 KiB
Python
112 lines
5.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."""
|
|
import app.downloads.pipeline as pipeline
|
|
|
|
|
|
class TestProgressWriter:
|
|
def test_writes_the_first_value_immediately(self, monkeypatch):
|
|
writes = []
|
|
monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
|
|
pipeline._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(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
|
|
write = pipeline._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(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
|
|
write = pipeline._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(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
|
|
monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: None)
|
|
write = pipeline._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(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
|
|
monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: touches.append(job_id))
|
|
write = pipeline._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
|
|
|
|
def test_no_redundant_heartbeat_when_a_progress_write_just_happened(self, monkeypatch):
|
|
"""A progress write already refreshes updated_at, so an unchanged-percent call in the SAME
|
|
interval must NOT also heartbeat — that would be a redundant DB write (~2x during an active
|
|
download). The heartbeat only fires once no write/beat touched the row for an interval."""
|
|
writes, touches = [], []
|
|
clock = {"t": 1000.0}
|
|
monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
|
|
monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: touches.append(job_id))
|
|
write = pipeline._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
|
|
write(6.0) # progress write at t=1000 → refreshes updated_at
|
|
clock["t"] += 0.3
|
|
write(6.4) # same percent, SAME interval → the write already touched the row, so no heartbeat
|
|
assert [w["progress"] for w in writes] == [6]
|
|
assert touches == []
|
|
|
|
|
|
class TestCancelPoll:
|
|
def test_asks_once_then_reuses_the_answer(self, monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(pipeline, "_job_status", lambda job_id: calls.append(job_id) or "running")
|
|
cancelled = pipeline._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(pipeline, "_job_status", status)
|
|
cancelled = pipeline._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(pipeline, "_job_status", lambda job_id: state["status"])
|
|
cancelled = pipeline._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
|