From 64af066a116dd965a9df74bf680e3ada54af9924 Mon Sep 17 00:00:00 2001 From: Lincoln Stein Date: Mon, 29 Jun 2026 12:42:50 -0400 Subject: [PATCH] fix(multiuser): keep queue badge & progress bar live across users MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In multiuser mode, queue item status changes and batch-enqueued events are emitted only to the owning user's socket room (for privacy). The frontend badge and progress bar only refetch the global queue status in response to those events, so a non-admin only learns about queue changes from their *own* jobs. Once their jobs finish they get no further signal: the global total (the "/Y" in "X/Y") and the in_progress count freeze, leaving the badge wrong and the progress bar animating indefinitely. Broadcast a content-free `queue_counts_changed` event to the whole queue room whenever counts change (a status transition or an enqueue). It carries only the queue_id — no user_id, batch_id, session_id, or counts — so every subscriber can refetch GET /queue/{id}/status, which already redacts per-user data. Nothing private is leaked, and it never fires on high-frequency invocation progress events. Co-Authored-By: Claude Opus 4.8 (1M context) --- invokeai/app/api/sockets.py | 23 +++++++++++++++++++ .../src/services/events/setEventListeners.tsx | 10 ++++++++ .../frontend/web/src/services/events/types.ts | 5 ++++ 3 files changed, 38 insertions(+) diff --git a/invokeai/app/api/sockets.py b/invokeai/app/api/sockets.py index 7e93c332d64..144a1e9a2c0 100644 --- a/invokeai/app/api/sockets.py +++ b/invokeai/app/api/sockets.py @@ -260,6 +260,21 @@ async def _handle_sub_bulk_download(self, sid: str, data: Any) -> None: async def _handle_unsub_bulk_download(self, sid: str, data: Any) -> None: await self._sio.leave_room(sid, BulkDownloadSubscriptionEvent(**data).bulk_download_id) + async def _broadcast_queue_counts_changed(self, queue_id: str): + """Broadcast a content-free signal that the queue's aggregate counts may have changed. + + The detailed queue item events (status changes, enqueues) are private to their owner + and admins, so non-admin subscribers never learn when *other* users' jobs are queued, + started, or finished. Without that signal their badge's global total — and the + in_progress count that drives the progress animation — go stale and get stuck once + their own jobs finish. + + This event carries nothing but the queue_id: no user_id, batch_id, session_id, or + counts. Every subscriber simply refetches GET /queue/{queue_id}/status, which already + redacts per-user data, so broadcasting it to the whole queue room leaks nothing. + """ + await self._sio.emit(event="queue_counts_changed", data={"queue_id": queue_id}, room=queue_id) + async def _handle_queue_event(self, event: FastAPIEvent[QueueEventBase]): """Handle queue events with user isolation. @@ -313,6 +328,10 @@ async def _handle_queue_event(self, event: FastAPIEvent[QueueEventBase]): logger.debug(f"Emitted private queue item event {event_name} to user room {user_room} and admin room") + # A status change shifts the global pending/in_progress totals, so nudge every + # subscriber to refetch their (redacted) queue status — see _broadcast_queue_counts_changed. + await self._broadcast_queue_counts_changed(event_data.queue_id) + # RecallParametersUpdatedEvent is private - only emit to owner + admins. # # Emit to the union of the owner room and the admin room in a SINGLE @@ -340,6 +359,10 @@ async def _handle_queue_event(self, event: FastAPIEvent[QueueEventBase]): await self._sio.emit(event=event_name, data=event_data.model_dump(mode="json"), room="admin") logger.debug(f"Emitted private batch_enqueued event to user room {user_room} and admin room") + # Newly enqueued items raise the global total, so nudge every subscriber to + # refetch their (redacted) queue status — see _broadcast_queue_counts_changed. + await self._broadcast_queue_counts_changed(event_data.queue_id) + else: # For remaining queue events (e.g. QueueClearedEvent) that do not # carry user identity, emit to all subscribers in the queue room. diff --git a/invokeai/frontend/web/src/services/events/setEventListeners.tsx b/invokeai/frontend/web/src/services/events/setEventListeners.tsx index e6010ce4ca1..05b23e74428 100644 --- a/invokeai/frontend/web/src/services/events/setEventListeners.tsx +++ b/invokeai/frontend/web/src/services/events/setEventListeners.tsx @@ -451,6 +451,16 @@ export const setEventListeners = ({ socket, store, setIsConnected }: SetEventLis } }); + socket.on('queue_counts_changed', (data) => { + // A content-free broadcast that the queue's global counts may have changed — typically + // because *another* user enqueued or one of their jobs changed status. We never receive + // those users' private queue item events, so this is the only signal that lets a non-admin's + // badge keep its global total (the "/Y" in "X/Y") and the in_progress-driven progress bar + // current. Only the redacted SessionQueueStatus is refetched; no per-user/batch caches. + log.trace({ data }, 'Queue counts changed'); + dispatch(queueApi.util.invalidateTags(['SessionQueueStatus'])); + }); + socket.on('queue_cleared', (data) => { log.debug({ data }, 'Queue cleared'); dispatch( diff --git a/invokeai/frontend/web/src/services/events/types.ts b/invokeai/frontend/web/src/services/events/types.ts index 8937dcc451d..e8dff693cb4 100644 --- a/invokeai/frontend/web/src/services/events/types.ts +++ b/invokeai/frontend/web/src/services/events/types.ts @@ -26,6 +26,11 @@ export type ServerToClientEvents = { model_install_cancelled: (payload: S['ModelInstallCancelledEvent']) => void; model_load_complete: (payload: S['ModelLoadCompleteEvent']) => void; queue_item_status_changed: (payload: S['QueueItemStatusChangedEvent']) => void; + // Content-free broadcast to the whole queue room: the global queue counts may have changed + // (some user enqueued or a job changed status). Carries only queue_id — no per-user data — + // so every subscriber can refetch the redacted queue status. Emitted by the backend socket + // layer, not the event bus, so it has no generated `S[...]` schema type. + queue_counts_changed: (payload: { queue_id: string }) => void; queue_cleared: (payload: S['QueueClearedEvent']) => void; batch_enqueued: (payload: S['BatchEnqueuedEvent']) => void; queue_items_retried: (payload: S['QueueItemsRetriedEvent']) => void;