diff --git a/backend/app/downloads/edit.py b/backend/app/downloads/edit.py index 97f102b..99235ae 100644 --- a/backend/app/downloads/edit.py +++ b/backend/app/downloads/edit.py @@ -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) diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index bb41b0e..f4f1109 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -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: diff --git a/backend/app/worker.py b/backend/app/worker.py index e319a55..faf5086 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -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") diff --git a/backend/tests/test_ffmpeg_callbacks.py b/backend/tests/test_ffmpeg_callbacks.py new file mode 100644 index 0000000..c2644f5 --- /dev/null +++ b/backend/tests/test_ffmpeg_callbacks.py @@ -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 diff --git a/backend/tests/test_run_ffmpeg.py b/backend/tests/test_run_ffmpeg.py index 5dec4dc..d72bde2 100644 --- a/backend/tests/test_run_ffmpeg.py +++ b/backend/tests/test_run_ffmpeg.py @@ -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): diff --git a/frontend/src/components/PlayerModal.tsx b/frontend/src/components/PlayerModal.tsx index 34f175b..ab340e1 100644 --- a/frontend/src/components/PlayerModal.tsx +++ b/frontend/src/components/PlayerModal.tsx @@ -35,6 +35,7 @@ import { } from "../lib/format"; import { renderDescription } from "../lib/descriptionLinks"; import { useBackToClose } from "../lib/history"; +import { DEFAULT_VOLUME, normalizeVolume, stepVolume } from "../lib/playerVolume"; import { LS, readAccountMerged, writeAccount } from "../lib/storage"; import { useScrollFade } from "../lib/useScrollFade"; @@ -188,12 +189,12 @@ export default function PlayerModal({ // (the on-player flash has its own `volumeUi` state), the once-bound wheel/key listeners need a // ref anyway, and the hook's write lives inside a `setState` updater — which is both a second // write on every settle and not something React must run while the component is unmounting. - // `readAccountMerged` (not readAccount) so a partial/older stored object still gets the default - // — a missing `volume` would otherwise travel as undefined into the arithmetic below. Read via - // a lazy `useState` initializer so it happens once, not on every render (the house pattern — - // see the frozen queue above). - const [storedVolume] = useState( - () => readAccountMerged(LS.ytPlayerPrefs, { volume: 100 }).volume, + // Sanitised HERE, at the boundary where the untrusted value enters (`normalizeVolume`, see + // lib/playerVolume): a stored `null` / `"80"` / missing key becomes the default once, instead of + // travelling into the arithmetic. Read via a lazy `useState` initializer so it happens once, not + // on every render (the house pattern — see the frozen queue above). + const [storedVolume] = useState(() => + normalizeVolume(readAccountMerged(LS.ytPlayerPrefs, { volume: DEFAULT_VOLUME }).volume), ); const volumeRef = useRef(storedVolume); const volSaveTimerRef = useRef(undefined); @@ -227,10 +228,10 @@ export default function PlayerModal({ }; /** Push a level (0–100) to the player and remember it; 0 also mutes (a player left at zero * volume but unmuted shows YouTube's speaker icon as ON, which reads as broken). */ - const applyVolume = (level: number, flash = true) => { - // A non-finite level (a corrupt stored value, a player that answered NaN) must not become a - // clamped NaN: that would mute the player AND persist `null`, poisoning every later session. - const next = Number.isFinite(level) ? Math.max(0, Math.min(100, Math.round(level))) : 100; + const applyVolume = (level: unknown, flash = true) => { + // An unusable level (a player that answered NaN) falls back to the LAST GOOD one, never to the + // default: jumping a quiet session to full blast is the surprise this feature exists to avoid. + const next = normalizeVolume(level, volumeRef.current); const p = playerRef.current; if (p && typeof p.setVolume === "function") { if (next > 0) p.unMute?.(); @@ -255,12 +256,8 @@ export default function PlayerModal({ const now = Date.now(); const continuing = now - lastNudgeAtRef.current < 1000; lastNudgeAtRef.current = now; - const cur = continuing - ? volumeRef.current - : p.isMuted?.() - ? 0 - : (p.getVolume?.() ?? volumeRef.current); - applyVolume(cur + delta); + const cur = continuing ? volumeRef.current : p.isMuted?.() ? 0 : p.getVolume?.(); + applyVolume(stepVolume(cur, delta, volumeRef.current), true); }; // The player can navigate to other videos (YouTube links in a description). The diff --git a/frontend/src/lib/playerVolume.test.ts b/frontend/src/lib/playerVolume.test.ts new file mode 100644 index 0000000..0915eb9 --- /dev/null +++ b/frontend/src/lib/playerVolume.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, it } from "vitest"; + +import { DEFAULT_VOLUME, normalizeVolume, stepVolume } from "./playerVolume"; + +describe("normalizeVolume", () => { + it("keeps a valid level", () => { + expect(normalizeVolume(0)).toBe(0); + expect(normalizeVolume(37)).toBe(37); + expect(normalizeVolume(100)).toBe(100); + }); + + it("clamps out-of-range levels and rounds to whole percents", () => { + expect(normalizeVolume(-20)).toBe(0); + expect(normalizeVolume(150)).toBe(100); + expect(normalizeVolume(42.6)).toBe(43); + }); + + it.each([ + ["a missing key", undefined], + ["an explicit null", null], + ["a stringified number", "80"], + ["NaN from a broken player", NaN], + ["a whole object", { volume: 50 }], + ])("falls back on %s", (_label, raw) => { + // The regression: a partial stored object made the level undefined, which Math.round turned + // into NaN — muting playback and persisting `null` for every later session. + expect(normalizeVolume(raw, 15)).toBe(15); + }); + + it("prefers the last known level over the default, so a quiet session stays quiet", () => { + expect(normalizeVolume(NaN, 15)).toBe(15); + expect(normalizeVolume(NaN, 15)).not.toBe(DEFAULT_VOLUME); + }); + + it("uses the default when the fallback is unusable too", () => { + expect(normalizeVolume(NaN, NaN)).toBe(DEFAULT_VOLUME); + expect(normalizeVolume(null, undefined as unknown as number)).toBe(DEFAULT_VOLUME); + }); +}); + +describe("stepVolume", () => { + it("steps up and down from the current level", () => { + expect(stepVolume(60, 5, 100)).toBe(65); + expect(stepVolume(60, -5, 100)).toBe(55); + }); + + it("clamps at both ends instead of wrapping", () => { + expect(stepVolume(98, 5, 100)).toBe(100); + expect(stepVolume(3, -5, 100)).toBe(0); + }); + + it("steps from the fallback when the player answers garbage", () => { + // A spin must not lose notches to a stale/unusable reading — it continues from what we last set. + expect(stepVolume(undefined, -5, 90)).toBe(85); + expect(stepVolume(NaN, -5, 90)).toBe(85); + }); +}); diff --git a/frontend/src/lib/playerVolume.ts b/frontend/src/lib/playerVolume.ts new file mode 100644 index 0000000..552bc2a --- /dev/null +++ b/frontend/src/lib/playerVolume.ts @@ -0,0 +1,30 @@ +// The YouTube player's volume arithmetic, kept out of PlayerModal so it can be unit-tested. Every +// rule here was established by a bug: a wheel spin that lost notches to the iframe's lagging +// getVolume, a partial stored object that became NaN and persisted as `null`, and a "corrupt → +// default" fallback that jumped an already-quiet session to full blast. + +/** Level used when nothing trustworthy is available. */ +export const DEFAULT_VOLUME = 100; + +const clamp = (n: number) => Math.max(0, Math.min(100, Math.round(n))); + +/** + * Coerce anything that claims to be a volume into a usable 0–100 level. + * + * Only a real, finite `number` is accepted: the stored blob is JSON, so a level is a number or it + * is corrupt — `null`, `undefined`, `"80"` and `NaN` all mean "we don't know", and `Number(null)` + * being `0` (silently muting the player) is exactly the coercion to avoid. `fallback` is the last + * level known to work; it is itself validated, so a corrupt value can never travel through it. + */ +export function normalizeVolume(raw: unknown, fallback: number = DEFAULT_VOLUME): number { + if (typeof raw === "number" && Number.isFinite(raw)) return clamp(raw); + if (typeof fallback === "number" && Number.isFinite(fallback)) return clamp(fallback); + return DEFAULT_VOLUME; +} + +/** One volume step (wheel notch / arrow key) from `current`, falling back to `fallback` when the + * player answered something unusable. */ +export function stepVolume(current: unknown, delta: number, fallback: number): number { + const base = normalizeVolume(current, fallback); + return clamp(base + delta); +}