Fix race between integrated auto-start and manual connect leaving a stale subscriber-less proxy

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,22 +130,25 @@ 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,
         ready_timeout=10.0,
         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
 # --------------------------------------------------------------------------

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,20 +179,32 @@ 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.
     if is_port_listening is None:
         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(

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",
 ]

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