diff --git a/WebATM-integrated/tests/test_auto_start.py b/WebATM-integrated/tests/test_auto_start.py index 48545ba..bdc6ce3 100644 --- a/WebATM-integrated/tests/test_auto_start.py +++ b/WebATM-integrated/tests/test_auto_start.py @@ -118,7 +118,7 @@ def test_claim_degrades_to_true_when_marker_uncreatable(tmp_path): def test_connects_once_ports_come_up(): """Ports come up after a couple of polls; then connect + subscribe, in order.""" proxy = FakeProxy(server_ip="localhost") - registered: list[str] = [] + registered: list[object] = [] ready = {"v": False} sleeps = {"n": 0} @@ -130,9 +130,9 @@ def fake_sleep(_): def fake_lister(port, timeout, host): return ready["v"] - def fake_register(): + def fake_register(p): proxy.events.append("register_subscribers") - registered.append("ok") + registered.append(p) result = auto_start.connect_proxy_when_ready( proxy, @@ -140,12 +140,15 @@ def fake_register(): poll_interval=0.01, is_port_listening=fake_lister, register_subscribers=fake_register, + get_proxy=lambda: proxy, sleep=fake_sleep, ) assert result is True assert proxy.start_client_hosts == ["localhost"] - assert registered == ["ok"] + # Subscribers must attach to the exact proxy that was connected, not to + # whatever the global happens to be at registration time. + assert registered == [proxy] # Subscribers must be registered AFTER the client is created by start_client. assert proxy.events == ["start_client", "register_subscribers"] assert sleeps["n"] >= 2 @@ -154,14 +157,15 @@ def fake_register(): def test_gives_up_when_ports_never_listen(): """No port ever opens -> no connect attempt, returns False.""" proxy = FakeProxy() - registered: list[str] = [] + registered: list[object] = [] result = auto_start.connect_proxy_when_ready( proxy, ready_timeout=0.05, poll_interval=0.01, is_port_listening=_always(False), - register_subscribers=lambda: registered.append("ok"), + register_subscribers=registered.append, + get_proxy=lambda: proxy, sleep=lambda _: None, ) @@ -219,7 +223,8 @@ def fake_lister(port, timeout, host): proxy, host="explicit", is_port_listening=fake_lister, - register_subscribers=lambda: None, + register_subscribers=lambda p: None, + get_proxy=lambda: proxy, sleep=lambda _: None, ) @@ -235,7 +240,8 @@ def test_host_falls_back_to_env_when_proxy_has_none(monkeypatch): auto_start.connect_proxy_when_ready( proxy, is_port_listening=_always(True), - register_subscribers=lambda: None, + register_subscribers=lambda p: None, + get_proxy=lambda: proxy, sleep=lambda _: None, ) @@ -245,12 +251,13 @@ def test_host_falls_back_to_env_when_proxy_has_none(monkeypatch): def test_connect_failure_is_caught(): """A start_client that raises is reported as a failed (False) connect.""" proxy = FakeProxy(raise_on_start=True) - registered: list[str] = [] + registered: list[object] = [] result = auto_start.connect_proxy_when_ready( proxy, is_port_listening=_always(True), - register_subscribers=lambda: registered.append("ok"), + register_subscribers=registered.append, + get_proxy=lambda: proxy, sleep=lambda _: None, ) @@ -260,6 +267,71 @@ def test_connect_failure_is_caught(): assert registered == [] +def test_stands_down_when_manual_connect_replaced_proxy(): + """A manual connect during the port wait wins; the stale proxy stays down. + + /api/server/config replaces the global proxy (and closes the boot-time one + captured by auto-start). Connecting the captured proxy anyway would leave a + second, subscriber-less ZMQ client alive whose data-flow timeout later + broadcasts a bogus disconnect -- so auto-start must not touch it. + """ + boot_proxy = FakeProxy() + manual_proxy = FakeProxy() # what /api/server/config installed meanwhile + registered: list[object] = [] + + result = auto_start.connect_proxy_when_ready( + boot_proxy, + is_port_listening=_always(True), + register_subscribers=registered.append, + get_proxy=lambda: manual_proxy, + sleep=lambda _: None, + ) + + assert result is False + assert boot_proxy.start_client_hosts == [] + assert manual_proxy.start_client_hosts == [] + assert registered == [] + + +def test_takeover_check_and_connect_run_under_the_lock(): + """The global-proxy re-check and the connect are one atomic critical section. + + The lock is shared with /api/server/config, so the route can never swap the + global proxy between auto-start's identity check and its start_client. + """ + proxy = FakeProxy() + + class FakeLock: + def __enter__(self): + proxy.events.append("lock_enter") + + def __exit__(self, *exc): + proxy.events.append("lock_exit") + return False + + def fake_get_proxy(): + proxy.events.append("get_proxy") + return proxy + + result = auto_start.connect_proxy_when_ready( + proxy, + is_port_listening=_always(True), + register_subscribers=lambda p: proxy.events.append("register_subscribers"), + get_proxy=fake_get_proxy, + lock=FakeLock(), + sleep=lambda _: None, + ) + + assert result is True + assert proxy.events == [ + "lock_enter", + "get_proxy", + "start_client", + "register_subscribers", + "lock_exit", + ] + + # -------------------------------------------------------------------------- # _run_auto_start # -------------------------------------------------------------------------- diff --git a/WebATM-integrated/webatm_integrated/auto_start.py b/WebATM-integrated/webatm_integrated/auto_start.py index b06a1b3..a7ef869 100644 --- a/WebATM-integrated/webatm_integrated/auto_start.py +++ b/WebATM-integrated/webatm_integrated/auto_start.py @@ -10,6 +10,10 @@ This mirrors -- server-side and automatically -- the exact connect sequence the manual ``/api/server/config`` route performs (``start_client`` then ``register_subscribers``; subscribers can only attach once the client exists). +If the user connects manually while auto-start is still waiting for BlueSky's +ports, the manual route wins: auto-start detects that the global proxy was +replaced and stands down instead of reviving the stale boot-time proxy (see +``connect_proxy_when_ready``). Opt out with ``WEBATM_AUTO_START=0`` (e.g. for tests, or deployments that want the manual Start button to drive the lifecycle). The core ``webatm`` package @@ -22,6 +26,7 @@ import os import time from collections.abc import Callable +from contextlib import AbstractContextManager from WebATM.logger import get_logger @@ -135,7 +140,9 @@ def connect_proxy_when_ready( ready_timeout: float = 60.0, poll_interval: float = 0.5, is_port_listening: Callable[..., bool] | None = None, - register_subscribers: Callable[[], None] | None = None, + register_subscribers: Callable[..., None] | None = None, + get_proxy: Callable[[], object] | None = None, + lock: AbstractContextManager | None = None, sleep: Callable[[float], None] | None = None, ) -> bool: """Wait for BlueSky to accept connections, then connect the WebATM proxy. @@ -145,8 +152,21 @@ def connect_proxy_when_ready( ``start_client`` followed by ``register_subscribers`` (subscribers attach to the client created by ``start_client``). - The port probe, subscriber registration and sleep are injectable so this can - be unit-tested without a real BlueSky server or wall-clock delays. + The port wait can span many seconds on a cold start, and during it the user + may connect manually: the ``/api/server/config`` route replaces the global + proxy with a fresh one and closes the boot-time proxy captured here. The + connect step therefore runs under the shared ``WebATM.proxy.connect_lock`` + and first re-checks that the global proxy is still ``bluesky_proxy``; if a + manual connect took over, auto-start stands down. Without this, connecting + the stale proxy would leave a second ZMQ client alive with no subscribers + (``register_subscribers`` resolves the *global* proxy), whose data-flow + timeout would later broadcast a bogus disconnect and blank the map while + the real proxy streams fine. Subscribers are likewise registered on + ``bluesky_proxy`` explicitly, never on whatever the global happens to be. + + The port probe, subscriber registration, proxy getter, lock and sleep are + injectable so this can be unit-tested without a real BlueSky server or + wall-clock delays. Args: bluesky_proxy (BlueSkyProxy): Core proxy to connect. @@ -159,13 +179,21 @@ def connect_proxy_when_ready( is_port_listening (Callable | None): Port probe taking ``(port, timeout, host)``. Defaults to ``WebATM.server.bluesky_server_status.is_port_listening``. - register_subscribers (Callable | None): Subscriber-registration hook. - Defaults to ``WebATM.proxy.register_subscribers``. + register_subscribers (Callable | None): Subscriber-registration hook + taking the proxy to attach to. Defaults to + ``WebATM.proxy.register_subscribers``. + get_proxy (Callable | None): Returns the current global proxy, used to + detect a manual connect having replaced it. Defaults to + ``WebATM.proxy.get_bluesky_proxy``. + lock (AbstractContextManager | None): Lock held around the connect + step. Defaults to ``WebATM.proxy.connect_lock``, shared with the + manual ``/api/server/config`` route. sleep (Callable | None): Sleep function. Defaults to ``time.sleep``. Returns: bool: True if the proxy connect succeeded, False if BlueSky never came - up within the timeout or the connect attempt raised. + up within the timeout, the connect attempt raised, or a manual + connect replaced the proxy first. """ # Deferred imports: keep this module light for unit tests and avoid pulling # the Flask/ZMQ-laden core packages unless we actually connect. @@ -173,6 +201,10 @@ def connect_proxy_when_ready( from WebATM.server.bluesky_server_status import is_port_listening if register_subscribers is None: from WebATM.proxy import register_subscribers + if get_proxy is None: + from WebATM.proxy import get_bluesky_proxy as get_proxy + if lock is None: + from WebATM.proxy import connect_lock as lock if sleep is None: sleep = time.sleep @@ -191,18 +223,26 @@ def connect_proxy_when_ready( ) return False - try: - bluesky_proxy.server_ip = host - bluesky_proxy.start_client(hostname=host) - # Subscribers can only be registered once the client exists, which - # start_client creates -- this is the same ordering the manual - # /api/server/config route relies on. - register_subscribers() - logger.info(f"Auto-start: WebATM proxy connected to BlueSky at '{host}'") - return True - except Exception as e: - logger.error(f"Auto-start: failed to connect proxy to BlueSky: {e}") - return False + with lock: + if get_proxy() is not bluesky_proxy: + logger.info( + "Auto-start: a manual connect replaced the proxy while waiting " + "for BlueSky; standing down" + ) + return False + try: + bluesky_proxy.server_ip = host + bluesky_proxy.start_client(hostname=host) + # Subscribers can only be registered once the client exists, which + # start_client creates -- this is the same ordering the manual + # /api/server/config route relies on. They attach to bluesky_proxy + # explicitly, so they always land on the client just started. + register_subscribers(bluesky_proxy) + logger.info(f"Auto-start: WebATM proxy connected to BlueSky at '{host}'") + return True + except Exception as e: + logger.error(f"Auto-start: failed to connect proxy to BlueSky: {e}") + return False def _wait_for_ports( diff --git a/WebATM/proxy/__init__.py b/WebATM/proxy/__init__.py index 0790863..3f085ad 100644 --- a/WebATM/proxy/__init__.py +++ b/WebATM/proxy/__init__.py @@ -8,12 +8,20 @@ - Subscriber registration for network events """ +import threading + from .core import BlueSkyProxy from .subscribers import register_subscribers # Global BlueSky proxy instance to be set by the app _bluesky_proxy = None +# Serializes replacing and connecting the global proxy. Held by the manual +# /api/server/config route and by the integrated auto-start's connect step, so +# one of them can never revive a proxy the other has just replaced/closed +# (which would leave a second, subscriber-less ZMQ client alive). +connect_lock = threading.Lock() + def get_bluesky_proxy(): """Get the current BlueSky proxy instance. @@ -41,4 +49,5 @@ def set_bluesky_proxy(proxy): "register_subscribers", "get_bluesky_proxy", "set_bluesky_proxy", + "connect_lock", ] diff --git a/WebATM/proxy/subscribers.py b/WebATM/proxy/subscribers.py index 5f05417..5d605b1 100644 --- a/WebATM/proxy/subscribers.py +++ b/WebATM/proxy/subscribers.py @@ -47,22 +47,31 @@ ] -def register_subscribers(): +def register_subscribers(proxy=None): """Register all handler callbacks with the proxy's BlueSky client. Iterates over ``SUBSCRIPTIONS`` and subscribes each (topic, callback, - actonly) triple on the global proxy's network client. Topics flagged - ``actonly`` only deliver data for the active node and are re-subscribed - when the active node changes. + actonly) triple on the proxy's network client. Topics flagged ``actonly`` + only deliver data for the active node and are re-subscribed when the + active node changes. - Logs an error and returns early if no global proxy is set, or a warning + Logs an error and returns early if no proxy is available, or a warning if the proxy has no connected BlueSky client yet. + + Args: + proxy (BlueSkyProxy | None): Proxy whose client to attach to. Defaults + to the globally registered proxy. Callers that just connected a + specific proxy instance should pass it explicitly, so a concurrent + reconnect swapping the global can never leave the client they + started without subscribers. """ - # Imported lazily: WebATM.proxy imports this module while it is still being - # initialised, so get_bluesky_proxy does not exist at module-load time yet. - from . import get_bluesky_proxy + if proxy is None: + # Imported lazily: WebATM.proxy imports this module while it is still + # being initialised, so get_bluesky_proxy does not exist at module-load + # time yet. + from . import get_bluesky_proxy - proxy = get_bluesky_proxy() + proxy = get_bluesky_proxy() if not proxy: logger.error("No proxy available for subscriber registration") return diff --git a/WebATM/server/routes.py b/WebATM/server/routes.py index f186bba..09d2395 100644 --- a/WebATM/server/routes.py +++ b/WebATM/server/routes.py @@ -264,30 +264,41 @@ def update_server_config(): server_ip = data.get("server_ip", "localhost").strip() or "localhost" logger.info(f"User requested connection to BlueSky server at {server_ip}") - from ..proxy import BlueSkyProxy, register_subscribers, set_bluesky_proxy + from ..proxy import ( + BlueSkyProxy, + connect_lock, + register_subscribers, + set_bluesky_proxy, + ) # Every (re)connect gets a completely fresh proxy: recreating the # ZMQ client is the reliable way to shed any half-dead connection # state. Only the Socket.IO wiring carries over. The old proxy is # replaced in place (never deleted) so concurrent requests always - # find a usable current_app.bluesky_proxy. - old_proxy = getattr(current_app, "bluesky_proxy", None) - if old_proxy is not None: - if old_proxy.running: - old_proxy.stop_client() - time.sleep(0.3) # let ZMQ teardown settle before reconnecting - old_proxy.close() - - proxy = BlueSkyProxy() - proxy.socketio = old_proxy.socketio if old_proxy else None - proxy.connected_clients = old_proxy.connected_clients if old_proxy else 0 - current_app.bluesky_proxy = proxy - set_bluesky_proxy(proxy) # update the global the subscribers use - - proxy.server_ip = server_ip - proxy.start_client(hostname=server_ip) - # Subscribers attach to the client start_client just created. - register_subscribers() + # find a usable current_app.bluesky_proxy. The swap-and-connect + # runs under connect_lock so the integrated auto-start (or another + # concurrent connect request) can never revive the proxy this + # request is tearing down. + with connect_lock: + old_proxy = getattr(current_app, "bluesky_proxy", None) + if old_proxy is not None: + if old_proxy.running: + old_proxy.stop_client() + time.sleep(0.3) # let ZMQ teardown settle before reconnecting + old_proxy.close() + + proxy = BlueSkyProxy() + proxy.socketio = old_proxy.socketio if old_proxy else None + proxy.connected_clients = ( + old_proxy.connected_clients if old_proxy else 0 + ) + current_app.bluesky_proxy = proxy + set_bluesky_proxy(proxy) # update the global the subscribers use + + proxy.server_ip = server_ip + proxy.start_client(hostname=server_ip) + # Subscribers attach to the client start_client just created. + register_subscribers(proxy) # Confirm the server is real: wait for node detection. timeout = 10.0 diff --git a/tests/test_app.py b/tests/test_app.py index a94d066..d9d99ed 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -150,9 +150,7 @@ def test_post_config_replaces_proxy_and_connects(self, app_and_client, monkeypat created: list = [] registered: list = [] monkeypatch.setattr(proxy_pkg, "BlueSkyProxy", _fake_proxy_class(created)) - monkeypatch.setattr( - proxy_pkg, "register_subscribers", lambda: registered.append(True) - ) + monkeypatch.setattr(proxy_pkg, "register_subscribers", registered.append) old_proxy = app.bluesky_proxy old_proxy.connected_clients = 3 @@ -172,7 +170,8 @@ def test_post_config_replaces_proxy_and_connects(self, app_and_client, monkeypat assert new_proxy.server_ip == "10.0.0.5" assert new_proxy.socketio is old_proxy.socketio assert new_proxy.connected_clients == 3 - assert registered == [True] + # Subscribers are registered on the freshly connected proxy explicitly. + assert registered == [new_proxy] def test_post_config_connect_failure_returns_500(self, app_and_client, monkeypatch): import WebATM.proxy as proxy_pkg @@ -186,7 +185,7 @@ def boom(proxy, hostname): monkeypatch.setattr( proxy_pkg, "BlueSkyProxy", _fake_proxy_class(created, start_client=boom) ) - monkeypatch.setattr(proxy_pkg, "register_subscribers", lambda: None) + monkeypatch.setattr(proxy_pkg, "register_subscribers", lambda proxy: None) resp = client.post("/api/server/config", json={"server_ip": "10.0.0.5"}) diff --git a/tests/test_subscribers.py b/tests/test_subscribers.py index 0424e7b..ffe6df5 100644 --- a/tests/test_subscribers.py +++ b/tests/test_subscribers.py @@ -55,6 +55,25 @@ def test_double_registration_does_not_duplicate_callbacks(self): finally: set_bluesky_proxy(None) + def test_explicit_proxy_overrides_global(self): + # Callers that just connected a specific proxy (the /api/server/config + # route, the integrated auto-start) pass it explicitly; subscribers must + # land on that proxy's client even when the global points elsewhere. + explicit = BlueSkyProxy() + explicit.bluesky_client = BlueSkyClient() + other = BlueSkyProxy() + other.bluesky_client = BlueSkyClient() + set_bluesky_proxy(other) + try: + register_subscribers(explicit) + explicit_subs = explicit.bluesky_client.subscriber.subscribers + other_subs = other.bluesky_client.subscriber.subscribers + for topic, _, _ in SUBSCRIPTIONS: + assert topic in explicit_subs + assert topic not in other_subs + finally: + set_bluesky_proxy(None) + def test_no_proxy_is_safe(self): set_bluesky_proxy(None) # Should log an error and return without raising.