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

Add time-based throttling to StatusTracker to reduce item_updated websocket events

Browse source 
Co-authored-by: jbatesy <51190172+jbatesy@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot] 2026-03-09 05:28:17 +00:00
parent d03bab3cd9
commit c32c6a2054
2 changed files with 110 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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:

74
app/tests/test_download.py
View file

@ -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"