Elect single history-audit monitor; add standalone daemon

Every webapp process was running its own HistoryAuditMonitor (and the
per-process Postgres LISTEN that comes with it). Move to a single elected
producer via a new is_history_audit_monitor role on DatabaseHeartbeat.
The election prefers any WorkerProcess with app_type=sse_monitor, falling
back to the max-server_name webapp when no dedicated daemon is running,
so the monitor migrates cleanly when the leader dies.

Add a standalone galaxy-sse-monitor process (new lib/galaxy/sse_monitor
package, galaxy-sse-monitor console script) that builds a minimal
GalaxyManagerApplication, registers itself as sse_monitor.<host>.<pid>,
and blocks on SIGTERM. Running this daemon moves SSE event production
out of the webapp so its dispatch can't stall behind a webapp's GIL.

Gate HistoryAuditMonitor.start/stop on the role callback rather than a
postfork hook; make start() restart-safe (clear _exit) and shutdown()
idempotent so role transitions can cycle the monitor.

lib/galaxy/app/__init__.py

@@ -1011,12 +1011,17 @@ 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
+        # History audit monitor for SSE-based history updates. The monitor only
+        # runs on the single process elected via DatabaseHeartbeat's
+        # is_history_audit_monitor role — a standalone ``galaxy-sse-monitor``
+        # daemon wins that election when present, otherwise one webapp picks it
+        # up. start/stop are driven by heartbeat role transitions rather than
+        # postfork, so the monitor cleanly migrates when the leader dies.
         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)
+            self.database_heartbeat.add_audit_monitor_change_callback(self._history_audit_monitor.on_role_change)
 
         # Start web stack message handling
         self.application_stack.register_postfork_function(self.application_stack.start)

lib/galaxy/managers/history_audit_monitor.py

@@ -130,6 +130,7 @@ class HistoryAuditMonitor:
         if self._active:
             return
         self._active = True
+        self._exit.clear()  # allow restart after a previous shutdown
         target = self._listen_postgres if self._is_postgres else self._poll_audit_table
         self._thread = threading.Thread(
             target=target,
@@ -144,10 +145,21 @@ class HistoryAuditMonitor:
         )
 
     def shutdown(self) -> None:
+        if not self._active:
+            return
         self._active = False
         self._exit.set()
         if self._thread:
             self._thread.join(timeout=5)
+        self._thread = None
+        log.info("HistoryAuditMonitor stopped")
+
+    def on_role_change(self, is_leader: bool) -> None:
+        """Heartbeat callback: start/stop the monitor as this process's election state changes."""
+        if is_leader:
+            self.start()
+        else:
+            self.shutdown()
 
     # --- PostgreSQL LISTEN/NOTIFY mode ---
 

lib/galaxy/model/database_heartbeat.py

@@ -17,6 +17,8 @@ from galaxy.model.orm.now import now
 log = logging.getLogger(__name__)
 
 WEBAPP = "webapp"  # WorkerProcess.app_type for web apps.
+SSE_MONITOR = "sse_monitor"  # WorkerProcess.app_type for the standalone SSE monitor process.
+SSE_MONITOR_SERVER_PREFIX = "sse_monitor."  # server_name prefix used by the standalone SSE monitor process.
 
 
 class DatabaseHeartbeat:
@@ -27,7 +29,9 @@ class DatabaseHeartbeat:
         self.hostname = socket.gethostname()
         self._engine = application_stack.app.model.engine
         self._is_config_watcher = False
+        self._is_history_audit_monitor = False
         self._observers = []
+        self._audit_monitor_observers = []
         self.exit = threading.Event()
         self.thread = None
         self.active = False
@@ -72,6 +76,9 @@ class DatabaseHeartbeat:
     def add_change_callback(self, callback):
         self._observers.append(callback)
 
+    def add_audit_monitor_change_callback(self, callback):
+        self._audit_monitor_observers.append(callback)
+
     @property
     def is_config_watcher(self):
         return self._is_config_watcher
@@ -83,6 +90,28 @@ class DatabaseHeartbeat:
         for callback in self._observers:
             callback(self._is_config_watcher)
 
+    @property
+    def is_history_audit_monitor(self):
+        return self._is_history_audit_monitor
+
+    @is_history_audit_monitor.setter
+    def is_history_audit_monitor(self, value):
+        self._is_history_audit_monitor = value
+        log.debug(
+            "%s %s history audit monitor",
+            self.server_name,
+            "is" if self._is_history_audit_monitor else "is not",
+        )
+        for callback in self._audit_monitor_observers:
+            callback(self._is_history_audit_monitor)
+
+    def _app_type(self):
+        if self.application_stack.app.is_webapp:
+            return WEBAPP
+        if self.server_name.startswith(SSE_MONITOR_SERVER_PREFIX):
+            return SSE_MONITOR
+        return None
+
     def update_watcher_designation(self):
         expression = self._worker_process_identifying_clause()
         stmt = select(WorkerProcess).with_for_update(of=WorkerProcess).where(expression)
@@ -90,19 +119,37 @@ class DatabaseHeartbeat:
             worker_process = session.scalars(stmt).first()
             if not worker_process:
                 worker_process = WorkerProcess(server_name=self.server_name, hostname=self.hostname)
-            if self.application_stack.app.is_webapp:
-                worker_process.app_type = WEBAPP
+            app_type = self._app_type()
+            if app_type is not None:
+                worker_process.app_type = app_type
             worker_process.update_time = now()
             worker_process.pid = self.pid
             session.add(worker_process)
+        active = list(self.get_active_processes(self.heartbeat_interval + 1))
         # We only want a single process watching the various config files on the file system.
         # We just pick the max server name for simplicity
-        webapp_servers = [
-            p.server_name for p in self.get_active_processes(self.heartbeat_interval + 1) if p.app_type == WEBAPP
-        ]
+        webapp_servers = [p.server_name for p in active if p.app_type == WEBAPP]
         is_config_watcher = bool(webapp_servers) and self.server_name == max(webapp_servers)
         if is_config_watcher != self.is_config_watcher:
             self.is_config_watcher = is_config_watcher
+        # The history-audit monitor is a single elected process too, but preference
+        # goes to a standalone sse_monitor daemon when one is running so the
+        # monitor's postgres LISTEN isn't blocked by webapp GIL pauses. If no
+        # dedicated process is registered we fall back to a webapp (same
+        # max-server_name tiebreaker as config_watcher).
+        audit_leader = self._elect_audit_leader(active, webapp_servers)
+        is_history_audit_monitor = audit_leader is not None and self.server_name == audit_leader
+        if is_history_audit_monitor != self.is_history_audit_monitor:
+            self.is_history_audit_monitor = is_history_audit_monitor
+
+    @staticmethod
+    def _elect_audit_leader(active, webapp_servers):
+        monitor_servers = [p.server_name for p in active if p.app_type == SSE_MONITOR]
+        if monitor_servers:
+            return min(monitor_servers)
+        if webapp_servers:
+            return max(webapp_servers)
+        return None
 
     def send_database_heartbeat(self):
         if self.active:

lib/galaxy/sse_monitor/__init__.py

@@ -0,0 +1,30 @@
+"""Standalone SSE monitor process.
+
+Runs the :class:`~galaxy.managers.history_audit_monitor.HistoryAuditMonitor`
+(and, in principle, any future SSE event producer) in its own OS process so a
+webapp's GIL pauses or fork-lifecycle events can never block history-update
+dispatch. Webapps still consume SSE events normally; only the *producer* moves.
+
+The process registers itself via :class:`~galaxy.model.database_heartbeat.DatabaseHeartbeat`
+with a server_name beginning with ``sse_monitor.``. The heartbeat's
+``is_history_audit_monitor`` election prefers any such process, so the monitor
+automatically migrates here when this daemon is running. If it stops, a webapp
+is re-elected within one heartbeat interval (~60s).
+
+Starting the daemon
+-------------------
+
+Installed::
+
+    GALAXY_CONFIG_FILE=/etc/galaxy/galaxy.yml galaxy-sse-monitor
+
+From a source checkout::
+
+    GALAXY_CONFIG_FILE=config/galaxy.yml python -m galaxy.sse_monitor
+
+For production deployments, run under systemd / supervisord alongside the
+webapp. A native ``gravity`` (``galaxyctl``) service type for this daemon is
+tracked as a follow-up against the ``gravity`` package — until that lands,
+either start the process via your process manager of choice, or omit it and
+let a webapp fall back to the monitor role.
+"""

lib/galaxy/sse_monitor/__main__.py

@@ -0,0 +1,101 @@
+"""Entry point for the standalone ``galaxy-sse-monitor`` daemon.
+
+Loads a minimal :class:`GalaxyManagerApplication` (same shape the Celery
+workers use), forces the server_name to ``sse_monitor.<host>.<pid>`` so
+DatabaseHeartbeat's election picks this process as the history-audit monitor,
+and blocks on SIGINT/SIGTERM.
+"""
+
+import logging
+import os
+import signal
+import socket
+import sys
+import threading
+
+from galaxy.celery import get_app_properties
+from galaxy.model.database_heartbeat import DatabaseHeartbeat
+
+log = logging.getLogger("galaxy.sse_monitor")
+
+
+def _build_server_name() -> str:
+    return f"sse_monitor.{socket.gethostname()}.{os.getpid()}"
+
+
+def _build_app(server_name: str):
+    kwargs = get_app_properties() or {}
+    if not kwargs:
+        raise RuntimeError(
+            "GALAXY_CONFIG_FILE (or GALAXY_ROOT_DIR with an on-disk Galaxy config) is required "
+            "to start galaxy-sse-monitor"
+        )
+    kwargs = dict(kwargs)
+    kwargs["check_migrate_databases"] = False
+    kwargs["use_display_applications"] = False
+    kwargs["use_converters"] = False
+    kwargs["server_name"] = server_name
+
+    import galaxy.app
+
+    return galaxy.app.GalaxyManagerApplication(configure_logging=True, **kwargs)
+
+
+def main() -> int:
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
+
+    server_name = _build_server_name()
+    log.info("Starting galaxy-sse-monitor as %s", server_name)
+
+    app = _build_app(server_name)
+
+    # GalaxyManagerApplication doesn't wire a DatabaseHeartbeat (that lives on
+    # UniverseApplication, which pulls in the webapp stack we don't need). We
+    # spin up our own and register the audit-monitor callback so election
+    # transitions start/stop the producer cleanly.
+    heartbeat = DatabaseHeartbeat(application_stack=app.application_stack)
+
+    monitor = None
+    if app.config.enable_sse_history_updates:
+        from galaxy.managers.history_audit_monitor import HistoryAuditMonitor
+
+        monitor = app[HistoryAuditMonitor]
+        heartbeat.add_audit_monitor_change_callback(monitor.on_role_change)
+    else:
+        log.warning(
+            "enable_sse_history_updates is False — galaxy-sse-monitor will idle with no producers"
+        )
+
+    heartbeat.start()
+
+    shutdown = threading.Event()
+
+    def _handle_signal(signum, _frame):
+        log.info("Received signal %s, shutting down galaxy-sse-monitor", signum)
+        shutdown.set()
+
+    signal.signal(signal.SIGINT, _handle_signal)
+    signal.signal(signal.SIGTERM, _handle_signal)
+
+    try:
+        shutdown.wait()
+    finally:
+        if monitor is not None:
+            try:
+                monitor.shutdown()
+            except Exception:
+                log.exception("Error shutting down HistoryAuditMonitor")
+        try:
+            heartbeat.shutdown()
+        except Exception:
+            log.exception("Error shutting down database heartbeat")
+        try:
+            app.shutdown()
+        except Exception:
+            log.exception("Error shutting down GalaxyManagerApplication")
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

packages/app/setup.cfg

@@ -101,6 +101,7 @@ test =
 console_scripts =
         galaxy-main = galaxy.main:main
         galaxy-dependencies = galaxy.dependencies.script:main
+        galaxy-sse-monitor = galaxy.sse_monitor.__main__:main
 
 [options.packages.find]
 where = src

packages/app/src/galaxy/sse_monitor

@@ -0,0 +1 @@
+../../../../lib/galaxy/sse_monitor