diff --git a/custom_components/meshcore/config_flow.py b/custom_components/meshcore/config_flow.py index 6f77580..659a0d4 100644 --- a/custom_components/meshcore/config_flow.py +++ b/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") diff --git a/custom_components/meshcore/const.py b/custom_components/meshcore/const.py index 433ad7e..afeadc3 100644 --- a/custom_components/meshcore/const.py +++ b/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 diff --git a/custom_components/meshcore/coordinator.py b/custom_components/meshcore/coordinator.py index 5107096..f5af0e2 100644 --- a/custom_components/meshcore/coordinator.py +++ b/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 diff --git a/custom_components/meshcore/translations/en.json b/custom_components/meshcore/translations/en.json index 06ad7d3..75464b3 100644 --- a/custom_components/meshcore/translations/en.json +++ b/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.", diff --git a/tests/test_path_recovery.py b/tests/test_path_recovery.py new file mode 100644 index 0000000..c19273d --- /dev/null +++ b/tests/test_path_recovery.py @@ -0,0 +1,179 @@ +"""Tests for the stale-path recovery used by repeater/telemetry polling. + +Mirrors MeshCoreDataUpdateCoordinator._call_with_path_recovery: on a timed-out +mesh request with a stored *direct* path, reset the path to flood and retry once +within the same poll. This is what lets a freshly-added (or moved) repeater go +green on its first poll instead of waiting MAX_FAILURES_BEFORE_PATH_RESET cycles. +""" +import asyncio +from unittest.mock import AsyncMock, MagicMock + +import pytest + + +async def _call_with_path_recovery( + coord, command_factory, contact, node_config, pubkey_prefix, label, timeout=0.05 +): + """Standalone copy of the coordinator method for testability.""" + node_name = node_config.get("name", "unknown") + + async def _bounded(c): + try: + return await asyncio.wait_for(command_factory(c), timeout=timeout) + except TimeoutError: + coord.logger.warning( + f"{label} for {node_name} exceeded {timeout}s; treating as no response" + ) + return None + + result = await _bounded(contact) + if result or not contact or contact.get("out_path_len", -1) <= -1: + return result, contact + + coord.logger.info( + f"No response for {label} from {node_name}; resetting path to flood and retrying" + ) + if not await coord._reset_node_path(contact, node_config): + return result, contact + + await asyncio.sleep(0) + contact = coord.api.mesh_core.get_contact_by_key_prefix(pubkey_prefix) or contact + result = await _bounded(contact) + return result, contact + + +def _make_coordinator(reset_ok=True, refreshed_contact=None): + coord = MagicMock() + coord.logger = MagicMock() + coord._reset_node_path = AsyncMock(return_value=reset_ok) + coord.api = MagicMock() + coord.api.mesh_core.get_contact_by_key_prefix = MagicMock(return_value=refreshed_contact) + return coord + + +@pytest.mark.asyncio +async def test_success_first_try_no_reset(): + """A successful request never touches the path.""" + coord = _make_coordinator() + contact = {"out_path_len": 3} + factory = AsyncMock(return_value={"uptime": 100}) + + result, out_contact = await _call_with_path_recovery( + coord, factory, contact, {"name": "rptr"}, "ab12", "status request" + ) + + assert result == {"uptime": 100} + assert out_contact is contact + coord._reset_node_path.assert_not_awaited() + assert factory.await_count == 1 + + +@pytest.mark.asyncio +async def test_timeout_with_direct_path_resets_and_retries(): + """A timeout with a stored direct path resets to flood and retries once.""" + refreshed = {"out_path_len": -1} + coord = _make_coordinator(reset_ok=True, refreshed_contact=refreshed) + contact = {"out_path_len": 5} + factory = AsyncMock(side_effect=[None, {"uptime": 42}]) + + result, out_contact = await _call_with_path_recovery( + coord, factory, contact, {"name": "rptr"}, "ab12", "status request" + ) + + assert result == {"uptime": 42} + assert out_contact is refreshed # contact refreshed after reset + coord._reset_node_path.assert_awaited_once() + assert factory.await_count == 2 + + +@pytest.mark.asyncio +async def test_timeout_while_flooding_does_not_reset(): + """No stored direct path (already flooding) → nothing to reset, no retry.""" + coord = _make_coordinator() + contact = {"out_path_len": -1} + factory = AsyncMock(return_value=None) + + result, out_contact = await _call_with_path_recovery( + coord, factory, contact, {"name": "rptr"}, "ab12", "status request" + ) + + assert result is None + assert out_contact is contact + coord._reset_node_path.assert_not_awaited() + assert factory.await_count == 1 + + +@pytest.mark.asyncio +async def test_reset_disabled_skips_retry(): + """If the path reset is refused (disabled), don't retry the request.""" + coord = _make_coordinator(reset_ok=False) + contact = {"out_path_len": 5} + factory = AsyncMock(return_value=None) + + result, out_contact = await _call_with_path_recovery( + coord, factory, contact, {"name": "rptr"}, "ab12", "status request" + ) + + assert result is None + assert out_contact is contact + coord._reset_node_path.assert_awaited_once() + assert factory.await_count == 1 # no retry + + +@pytest.mark.asyncio +async def test_missing_contact_returns_without_reset(): + """A None contact short-circuits without attempting a reset.""" + coord = _make_coordinator() + factory = AsyncMock(return_value=None) + + result, out_contact = await _call_with_path_recovery( + coord, factory, None, {"name": "rptr"}, "ab12", "status request" + ) + + assert result is None + assert out_contact is None + coord._reset_node_path.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_hanging_command_is_treated_as_no_response_and_recovers(): + """A command that hangs past the timeout must be treated as no-response so a + wedged link can't freeze the poll task — it then triggers path recovery.""" + refreshed = {"out_path_len": -1} + coord = _make_coordinator(reset_ok=True, refreshed_contact=refreshed) + contact = {"out_path_len": 5} + + calls = {"n": 0} + + async def factory(c): + calls["n"] += 1 + if calls["n"] == 1: + await asyncio.sleep(10) # hang well past the timeout + return {"uptime": 1} + return {"uptime": 99} # retry after path reset succeeds fast + + result, out_contact = await _call_with_path_recovery( + coord, factory, contact, {"name": "rptr"}, "ab12", "status request", timeout=0.02 + ) + + assert result == {"uptime": 99} + assert out_contact is refreshed + coord._reset_node_path.assert_awaited_once() + coord.logger.warning.assert_called() # logged the timeout + + +@pytest.mark.asyncio +async def test_retry_still_times_out_returns_none(): + """If the retry after reset also times out, the falsy result is returned.""" + refreshed = {"out_path_len": -1} + coord = _make_coordinator(reset_ok=True, refreshed_contact=refreshed) + contact = {"out_path_len": 5} + factory = AsyncMock(side_effect=[None, None]) + + result, out_contact = await _call_with_path_recovery( + coord, factory, contact, {"name": "rptr"}, "ab12", "status request" + ) + + assert result is None + assert out_contact is refreshed + assert factory.await_count == 2