From f1365f6fba84cd64baa86f1a3cffea50f9e042df Mon Sep 17 00:00:00 2001 From: Jesse Bate Date: Wed, 10 Jun 2026 23:47:25 +0930 Subject: [PATCH] perf: extend WebSocket batching to item_moved events Bursts of completions (e.g. small playlist entries) fire one item_moved frame per item. Route item_moved through the same ItemBatcher mechanism as item_updated: one frame per 500ms tick, payload always a list. ItemBatcher gains an optional key callable since item_moved payloads are {"to", "preset", "item"} dicts rather than ItemDTOs; moves coalesce by item _id with last write wins, so an item that moves twice within a tick delivers only its final destination. The notifications service consumes ITEM_MOVED from the event bus directly and is unaffected. BREAKING CHANGE: the item_moved WebSocket event payload (data) is now always an array of {to, preset, item} objects instead of a single one. --- app/library/HttpSocket.py | 32 +++++++++++++++++--------- app/library/ItemBatcher.py | 22 ++++++++++++------ app/tests/test_item_batcher.py | 33 +++++++++++++++++++++++++++ ui/app/composables/useAppSocket.ts | 18 ++++++++------- ui/app/composables/useHistoryState.ts | 8 +++++-- ui/app/types/sockets.d.ts | 2 +- 6 files changed, 86 insertions(+), 29 deletions(-) diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index 21753e96..9f47cfb8 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -86,20 +86,29 @@ 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))) + def _make_emit(event_name: str): + async def _emit_items(items: list) -> None: + ev = Event(event=event_name, data=items) + await self.sio.emit(event=event_name, data=json.loads(encoder.encode(ev))) - self._batcher = ItemBatcher(emit=_emit_items) + return _emit_items + + self._batchers: dict[str, ItemBatcher] = { + Events.ITEM_UPDATED: ItemBatcher(emit=_make_emit(Events.ITEM_UPDATED)), + Events.ITEM_MOVED: ItemBatcher( + emit=_make_emit(Events.ITEM_MOVED), + key=lambda d: getattr(d.get("item"), "_id", None) if isinstance(d, dict) else None, + ), + } 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: + if batcher := self._batchers.get(e.event): + if not kwargs: + await batcher.add(e.data) + return # 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]) + # wire shape is always an array for batched events. + targeted = Event(event=e.event, data=[e.data]) payload = json.loads(encoder.encode(targeted)) await self.sio.emit(event=e.event, data=payload, **kwargs) return @@ -138,7 +147,8 @@ class HttpSocket: async def on_shutdown(self, _: web.Application): LOG.debug("Shutting down socket server.") - await self._batcher.flush() + for batcher in self._batchers.values(): + await 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 index 26cb0b6f..d546683d 100644 --- a/app/library/ItemBatcher.py +++ b/app/library/ItemBatcher.py @@ -1,6 +1,6 @@ -"""Global coalescing batcher for ITEM_UPDATED WebSocket events. +"""Global coalescing batcher for high-frequency WebSocket item events. -Collects dirty ItemDTOs keyed by ``_id`` (last write wins) and flushes them +Collects dirty items keyed per item (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 @@ -21,19 +21,27 @@ LOG = get_logger() class ItemBatcher: - """Coalescing batcher for WebSocket item-updated payloads. + """Coalescing batcher for WebSocket item event 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. + key: Optional callable that extracts the coalescing key from an item. + Defaults to the item's ``_id`` attribute. """ - def __init__(self, emit: Callable[[list], Awaitable[None]], interval: float = 0.5) -> None: + def __init__( + self, + emit: Callable[[list], Awaitable[None]], + interval: float = 0.5, + key: Callable[[Any], str | None] | None = None, + ) -> None: self._emit = emit self._interval = interval + self._key = key self._pending: dict[str, Any] = {} self._handle: asyncio.TimerHandle | None = None self._last_flush: float = 0.0 # monotonic @@ -50,11 +58,11 @@ class ItemBatcher: flush. Args: - item: The ItemDTO (or any object with a ``_id`` attribute) to - batch. + item: The payload to batch. Without a ``key`` callable, this is + expected to expose a ``_id`` attribute. """ - item_id: str = getattr(item, "_id", None) or str(id(item)) + item_id: str = (self._key(item) if self._key else getattr(item, "_id", None)) or str(id(item)) # interval <= 0 → always emit immediately, no batching if self._interval <= 0: diff --git a/app/tests/test_item_batcher.py b/app/tests/test_item_batcher.py index c701fb8e..f07edbdf 100644 --- a/app/tests/test_item_batcher.py +++ b/app/tests/test_item_batcher.py @@ -189,6 +189,39 @@ class TestItemBatcher: assert emitted == [] + @pytest.mark.asyncio + async def test_custom_key_coalesces_dict_payloads(self) -> None: + """A key callable extracts the coalescing key from non-ItemDTO payloads.""" + emitted: list[list] = [] + + async def emit(items: list) -> None: + emitted.append(items) + + batcher = ItemBatcher( + emit=emit, + interval=0.1, + key=lambda d: getattr(d.get("item"), "_id", None) if isinstance(d, dict) else None, + ) + + # Consume the leading-edge slot + await batcher.add({"to": "history", "item": make_item("id1")}) + assert len(emitted) == 1 + + # Two moves for the same item within the interval → last wins + await batcher.add({"to": "history", "item": make_item("id2", "first")}) + await batcher.add({"to": "queue", "item": make_item("id2", "second")}) + await batcher.add({"to": "history", "item": make_item("id3")}) + + await asyncio.sleep(0.15) + + assert len(emitted) == 2 + flush_items = emitted[1] + assert len(flush_items) == 2 + by_id = {moved["item"]._id: moved for moved in flush_items} + assert set(by_id) == {"id2", "id3"} + assert by_id["id2"]["to"] == "queue" + assert by_id["id2"]["item"].title == "second" + @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.""" diff --git a/ui/app/composables/useAppSocket.ts b/ui/app/composables/useAppSocket.ts index e5a90447..3a803f6e 100644 --- a/ui/app/composables/useAppSocket.ts +++ b/ui/app/composables/useAppSocket.ts @@ -345,16 +345,18 @@ on('item_updated', (data: WSEP['item_updated']) => { on('item_moved', (data: WSEP['item_moved']) => { const queueState = getQueueState(); - const to = data.data.to; - const id = data.data.item._id; - if ('queue' === to) { - queueState.add(id, data.data.item); - } + for (const moved of data.data) { + const id = moved.item._id; - if ('history' === to) { - if (true === queueState.has(id)) { - queueState.remove(id); + if ('queue' === moved.to) { + queueState.add(id, moved.item); + } + + if ('history' === moved.to) { + if (true === queueState.has(id)) { + queueState.remove(id); + } } } }); diff --git a/ui/app/composables/useHistoryState.ts b/ui/app/composables/useHistoryState.ts index c10781a6..f48ef5bd 100644 --- a/ui/app/composables/useHistoryState.ts +++ b/ui/app/composables/useHistoryState.ts @@ -231,11 +231,15 @@ const moveHandler = ( shouldHandle: () => boolean = () => isLoaded.value, ): ((payload: WSEP['item_moved']) => void) => { return (payload: WSEP['item_moved']): void => { - if ('history' !== payload.data.to || !shouldHandle()) { + if (!shouldHandle()) { return; } - upsert(payload.data.item); + for (const moved of payload.data) { + if ('history' === moved.to) { + upsert(moved.item); + } + } }; }; diff --git a/ui/app/types/sockets.d.ts b/ui/app/types/sockets.d.ts index 3adb3a29..fb3fee41 100644 --- a/ui/app/types/sockets.d.ts +++ b/ui/app/types/sockets.d.ts @@ -26,7 +26,7 @@ export type WSEP = { item_cancelled: EventPayload; item_deleted: EventPayload; item_bulk_deleted: EventPayload<{ count: number; status?: string; ids?: string[] }>; - item_moved: EventPayload<{ to: 'queue' | 'history'; item: StoreItem }>; + item_moved: EventPayload>; item_status: EventPayload<{ status?: string; msg?: string; preset?: string }>; paused: EventPayload<{ paused?: boolean }>; resumed: EventPayload<{ paused?: boolean }>;