Add thread safety to MeshConnector node caches and StatusData

- 4a: Add threading.Lock to MeshConnector protecting _node_names and
  _node_positions dicts that are read/written from Meshtastic's pubsub
  thread callbacks (_on_receive, _on_node_update, _cache_node_info)
  and read from async code (get_node_position, get_node_name)
- 4b: Add threading.Lock to StatusData protecting counters and activity
  list that are written from the async event loop and read from the
  HTTP server thread in to_dict()

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Ubuntu 2026-02-23 20:13:12 +00:00
commit dca03500ec
2 changed files with 75 additions and 60 deletions

View file

@ -2,6 +2,7 @@
import asyncio import asyncio
import logging import logging
import threading
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Callable, Optional from typing import Callable, Optional
@ -46,6 +47,7 @@ class MeshConnector:
self._node_names: dict[str, str] = {} self._node_names: dict[str, str] = {}
self._connected = False self._connected = False
self._loop: Optional[asyncio.AbstractEventLoop] = None self._loop: Optional[asyncio.AbstractEventLoop] = None
self._lock = threading.Lock()
@property @property
def connected(self) -> bool: def connected(self) -> bool:
@ -127,6 +129,7 @@ class MeshConnector:
if not self._interface: if not self._interface:
return return
with self._lock:
for node_id, node in self._interface.nodes.items(): for node_id, node in self._interface.nodes.items():
# Cache name # Cache name
if user := node.get("user"): if user := node.get("user"):
@ -144,6 +147,7 @@ class MeshConnector:
"""Handle node info updates.""" """Handle node info updates."""
node_id = f"!{node['num']:08x}" node_id = f"!{node['num']:08x}"
with self._lock:
# Update name cache # Update name cache
if user := node.get("user"): if user := node.get("user"):
name = user.get("shortName") or user.get("longName") or node_id name = user.get("shortName") or user.get("longName") or node_id
@ -175,8 +179,11 @@ class MeshConnector:
# Determine if DM (sent directly to us, not broadcast) # Determine if DM (sent directly to us, not broadcast)
is_dm = to_num == self._my_node_id is_dm = to_num == self._my_node_id
with self._lock:
# Get sender name # Get sender name
sender_name = self._node_names.get(sender_num, sender_num) sender_name = self._node_names.get(sender_num, sender_num)
# Get position if available
position = self._node_positions.get(sender_num)
# Create message object # Create message object
msg = MeshMessage( msg = MeshMessage(
@ -189,8 +196,8 @@ class MeshConnector:
) )
# Attach position if available # Attach position if available
if sender_num in self._node_positions: if position:
msg._position = self._node_positions[sender_num] msg._position = position
# Schedule callback on event loop # Schedule callback on event loop
self._loop.call_soon_threadsafe( self._loop.call_soon_threadsafe(
@ -258,6 +265,7 @@ class MeshConnector:
Returns: Returns:
Tuple of (latitude, longitude) or None if not available Tuple of (latitude, longitude) or None if not available
""" """
with self._lock:
return self._node_positions.get(node_id) return self._node_positions.get(node_id)
def get_node_name(self, node_id: str) -> str: def get_node_name(self, node_id: str) -> str:
@ -269,4 +277,5 @@ class MeshConnector:
Returns: Returns:
Node name or the node ID if name not available Node name or the node ID if name not available
""" """
with self._lock:
return self._node_names.get(node_id, node_id) return self._node_names.get(node_id, node_id)

View file

@ -3,6 +3,7 @@
import asyncio import asyncio
import json import json
import logging import logging
import threading
import time import time
from datetime import datetime from datetime import datetime
from http.server import BaseHTTPRequestHandler, HTTPServer from http.server import BaseHTTPRequestHandler, HTTPServer
@ -18,6 +19,7 @@ class StatusData:
"""Container for status information.""" """Container for status information."""
def __init__(self): def __init__(self):
self._lock = threading.Lock()
self.start_time = time.time() self.start_time = time.time()
self.message_count = 0 self.message_count = 0
self.response_count = 0 self.response_count = 0
@ -29,6 +31,7 @@ class StatusData:
def record_message(self, sender_id: str, sender_name: str): def record_message(self, sender_id: str, sender_name: str):
"""Record an incoming message.""" """Record an incoming message."""
with self._lock:
self.message_count += 1 self.message_count += 1
self.last_message_time = time.time() self.last_message_time = time.time()
self.connected_nodes.add(sender_id) self.connected_nodes.add(sender_id)
@ -43,10 +46,12 @@ class StatusData:
def record_response(self): def record_response(self):
"""Record an outgoing response.""" """Record an outgoing response."""
with self._lock:
self.response_count += 1 self.response_count += 1
def record_error(self, error: str): def record_error(self, error: str):
"""Record an error.""" """Record an error."""
with self._lock:
self.error_count += 1 self.error_count += 1
self.recent_activity.append({ self.recent_activity.append({
"type": "error", "type": "error",
@ -75,6 +80,7 @@ class StatusData:
def to_dict(self, include_activity: bool = False) -> dict: def to_dict(self, include_activity: bool = False) -> dict:
"""Convert to dictionary for JSON response.""" """Convert to dictionary for JSON response."""
with self._lock:
data = { data = {
"status": "online", "status": "online",
"uptime": self.get_uptime(), "uptime": self.get_uptime(),
@ -90,7 +96,7 @@ class StatusData:
data["last_message_ago"] = int(time.time() - self.last_message_time) data["last_message_ago"] = int(time.time() - self.last_message_time)
if include_activity: if include_activity:
data["recent_activity"] = self.recent_activity data["recent_activity"] = list(self.recent_activity)
return data return data