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;