mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-08-08 01:42:49 +02:00
Refactor packet handling in RepeaterHandler and PacketRouter
- Introduced helper methods `_path_hash_display` and `_packet_record_src_dst` in `RepeaterHandler` to streamline path hash and source/destination hash extraction. - Updated `record_packet` method to utilize a new `_build_packet_record` method for improved readability and maintainability. - Enhanced `PacketRouter` comments for clarity on handling remote destinations and packet processing, ensuring better understanding of the routing logic.
This commit is contained in:
agessaman
@@ -0,0 +1,66 @@
|
||||
# DIRECT packets not forwarded (router consumes them)
|
||||
|
||||
## Summary
|
||||
|
||||
Besides ANON_REQ (fixed), these payload types are **always** marked `processed_by_injection` in the router and **never** passed to the engine, so they are **never forwarded** even when this repeater is a middle hop on a DIRECT path:
|
||||
|
||||
| Payload type | Router behavior | Can be DIRECT? | Should forward when middle hop? |
|
||||
|---------------------|-----------------|-----------------|----------------------------------|
|
||||
| **ACK** | Deliver to all companion bridges, set processed | Yes (return path) | Yes |
|
||||
| **PATH** | Deliver to companion(s) or all bridges (anon), set processed | Yes (path response) | Yes |
|
||||
| **LoginResponse** | Deliver to bridge(s), set processed | Yes (login response) | Yes |
|
||||
| **ProtocolResponse**| Deliver to companions, set processed | Yes (telemetry etc.) | Yes |
|
||||
| **Trace** | trace_helper only, set processed | Yes/No | No (diagnostic, not forwarded by design) |
|
||||
|
||||
So **ACK, PATH, LoginResponse (RESPONSE), and ProtocolResponse** are DIRECT packet types that we should be forwarding when we're in the path but currently are not, because the router consumes them every time.
|
||||
|
||||
## Detail by type
|
||||
|
||||
### ACK (AckHandler)
|
||||
|
||||
- **Router:** Delivers to all companion bridges (so sender sees send_confirmed), then sets `processed_by_injection = True`. Never passes to engine.
|
||||
- **Use case:** ACKs travel back along the path. When we're a middle hop we should forward the ACK toward the sender.
|
||||
- **Conclusion:** Should be passed to engine after companion delivery so DIRECT ACKs can be forwarded.
|
||||
|
||||
### PATH (PathHandler)
|
||||
|
||||
- **Router:** If dest in companion_bridges → deliver to that bridge, set processed. If dest not in bridges but we have companions → deliver to all bridges (anon), set processed. Only when path_helper runs (and no companion anon delivery) do we *not* set processed, so the packet can reach the engine.
|
||||
- **Use case:** PATH responses come back along the path. When we're a middle hop we should forward. When we're final hop (dest in companion or we're path_helper) we should not forward.
|
||||
- **Conclusion:** When we deliver to companions we still need to pass to engine so we can forward when we're a middle hop. Today we set processed in both companion branches so we never forward PATH.
|
||||
|
||||
### LoginResponse (RESPONSE, LoginResponseHandler)
|
||||
|
||||
- **Router:** All three branches (dest in companion, dest == local_hash, or anon to all bridges) set `processed_by_injection = True`. Never passes to engine.
|
||||
- **Use case:** Login responses come back along the path. When we're a middle hop we should forward.
|
||||
- **Conclusion:** Should be passed to engine after companion delivery so DIRECT RESPONSE can be forwarded.
|
||||
|
||||
### ProtocolResponse (ProtocolResponseHandler)
|
||||
|
||||
- **Router:** When we have companion_bridges we deliver and set `processed_by_injection = True`. Never passes to engine.
|
||||
- **Use case:** Protocol responses (telemetry, etc.) come back along the path. When we're a middle hop we should forward.
|
||||
- **Conclusion:** Should be passed to engine after companion delivery so DIRECT ProtocolResponse can be forwarded.
|
||||
|
||||
## Recommended fix (two parts)
|
||||
|
||||
### 1. Router: pass to engine after companion delivery
|
||||
|
||||
For **ACK, PATH, LoginResponse, ProtocolResponse**, do **not** set `processed_by_injection` so the packet is always passed to the engine after any companion delivery. That implies:
|
||||
|
||||
- **ACK:** Remove the unconditional `processed_by_injection = True` (or only set it when we have no companions and are definitely final). Then always pass to engine; engine will forward when we're next hop, drop when "Direct: not for us" or duplicate.
|
||||
- **PATH:** In both branches where we set processed (dest in companion, anon to all), stop setting `processed_by_injection` so the packet also goes to the engine. Keep companion delivery and `_record_for_ui` as today.
|
||||
- **LoginResponse:** In all three branches, stop setting `processed_by_injection` so the packet also goes to the engine.
|
||||
- **ProtocolResponse:** When we have companions, stop setting `processed_by_injection` so the packet also goes to the engine.
|
||||
|
||||
We still deliver to companions and call `_record_for_ui` where we do today; we just also pass the packet to the engine so it can forward when we're a middle hop.
|
||||
|
||||
### 2. Engine: do not forward when we're final hop (optional but recommended)
|
||||
|
||||
In `direct_forward`, after stripping our hash from the path, if the path is empty (hop_count was 1), we're the final destination and should not forward. Today we return the packet and would schedule a transmit with an empty path. Add:
|
||||
|
||||
- After `packet.path = bytearray(packet.path[hash_size:])` and updating `path_len`, if `hop_count - 1 == 0` (or `len(packet.path) == 0`), set e.g. `packet.drop_reason = "Direct: final hop (deliver only)"` and return `None` so we don't transmit.
|
||||
|
||||
That avoids transmitting when we're the final hop and only delivering to companions.
|
||||
|
||||
## Recording for UI
|
||||
|
||||
When we pass these packets to the engine, the engine will record them (forwarded or dropped with reason). We can keep calling `_record_for_ui` before passing to the engine for consistency, or rely on the engine's recording; if we do both we might double-record. Prefer: only the engine records when we pass to it (no `_record_for_ui` for these when we're also passing to engine), or keep a single record in the router and don't pass to engine for recording (pass only for forwarding). Simplest is: pass to engine, let engine do the only record (it already builds packet_record for every packet it sees). So we may remove `_record_for_ui` for these four types when we add the "pass to engine" path, to avoid duplicate entries. Alternatively we could keep _record_for_ui and have the engine skip recording when it's a type that the router already recorded—more complex. Easiest: pass to engine, remove the router's _record_for_ui for these four so the engine is the single place that records them.
|
||||
+127
-112
@@ -151,6 +151,9 @@ class RepeaterHandler(BaseHandler):
|
||||
transmitted = False
|
||||
tx_delay_ms = 0.0
|
||||
drop_reason = None
|
||||
lbt_attempts = 0
|
||||
lbt_backoff_delays_ms = None
|
||||
lbt_channel_busy = False
|
||||
|
||||
original_path_hashes = packet.get_path_hashes_hex()
|
||||
path_hash_size = packet.get_path_hash_size()
|
||||
@@ -292,70 +295,33 @@ class RepeaterHandler(BaseHandler):
|
||||
if is_dupe and drop_reason is None:
|
||||
drop_reason = "Duplicate"
|
||||
|
||||
path_hash = None
|
||||
display_hashes = (
|
||||
original_path_hashes if original_path_hashes else packet.get_path_hashes_hex()
|
||||
)
|
||||
if display_hashes:
|
||||
display = display_hashes[:8]
|
||||
if len(display_hashes) > 8:
|
||||
display = list(display) + ["..."]
|
||||
path_hash = "[" + ", ".join(display) + "]"
|
||||
|
||||
src_hash = None
|
||||
dst_hash = None
|
||||
|
||||
# Payload types with dest_hash and src_hash as first 2 bytes
|
||||
if payload_type in [0x00, 0x01, 0x02, 0x08]:
|
||||
if hasattr(packet, "payload") and packet.payload and len(packet.payload) >= 2:
|
||||
dst_hash = f"{packet.payload[0]:02X}"
|
||||
src_hash = f"{packet.payload[1]:02X}"
|
||||
|
||||
# ADVERT packets have source identifier as first byte
|
||||
elif payload_type == PAYLOAD_TYPE_ADVERT:
|
||||
if hasattr(packet, "payload") and packet.payload and len(packet.payload) >= 1:
|
||||
src_hash = f"{packet.payload[0]:02X}"
|
||||
path_hash = self._path_hash_display(display_hashes)
|
||||
src_hash, dst_hash = self._packet_record_src_dst(packet, payload_type)
|
||||
|
||||
# Record packet for charts
|
||||
packet_record = {
|
||||
"timestamp": time.time(),
|
||||
"header": (
|
||||
f"0x{packet.header:02X}"
|
||||
if hasattr(packet, "header") and packet.header is not None
|
||||
else None
|
||||
),
|
||||
"payload": (
|
||||
packet.payload.hex() if hasattr(packet, "payload") and packet.payload else None
|
||||
),
|
||||
"payload_length": (
|
||||
len(packet.payload) if hasattr(packet, "payload") and packet.payload else 0
|
||||
),
|
||||
"type": payload_type,
|
||||
"route": route_type,
|
||||
"length": len(packet.payload or b""),
|
||||
"rssi": rssi,
|
||||
"snr": snr,
|
||||
"score": self.calculate_packet_score(
|
||||
snr, len(packet.payload or b""), self.radio_config["spreading_factor"]
|
||||
),
|
||||
"tx_delay_ms": tx_delay_ms,
|
||||
"transmitted": transmitted,
|
||||
"is_duplicate": is_dupe,
|
||||
"packet_hash": pkt_hash[:16],
|
||||
"drop_reason": drop_reason,
|
||||
"path_hash": path_hash,
|
||||
"src_hash": src_hash,
|
||||
"dst_hash": dst_hash,
|
||||
"original_path": original_path_hashes or None,
|
||||
"forwarded_path": forwarded_path_hashes,
|
||||
"path_hash_size": path_hash_size,
|
||||
"raw_packet": packet.write_to().hex() if hasattr(packet, "write_to") else None,
|
||||
"lbt_attempts": lbt_attempts if transmitted else 0,
|
||||
"lbt_backoff_delays_ms": (
|
||||
lbt_backoff_delays_ms if transmitted and lbt_backoff_delays_ms else None
|
||||
),
|
||||
"lbt_channel_busy": lbt_channel_busy if transmitted else False,
|
||||
}
|
||||
packet_record = self._build_packet_record(
|
||||
packet,
|
||||
payload_type,
|
||||
route_type,
|
||||
rssi,
|
||||
snr,
|
||||
original_path_hashes,
|
||||
path_hash_size,
|
||||
path_hash,
|
||||
src_hash,
|
||||
dst_hash,
|
||||
transmitted=transmitted,
|
||||
drop_reason=drop_reason,
|
||||
is_duplicate=is_dupe,
|
||||
forwarded_path=forwarded_path_hashes,
|
||||
tx_delay_ms=tx_delay_ms,
|
||||
lbt_attempts=lbt_attempts,
|
||||
lbt_backoff_delays_ms=lbt_backoff_delays_ms,
|
||||
lbt_channel_busy=lbt_channel_busy,
|
||||
)
|
||||
|
||||
# Store packet record to persistent storage
|
||||
# Skip LetsMesh only for invalid packets (not duplicates or operational drops)
|
||||
@@ -426,61 +392,22 @@ class RepeaterHandler(BaseHandler):
|
||||
header_info = PacketHeaderUtils.parse_header(packet.header)
|
||||
payload_type = header_info["payload_type"]
|
||||
route_type = header_info["route_type"]
|
||||
pkt_hash = packet.calculate_packet_hash().hex().upper()
|
||||
original_path_hashes = packet.get_path_hashes_hex()
|
||||
path_hash_size = packet.get_path_hash_size()
|
||||
display_hashes = original_path_hashes
|
||||
path_hash = None
|
||||
if display_hashes:
|
||||
display = display_hashes[:8]
|
||||
if len(display_hashes) > 8:
|
||||
display = list(display) + ["..."]
|
||||
path_hash = "[" + ", ".join(display) + "]"
|
||||
src_hash = None
|
||||
dst_hash = None
|
||||
if payload_type in [0x00, 0x01, 0x02, 0x08]:
|
||||
if hasattr(packet, "payload") and packet.payload and len(packet.payload) >= 2:
|
||||
dst_hash = f"{packet.payload[0]:02X}"
|
||||
src_hash = f"{packet.payload[1]:02X}"
|
||||
elif payload_type == PAYLOAD_TYPE_ADVERT:
|
||||
if hasattr(packet, "payload") and packet.payload and len(packet.payload) >= 1:
|
||||
src_hash = f"{packet.payload[0]:02X}"
|
||||
elif payload_type == PAYLOAD_TYPE_ANON_REQ:
|
||||
if hasattr(packet, "payload") and packet.payload and len(packet.payload) >= 1:
|
||||
dst_hash = f"{packet.payload[0]:02X}"
|
||||
packet_record = {
|
||||
"timestamp": time.time(),
|
||||
"header": f"0x{packet.header:02X}",
|
||||
"payload": (
|
||||
packet.payload.hex() if hasattr(packet, "payload") and packet.payload else None
|
||||
),
|
||||
"payload_length": (
|
||||
len(packet.payload) if hasattr(packet, "payload") and packet.payload else 0
|
||||
),
|
||||
"type": payload_type,
|
||||
"route": route_type,
|
||||
"length": len(packet.payload or b""),
|
||||
"rssi": rssi,
|
||||
"snr": snr,
|
||||
"score": self.calculate_packet_score(
|
||||
snr, len(packet.payload or b""), self.radio_config["spreading_factor"]
|
||||
),
|
||||
"tx_delay_ms": 0.0,
|
||||
"transmitted": False,
|
||||
"is_duplicate": False,
|
||||
"packet_hash": pkt_hash[:16],
|
||||
"drop_reason": None,
|
||||
"path_hash": path_hash,
|
||||
"src_hash": src_hash,
|
||||
"dst_hash": dst_hash,
|
||||
"original_path": original_path_hashes or None,
|
||||
"forwarded_path": None,
|
||||
"path_hash_size": path_hash_size,
|
||||
"raw_packet": packet.write_to().hex() if hasattr(packet, "write_to") else None,
|
||||
"lbt_attempts": 0,
|
||||
"lbt_backoff_delays_ms": None,
|
||||
"lbt_channel_busy": False,
|
||||
}
|
||||
path_hash = self._path_hash_display(original_path_hashes)
|
||||
src_hash, dst_hash = self._packet_record_src_dst(packet, payload_type)
|
||||
packet_record = self._build_packet_record(
|
||||
packet,
|
||||
payload_type,
|
||||
route_type,
|
||||
rssi,
|
||||
snr,
|
||||
original_path_hashes,
|
||||
path_hash_size,
|
||||
path_hash,
|
||||
src_hash,
|
||||
dst_hash,
|
||||
)
|
||||
try:
|
||||
self.storage.record_packet(packet_record, skip_letsmesh_if_invalid=False)
|
||||
except Exception as e:
|
||||
@@ -498,6 +425,94 @@ class RepeaterHandler(BaseHandler):
|
||||
for k in expired:
|
||||
del self.seen_packets[k]
|
||||
|
||||
def _path_hash_display(self, display_hashes) -> Optional[str]:
|
||||
"""Build path hash string for packet record from path hashes list."""
|
||||
if not display_hashes:
|
||||
return None
|
||||
display = display_hashes[:8]
|
||||
if len(display_hashes) > 8:
|
||||
display = list(display) + ["..."]
|
||||
return "[" + ", ".join(display) + "]"
|
||||
|
||||
def _packet_record_src_dst(
|
||||
self, packet: Packet, payload_type: int
|
||||
) -> Tuple[Optional[str], Optional[str]]:
|
||||
"""Return (src_hash, dst_hash) for packet_record from packet and payload_type."""
|
||||
src_hash = None
|
||||
dst_hash = None
|
||||
payload = getattr(packet, "payload", None)
|
||||
if payload_type in [0x00, 0x01, 0x02, 0x08]:
|
||||
if payload and len(payload) >= 2:
|
||||
dst_hash = f"{payload[0]:02X}"
|
||||
src_hash = f"{payload[1]:02X}"
|
||||
elif payload_type == PAYLOAD_TYPE_ADVERT:
|
||||
if payload and len(payload) >= 1:
|
||||
src_hash = f"{payload[0]:02X}"
|
||||
elif payload_type == PAYLOAD_TYPE_ANON_REQ:
|
||||
if payload and len(payload) >= 1:
|
||||
dst_hash = f"{payload[0]:02X}"
|
||||
return (src_hash, dst_hash)
|
||||
|
||||
def _build_packet_record(
|
||||
self,
|
||||
packet: Packet,
|
||||
payload_type: int,
|
||||
route_type: int,
|
||||
rssi: int,
|
||||
snr: float,
|
||||
original_path_hashes,
|
||||
path_hash_size: int,
|
||||
path_hash: Optional[str],
|
||||
src_hash: Optional[str],
|
||||
dst_hash: Optional[str],
|
||||
*,
|
||||
transmitted: bool = False,
|
||||
drop_reason: Optional[str] = None,
|
||||
is_duplicate: bool = False,
|
||||
forwarded_path=None,
|
||||
tx_delay_ms: float = 0.0,
|
||||
lbt_attempts: int = 0,
|
||||
lbt_backoff_delays_ms=None,
|
||||
lbt_channel_busy: bool = False,
|
||||
) -> dict:
|
||||
"""Build a single packet_record dict for storage and recent_packets."""
|
||||
pkt_hash = packet.calculate_packet_hash().hex().upper()
|
||||
payload = getattr(packet, "payload", None)
|
||||
payload_len = len(payload or b"")
|
||||
return {
|
||||
"timestamp": time.time(),
|
||||
"header": (
|
||||
f"0x{packet.header:02X}"
|
||||
if hasattr(packet, "header") and packet.header is not None
|
||||
else None
|
||||
),
|
||||
"payload": payload.hex() if payload else None,
|
||||
"payload_length": len(payload) if payload else 0,
|
||||
"type": payload_type,
|
||||
"route": route_type,
|
||||
"length": payload_len,
|
||||
"rssi": rssi,
|
||||
"snr": snr,
|
||||
"score": self.calculate_packet_score(
|
||||
snr, payload_len, self.radio_config["spreading_factor"]
|
||||
),
|
||||
"tx_delay_ms": tx_delay_ms,
|
||||
"transmitted": transmitted,
|
||||
"is_duplicate": is_duplicate,
|
||||
"packet_hash": pkt_hash[:16],
|
||||
"drop_reason": drop_reason,
|
||||
"path_hash": path_hash,
|
||||
"src_hash": src_hash,
|
||||
"dst_hash": dst_hash,
|
||||
"original_path": original_path_hashes or None,
|
||||
"forwarded_path": forwarded_path,
|
||||
"path_hash_size": path_hash_size,
|
||||
"raw_packet": packet.write_to().hex() if hasattr(packet, "write_to") else None,
|
||||
"lbt_attempts": lbt_attempts,
|
||||
"lbt_backoff_delays_ms": lbt_backoff_delays_ms,
|
||||
"lbt_channel_busy": lbt_channel_busy,
|
||||
}
|
||||
|
||||
def _get_drop_reason(self, packet: Packet) -> str:
|
||||
|
||||
if self.is_duplicate(packet):
|
||||
|
||||
@@ -182,7 +182,8 @@ class PacketRouter:
|
||||
|
||||
elif payload_type == LoginServerHandler.payload_type():
|
||||
# Route to companion if dest is a companion; else to login_helper (for logging into this repeater).
|
||||
# If dest is remote (no local handler), mark processed so we don't pass our own outbound login TX to the repeater as RX.
|
||||
# When dest is remote (not handled), pass to engine so DIRECT/FLOOD ANON_REQ can be forwarded.
|
||||
# Our own injected ANON_REQ is suppressed by the engine's duplicate (mark_seen) check.
|
||||
dest_hash = packet.payload[0] if packet.payload else None
|
||||
companion_bridges = getattr(self.daemon, "companion_bridges", {})
|
||||
if dest_hash is not None and dest_hash in companion_bridges:
|
||||
@@ -192,9 +193,6 @@ class PacketRouter:
|
||||
handled = await self.daemon.login_helper.process_login_packet(packet)
|
||||
if handled:
|
||||
processed_by_injection = True
|
||||
else:
|
||||
# Login request for remote repeater (we already TXed it via inject); don't treat as RX.
|
||||
processed_by_injection = True
|
||||
if processed_by_injection:
|
||||
self._record_for_ui(packet, metadata)
|
||||
|
||||
@@ -220,6 +218,7 @@ class PacketRouter:
|
||||
handled = await self.daemon.text_helper.process_text_packet(packet)
|
||||
if handled:
|
||||
processed_by_injection = True
|
||||
self._record_for_ui(packet, metadata)
|
||||
|
||||
elif payload_type == PathHandler.payload_type():
|
||||
dest_hash = packet.payload[0] if packet.payload else None
|
||||
|
||||
Reference in New Issue
Block a user