From a37be4b2f7f75dc467cb3bc0fe2fe3736eacb3b3 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 12 Sep 2026 17:32:41 -0700 Subject: [PATCH] fix(neighbours): only treat a reception as our echo while that send is on air The self-echo guard registered a packet's hash and forwarded path when the engine decided to forward it and kept it for 30 s. Two things could then hide a real neighbour from the link tracker: - A neighbour whose 1-byte path hash collides with ours, repeating the same packet from the same upstream path, carries exactly the path we would have sent. - The registration happened before the duty-cycle gate, so a packet that was never transmitted still suppressed matching receptions. Registration now happens at the physical send, inside the TX lock after the duty-cycle gate, and records which radio is sending. A send that fails is forgotten. A reception counts as our echo only when it arrives on a radio other than the one that sent it, and while that send is on the air or within 2 s of it finishing, which covers a remote modem's delivery delay. Single-radio nodes keep no records at all, since a lone radio cannot hear itself. A colliding neighbour is now mistaken for our echo only if it repeats the same packet on the same path, on the other radio, inside that window. --- repeater/engine.py | 102 +++++++++++++++++++++++++++--------- tests/test_fabric_fanout.py | 65 +++++++++++++++++++++++ tests/test_tx_lock.py | 2 + 3 files changed, 143 insertions(+), 26 deletions(-) diff --git a/repeater/engine.py b/repeater/engine.py index e961e61..a8ee19c 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -208,9 +208,9 @@ class RepeaterHandler(BaseHandler): config.get("mesh", {}).get("loop_detect", LOOP_DETECT_OFF) ) self.neighbour_link_tracker = NeighbourLinkTracker(config) - # (packet hash, path) of recent transmissions. Two radios sharing one - # frequency hear each other send; that echo ends in our own path hash and - # must not make this node its own neighbour. + # (packet hash, path) -> recent physical sends, each with its radio and + # timing. Two radios sharing one frequency hear each other send; that echo + # ends in our own path hash and must not make this node its own neighbour. self._recent_own_tx: OrderedDict = OrderedDict() radio = dispatcher.radio if dispatcher else None @@ -405,7 +405,7 @@ class RepeaterHandler(BaseHandler): self.radio_config["coding_rate"], self.radio_config["preamble_length"], ).score - if not self._is_own_echo(pkt_hash_full, original_path_hashes): + if not self._is_own_echo(pkt_hash_full, original_path_hashes, rx_radio_id): self.neighbour_link_tracker.observe( packet, route_type=route_type, @@ -443,9 +443,6 @@ class RepeaterHandler(BaseHandler): # Capture the forwarded path (after modification) forwarded_path_hashes = fwd_pkt.get_path_hashes_hex() - # Before the first send: an echo can arrive while a later egress of - # the same fan-out is still on the air. - self._note_own_transmission(pkt_hash_full, forwarded_path_hashes) # MeshCore queues multi-ack redundancy copies ahead of the primary # ACK at the same scheduled time, so create their TX tasks first. @@ -749,34 +746,83 @@ class RepeaterHandler(BaseHandler): return self._append_recent_packet(packet_record) - OWN_ECHO_TTL_SECONDS = 30.0 + # Another radio of this node hears a send only while it is on the air or + # shortly after, allowing for a remote modem's delivery delay. + OWN_ECHO_MARGIN_SECONDS = 2.0 + # A send that never reports back stops counting after this long. + OWN_ECHO_INFLIGHT_LIMIT_SECONDS = 60.0 OWN_ECHO_MAX_ENTRIES = 256 - def _note_own_transmission(self, packet_hash: str, path_hashes) -> None: - """Remember a packet about to be sent, so its echo can be recognised.""" + def _note_own_send_started(self, packet: Packet, radio_id: Optional[str]): + """Record a physical send about to go on air; return a handle for its outcome. + + Only on a node with two or more radios: a lone radio cannot hear itself. + """ + _, radio_ids = self._fabric_endpoints() + if not radio_ids or len(radio_ids) < 2: + return None + path_hashes = packet.get_path_hashes_hex() if not path_hashes: - return - key = (packet_hash, tuple(path_hashes)) - self._recent_own_tx.pop(key, None) - self._recent_own_tx[key] = time.monotonic() + return None + key = (packet.calculate_packet_hash().hex().upper(), tuple(path_hashes)) + record = {"radio_id": radio_id, "started": time.monotonic(), "finished": None} + sends = self._recent_own_tx.pop(key, []) + sends.append(record) + self._recent_own_tx[key] = sends while len(self._recent_own_tx) > self.OWN_ECHO_MAX_ENTRIES: self._recent_own_tx.popitem(last=False) + return key, record - def _is_own_echo(self, packet_hash: str, path_hashes) -> bool: - """True when a reception carries exactly the path this node just sent. + def _note_own_send_finished(self, handle, packet: Packet, *, sent: bool) -> None: + """Close a send: keep it for its echo window, or forget it if nothing went on air.""" + if handle is None: + return + key, record = handle + if record["finished"] is not None: + return + record["finished"] = time.monotonic() + if not sent: + sends = [r for r in self._recent_own_tx.get(key, []) if r is not record] + if sends: + self._recent_own_tx[key] = sends + else: + self._recent_own_tx.pop(key, None) + return + if record["radio_id"] is None: + metadata = getattr(packet, "_tx_metadata", None) + if isinstance(metadata, dict): + record["radio_id"] = metadata.get("radio_id") or metadata.get("tx_radio_id") - A neighbour repeating our copy appends its own hash, so only our own - transmission, heard by a second radio on the same frequency, matches. + def _own_send_is_live(self, record: dict, now: float) -> bool: + if record["finished"] is None: + return now - record["started"] <= self.OWN_ECHO_INFLIGHT_LIMIT_SECONDS + return now - record["finished"] <= self.OWN_ECHO_MARGIN_SECONDS + + def _is_own_echo(self, packet_hash: str, path_hashes, rx_radio_id) -> bool: + """True when another radio of this node is hearing our own transmission. + + The reception must carry exactly the path we sent, arrive on a radio other + than the one that sent it, and arrive while that send is on the air or + within OWN_ECHO_MARGIN_SECONDS of it finishing. A neighbour repeating our + copy appends its own hash; one whose 1-byte hash collides with ours is only + mistaken for an echo if it repeats the same packet on the same path inside + that window. """ - if not path_hashes or not self._recent_own_tx: + if rx_radio_id is None or not path_hashes or not self._recent_own_tx: return False now = time.monotonic() while self._recent_own_tx: - oldest = next(iter(self._recent_own_tx.values())) - if now - oldest <= self.OWN_ECHO_TTL_SECONDS: + sends = next(iter(self._recent_own_tx.values())) + if any(self._own_send_is_live(record, now) for record in sends): break self._recent_own_tx.popitem(last=False) - return (packet_hash, tuple(path_hashes)) in self._recent_own_tx + for record in self._recent_own_tx.get((packet_hash, tuple(path_hashes)), ()): + if not self._own_send_is_live(record, now): + continue + sender = record["radio_id"] + if sender is None or str(sender) != str(rx_radio_id): + return True + return False def record_duplicate(self, packet: Packet, rssi: int = 0, snr: float = 0.0) -> None: """Record a known-duplicate packet for UI/storage visibility without forwarding. @@ -812,7 +858,8 @@ class RepeaterHandler(BaseHandler): self.radio_config["coding_rate"], self.radio_config["preamble_length"], ).score - if not self._is_own_echo(pkt_hash_full, original_path_hashes): + rx_radio_id = getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None) + if not self._is_own_echo(pkt_hash_full, original_path_hashes, rx_radio_id): self.neighbour_link_tracker.observe( packet, route_type=route_type, @@ -821,9 +868,7 @@ class RepeaterHandler(BaseHandler): snr=float(snr), score=score, is_duplicate=True, - rx_radio_id=( - getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None) - ), + rx_radio_id=rx_radio_id, ) packet_record = self._build_packet_record( @@ -1780,6 +1825,9 @@ class RepeaterHandler(BaseHandler): ) return False + # Registered as the send starts: another radio of this node can + # hear it while it is still on the air. + own_send = self._note_own_send_started(fwd_pkt, preferred_tx_radio_id) try: if preferred_tx_radio_id: try: @@ -1794,6 +1842,7 @@ class RepeaterHandler(BaseHandler): ) else: sent = await self.dispatcher.send_packet(fwd_pkt, wait_for_ack=False) + self._note_own_send_finished(own_send, fwd_pkt, sent=bool(sent)) if not sent: logger.warning( "Retransmit failed (attempt %d): dispatcher returned false", @@ -1812,6 +1861,7 @@ class RepeaterHandler(BaseHandler): ) return True except Exception as e: + self._note_own_send_finished(own_send, fwd_pkt, sent=False) logger.error(f"Retransmit failed (attempt {attempt + 1}): {e}") if local_transmission and attempt == 0: pass # release lock, outer loop sleeps, then retries diff --git a/tests/test_fabric_fanout.py b/tests/test_fabric_fanout.py index 5ca3f64..5975582 100644 --- a/tests/test_fabric_fanout.py +++ b/tests/test_fabric_fanout.py @@ -289,6 +289,71 @@ async def test_own_transmission_heard_on_the_other_radio_is_not_a_neighbour(): assert [r["drop_reason"] for r in rig.records()[1:]] == ["Duplicate"] * 3 +@pytest.mark.asyncio +async def test_a_copy_heard_on_the_radio_that_sent_it_is_not_our_echo(): + """A radio cannot hear its own transmission, so that copy came from a neighbour.""" + rig = _Rig() # bridge only: heard on local, sent on link + payload = b"\x51\x52\x53" + assert await rig.handler(_flood(payload=payload, path=b"\x11", rx="local"), _rx("local")) + assert rig.air.order == ["link"] + + collided = _flood(payload=payload, path=bytes([0x11, LOCAL_HASH]), rx="link") + assert await rig.handler(collided, _rx("link")) is False + assert f"1:{LOCAL_HASH:02X}" in rig.handler.neighbour_link_tracker.links + + echo = _flood(payload=payload, path=bytes([0x11, LOCAL_HASH]), rx="local") + rig.handler.neighbour_link_tracker.links.clear() + assert await rig.handler(echo, _rx("local")) is False + assert f"1:{LOCAL_HASH:02X}" not in rig.handler.neighbour_link_tracker.links + + +@pytest.mark.asyncio +async def test_a_matching_copy_after_the_echo_window_is_a_neighbour(): + rig = _Rig({"repeat_on_ingress": True}) + rig.handler.OWN_ECHO_MARGIN_SECONDS = -1.0 # every finished send is already outside it + payload = b"\x61\x62\x63" + assert await rig.handler(_flood(payload=payload, path=b"\x11", rx="local"), _rx("local")) + + late = _flood(payload=payload, path=bytes([0x11, LOCAL_HASH]), rx="link") + assert await rig.handler(late, _rx("link")) is False + + assert f"1:{LOCAL_HASH:02X}" in rig.handler.neighbour_link_tracker.links + assert not rig.handler._recent_own_tx # expired sends are dropped + + +@pytest.mark.asyncio +async def test_a_send_refused_by_duty_cycle_leaves_no_echo_to_match(): + rig = _Rig({"repeat_on_ingress": True}) + rig.handler.airtime_mgr.can_transmit = MagicMock(return_value=(False, 5.0)) + payload = b"\x71\x72\x73" + assert ( + await rig.handler(_flood(payload=payload, path=b"\x11", rx="local"), _rx("local")) is False + ) + assert rig.air.frames == [] + + collided = _flood(payload=payload, path=bytes([0x11, LOCAL_HASH]), rx="link") + assert await rig.handler(collided, _rx("link")) is False + + assert not rig.handler._recent_own_tx + assert f"1:{LOCAL_HASH:02X}" in rig.handler.neighbour_link_tracker.links + + +@pytest.mark.asyncio +async def test_a_failed_send_is_forgotten(): + rig = _Rig() + rig.link.fail = True + assert await rig.handler(_flood(path=b"\x11", rx="local"), _rx("local")) is False + assert not rig.handler._recent_own_tx + + +@pytest.mark.asyncio +async def test_a_single_radio_node_keeps_no_echo_records(): + rig = _Rig() + with patch.object(rig.handler, "_fabric_endpoints", return_value=(None, ["local"])): + assert await rig.handler(_flood(path=b"\x11", rx="local"), _rx("local")) + assert not rig.handler._recent_own_tx + + @pytest.mark.asyncio async def test_egress_follows_captured_ingress_not_the_latest_rx(): rig = _Rig({"repeat_on_ingress": True}) diff --git a/tests/test_tx_lock.py b/tests/test_tx_lock.py index 31c3768..5de10b4 100644 --- a/tests/test_tx_lock.py +++ b/tests/test_tx_lock.py @@ -17,6 +17,7 @@ or: import asyncio import time import unittest +from collections import OrderedDict from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock @@ -61,6 +62,7 @@ def _make_handler(): h.airtime_mgr.can_transmit.return_value = (True, 0.0) h.airtime_mgr.calculate_airtime.return_value = 100.0 h._tx_lock = asyncio.Lock() + h._recent_own_tx = OrderedDict() h.sent_flood_count = 0 h.sent_direct_count = 0 # Stub out _record_packet_sent so it doesn't touch packet.header constants