Unify SSE: single shared EventSource, drop /api/notifications/stream

The three SSE-driven stores (history, notifications, entry points) each
called useSSE() in their own setup(), producing three separate
EventSource connections to /api/events/stream. HTTP/1.1 caps
per-origin connections at six, so SSE alone consumed half the budget
— and under iframe-heavy flows (scratchbook) it starved the pool and
hung the tab.

useSSE now multiplexes every subscriber through a single module-scoped
EventSource with a per-event-type dispatch registry. Refcounted
open/close so the socket is only created on first subscribe and closed
when the last subscriber goes away. Existing store call sites are
unchanged.

While here: drop the dead /api/notifications/stream route. The generic
/api/events/stream endpoint (lib/galaxy/webapps/galaxy/api/events.py)
already serves notifications alongside history_update and
entry_point_update events via the same SSEConnectionManager, and no
frontend code referenced the notification-only path. Delete the route,
its service method, and the auto-generated schema entries.

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;

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<SSEEventType, Set<Handler>> = new Map();
+// Track the per-type dispatchers we registered so ``closeSource`` removes the
+// exact same listeners (``addEventListener`` matches by reference).
+const dispatchers: Map<SSEEventType, Handler> = 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 };
 }
 
 /**

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.",

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]: