fix(coordinator): re-arm the self-telemetry interval on failure

custom_components/meshcore/coordinator.py

@@ -92,6 +92,44 @@ def _log_get_msg_error(action: str, payload: Any) -> None:
         _LOGGER.error("Error %s messages: %s", action, payload)
 
 
+# Companion-protocol error codes that mean the node can never satisfy the
+# request, however many times it is repeated -- as opposed to a transient
+# fault that a later attempt may clear.
+_TELEMETRY_UNSUPPORTED_CODES = frozenset(
+    {"ERR_CODE_ILLEGAL_ARG", "ERR_CODE_UNSUPPORTED_CMD"}
+)
+
+
+def _log_self_telemetry_error(payload: Any, already_reported: bool) -> None:
+    """Log a failed self-telemetry result at the appropriate level.
+
+    ``get_self_telemetry()`` sends the four-byte "self" form of
+    ``CMD_SEND_TELEMETRY_REQ`` (opcode plus three reserved bytes, no pub key).
+    Firmware answers it, but a non-firmware companion need not: an openHop
+    virtual companion up to and including 1.1.1 requires the 36-byte contact
+    form and rejects the short frame with ``ERR_CODE_ILLEGAL_ARG``.
+
+    That is a fixed property of the peer, not a fault, so it is reported once
+    at WARNING naming the cause and the remedy, then demoted to DEBUG. Every
+    other failure keeps ERROR on every occurrence. The caller clears its flag
+    on success, so a genuine later regression is reported afresh.
+    """
+    code = payload.get("code_string") if isinstance(payload, dict) else None
+    if code in _TELEMETRY_UNSUPPORTED_CODES:
+        if already_reported:
+            _LOGGER.debug("Self telemetry still unsupported by this node: %s", payload)
+        else:
+            _LOGGER.warning(
+                "This node rejected the self-telemetry request (%s): it does not "
+                "implement the 'self' form of CMD_SEND_TELEMETRY_REQ. Turn off "
+                "self telemetry for this entry, or update the node's firmware or "
+                "companion software. Further occurrences are logged at DEBUG.",
+                code,
+            )
+    else:
+        _LOGGER.error("Failed to get self telemetry: %s", payload)
+
+
 class MeshCoreDataUpdateCoordinator(DataUpdateCoordinator):
     """Class to manage fetching data from the MeshCore node and trigger event-generating commands."""
 
@@ -190,6 +228,10 @@ def __init__(
 
         # Self telemetry tracking
         self._last_self_telemetry_update = 0
+        # Set once a self-telemetry failure has been reported, so a node that
+        # cannot answer the request is named once rather than on every cycle.
+        # Cleared on the next success.
+        self._self_telemetry_error_reported = False
         self._self_telemetry_enabled = config_entry.data.get(CONF_SELF_TELEMETRY_ENABLED, False)
         self._self_telemetry_interval = config_entry.data.get(CONF_SELF_TELEMETRY_INTERVAL, DEFAULT_SELF_TELEMETRY_INTERVAL)
 
@@ -1619,13 +1661,24 @@ async def _async_update_data(self) -> Dict[str, Any]:
         if self._self_telemetry_enabled:
             if current_time - self._last_self_telemetry_update >= self._self_telemetry_interval:
                 self.logger.debug(f"Getting self telemetry (interval: {self._self_telemetry_interval}s)")
+                # The interval gates *attempts*, not successes. Recording the
+                # attempt before it runs keeps a failing node on the configured
+                # cadence; advancing this only in the success branch left the
+                # gate permanently open against a node that cannot answer, so a
+                # 300 s setting collapsed to the coordinator tick and re-sent a
+                # request that could never succeed (~17k ERROR lines a day at a
+                # 5 s tick).
+                self._last_self_telemetry_update = current_time
                 try:
                     telemetry_result = await self.api.mesh_core.commands.get_self_telemetry()
                     if telemetry_result.type == EventType.TELEMETRY_RESPONSE:
                         self.logger.debug(f"Self telemetry received: {telemetry_result.payload}")
-                        self._last_self_telemetry_update = current_time
+                        self._self_telemetry_error_reported = False
                     else:
-                        self.logger.error(f"Failed to get self telemetry: {telemetry_result.payload}")
+                        _log_self_telemetry_error(
+                            telemetry_result.payload, self._self_telemetry_error_reported
+                        )
+                        self._self_telemetry_error_reported = True
                 except Exception as ex:
                     self.logger.error(f"Exception getting self telemetry: {ex}")
             else:

tests/test_self_telemetry_retry.py

@@ -0,0 +1,166 @@
+"""Regression tests for the self-telemetry retry storm and its log level.
+
+``get_self_telemetry()`` sends the four-byte "self" form of
+``CMD_SEND_TELEMETRY_REQ``. Firmware answers it; a non-firmware companion need
+not. An openHop virtual companion up to and including 1.1.1 requires the
+36-byte contact form and rejects the short frame with ``ERR_CODE_ILLEGAL_ARG``
+(``openhop_core/companion/frame_server.py``: ``if len(data) < 35``).
+
+Two bugs compounded that on a live instance:
+
+1. ``_last_self_telemetry_update`` advanced only in the success branch, so a
+   node that always fails never re-armed the interval gate. The configured
+   300 s interval collapsed to the coordinator tick -- a measured ~357 requests
+   per 30 minutes (~5 s apart), half of every companion frame the integration
+   sent, none of which could succeed.
+2. Every failure logged at ERROR, so that storm surfaced as roughly 17k
+   ERROR-tier lines a day.
+
+``coordinator.py`` cannot be imported whole under the conftest stubs, so these
+tests AST-extract the real module-level helper and run it with its free names
+bound, exercising production source. A second test guards the ordering fix
+structurally: the attempt must be recorded before the request is issued, not
+inside the ``try`` that only records it on success.
+"""
+import ast
+import logging
+import os
+
+from typing import Any
+
+_BASE = os.path.join(
+    os.path.dirname(os.path.dirname(__file__)), "custom_components", "meshcore"
+)
+_COORDINATOR_PY = os.path.join(_BASE, "coordinator.py")
+
+_LOGGER_NAME = "custom_components.meshcore.coordinator"
+_LOGGER = logging.getLogger(_LOGGER_NAME)
+
+
+def _read_source():
+    with open(_COORDINATOR_PY, encoding="utf-8") as fh:
+        return fh.read()
+
+
+def _extract_log_self_telemetry_error():
+    """Compile the production helper plus the constant it closes over."""
+    tree = ast.parse(_read_source())
+    wanted = ("_TELEMETRY_UNSUPPORTED_CODES", "_log_self_telemetry_error")
+    body = [
+        node
+        for node in tree.body
+        if (isinstance(node, ast.FunctionDef) and node.name in wanted)
+        or (
+            isinstance(node, ast.Assign)
+            and any(
+                isinstance(t, ast.Name) and t.id in wanted for t in node.targets
+            )
+        )
+    ]
+    assert len(body) == 2, f"expected both {wanted} at module level, got {len(body)}"
+    module = ast.Module(body=body, type_ignores=[])
+    ast.fix_missing_locations(module)
+    code = compile(module, _COORDINATOR_PY, "exec")
+    namespace = {"_LOGGER": _LOGGER, "Any": Any, "frozenset": frozenset}
+    exec(code, namespace)  # noqa: S102 -- executing our own production source
+    return namespace["_log_self_telemetry_error"]
+
+
+_log_self_telemetry_error = _extract_log_self_telemetry_error()
+
+
+def _records(caplog):
+    return [(r.levelno, r.getMessage()) for r in caplog.records]
+
+
+def test_unsupported_code_warns_once_then_debugs(caplog):
+    """A node that cannot answer is named once, then demoted."""
+    payload = {"error_code": 6, "code_string": "ERR_CODE_ILLEGAL_ARG"}
+
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        _log_self_telemetry_error(payload, already_reported=False)
+    levels = [level for level, _ in _records(caplog)]
+    assert levels == [logging.WARNING]
+    assert "ERR_CODE_ILLEGAL_ARG" in caplog.records[0].getMessage()
+
+    caplog.clear()
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        _log_self_telemetry_error(payload, already_reported=True)
+    assert [level for level, _ in _records(caplog)] == [logging.DEBUG]
+
+
+def test_unsupported_cmd_code_is_also_demoted(caplog):
+    """openHop >= 1.1.2 answers the self form and returns UNSUPPORTED_CMD for
+    other malformed lengths; treat it the same way."""
+    payload = {"error_code": 2, "code_string": "ERR_CODE_UNSUPPORTED_CMD"}
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        _log_self_telemetry_error(payload, already_reported=True)
+    assert [level for level, _ in _records(caplog)] == [logging.DEBUG]
+
+
+def test_other_failures_keep_error(caplog):
+    """Transient or unknown failures stay at ERROR on every occurrence."""
+    for payload in (
+        {"error_code": 4, "code_string": "ERR_CODE_TABLE_FULL"},
+        {"reason": "no_event_received"},
+        None,
+    ):
+        caplog.clear()
+        with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+            _log_self_telemetry_error(payload, already_reported=True)
+        assert [level for level, _ in _records(caplog)] == [logging.ERROR], payload
+
+
+def _self_telemetry_gate():
+    """Return the ``if`` node holding the self-telemetry attempt."""
+    for node in ast.walk(ast.parse(_read_source())):
+        if not isinstance(node, ast.Try):
+            continue
+        calls = [
+            n
+            for n in ast.walk(node)
+            if isinstance(n, ast.Attribute) and n.attr == "get_self_telemetry"
+        ]
+        if calls:
+            return node
+    raise AssertionError("no try block issuing get_self_telemetry() found")
+
+
+def _assignments_to(node, name):
+    return [
+        n
+        for n in ast.walk(node)
+        if isinstance(n, ast.Assign)
+        and any(
+            isinstance(t, ast.Attribute) and t.attr == name for t in n.targets
+        )
+    ]
+
+
+def test_attempt_recorded_before_request_not_only_on_success():
+    """The interval gate must re-arm on failure, not just on success.
+
+    Structural guard against the original bug returning: no assignment to
+    ``_last_self_telemetry_update`` may live inside the ``try`` that issues the
+    request, because every path in there other than success would skip it.
+    """
+    try_node = _self_telemetry_gate()
+    inside = _assignments_to(try_node, "_last_self_telemetry_update")
+    assert not inside, (
+        "_last_self_telemetry_update is assigned inside the get_self_telemetry() "
+        "try block; a failing node would never re-arm the interval and would be "
+        "retried on every coordinator tick"
+    )
+
+
+def test_error_reported_flag_is_cleared_on_success():
+    """A success must reset the flag so a later regression is reported again."""
+    try_node = _self_telemetry_gate()
+    assigns = _assignments_to(try_node, "_self_telemetry_error_reported")
+    values = {
+        n.value.value for n in assigns if isinstance(n.value, ast.Constant)
+    }
+    assert values == {True, False}, (
+        "expected _self_telemetry_error_reported to be set True on failure and "
+        f"cleared False on success, found {values}"
+    )