Harden MeshCore messaging against contact-fetch bus hammering.

Cooldown soft-fail cooldown, pause auto-fetch around TX/sync, throttle DM contact refresh, and serve the contact picker from SQLite so live get_contacts cannot wedge channel sends.

Author: KMX415

src/api/routes/messages.py

@@ -237,6 +237,12 @@ async def get_channels():
 
 @router.get("/contacts")
 async def get_contacts():
+    """Return messaging contacts from the local node roster only.
+
+    Does not call live MeshCore ``get_contacts``: that command shares
+    the companion serial bus with channel TX and was wedging sends
+    whenever the contact picker or enrichment path refreshed.
+    """
     contacts = []
 
     _synthetic = {"rf_log", "raw", "mc:channel", "unknown", ""}
@@ -254,21 +260,6 @@ async def get_contacts():
                 "last_heard": n.get("last_heard", ""),
             })
 
-    if _meshcore_tx and _meshcore_tx.connected:
-        mc_contacts = await _meshcore_tx.get_contacts()
-        for contact in mc_contacts:
-            pk = contact.get("public_key", "")
-            canonical = pk[:12].lower() if len(pk) >= 12 else pk.lower()
-            name = contact.get("name", "")
-            if not name or name.lower() == canonical:
-                name = await _resolve_display_name(canonical, "meshcore") or canonical
-            contacts.append({
-                "node_id": canonical,
-                "name": name,
-                "protocol": "meshcore",
-                "last_heard": "",
-            })
-
     return contacts
 
 

src/api/server.py

@@ -1014,6 +1014,8 @@ def _setup_message_interception(
 
     mc_name_cache: dict[str, str] = {}
     mc_pubkey_canon: dict[str, str] = {}
+    _mc_refresh_state = {"last": 0.0}
+    _MC_REFRESH_MIN_INTERVAL_S = 60.0
 
     from src.api.channel_hash_resolver import ChannelHashResolver
 
@@ -1038,6 +1040,12 @@ async def _refresh_mc_contacts() -> None:
         if not meshcore_tx or not meshcore_tx.connected:
             logger.debug("MC contact refresh skipped: not connected")
             return
+        import time as _time
+
+        now = _time.monotonic()
+        if now - _mc_refresh_state["last"] < _MC_REFRESH_MIN_INTERVAL_S:
+            return
+        _mc_refresh_state["last"] = now
         try:
             contacts = await meshcore_tx.get_contacts()
             for c in contacts:

src/transmit/meshcore_contacts.py

@@ -45,6 +45,15 @@ def store(self, rows: list[dict]) -> None:
         self._rows = list(rows)
         self._fetched_at = time.monotonic()
 
+    def note_soft_fail(self) -> None:
+        """Refresh TTL after a timeout/error so callers back off.
+
+        Without this, an empty cache causes every Messages/enrichment
+        caller to burn another 5s live ``get_contacts`` and wedge the
+        companion command channel.
+        """
+        self._fetched_at = time.monotonic()
+
     def invalidate(self) -> None:
         self._fetched_at = 0.0
 

src/transmit/meshcore_tx_client.py

@@ -125,7 +125,9 @@ def _fail_timeout(self, reason: str) -> SendResult:
 
         Credit: javastraat/meshpoint b04e91c
         """
-        self._contact_cache.invalidate()
+        # Cooldown, do not wipe: keep stale names and stop contact
+        # fetchers from immediately re-hammering a wedged companion.
+        self._contact_cache.note_soft_fail()
         trigger = getattr(self._source, "_trigger_reconnect", None)
         if callable(trigger):
             trigger(reason)
@@ -138,14 +140,23 @@ async def _run_tx_command(
         success_log: str,
         timeout_label: str,
     ) -> SendResult:
-        """Run one companion command under the serial lock."""
+        """Run one companion command under the serial lock.
+
+        Pauses auto message fetching for the command window so the
+        library's background poll cannot steal OK/ERROR events
+        (same pattern as set_radio).
+        """
         if not self.connected:
             return SendResult(success=False, error="Not connected")
         try:
             async with self._cmd_lock:
                 if not self.connected or self._mc is None:
                     return SendResult(success=False, error="Not connected")
-                result = await asyncio.wait_for(factory(), timeout=10.0)
+                await self._pause_auto_fetch()
+                try:
+                    result = await asyncio.wait_for(factory(), timeout=10.0)
+                finally:
+                    await self._resume_auto_fetch()
         except asyncio.TimeoutError:
             return self._fail_timeout(timeout_label)
         except Exception as exc:
@@ -191,6 +202,42 @@ async def _run_tx_command(
         await self._run_post_command()
         return SendResult(success=True, event_type=event_type)
 
+    async def _pause_auto_fetch(self) -> None:
+        mc = self._mc
+        if mc is None:
+            return
+        stop = getattr(mc, "stop_auto_message_fetching", None)
+        if not callable(stop):
+            return
+        try:
+            await stop()
+        except Exception:
+            logger.debug("Could not pause MeshCore auto-fetch", exc_info=True)
+
+    async def _resume_auto_fetch(self) -> None:
+        # Prefer the capture-source restart (rebinds subscriptions) when
+        # bound; otherwise poke the library directly.
+        restart = getattr(self._source, "restart_auto_fetching", None)
+        if callable(restart):
+            try:
+                await restart()
+                return
+            except Exception:
+                logger.debug(
+                    "Could not restart MeshCore auto-fetch via source",
+                    exc_info=True,
+                )
+        mc = self._mc
+        if mc is None:
+            return
+        start = getattr(mc, "start_auto_message_fetching", None)
+        if not callable(start):
+            return
+        try:
+            await start()
+        except Exception:
+            logger.debug("Could not resume MeshCore auto-fetch", exc_info=True)
+
     async def create_connection(
         self,
         port: str,
@@ -373,10 +420,18 @@ async def sync_channels(self, channel_keys: dict) -> None:
         if not self.connected:
             logger.debug("sync_channels: not connected, skipping")
             return
-        await MeshcoreChannelSync(
-            self._mc,
-            post_command=self._run_post_command,
-        ).sync(channel_keys)
+        async with self._cmd_lock:
+            if not self.connected or self._mc is None:
+                return
+            await self._pause_auto_fetch()
+            try:
+                await MeshcoreChannelSync(
+                    self._mc,
+                    post_command=None,
+                ).sync(channel_keys)
+            finally:
+                await self._resume_auto_fetch()
+            await self._run_post_command()
 
     async def get_contacts(self, *, force: bool = False) -> list[dict]:
         """Retrieve the companion's contact list.
@@ -407,24 +462,30 @@ async def get_contacts(self, *, force: bool = False) -> list[dict]:
                 )
         except asyncio.TimeoutError:
             logger.warning("get_contacts timed out waiting for companion")
+            self._contact_cache.note_soft_fail()
             return self._contact_cache.get_stale()
         except Exception:
             logger.exception("Failed to retrieve MeshCore contacts")
+            self._contact_cache.note_soft_fail()
             return self._contact_cache.get_stale()
 
         contacts = MeshcoreContactParser.from_command_result(result)
         soft_fail = result is None or MeshcoreContactParser._is_error_event(
             result
         )
         if soft_fail:
+            # Stamp TTL even on failure so queued callers do not each
+            # burn another live 5s get_contacts on a wedged companion.
+            self._contact_cache.note_soft_fail()
             stale = self._contact_cache.get_stale()
             if stale:
                 logger.info(
                     "get_contacts: soft fail, using stale cache (%d)",
                     len(stale),
                 )
                 return stale
-        else:
-            self._contact_cache.store(contacts)
+            logger.info("get_contacts: soft fail, empty roster (cooling down)")
+            return []
+        self._contact_cache.store(contacts)
         logger.info("get_contacts: %d contacts parsed", len(contacts))
         return contacts

tests/test_meshcore_contacts.py

@@ -89,6 +89,13 @@ def test_fresh_then_stale(self):
         self.assertIsNone(cache.get_fresh())
         self.assertEqual(len(cache.get_stale()), 1)
 
+    def test_soft_fail_starts_cooldown_with_empty_roster(self):
+        cache = MeshcoreContactCache(ttl_seconds=60.0)
+        self.assertIsNone(cache.get_fresh())
+        cache.note_soft_fail()
+        # Empty list (not None) so callers skip another live fetch.
+        self.assertEqual(cache.get_fresh(), [])
+
 
 if __name__ == "__main__":
     unittest.main()