From 201be3e5462124cd9ecd44388faa7e1f99200535 Mon Sep 17 00:00:00 2001 From: Nick Edson <35981293+s00rz@users.noreply.github.com> Date: Sat, 11 Jul 2026 08:27:57 -0700 Subject: [PATCH] fix(gateway): reserve retrying Photon listener ownership --- gateway/run.py | 39 +++++++++++++- .../test_multiplex_adapter_registry.py | 52 ++++++++++++++++++- 2 files changed, 87 insertions(+), 4 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 40d44e34f99f..13f2743d962f 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -8475,6 +8475,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "config": platform_config, "attempts": 1, "next_retry": time.monotonic() + 30, + "credential_claim": self._adapter_credential_claim( + platform, adapter + ), + "listener_claim": self._adapter_listener_claim( + platform, adapter + ), } else: self._update_platform_runtime_status( @@ -8491,6 +8497,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "config": platform_config, "attempts": 1, "next_retry": time.monotonic() + 30, + "credential_claim": self._adapter_credential_claim( + platform, adapter + ), + "listener_claim": self._adapter_listener_claim( + platform, adapter + ), } except Exception as e: logger.error("✗ %s error: %s", platform.value, e) @@ -8510,6 +8522,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "config": platform_config, "attempts": 1, "next_retry": time.monotonic() + 30, + "credential_claim": self._adapter_credential_claim( + platform, adapter + ), + "listener_claim": self._adapter_listener_claim( + platform, adapter + ), } if await self._abort_startup_if_shutdown_requested(): return True @@ -10231,6 +10249,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew listener_claim = self._adapter_listener_claim(_plat, _ad) if listener_claim is not None: claimed[listener_claim] = active + # A retryable primary still owns its configured credential and listener. + # Reserve both while it is queued so a secondary cannot take the endpoint + # before the reconnect watcher retries the primary adapter. + for retry_info in getattr(self, "_failed_platforms", {}).values(): + for claim_name in ("credential_claim", "listener_claim"): + retry_claim = retry_info.get(claim_name) + if isinstance(retry_claim, tuple): + claimed[retry_claim] = active for profile_name, profile_home in profiles_to_serve(multiplex=True): if profile_name == active: @@ -10339,8 +10365,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew continue # Same-token conflict detection — refuse a duplicate poll. - fp = self._adapter_credential_fingerprint(adapter) - credential_claim = (platform, fp) if fp is not None else None + credential_claim = self._adapter_credential_claim(platform, adapter) if credential_claim is not None: owner = claimed.get(credential_claim) if owner is not None: @@ -10619,6 +10644,16 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew return _handler + @staticmethod + def _adapter_credential_claim( + platform: Platform, adapter: Any + ) -> Optional[tuple]: + """Return the exclusive credential resource claimed by an adapter.""" + fingerprint = GatewayRunner._adapter_credential_fingerprint(adapter) + if fingerprint is None: + return None + return (platform, fingerprint) + @staticmethod def _adapter_listener_claim(platform: Platform, adapter: Any) -> Optional[tuple]: """Return the exclusive listener resource claimed by an adapter. diff --git a/tests/gateway/test_multiplex_adapter_registry.py b/tests/gateway/test_multiplex_adapter_registry.py index 6643e4d73442..ca3b2fbe2895 100644 --- a/tests/gateway/test_multiplex_adapter_registry.py +++ b/tests/gateway/test_multiplex_adapter_registry.py @@ -878,7 +878,9 @@ class TestSecondaryProfileConfigHandling: monkeypatch.setattr("gateway.config.load_gateway_config", lambda: reviewer_cfg) monkeypatch.setattr(runner, "_create_adapter", lambda p, c: secondary) monkeypatch.setattr(runner, "_connect_adapter_with_timeout", _connect) - monkeypatch.setattr(runner, "_make_adapter_auth_check", lambda p: None) + monkeypatch.setattr( + runner, "_make_adapter_auth_check", lambda p, **kwargs: None + ) connected = await runner._start_one_profile_adapters( "reviewer", "/tmp/x", claimed @@ -934,7 +936,9 @@ class TestSecondaryProfileConfigHandling: monkeypatch.setattr("gateway.config.load_gateway_config", lambda: profile_cfg) monkeypatch.setattr(runner, "_create_adapter", lambda p, c: next(adapters)) monkeypatch.setattr(runner, "_connect_adapter_with_timeout", _connect) - monkeypatch.setattr(runner, "_make_adapter_auth_check", lambda p: None) + monkeypatch.setattr( + runner, "_make_adapter_auth_check", lambda p, **kwargs: None + ) first = await runner._start_one_profile_adapters("broken", "/tmp/x", claimed) second = await runner._start_one_profile_adapters("later", "/tmp/y", claimed) @@ -944,6 +948,50 @@ class TestSecondaryProfileConfigHandling: assert second == 1 assert runner._profile_adapters["later"][photon] is later + @pytest.mark.asyncio + async def test_failed_primary_photon_listener_is_reserved_for_retry( + self, monkeypatch + ): + """A retrying primary keeps secondaries off its sidecar endpoint.""" + from gateway.config import GatewayConfig, Platform + + runner = GatewayRunner.__new__(GatewayRunner) + runner.config = GatewayConfig(multiplex_profiles=True) + runner.adapters = {} + runner._profile_adapters = {} + runner.pairing_stores = {} + + photon = Platform("photon") + listener_claim = ("listener", "photon", "127.0.0.1", 8789) + runner._failed_platforms = { + photon: { + "config": object(), + "attempts": 1, + "next_retry": 0, + "listener_claim": listener_claim, + } + } + seen = {} + + async def _start(profile_name, profile_home, claimed): + seen.update(claimed) + return 0 + + monkeypatch.setattr( + "hermes_cli.profiles.profiles_to_serve", + lambda multiplex=True: (("default", "/tmp/default"), ("reviewer", "/tmp/reviewer")), + ) + monkeypatch.setattr( + "hermes_cli.profiles.get_active_profile_name", lambda: "default" + ) + monkeypatch.setattr("gateway.status.write_runtime_status", lambda **kwargs: None) + monkeypatch.setattr(runner, "_start_one_profile_adapters", _start) + + connected = await runner._start_secondary_profile_adapters() + + assert connected == 0 + assert seen[listener_claim] == "default" + def test_port_binding_set_covers_known_listeners(self): from gateway.run import _PORT_BINDING_PLATFORM_VALUES # Every adapter that binds a TCP port must be in the guard set.