diff --git a/client/src/api/schema/schema.ts b/client/src/api/schema/schema.ts index 62a3f2f0a51..09a570680d0 100644 --- a/client/src/api/schema/schema.ts +++ b/client/src/api/schema/schema.ts @@ -3995,32 +3995,6 @@ export interface paths { patch?: never; trace?: never; }; - "/api/notifications/stream": { - parameters: { - query?: never; - header?: never; - path?: never; - cookie?: never; - }; - /** - * Server-Sent Events stream for real-time notification updates. - * @description Opens a Server-Sent Events (SSE) connection that pushes notification updates in real-time. - * - * On reconnect, the browser sends the ``Last-Event-ID`` header automatically. - * Any notifications created since that timestamp are delivered as a catch-up - * ``notification_status`` event before the stream begins. - * - * Anonymous users receive only broadcast events. - */ - get: operations["stream_notifications_api_notifications_stream_get"]; - put?: never; - post?: never; - delete?: never; - options?: never; - head?: never; - patch?: never; - trace?: never; - }; "/api/notifications/{notification_id}": { parameters: { query?: never; @@ -42500,46 +42474,6 @@ export interface operations { }; }; }; - stream_notifications_api_notifications_stream_get: { - parameters: { - query?: never; - header?: { - "Last-Event-ID"?: string | null; - /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */ - "run-as"?: string | null; - }; - path?: never; - cookie?: never; - }; - requestBody?: never; - responses: { - /** @description Successful Response */ - 200: { - headers: { - [name: string]: unknown; - }; - content?: never; - }; - /** @description Request Error */ - "4XX": { - headers: { - [name: string]: unknown; - }; - content: { - "application/json": components["schemas"]["MessageExceptionModel"]; - }; - }; - /** @description Server Error */ - "5XX": { - headers: { - [name: string]: unknown; - }; - content: { - "application/json": components["schemas"]["MessageExceptionModel"]; - }; - }; - }; - }; show_notification_api_notifications__notification_id__get: { parameters: { query?: never; diff --git a/client/src/composables/useNotificationSSE.ts b/client/src/composables/useNotificationSSE.ts index 56cf861a4a6..8bdae964f2e 100644 --- a/client/src/composables/useNotificationSSE.ts +++ b/client/src/composables/useNotificationSSE.ts @@ -24,25 +24,64 @@ function sseGlobals(): SSEDebugGlobals { return window as unknown as SSEDebugGlobals; } -/** - * Composable for connecting to the unified SSE event stream. - * - * The browser's EventSource handles reconnection automatically and - * sends the Last-Event-ID header so the server can catch up on missed events. - * - * @param onEvent - callback invoked for every SSE event - * @param eventTypes - subset of event types to listen to (defaults to all) - */ -export function useSSE(onEvent: (event: MessageEvent) => void, eventTypes: readonly SSEEventType[] = SSE_EVENT_TYPES) { - const connected = ref(false); - let eventSource: EventSource | null = null; +// --------------------------------------------------------------------------- +// Module-level shared EventSource. +// +// Every call to ``useSSE`` registers its handler against this one socket so +// the tab opens a single ``/api/events/stream`` connection no matter how many +// stores listen. HTTP/1.1 caps simultaneous connections per origin at six; +// before this consolidation we burned three slots on SSE alone (history, +// notifications, entry points), which is what starved the scratchbook iframe +// flow — see the fix in ``client/src/entry/analysis/App.vue``. +// --------------------------------------------------------------------------- - // Selenium tests watch __galaxy_sse_last_event_ts to prove that an - // observable state change came from an SSE push and not the polling - // fallback (where __galaxy_sse_last_event_ts would never advance). - const trackedOnEvent = (event: MessageEvent) => { - sseGlobals().__galaxy_sse_last_event_ts = Date.now(); - onEvent(event); +type Handler = (event: MessageEvent) => void; + +let sharedSource: EventSource | null = null; +const sharedConnected = ref(false); +const subscribers: Map> = new Map(); +// Track the per-type dispatchers we registered so ``closeSource`` removes the +// exact same listeners (``addEventListener`` matches by reference). +const dispatchers: Map = new Map(); + +function openSourceIfNeeded() { + if (sharedSource) { + return; + } + sharedSource = new EventSource(withPrefix("/api/events/stream")); + + for (const eventType of SSE_EVENT_TYPES) { + const dispatcher: Handler = (event) => { + // Selenium tests watch ``__galaxy_sse_last_event_ts`` to prove that + // an observable state change came from an SSE push and not the + // polling fallback (where the global would never advance). + sseGlobals().__galaxy_sse_last_event_ts = Date.now(); + const subs = subscribers.get(eventType); + if (!subs) { + return; + } + for (const handler of subs) { + handler(event); + } + }; + dispatchers.set(eventType, dispatcher); + sharedSource.addEventListener(eventType, dispatcher); + } + + sharedSource.onopen = () => { + sharedConnected.value = true; + // Global readiness flag so Selenium tests can distinguish a working + // SSE pipeline from the polling fallback. + sseGlobals().__galaxy_sse_connected = true; + }; + + sharedSource.onerror = () => { + // EventSource auto-reconnects natively; SSE-vs-polling is a + // config-level decision (see historyStore / notificationsStore), so + // we must not give up on transient errors here — doing so would leave + // the client with no updates at all. + sharedConnected.value = false; + sseGlobals().__galaxy_sse_connected = false; }; // Browser EventSource teardown during a full-page navigation @@ -50,62 +89,100 @@ export function useSSE(onEvent: (event: MessageEvent) => void, eventTypes: reado // browser issues requests for the new page — we've seen Chrome keep the // stream alive long enough that a login/register POST reload races the // close, and the new page then loads with a stale auth view. Force a - // synchronous ``eventSource.close()`` during ``pagehide`` (fires for both - // reloads and tab-close, unlike ``beforeunload``) to close that window. - // The listener is registered only while a connection is live so composables - // that never ``connect()`` don't leave dangling listeners behind. - const onPageHide = () => disconnect(); + // synchronous ``close()`` during ``pagehide`` (fires for both reloads and + // tab-close, unlike ``beforeunload``) to close that window. + if (typeof window !== "undefined") { + window.addEventListener("pagehide", closeSource); + } +} + +function closeSource() { + if (!sharedSource) { + return; + } + for (const [eventType, dispatcher] of dispatchers) { + sharedSource.removeEventListener(eventType, dispatcher); + } + dispatchers.clear(); + sharedSource.close(); + sharedSource = null; + sharedConnected.value = false; + sseGlobals().__galaxy_sse_connected = false; + if (typeof window !== "undefined") { + window.removeEventListener("pagehide", closeSource); + } +} + +function addSubscriber(onEvent: Handler, eventTypes: readonly SSEEventType[]) { + for (const eventType of eventTypes) { + let subs = subscribers.get(eventType); + if (!subs) { + subs = new Set(); + subscribers.set(eventType, subs); + } + subs.add(onEvent); + } +} + +function removeSubscriber(onEvent: Handler, eventTypes: readonly SSEEventType[]): boolean { + let anyRemaining = false; + for (const eventType of eventTypes) { + const subs = subscribers.get(eventType); + if (subs) { + subs.delete(onEvent); + if (subs.size === 0) { + subscribers.delete(eventType); + } + } + } + for (const subs of subscribers.values()) { + if (subs.size > 0) { + anyRemaining = true; + break; + } + } + return anyRemaining; +} + +/** + * Composable for subscribing to events on the shared SSE stream. + * + * The browser's EventSource handles reconnection automatically and sends the + * ``Last-Event-ID`` header so the server can catch up on missed events. Only + * one EventSource is opened per tab regardless of how many callers invoke + * this composable; the composable multiplexes dispatch per event type. + * + * @param onEvent - callback invoked for every matching SSE event + * @param eventTypes - subset of event types to listen to (defaults to all) + */ +export function useSSE(onEvent: Handler, eventTypes: readonly SSEEventType[] = SSE_EVENT_TYPES) { + let connected_: boolean = false; function connect() { - disconnect(); - const url = withPrefix("/api/events/stream"); - eventSource = new EventSource(url); - - for (const eventType of eventTypes) { - eventSource.addEventListener(eventType, trackedOnEvent); - } - - eventSource.onopen = () => { - connected.value = true; - // Expose a global readiness flag so Selenium tests can distinguish - // a working SSE pipeline from the polling fallback. - sseGlobals().__galaxy_sse_connected = true; - }; - - eventSource.onerror = () => { - // EventSource auto-reconnects natively; SSE-vs-polling is a - // config-level decision (see historyStore / notificationsStore), - // so we must not give up on transient errors here — doing so - // would leave the client with no updates at all. - connected.value = false; - sseGlobals().__galaxy_sse_connected = false; - }; - - if (typeof window !== "undefined") { - window.addEventListener("pagehide", onPageHide); + if (connected_) { + return; } + connected_ = true; + addSubscriber(onEvent, eventTypes); + openSourceIfNeeded(); } function disconnect() { - if (eventSource) { - for (const eventType of eventTypes) { - eventSource.removeEventListener(eventType, trackedOnEvent); - } - eventSource.close(); - eventSource = null; + if (!connected_) { + return; } - if (typeof window !== "undefined") { - window.removeEventListener("pagehide", onPageHide); + connected_ = false; + const anyRemaining = removeSubscriber(onEvent, eventTypes); + if (!anyRemaining) { + closeSource(); } - connected.value = false; - sseGlobals().__galaxy_sse_connected = false; } onScopeDispose(() => { disconnect(); }); - return { connect, disconnect, connected }; + return { connect, disconnect, connected: sharedConnected }; } /** diff --git a/lib/galaxy/webapps/galaxy/api/notifications.py b/lib/galaxy/webapps/galaxy/api/notifications.py index d14d004fb91..f1bfbcab88a 100644 --- a/lib/galaxy/webapps/galaxy/api/notifications.py +++ b/lib/galaxy/webapps/galaxy/api/notifications.py @@ -10,13 +10,10 @@ from typing import ( from fastapi import ( Body, - Header, Query, - Request, Response, status, ) -from starlette.responses import StreamingResponse from galaxy.managers.context import ProvidesUserContext from galaxy.schema.notifications import ( @@ -55,35 +52,6 @@ router = Router(tags=["notifications"]) class FastAPINotifications: service: NotificationService = depends(NotificationService) - @router.get( - "/api/notifications/stream", - summary="Server-Sent Events stream for real-time notification updates.", - response_class=StreamingResponse, - ) - async def stream_notifications( - self, - request: Request, - trans: ProvidesUserContext = DependsOnTrans, - last_event_id: Optional[str] = Header(None, alias="Last-Event-ID"), - ) -> StreamingResponse: - """Opens a Server-Sent Events (SSE) connection that pushes notification updates in real-time. - - On reconnect, the browser sends the ``Last-Event-ID`` header automatically. - Any notifications created since that timestamp are delivered as a catch-up - ``notification_status`` event before the stream begins. - - Anonymous users receive only broadcast events. - """ - return StreamingResponse( - self.service.open_stream(trans, last_event_id, request.is_disconnected), - media_type="text/event-stream", - headers={ - "Cache-Control": "no-cache", - "Connection": "keep-alive", - "X-Accel-Buffering": "no", - }, - ) - @router.get( "/api/notifications/status", summary="Returns the current status summary of the user's notifications since a particular date.", diff --git a/lib/galaxy/webapps/galaxy/services/notifications.py b/lib/galaxy/webapps/galaxy/services/notifications.py index 046f5d1718a..0fa3091c6c0 100644 --- a/lib/galaxy/webapps/galaxy/services/notifications.py +++ b/lib/galaxy/webapps/galaxy/services/notifications.py @@ -1,4 +1,3 @@ -from collections.abc import AsyncIterator from datetime import datetime from typing import ( NoReturn, @@ -15,7 +14,6 @@ from galaxy.exceptions import ( from galaxy.managers.context import ProvidesUserContext from galaxy.managers.notification import NotificationManager from galaxy.managers.sse import ( - IsDisconnected, make_event_id, parse_event_id, SSEConnectionManager, @@ -50,22 +48,6 @@ class NotificationService(ServiceBase): self.notification_manager = notification_manager self.sse_manager = sse_manager - def open_stream( - self, - user_context: ProvidesUserContext, - last_event_id: Optional[str], - is_disconnected: IsDisconnected, - ) -> AsyncIterator[str]: - """Open an SSE notification stream for ``user_context``. - - Enforces the notifications-enabled guard, builds the optional catch-up, - and resolves the user id so the controller stays a thin wrapper. - """ - self.notification_manager.ensure_notifications_enabled() - user_id = user_context.user.id if not user_context.anonymous else None - catch_up = self.build_status_catchup(user_context, last_event_id) - return self.sse_manager.stream(is_disconnected, user_id, catch_up=catch_up) - def send_notification( self, sender_context: ProvidesUserContext, payload: NotificationCreateRequestBody ) -> Union[NotificationCreatedResponse, AsyncTaskResultSummary]: