fix: elevate broadcast hostname resolution errors from DEBUG to ERROR

packages/opal-server/opal_server/pubsub.py

@@ -1,8 +1,10 @@
+import socket
 import time
 from contextlib import contextmanager
 from contextvars import ContextVar
 from threading import Lock
 from typing import Dict, Generator, List, Optional, Set, Tuple, Union, cast
+from urllib.parse import urlparse
 from uuid import UUID, uuid4
 
 from fastapi import APIRouter, Depends, WebSocket
@@ -141,7 +143,10 @@ def __init__(self, signer: JWTSigner, broadcaster_uri: str = None):
 
         self.broadcaster = None
         if broadcaster_uri is not None:
-            logger.info(f"Initializing broadcaster for server<->server communication")
+            logger.info(
+                "Initializing broadcaster for server<->server communication"
+            )
+            self._validate_broadcast_uri(broadcaster_uri)
             self.broadcaster = EventBroadcaster(
                 broadcaster_uri,
                 notifier=self.notifier,
@@ -202,6 +207,35 @@ async def websocket_rpc_endpoint(
             finally:
                 await websocket.close()
 
+    @staticmethod
+    def _validate_broadcast_uri(uri: str):
+        """Validate that the broadcast URI hostname can be resolved.
+
+        Logs an ERROR if DNS resolution fails so that operators can
+        quickly diagnose a misconfigured BROADCAST_URI.  Does not raise
+        -- the broadcaster is still created so existing behaviour is
+        preserved.
+        """
+        parsed = urlparse(uri)
+        hostname = parsed.hostname
+        if not hostname:
+            logger.error(
+                "Broadcast URI has no hostname: {uri}. "
+                "Check your OPAL_BROADCAST_URI configuration.",
+                uri=uri,
+            )
+            return
+        try:
+            socket.getaddrinfo(hostname, None)
+        except socket.gaierror as exc:
+            logger.error(
+                "Cannot resolve broadcast URI hostname '{hostname}': {error}. "
+                "Broadcasting will fail. "
+                "Check your OPAL_BROADCAST_URI configuration.",
+                hostname=hostname,
+                error=str(exc),
+            )
+
     @staticmethod
     async def _verify_permitted_topics(
         topics: Union[TopicList, ALL_TOPICS], channel: RpcChannel

packages/opal-server/opal_server/server.py

@@ -329,7 +329,7 @@ async def start_server_background_tasks(self):
                         await self.broadcast_listening_context.__aenter__()
                         # if the broadcast channel is closed, we want to restart worker process because statistics can't be reliable anymore
                         self.broadcast_listening_context._event_broadcaster.get_reader_task().add_done_callback(
-                            lambda _: self._graceful_shutdown()
+                            self._on_broadcaster_disconnected
                         )
                     asyncio.create_task(self.opal_statistics.run())
                     self.pubsub.endpoint.notifier.register_unsubscribe_event(
@@ -398,6 +398,29 @@ async def stop_server_background_tasks(self):
         except Exception:
             logger.exception("exception while shutting down background tasks")
 
+    def _on_broadcaster_disconnected(self, task: asyncio.Task):
+        """Callback fired when the broadcast listener task finishes.
+
+        Logs the underlying exception (e.g. ``gaierror``) at ERROR level
+        so that operators can diagnose misconfigured broadcast URIs
+        instead of having the error silently swallowed at DEBUG level.
+        """
+        try:
+            exc = task.exception()
+        except (asyncio.CancelledError, asyncio.InvalidStateError):
+            exc = None
+
+        if exc is not None:
+            logger.error(
+                "Broadcast channel connection failed: {error}. "
+                "Check your OPAL_BROADCAST_URI configuration.",
+                error=exc,
+            )
+        else:
+            logger.warning("Broadcast channel disconnected.")
+
+        self._graceful_shutdown()
+
     def _graceful_shutdown(self):
         logger.info("Trigger worker graceful shutdown")
         os.kill(os.getpid(), signal.SIGTERM)