diff --git a/repeater/data_acquisition/letsmesh_handler.py b/repeater/data_acquisition/letsmesh_handler.py index e1b74f7..44a45f4 100644 --- a/repeater/data_acquisition/letsmesh_handler.py +++ b/repeater/data_acquisition/letsmesh_handler.py @@ -198,7 +198,7 @@ class _BrokerConnection: # Stop the loop if it's still running (websocket mode requires clean restart) try: self.client.loop_stop() - except: + except Exception: pass self._set_jwt_credentials() diff --git a/repeater/data_acquisition/sqlite_handler.py b/repeater/data_acquisition/sqlite_handler.py index 7b9e64c..22f94a2 100644 --- a/repeater/data_acquisition/sqlite_handler.py +++ b/repeater/data_acquisition/sqlite_handler.py @@ -14,6 +14,11 @@ class SQLiteHandler: def __init__(self, storage_dir: Path): self.storage_dir = storage_dir self.sqlite_path = self.storage_dir / "repeater.db" + self._api_token_last_used_updates = {} + self._api_token_last_used_interval_sec = 300 + self._hot_cache_ttl_sec = 60 + self._packet_stats_cache = {} + self._neighbors_cache = {"timestamp": 0.0, "value": None} self._init_database() self._run_migrations() @@ -24,6 +29,10 @@ class SQLiteHandler: conn.execute("PRAGMA busy_timeout=30000") return conn + def _invalidate_hot_caches(self) -> None: + self._packet_stats_cache.clear() + self._neighbors_cache = {"timestamp": 0.0, "value": None} + def _init_database(self): try: with self._connect() as conn: @@ -494,19 +503,23 @@ class SQLiteHandler: try: with self._connect() as conn: cursor = conn.execute( - "SELECT id, name, created_at FROM api_tokens WHERE token_hash = ?", + "SELECT id, name, created_at, last_used FROM api_tokens WHERE token_hash = ?", (token_hash,), ) row = cursor.fetchone() if row: - token_id, name, created_at = row + token_id, name, created_at, _last_used = row + now = time.time() - # Update last_used timestamp - conn.execute( - "UPDATE api_tokens SET last_used = ? WHERE id = ?", (time.time(), token_id) - ) - conn.commit() + # Throttle last_used updates to reduce write-lock contention. + last_update = self._api_token_last_used_updates.get(token_id, 0.0) + if now - last_update >= self._api_token_last_used_interval_sec: + conn.execute( + "UPDATE api_tokens SET last_used = ? WHERE id = ?", (now, token_id) + ) + conn.commit() + self._api_token_last_used_updates[token_id] = now return {"id": token_id, "name": name, "created_at": created_at} return None @@ -598,6 +611,7 @@ class SQLiteHandler: int(bool(record.get("lbt_channel_busy", False))), ), ) + self._invalidate_hot_caches() except Exception as e: logger.error(f"Failed to store packet in SQLite: {e}") @@ -687,6 +701,8 @@ class SQLiteHandler: ), ) + self._invalidate_hot_caches() + except Exception as e: logger.error(f"Failed to store advert in SQLite: {e}") @@ -754,7 +770,12 @@ class SQLiteHandler: def get_packet_stats(self, hours: int = 24) -> dict: try: - cutoff = time.time() - (hours * 3600) + now = time.time() + cached = self._packet_stats_cache.get(hours) + if cached and (now - cached["timestamp"]) < self._hot_cache_ttl_sec: + return cached["value"] + + cutoff = now - (hours * 3600) with self._connect() as conn: conn.row_factory = sqlite3.Row @@ -798,7 +819,7 @@ class SQLiteHandler: (cutoff,), ).fetchall() - return { + result = { "total_packets": stats["total_packets"], "transmitted_packets": stats["transmitted_packets"], "dropped_packets": stats["dropped_packets"], @@ -814,6 +835,9 @@ class SQLiteHandler: ], } + self._packet_stats_cache[hours] = {"timestamp": now, "value": result} + return result + except Exception as e: logger.error(f"Failed to get packet stats: {e}") return {} @@ -1009,20 +1033,27 @@ class SQLiteHandler: with self._connect() as conn: conn.row_factory = sqlite3.Row + type_rows = conn.execute( + """ + SELECT type, COUNT(*) as count + FROM packets + WHERE timestamp > ? + GROUP BY type + """, + (cutoff,), + ).fetchall() + type_counts = {} - for packet_type in range(16): - count = conn.execute( - "SELECT COUNT(*) FROM packets WHERE type = ? AND timestamp > ?", - (packet_type, cutoff), - ).fetchone()[0] - - type_name = packet_type_names.get(packet_type, f"Type {packet_type}") - if count > 0: + other_count = 0 + for row in type_rows: + pkt_type = int(row["type"]) + count = int(row["count"]) + if pkt_type <= 15: + type_name = packet_type_names.get(pkt_type, f"Type {pkt_type}") type_counts[type_name] = count + else: + other_count += count - other_count = conn.execute( - "SELECT COUNT(*) FROM packets WHERE type > 15 AND timestamp > ?", (cutoff,) - ).fetchone()[0] if other_count > 0: type_counts["Other Types (>15)"] = other_count @@ -1046,23 +1077,29 @@ class SQLiteHandler: with self._connect() as conn: conn.row_factory = sqlite3.Row + route_rows = conn.execute( + """ + SELECT route, COUNT(*) as count + FROM packets + WHERE timestamp > ? + GROUP BY route + """, + (cutoff,), + ).fetchall() + route_counts = {} route_names = {0: "Transport Flood", 1: "Flood", 2: "Direct", 3: "Transport Direct"} + other_count = 0 - for route_type in range(4): - count = conn.execute( - "SELECT COUNT(*) FROM packets WHERE route = ? AND timestamp > ?", - (route_type, cutoff), - ).fetchone()[0] - - route_name = route_names.get(route_type, f"Route {route_type}") - if count > 0: + for row in route_rows: + route_type = int(row["route"]) + count = int(row["count"]) + if route_type <= 3: + route_name = route_names.get(route_type, f"Route {route_type}") route_counts[route_name] = count + else: + other_count += count - # Count any other route types > 3 - other_count = conn.execute( - "SELECT COUNT(*) FROM packets WHERE route > 3 AND timestamp > ?", (cutoff,) - ).fetchone()[0] if other_count > 0: route_counts["Other Routes (>3)"] = other_count @@ -1080,6 +1117,12 @@ class SQLiteHandler: def get_neighbors(self) -> dict: try: + now = time.time() + cached = self._neighbors_cache.get("value") + cached_ts = float(self._neighbors_cache.get("timestamp", 0.0)) + if cached is not None and (now - cached_ts) < self._hot_cache_ttl_sec: + return cached + with self._connect() as conn: conn.row_factory = sqlite3.Row @@ -1087,12 +1130,14 @@ class SQLiteHandler: """ SELECT pubkey, node_name, is_repeater, route_type, contact_type, latitude, longitude, first_seen, last_seen, rssi, snr, advert_count, zero_hop - FROM adverts a1 - WHERE last_seen = ( - SELECT MAX(last_seen) - FROM adverts a2 - WHERE a2.pubkey = a1.pubkey - ) + FROM ( + SELECT + pubkey, node_name, is_repeater, route_type, contact_type, + latitude, longitude, first_seen, last_seen, rssi, snr, advert_count, zero_hop, + ROW_NUMBER() OVER (PARTITION BY pubkey ORDER BY last_seen DESC) AS rn + FROM adverts + ) latest + WHERE rn = 1 ORDER BY last_seen DESC """ ).fetchall() @@ -1114,6 +1159,7 @@ class SQLiteHandler: "zero_hop": bool(row["zero_hop"]), } + self._neighbors_cache = {"timestamp": now, "value": result} return result except Exception as e: @@ -1320,30 +1366,36 @@ class SQLiteHandler: def get_cumulative_counts(self) -> dict: try: with self._connect() as conn: - type_counts = {} - for i in range(16): - count = conn.execute( - "SELECT COUNT(*) FROM packets WHERE type = ?", (i,) - ).fetchone()[0] - type_counts[f"type_{i}"] = count + conn.row_factory = sqlite3.Row - other_count = conn.execute( - "SELECT COUNT(*) FROM packets WHERE type > 15" - ).fetchone()[0] - type_counts["type_other"] = other_count + type_rows = conn.execute( + "SELECT type, COUNT(*) as count FROM packets GROUP BY type" + ).fetchall() - rx_total = conn.execute("SELECT COUNT(*) FROM packets").fetchone()[0] - tx_total = conn.execute( - "SELECT COUNT(*) FROM packets WHERE transmitted = 1" - ).fetchone()[0] - drop_total = conn.execute( - "SELECT COUNT(*) FROM packets WHERE transmitted = 0" - ).fetchone()[0] + type_counts = {f"type_{i}": 0 for i in range(16)} + type_counts["type_other"] = 0 + for row in type_rows: + pkt_type = int(row["type"]) + count = int(row["count"]) + if pkt_type <= 15: + type_counts[f"type_{pkt_type}"] = count + else: + type_counts["type_other"] += count + + totals = conn.execute( + """ + SELECT + COUNT(*) AS rx_total, + SUM(CASE WHEN transmitted = 1 THEN 1 ELSE 0 END) AS tx_total, + SUM(CASE WHEN transmitted = 0 THEN 1 ELSE 0 END) AS drop_total + FROM packets + """ + ).fetchone() return { - "rx_total": rx_total, - "tx_total": tx_total, - "drop_total": drop_total, + "rx_total": int(totals["rx_total"] or 0), + "tx_total": int(totals["tx_total"] or 0), + "drop_total": int(totals["drop_total"] or 0), "type_counts": type_counts, } diff --git a/repeater/engine.py b/repeater/engine.py index 533c63f..ad8711a 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -3,7 +3,7 @@ import copy import logging import struct import time -from collections import OrderedDict +from collections import OrderedDict, deque from typing import Optional, Tuple from pymc_core.node.handlers.base import BaseHandler @@ -99,8 +99,9 @@ class RepeaterHandler(BaseHandler): self.rx_count = 0 self.forwarded_count = 0 self.dropped_count = 0 - self.recent_packets = [] self.max_recent_packets = 50 + self.recent_packets = deque(maxlen=self.max_recent_packets) + self._recent_hash_index = {} self.start_time = time.time() # Flood/direct and duplicate counters (for GET_STATUS / firmware RepeaterStats) self.recv_flood_count = 0 @@ -126,6 +127,7 @@ class RepeaterHandler(BaseHandler): self.last_db_cleanup = time.time() self.noise_floor_interval = NOISE_FLOOR_INTERVAL # 30 seconds self._background_task = None + self._cached_noise_floor = None self._last_crc_error_count = 0 # Track radio counter for delta persistence # Cache transport keys for efficient lookup @@ -157,6 +159,7 @@ class RepeaterHandler(BaseHandler): pass route_type = packet.header & PH_ROUTE_MASK + pkt_hash_full = packet.calculate_packet_hash().hex().upper() # TX mode: forward (repeat on), monitor (no repeat, tenants can TX), no_tx (all TX off) mode = self.config.get("repeater", {}).get("mode", "forward") @@ -197,7 +200,7 @@ class RepeaterHandler(BaseHandler): # For local transmissions, create a direct transmission result (if local TX allowed) if local_transmission and allow_local_tx: # Mark local packet as seen to prevent duplicate processing when received back - self.mark_seen(packet) + self.mark_seen(packet, packet_hash=pkt_hash_full) # Calculate transmission delay for local packets delay = self._calculate_tx_delay(packet, snr) result = (packet, delay) @@ -318,8 +321,7 @@ class RepeaterHandler(BaseHandler): ) # Check if this is a duplicate - pkt_hash = packet.calculate_packet_hash().hex().upper() - is_dupe = pkt_hash in self.seen_packets and not transmitted + is_dupe = pkt_hash_full in self.seen_packets and not transmitted # Set drop reason for duplicates and count flood vs direct dups if is_dupe and drop_reason is None: @@ -356,6 +358,7 @@ class RepeaterHandler(BaseHandler): lbt_attempts=lbt_attempts, lbt_backoff_delays_ms=lbt_backoff_delays_ms, lbt_channel_busy=lbt_channel_busy, + packet_hash=pkt_hash_full, ) # Store packet record to persistent storage @@ -371,30 +374,24 @@ class RepeaterHandler(BaseHandler): # If this is a duplicate, try to attach it to the original packet if is_dupe and len(self.recent_packets) > 0: - # Find the original packet with same hash - for idx in range(len(self.recent_packets) - 1, -1, -1): - prev_pkt = self.recent_packets[idx] - if prev_pkt.get("packet_hash") == packet_record["packet_hash"]: - # Add duplicate to original packet's duplicate list - if "duplicates" not in prev_pkt: - prev_pkt["duplicates"] = [] - if len(prev_pkt["duplicates"]) < self.max_duplicates_per_packet: - prev_pkt["duplicates"].append(packet_record) - # Don't add duplicate to main list, just track in original - break + prev_pkt = self._recent_hash_index.get(packet_record["packet_hash"]) + if prev_pkt is not None: + # Add duplicate to original packet's duplicate list + if "duplicates" not in prev_pkt: + prev_pkt["duplicates"] = [] + if len(prev_pkt["duplicates"]) < self.max_duplicates_per_packet: + prev_pkt["duplicates"].append(packet_record) + # Don't add duplicate to main list, just track in original else: # Original not found, add as regular packet - self.recent_packets.append(packet_record) + self._append_recent_packet(packet_record) else: # Not a duplicate or first occurrence - self.recent_packets.append(packet_record) - - if len(self.recent_packets) > self.max_recent_packets: - self.recent_packets.pop(0) + self._append_recent_packet(packet_record) def log_trace_record(self, packet_record: dict) -> None: """Manually log a packet trace record (used by external callers)""" - self.recent_packets.append(packet_record) + self._append_recent_packet(packet_record) self.rx_count += 1 if packet_record.get("transmitted", False): @@ -409,9 +406,6 @@ class RepeaterHandler(BaseHandler): except Exception as e: logger.error(f"Failed to store packet record: {e}") - if len(self.recent_packets) > self.max_recent_packets: - self.recent_packets.pop(0) - def record_packet_only(self, packet: Packet, metadata: dict) -> None: """Record a packet for UI/storage without running forwarding or duplicate logic. @@ -448,15 +442,14 @@ class RepeaterHandler(BaseHandler): path_hash, src_hash, dst_hash, + packet_hash=packet.calculate_packet_hash().hex().upper(), ) try: self.storage.record_packet(packet_record, skip_letsmesh_if_invalid=False) except Exception as e: logger.error(f"Failed to store packet record (record_packet_only): {e}") return - self.recent_packets.append(packet_record) - if len(self.recent_packets) > self.max_recent_packets: - self.recent_packets.pop(0) + self._append_recent_packet(packet_record) 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. @@ -489,6 +482,7 @@ class RepeaterHandler(BaseHandler): transmitted=False, drop_reason="Duplicate", is_duplicate=True, + packet_hash=packet.calculate_packet_hash().hex().upper(), ) if self.storage: @@ -499,20 +493,15 @@ class RepeaterHandler(BaseHandler): # Group under original in recent_packets if len(self.recent_packets) > 0: - for idx in range(len(self.recent_packets) - 1, -1, -1): - prev_pkt = self.recent_packets[idx] - if prev_pkt.get("packet_hash") == packet_record["packet_hash"]: - if "duplicates" not in prev_pkt: - prev_pkt["duplicates"] = [] - prev_pkt["duplicates"].append(packet_record) - break + prev_pkt = self._recent_hash_index.get(packet_record["packet_hash"]) + if prev_pkt is not None: + if "duplicates" not in prev_pkt: + prev_pkt["duplicates"] = [] + prev_pkt["duplicates"].append(packet_record) else: - self.recent_packets.append(packet_record) + self._append_recent_packet(packet_record) else: - self.recent_packets.append(packet_record) - - if len(self.recent_packets) > self.max_recent_packets: - self.recent_packets.pop(0) + self._append_recent_packet(packet_record) def cleanup_cache(self): @@ -570,9 +559,10 @@ class RepeaterHandler(BaseHandler): lbt_attempts: int = 0, lbt_backoff_delays_ms=None, lbt_channel_busy: bool = False, + packet_hash: Optional[str] = None, ) -> dict: """Build a single packet_record dict for storage and recent_packets.""" - pkt_hash = packet.calculate_packet_hash().hex().upper() + pkt_hash = packet_hash or packet.calculate_packet_hash().hex().upper() payload = getattr(packet, "payload", None) payload_len = len(payload or b"") return { @@ -609,6 +599,19 @@ class RepeaterHandler(BaseHandler): "lbt_channel_busy": lbt_channel_busy, } + def _append_recent_packet(self, packet_record: dict) -> None: + """Append packet to bounded recent list and keep hash index aligned.""" + if len(self.recent_packets) >= self.max_recent_packets: + oldest = self.recent_packets.popleft() + oldest_hash = oldest.get("packet_hash") if isinstance(oldest, dict) else None + if oldest_hash and self._recent_hash_index.get(oldest_hash) is oldest: + del self._recent_hash_index[oldest_hash] + + self.recent_packets.append(packet_record) + pkt_hash = packet_record.get("packet_hash") if isinstance(packet_record, dict) else None + if pkt_hash: + self._recent_hash_index[pkt_hash] = packet_record + def _get_drop_reason(self, packet: Packet) -> str: if self.is_duplicate(packet): @@ -646,9 +649,9 @@ class RepeaterHandler(BaseHandler): return True return False - def mark_seen(self, packet: Packet): + def mark_seen(self, packet: Packet, packet_hash: Optional[str] = None): - pkt_hash = packet.calculate_packet_hash().hex().upper() + pkt_hash = packet_hash or packet.calculate_packet_hash().hex().upper() self.seen_packets[pkt_hash] = time.time() if len(self.seen_packets) > self.max_cache_size: @@ -1047,6 +1050,10 @@ class RepeaterHandler(BaseHandler): logger.debug(f"Failed to get noise floor: {e}") return None + def get_cached_noise_floor(self) -> Optional[float]: + """Return the last asynchronously-sampled noise floor value.""" + return self._cached_noise_floor + def get_stats(self) -> dict: uptime_seconds = time.time() - self.start_time @@ -1065,8 +1072,8 @@ class RepeaterHandler(BaseHandler): rx_per_hour = len(packets_last_hour) forwarded_per_hour = sum(1 for p in packets_last_hour if p.get("transmitted", False)) - # Get current noise floor from radio - noise_floor_dbm = self.get_noise_floor() + # Use cached value sampled by the background timer to avoid serial I/O on stats requests. + noise_floor_dbm = self.get_cached_noise_floor() # Get CRC error count from radio hardware radio = self.dispatcher.radio if self.dispatcher else None @@ -1097,7 +1104,7 @@ class RepeaterHandler(BaseHandler): "direct_dup_count": self.direct_dup_count, "rx_per_hour": rx_per_hour, "forwarded_per_hour": forwarded_per_hour, - "recent_packets": self.recent_packets, + "recent_packets": list(self.recent_packets), "neighbors": neighbors, "uptime_seconds": uptime_seconds, "noise_floor_dbm": noise_floor_dbm, @@ -1212,6 +1219,7 @@ class RepeaterHandler(BaseHandler): loop = asyncio.get_running_loop() noise_floor = await loop.run_in_executor(None, self.get_noise_floor) if noise_floor is not None: + self._cached_noise_floor = noise_floor self.storage.record_noise_floor(noise_floor) logger.debug(f"Recorded noise floor: {noise_floor} dBm") else: diff --git a/repeater/handler_helpers/discovery.py b/repeater/handler_helpers/discovery.py index 4836161..8153747 100644 --- a/repeater/handler_helpers/discovery.py +++ b/repeater/handler_helpers/discovery.py @@ -44,11 +44,26 @@ class DiscoveryHelper: log_fn=log_fn or logger.info, debug_log_fn=debug_log_fn, ) + self._pending_tasks = set() # Set up the request callback self.control_handler.set_request_callback(self._on_discovery_request) logger.debug("Discovery handler initialized") + def _track_task(self, task: asyncio.Task) -> None: + self._pending_tasks.add(task) + + def _on_done(done_task: asyncio.Task) -> None: + self._pending_tasks.discard(done_task) + try: + done_task.result() + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Background discovery task failed: {e}", exc_info=True) + + task.add_done_callback(_on_done) + def _on_discovery_request(self, request_data: dict) -> None: """ Handle incoming discovery request. @@ -115,7 +130,8 @@ class DiscoveryHelper: # Send response via router injection if self.packet_injector: - asyncio.create_task(self._send_packet_async(response_packet, tag)) + task = asyncio.create_task(self._send_packet_async(response_packet, tag)) + self._track_task(task) else: logger.warning("No packet injector available - discovery response not sent") diff --git a/repeater/handler_helpers/login.py b/repeater/handler_helpers/login.py index 3e9924b..1561dec 100644 --- a/repeater/handler_helpers/login.py +++ b/repeater/handler_helpers/login.py @@ -22,6 +22,21 @@ class LoginHelper: self.handlers = {} self.acls = {} # Per-identity ACLs keyed by hash_byte + self._pending_tasks = set() + + def _track_task(self, task: asyncio.Task) -> None: + self._pending_tasks.add(task) + + def _on_done(done_task: asyncio.Task) -> None: + self._pending_tasks.discard(done_task) + try: + done_task.result() + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Background login task failed: {e}", exc_info=True) + + task.add_done_callback(_on_done) def register_identity( self, name: str, identity, identity_type: str = "room_server", config: dict = None @@ -141,7 +156,8 @@ class LoginHelper: def _send_packet_with_delay(self, packet, delay_ms: int): if self.packet_injector: - asyncio.create_task(self._delayed_send(packet, delay_ms)) + task = asyncio.create_task(self._delayed_send(packet, delay_ms)) + self._track_task(task) else: logger.error("No packet injector configured, cannot send login response") diff --git a/repeater/handler_helpers/text.py b/repeater/handler_helpers/text.py index 9d6f59d..4d9c432 100644 --- a/repeater/handler_helpers/text.py +++ b/repeater/handler_helpers/text.py @@ -65,6 +65,21 @@ class TextHelper: # Initialize CLI handler later when repeater identity is registered self.cli = None + self._pending_tasks = set() + + def _track_task(self, task: asyncio.Task) -> None: + self._pending_tasks.add(task) + + def _on_done(done_task: asyncio.Task) -> None: + self._pending_tasks.discard(done_task) + try: + done_task.result() + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Background text task failed: {e}", exc_info=True) + + task.add_done_callback(_on_done) def register_identity( self, name: str, identity, identity_type: str = "room_server", radio_config=None @@ -152,7 +167,8 @@ class TextHelper: self.room_servers[hash_byte] = room_server # Start sync loop - asyncio.create_task(room_server.start()) + start_task = asyncio.create_task(room_server.start()) + self._track_task(start_task) logger.info( f"Registered room server '{name}': hash=0x{hash_byte:02X}, " diff --git a/repeater/main.py b/repeater/main.py index d74d6d3..bb8b4c8 100644 --- a/repeater/main.py +++ b/repeater/main.py @@ -940,7 +940,7 @@ class RepeaterDaemon: "queue_len": min(255, queue_len), } if stats_type == STATS_TYPE_RADIO: - noise_floor = int(engine.get_noise_floor() or 0) + noise_floor = int(engine.get_cached_noise_floor() or 0) radio = getattr(self, "dispatcher", None) and getattr(self.dispatcher, "radio", None) if radio: _r = getattr(radio, "get_last_rssi", lambda: 0) diff --git a/repeater/packet_router.py b/repeater/packet_router.py index 187eeef..6700f5c 100644 --- a/repeater/packet_router.py +++ b/repeater/packet_router.py @@ -24,6 +24,7 @@ logger = logging.getLogger("PacketRouter") # Deliver PATH and protocol-response (PATH) to companion at most once per logical packet # so the client is not spammed with duplicate telemetry when the mesh delivers multiple copies. _COMPANION_DEDUPE_TTL_SEC = 60.0 +_COMPANION_DEDUPE_PRUNE_THRESHOLD = 1000 def _companion_dedup_key(packet) -> str | None: @@ -50,6 +51,7 @@ class PacketRouter: self.queue = asyncio.Queue(maxsize=500) self.running = False self.router_task = None + self._route_tasks = set() # Serialize injects so one local TX completes before the next is processed self._inject_lock = asyncio.Lock() # Hash -> expiry time; skip delivering same PATH/protocol-response to companions more than once @@ -68,7 +70,22 @@ class PacketRouter: await self.router_task except asyncio.CancelledError: pass + # Cancel in-flight packet routing tasks during shutdown. + if self._route_tasks: + tasks = list(self._route_tasks) + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) logger.info("Packet router stopped") + + def _on_route_task_done(self, task: asyncio.Task) -> None: + self._route_tasks.discard(task) + try: + task.result() + except asyncio.CancelledError: + pass + except Exception as e: + logger.error("Router packet task error: %s", e, exc_info=True) def _should_deliver_path_to_companions(self, packet) -> bool: """Return True if this PATH/protocol-response should be delivered to companions (first of duplicates).""" @@ -76,8 +93,11 @@ class PacketRouter: if not key: return True now = time.time() - # Prune expired - self._companion_delivered = {k: v for k, v in self._companion_delivered.items() if v > now} + # Prune expired entries only when map grows beyond threshold to avoid per-packet full sweeps. + if len(self._companion_delivered) > _COMPANION_DEDUPE_PRUNE_THRESHOLD: + self._companion_delivered = { + k: v for k, v in self._companion_delivered.items() if v > now + } if key in self._companion_delivered: return False self._companion_delivered[key] = now + _COMPANION_DEDUPE_TTL_SEC @@ -146,7 +166,9 @@ class PacketRouter: while self.running: try: packet = await asyncio.wait_for(self.queue.get(), timeout=0.1) - await self._route_packet(packet) + task = asyncio.create_task(self._route_packet(packet)) + self._route_tasks.add(task) + task.add_done_callback(self._on_route_task_done) except asyncio.TimeoutError: continue except Exception as e: