Fix ingestor reconnection detection (#361)

data/mesh_ingestor/daemon.py

@@ -167,6 +167,45 @@ def _is_ble_interface(iface_obj) -> bool:
     return "ble_interface" in module_name
 
 
+def _connected_state(candidate) -> bool | None:
+    """Return the connection state advertised by ``candidate``.
+
+    Parameters:
+        candidate: Attribute returned from ``iface.isConnected`` on a
+            Meshtastic interface. The value may be a boolean, a callable that
+            yields a boolean, or a :class:`threading.Event` instance.
+
+    Returns:
+        ``True`` when the interface is believed to be connected, ``False``
+        when it appears disconnected, and ``None`` when the state cannot be
+        determined from the provided attribute.
+    """
+
+    if candidate is None:
+        return None
+
+    if isinstance(candidate, threading.Event):
+        return candidate.is_set()
+
+    is_set_method = getattr(candidate, "is_set", None)
+    if callable(is_set_method):
+        try:
+            return bool(is_set_method())
+        except Exception:
+            return None
+
+    if callable(candidate):
+        try:
+            return bool(candidate())
+        except Exception:
+            return None
+
+    try:
+        return bool(candidate)
+    except Exception:  # pragma: no cover - defensive guard
+        return None
+
+
 def main() -> None:
     """Run the mesh ingestion daemon until interrupted."""
 
@@ -411,13 +450,20 @@ def main() -> None:
 
                 connected_attr = getattr(iface, "isConnected", None)
                 believed_disconnected = False
-                if callable(connected_attr):
-                    try:
-                        believed_disconnected = not bool(connected_attr())
-                    except Exception:
-                        believed_disconnected = False
-                elif connected_attr is not None:
-                    believed_disconnected = not bool(connected_attr)
+                connected_state = _connected_state(connected_attr)
+                if connected_state is None:
+                    if callable(connected_attr):
+                        try:
+                            believed_disconnected = not bool(connected_attr())
+                        except Exception:
+                            believed_disconnected = False
+                    elif connected_attr is not None:
+                        try:
+                            believed_disconnected = not bool(connected_attr)
+                        except Exception:  # pragma: no cover - defensive guard
+                            believed_disconnected = False
+                else:
+                    believed_disconnected = not connected_state
 
                 should_reconnect = believed_disconnected or (
                     inactivity_elapsed >= inactivity_reconnect_secs
@@ -468,5 +514,6 @@ __all__ = [
     "_node_items_snapshot",
     "_subscribe_receive_topics",
     "_is_ble_interface",
+    "_connected_state",
     "main",
 ]

tests/test_mesh.py

@@ -1201,6 +1201,81 @@ def test_main_retries_interface_creation(mesh_module, monkeypatch):
     assert iface.closed is True
 
 
+def test_connected_state_handles_threading_event(mesh_module):
+    mesh = mesh_module
+
+    event = mesh.threading.Event()
+    assert mesh._connected_state(event) is False
+
+    event.set()
+    assert mesh._connected_state(event) is True
+
+
+def test_main_reconnects_when_connection_event_clears(mesh_module, monkeypatch):
+    mesh = mesh_module
+
+    attempts = []
+    interfaces = []
+    current_iface = {"obj": None}
+    import threading as real_threading_module
+
+    real_event_cls = real_threading_module.Event
+
+    class DummyInterface:
+        def __init__(self):
+            self.nodes = {}
+            self.isConnected = real_event_cls()
+            self.isConnected.set()
+            self.close_calls = 0
+
+        def close(self):
+            self.close_calls += 1
+
+    def fake_create(port):
+        iface = DummyInterface()
+        attempts.append(port)
+        interfaces.append(iface)
+        current_iface["obj"] = iface
+        return iface, port
+
+    class DummyStopEvent:
+        def __init__(self):
+            self._flag = False
+            self.wait_calls = 0
+
+        def is_set(self):
+            return self._flag
+
+        def set(self):
+            self._flag = True
+
+        def wait(self, timeout):
+            self.wait_calls += 1
+            if self.wait_calls == 1:
+                iface = current_iface["obj"]
+                assert iface is not None, "interface should be available"
+                iface.isConnected.clear()
+                return self._flag
+            self._flag = True
+            return True
+
+    monkeypatch.setattr(mesh, "PORT", "/dev/ttyTEST")
+    monkeypatch.setattr(mesh, "_create_serial_interface", fake_create)
+    monkeypatch.setattr(mesh.threading, "Event", DummyStopEvent)
+    monkeypatch.setattr(mesh.signal, "signal", lambda *_, **__: None)
+    monkeypatch.setattr(mesh, "SNAPSHOT_SECS", 0)
+    monkeypatch.setattr(mesh, "_RECONNECT_INITIAL_DELAY_SECS", 0)
+    monkeypatch.setattr(mesh, "_RECONNECT_MAX_DELAY_SECS", 0)
+    monkeypatch.setattr(mesh, "_CLOSE_TIMEOUT_SECS", 0)
+
+    mesh.main()
+
+    assert len(attempts) == 2
+    assert len(interfaces) == 2
+    assert interfaces[0].close_calls >= 1
+    assert interfaces[1].close_calls >= 1
+
+
 def test_main_recreates_interface_after_snapshot_error(mesh_module, monkeypatch):
     mesh = mesh_module