diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index aed12564..21753e96 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -15,6 +15,7 @@ from app.library.Utils import load_modules from .config import Config from .encoder import Encoder from .Events import Event, EventBus, Events +from .ItemBatcher import ItemBatcher from .ItemDTO import Item LOG = get_logger() @@ -85,7 +86,23 @@ class HttpSocket: self.sio = sio or WebSocketHub(encoder=encoder) self.rootPath: Path = root_path + async def _emit_items(items: list) -> None: + ev = Event(event=Events.ITEM_UPDATED, data=items) + await self.sio.emit(event=Events.ITEM_UPDATED, data=json.loads(encoder.encode(ev))) + + self._batcher = ItemBatcher(emit=_emit_items) + async def event_handler(e: Event, _, **kwargs): + if Events.ITEM_UPDATED == e.event and not kwargs: + await self._batcher.add(e.data) + return + if Events.ITEM_UPDATED == e.event and kwargs: + # Targeted delivery (e.g. to=sid): wrap data in a list so the + # wire shape is always an array for item_updated. + targeted = Event(event=Events.ITEM_UPDATED, data=[e.data]) + payload = json.loads(encoder.encode(targeted)) + await self.sio.emit(event=e.event, data=payload, **kwargs) + return payload = json.loads(encoder.encode(e)) await self.sio.emit(event=e.event, data=payload, **kwargs) @@ -121,6 +138,7 @@ class HttpSocket: async def on_shutdown(self, _: web.Application): LOG.debug("Shutting down socket server.") + await self._batcher.flush() await self.sio.disconnect_all() LOG.debug("Socket server shutdown complete.") diff --git a/app/library/ItemBatcher.py b/app/library/ItemBatcher.py new file mode 100644 index 00000000..26cb0b6f --- /dev/null +++ b/app/library/ItemBatcher.py @@ -0,0 +1,120 @@ +"""Global coalescing batcher for ITEM_UPDATED WebSocket events. + +Collects dirty ItemDTOs keyed by ``_id`` (last write wins) and flushes them +as a single list to an async emit callback at most once per *interval* seconds. + +Leading-edge semantics: the very first ``add()`` after an idle period fires +immediately so single-item updates feel instant. Subsequent adds within the +same interval are coalesced into a single trailing flush. +""" + +from __future__ import annotations + +import asyncio +import time +from collections.abc import Awaitable, Callable +from typing import Any + +from app.library.log import get_logger + +LOG = get_logger() + + +class ItemBatcher: + """Coalescing batcher for WebSocket item-updated payloads. + + Args: + emit: Async callable that receives a list of items and delivers them + to all connected WebSocket clients. + interval: Flush interval in seconds. ``0`` or negative disables + batching – every ``add()`` emits immediately. + + """ + + def __init__(self, emit: Callable[[list], Awaitable[None]], interval: float = 0.5) -> None: + self._emit = emit + self._interval = interval + self._pending: dict[str, Any] = {} + self._handle: asyncio.TimerHandle | None = None + self._last_flush: float = 0.0 # monotonic + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + async def add(self, item: Any) -> None: + """Add (or coalesce) an item into the pending batch. + + If the batcher is idle and the leading-edge condition is met the item + is emitted immediately; otherwise it is queued for the next trailing + flush. + + Args: + item: The ItemDTO (or any object with a ``_id`` attribute) to + batch. + + """ + item_id: str = getattr(item, "_id", None) or str(id(item)) + + # interval <= 0 → always emit immediately, no batching + if self._interval <= 0: + await self._emit_items([item]) + return + + now = time.monotonic() + + # Leading edge: nothing queued yet and enough time has elapsed + if not self._pending and (now - self._last_flush) >= self._interval: + self._last_flush = now + await self._emit_items([item]) + return + + # Coalesce into the pending dict (last write wins per _id) + self._pending[item_id] = item + + # Schedule a trailing flush only if one is not already pending + if self._handle is None: + elapsed = now - self._last_flush + remaining = max(0.0, self._interval - elapsed) + loop = asyncio.get_running_loop() + self._handle = loop.call_later(remaining, lambda: loop.create_task(self._flush_pending())) + + async def flush(self) -> None: + """Cancel any scheduled flush and immediately emit all pending items. + + Intended for use during shutdown so that the last batch reaches clients + before the WebSocket is torn down. + """ + if self._handle is not None: + self._handle.cancel() + self._handle = None + + if self._pending: + await self._flush_pending() + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + async def _flush_pending(self) -> None: + """Emit everything that has accumulated in ``_pending``. + + State is cleared **before** awaiting the emit so that a slow or + failing send can never wedge the batcher. + """ + items = list(self._pending.values()) + self._pending = {} + self._handle = None + self._last_flush = time.monotonic() + + if not items: + return + + try: + await self._emit_items(items) + except Exception: + LOG.exception("ItemBatcher: emit callback raised an exception; %d item(s) lost.", len(items)) + + async def _emit_items(self, items: list) -> None: + """Delegate to the provided emit callback.""" + await self._emit(items) diff --git a/app/tests/test_item_batcher.py b/app/tests/test_item_batcher.py new file mode 100644 index 00000000..c701fb8e --- /dev/null +++ b/app/tests/test_item_batcher.py @@ -0,0 +1,208 @@ +"""Unit tests for app.library.ItemBatcher.""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace + +import pytest + +from app.library.ItemBatcher import ItemBatcher + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def make_item(id_: str = "id1", title: str = "Test") -> SimpleNamespace: + """Return a minimal stand-in for ItemDTO with a ``_id`` attribute.""" + return SimpleNamespace(_id=id_, title=title) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestItemBatcher: + """Tests for ItemBatcher leading-edge and trailing-flush semantics.""" + + @pytest.mark.asyncio + async def test_leading_edge_emits_immediately(self) -> None: + """First add after idle fires immediately as a one-item list.""" + emitted: list[list] = [] + + async def emit(items: list) -> None: + emitted.append(items) + + batcher = ItemBatcher(emit=emit, interval=0.05) + item = make_item("id1") + await batcher.add(item) + + assert len(emitted) == 1 + assert emitted[0] == [item] + + @pytest.mark.asyncio + async def test_rapid_adds_same_id_coalesces_to_last(self) -> None: + """Multiple adds with same _id within interval → one flush, last item wins.""" + emitted: list[list] = [] + + async def emit(items: list) -> None: + emitted.append(items) + + batcher = ItemBatcher(emit=emit, interval=0.1) + + # Consume the leading-edge slot + item_a = make_item("id1", "first") + await batcher.add(item_a) + assert len(emitted) == 1 + + # Now add more within the interval – these should batch + item_b = make_item("id1", "second") + item_c = make_item("id1", "third") + await batcher.add(item_b) + await batcher.add(item_c) + + # Allow the trailing flush to fire + await asyncio.sleep(0.15) + + assert len(emitted) == 2 + flush_items = emitted[1] + assert len(flush_items) == 1 + assert flush_items[0].title == "third" + + @pytest.mark.asyncio + async def test_multiple_different_ids_flushed_together(self) -> None: + """Adds for N different IDs within interval → one flush with N items.""" + emitted: list[list] = [] + + async def emit(items: list) -> None: + emitted.append(items) + + batcher = ItemBatcher(emit=emit, interval=0.1) + + # Consume the leading-edge slot for id1 + await batcher.add(make_item("id1")) + assert len(emitted) == 1 + + # Queue id2, id3 within the interval + item2 = make_item("id2") + item3 = make_item("id3") + await batcher.add(item2) + await batcher.add(item3) + + await asyncio.sleep(0.15) + + assert len(emitted) == 2 + flush_items = emitted[1] + ids = {i._id for i in flush_items} + assert ids == {"id2", "id3"} + + @pytest.mark.asyncio + async def test_only_one_timer_scheduled(self) -> None: + """Only one timer handle is created while a flush is pending.""" + emitted: list[list] = [] + + async def emit(items: list) -> None: + emitted.append(items) + + batcher = ItemBatcher(emit=emit, interval=0.1) + + # Consume the leading edge + await batcher.add(make_item("id1")) + + # Add several more to build up the batch + await batcher.add(make_item("id2")) + handle_after_first = batcher._handle + + await batcher.add(make_item("id3")) + handle_after_second = batcher._handle + + # The same handle object should be reused (not replaced) + assert handle_after_first is handle_after_second + + await asyncio.sleep(0.15) + + @pytest.mark.asyncio + async def test_emit_exception_clears_state_and_allows_next_add(self) -> None: + """A failing emit callback is logged but leaves the batcher functional.""" + emit_count = 0 + + async def emit(items: list) -> None: + nonlocal emit_count + emit_count += 1 + if emit_count == 2: + raise RuntimeError("simulated flush failure") + + batcher = ItemBatcher(emit=emit, interval=0.1) + await batcher.add(make_item("id1")) # leading edge ok (count=1) + + # Trigger trailing flush that will fail (count=2) + await batcher.add(make_item("id2")) + await asyncio.sleep(0.15) # trailing flush fires → raises → caught + + # State must be cleared; next add should work as a fresh leading-edge + await asyncio.sleep(0.05) # ensure _last_flush is old enough + await batcher.add(make_item("id3")) # leading edge ok (count=3) + + assert emit_count == 3 + # Batcher handle must be None (no dangling timer) + assert batcher._handle is None + + @pytest.mark.asyncio + async def test_flush_cancels_timer_and_emits_pending(self) -> None: + """flush() cancels a scheduled timer and emits remaining items.""" + emitted: list[list] = [] + + async def emit(items: list) -> None: + emitted.append(items) + + batcher = ItemBatcher(emit=emit, interval=10.0) # very long interval + + # Consume leading edge + await batcher.add(make_item("id1")) + assert len(emitted) == 1 + + # Queue an item that would normally wait 10 s + item2 = make_item("id2") + await batcher.add(item2) + assert batcher._handle is not None + + # flush() must fire immediately + await batcher.flush() + + assert batcher._handle is None + assert len(emitted) == 2 + assert emitted[1] == [item2] + + @pytest.mark.asyncio + async def test_flush_noop_when_empty(self) -> None: + """flush() on an idle batcher does not call emit.""" + emitted: list[list] = [] + + async def emit(items: list) -> None: + emitted.append(items) + + batcher = ItemBatcher(emit=emit, interval=0.05) + await batcher.flush() + + assert emitted == [] + + @pytest.mark.asyncio + async def test_interval_zero_emits_every_add_immediately(self) -> None: + """interval=0 → every add emits immediately as a single-item list.""" + emitted: list[list] = [] + + async def emit(items: list) -> None: + emitted.append(items) + + batcher = ItemBatcher(emit=emit, interval=0) + + for i in range(5): + await batcher.add(make_item(f"id{i}")) + + assert len(emitted) == 5 + for i, batch in enumerate(emitted): + assert len(batch) == 1 + assert batch[0]._id == f"id{i}" diff --git a/ui/app/composables/useAppSocket.ts b/ui/app/composables/useAppSocket.ts index 6d562565..e5a90447 100644 --- a/ui/app/composables/useAppSocket.ts +++ b/ui/app/composables/useAppSocket.ts @@ -336,10 +336,10 @@ on('item_deleted', (data: WSEP['item_deleted']) => { on('item_updated', (data: WSEP['item_updated']) => { const queueState = getQueueState(); - const id = data.data._id; - - if (true === queueState.has(id)) { - queueState.update(id, data.data); + for (const item of data.data) { + if (true === queueState.has(item._id)) { + queueState.update(item._id, item); + } } }); diff --git a/ui/app/types/sockets.d.ts b/ui/app/types/sockets.d.ts index 23851cf6..3adb3a29 100644 --- a/ui/app/types/sockets.d.ts +++ b/ui/app/types/sockets.d.ts @@ -22,7 +22,7 @@ export type WSEP = { connect_error: { message?: string }; connected: EventPayload<{ sid: string }>; item_added: EventPayload; - item_updated: EventPayload; + item_updated: EventPayload; item_cancelled: EventPayload; item_deleted: EventPayload; item_bulk_deleted: EventPayload<{ count: number; status?: string; ids?: string[] }>;