diff --git a/packages/opal-server/opal_server/pubsub.py b/packages/opal-server/opal_server/pubsub.py index c7a3b875e..afd24add2 100644 --- a/packages/opal-server/opal_server/pubsub.py +++ b/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 diff --git a/packages/opal-server/opal_server/server.py b/packages/opal-server/opal_server/server.py index f320484ea..380ef5a8a 100644 --- a/packages/opal-server/opal_server/server.py +++ b/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)