Repeater login & polling resilience: rejection vs timeout, auto path-reset, bounded mesh calls

custom_components/meshcore/config_flow.py

@@ -208,6 +208,77 @@ async def validate_tcp_input(hass: HomeAssistant, data: Dict[str, Any]) -> Dict[
     return await validate_common(api)
 
 
+async def _attempt_login(meshcore, contact, password, timeout: float) -> str:
+    """Single login attempt. Returns "success", "rejected", or "timeout".
+
+    The SDK's send_login_sync only waits for LOGIN_SUCCESS within a tight
+    window (suggested_timeout / 800 — a few seconds), so a multi-hop or slow
+    repeater reply times out, and a rejected password (which the firmware
+    reports as a distinct LOGIN_FAILED frame) is indistinguishable from
+    no-response. This waits for BOTH LOGIN_SUCCESS and LOGIN_FAILED with more
+    headroom, so the caller can tell "password rejected" from "no response".
+    """
+    # Register the waiters before sending so a fast reply can't be missed.
+    success = asyncio.ensure_future(
+        meshcore.wait_for_event(EventType.LOGIN_SUCCESS, timeout=timeout)
+    )
+    failed = asyncio.ensure_future(
+        meshcore.wait_for_event(EventType.LOGIN_FAILED, timeout=timeout)
+    )
+    await asyncio.sleep(0)  # let the waiters subscribe before we transmit
+
+    try:
+        send = await meshcore.commands.send_login(contact, password)
+    except Exception as ex:
+        _LOGGER.error("Error sending login to repeater: %s", ex)
+        send = None
+
+    if send is None or getattr(send, "type", None) == EventType.ERROR:
+        for task in (success, failed):
+            task.cancel()
+        return "timeout"
+
+    try:
+        done, _pending = await asyncio.wait(
+            {success, failed}, return_when=asyncio.FIRST_COMPLETED
+        )
+    finally:
+        for task in (success, failed):
+            if not task.done():
+                task.cancel()
+
+    if success in done and not success.cancelled() and success.result():
+        return "success"
+    if failed in done and not failed.cancelled() and failed.result():
+        return "rejected"
+    return "timeout"
+
+
+async def _login_to_repeater(meshcore, contact, password) -> str:
+    """Log in to a repeater, recovering from a stale path on timeout.
+
+    A direct login over a stale stored path gets no reply and times out (the
+    repeater never receives it). Mirror the manual fix — reset the path so the
+    next attempt floods — and retry once. A password rejection (LOGIN_FAILED)
+    is returned immediately without resetting the path. Returns "success",
+    "rejected", or "timeout".
+    """
+    outcome = await _attempt_login(meshcore, contact, password, timeout=12.0)
+    if outcome != "timeout":
+        return outcome
+
+    # No reply — the stored path is likely stale. Reset it (forces flood) and
+    # retry once with more headroom for the multi-hop flood round trip.
+    _LOGGER.info("Login timed out; resetting repeater path to flood and retrying")
+    try:
+        await meshcore.commands.reset_path(contact)
+        await asyncio.sleep(0.5)
+    except Exception as ex:
+        _LOGGER.debug("Path reset before login retry failed: %s", ex)
+
+    return await _attempt_login(meshcore, contact, password, timeout=20.0)
+
+
 class MeshCoreConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): # type: ignore
     """Handle a config flow for MeshCore."""
 
@@ -715,14 +786,19 @@ async def async_step_add_repeater(self, user_input=None):
             errors["base"] = "Contact not found"
             return self._show_add_repeater_form(repeater_dict, errors, user_input)
 
-        # Try to login
-        result = await meshcore.commands.send_login_sync(contact, password)
-        if not result:
-            _LOGGER.error("Login to repeater failed or timed out")
-            errors["base"] = "Failed to log in to repeater. Check password and try again."
+        # Try to login. Distinguish a rejected password from no-response so the
+        # user gets an actionable message (and give a slow repeater reply room).
+        outcome = await _login_to_repeater(meshcore, contact, password)
+        if outcome == "rejected":
+            _LOGGER.error("Repeater rejected login (incorrect password)")
+            errors["base"] = "login_failed"
             return self._show_add_repeater_form(repeater_dict, errors, user_input)
-
-
+        if outcome != "success":
+            _LOGGER.error("No login response from repeater (timed out)")
+            errors["base"] = "login_timeout"
+            return self._show_add_repeater_form(repeater_dict, errors, user_input)
+
+
         # Login successful, now optionally check for version
         send_result = await meshcore.commands.send_cmd(contact, "ver")
 

custom_components/meshcore/const.py

@@ -183,6 +183,14 @@ def get_contact_discovery_mode(config_entry) -> str:
 # Other constants
 CONNECTION_TIMEOUT: Final = 10  # seconds
 
+# Hard upper bound (seconds) on a single mesh SDK command. The "_sync" SDK
+# commands have their own short internal timeouts, but a wedged BLE/serial link
+# can leave an await hanging indefinitely; without a cap that hangs the whole
+# repeater poll task, which the "still running, skipping" guard then skips every
+# tick forever. These caps turn a hang into an ordinary (recoverable) failure.
+MESH_COMMAND_TIMEOUT: Final = 60
+NEIGHBOR_FETCH_TIMEOUT: Final = 60  # fetch_all_neighbours can be slow with many neighbors
+
 # Rate limiter settings
 RATE_LIMITER_CAPACITY: Final = 20
 RATE_LIMITER_REFILL_RATE_SECONDS: Final = 120

custom_components/meshcore/coordinator.py

@@ -36,6 +36,7 @@
     REPEATER_BACKOFF_BASE,
     MAX_FAILURES_BEFORE_PATH_RESET,
     MAX_RANDOM_DELAY,
+    MESH_COMMAND_TIMEOUT,
     CONF_REPEATER_TELEMETRY_ENABLED,
     CONF_SELF_TELEMETRY_ENABLED,
     CONF_SELF_TELEMETRY_INTERVAL,
@@ -52,6 +53,7 @@
     RATE_LIMITER_REFILL_RATE_SECONDS,
     RX_LOG_CACHE_MAX_SIZE,
     RX_LOG_CACHE_TTL_SECONDS,
+    NEIGHBOR_FETCH_TIMEOUT,
     NEIGHBOR_PUBKEY_PREFIX_LENGTH,
     NEIGHBOR_STALE_THRESHOLD,
     SEEN_WINDOW_SECS,
@@ -803,7 +805,10 @@ async def _reset_node_path(self, contact, node_config: dict) -> bool:
             return False
 
         try:
-            result = await self.api.mesh_core.commands.reset_path(contact)
+            result = await asyncio.wait_for(
+                self.api.mesh_core.commands.reset_path(contact),
+                timeout=MESH_COMMAND_TIMEOUT,
+            )
             if result and result.type != EventType.ERROR:
                 self.logger.info(f"Successfully reset path for {node_name}")
                 return True
@@ -814,7 +819,50 @@ async def _reset_node_path(self, contact, node_config: dict) -> bool:
         except Exception as ex:
             self.logger.warning(f"Exception resetting path for {node_name}: {ex}")
             return False
-
+
+    async def _call_with_path_recovery(self, command_factory, contact, node_config, pubkey_prefix, label):
+        """Run a mesh command; on timeout reset a stale path to flood and retry once.
+
+        A stored direct path that no longer reaches the node (e.g. right after
+        adding the node, or after it moves) makes the request silently time out.
+        Rather than waiting MAX_FAILURES_BEFORE_PATH_RESET cycles, reset the path
+        to flood and retry immediately so the node recovers on this same poll.
+
+        command_factory is called with the (possibly refreshed) contact and must
+        return an awaitable. Returns ``(result, contact)`` — contact may be
+        refreshed after a path reset.
+        """
+        node_name = node_config.get("name", "unknown")
+
+        async def _bounded(c):
+            # Cap the call so a wedged link can't hang the whole poll task.
+            try:
+                return await asyncio.wait_for(command_factory(c), timeout=MESH_COMMAND_TIMEOUT)
+            except TimeoutError:
+                self.logger.warning(
+                    f"{label} for {node_name} exceeded {MESH_COMMAND_TIMEOUT}s; treating as no response"
+                )
+                return None
+
+        result = await _bounded(contact)
+        # Only intervene when the request timed out AND a direct path exists to
+        # clear. Once flooding (out_path_len == -1) there is nothing to reset, so
+        # this never spins into a flood-spam loop.
+        if result or not contact or contact.get("out_path_len", -1) <= -1:
+            return result, contact
+
+        self.logger.info(
+            f"No response for {label} from {node_name}; resetting path to flood and retrying"
+        )
+        if not await self._reset_node_path(contact, node_config):
+            return result, contact
+
+        await asyncio.sleep(0.5)
+        # Re-fetch the contact in case the path reset mutated the stored entry.
+        contact = self.api.mesh_core.get_contact_by_key_prefix(pubkey_prefix) or contact
+        result = await _bounded(contact)
+        return result, contact
+
     def update_telemetry_settings(self, config_entry: ConfigEntry) -> None:
         """Update telemetry settings from config entry."""
         self._self_telemetry_enabled = config_entry.data.get(CONF_SELF_TELEMETRY_ENABLED, False)
@@ -870,9 +918,19 @@ async def _fetch_repeater_neighbors(self, contact, repeater_name: str, pubkey_pr
                 return
 
             self.logger.debug(f"Fetching neighbors for repeater {repeater_name} ({pubkey_prefix})")
-            result = await self.api.mesh_core.commands.fetch_all_neighbours(
-                contact, pubkey_prefix_length=NEIGHBOR_PUBKEY_PREFIX_LENGTH
-            )
+            try:
+                result = await asyncio.wait_for(
+                    self.api.mesh_core.commands.fetch_all_neighbours(
+                        contact, pubkey_prefix_length=NEIGHBOR_PUBKEY_PREFIX_LENGTH
+                    ),
+                    timeout=NEIGHBOR_FETCH_TIMEOUT,
+                )
+            except TimeoutError:
+                self.logger.warning(
+                    f"Neighbor fetch for {repeater_name} exceeded {NEIGHBOR_FETCH_TIMEOUT}s; "
+                    f"skipping this cycle"
+                )
+                return
 
             if not result or "neighbours" not in result:
                 self.logger.debug(f"No neighbor data returned for {repeater_name}")
@@ -1219,9 +1277,11 @@ async def _update_repeater(self, repeater_config):
                     return
 
                 try:
-                    login_result = await self.api.mesh_core.commands.send_login_sync(
-                        contact,
-                        repeater_config.get(CONF_REPEATER_PASSWORD, "")
+                    password = repeater_config.get(CONF_REPEATER_PASSWORD, "")
+                    # Reset the stale path to flood and retry if the login times out.
+                    login_result, contact = await self._call_with_path_recovery(
+                        lambda c: self.api.mesh_core.commands.send_login_sync(c, password),
+                        contact, repeater_config, pubkey_prefix, "login",
                     )
 
                     if login_result:
@@ -1254,7 +1314,11 @@ async def _update_repeater(self, repeater_config):
                 self._apply_repeater_backoff(pubkey_prefix, new_failure_count, update_interval)
                 return
 
-            result = await self.api.mesh_core.commands.req_status_sync(contact)
+            # Reset the stale path to flood and retry if the request times out.
+            result, contact = await self._call_with_path_recovery(
+                lambda c: self.api.mesh_core.commands.req_status_sync(c),
+                contact, repeater_config, pubkey_prefix, "status request",
+            )
             _LOGGER.debug(f"Status response received: {result}")
 
 
@@ -1385,8 +1449,12 @@ async def _update_node_telemetry(self, contact, node_config: dict):
                 self._apply_backoff(pubkey_prefix, new_failure_count, update_interval, "telemetry")
                 return
 
-            telemetry_result = await self.api.mesh_core.commands.req_telemetry_sync(contact)
-
+            # Reset the stale path to flood and retry if the request times out.
+            telemetry_result, contact = await self._call_with_path_recovery(
+                lambda c: self.api.mesh_core.commands.req_telemetry_sync(c),
+                contact, node_config, pubkey_prefix, "telemetry request",
+            )
+
             if telemetry_result:
                 self.logger.debug(f"Telemetry response received from {node_name}: {telemetry_result}")
                 # Reset failure count on success

custom_components/meshcore/translations/en.json

@@ -8,8 +8,8 @@
       "unknown": "Unexpected error",
       "already_configured": "Repeater is already configured",
       "client_already_tracked": "Client is already being tracked",
-      "login_failed": "Failed to log in to repeater. Check password and try again.",
-      "login_timeout": "Timed out waiting for login response",
+      "login_failed": "The repeater rejected the password. Check it and try again.",
+      "login_timeout": "No response from the repeater, even after resetting the path to flood. The password was not rejected — the repeater may be offline or out of range. Try again.",
       "contact_not_found": "Contact not found",
       "no_repeaters_found": "No repeaters found. Refresh contacts first.",
       "no_clients_found": "No clients found. Refresh contacts first.",