Add SSE entry-point channel, dispatch observability, declare-queue cache

Three related additions to the SSE notification pipeline, bundled here
because they share ``SSEEventDispatcher._send`` as their modification
point:

1. Interactive-tool entry-point SSE channel
   - ``entry_point_update`` dispatcher method + queue-worker handler.
   - ``InteractiveToolManager.configure_entry_points`` dispatches a
     wake-up event after the DB commit; the client refetches
     ``/api/entry_points`` on receipt (no payload).
   - Frontend: new SSE event type, store subscription with XOR polling
     fallback via ``enable_sse_entry_point_updates`` config flag.
   - Integration + selenium tests.

2. Queue and SSE observability metrics
   - Counters, timers, and periodic gauges for SSE dispatch, control-queue
     task execution, control-queue depth (via kombu passive declare),
     active SSE connections, dropped events, and active WorkerProcess
     rows. Flow through the existing ``galaxy_statsd_client`` — no new
     infra. Gauges are scheduled via Celery beat at ``queue_metrics_interval``
     seconds (default 15). All instrumentation no-ops when statsd isn't
     configured.
   - Sub-emitter failures log once at WARNING and bump a
     ``galaxy.queue_metrics.error`` counter tagged by emitter so broken
     emitters are visible in metrics without log spam.

3. Active-worker control-queue cache
   - 30 s TTL cache on ``all_control_queues_for_declare`` with RLock
     stampede protection. At 1000+ events/s this eliminates ~30 DB
     round-trips/s per webapp for data that only changes on 60 s
     heartbeat cadence. Empty results are not cached — would otherwise
     silently drop every SSE event during the startup window.

Author: mvdbeek

client/src/composables/useNotificationSSE.ts

@@ -10,6 +10,7 @@ export const SSE_EVENT_TYPES = [
     "broadcast_update",
     "notification_status",
     "history_update",
+    "entry_point_update",
 ] as const;
 
 export type SSEEventType = (typeof SSE_EVENT_TYPES)[number];

client/src/stores/_testing/sseStoreSupport.ts

@@ -13,20 +13,27 @@
  * files in this directory for the pattern.
  */
 
+import { ref, type Ref } from "vue";
 import { vi } from "vitest";
 
 export interface SSEMockState {
     onEvent: ((event: MessageEvent) => void) | null;
     connect: ReturnType<typeof vi.fn>;
     disconnect: ReturnType<typeof vi.fn>;
+    connected: Ref<boolean>;
 }
 
 /** Build the factory used with ``vi.mock("@/composables/useNotificationSSE", ...)``. */
 export function sseMockFactory(state: SSEMockState) {
+    // Lazily initialize ``connected`` so existing callers that don't pass it
+    // still get a working ref.
+    if (!state.connected) {
+        state.connected = ref(false);
+    }
     return {
         useSSE: vi.fn((onEvent: (event: MessageEvent) => void) => {
             state.onEvent = onEvent;
-            return { connect: state.connect, disconnect: state.disconnect };
+            return { connect: state.connect, disconnect: state.disconnect, connected: state.connected };
         }),
     };
 }

client/src/stores/entryPointStore.test.js

@@ -1,12 +1,25 @@
 import flushPromises from "flush-promises";
 import { createPinia, setActivePinia } from "pinia";
-import { beforeEach, describe, expect, it } from "vitest";
+import { beforeEach, describe, expect, it, vi } from "vitest";
 
 import { HttpResponse, useServerMock } from "@/api/client/__mocks__";
 
 import testInteractiveToolsResponse from "../components/InteractiveTools/testData/testInteractiveToolsResponse";
+import { sseMockFactory } from "./_testing/sseStoreSupport";
 import { useEntryPointStore } from "./entryPointStore";
 
+// ``vi.mock`` is hoisted above module-level declarations, so the capture-state
+// has to be built via ``vi.hoisted`` to be visible to the factory. Prevents
+// these tests from opening a real EventSource against ``/api/events/stream``
+// when ``useEntryPointStore()`` is invoked.
+const sseState = vi.hoisted(() => ({
+    onEvent: null,
+    connect: vi.fn(),
+    disconnect: vi.fn(),
+    connected: null,
+}));
+vi.mock("@/composables/useNotificationSSE", () => sseMockFactory(sseState));
+
 const { server, http } = useServerMock();
 
 describe("stores/EntryPointStore", () => {

client/src/stores/entryPointStore.ts

@@ -1,10 +1,12 @@
 import axios from "axios";
 import isEqual from "lodash.isequal";
 import { defineStore } from "pinia";
-import { computed, ref } from "vue";
+import { computed, ref, watch } from "vue";
 
 import { useResourceWatcher } from "@/composables/resourceWatcher";
+import { useSSE } from "@/composables/useNotificationSSE";
 import { getAppRoot } from "@/onload/loadConfig";
+import { useConfigStore } from "@/stores/configurationStore";
 import { rethrowSimple } from "@/utils/simple-error";
 
 const ACTIVE_POLLING_INTERVAL = 10000;
@@ -23,22 +25,8 @@ interface EntryPoint {
 }
 
 export const useEntryPointStore = defineStore("entryPointStore", () => {
-    const { startWatchingResource: startWatchingEntryPoints } = useResourceWatcher(fetchEntryPoints, {
-        shortPollingInterval: ACTIVE_POLLING_INTERVAL,
-        enableBackgroundPolling: false, // No need to poll in the background
-    });
-
     const entryPoints = ref<EntryPoint[]>([]);
 
-    const entryPointsForJob = computed(() => {
-        return (jobId: string) => entryPoints.value.filter((entryPoint) => entryPoint["job_id"] === jobId);
-    });
-
-    const entryPointsForHda = computed(() => {
-        return (hdaId: string) =>
-            entryPoints.value.filter((entryPoint) => entryPoint["output_datasets_ids"].includes(hdaId));
-    });
-
     async function fetchEntryPoints() {
         const url = `${getAppRoot()}api/entry_points`;
         const params = { running: true };
@@ -50,6 +38,73 @@ export const useEntryPointStore = defineStore("entryPointStore", () => {
         }
     }
 
+    // SSE-driven path: on each entry_point_update signal, refetch the canonical
+    // list from REST. The event carries no data — it's a pure wake-up.
+    function handleEntryPointSSEEvent(_event: MessageEvent) {
+        fetchEntryPoints().catch((err) => console.error("Error refreshing entry points from SSE push:", err));
+    }
+    const { connect: sseConnect, connected: sseConnected } = useSSE(handleEntryPointSSEEvent, ["entry_point_update"]);
+
+    let watchingInitialized = false;
+
+    // Callers opt in via ``startWatchingEntryPoints()`` (App.vue gates this on
+    // ``interactivetools_enable``). We then pick SSE or polling based on the
+    // server flag — mutually exclusive, mirroring historyStore / notificationsStore.
+    // ``useConfigStore`` is resolved lazily here so tests that only exercise
+    // the data methods don't need a ``/api/configuration`` handler registered.
+    function startWatchingEntryPoints() {
+        if (watchingInitialized) {
+            return;
+        }
+        watchingInitialized = true;
+        const configStore = useConfigStore();
+
+        const decide = () => {
+            if (configStore.config?.enable_sse_entry_point_updates) {
+                // Baseline fetch + SSE. Reconnect-refetch closes the "user
+                // navigated away and missed events" window.
+                fetchEntryPoints().catch((err) => console.warn("Initial entry-point load failed", err));
+                sseConnect();
+                watch(sseConnected, (isConnected, wasConnected) => {
+                    if (isConnected && !wasConnected) {
+                        fetchEntryPoints().catch((err) =>
+                            console.error("Error refreshing entry points on SSE reconnect:", err),
+                        );
+                    }
+                });
+            } else {
+                const { startWatchingResource } = useResourceWatcher(fetchEntryPoints, {
+                    shortPollingInterval: ACTIVE_POLLING_INTERVAL,
+                    enableBackgroundPolling: false,
+                });
+                startWatchingResource();
+            }
+        };
+
+        if (configStore.isLoaded) {
+            decide();
+        } else {
+            const stop = watch(
+                () => configStore.isLoaded,
+                (loaded) => {
+                    if (loaded) {
+                        stop();
+                        decide();
+                    }
+                },
+            );
+        }
+    }
+
+    const entryPointsForJob = computed(() => {
+        return (jobId: string) => entryPoints.value.filter((entryPoint) => entryPoint["job_id"] === jobId);
+    });
+
+    const entryPointsForHda = computed(() => {
+        return (hdaId: string) =>
+            entryPoints.value.filter((entryPoint) => entryPoint["output_datasets_ids"].includes(hdaId));
+    });
+
     function updateEntryPoints(data: EntryPoint[]) {
         let hasChanged = entryPoints.value.length !== data.length ? true : false;
         if (entryPoints.value.length === 0) {

doc/source/admin/galaxy_options.rst

@@ -3394,6 +3394,18 @@
 :Type: bool
 
 
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+``queue_metrics_interval``
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+:Description:
+    How often (in seconds) the Celery beat task emits queue-depth,
+    SSE-connection, and WorkerProcess gauges. Only active when
+    statsd_host is set. Set to 0 to disable.
+:Default: ``15``
+:Type: int
+
+
 ~~~~~~~~~~~~~~~~~~~~~~
 ``library_import_dir``
 ~~~~~~~~~~~~~~~~~~~~~~
@@ -5804,6 +5816,20 @@
 :Type: bool
 
 
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+``enable_sse_entry_point_updates``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+:Description:
+    Enables real-time interactive-tool entry-point update
+    notifications via Server-Sent Events. When enabled, the client
+    subscribes to entry_point_update SSE events and refetches the
+    entry-point list on each event, replacing the 10-second polling
+    loop. When disabled, polling remains the source of updates.
+:Default: ``false``
+:Type: bool
+
+
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 ``history_audit_monitor_poll_interval``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

lib/galaxy/app/__init__.py

@@ -693,6 +693,7 @@ def __init__(
             SSEEventDispatcher(
                 queue_worker=getattr(self, "queue_worker", None),
                 application_stack=self.application_stack,
+                statsd_client=self.execution_timer_factory.galaxy_statsd_client,
             ),
         )
         self.notification_manager = self._register_singleton(NotificationManager)
@@ -856,7 +857,12 @@ def __init__(self, **kwargs) -> None:
         # SSE connection manager for real-time notification push.
         # Consumed via ``depends(SSEConnectionManager)`` / ``app[SSEConnectionManager]``,
         # so no module-level attribute is needed — keep the container wiring only.
-        self._register_singleton(SSEConnectionManager)
+        self._register_singleton(
+            SSEConnectionManager,
+            SSEConnectionManager(
+                statsd_client=self.execution_timer_factory.galaxy_statsd_client,
+            ),
+        )
 
         # AI agent registry and service
         agent_registry = build_agent_registry(self.config)

lib/galaxy/app_unittest_utils/galaxy_mock.py

@@ -159,7 +159,7 @@ def __init__(self, config=None, **kwargs) -> None:
         self.application_stack = ApplicationStack()
         self.auth_manager = AuthManager(self.config)
         self.user_manager = UserManager(cast(BasicSharedApp, self))
-        self.execution_timer_factory = Bunch(get_timer=StructuredExecutionTimer)
+        self.execution_timer_factory = Bunch(get_timer=StructuredExecutionTimer, galaxy_statsd_client=None)
         self.interactivetool_manager = Bunch(create_interactivetool=lambda *args, **kwargs: None)
         self.is_job_handler = False
         self.biotools_metadata_source = None

lib/galaxy/authnz/psa_authnz.py

@@ -798,6 +798,7 @@ def _send_oidc_profile_update_notification(trans, user, updates: list[str]) -> N
             NotificationVariant,
             PersonalNotificationCategory,
         )
+
         labels: dict[str, str] = {
             "email": "email address",
             "username": "public name",

lib/galaxy/celery/__init__.py

@@ -252,6 +252,9 @@ def schedule_task(task, interval):
     schedule_task("prune_history_audit_table", config.history_audit_table_prune_interval)
     schedule_task("cleanup_short_term_storage", config.short_term_storage_cleanup_interval)
 
+    if config.statsd_host:
+        schedule_task("emit_queue_metrics_task", config.queue_metrics_interval)
+
     if config.enable_notification_system:
         schedule_task("cleanup_expired_notifications", config.expired_notifications_cleanup_interval)
         schedule_task("dispatch_pending_notifications", config.dispatch_notifications_interval)

lib/galaxy/celery/tasks.py

@@ -622,6 +622,14 @@ def dispatch_pending_notifications(notification_manager: NotificationManager):
         log.info(f"Successfully dispatched {count} notifications.")
 
 
+@galaxy_task(action="emit queue and SSE observability metrics")
+def emit_queue_metrics_task(app: MinimalManagerApp):
+    """Sample control-queue depth, SSE connection count, and worker rows → statsd."""
+    from galaxy.webapps.galaxy.metrics.queue_metrics import emit_queue_metrics
+
+    emit_queue_metrics(app)
+
+
 @galaxy_task(action="clean up job working directories")
 def cleanup_jwds(sa_session: galaxy_scoped_session, object_store: BaseObjectStore, config: GalaxyAppConfiguration):
     """Cleanup job working directories for failed jobs that are older than X days"""