diff --git a/backend/app/downloads/pipeline.py b/backend/app/downloads/pipeline.py index 72eafc7b..ae61a6c9 100644 --- a/backend/app/downloads/pipeline.py +++ b/backend/app/downloads/pipeline.py @@ -196,26 +196,30 @@ def _ffmpeg_progress_writer(job_id: int, phase: str, *, now=time.monotonic): """`on_progress` for run_ffmpeg: persist a progress row write at most once per interval and only when the whole-percent value changed (the bar shows integers), plus a separate once-per-interval heartbeat while the percent is stuck (so a slow re-encode isn't mistaken for a stall).""" - # Two independent 1/interval budgets so a heartbeat never consumes the progress-write slot — a - # percent change right after a heartbeat must still write immediately, not wait out the interval. + # Progress writes have their OWN 1/interval budget so a heartbeat can never delay a real write (the + # round-1 concern). The heartbeat, in turn, is gated on `last_touch` — the last DB write OR beat — + # so a progress write (which already refreshes updated_at) suppresses a redundant heartbeat that + # same interval, keeping active downloads at ~1 write/interval, not two. allow_write = _rate_limiter(_DB_WRITE_INTERVAL_S, now=now) - allow_beat = _rate_limiter(_DB_WRITE_INTERVAL_S, now=now) state = {"pct": -1} + last_touch: dict = {"t": None} # monotonic time of the last DB write/heartbeat def write(pct: float) -> None: whole = int(pct) - if whole == state["pct"]: - # Same integer percent, but ffmpeg IS still working — a slow re-encode (AV1/VP9 -> H.264 - # or a crop on the CPU-only host) can sit on one percent for minutes. Heartbeat the row's - # updated_at (on its OWN budget) so the stall watchdog doesn't kill a live job as "stalled" - # (B-sub-4), without rewriting the unchanged bar or delaying the next real write. - if allow_beat(): - _touch_job(job_id) + if whole != state["pct"]: + if allow_write(): + state["pct"] = whole + _set_job(job_id, progress=whole, phase=phase) + last_touch["t"] = now() return - if not allow_write(): - return - state["pct"] = whole - _set_job(job_id, progress=whole, phase=phase) + # Same integer percent, but ffmpeg IS still working — a slow re-encode (AV1/VP9 -> H.264 or a + # crop on the CPU-only host) can sit on one percent for minutes. Heartbeat updated_at so the + # stall watchdog doesn't kill a live job (B-sub-4) — but only if no write/beat already touched + # the row this interval, so a progressing download doesn't pay a redundant write. + t = now() + if last_touch["t"] is None or t - last_touch["t"] >= _DB_WRITE_INTERVAL_S: + _touch_job(job_id) + last_touch["t"] = t return write diff --git a/backend/tests/test_ffmpeg_callbacks.py b/backend/tests/test_ffmpeg_callbacks.py index b351f03c..069e70bc 100644 --- a/backend/tests/test_ffmpeg_callbacks.py +++ b/backend/tests/test_ffmpeg_callbacks.py @@ -60,6 +60,21 @@ class TestProgressWriter: 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):