danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

perf: throttle item_updated WebSocket events to 500ms intervals

Browse source 
yt-dlp fires progress hooks faster than the browser can usefully render
them. Previously every callback triggered a WebSocket emit → Vue
reactivity update → DOM re-render, causing unnecessary main-thread work
during active downloads.

Now status changes (e.g. downloading → finished) are always emitted
immediately, while progress-only updates are capped at one per 500ms
per download. A flush_pending() call at loop termination ensures the
final state is never dropped.
This commit is contained in:
Jesse Bate 2026-06-10 19:23:45 +09:30
parent d6318fca0f
commit e0dbc2d15d
2 changed files with 92 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
app/library/downloads/status_tracker.py
View file

@ -56,6 +56,26 @@ class StatusTracker:
self._terminator_sent: bool = False
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:
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:
if self._pending_emit:
self._emit_update(force=True)
async def _finalize_file(self, filepath: Path) -> None:
"""
@ -235,7 +255,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:
"""
@ -248,6 +268,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):
@ -295,6 +316,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:

68
app/tests/test_download.py
View file

@ -1211,6 +1211,74 @@ class TestStatusTracker:
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))
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})
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))
await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 100})
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})
before = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED]
assert len(before) == 1, "Second update should be throttled"
st.flush_pending()
after = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED]
assert len(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_disabled_when_interval_is_zero(self, mock_config: dict) -> None:
st = StatusTracker(**mock_config)
st._emit_interval = 0.0 # disable throttle
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"
class TestQueueManager:
@staticmethod