diff --git a/client/src/api/schema/schema.ts b/client/src/api/schema/schema.ts index b171e21e0f3..62a3f2f0a51 100644 --- a/client/src/api/schema/schema.ts +++ b/client/src/api/schema/schema.ts @@ -1292,6 +1292,33 @@ export interface paths { patch?: never; trace?: never; }; + "/api/events/stream": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * Server-Sent Events stream for real-time updates. + * @description Opens a Server-Sent Events (SSE) connection that pushes real-time + * updates for notifications, history changes, and other events. + * + * On reconnect, the browser sends the ``Last-Event-ID`` header automatically. + * If the notification system is enabled, any notifications created since that + * timestamp are delivered as a catch-up ``notification_status`` event. + * + * Anonymous users receive only broadcast events. + */ + get: operations["stream_events_api_events_stream_get"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/exports": { parameters: { query?: never; @@ -3968,6 +3995,32 @@ 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; @@ -33209,6 +33262,46 @@ export interface operations { }; }; }; + stream_events_api_events_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"]; + }; + }; + }; + }; index_api_exports_get: { parameters: { query?: { @@ -42407,6 +42500,46 @@ 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 new file mode 100644 index 00000000000..84b03373cfa --- /dev/null +++ b/client/src/composables/useNotificationSSE.ts @@ -0,0 +1,84 @@ +import { onScopeDispose, ref } from "vue"; + +import { withPrefix } from "@/utils/redirect"; + +/** + * All SSE event types the server may emit. + */ +export const SSE_EVENT_TYPES = [ + "notification_update", + "broadcast_update", + "notification_status", + "history_update", +] as const; + +export type SSEEventType = (typeof SSE_EVENT_TYPES)[number]; + +/** + * 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; + let consecutiveErrors = 0; + + function connect() { + disconnect(); + consecutiveErrors = 0; + const url = withPrefix("/api/events/stream"); + eventSource = new EventSource(url); + + for (const eventType of eventTypes) { + eventSource.addEventListener(eventType, onEvent); + } + + eventSource.onopen = () => { + connected.value = true; + consecutiveErrors = 0; + // Expose a global readiness flag so Selenium tests can distinguish + // a working SSE pipeline from the polling fallback. + (window as unknown as { __galaxy_sse_connected?: boolean }).__galaxy_sse_connected = true; + }; + + eventSource.onerror = () => { + connected.value = false; + (window as unknown as { __galaxy_sse_connected?: boolean }).__galaxy_sse_connected = false; + consecutiveErrors++; + // EventSource auto-reconnects, but if we get too many errors + // in a row, the server likely doesn't support SSE — give up + // and let the caller fall back to polling. + if (consecutiveErrors > 5) { + disconnect(); + } + }; + } + + function disconnect() { + if (eventSource) { + for (const eventType of eventTypes) { + eventSource.removeEventListener(eventType, onEvent); + } + eventSource.close(); + eventSource = null; + } + connected.value = false; + (window as unknown as { __galaxy_sse_connected?: boolean }).__galaxy_sse_connected = false; + } + + onScopeDispose(() => { + disconnect(); + }); + + return { connect, disconnect, connected }; +} + +/** + * @deprecated Use `useSSE` instead. This alias exists for backward compatibility. + */ +export const useNotificationSSE = useSSE; diff --git a/client/src/stores/historyStore.ts b/client/src/stores/historyStore.ts index 019c57de8ae..1d6c40acebc 100644 --- a/client/src/stores/historyStore.ts +++ b/client/src/stores/historyStore.ts @@ -15,6 +15,7 @@ import type { ArchivedHistoryDetailed } from "@/api/histories.archived"; import { getGalaxyInstance } from "@/app"; import { HistoryFilters } from "@/components/History/HistoryFilters"; import { useResourceWatcher } from "@/composables/resourceWatcher"; +import { useSSE } from "@/composables/useNotificationSSE"; import { useUserLocalStorage } from "@/composables/userLocalStorage"; import { createAndSelectNewHistory, @@ -31,6 +32,7 @@ import { ACTIVE_POLLING_INTERVAL, INACTIVE_POLLING_INTERVAL, watchHistory as watchHistorySuppliedApp, + watchHistoryOnce as watchHistoryOnceSuppliedApp, } from "@/watch/watchHistory"; const PAGINATION_LIMIT = 10; @@ -391,6 +393,26 @@ export const useHistoryStore = defineStore("historyStore", () => { return watchHistorySuppliedApp(app); } + // SSE-driven history updates: when we receive a history_update event, + // immediately trigger a refresh of the current history + const SSE_HISTORY_EVENT_TYPES = ["history_update"] as const; + const { connect: sseHistoryConnect } = useSSE(handleHistorySSEEvent, SSE_HISTORY_EVENT_TYPES); + + function handleHistorySSEEvent(event: MessageEvent) { + try { + const data = JSON.parse(event.data); + const changedHistoryIds: string[] = data.history_ids ?? []; + // If the current history was updated, trigger a refresh + if (currentHistoryId.value && changedHistoryIds.includes(currentHistoryId.value)) { + const app = getGalaxyInstance(); + watchHistoryOnceSuppliedApp(app); + } + } catch (e) { + console.error("Error handling history SSE event:", e); + } + } + + // Polling fallback — keeps running as a safety net even when SSE is connected const { startWatchingResource: startWatchingHistory, stopWatchingResource: stopWatchingHistory, @@ -400,6 +422,13 @@ export const useHistoryStore = defineStore("historyStore", () => { longPollingInterval: INACTIVE_POLLING_INTERVAL, }); + function startWatchingHistoryWithSSE() { + // Always start polling as a baseline + startWatchingHistory(); + // Also connect SSE for instant updates + sseHistoryConnect(); + } + async function loadHistoryById(historyId: string) { if (!isLoadingHistory.has(historyId)) { isLoadingHistory.add(historyId); @@ -525,7 +554,7 @@ export const useHistoryStore = defineStore("historyStore", () => { restoreHistory, restoreHistories, handleTotalCountChange, - startWatchingHistory, + startWatchingHistory: startWatchingHistoryWithSSE, stopWatchingHistory, isWatchingHistory, loadCurrentHistory, diff --git a/client/src/stores/notificationsStore.ts b/client/src/stores/notificationsStore.ts index 2a154885d99..220bd8e796a 100644 --- a/client/src/stores/notificationsStore.ts +++ b/client/src/stores/notificationsStore.ts @@ -1,9 +1,10 @@ import { defineStore } from "pinia"; -import { computed, ref } from "vue"; +import { computed, ref, watch } from "vue"; import { GalaxyApi } from "@/api"; import type { NotificationChanges, UserNotification, UserNotificationsBatchUpdateRequest } from "@/api/notifications"; import { useResourceWatcher } from "@/composables/resourceWatcher"; +import { useSSE } from "@/composables/useNotificationSSE"; import { rethrowSimple } from "@/utils/simple-error"; import { mergeObjectListsById } from "@/utils/utils"; @@ -13,11 +14,6 @@ const ACTIVE_POLLING_INTERVAL = 30000; // 30 seconds const INACTIVE_POLLING_INTERVAL = ACTIVE_POLLING_INTERVAL * 20; // 10 minutes export const useNotificationsStore = defineStore("notificationsStore", () => { - const { startWatchingResource: startWatchingNotifications, stopWatchingResource: stopWatchingNotifications } = - useResourceWatcher(getNotificationStatus, { - shortPollingInterval: ACTIVE_POLLING_INTERVAL, - longPollingInterval: INACTIVE_POLLING_INTERVAL, - }); const broadcastsStore = useBroadcastsStore(); const totalUnreadCount = ref(0); @@ -25,9 +21,82 @@ export const useNotificationsStore = defineStore("notificationsStore", () => { const loadingNotifications = ref(false); const lastNotificationUpdate = ref(null); + const wantSSE = ref(true); const unreadNotifications = computed(() => notifications.value.filter((n) => !n.seen_time)); + // --- SSE setup (listen only for notification event types) --- + const NOTIFICATION_EVENT_TYPES = ["notification_update", "broadcast_update", "notification_status"] as const; + const { + connect: sseConnect, + disconnect: sseDisconnect, + connected: sseConnected, + } = useSSE(handleSSEEvent, NOTIFICATION_EVENT_TYPES); + + // --- Polling fallback --- + const { startWatchingResource: startPolling, stopWatchingResource: stopPolling } = useResourceWatcher( + getNotificationStatus, + { + shortPollingInterval: ACTIVE_POLLING_INTERVAL, + longPollingInterval: INACTIVE_POLLING_INTERVAL, + }, + ); + + function stopWatchingNotifications() { + sseDisconnect(); + stopPolling(); + } + + // When SSE connection drops and doesn't recover, fall back to polling + watch(sseConnected, (isConnected) => { + if (!isConnected && wantSSE.value) { + // SSE disconnected but we still want updates — don't start polling + // immediately, EventSource will auto-reconnect. Only if useSSE is + // set to false (after too many errors) do we fall back. + } + }); + + watch(wantSSE, (wantSSE) => { + if (!wantSSE) { + sseDisconnect(); + startPolling(); + } + }); + + function handleSSEEvent(event: MessageEvent) { + try { + const data = JSON.parse(event.data); + switch (event.type) { + case "notification_update": + notifications.value = mergeObjectListsById( + notifications.value, + [data as UserNotification], + "create_time", + "desc", + ); + updateUnreadCount(); + break; + case "broadcast_update": + broadcastsStore.updateBroadcasts([data]); + break; + case "notification_status": + // Full catch-up on reconnect (same shape as GET /api/notifications/status) + totalUnreadCount.value = data.total_unread_count; + notifications.value = mergeObjectListsById( + notifications.value, + data.notifications as UserNotification[], + "create_time", + "desc", + ); + broadcastsStore.updateBroadcasts(data.broadcasts); + break; + } + lastNotificationUpdate.value = new Date(); + } catch (e) { + console.error("Error handling SSE event:", e); + } + } + async function loadNotifications() { const { data, error } = await GalaxyApi().GET("/api/notifications"); @@ -76,6 +145,29 @@ export const useNotificationsStore = defineStore("notificationsStore", () => { } } + async function startWatchingNotifications() { + // Always do an initial load first + if (!lastNotificationUpdate.value) { + try { + loadingNotifications.value = true; + await broadcastsStore.loadBroadcasts(); + await loadNotifications(); + updateUnreadCount(); + lastNotificationUpdate.value = new Date(); + } catch (e) { + console.error(e); + } finally { + loadingNotifications.value = false; + } + } + + if (wantSSE.value) { + sseConnect(); + } else { + startPolling(); + } + } + async function updateBatchNotification(request: UserNotificationsBatchUpdateRequest) { const { error } = await GalaxyApi().PUT("/api/notifications", { body: request, @@ -88,7 +180,10 @@ export const useNotificationsStore = defineStore("notificationsStore", () => { if (request.changes.deleted) { notifications.value = notifications.value.filter((n) => !request.notification_ids.includes(n.id)); } - startWatchingNotifications(); + // If not using SSE, trigger a poll to refresh state + if (!sseConnected.value) { + startWatchingNotifications(); + } } async function updateNotification(notification: UserNotification, changes: NotificationChanges) { diff --git a/doc/source/admin/galaxy_options.rst b/doc/source/admin/galaxy_options.rst index 47e3aaacf34..da52bf02d82 100644 --- a/doc/source/admin/galaxy_options.rst +++ b/doc/source/admin/galaxy_options.rst @@ -5804,6 +5804,34 @@ :Type: str +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +``enable_sse_history_updates`` +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +:Description: + Enables real-time history update notifications via Server-Sent + Events (SSE). When enabled, a background monitor watches for + history changes (via PostgreSQL LISTEN/NOTIFY or audit table + polling as a fallback for SQLite) and pushes update signals to + connected browsers, replacing aggressive 3-second polling. +:Default: ``false`` +:Type: bool + + +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +``history_audit_monitor_poll_interval`` +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +:Description: + The interval in seconds between history audit table polls when + using the polling fallback (SQLite or when PostgreSQL + LISTEN/NOTIFY is unavailable). Only used when + enable_sse_history_updates is true. Lower values mean faster + updates but more database queries. Recommended range: 1-5 seconds. +:Default: ``2`` +:Type: int + + ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ``enable_notification_system`` ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/lib/galaxy/app/__init__.py b/lib/galaxy/app/__init__.py index e8b8ef1b941..672efde7dc6 100644 --- a/lib/galaxy/app/__init__.py +++ b/lib/galaxy/app/__init__.py @@ -77,6 +77,10 @@ from galaxy.managers.notification import NotificationManager from galaxy.managers.object_store_instances import UserObjectStoreResolverImpl from galaxy.managers.roles import RoleManager from galaxy.managers.session import GalaxySessionManager +from galaxy.managers.sse import ( + SSEConnectionManager, + SSEEventDispatcher, +) from galaxy.managers.tasks import ( AsyncTasksManager, CeleryAsyncTasksManager, @@ -150,8 +154,12 @@ from galaxy.structured_app import ( ) from galaxy.tool_shed.cache import ToolShedRepositoryCache from galaxy.tool_shed.galaxy_install.client import InstallationTarget -from galaxy.tool_shed.galaxy_install.installed_repository_manager import InstalledRepositoryManager -from galaxy.tool_shed.galaxy_install.update_repository_manager import UpdateRepositoryManager +from galaxy.tool_shed.galaxy_install.installed_repository_manager import ( + InstalledRepositoryManager, +) +from galaxy.tool_shed.galaxy_install.update_repository_manager import ( + UpdateRepositoryManager, +) from galaxy.tool_util.data import ToolDataTableManager as BaseToolDataTableManager from galaxy.tool_util.deps import containers from galaxy.tool_util.deps.dependencies import AppInfo @@ -250,7 +258,11 @@ class SentryClientMixin: # "cannot find 'file_name' while searching for 'species_chromosomes.file_name'"] # If we don't do this issues are never properly grouped since by default the calling stack is inspected, # and that is always unique in cheetah as it is dynamically generated. - event["fingerprint"] = [str(exc_value), str(exc_value.tool_version), str(exc_value.__cause__)] + event["fingerprint"] = [ + str(exc_value), + str(exc_value.tool_version), + str(exc_value.__cause__), + ] event.setdefault("tags", {}).update( { "tool_is_latest": exc_value.is_latest, @@ -307,7 +319,10 @@ class MinimalGalaxyApplication(BasicSharedApp, HaltableContainer, SentryClientMi config_file = kwargs.get("global_conf", {}).get("__file__", None) if config_file: log.debug('Using "galaxy.ini" config file: %s', config_file) - self._configure_models(check_migrate_databases=self.config.check_migrate_databases, config_file=config_file) + self._configure_models( + check_migrate_databases=self.config.check_migrate_databases, + config_file=config_file, + ) # Security helper self._configure_security() self._register_singleton(IdEncodingHelper, self.security) @@ -411,7 +426,11 @@ class MinimalGalaxyApplication(BasicSharedApp, HaltableContainer, SentryClientMi index_help = getattr(self.config, "index_tool_help", True) self.toolbox_search = self._register_singleton( ToolBoxSearch, - ToolBoxSearch(self.toolbox, index_dir=self.config.tool_search_index_dir, index_help=index_help), + ToolBoxSearch( + self.toolbox, + index_dir=self.config.tool_search_index_dir, + index_help=index_help, + ), ) @property @@ -496,8 +515,9 @@ class MinimalGalaxyApplication(BasicSharedApp, HaltableContainer, SentryClientMi templates = ConfiguredObjectStoreTemplates.from_app_config(self.config, vault_configured=vault_configured) self.object_store_templates = self._register_singleton(ConfiguredObjectStoreTemplates, templates) user_object_store_resolver = self._register_abstract_singleton( - UserObjectStoreResolver, UserObjectStoreResolverImpl # type: ignore[type-abstract] - ) # Ignored because of https://github.com/python/mypy/issues/4717 + UserObjectStoreResolver, # type: ignore[type-abstract] # https://github.com/python/mypy/issues/4717 + UserObjectStoreResolverImpl, + ) kwds["user_object_store_resolver"] = user_object_store_resolver self.object_store = build_object_store_from_config(self.config, **kwds) @@ -600,7 +620,13 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication): model: GalaxyModelMapping - def __init__(self, configure_logging=True, use_converters=True, use_display_applications=True, **kwargs): + def __init__( + self, + configure_logging=True, + use_converters=True, + use_display_applications=True, + **kwargs, + ): super().__init__(**kwargs) self._register_singleton(MinimalManagerApp, self) # type: ignore[type-abstract] self.execution_timer_factory = self._register_singleton( @@ -617,7 +643,8 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication): # Initialize job metrics manager, needs to be in place before # config so per-destination modifications can be made. self.job_metrics = self._register_singleton( - JobMetrics, JobMetrics(self.config.job_metrics_config_file, self.config.job_metrics, app=self) + JobMetrics, + JobMetrics(self.config.job_metrics_config_file, self.config.job_metrics, app=self), ) # Initialize the job management configuration self.job_config = self._register_singleton(jobs.JobConfiguration) @@ -655,11 +682,15 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication): self.role_manager = self._register_singleton(RoleManager) self.job_manager = self._register_singleton(JobManager) + # SSE dispatcher must be registered before NotificationManager so Lagom + # can auto-inject the Optional[SSEEventDispatcher] constructor arg. + self._register_singleton(SSEEventDispatcher, SSEEventDispatcher(self)) self.notification_manager = self._register_singleton(NotificationManager) self.interactivetool_manager = InteractiveToolManager(self) self.task_manager = self._register_abstract_singleton( - AsyncTasksManager, CeleryAsyncTasksManager # type: ignore[type-abstract] # https://github.com/python/mypy/issues/4717 + AsyncTasksManager, # type: ignore[type-abstract] # https://github.com/python/mypy/issues/4717 + CeleryAsyncTasksManager, ) # ConfiguredFileSources @@ -671,10 +702,12 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication): self._register_singleton(FileSourcePluginLoader, file_source_plugin_loader) self.file_source_templates = self._register_singleton(ConfiguredFileSourceTemplates, templates) self._register_singleton( - UserDefinedFileSourcesConfig, UserDefinedFileSourcesConfig.from_app_config(self.config) + UserDefinedFileSourcesConfig, + UserDefinedFileSourcesConfig.from_app_config(self.config), ) user_defined_file_sources = self._register_abstract_singleton( - UserDefinedFileSources, UserDefinedFileSourcesImpl # type: ignore[type-abstract] # https://github.com/python/mypy/issues/4717 + UserDefinedFileSources, # type: ignore[type-abstract] # https://github.com/python/mypy/issues/4717 + UserDefinedFileSourcesImpl, ) configured_file_source_conf: ConfiguredFileSourcesConf = ConfiguredFileSourcesConf.from_app_config(self.config) file_sources = ConfiguredFileSources( @@ -690,7 +723,8 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication): # Load security policy. self.security_agent = self.model.security_agent self.host_security_agent = galaxy.model.security.HostAgent( - self.security_agent.sa_session, permitted_actions=self.security_agent.permitted_actions + self.security_agent.sa_session, + permitted_actions=self.security_agent.permitted_actions, ) # We need the datatype registry for running certain tasks that modify HDAs, and to build the registry we need @@ -791,6 +825,7 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT ("queue worker", self._shutdown_queue_worker), ("file watcher", self._shutdown_watcher), ("database heartbeat", self._shutdown_database_heartbeat), + ("history audit monitor", self._shutdown_history_audit_monitor), ("workflow scheduler", self._shutdown_scheduling_manager), ("object store", self._shutdown_object_store), ("job manager", self._shutdown_job_manager), @@ -809,17 +844,23 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT # queue_worker *can* be initialized with a queue, but here we don't # want to and we'll allow postfork to bind and start it. self.queue_worker = self._register_singleton(GalaxyQueueWorker, GalaxyQueueWorker(self)) + # SSE connection manager for real-time notification push + self.sse_connection_manager = self._register_singleton(SSEConnectionManager) # AI agent registry and service agent_registry = build_agent_registry(self.config) self._register_singleton(AgentRegistry, agent_registry) - self._register_singleton(AgentService, AgentService(self.config, JobQueryManager(self), agent_registry)) + self._register_singleton( + AgentService, + AgentService(self.config, JobQueryManager(self), agent_registry), + ) self.dependency_resolvers_view = self._register_singleton( DependencyResolversView, DependencyResolversView(self) ) self.test_data_resolver = self._register_singleton( - TestDataResolver, TestDataResolver(file_dirs=self.config.tool_test_data_directories) + TestDataResolver, + TestDataResolver(file_dirs=self.config.tool_test_data_directories), ) self.api_keys_manager = self._register_singleton(ApiKeyManager) @@ -876,7 +917,9 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT # Start the heartbeat process if configured and available if self.config.use_heartbeat: self.heartbeat = heartbeat.Heartbeat( - self.config, period=self.config.heartbeat_interval, fname=self.config.heartbeat_log + self.config, + period=self.config.heartbeat_interval, + fname=self.config.heartbeat_log, ) self.heartbeat.daemon = True self.application_stack.register_postfork_function(self.heartbeat.start) @@ -886,7 +929,9 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT from galaxy.authnz import managers self.authnz_manager = managers.AuthnzManager( - self, self.config.oidc_config_file, self.config.oidc_backends_config_file + self, + self.config.oidc_config_file, + self.config.oidc_backends_config_file, ) # If there is only a single external authentication provider in use @@ -923,7 +968,12 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT self.workflow_completion_hook_registry, ) self.application_stack.register_postfork_function(self.workflow_completion_monitor.start) - self.haltables.append(("WorkflowCompletionMonitor", self.workflow_completion_monitor.shutdown_monitor)) + self.haltables.append( + ( + "WorkflowCompletionMonitor", + self.workflow_completion_monitor.shutdown_monitor, + ) + ) # Start the job manager self.application_stack.register_postfork_function(self.job_manager.start) @@ -942,12 +992,20 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT self.database_heartbeat.add_change_callback(self.watchers.change_state) self.application_stack.register_postfork_function(self.database_heartbeat.start) + # History audit monitor for SSE-based history updates + if self.config.enable_sse_history_updates: + from galaxy.managers.history_audit_monitor import HistoryAuditMonitor + + self._history_audit_monitor = self._register_singleton(HistoryAuditMonitor) + self.application_stack.register_postfork_function(self._history_audit_monitor.start) + # Start web stack message handling self.application_stack.register_postfork_function(self.application_stack.start) self.application_stack.register_postfork_function(self.queue_worker.bind_and_start) # Reload toolbox to pick up changes to toolbox made after master was ready self.application_stack.register_postfork_function( - lambda: reload_toolbox(self, save_integrated_tool_panel=False), post_fork_only=True + lambda: reload_toolbox(self, save_integrated_tool_panel=False), + post_fork_only=True, ) # Delay toolbox index until after startup self.application_stack.register_postfork_function( @@ -975,6 +1033,11 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT def _shutdown_database_heartbeat(self): self.database_heartbeat.shutdown() + def _shutdown_history_audit_monitor(self): + monitor = getattr(self, "_history_audit_monitor", None) + if monitor: + monitor.shutdown() + def _shutdown_scheduling_manager(self): self.workflow_scheduling_manager.shutdown() diff --git a/lib/galaxy/config/sample/galaxy.yml.sample b/lib/galaxy/config/sample/galaxy.yml.sample index 524ad8dcf77..34ae86e4b1b 100644 --- a/lib/galaxy/config/sample/galaxy.yml.sample +++ b/lib/galaxy/config/sample/galaxy.yml.sample @@ -3129,6 +3129,20 @@ galaxy: # Message to display on the export citations tool page #citations_export_message_html: When writing up your analysis, remember to include all references that should be cited in order to completely describe your work. Also, please remember to cite Galaxy. + # Enables real-time history update notifications via Server-Sent + # Events (SSE). When enabled, a background monitor watches for history + # changes (via PostgreSQL LISTEN/NOTIFY or audit table polling as a + # fallback for SQLite) and pushes update signals to connected + # browsers, replacing aggressive 3-second polling. + #enable_sse_history_updates: false + + # The interval in seconds between history audit table polls when using + # the polling fallback (SQLite or when PostgreSQL LISTEN/NOTIFY is + # unavailable). Only used when enable_sse_history_updates is true. + # Lower values mean faster updates but more database queries. + # Recommended range: 1-5 seconds. + #history_audit_monitor_poll_interval: 2 + # Enables the Notification System integrated in Galaxy. # Users can receive automatic notifications when a certain resource is # shared with them or when some long running operations have finished, diff --git a/lib/galaxy/config/schemas/config_schema.yml b/lib/galaxy/config/schemas/config_schema.yml index 05c50b6eb39..aba88b87cee 100644 --- a/lib/galaxy/config/schemas/config_schema.yml +++ b/lib/galaxy/config/schemas/config_schema.yml @@ -4293,6 +4293,26 @@ mapping: desc: | Message to display on the export citations tool page + enable_sse_history_updates: + type: bool + default: false + required: false + desc: | + Enables real-time history update notifications via Server-Sent Events (SSE). + When enabled, a background monitor watches for history changes (via PostgreSQL + LISTEN/NOTIFY or audit table polling as a fallback for SQLite) and pushes update + signals to connected browsers, replacing aggressive 3-second polling. + + history_audit_monitor_poll_interval: + type: int + default: 2 + required: false + desc: | + The interval in seconds between history audit table polls when using the polling + fallback (SQLite or when PostgreSQL LISTEN/NOTIFY is unavailable). Only used when + enable_sse_history_updates is true. Lower values mean faster updates but more + database queries. Recommended range: 1-5 seconds. + enable_notification_system: type: bool default: false diff --git a/lib/galaxy/managers/history_audit_monitor.py b/lib/galaxy/managers/history_audit_monitor.py new file mode 100644 index 00000000000..4a9759bd837 --- /dev/null +++ b/lib/galaxy/managers/history_audit_monitor.py @@ -0,0 +1,263 @@ +"""Monitor for history audit table changes. + +Detects history changes via PostgreSQL LISTEN/NOTIFY (instant) or by polling +the history_audit table (SQLite fallback). Dispatches SSE events to connected +users via Kombu control tasks. + +Only active when ``enable_sse_history_updates`` is True in the Galaxy config. +""" + +import logging +import select +import threading +import time +from collections import ( + defaultdict, + OrderedDict, +) +from datetime import ( + datetime, + timedelta, +) +from typing import ( + Any, + Iterator, + Optional, +) + +from sqlalchemy import select as sa_select +from sqlalchemy.engine import Engine + +from galaxy.config import GalaxyAppConfiguration +from galaxy.managers.sse import SSEEventDispatcher +from galaxy.model import ( + History, + HistoryAudit, +) +from galaxy.model.mapping import GalaxyModelMapping + +log = logging.getLogger(__name__) + +CHANNEL_NAME = "galaxy_history_update" +OWNER_CACHE_MAX = 10_000 +DEBOUNCE_SECONDS = 0.2 + + +class _PgListenAdapter: + """Thin DBAPI-level adapter for PostgreSQL LISTEN/NOTIFY. + + Hides the receiving-API differences between psycopg2 (``conn.poll()`` + + ``conn.notifies`` list, driven by ``select.select``) and psycopg3 + (``conn.notifies(timeout=...)`` generator). The SA URL is used to inherit + DSN / SSL / auth config, but the connection itself is opened directly with + the DBAPI driver so it stays outside the SA pool — LISTEN connections must + live for the lifetime of the monitor and never be returned to the pool. + """ + + def __init__(self, engine: Engine) -> None: + # Strip the SA ``+driver`` suffix so the raw DBAPI libraries accept the URL. + dsn = engine.url.set(drivername="postgresql").render_as_string(hide_password=False) + driver = engine.dialect.driver + if driver == "psycopg": + import psycopg # conditional: psycopg3 driver + + self._conn: Any = psycopg.connect(dsn, autocommit=True) + self.driver = "psycopg3" + else: + import psycopg2 # conditional: psycopg2 driver + + self._conn = psycopg2.connect(dsn) + self._conn.autocommit = True # same effect as set_isolation_level(AUTOCOMMIT) + self.driver = "psycopg2" + + def listen(self, channel: str) -> None: + with self._conn.cursor() as cursor: + cursor.execute(f"LISTEN {channel};") + + def close(self) -> None: + try: + self._conn.close() + except Exception: + log.debug("Error closing LISTEN connection", exc_info=True) + + def poll(self, timeout: float) -> Iterator[str]: + """Block up to ``timeout`` seconds and yield notification payloads. + + Returns an empty iterator on timeout so callers can uniformly treat + "nothing received in this tick" regardless of driver. + """ + if self.driver == "psycopg3": + # psycopg3: notifies() is a blocking generator bounded by ``timeout``. + yield from (n.payload for n in self._conn.notifies(timeout=timeout)) + return + # psycopg2: block on the socket via select(), then drain notifies list. + if select.select([self._conn], [], [], timeout) == ([], [], []): + return + self._conn.poll() + while self._conn.notifies: + yield self._conn.notifies.pop(0).payload + + +class HistoryAuditMonitor: + """Background thread that monitors history_audit for changes and dispatches SSE events. + + On PostgreSQL: uses LISTEN/NOTIFY for instant notification. + On SQLite: polls history_audit table at a configurable interval. + """ + + def __init__( + self, + config: GalaxyAppConfiguration, + model: GalaxyModelMapping, + sse_dispatcher: SSEEventDispatcher, + ) -> None: + self._config = config + self._model = model + self._dispatcher = sse_dispatcher + self.poll_interval: int = config.history_audit_monitor_poll_interval + self._is_postgres: bool = "postgres" in model.engine.name + self._exit = threading.Event() + self._thread: Optional[threading.Thread] = None + self._active = False + # Bounded LRU cache: history_id -> user_id, refreshed on miss. + self._history_owner_cache: "OrderedDict[int, int]" = OrderedDict() + + def start(self) -> None: + if self._active: + return + self._active = True + target = self._listen_postgres if self._is_postgres else self._poll_audit_table + self._thread = threading.Thread( + target=target, + name="history_audit_monitor", + daemon=True, + ) + self._thread.start() + log.info( + "HistoryAuditMonitor started (mode=%s, interval=%ds)", + "pg_listen" if self._is_postgres else "poll", + self.poll_interval, + ) + + def shutdown(self) -> None: + self._active = False + self._exit.set() + if self._thread: + self._thread.join(timeout=5) + + # --- PostgreSQL LISTEN/NOTIFY mode --- + + def _listen_postgres(self) -> None: + """LISTEN for history update notifications. + + Works against both psycopg2 and psycopg3 — whichever driver the SA + engine was built with. Falls back to the SQLite polling path if the + DBAPI driver can't be imported or the initial LISTEN fails. + """ + try: + adapter = _PgListenAdapter(self._model.engine) + adapter.listen(CHANNEL_NAME) + except Exception: + log.warning( + "Failed to establish PostgreSQL LISTEN connection, falling back to polling", + exc_info=True, + ) + self._poll_audit_table() + return + + log.debug("LISTEN %s established (driver=%s)", CHANNEL_NAME, adapter.driver) + pending: dict[int, float] = {} # history_id -> first_seen_time + + try: + while not self._exit.is_set(): + received_any = False + for payload in adapter.poll(self.poll_interval): + received_any = True + try: + history_id = int(payload) + except (ValueError, TypeError): + continue + pending.setdefault(history_id, time.monotonic()) + + if not received_any: + # Timeout — flush anything that's been pending since last tick + if pending: + self._dispatch_history_updates(set(pending.keys())) + pending.clear() + continue + + # Debounce: dispatch events that have been pending long enough + now = time.monotonic() + ready = {hid for hid, ts in pending.items() if now - ts >= DEBOUNCE_SECONDS} + if ready: + self._dispatch_history_updates(ready) + for hid in ready: + del pending[hid] + except Exception: + log.exception("HistoryAuditMonitor LISTEN loop error") + finally: + adapter.close() + + # --- SQLite polling fallback --- + + def _poll_audit_table(self) -> None: + """Poll history_audit for recent changes.""" + last_check = datetime.utcnow() - timedelta(seconds=self.poll_interval) + + while not self._exit.is_set(): + try: + check_time = datetime.utcnow() + stmt = ( + sa_select(HistoryAudit.history_id) + .where(HistoryAudit.update_time > last_check) + .group_by(HistoryAudit.history_id) + ) + with self._model.new_session() as session: + changed_ids = set(session.scalars(stmt).all()) + + if changed_ids: + self._dispatch_history_updates(changed_ids) + + last_check = check_time + except Exception: + log.exception("HistoryAuditMonitor poll error") + + self._exit.wait(self.poll_interval) + + # --- Common dispatch logic --- + + def _dispatch_history_updates(self, history_ids: set[int]) -> None: + """Map history_ids to user_ids and send Kombu control task. + + Raw integer history IDs are sent across the control queue; encoding is + deferred to the ``history_update`` task handler on the receiving side, + keeping this manager free of presentation concerns. + """ + # Resolve owners for unknown history_ids + unknown = history_ids - self._history_owner_cache.keys() + if unknown: + self._refresh_owner_cache(unknown) + + user_updates: dict[str, list[int]] = defaultdict(list) + for history_id in history_ids: + user_id = self._history_owner_cache.get(history_id) + if user_id is not None: + user_updates[str(user_id)].append(history_id) + + if not user_updates: + return + + self._dispatcher.history_update(user_updates=dict(user_updates)) + + def _refresh_owner_cache(self, history_ids: set[int]) -> None: + """Look up user_id for given history_ids and update the bounded cache.""" + try: + stmt = sa_select(History.id, History.user_id).where(History.id.in_(history_ids)) + with self._model.new_session() as session: + for row in session.execute(stmt): + self._history_owner_cache[row[0]] = row[1] + self._history_owner_cache.move_to_end(row[0]) + while len(self._history_owner_cache) > OWNER_CACHE_MAX: + self._history_owner_cache.popitem(last=False) + except Exception: + log.debug("Failed to refresh history owner cache", exc_info=True) diff --git a/lib/galaxy/managers/notification.py b/lib/galaxy/managers/notification.py index f2c654ca6f9..d82140549b5 100644 --- a/lib/galaxy/managers/notification.py +++ b/lib/galaxy/managers/notification.py @@ -37,6 +37,7 @@ from galaxy.exceptions import ( ObjectNotFound, ) from galaxy.managers.markdown_util import to_html +from galaxy.managers.sse import SSEEventDispatcher from galaxy.model import ( GroupRoleAssociation, Notification, @@ -49,6 +50,7 @@ from galaxy.model.scoped_session import galaxy_scoped_session from galaxy.schema.notifications import ( AnyNotificationContent, BroadcastNotificationCreateRequest, + BroadcastNotificationResponse, MandatoryNotificationCategory, MessageNotificationContent, NewSharedItemNotificationContent, @@ -58,6 +60,7 @@ from galaxy.schema.notifications import ( NotificationCreateData, NotificationCreateRequest, NotificationRecipients, + NotificationResponse, NotificationVariant, PersonalNotificationCategory, UpdateUserNotificationPreferencesRequest, @@ -94,9 +97,15 @@ class NotificationChannelPlugin(Protocol): class NotificationManager: """Manager class to interact with the database models related with Notifications.""" - def __init__(self, sa_session: galaxy_scoped_session, config: GalaxyAppConfiguration): + def __init__( + self, + sa_session: galaxy_scoped_session, + config: GalaxyAppConfiguration, + sse_dispatcher: Optional[SSEEventDispatcher] = None, + ): self.sa_session = sa_session self.config = config + self.sse_dispatcher = sse_dispatcher self.recipient_resolver = NotificationRecipientResolver(strategy=DefaultStrategy(sa_session)) self.user_notification_columns: list[InstrumentedAttribute] = [ Notification.id, @@ -164,6 +173,10 @@ class NotificationManager: notifications_sent = self._create_associations(notification, recipient_users) self.sa_session.commit() + # Push SSE events to connected users via control queue + user_ids = [user.id for user in recipient_users] + self._notify_users_via_sse(user_ids, notification) + return notification, notifications_sent def _create_associations(self, notification: Notification, users: list[User]) -> int: @@ -179,6 +192,26 @@ class NotificationManager: continue return success_count + def _notify_users_via_sse(self, user_ids: list[int], notification: Notification) -> None: + """Broadcast a control task to all workers to push SSE events to connected users.""" + if not self.sse_dispatcher or not user_ids: + return + try: + payload = NotificationResponse.model_validate(notification).model_dump_json() + self.sse_dispatcher.notify_users(user_ids, payload) + except Exception: + log.warning("Failed to send SSE notification event", exc_info=True) + + def _notify_broadcast_via_sse(self, notification: Notification) -> None: + """Broadcast a control task to all workers to push SSE broadcast events.""" + if not self.sse_dispatcher: + return + try: + payload = BroadcastNotificationResponse.model_validate(notification).model_dump_json() + self.sse_dispatcher.notify_broadcast(payload) + except Exception: + log.warning("Failed to send SSE broadcast event", exc_info=True) + def dispatch_pending_notifications_via_channels(self) -> int: """ Dispatches all pending notifications to the users depending on the configured channels. @@ -273,6 +306,7 @@ class NotificationManager: notification = self._create_notification_model(request) self.sa_session.add(notification) self.sa_session.commit() + self._notify_broadcast_via_sse(notification) return notification def get_user_notification(self, user: User, notification_id: int, active_only: Optional[bool] = True): @@ -353,7 +387,10 @@ class NotificationManager: return result def update_user_notifications( - self, user: User, notification_ids: set[int], request: UserNotificationUpdateRequest + self, + user: User, + notification_ids: set[int], + request: UserNotificationUpdateRequest, ) -> int: """Updates a batch of notifications associated with the user using the requested values.""" updated_row_count = 0 @@ -447,7 +484,8 @@ class NotificationManager: UserNotificationAssociation.notification_id.in_(expired_notifications_stmt) ) result = cast( - CursorResult, self.sa_session.execute(delete_stmt, execution_options={"synchronize_session": False}) + CursorResult, + self.sa_session.execute(delete_stmt, execution_options={"synchronize_session": False}), ) deleted_associations_count = result.rowcount @@ -474,7 +512,10 @@ class NotificationManager: return notification def _user_notifications_query( - self, user: User, since: Optional[datetime] = None, active_only: Optional[bool] = True + self, + user: User, + since: Optional[datetime] = None, + active_only: Optional[bool] = True, ): stmt = ( select(*self.user_notification_columns) @@ -552,7 +593,7 @@ class DefaultStrategy(NotificationRecipientResolverStrategy): user_ids_from_roles_stmt = self._get_all_user_ids_from_roles_query(all_role_ids) union_stmt = union(user_ids_from_groups_stmt, user_ids_from_roles_stmt) - user_ids_from_groups_and_roles = {id for id, in self.sa_session.execute(union_stmt)} + user_ids_from_groups_and_roles = {id for (id,) in self.sa_session.execute(union_stmt)} unique_user_ids.update(user_ids_from_groups_and_roles) stmt = select(User).where(User.id.in_(unique_user_ids)) @@ -591,7 +632,7 @@ class DefaultStrategy(NotificationRecipientResolverStrategy): .where(GroupRoleAssociation.role_id.in_(role_ids)) .distinct() ) - group_ids_from_roles = {id for id, in self.sa_session.execute(stmt) if id is not None} + group_ids_from_roles = {id for (id,) in self.sa_session.execute(stmt) if id is not None} new_group_ids = group_ids_from_roles - processed_group_ids # Get role IDs associated with any of the given group IDs @@ -601,7 +642,7 @@ class DefaultStrategy(NotificationRecipientResolverStrategy): .where(GroupRoleAssociation.group_id.in_(group_ids)) .distinct() ) - role_ids_from_groups = {id for id, in self.sa_session.execute(stmt) if id is not None} + role_ids_from_groups = {id for (id,) in self.sa_session.execute(stmt) if id is not None} new_role_ids = role_ids_from_groups - processed_role_ids # Stop if there are no new group or role IDs to process @@ -713,7 +754,6 @@ class EmailNotificationTemplateBuilder(Protocol): class MessageEmailNotificationTemplateBuilder(EmailNotificationTemplateBuilder): - markdown_to = { TemplateFormats.HTML: to_html, TemplateFormats.TXT: lambda x: x, # TODO: strip markdown? @@ -730,9 +770,10 @@ class MessageEmailNotificationTemplateBuilder(EmailNotificationTemplateBuilder): class NewSharedItemEmailNotificationTemplateBuilder(EmailNotificationTemplateBuilder): - def get_content(self, template_format: TemplateFormats) -> AnyNotificationContent: - content = NewSharedItemNotificationContent.model_construct(**self.notification.content) # type: ignore[arg-type] + content = NewSharedItemNotificationContent.model_construct( + **self.notification.content + ) # type: ignore[arg-type] return content def get_subject(self) -> str: @@ -741,7 +782,6 @@ class NewSharedItemEmailNotificationTemplateBuilder(EmailNotificationTemplateBui class EmailNotificationChannelPlugin(NotificationChannelPlugin): - # Register the supported email templates here email_templates_by_category: dict[PersonalNotificationCategory, type[EmailNotificationTemplateBuilder]] = { PersonalNotificationCategory.message: MessageEmailNotificationTemplateBuilder, diff --git a/lib/galaxy/managers/sse.py b/lib/galaxy/managers/sse.py new file mode 100644 index 00000000000..d63532294f9 --- /dev/null +++ b/lib/galaxy/managers/sse.py @@ -0,0 +1,227 @@ +"""Server-Sent Events (SSE) connection manager for real-time notifications. + +Manages per-worker in-memory mapping of user IDs to asyncio.Queue instances, +enabling push of events from any thread (e.g. Kombu control queue worker) +to async SSE endpoint handlers running in the uvicorn event loop. +""" + +import asyncio +import logging +from collections import defaultdict +from dataclasses import dataclass +from datetime import datetime +from typing import ( + AsyncIterator, + Optional, + TYPE_CHECKING, +) + +if TYPE_CHECKING: + from starlette.requests import Request + + from galaxy.structured_app import MinimalManagerApp + +log = logging.getLogger(__name__) + + +@dataclass +class SSEEvent: + """An event to be sent to an SSE client.""" + + event: str # e.g. "notification_update", "broadcast_update", "notification_status" + data: str # JSON payload + id: Optional[str] = None # ISO timestamp, used by EventSource as Last-Event-ID on reconnect + + def to_wire(self) -> str: + """Serialize this event to the SSE wire format (``event:…\\ndata:…\\n[id:…\\n]\\n``).""" + frame = f"event: {self.event}\ndata: {self.data}\n" + if self.id: + frame += f"id: {self.id}\n" + return frame + "\n" + + +class SSEConnectionManager: + """Per-worker manager for SSE connections. + + Maps user_ids to sets of asyncio.Queue instances. Each SSE connection + gets its own queue. The manager is thread-safe for push operations + via ``loop.call_soon_threadsafe``. + + Lifecycle: + - Instantiated once per Galaxy worker process (on app object). + - ``connect()`` is called from the SSE async endpoint (event loop thread). + - ``disconnect()`` is called from the SSE endpoint's ``finally`` block. + - ``push_to_user()`` / ``push_broadcast()`` are called from ANY thread + (typically the Kombu daemon thread via control task handlers). + """ + + def __init__(self) -> None: + self._connections: dict[int, set[asyncio.Queue]] = defaultdict(set) + self._broadcast_connections: set[asyncio.Queue] = set() + self._loop: Optional[asyncio.AbstractEventLoop] = None + + def _ensure_loop(self) -> None: + """Capture the running asyncio event loop. Must be called from async context.""" + if self._loop is None or self._loop.is_closed(): + self._loop = asyncio.get_running_loop() + + # -- Called from ASYNC context (uvicorn event loop thread) -- + + def connect(self, user_id: Optional[int]) -> asyncio.Queue: + """Register a new SSE connection. Returns a queue to await events from. + + Called from the SSE endpoint handler (async context). A ``ready`` event is + enqueued immediately so that clients (and tests) can synchronize on the + server-side subscription rather than the underlying socket open event. + """ + self._ensure_loop() + queue: asyncio.Queue = asyncio.Queue(maxsize=64) + if user_id is not None: + self._connections[user_id].add(queue) + self._broadcast_connections.add(queue) + queue.put_nowait(SSEEvent(event="ready", data="")) + log.debug( + "SSE connection opened for user_id=%s (total=%d)", + user_id, + len(self._broadcast_connections), + ) + return queue + + def disconnect(self, user_id: Optional[int], queue: asyncio.Queue) -> None: + """Unregister an SSE connection. + + Called from the SSE endpoint's ``finally`` block (async context). + """ + if user_id is not None: + self._connections[user_id].discard(queue) + if not self._connections[user_id]: + del self._connections[user_id] + self._broadcast_connections.discard(queue) + log.debug( + "SSE connection closed for user_id=%s (total=%d)", + user_id, + len(self._broadcast_connections), + ) + + # -- Called from ANY thread (Kombu thread or async) -- + + def push_to_user(self, user_id: int, event: SSEEvent) -> None: + """Thread-safe. Push an event to all SSE connections for a specific user.""" + for queue in list(self._connections.get(user_id, [])): + self._safe_put(queue, event) + + def push_broadcast(self, event: SSEEvent) -> None: + """Thread-safe. Push an event to ALL connected SSE clients.""" + for queue in list(self._broadcast_connections): + self._safe_put(queue, event) + + def _safe_put(self, queue: asyncio.Queue, event: SSEEvent) -> None: + """Cross the thread boundary safely using ``call_soon_threadsafe``.""" + if self._loop is None or self._loop.is_closed(): + return + try: + self._loop.call_soon_threadsafe(self._do_put, queue, event) + except RuntimeError: + # Event loop is closed or shutting down + pass + + @staticmethod + def _do_put(queue: asyncio.Queue, event: SSEEvent) -> None: + """Runs ON the event loop thread. Safe to touch asyncio.Queue here.""" + try: + queue.put_nowait(event) + except asyncio.QueueFull: + log.warning("SSE queue full, dropping event: %s", event.event) + + @property + def connected_user_ids(self) -> set[int]: + return set(self._connections.keys()) + + @property + def total_connections(self) -> int: + return len(self._broadcast_connections) + + # -- High-level streaming helper -- + + async def stream( + self, + request: "Request", + user_id: Optional[int], + catch_up: Optional[SSEEvent] = None, + keepalive: float = 30.0, + ) -> AsyncIterator[str]: + """Yield SSE-framed strings for one connected client. + + Handles ``connect``, optional catch-up event priming, the main event + loop with a keepalive comment on timeout, disconnect detection, and + ``disconnect`` in ``finally``. Controllers should call this and return + the iterator wrapped in a ``StreamingResponse``. + """ + queue = self.connect(user_id) + if catch_up is not None: + await queue.put(catch_up) + try: + while True: + if await request.is_disconnected(): + break + try: + event: SSEEvent = await asyncio.wait_for(queue.get(), timeout=keepalive) + yield event.to_wire() + except asyncio.TimeoutError: + yield ": keepalive\n\n" + finally: + self.disconnect(user_id, queue) + + +class SSEEventDispatcher: + """Fans out SSE events across all Galaxy worker processes via the control queue. + + This is a thin wrapper around ``send_control_task`` so that managers can + depend on a narrow, injectable collaborator instead of importing the + queue-worker module directly. + + In Celery / background-task processes the app object has no ``queue_worker`` + (it's only built in ``UniverseApplication``), so dispatch is silently a + no-op there — notifications created from Celery tasks will still be + delivered the next time a client polls, just not pushed in real time. + """ + + def __init__(self, app: "MinimalManagerApp") -> None: + self._app = app + + def _send(self, task: str, kwargs: dict) -> None: + if getattr(self._app, "queue_worker", None) is None: + # No control-queue publisher available (e.g. Celery worker context). + log.debug("SSE dispatch skipped: app has no queue_worker (task=%s)", task) + return + from galaxy.queue_worker import send_control_task # circular: queue_worker -> app -> managers + + send_control_task(self._app, task, kwargs=kwargs, expiration=10) + + def notify_users(self, user_ids: list[int], payload: str, event_id: Optional[str] = None) -> None: + self._send( + "notify_users", + { + "user_ids": user_ids, + "payload": payload, + "event_id": event_id or datetime.utcnow().isoformat(), + }, + ) + + def notify_broadcast(self, payload: str, event_id: Optional[str] = None) -> None: + self._send( + "notify_broadcast", + { + "payload": payload, + "event_id": event_id or datetime.utcnow().isoformat(), + }, + ) + + def history_update(self, user_updates: dict[str, list], event_id: Optional[str] = None) -> None: + self._send( + "history_update", + { + "user_updates": user_updates, + "event_id": event_id or datetime.utcnow().isoformat(), + }, + ) diff --git a/lib/galaxy/model/triggers/update_audit_table.py b/lib/galaxy/model/triggers/update_audit_table.py index 3c06c05830e..9ce71e132a7 100644 --- a/lib/galaxy/model/triggers/update_audit_table.py +++ b/lib/galaxy/model/triggers/update_audit_table.py @@ -54,12 +54,18 @@ def _postgres_install(engine): RETURNS TRIGGER LANGUAGE 'plpgsql' AS $BODY$ + DECLARE + _history_id integer; BEGIN INSERT INTO history_audit (history_id, update_time) SELECT DISTINCT {id_field}, clock_timestamp() AT TIME ZONE 'UTC' FROM new_table WHERE {id_field} IS NOT NULL ON CONFLICT DO NOTHING; + FOR _history_id IN SELECT DISTINCT {id_field} FROM new_table WHERE {id_field} IS NOT NULL + LOOP + PERFORM pg_notify('galaxy_history_update', _history_id::text); + END LOOP; RETURN NULL; END; $BODY$ @@ -77,6 +83,7 @@ def _postgres_install(engine): INSERT INTO history_audit (history_id, update_time) VALUES (NEW.{id_field}, clock_timestamp() AT TIME ZONE 'UTC') ON CONFLICT DO NOTHING; + PERFORM pg_notify('galaxy_history_update', NEW.{id_field}::text); RETURN NULL; END; $BODY$ diff --git a/lib/galaxy/queue_worker/__init__.py b/lib/galaxy/queue_worker/__init__.py index 0e3d0f719b0..0da4b289117 100644 --- a/lib/galaxy/queue_worker/__init__.py +++ b/lib/galaxy/queue_worker/__init__.py @@ -4,6 +4,7 @@ reloading the toolbox, etc., across multiple processes. """ import importlib +import json import logging import math import socket @@ -27,6 +28,10 @@ from kombu.pools import producers import galaxy.queues from galaxy import util from galaxy.config import reload_config_options +from galaxy.managers.sse import ( + SSEConnectionManager, + SSEEvent, +) from galaxy.model import User from galaxy.tools import ToolBox from galaxy.tools.data_manager.manager import DataManagers @@ -43,7 +48,12 @@ if TYPE_CHECKING: ) -def send_local_control_task(app: "StructuredApp", task: str, get_response: bool = False, kwargs: Optional[dict] = None): +def send_local_control_task( + app: "StructuredApp", + task: str, + get_response: bool = False, + kwargs: Optional[dict] = None, +): """ This sends a message to the process-local control worker, which is useful for one-time asynchronous tasks like recalculating user disk usage. @@ -57,7 +67,15 @@ def send_local_control_task(app: "StructuredApp", task: str, get_response: bool return control_task.send_task(payload, routing_key, local=True, get_response=get_response) -def send_control_task(app, task, noop_self=False, get_response=False, routing_key="control.*", kwargs=None): +def send_control_task( + app, + task, + noop_self=False, + get_response=False, + routing_key="control.*", + kwargs=None, + expiration: Optional[int] = None, +): """ This sends a control task out to all processes, useful for things like reloading a data table, which needs to happen individually in all @@ -65,6 +83,7 @@ def send_control_task(app, task, noop_self=False, get_response=False, routing_ke Set noop_self to True to not run task for current process. Set get_response to True to wait for and return the task results as a list. + Set expiration to a number of seconds for message TTL. """ if kwargs is None: kwargs = {} @@ -73,7 +92,9 @@ def send_control_task(app, task, noop_self=False, get_response=False, routing_ke if noop_self: payload["noop"] = app.config.server_name control_task = ControlTask(app.queue_worker) - return control_task.send_task(payload=payload, routing_key=routing_key, get_response=get_response) + return control_task.send_task( + payload=payload, routing_key=routing_key, get_response=get_response, expiration=expiration + ) class ControlTask: @@ -107,7 +128,15 @@ class ControlTask: if message.properties["correlation_id"] == self.correlation_id: self.response = message.payload["result"] - def send_task(self, payload, routing_key, local=False, get_response=False, timeout=10): + def send_task( + self, + payload, + routing_key, + local=False, + get_response=False, + timeout=10, + expiration: Optional[int] = None, + ): if local: declare_queues = self.control_queues else: @@ -129,14 +158,24 @@ class ControlTask: correlation_id=self.correlation_id, retry=True, headers={"epoch": time.time()}, + expiration=expiration, ) if get_response: - with Consumer(self.connection, on_message=self.on_response, queues=callback_queue, no_ack=True): + with Consumer( + self.connection, + on_message=self.on_response, + queues=callback_queue, + no_ack=True, + ): while self.response is self._response: self.connection.drain_events(timeout=timeout) return self.response except TimeoutError: - log.exception("Error waiting for task: '%s' sent with routing key '%s'", payload, routing_key) + log.exception( + "Error waiting for task: '%s' sent with routing key '%s'", + payload, + routing_key, + ) except Exception: log.exception("Error queueing async task: '%s'. for %s", payload, routing_key) @@ -189,7 +228,10 @@ def _get_new_toolbox(app: "UniverseApplication", save_integrated_tool_panel: boo tool_configs = app.config.tool_configs new_toolbox = ToolBox( - tool_configs, app.config.tool_path, app, save_integrated_tool_panel=save_integrated_tool_panel + tool_configs, + app.config.tool_path, + app, + save_integrated_tool_panel=save_integrated_tool_panel, ) new_toolbox.data_manager_tools = app.toolbox.data_manager_tools app.datatypes_registry.load_datatype_converters(new_toolbox, use_cached=True) @@ -309,6 +351,44 @@ def admin_job_lock(app, **kwargs): log.info(f"Administrative Job Lock is now set to {job_lock}. Jobs will {'not' if job_lock else 'now'} dispatch.") +def notify_users(app, **kwargs): + """Push SSE events to connected users on this worker process.""" + sse_manager = app[SSEConnectionManager] + user_ids = kwargs.get("user_ids", []) + payload = kwargs.get("payload", "{}") + event_id = kwargs.get("event_id") + event = SSEEvent(event="notification_update", data=payload, id=event_id) + for user_id in user_ids: + sse_manager.push_to_user(user_id, event) + + +def notify_broadcast(app, **kwargs): + """Push SSE broadcast events to all connected clients on this worker process.""" + sse_manager = app[SSEConnectionManager] + payload = kwargs.get("payload", "{}") + event_id = kwargs.get("event_id") + event = SSEEvent(event="broadcast_update", data=payload, id=event_id) + sse_manager.push_broadcast(event) + + +def history_update(app, **kwargs): + """Push SSE history update events to connected users on this worker process. + + Encodes integer history IDs here (not in the monitor) so the manager layer + stays free of presentation/security concerns. + """ + sse_manager = app[SSEConnectionManager] + user_updates = kwargs.get("user_updates", {}) + event_id = kwargs.get("event_id") + encode = app.security.encode_id + for user_id_str, history_ids in user_updates.items(): + user_id = int(user_id_str) + encoded_ids = [encode(hid) for hid in history_ids] + data = json.dumps({"history_ids": encoded_ids}) + event = SSEEvent(event="history_update", data=data, id=event_id) + sse_manager.push_to_user(user_id, event) + + control_message_to_task = { "create_panel_section": create_panel_section, "reload_tool": reload_tool, @@ -324,6 +404,9 @@ control_message_to_task = { "reconfigure_watcher": reconfigure_watcher, "reload_tour": reload_tour, "reload_core_config": reload_core_config, + "notify_users": notify_users, + "notify_broadcast": notify_broadcast, + "history_update": history_update, } @@ -354,7 +437,14 @@ class GalaxyQueueWorker(ConsumerProducerMixin, threading.Thread): self.control_queues = [] self.epoch = 0 - def send_control_task(self, task, noop_self=False, get_response=False, routing_key="control.*", kwargs=None): + def send_control_task( + self, + task, + noop_self=False, + get_response=False, + routing_key="control.*", + kwargs=None, + ): return send_control_task( app=self.app, task=task, @@ -374,7 +464,10 @@ class GalaxyQueueWorker(ConsumerProducerMixin, threading.Thread): def bind_and_start(self): # This is post-forking, so we got the correct sever name - log.info("Binding and starting galaxy control worker for %s", self.app.config.server_name) + log.info( + "Binding and starting galaxy control worker for %s", + self.app.config.server_name, + ) self.exchange_queue, self.direct_queue = galaxy.queues.control_queues_from_config(self.app.config) self.control_queues = [self.exchange_queue, self.direct_queue] self.epoch = time.time() diff --git a/lib/galaxy/structured_app/__init__.py b/lib/galaxy/structured_app/__init__.py index f0fb04cb894..75b53be2ff1 100644 --- a/lib/galaxy/structured_app/__init__.py +++ b/lib/galaxy/structured_app/__init__.py @@ -50,6 +50,7 @@ if TYPE_CHECKING: from galaxy.managers.histories import HistoryManager from galaxy.managers.interactivetool import InteractiveToolManager from galaxy.managers.jobs import JobSearch + from galaxy.managers.sse import SSEConnectionManager from galaxy.managers.tools import DynamicToolManager from galaxy.managers.users import UserManager from galaxy.managers.workflows import ( @@ -57,7 +58,9 @@ if TYPE_CHECKING: WorkflowsManager, ) from galaxy.tool_shed.galaxy_install.client import DataManagersInterface - from galaxy.tool_shed.galaxy_install.installed_repository_manager import InstalledRepositoryManager + from galaxy.tool_shed.galaxy_install.installed_repository_manager import ( + InstalledRepositoryManager, + ) from galaxy.tool_util.data import ToolDataTableManager from galaxy.tools import ToolBox from galaxy.tools.cache import ToolCache @@ -174,6 +177,7 @@ class StructuredApp(MinimalManagerApp): vault: Vault webhooks_registry: WebhooksRegistry queue_worker: Any # 'galaxy.queue_worker.GalaxyQueueWorker' + sse_connection_manager: "SSEConnectionManager" data_provider_registry: Any # 'galaxy.visualization.data_providers.registry.DataProviderRegistry' tool_cache: "ToolCache" tool_shed_repository_cache: Optional[ToolShedRepositoryCache] diff --git a/lib/galaxy/webapps/galaxy/api/events.py b/lib/galaxy/webapps/galaxy/api/events.py new file mode 100644 index 00000000000..ad9a3b2e9d9 --- /dev/null +++ b/lib/galaxy/webapps/galaxy/api/events.py @@ -0,0 +1,66 @@ +""" +API endpoint for Server-Sent Events (SSE) stream. + +Provides a unified event stream for all real-time push events (notifications, +history updates, etc.) independent of the notification system configuration. +""" + +import logging +from typing import Optional + +from fastapi import ( + Header, + Request, +) +from starlette.responses import StreamingResponse + +from galaxy.managers.context import ProvidesUserContext +from galaxy.managers.sse import SSEConnectionManager +from galaxy.webapps.galaxy.services.notifications import NotificationService +from . import ( + depends, + DependsOnTrans, + Router, +) + +log = logging.getLogger(__name__) + +router = Router(tags=["events"]) + + +@router.cbv +class FastAPIEvents: + sse_manager: SSEConnectionManager = depends(SSEConnectionManager) + notifications: NotificationService = depends(NotificationService) + + @router.get( + "/api/events/stream", + summary="Server-Sent Events stream for real-time updates.", + response_class=StreamingResponse, + ) + async def stream_events( + self, + request: Request, + trans: ProvidesUserContext = DependsOnTrans, + last_event_id: Optional[str] = Header(None, alias="Last-Event-ID"), + ): + """Opens a Server-Sent Events (SSE) connection that pushes real-time + updates for notifications, history changes, and other events. + + On reconnect, the browser sends the ``Last-Event-ID`` header automatically. + If the notification system is enabled, any notifications created since that + timestamp are delivered as a catch-up ``notification_status`` event. + + Anonymous users receive only broadcast events. + """ + user_id = trans.user.id if not trans.anonymous else None + catch_up = self.notifications.build_status_catchup(trans, last_event_id) + return StreamingResponse( + self.sse_manager.stream(request, user_id, catch_up=catch_up), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) diff --git a/lib/galaxy/webapps/galaxy/api/notifications.py b/lib/galaxy/webapps/galaxy/api/notifications.py index f1bfbcab88a..ec490dc0dd1 100644 --- a/lib/galaxy/webapps/galaxy/api/notifications.py +++ b/lib/galaxy/webapps/galaxy/api/notifications.py @@ -10,12 +10,16 @@ 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.managers.sse import SSEConnectionManager from galaxy.schema.notifications import ( BroadcastNotificationCreateRequest, BroadcastNotificationListResponse, @@ -51,6 +55,39 @@ router = Router(tags=["notifications"]) @router.cbv class FastAPINotifications: service: NotificationService = depends(NotificationService) + sse_manager: SSEConnectionManager = depends(SSEConnectionManager) + + @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"), + ): + """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. + """ + self.service.notification_manager.ensure_notifications_enabled() + user_id = trans.user.id if not trans.anonymous else None + catch_up = self.service.build_status_catchup(trans, last_event_id) + return StreamingResponse( + self.sse_manager.stream(request, user_id, catch_up=catch_up), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) @router.get( "/api/notifications/status", diff --git a/lib/galaxy/webapps/galaxy/services/notifications.py b/lib/galaxy/webapps/galaxy/services/notifications.py index ae3e4494719..c11e0af8737 100644 --- a/lib/galaxy/webapps/galaxy/services/notifications.py +++ b/lib/galaxy/webapps/galaxy/services/notifications.py @@ -14,6 +14,7 @@ from galaxy.exceptions import ( ) from galaxy.managers.context import ProvidesUserContext from galaxy.managers.notification import NotificationManager +from galaxy.managers.sse import SSEEvent from galaxy.model import User from galaxy.schema.fields import Security from galaxy.schema.notifications import ( @@ -99,6 +100,28 @@ class NotificationService(ServiceBase): total_notifications_sent=1, notification=NotificationResponse.model_validate(notification) ) + def build_status_catchup( + self, user_context: ProvidesUserContext, last_event_id: Optional[str] + ) -> Optional[SSEEvent]: + """Build a ``notification_status`` SSE event covering everything since ``last_event_id``. + + Returns ``None`` when catch-up isn't possible (no ``Last-Event-ID``, + unparseable timestamp, or notifications disabled) so callers can simply + pass the result to ``SSEConnectionManager.stream`` without extra guards. + """ + if not last_event_id or not self.notification_manager.notifications_enabled: + return None + try: + since = datetime.fromisoformat(last_event_id) + except (ValueError, TypeError): + return None + catchup = self.get_notifications_status(user_context, since) + return SSEEvent( + event="notification_status", + data=catchup.model_dump_json(), + id=datetime.utcnow().isoformat(), + ) + def get_notifications_status(self, user_context: ProvidesUserContext, since: datetime) -> NotificationStatusSummary: """Returns the status of (unread or updated) notifications received by the user **since** a particular date and time. diff --git a/lib/galaxy_test/base/sse.py b/lib/galaxy_test/base/sse.py new file mode 100644 index 00000000000..47ee4172be4 --- /dev/null +++ b/lib/galaxy_test/base/sse.py @@ -0,0 +1,136 @@ +"""Shared helpers for SSE integration tests. + +The stream layer emits a ``ready`` event as the first frame on every connection +so tests can synchronize on the server-side subscription rather than the +underlying TCP socket. ``SSELineListener`` waits for that ``ready`` event before +``start()`` returns, and propagates listener-thread exceptions back to the main +thread instead of silently swallowing them. +""" + +import queue +import threading +from typing import ( + Optional, +) + +import requests + +from galaxy.util.wait import wait_on + +CONNECT_TIMEOUT = 15 +DEFAULT_WAIT_TIMEOUT = 15 + + +def parse_sse_events(raw: str) -> list[dict]: + """Parse raw SSE text into a list of event dicts with ``event``, ``data``, and ``id`` keys.""" + events: list[dict] = [] + current: dict[str, str] = {} + for line in raw.split("\n"): + if line.startswith(":"): + continue # comment / keepalive + if line == "": + if current: + events.append(current) + current = {} + continue + if ": " in line: + field, _, value = line.partition(": ") + else: + field, value = line.rstrip(":"), "" + if field in ("event", "data", "id"): + current[field] = value + if current: + events.append(current) + return events + + +class SSEListenerError(Exception): + """Wraps an exception raised inside the listener thread.""" + + +class SSELineListener: + """Runs an SSE connection on a background thread and collects raw chunks. + + ``start()`` blocks until the server-side ``ready`` event has been received, + guaranteeing that any event *posted after* ``start()`` returns will be seen + by this listener. Failures in the background thread are surfaced via + ``wait_for_event`` instead of silently timing out. + """ + + def __init__( + self, + url: str, + api_key: str, + headers: Optional[dict] = None, + timeout: int = 30, + ): + self.url = url + self.api_key = api_key + self.headers = headers or {} + self.timeout = timeout + self._collected: list[str] = [] + self._stop = threading.Event() + self._ready = threading.Event() + self._errors: "queue.Queue[BaseException]" = queue.Queue() + self._thread = threading.Thread(target=self._listen, daemon=True) + + def start(self) -> None: + self._thread.start() + wait_on( + lambda: True if self._ready.is_set() else None, + "SSE `ready` event", + timeout=CONNECT_TIMEOUT, + ) + self._raise_if_errored() + + def stop(self) -> None: + self._stop.set() + self._thread.join(timeout=5) + + def wait_for_event(self, event_type: str, timeout: int = DEFAULT_WAIT_TIMEOUT) -> list[dict]: + """Block until at least one event of ``event_type`` has been observed, then return all such events.""" + + def _check(): + self._raise_if_errored() + events = self.get_events(event_type) + return events if events else None + + return wait_on(_check, f"SSE {event_type} event", timeout=timeout) + + def get_events(self, event_type: Optional[str] = None) -> list[dict]: + """Return all collected events so far, optionally filtered by type.""" + all_events = parse_sse_events("".join(self._collected)) + if event_type is None: + return all_events + return [e for e in all_events if e.get("event") == event_type] + + def _raise_if_errored(self) -> None: + try: + err = self._errors.get_nowait() + except queue.Empty: + return + raise SSEListenerError(f"SSE listener thread failed: {err!r}") from err + + def _listen(self) -> None: + try: + resp = requests.get( + self.url, + params={"key": self.api_key}, + headers=self.headers, + stream=True, + timeout=self.timeout, + ) + if resp.status_code != 200: + raise RuntimeError(f"SSE connect returned HTTP {resp.status_code}: {resp.text[:200]}") + for chunk in resp.iter_content(chunk_size=None, decode_unicode=True): + if chunk: + self._collected.append(chunk) + if not self._ready.is_set() and "event: ready" in "".join(self._collected): + self._ready.set() + if self._stop.is_set(): + break + resp.close() + except BaseException as exc: + self._errors.put(exc) + # Ensure start() doesn't hang forever on connection failure. + self._ready.set() diff --git a/test/integration/test_history_sse.py b/test/integration/test_history_sse.py new file mode 100644 index 00000000000..6dd15c7afea --- /dev/null +++ b/test/integration/test_history_sse.py @@ -0,0 +1,131 @@ +"""Integration tests for SSE-based history update notifications.""" + +import json +from urllib.parse import urljoin +from uuid import uuid4 + +import requests + +from galaxy_test.base.populators import DatasetPopulator +from galaxy_test.base.sse import SSELineListener +from galaxy_test.driver.integration_util import IntegrationTestCase + + +class TestHistorySSEIntegration(IntegrationTestCase): + dataset_populator: DatasetPopulator + framework_tool_and_types = True + + @classmethod + def handle_galaxy_config_kwds(cls, config): + super().handle_galaxy_config_kwds(config) + config["enable_celery_tasks"] = False + config["enable_sse_history_updates"] = True + + def setUp(self): + super().setUp() + self.dataset_populator = DatasetPopulator(self.galaxy_interactor) + + def _events_stream_url(self) -> str: + return urljoin(self.url, "api/events/stream") + + def _create_history(self, name=None) -> str: + name = name or f"test_history_{uuid4()}" + response = self._post("histories", data={"name": name}, json=True) + self._assert_status_code_is_ok(response) + return response.json()["id"] + + def test_sse_events_endpoint_returns_event_stream(self): + """The /api/events/stream endpoint should return content-type text/event-stream.""" + response = requests.get( + self._events_stream_url(), + params={"key": self.galaxy_interactor.api_key}, + stream=True, + timeout=5, + ) + assert response.status_code == 200 + assert "text/event-stream" in response.headers.get("content-type", "") + response.close() + + def test_sse_receives_history_update_on_dataset_upload(self): + """When a dataset is uploaded, a history_update SSE event should be received.""" + history_id = self._create_history() + + listener = SSELineListener(self._events_stream_url(), self.galaxy_interactor.api_key) + listener.start() + try: + self.dataset_populator.new_dataset(history_id, wait=False) + history_events = listener.wait_for_event("history_update") + assert len(history_events) > 0 + finally: + listener.stop() + + def test_history_update_contains_current_history_id(self): + """The history_update event should contain the history's encoded ID.""" + history_id = self._create_history() + + listener = SSELineListener(self._events_stream_url(), self.galaxy_interactor.api_key) + listener.start() + try: + self.dataset_populator.new_dataset(history_id, wait=False) + history_events = listener.wait_for_event("history_update") + found = any(history_id in json.loads(e["data"]).get("history_ids", []) for e in history_events) + assert found, f"Expected history_id '{history_id}' in history_update events, got: {history_events}" + finally: + listener.stop() + + def test_history_update_is_scoped_to_owning_user(self): + """User A must only see history_update events for their own histories. + + Inverted positive assertion: after user B's upload, user A uploads to + their own history and we assert A's stream contains A's encoded id and + not B's. This avoids a sleep-based "no events" test that was prone to + flaking under slow CI. + """ + user_b = self._setup_user(f"{uuid4()}@galaxy.test") + _, user_b_api_key = self._setup_user_get_key(user_b["email"]) + + user_a_history_id = self._create_history() + + listener = SSELineListener(self._events_stream_url(), self.galaxy_interactor.api_key) + listener.start() + try: + # User B creates a history and uploads to it. User A must NOT see this. + create_resp = requests.post( + urljoin(self.url, "api/histories"), + params={"key": user_b_api_key}, + json={"name": "User B History"}, + ) + assert create_resp.status_code == 200 + user_b_history_id = create_resp.json()["id"] + + requests.post( + urljoin(self.url, f"api/histories/{user_b_history_id}/contents"), + params={"key": user_b_api_key}, + json={"from_hda_id": None, "source": "pasted", "content": "user b content"}, + ) + + # User A uploads to their own history — this is what A's stream must observe. + self.dataset_populator.new_dataset(user_a_history_id, wait=False) + history_events = listener.wait_for_event("history_update") + finally: + listener.stop() + + seen_ids: set[str] = set() + for event in history_events: + seen_ids.update(json.loads(event["data"]).get("history_ids", [])) + assert user_a_history_id in seen_ids, f"User A missed its own history_update: {history_events}" + assert ( + user_b_history_id not in seen_ids + ), f"User A received history_update for user B's history ({user_b_history_id}): {history_events}" + + def test_existing_polling_api_still_works(self): + """The existing current_history_json endpoint should continue to work.""" + url = urljoin(self.url, "history/current_history_json") + response = requests.get( + url, + params={"key": self.galaxy_interactor.api_key}, + ) + assert response.status_code == 200 + data = response.json() + assert "id" in data + assert "update_time" in data diff --git a/test/integration/test_notification_sse.py b/test/integration/test_notification_sse.py new file mode 100644 index 00000000000..5c8622514a4 --- /dev/null +++ b/test/integration/test_notification_sse.py @@ -0,0 +1,186 @@ +"""Integration tests for the notification SSE (Server-Sent Events) endpoint.""" + +from datetime import datetime +from typing import Optional +from urllib.parse import urljoin +from uuid import uuid4 + +import requests + +from galaxy_test.base.populators import DatasetPopulator +from galaxy_test.base.sse import SSELineListener +from galaxy_test.driver.integration_util import IntegrationTestCase + + +def notification_test_data(subject: Optional[str] = None, message: Optional[str] = None) -> dict: + return { + "source": "integration_tests", + "variant": "info", + "category": "message", + "content": { + "category": "message", + "subject": subject or "Testing Subject", + "message": message or "Testing Message", + }, + } + + +def notification_broadcast_test_data(subject: Optional[str] = None, message: Optional[str] = None) -> dict: + return { + "source": "integration_tests", + "variant": "info", + "category": "broadcast", + "content": { + "category": "broadcast", + "subject": subject or "Testing Broadcast Subject", + "message": message or "Testing Broadcast Message", + }, + } + + +class TestNotificationSSEIntegration(IntegrationTestCase): + dataset_populator: DatasetPopulator + framework_tool_and_types = False + + @classmethod + def handle_galaxy_config_kwds(cls, config): + super().handle_galaxy_config_kwds(config) + config["enable_celery_tasks"] = False + config["enable_notification_system"] = True + + def setUp(self): + super().setUp() + self.dataset_populator = DatasetPopulator(self.galaxy_interactor) + + def _stream_url(self) -> str: + return urljoin(self.url, "api/notifications/stream") + + def test_sse_endpoint_returns_event_stream(self): + """The SSE endpoint should return content-type text/event-stream.""" + response = requests.get( + self._stream_url(), + params={"key": self.galaxy_interactor.api_key}, + stream=True, + timeout=5, + ) + assert response.status_code == 200 + assert "text/event-stream" in response.headers.get("content-type", "") + response.close() + + def test_sse_receives_notification_events(self): + """When a notification is created, the SSE stream should receive it.""" + user = self._setup_user(f"{uuid4()}@galaxy.test") + _, user_api_key = self._setup_user_get_key(user["email"]) + + listener = SSELineListener(self._stream_url(), user_api_key) + listener.start() + try: + subject = f"sse_test_{uuid4()}" + request = { + "recipients": {"user_ids": [user["id"]]}, + "notification": notification_test_data(subject=subject, message="SSE test notification"), + } + response = self._post("notifications", data=request, admin=True, json=True) + self._assert_status_code_is_ok(response) + + notification_events = listener.wait_for_event("notification_update") + finally: + listener.stop() + + assert any( + subject in e.get("data", "") for e in notification_events + ), f"Expected subject '{subject}' in SSE events, got: {notification_events}" + + def test_sse_receives_broadcast_events(self): + """When a broadcast is created, the SSE stream should receive it.""" + listener = SSELineListener(self._stream_url(), self.galaxy_interactor.api_key) + listener.start() + try: + subject = f"broadcast_sse_test_{uuid4()}" + payload = notification_broadcast_test_data(subject=subject) + response = self._post("notifications/broadcast", data=payload, admin=True, json=True) + self._assert_status_code_is_ok(response) + + broadcast_events = listener.wait_for_event("broadcast_update") + finally: + listener.stop() + + assert any( + subject in e.get("data", "") for e in broadcast_events + ), f"Expected subject '{subject}' in broadcast SSE events, got: {broadcast_events}" + + def test_sse_catchup_on_reconnect(self): + """Reconnecting with Last-Event-ID should replay a catch-up notification_status event. + + The ``Last-Event-ID`` value is the server-issued ID from a prior event, + not a client-side ``datetime.utcnow()``. This avoids clock-skew flake + between the test runner and the app in containerized CI. + """ + user = self._setup_user(f"{uuid4()}@galaxy.test") + _, user_api_key = self._setup_user_get_key(user["email"]) + + # First connection: capture the server-issued event id of the first notification. + listener_1 = SSELineListener(self._stream_url(), user_api_key) + listener_1.start() + try: + subject_1 = f"first_{uuid4()}" + request = { + "recipients": {"user_ids": [user["id"]]}, + "notification": notification_test_data(subject=subject_1), + } + response = self._post("notifications", data=request, admin=True, json=True) + self._assert_status_code_is_ok(response) + first_events = listener_1.wait_for_event("notification_update") + finally: + listener_1.stop() + + last_event_id = next((e["id"] for e in first_events if e.get("id")), None) + assert last_event_id, f"No server-issued id on first notification event: {first_events}" + + # Emit a second notification while disconnected; it should appear in the catch-up. + subject_2 = f"catchup_{uuid4()}" + request = { + "recipients": {"user_ids": [user["id"]]}, + "notification": notification_test_data(subject=subject_2, message="Catch-up test"), + } + response = self._post("notifications", data=request, admin=True, json=True) + self._assert_status_code_is_ok(response) + + # Reconnect with Last-Event-ID = the captured id. The server catch-up runs before + # the `ready` event and must include the missed notification. + listener_2 = SSELineListener( + self._stream_url(), + user_api_key, + headers={"Last-Event-ID": last_event_id}, + ) + listener_2.start() + try: + status_events = listener_2.wait_for_event("notification_status") + finally: + listener_2.stop() + + assert any( + subject_2 in e.get("data", "") for e in status_events + ), f"Expected subject '{subject_2}' in catch-up event, got: {status_events}" + + def test_existing_polling_api_still_works(self): + """The existing polling endpoint should continue to work alongside SSE.""" + user = self._setup_user(f"{uuid4()}@galaxy.test") + + before = datetime.utcnow() + + subject = f"polling_test_{uuid4()}" + request = { + "recipients": {"user_ids": [user["id"]]}, + "notification": notification_test_data(subject=subject), + } + response = self._post("notifications", data=request, admin=True, json=True) + self._assert_status_code_is_ok(response) + + with self._different_user(user["email"]): + status_response = self._get(f"notifications/status?since={before.isoformat()}") + self._assert_status_code_is_ok(status_response) + status = status_response.json() + assert status["total_unread_count"] == 1 + assert len(status["notifications"]) == 1 + assert status["notifications"][0]["content"]["subject"] == subject diff --git a/test/integration_selenium/test_notification_sse.py b/test/integration_selenium/test_notification_sse.py new file mode 100644 index 00000000000..e3e9a564c84 --- /dev/null +++ b/test/integration_selenium/test_notification_sse.py @@ -0,0 +1,110 @@ +"""Playwright E2E test for the notification SSE pipeline. + +Verifies that when an admin creates a notification via the API, +a logged-in user sees it appear in the UI in real-time (within seconds) +without a page refresh, proving the SSE push pipeline works end-to-end. +""" + +from uuid import uuid4 + +from galaxy.util.wait import wait_on +from galaxy_test.selenium.framework import ( + managed_history, + selenium_test, +) +from .framework import SeleniumIntegrationTestCase + +SSE_CONNECT_TIMEOUT_SECONDS = 15 + + +class TestNotificationSSESeleniumIntegration(SeleniumIntegrationTestCase): + ensure_registered = True + + @classmethod + def handle_galaxy_config_kwds(cls, config): + super().handle_galaxy_config_kwds(config) + config["enable_notification_system"] = True + config["enable_celery_tasks"] = False + + def _wait_for_sse_connected(self) -> None: + """Block until the frontend confirms the SSE pipeline is live. + + Without this gate, the 30 s polling fallback silently masks a broken + SSE pipeline — the UI would still update, but via polling, and the + test would falsely pass. + """ + wait_on( + lambda: True if self.driver.execute_script("return window.__galaxy_sse_connected === true") else None, + "window.__galaxy_sse_connected === true", + timeout=SSE_CONNECT_TIMEOUT_SECONDS, + ) + + @selenium_test + @managed_history + def test_notification_appears_via_sse(self): + """Send a notification via the API and verify it appears in the UI without refresh.""" + # Get the logged-in user's info so we can send a notification to them + user_info = self._get("users/current").json() + user_id = user_info["id"] + + # Navigate to notifications page so the store is watching + self.driver.get(f"{self.target_url_from_selenium}/user/notifications") + self._wait_for_sse_connected() + self.screenshot("notification_sse_before") + + # Send a notification to this user via the admin API + subject = f"SSE E2E Test {uuid4()}" + notification_request = { + "recipients": {"user_ids": [user_id]}, + "notification": { + "source": "integration_tests", + "variant": "info", + "category": "message", + "content": { + "category": "message", + "subject": subject, + "message": "This notification was pushed via SSE", + }, + }, + } + response = self._post("notifications", data=notification_request, admin=True, json=True) + self._assert_status_code_is_ok(response) + + # Wait for the notification to appear in the UI — SSE should push it + # within a few seconds, without needing a page refresh. + # We wait up to 15 seconds checking for the subject text to appear. + self.driver.wait_for_selector_visible(f"text={subject}", timeout=15000) + self.screenshot("notification_sse_after") + + @selenium_test + @managed_history + def test_notification_bell_updates_via_sse(self): + """The notification bell indicator should update when a new notification arrives via SSE.""" + user_info = self._get("users/current").json() + user_id = user_info["id"] + + # Go to home page (bell is in masthead) + self.home() + self._wait_for_sse_connected() + + # Send a notification + subject = f"Bell Test {uuid4()}" + notification_request = { + "recipients": {"user_ids": [user_id]}, + "notification": { + "source": "integration_tests", + "variant": "info", + "category": "message", + "content": { + "category": "message", + "subject": subject, + "message": "Testing bell indicator update via SSE", + }, + }, + } + response = self._post("notifications", data=notification_request, admin=True, json=True) + self._assert_status_code_is_ok(response) + + # The indicator dot should appear on the bell (within the #activity-notifications element) + self.driver.wait_for_selector_visible("#activity-notifications .indicator", timeout=15000) + self.screenshot("notification_bell_indicator")