From bb6fd304658a2a04f288c6b01ba2fe9886f853ca Mon Sep 17 00:00:00 2001 From: Andres Morfin Veytia <78442543+amorfinv@users.noreply.github.com> Date: Wed, 19 Aug 2026 20:55:28 +0200 Subject: [PATCH 1/2] fix(proxy): serialize proxy swap-and-connect and register subscribers on an explicit proxy Add a shared connect_lock to WebATM.proxy and hold it in /api/server/config around the tear-down/replace/connect sequence, so a concurrent connector can never revive the proxy this request is replacing. register_subscribers now takes the proxy to attach to (falling back to the global), so subscribers always land on the client that was just started even if the global is swapped concurrently. Co-Authored-By: Claude Fable 5 --- WebATM/proxy/__init__.py | 9 +++++++ WebATM/proxy/subscribers.py | 27 +++++++++++++------- WebATM/server/routes.py | 49 +++++++++++++++++++++++-------------- tests/test_app.py | 9 +++---- tests/test_subscribers.py | 19 ++++++++++++++ 5 files changed, 80 insertions(+), 33 deletions(-) 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. From 31fd2351abd52293cd15baad3f8722f63fa5c8bc Mon Sep 17 00:00:00 2001 From: Andres Morfin Veytia <78442543+amorfinv@users.noreply.github.com> Date: Wed, 19 Aug 2026 20:55:38 +0200 Subject: [PATCH 2/2] fix(integrated): stand down auto-start when a manual connect replaces the proxy The port wait can span many seconds on a cold start; if the user connects manually meanwhile, /api/server/config replaces the global proxy and closes the boot-time one auto-start captured. Connecting it anyway left a second, subscriber-less ZMQ client alive whose data-flow timeout later broadcast a bogus disconnect and blanked the map. Auto-start now runs its connect step under the shared connect_lock, re-checks the global proxy is still its own, and stands down if a manual connect took over. Subscribers are registered on the connected proxy explicitly. Co-Authored-By: Claude Fable 5 --- WebATM-integrated/tests/test_auto_start.py | 92 +++++++++++++++++-- .../webatm_integrated/auto_start.py | 76 +++++++++++---- 2 files changed, 140 insertions(+), 28 deletions(-) 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(