Merge pull request #341 from agessaman/fix/trace-late-response

fix(trace): accept a late reply to an earlier attempt, and allow for repeater TX delay

CHANGELOG.md

@@ -38,6 +38,7 @@ semantic versioning.
 
 ### Fixed
 
+- `trace` no longer gives up on a reply that is on its way. Each attempt listened only for its own tag, so when the first trace's reply arrived after its timeout, the retry had already replaced it and the reply was dropped; on a mesh where repeaters wait a second or more before forwarding, a three-hop trace always failed. A reply to any attempt now counts until the last attempt times out, and one that arrives during the retry delay ends the trace without sending again. The timeout defaults are now 2.0 s base and 1.5 s per hop (were 1.0 s and 0.5 s, about 2.5 s for three hops); a config that sets them keeps its values.
 - Repeat counts for the bot's own messages (the web viewer's command log, the DARC MoWaS retry check) count the bot's own packets. A sent message was matched to the first packet the radio heard within 30 seconds, whatever it was, so on a busy mesh it usually took someone else's packet: its repeats were counted as the bot's, and the bot's real repeats were not. A channel message is now matched only by a packet that decrypts on the same channel to the radio's name and the same text, a DM by a direct message addressed to the recipient from the bot, and a trace by its tag. The repeat counts are written to the database off the event loop, and each repeat no longer adds an entry to Python's import path.
 - A successful `trace` now adds its links to the mesh graph, as `[Trace_Command] update_graph_one_byte` and `update_graph_two_byte` (both on by default) describe. The bot's own prefix was read from a radio attribute meshcore_py doesn't provide, so it was always unknown and every trace's graph update was skipped; it now comes from the radio's reported public key, at the trace's hash width. `update_graph_two_byte` was never read: 2-byte traces followed `update_graph_one_byte`, and now follow their own setting.
 - A new contact's route comes from its own advert. NEW_CONTACT took the route, packet hash and signal data from whichever of the last ten received packets came first, usually a different node's message or advert, and wrote that route into the contact's `out_path` (sent to the radio with `auto_manage_contacts = bot`), the mesh graph and the observed paths. It now uses the first copy heard of the advert NEW_CONTACT reports, matched by public key and advert timestamp, and leaves the mesh graph and observed paths to the advert packet itself, which already recorded them (a correct match was counted twice). With `auto_manage_contacts = bot`, a companion is still added to the radio once its advert has been tracked: the add was skipped whenever the packet hash had been seen before, which the correct hash always has.

config.ini.example

@@ -1734,9 +1734,11 @@ enabled = true
 maximum_hops = 5
 # Trace mode: one_byte (default) or two_byte (when firmware supports it)
 trace_mode = one_byte
-# Timeout: total = timeout_base_seconds + (path hops * timeout_per_hop_seconds); typical 6 hops ~1s, 10 ~2s
-timeout_base_seconds = 1.0
-timeout_per_hop_seconds = 0.5
+# Timeout per attempt: timeout_base_seconds + (path hops * timeout_per_hop_seconds). Each repeater waits a
+# random TX delay before forwarding, so allow a second or more per hop; a reply to an earlier attempt still
+# counts until the last attempt times out
+timeout_base_seconds = 2.0
+timeout_per_hop_seconds = 1.5
 # Retries: max attempts (default 2 = try once, then once more after delay)
 trace_retry_count = 2
 # Seconds to wait before retrying a failed trace

docs/command-reference.md

@@ -771,7 +771,7 @@ tracer
 ```
 With no path, both use the path your message took to reach the bot (like the test command).
 
-**Config:** `[Trace_Command]` — `enabled`, `maximum_hops`, `trace_mode` (one_byte/two_byte), `timeout_base_seconds` (default 1.0), `timeout_per_hop_seconds` (default 0.5), `trace_retry_count` (default 2 attempts), `trace_retry_delay_seconds` (default 1.0), `update_graph_one_byte`, `update_graph_two_byte`. Total wait per attempt = base + (hops × per_hop). On failure, waits then retries up to `trace_retry_count` times.
+**Config:** `[Trace_Command]` — `enabled`, `maximum_hops`, `trace_mode` (one_byte/two_byte), `timeout_base_seconds` (default 2.0), `timeout_per_hop_seconds` (default 1.5), `trace_retry_count` (default 2 attempts), `trace_retry_delay_seconds` (default 1.0), `update_graph_one_byte`, `update_graph_two_byte`. Total wait per attempt = base + (hops × per_hop). On failure, waits then retries up to `trace_retry_count` times; a reply to an earlier attempt still counts until the last attempt times out.
 
 **Response:** Compact trace result: tag, hop count, SNR per hop, and optional graph update when enabled.
 

modules/trace_runner.py

@@ -28,23 +28,21 @@ class RunTraceResult:
 
 def _get_timeout_seconds(bot: Any, path: Optional[list[str]]) -> float:
     """Compute total timeout from path length and config."""
-    per_hop = bot.config.getfloat("Trace_Command", "timeout_per_hop_seconds", fallback=0.5)
-    base = bot.config.getfloat("Trace_Command", "timeout_base_seconds", fallback=1.0)
+    per_hop = bot.config.getfloat("Trace_Command", "timeout_per_hop_seconds", fallback=1.5)
+    base = bot.config.getfloat("Trace_Command", "timeout_base_seconds", fallback=2.0)
     hops = len(path) if path else 0
-    # Typical: ~1s for 6 hops, ~2s for 10 hops; base + per-hop gives margin
+    # Each repeater waits a random TX delay before forwarding, often a second or more per hop
     total = base + max(1, hops) * per_hop
     return total
 
 
-async def _run_trace_attempt(
+async def _send_trace_attempt(
     bot: Any,
-    path: Optional[list[str]],
     path_string: Optional[str],
     flags: int,
-    timeout_seconds: float,
     tag: int,
-) -> RunTraceResult:
-    """Execute one trace attempt: send_trace then wait for TRACE_DATA. Caller handles retries."""
+) -> Optional[str]:
+    """Record and send one trace. Returns an error message, or None once the trace is sent."""
     try:
         if hasattr(bot, "transmission_tracker") and bot.transmission_tracker:
             bot.transmission_tracker.record_transmission(
@@ -65,44 +63,20 @@ async def _run_trace_attempt(
             path=path_string,
         )
     except Exception as e:
-        return RunTraceResult(success=False, tag=tag, error_message=str(e))
+        return str(e)
 
     if result.type == EventType.ERROR:
-        reason = result.payload.get("reason", "unknown error")
-        return RunTraceResult(success=False, tag=tag, error_message=reason)
+        return str(result.payload.get("reason", "unknown error"))
+    return None
 
-    path_str = ",".join(path) if path else "(flood)"
+
+async def _wait_for(arrived: asyncio.Event, timeout: float) -> bool:
+    """Wait up to timeout seconds for arrived; True if it was set."""
     try:
-        event = await bot.meshcore.wait_for_event(
-            EventType.TRACE_DATA,
-            attribute_filters={"tag": tag},
-            timeout=timeout_seconds,
-        )
-    except Exception as e:
-        return RunTraceResult(
-            success=False,
-            tag=tag,
-            error_message=f"Timeout or error waiting for trace (path: {path_str}): {e}",
-        )
-
-    if not event:
-        return RunTraceResult(
-            success=False,
-            tag=tag,
-            error_message=f"No trace response within timeout (path: {path_str})",
-        )
-
-    payload = event.payload or {}
-    path_nodes = payload.get("path") or []
-    path_len = payload.get("path_len", 0)
-    flags_val = payload.get("flags", 0)
-    return RunTraceResult(
-        success=True,
-        tag=tag,
-        path_nodes=path_nodes,
-        path_len=path_len,
-        flags=flags_val,
-    )
+        await asyncio.wait_for(arrived.wait(), timeout)
+    except asyncio.TimeoutError:
+        pass
+    return arrived.is_set()
 
 
 async def run_trace(
@@ -114,6 +88,9 @@ async def run_trace(
     """
     Send a trace and wait for TRACE_DATA. Retries on failure per config (default 2 attempts, 1s delay).
 
+    A reply to any attempt counts until the last attempt's timeout runs out, so a slow mesh that
+    answers the first trace while the retry is out still succeeds.
+
     Args:
         bot: MeshCoreBot instance (must have meshcore, config, transmission_tracker).
         path: Optional list of 2-char hex node IDs (e.g. ["01", "7a", "55"]). None = flood.
@@ -141,30 +118,75 @@ async def run_trace(
     retry_delay = max(0.0, bot.config.getfloat("Trace_Command", "trace_retry_delay_seconds", fallback=1.0))
 
     path_str_debug = path_string if path_string else "(flood)"
-    last_result: Optional[RunTraceResult] = None
+    path_str = ",".join(path) if path else "(flood)"
 
-    for attempt in range(max_attempts):
-        tag = random.randint(1, 0xFFFFFFFF)
-        if attempt > 0:
-            bot.logger.debug("Trace retry %s/%s after %.1fs delay", attempt + 1, max_attempts, retry_delay)
-            await asyncio.sleep(retry_delay)
-        bot.logger.debug(
-            "Trace: path=%s hops=%s timeout=%.1fs tag=%s attempt=%s/%s",
-            path_str_debug,
-            len(path) if path else 0,
-            timeout_seconds,
-            tag,
-            attempt + 1,
-            max_attempts,
-        )
-        last_result = await _run_trace_attempt(
-            bot, path, path_string, flags, timeout_seconds, tag
-        )
-        if last_result.success:
-            return last_result
+    sent_tags: list[int] = []
+    replies: list[tuple[int, Any]] = []
+    arrived = asyncio.Event()
 
-    return last_result or RunTraceResult(
+    def _on_trace_data(event: Any) -> None:
+        attributes = getattr(event, "attributes", None) or {}
+        payload = getattr(event, "payload", None) or {}
+        tag = attributes.get("tag", payload.get("tag"))
+        if tag in sent_tags and not replies:
+            replies.append((tag, event))
+            arrived.set()
+
+    # Subscribe before the first send so neither a fast reply nor a late one is missed
+    subscription = bot.meshcore.subscribe(EventType.TRACE_DATA, _on_trace_data)
+    last_error: Optional[str] = None
+    last_tag = 0
+    try:
+        for attempt in range(max_attempts):
+            tag = random.randint(1, 0xFFFFFFFF)
+            if attempt > 0:
+                bot.logger.debug("Trace retry %s/%s after %.1fs delay", attempt + 1, max_attempts, retry_delay)
+                if await _wait_for(arrived, retry_delay):
+                    break
+            bot.logger.debug(
+                "Trace: path=%s hops=%s timeout=%.1fs tag=%s attempt=%s/%s",
+                path_str_debug,
+                len(path) if path else 0,
+                timeout_seconds,
+                tag,
+                attempt + 1,
+                max_attempts,
+            )
+            sent_tags.append(tag)
+            last_tag = tag
+            error = await _send_trace_attempt(bot, path_string, flags, tag)
+            if error is not None:
+                last_error = error
+                # An earlier attempt's reply can still arrive
+                if arrived.is_set():
+                    break
+                continue
+            if await _wait_for(arrived, timeout_seconds):
+                break
+            last_error = f"No trace response within timeout (path: {path_str})"
+    finally:
+        try:
+            subscription.unsubscribe()
+        except Exception as e:
+            bot.logger.debug(f"Trace runner: failed to unsubscribe: {e}")
+
+    if replies:
+        reply_tag, event = replies[0]
+        payload = event.payload or {}
+        if reply_tag != last_tag:
+            bot.logger.debug(
+                "Trace: reply for attempt %s/%s (tag=%s)", sent_tags.index(reply_tag) + 1, max_attempts, reply_tag
+            )
+        return RunTraceResult(
+            success=True,
+            tag=reply_tag,
+            path_nodes=payload.get("path") or [],
+            path_len=payload.get("path_len", 0),
+            flags=payload.get("flags", 0),
+        )
+
+    return RunTraceResult(
         success=False,
-        tag=0,
-        error_message="Trace failed (no attempts run)",
+        tag=last_tag,
+        error_message=last_error or "Trace failed (no attempts run)",
     )

tests/test_trace_runner.py

@@ -0,0 +1,142 @@
+"""Tests for modules.trace_runner: retries, late replies and timeouts."""
+
+import asyncio
+import configparser
+from types import SimpleNamespace
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+from meshcore import EventType
+
+from modules.trace_runner import _get_timeout_seconds, run_trace
+
+
+class _FakeMeshCore:
+    """Dispatches TRACE_DATA to subscribers; replies are scheduled per sent tag."""
+
+    def __init__(self, reply_delays=None, error_reason=None):
+        # reply_delays: {attempt index: seconds until that attempt's reply arrives}
+        self.reply_delays = reply_delays or {}
+        self.error_reason = error_reason
+        self.subscribers = []
+        self.sent_tags = []
+        self.commands = SimpleNamespace(send_trace=self._send_trace)
+
+    def subscribe(self, event_type, callback, attribute_filters=None):
+        assert event_type == EventType.TRACE_DATA
+        entry = [callback]
+        self.subscribers.append(entry)
+        return SimpleNamespace(unsubscribe=lambda: self.subscribers.remove(entry))
+
+    def dispatch(self, tag):
+        payload = {"tag": tag, "flags": 1, "path_len": 1, "path": [{"hash": "0101", "snr": 11.5}, {"snr": 12.5}]}
+        event = SimpleNamespace(type=EventType.TRACE_DATA, payload=payload, attributes={"tag": tag, "auth_code": 0})
+        for (callback,) in list(self.subscribers):
+            callback(event)
+
+    async def _send_trace(self, auth_code, tag, flags, path):
+        attempt = len(self.sent_tags)
+        self.sent_tags.append(tag)
+        if self.error_reason:
+            return SimpleNamespace(type=EventType.ERROR, payload={"reason": self.error_reason})
+        delay = self.reply_delays.get(attempt)
+        if delay == 0:
+            self.dispatch(tag)
+        elif delay is not None:
+            asyncio.get_running_loop().call_later(delay, self.dispatch, tag)
+        return SimpleNamespace(type=EventType.MSG_SENT, payload={})
+
+
+def _make_bot(meshcore, retry_delay="0.05", attempts="2"):
+    bot = MagicMock()
+    bot.logger = Mock()
+    config = configparser.ConfigParser()
+    config.add_section("Trace_Command")
+    config.set("Trace_Command", "trace_retry_count", attempts)
+    config.set("Trace_Command", "trace_retry_delay_seconds", retry_delay)
+    bot.config = config
+    bot.meshcore = meshcore
+    bot.transmission_tracker = None
+    return bot
+
+
+@pytest.fixture
+def tags():
+    with patch("modules.trace_runner.random.randint", side_effect=[111, 222, 333]):
+        yield
+
+
+@pytest.mark.usefixtures("tags")
+class TestRunTrace:
+    def test_reply_to_first_attempt(self):
+        mc = _FakeMeshCore(reply_delays={0: 0.02})
+        result = asyncio.run(run_trace(_make_bot(mc), path=["0101"], timeout_seconds=0.2))
+        assert result.success and result.tag == 111
+        assert result.path_nodes[0] == {"hash": "0101", "snr": 11.5}
+        assert mc.sent_tags == [111]
+        assert mc.subscribers == []
+
+    def test_late_reply_to_first_attempt_counts_during_retry(self):
+        # The first trace's reply arrives after its own timeout, while the retry waits
+        mc = _FakeMeshCore(reply_delays={0: 0.3})
+        result = asyncio.run(run_trace(_make_bot(mc), path=["0101"], timeout_seconds=0.2))
+        assert result.success and result.tag == 111
+        assert mc.sent_tags == [111, 222]
+
+    def test_late_reply_during_retry_delay_skips_the_retry(self):
+        mc = _FakeMeshCore(reply_delays={0: 0.22})
+        bot = _make_bot(mc, retry_delay="0.3")
+        result = asyncio.run(run_trace(bot, path=["0101"], timeout_seconds=0.2))
+        assert result.success and result.tag == 111
+        assert mc.sent_tags == [111]
+
+    def test_reply_dispatched_before_waiting_is_not_missed(self):
+        mc = _FakeMeshCore(reply_delays={0: 0})
+        result = asyncio.run(run_trace(_make_bot(mc), path=["0101"], timeout_seconds=0.2))
+        assert result.success and result.tag == 111
+        assert mc.sent_tags == [111]
+
+    def test_no_reply_fails_after_every_attempt(self):
+        mc = _FakeMeshCore()
+        result = asyncio.run(run_trace(_make_bot(mc), path=["0101", "7a2a"], timeout_seconds=0.05))
+        assert not result.success
+        assert result.error_message == "No trace response within timeout (path: 0101,7a2a)"
+        assert mc.sent_tags == [111, 222]
+        assert mc.subscribers == []
+
+    def test_reply_after_last_timeout_is_too_late(self):
+        mc = _FakeMeshCore(reply_delays={1: 0.5})
+        result = asyncio.run(run_trace(_make_bot(mc), path=["0101"], timeout_seconds=0.05))
+        assert not result.success
+
+    def test_other_tags_are_ignored(self):
+        mc = _FakeMeshCore()
+
+        async def go():
+            asyncio.get_running_loop().call_later(0.02, mc.dispatch, 999)
+            return await run_trace(_make_bot(mc, attempts="1"), path=["0101"], timeout_seconds=0.1)
+
+        result = asyncio.run(go())
+        assert not result.success
+
+    def test_send_error_is_reported(self):
+        mc = _FakeMeshCore(error_reason="no route")
+        result = asyncio.run(run_trace(_make_bot(mc), path=["0101"], timeout_seconds=0.05))
+        assert not result.success
+        assert result.error_message == "no route"
+        assert mc.sent_tags == [111, 222]
+        assert mc.subscribers == []
+
+
+class TestTimeout:
+    def test_defaults_allow_for_repeater_tx_delay(self):
+        bot = MagicMock()
+        bot.config = configparser.ConfigParser()
+        assert _get_timeout_seconds(bot, ["0101", "7a2a", "0101"]) == pytest.approx(6.5)
+
+    def test_configured_values(self):
+        bot = MagicMock()
+        bot.config = configparser.ConfigParser()
+        bot.config.read_dict({"Trace_Command": {"timeout_base_seconds": "6", "timeout_per_hop_seconds": "1.5"}})
+        assert _get_timeout_seconds(bot, ["01", "02", "03"]) == pytest.approx(10.5)
+        assert _get_timeout_seconds(bot, None) == pytest.approx(7.5)