From dc82a4c7b0b6e8bf85041dce6f29cf25b0a7b7d5 Mon Sep 17 00:00:00 2001 From: Louis King Date: Wed, 3 Sep 2025 10:51:34 +0100 Subject: [PATCH] feat: add message deduplication to prevent duplicate MQTT messages on reconnect --- meshcore_mqtt/meshcore_worker.py | 109 ++++++++++++++++++++++++- tests/test_deduplication.py | 131 +++++++++++++++++++++++++++++++ 2 files changed, 239 insertions(+), 1 deletion(-) create mode 100644 tests/test_deduplication.py diff --git a/meshcore_mqtt/meshcore_worker.py b/meshcore_mqtt/meshcore_worker.py index 7d1602c..21eefe0 100644 --- a/meshcore_mqtt/meshcore_worker.py +++ b/meshcore_mqtt/meshcore_worker.py @@ -3,7 +3,7 @@ import asyncio import logging import time -from typing import TYPE_CHECKING, Any, Optional +from typing import TYPE_CHECKING, Any, Dict, Optional if TYPE_CHECKING: import serial @@ -64,6 +64,11 @@ class MeshCoreWorker: self._consecutive_health_failures = 0 self._max_health_failures = 3 + # Message deduplication + self._message_cache: Dict[str, float] = {} + self._cache_max_size = 1000 + self._cache_ttl = 300 # 5 minutes + # Worker state self._running = False self._shutdown_event = asyncio.Event() @@ -200,6 +205,12 @@ class MeshCoreWorker: self._connected = True self._last_activity = time.time() + # Clear message deduplication cache on new connection + self._message_cache.clear() + self.logger.debug( + "Cleared message deduplication cache for fresh connection" + ) + # Send connection status await self._send_status_update(ComponentStatus.CONNECTED, "connected") @@ -379,6 +390,8 @@ class MeshCoreWorker: "connected": self._connected, "last_activity": self._last_activity, "auto_fetch_running": self._auto_fetch_running, + "message_cache_size": len(self._message_cache), + "message_cache_max_size": self._cache_max_size, }, ) await self.message_bus.send_message(response) @@ -582,6 +595,91 @@ class MeshCoreWorker: else: self.logger.error("🚨 MeshCore recovery failed permanently") + def _generate_message_fingerprint(self, event_data: Any) -> str: + """Generate a unique fingerprint for message deduplication.""" + import hashlib + + try: + # Create fingerprint based on key message attributes + fingerprint_data = [] + + # Add event type + event_type_name = getattr(event_data, "type", "UNKNOWN") + fingerprint_data.append(str(event_type_name)) + + # For message events, include message content and metadata + if hasattr(event_data, "payload") and isinstance(event_data.payload, dict): + payload = event_data.payload + + # Include message text, sender, channel for uniqueness + if "text" in payload: + fingerprint_data.append(payload["text"]) + if "from" in payload: + fingerprint_data.append(payload["from"]) + if "channel_idx" in payload: + fingerprint_data.append(str(payload["channel_idx"])) + if "timestamp" in payload: + fingerprint_data.append(str(payload["timestamp"])) + if "msg_id" in payload: + fingerprint_data.append(str(payload["msg_id"])) + + # For other events, include key identifying attributes + elif hasattr(event_data, "payload"): + fingerprint_data.append(str(event_data.payload)) + + # Create hash from combined data + combined = "|".join(fingerprint_data) + return hashlib.md5(combined.encode()).hexdigest()[:16] + + except Exception as e: + self.logger.debug(f"Error generating message fingerprint: {e}") + # Fallback: use object string representation + return hashlib.md5(str(event_data).encode()).hexdigest()[:16] + + def _is_duplicate_message(self, fingerprint: str) -> bool: + """Check if message is a duplicate and update cache.""" + current_time = time.time() + + # Clean expired entries from cache + self._clean_message_cache(current_time) + + # Check if this message was seen recently + if fingerprint in self._message_cache: + self.logger.debug(f"Duplicate message detected: {fingerprint}") + return True + + # Add to cache + self._message_cache[fingerprint] = current_time + + # Ensure cache doesn't exceed max size after adding + if len(self._message_cache) > self._cache_max_size: + sorted_items = sorted(self._message_cache.items(), key=lambda x: x[1]) + excess_count = len(self._message_cache) - self._cache_max_size + + for key, _ in sorted_items[:excess_count]: + del self._message_cache[key] + + return False + + def _clean_message_cache(self, current_time: float) -> None: + """Remove expired entries from message cache.""" + expired_keys = [ + key + for key, timestamp in self._message_cache.items() + if current_time - timestamp > self._cache_ttl + ] + + for key in expired_keys: + del self._message_cache[key] + + # If cache is still too large, remove oldest entries + if len(self._message_cache) > self._cache_max_size: + sorted_items = sorted(self._message_cache.items(), key=lambda x: x[1]) + excess_count = len(self._message_cache) - self._cache_max_size + + for key, _ in sorted_items[:excess_count]: + del self._message_cache[key] + def _on_meshcore_event(self, event_data: Any) -> None: """Handle MeshCore events and forward them to MQTT.""" try: @@ -604,6 +702,15 @@ class MeshCoreWorker: if event_name in ["CONNECTED", "DISCONNECTED"]: self.logger.info(f"MeshCore {event_name} event received: {event_data}") + # Check for duplicate messages (except for connection events) + if event_name not in ["CONNECTED", "DISCONNECTED"]: + fingerprint = self._generate_message_fingerprint(event_data) + if self._is_duplicate_message(fingerprint): + self.logger.debug( + f"Dropping duplicate {event_name} event: {fingerprint}" + ) + return + # Create message for MQTT worker message = Message.create( message_type=MessageType.MESHCORE_EVENT, diff --git a/tests/test_deduplication.py b/tests/test_deduplication.py new file mode 100644 index 0000000..e8782c1 --- /dev/null +++ b/tests/test_deduplication.py @@ -0,0 +1,131 @@ +"""Test message deduplication functionality.""" + +import pytest +from unittest.mock import Mock +from meshcore_mqtt.config import Config, MeshCoreConfig, MQTTConfig +from meshcore_mqtt.meshcore_worker import MeshCoreWorker + + +class TestMessageDeduplication: + """Test message deduplication in MeshCore worker.""" + + @pytest.fixture + def config(self): + """Create a test configuration.""" + return Config( + meshcore=MeshCoreConfig( + connection_type="tcp", + address="127.0.0.1", + port=12345, + events=["CONTACT_MSG_RECV"], + ), + mqtt=MQTTConfig( + broker="localhost", + port=1883, + topic_prefix="test/meshcore", + ), + ) + + @pytest.fixture + def worker(self, config): + """Create a MeshCore worker for testing.""" + return MeshCoreWorker(config) + + def test_fingerprint_generation_message_events(self, worker): + """Test fingerprint generation for message events.""" + # Mock message event data + mock_event = Mock() + mock_event.type = "CONTACT_MSG_RECV" + mock_event.payload = { + "text": "Hello World", + "from": "user123", + "channel_idx": 0, + "timestamp": 1627849200, + "msg_id": "msg_123", + } + + fingerprint1 = worker._generate_message_fingerprint(mock_event) + fingerprint2 = worker._generate_message_fingerprint(mock_event) + + # Same event should generate same fingerprint + assert fingerprint1 == fingerprint2 + assert len(fingerprint1) == 16 # MD5 hash truncated to 16 chars + + def test_fingerprint_generation_different_messages(self, worker): + """Test that different messages generate different fingerprints.""" + # Mock first message + mock_event1 = Mock() + mock_event1.type = "CONTACT_MSG_RECV" + mock_event1.payload = { + "text": "Hello World", + "from": "user123", + } + + # Mock second message (different text) + mock_event2 = Mock() + mock_event2.type = "CONTACT_MSG_RECV" + mock_event2.payload = { + "text": "Hello Universe", # Different text + "from": "user123", + } + + fingerprint1 = worker._generate_message_fingerprint(mock_event1) + fingerprint2 = worker._generate_message_fingerprint(mock_event2) + + assert fingerprint1 != fingerprint2 + + def test_duplicate_detection(self, worker): + """Test duplicate message detection.""" + fingerprint = "test123456789abc" + + # First message should not be duplicate + assert not worker._is_duplicate_message(fingerprint) + + # Same fingerprint should now be duplicate + assert worker._is_duplicate_message(fingerprint) + + # Different fingerprint should not be duplicate + assert not worker._is_duplicate_message("different456789xyz") + + def test_cache_expiry(self, worker): + """Test that old cache entries are expired.""" + import time + + fingerprint = "expire123456789" + + # Add message to cache + assert not worker._is_duplicate_message(fingerprint) + + # Manually set old timestamp (simulate expired entry) + worker._message_cache[fingerprint] = time.time() - 400 # 400 sec ago + + # Should not be duplicate after expiry + assert not worker._is_duplicate_message(fingerprint) + + def test_cache_size_limit(self, worker): + """Test that cache size is limited.""" + # Set small cache size for testing + worker._cache_max_size = 3 + + # Add messages up to limit + 2 + for i in range(6): + fingerprint = f"msg{i:016d}" + worker._is_duplicate_message(fingerprint) + + # Cache should not exceed max size after cleanup + assert len(worker._message_cache) <= worker._cache_max_size + + def test_connection_events_not_deduplicated(self, worker): + """Test that connection events are not subject to deduplication.""" + # Mock connection event + mock_event = Mock() + mock_event.type = "CONNECTED" + mock_event.payload = {"status": "connected"} + + # Connection events should always have unique fingerprints + # (because they're excluded from deduplication logic) + fingerprint1 = worker._generate_message_fingerprint(mock_event) + fingerprint2 = worker._generate_message_fingerprint(mock_event) + + # Even though fingerprints are same, connection events bypass deduplication + assert fingerprint1 == fingerprint2 \ No newline at end of file