From 5a3d6306f54d8715aa1986469a1d97a80e2b351b Mon Sep 17 00:00:00 2001 From: pedramhatef Date: Tue, 21 Jul 2026 05:40:06 +0800 Subject: [PATCH] Fix orchestrator registration: call missing REST endpoint, make NATS startup resilient MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BeamCore's NATS `register` binds to an existing orchestrator row, but that row is only created by REST POST /orchestrators/register — which the orchestrator neuron never calls (only the worker self-registers over REST). A fresh hotkey therefore gets orchestrator_not_routable forever, with no path to recover. See #2 for the full writeup and reproduction. - Add _ensure_rest_registration(), called at the top of each start_polling() attempt, mirroring the worker's existing self-registration. - Move the NATS control handshake into a background task with unbounded retry/backoff. Previously it was awaited inline during FastAPI startup, which blocked uvicorn from binding (so BeamCore's own reachability check against our port would fail) and permanently discarded the client after 12 attempts, with no self-healing path. - Log the outgoing registration payload and the full NATS error response, since the prior truncated error hid the actual rejection reason. Verified on mainnet SN105: an orchestrator stuck at orchestrator_not_routable for 5 days registered successfully on the first attempt after this change, with no other configuration changes. --- .../clients/subnet_core_client.py | 44 ++++++++++++++++++- neurons/orchestrator/core/orchestrator.py | 24 ++++++++-- 2 files changed, 63 insertions(+), 5 deletions(-) diff --git a/neurons/orchestrator/clients/subnet_core_client.py b/neurons/orchestrator/clients/subnet_core_client.py index 6ec52eb..e53f555 100644 --- a/neurons/orchestrator/clients/subnet_core_client.py +++ b/neurons/orchestrator/clients/subnet_core_client.py @@ -284,6 +284,42 @@ async def _ensure_api_key(self) -> Optional[str]: logger.error("Failed to get API key: %s", e) return None + async def _ensure_rest_registration(self) -> bool: + """Create/refresh our orchestrator row in BeamCore via REST. + + BeamCore's NATS `register` binds to an existing orchestrator row; if no + row exists it answers `orchestrator_not_routable` (nothing to do with + network reachability — BeamCore never probes the advertised URL, and + plenty of live orchestrators advertise RFC1918/localhost addresses). + Upstream never calls this endpoint, so a fresh hotkey can never register. + """ + api_key = await self._ensure_api_key() + if not api_key: + logger.warning("Cannot REST-register orchestrator: no API key") + return False + cfg = dict(self._registration_config or {}) + payload = {"hotkey": self.orchestrator_hotkey} + # /orchestrators/register has additionalProperties:false — send only + # fields it declares (no gateway_url, no ready). + for key in ("url", "region", "max_workers", "uid", "fee_percentage"): + if cfg.get(key) is not None: + payload[key] = cfg[key] + try: + client = await self._get_client() + resp = await client.post( + f"{self.base_url}/orchestrators/register", + json=payload, + headers={"Authorization": f"Bearer {api_key}"}, + ) + if resp.status_code != 200: + logger.error("Orchestrator REST registration failed: %s - %s", resp.status_code, resp.text[:300]) + return False + logger.info("Orchestrator REST registration OK: %s", resp.json().get("message")) + return True + except Exception as e: + logger.error("Orchestrator REST registration error: %s", e) + return False + async def start_polling(self): if self._running: logger.warning("Already running") @@ -292,6 +328,7 @@ async def start_polling(self): last_error: Optional[Exception] = None for attempt in range(1, STARTUP_CONNECT_ATTEMPTS + 1): try: + await self._ensure_rest_registration() await self._connect_nats_session() registered = await self._register_via_nats() if not registered: @@ -560,6 +597,11 @@ async def _register_via_nats(self) -> bool: self._desired_ready = self._registration_ready() cfg["ready"] = self._desired_ready cfg["signature"] = signature + logger.info( + "Registering via NATS: url=%s gateway_url=%s region=%s uid=%s ready=%s max_workers=%s", + cfg.get("url"), cfg.get("gateway_url"), cfg.get("region"), + cfg.get("uid"), cfg.get("ready"), cfg.get("max_workers"), + ) response = await self._send_nats_request("register", cfg, timeout=max(REQUEST_TIMEOUT, 3.0)) if response.get("type") == "register_ack": self._registered = True @@ -567,7 +609,7 @@ async def _register_via_nats(self) -> bool: self._schedule_ready_sync_if_needed() return True self._registered = False - logger.error("Registration failed: %s", response.get("message") or response.get("reason") or response) + logger.error("Registration failed, full response: %s", response) return False async def _heartbeat_loop(self) -> None: diff --git a/neurons/orchestrator/core/orchestrator.py b/neurons/orchestrator/core/orchestrator.py index ae35bac..0d76e65 100644 --- a/neurons/orchestrator/core/orchestrator.py +++ b/neurons/orchestrator/core/orchestrator.py @@ -1023,10 +1023,26 @@ async def _init_subnet_core_client(self) -> None: gateway_url, ) - # Start NATS control connection for real-time notifications and - # orchestrator control-plane requests. - await self.subnet_core_client.start_polling() - logger.info("SubnetCore NATS control connection started") + # Start NATS control connection in the background with infinite + # retry. Awaiting the handshake here blocks uvicorn from binding, + # but BeamCore's registration probe needs our HTTP/gateway port + # answering — so a first-time registration deadlocks and the + # client is torn down permanently after 12 attempts. + async def _start_nats_control_forever() -> None: + delay = 5.0 + while True: + try: + await self.subnet_core_client.start_polling() + logger.info("SubnetCore NATS control connection started") + return + except Exception as e: + logger.warning( + f"NATS control startup failed ({e}); retrying in {delay:.0f}s" + ) + await asyncio.sleep(delay) + delay = min(delay * 2, 120.0) + + self._nats_control_task = asyncio.create_task(_start_nats_control_forever()) except Exception as e: logger.warning(f"Failed to initialize SubnetCoreClient: {e}")