diff --git a/app/library/downloads/status_tracker.py b/app/library/downloads/status_tracker.py index edd248d8..e4805c05 100644 --- a/app/library/downloads/status_tracker.py +++ b/app/library/downloads/status_tracker.py @@ -57,6 +57,38 @@ class StatusTracker: self._candidate_filepath: Path | None = None self.update_task: asyncio.Task | None = None + self._last_emit_time: float = 0.0 + self._last_status: str | None = None + self._pending_emit: bool = False + self._emit_interval: float = 0.5 + + def _emit_update(self, force: bool = False) -> None: + """ + Emit an item_updated event with time-based throttling. + + Progress-only updates are throttled to at most once per ``_emit_interval`` + seconds. Status changes are always emitted immediately. + + Args: + force: If True, emit regardless of throttle interval. + + """ + now = time.monotonic() + status_changed = self.info.status != self._last_status + + if force or status_changed or (now - self._last_emit_time) >= self._emit_interval: + self._notify.emit(Events.ITEM_UPDATED, data=self.info) + self._last_emit_time = now + self._last_status = self.info.status + self._pending_emit = False + else: + self._pending_emit = True + + def flush_pending(self) -> None: + """Emit any pending throttled update so the final state is sent.""" + if self._pending_emit: + self._emit_update(force=True) + async def _finalize_file(self, filepath: Path) -> None: """ Set filename, file_size, and run ffprobe on completed file. @@ -164,7 +196,7 @@ class StatusTracker: await self._finalize_file(Path(final_name)) self.info.status = "finished" - self._notify.emit(Events.ITEM_UPDATED, data=self.info) + self._emit_update() async def progress_update(self) -> None: """ @@ -177,6 +209,7 @@ class StatusTracker: self.update_task = asyncio.get_running_loop().run_in_executor(None, self.status_queue.get) status = await self.update_task if status is None or isinstance(status, Terminator): + self.flush_pending() return await self.process_status_update(status) except (asyncio.CancelledError, OSError, FileNotFoundError, EOFError, BrokenPipeError, ConnectionError): @@ -212,6 +245,8 @@ class StatusTracker: except (queue.Empty, BrokenPipeError, ConnectionRefusedError, EOFError, OSError): continue + self.flush_pending() + def cancel_update_task(self) -> None: """Cancel the progress update task if it's running.""" try: diff --git a/app/tests/test_download.py b/app/tests/test_download.py index f32a7d8a..df61fd17 100644 --- a/app/tests/test_download.py +++ b/app/tests/test_download.py @@ -866,3 +866,77 @@ class TestStatusTracker: st.put_terminator() assert 1 == len(queue.items), "Should add terminator to queue" assert isinstance(queue.items[0], Terminator), "Should add Terminator instance" + + @pytest.mark.asyncio + async def test_throttle_skips_rapid_progress_updates(self, mock_config: dict) -> None: + st = StatusTracker(**mock_config) + emit_calls: list = [] + st._notify = Mock() + st._notify.emit = Mock(side_effect=lambda *a, **kw: emit_calls.append(a)) + + # First update: always emitted (time-based: _last_emit_time starts at 0) + await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 100}) + # Second update: should be throttled (within _emit_interval) + await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 200}) + # Third update: also throttled + await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 300}) + + item_updated_calls = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED] + assert len(item_updated_calls) == 1, "Should throttle rapid progress updates to one emission" + assert st._pending_emit is True, "Should mark pending emit for throttled updates" + + @pytest.mark.asyncio + async def test_throttle_always_emits_on_status_change(self, mock_config: dict) -> None: + st = StatusTracker(**mock_config) + emit_calls: list = [] + st._notify = Mock() + st._notify.emit = Mock(side_effect=lambda *a, **kw: emit_calls.append(a)) + + # First update with status "downloading" + await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 100}) + # Second update with a different status — should emit immediately + await st.process_status_update({"id": "test-id", "status": "error", "error": "fail"}) + + item_updated_calls = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED] + assert len(item_updated_calls) == 2, "Should emit immediately when status changes" + + @pytest.mark.asyncio + async def test_flush_pending_sends_throttled_update(self, mock_config: dict) -> None: + st = StatusTracker(**mock_config) + emit_calls: list = [] + st._notify = Mock() + st._notify.emit = Mock(side_effect=lambda *a, **kw: emit_calls.append(a)) + + await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 100}) + await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 200}) + + item_updated_before = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED] + assert len(item_updated_before) == 1, "Second update should be throttled" + + st.flush_pending() + + item_updated_after = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED] + assert len(item_updated_after) == 2, "flush_pending should emit the throttled update" + assert st._pending_emit is False, "Should clear pending flag after flush" + + def test_flush_pending_noop_when_no_pending(self, mock_config: dict) -> None: + st = StatusTracker(**mock_config) + st._notify = Mock() + st._notify.emit = Mock() + + st.flush_pending() + st._notify.emit.assert_not_called() + + @pytest.mark.asyncio + async def test_throttle_emits_after_interval(self, mock_config: dict) -> None: + st = StatusTracker(**mock_config) + st._emit_interval = 0.0 # disable throttle to allow all emissions + emit_calls: list = [] + st._notify = Mock() + st._notify.emit = Mock(side_effect=lambda *a, **kw: emit_calls.append(a)) + + await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 100}) + await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 200}) + + item_updated_calls = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED] + assert len(item_updated_calls) == 2, "Should emit all updates when interval is 0"