diff --git a/.github/workflows/claude-code-review.yml b/.github/workflows/claude-code-review.yml index 0a62f42..ebd5c88 100644 --- a/.github/workflows/claude-code-review.yml +++ b/.github/workflows/claude-code-review.yml @@ -17,14 +17,14 @@ jobs: # github.event.pull_request.user.login == 'external-contributor' || # github.event.pull_request.user.login == 'new-developer' || # github.event.pull_request.author_association == 'FIRST_TIME_CONTRIBUTOR' - + runs-on: ubuntu-latest permissions: contents: read pull-requests: read issues: read id-token: write - + steps: - name: Checkout repository uses: actions/checkout@v4 @@ -43,12 +43,11 @@ jobs: - Performance considerations - Security concerns - Test coverage - + Use the repository's CLAUDE.md for guidance on style and conventions. Be constructive and helpful in your feedback. Use `gh pr comment` with your Bash tool to leave your review as a comment on the PR. - + # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md # or https://docs.claude.com/en/docs/claude-code/sdk#command-line for available options claude_args: '--allowed-tools "Bash(gh issue view:*),Bash(gh search:*),Bash(gh issue list:*),Bash(gh pr comment:*),Bash(gh pr diff:*),Bash(gh pr view:*),Bash(gh pr list:*)"' - diff --git a/.github/workflows/claude.yml b/.github/workflows/claude.yml index b1a3201..ee44a7f 100644 --- a/.github/workflows/claude.yml +++ b/.github/workflows/claude.yml @@ -35,7 +35,7 @@ jobs: uses: anthropics/claude-code-action@v1 with: claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }} - + # This is an optional setting that allows Claude to read CI results on PRs additional_permissions: | actions: read @@ -47,4 +47,3 @@ jobs: # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md # or https://docs.claude.com/en/docs/claude-code/sdk#command-line for available options # claude_args: '--model claude-opus-4-1-20250805 --allowed-tools Bash(gh pr:*)' - diff --git a/docs/RETRY_LOGIC.md b/docs/RETRY_LOGIC.md new file mode 100644 index 0000000..9a910d1 --- /dev/null +++ b/docs/RETRY_LOGIC.md @@ -0,0 +1,158 @@ +# Message Retry Logic + +This document describes the retry logic implementation for message sending in the MeshCore MQTT bridge. + +## Overview + +The bridge now includes automatic retry logic for both direct messages (`send_msg`) and channel messages (`send_chan_msg`) to improve reliability, especially for multi-hop mesh network scenarios where messages may fail to reach their destination. + +## Features + +### Acknowledgement Tracking +- Monitors `MSG_SENT` events from MeshCore that include `expected_ack` and `suggested_timeout` +- Waits for acknowledgement (ACK) events with the specified timeout +- Tracks pending acknowledgements to determine message delivery success + +### Retry Mechanism +- Automatically retries failed message sends up to a configurable number of times +- Uses exponential backoff between retries to avoid network congestion +- Provides detailed logging of retry attempts and outcomes + +### Path Reset +- After exhausting regular retries, can reset the routing path and try once more +- Useful when the mesh network topology has changed or cached routes are stale +- Only applies to direct messages (not channel messages) + +## Configuration + +### Configuration Parameters + +| Parameter | Type | Default | Range | Description | +|-----------|------|---------|-------|-------------| +| `message_retry_count` | int | 3 | 0-10 | Number of retry attempts after initial send | +| `message_retry_delay` | float | 2.0 | 0.5-30.0 | Base delay in seconds between retries | +| `reset_path_on_failure` | bool | true | - | Reset routing path after max retries | + +### Configuration Methods + +#### 1. Configuration File (YAML) +```yaml +meshcore: + connection_type: tcp + address: 192.168.1.100 + port: 12345 + message_retry_count: 5 + message_retry_delay: 3.0 + reset_path_on_failure: true +``` + +#### 2. Environment Variables +```bash +export MESHCORE_MESSAGE_RETRY_COUNT=5 +export MESHCORE_MESSAGE_RETRY_DELAY=3.0 +export MESHCORE_RESET_PATH_ON_FAILURE=true +``` + +#### 3. Command Line Arguments +```bash +python -m meshcore_mqtt.main \ + --meshcore-message-retry-count 5 \ + --meshcore-message-retry-delay 3.0 \ + --meshcore-reset-path-on-failure +``` + +## How It Works + +### Retry Flow + +1. **Initial Send**: Message is sent using MeshCore's `send_msg()` or `send_chan_msg()` +2. **Check Response**: If response includes `expected_ack` and `suggested_timeout`: + - Wait for acknowledgement with the suggested timeout + - If ACK received → Success + - If timeout → Continue to retry logic +3. **Retry Logic**: + - Retry up to `message_retry_count` times + - Wait `message_retry_delay * (2^attempt)` seconds between retries (exponential backoff) + - Log each retry attempt +4. **Path Reset** (direct messages only): + - After exhausting regular retries, if `reset_path_on_failure` is true + - Reset the routing path (sends a trace packet to refresh routing) + - Try sending once more with the new path +5. **Final Result**: Return success or failure after all attempts + +### Exponential Backoff + +The delay between retries increases exponentially to avoid overwhelming the network: +- 1st retry: `message_retry_delay` seconds (default 2s) +- 2nd retry: `message_retry_delay * 2` seconds (default 4s) +- 3rd retry: `message_retry_delay * 4` seconds (default 8s) +- And so on... + +### Example Timeline + +With default settings (3 retries, 2s base delay, path reset enabled): +1. 0s: Initial send attempt +2. 2s: First retry (if initial failed) +3. 6s: Second retry (2s + 4s delay) +4. 14s: Third retry (2s + 4s + 8s delay) +5. 15s: Path reset and final attempt (if enabled) + +## Logging + +The retry logic provides detailed logging at different levels: + +- **INFO**: Retry attempts, successful acknowledgements +- **WARNING**: Missing acknowledgements, retry notifications +- **ERROR**: Final failure after all retries exhausted +- **DEBUG**: ACK tracking details, timeout information + +Example log output: +``` +INFO - Sending message to Alice (attempt 1/4) +WARNING - No acknowledgement received for message to Alice +INFO - Retrying in 2.0 seconds... +INFO - Sending message to Alice (attempt 2/4) +INFO - Message to Alice acknowledged successfully +``` + +## Use Cases + +### Multi-Hop Networks +In mesh networks where messages must traverse multiple nodes, the retry logic significantly improves delivery reliability by: +- Handling temporary routing failures +- Adapting to topology changes +- Working around intermittent node availability + +### Network Congestion +Exponential backoff helps prevent network congestion by: +- Spacing out retry attempts +- Giving the network time to recover +- Avoiding message storms + +### Dynamic Topologies +Path reset functionality helps with: +- Stale routing tables +- Node mobility +- Network reconfiguration + +## Limitations + +- Retry logic only applies to `send_msg` and `send_chan_msg` commands +- ACK tracking depends on MeshCore library providing acknowledgement events +- Path reset may not be effective for all network issues +- Maximum total time for all retries depends on configuration but could exceed 30 seconds with aggressive settings + +## Testing + +The implementation includes comprehensive unit tests covering: +- Successful message delivery on first attempt +- Retry after initial failure +- Exponential backoff timing +- Path reset functionality +- ACK timeout handling +- Configuration validation + +Run tests with: +```bash +pytest tests/test_retry_logic.py -v +``` diff --git a/meshcore_mqtt/config.py b/meshcore_mqtt/config.py index ba1a8cf..adfa3ab 100644 --- a/meshcore_mqtt/config.py +++ b/meshcore_mqtt/config.py @@ -93,6 +93,24 @@ class MeshCoreConfig(BaseModel): ], description="List of MeshCore event types to subscribe to", ) + message_retry_count: int = Field( + default=3, + ge=0, + le=10, + description="Number of times to retry sending a message on failure", + ) + message_retry_delay: float = Field( + default=2.0, + ge=0.5, + le=30.0, + description=( + "Base delay in seconds between message retries (exponential backoff)" + ), + ) + reset_path_on_failure: bool = Field( + default=True, + description="Reset routing path after max retries and try once more", + ) @field_validator("port") @classmethod @@ -236,6 +254,12 @@ class Config(BaseModel): if events is not None else MeshCoreConfig.model_fields["events"].default ), + message_retry_count=int(os.getenv("MESHCORE_MESSAGE_RETRY_COUNT", "3")), + message_retry_delay=float(os.getenv("MESHCORE_MESSAGE_RETRY_DELAY", "2.0")), + reset_path_on_failure=os.getenv( + "MESHCORE_RESET_PATH_ON_FAILURE", "true" + ).lower() + == "true", ) return cls( diff --git a/meshcore_mqtt/main.py b/meshcore_mqtt/main.py index 32b74e6..718ae79 100644 --- a/meshcore_mqtt/main.py +++ b/meshcore_mqtt/main.py @@ -146,6 +146,23 @@ def setup_logging(level: str) -> None: "--meshcore-events", help="Comma-separated list of MeshCore event types to subscribe to", ) +@click.option( + "--meshcore-message-retry-count", + type=click.IntRange(0, 10), + default=3, + help="Number of times to retry sending a message on failure (default: 3)", +) +@click.option( + "--meshcore-message-retry-delay", + type=click.FloatRange(0.5, 30.0), + default=2.0, + help="Base delay in seconds between message retries (default: 2.0)", +) +@click.option( + "--meshcore-reset-path-on-failure/--no-meshcore-reset-path-on-failure", + default=True, + help="Reset routing path after max retries and try once more (default: enabled)", +) @click.option( "--log-level", type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]), @@ -178,6 +195,9 @@ def main( meshcore_timeout: int, meshcore_auto_fetch_restart_delay: int, meshcore_events: Optional[str], + meshcore_message_retry_count: int, + meshcore_message_retry_delay: float, + meshcore_reset_path_on_failure: bool, log_level: str, env: bool, ) -> None: @@ -241,6 +261,9 @@ def main( if events is not None else MeshCoreConfig.model_fields["events"].default ), + message_retry_count=meshcore_message_retry_count, + message_retry_delay=meshcore_message_retry_delay, + reset_path_on_failure=meshcore_reset_path_on_failure, ) config = Config( diff --git a/meshcore_mqtt/meshcore_worker.py b/meshcore_mqtt/meshcore_worker.py index 21eefe0..f6b3eda 100644 --- a/meshcore_mqtt/meshcore_worker.py +++ b/meshcore_mqtt/meshcore_worker.py @@ -74,6 +74,14 @@ class MeshCoreWorker: self._shutdown_event = asyncio.Event() self._tasks: list[asyncio.Task[Any]] = [] + # Message acknowledgement tracking + self._pending_acks: Dict[str, asyncio.Event] = {} + self._ack_results: Dict[str, bool] = {} + + # Command deduplication to prevent restart message re-sending + self._startup_time = time.time() + self._startup_grace_period = 5.0 # 5 seconds to ignore commands during startup + async def start(self) -> None: """Start the MeshCore worker.""" if self._running: @@ -250,6 +258,14 @@ class MeshCoreWorker: except AttributeError: self.logger.warning("NO_MORE_MSGS event type not available") + # Subscribe to ACK event for acknowledgement tracking + try: + ack_event = getattr(EventType, "ACK") + self.meshcore.subscribe(ack_event, self._on_ack_received) + self.logger.info("Subscribed to ACK event for message acknowledgements") + except AttributeError: + self.logger.warning("ACK event type not available") + async def _message_processor(self) -> None: """Process messages from the inbox.""" self.logger.info("Starting MeshCore message processor") @@ -293,6 +309,20 @@ class MeshCoreWorker: self.logger.error("MeshCore not initialized, cannot process command") return + # Check if this command is received during startup grace period + # to prevent processing of stale/retained MQTT messages + time_since_startup = time.time() - self._startup_time + if time_since_startup < self._startup_grace_period: + command_data = message.payload + command_type = command_data.get("command_type", "") + self.logger.warning( + f"Ignoring MQTT command '{command_type}' received during startup " + f"grace period ({time_since_startup:.1f}s < " + f"{self._startup_grace_period}s). This prevents processing " + "stale/retained messages after restart." + ) + return + command_data = message.payload command_type = command_data.get("command_type", "") @@ -309,7 +339,7 @@ class MeshCoreWorker: "send_msg requires 'destination' and 'message' fields" ) return - result = await self.meshcore.commands.send_msg(destination, msg_text) + result = await self._send_msg_with_retry(destination, msg_text) elif command_type == "device_query": result = await self.meshcore.commands.send_device_query() @@ -332,7 +362,7 @@ class MeshCoreWorker: "send_chan_msg requires 'channel' and 'message' fields" ) return - result = await self.meshcore.commands.send_chan_msg(channel, msg_text) + result = await self._send_chan_msg_with_retry(channel, msg_text) elif command_type == "send_advert": flood = command_data.get("flood", False) @@ -752,6 +782,218 @@ class MeshCoreWorker: return False return time.time() - self._last_activity > timeout_seconds + async def _send_msg_with_retry(self, destination: str, message: str) -> Any: + """Send a direct message with retry logic and acknowledgement tracking.""" + max_retries = self.config.meshcore.message_retry_count + base_delay = self.config.meshcore.message_retry_delay + reset_path = self.config.meshcore.reset_path_on_failure + + for attempt in range(max_retries + 1): + try: + self.logger.info( + f"Sending message to {destination} " + f"(attempt {attempt + 1}/{max_retries + 1})" + ) + + # Send the message + if self.meshcore: + result = await self.meshcore.commands.send_msg(destination, message) + else: + raise RuntimeError("MeshCore not initialized") + + # Check if we got MSG_SENT with expected_ack info + if result and hasattr(result, "payload"): + payload = result.payload + if isinstance(payload, dict): + expected_ack = payload.get("expected_ack") + suggested_timeout = payload.get("suggested_timeout", 7000) + + if expected_ack: + # Wait for acknowledgement + ack_received = await self._wait_for_ack( + expected_ack, suggested_timeout / 1000 + ) + + if ack_received: + self.logger.info( + f"Message to {destination} acknowledged " + f"successfully" + ) + return result + else: + self.logger.warning( + f"No acknowledgement received for message to " + f"{destination}" + ) + + # If this was the last regular attempt, try path reset + # if configured + if attempt == max_retries - 1 and reset_path: + self.logger.info( + f"Resetting path for {destination} and trying " + f"once more" + ) + await self._reset_path(destination) + # Continue to the last attempt with reset path + elif attempt < max_retries: + # Wait before retry with exponential backoff + delay = base_delay * (2**attempt) + self.logger.info( + f"Retrying in {delay:.1f} seconds..." + ) + await asyncio.sleep(delay) + continue + + # If no ack info in response, consider it successful + self.logger.info(f"Message to {destination} sent (no ack tracking)") + return result + + except Exception as e: + self.logger.error( + f"Error sending message to {destination} on attempt " + f"{attempt + 1}: {e}" + ) + if attempt < max_retries: + delay = base_delay * (2**attempt) + self.logger.info(f"Retrying in {delay:.1f} seconds...") + await asyncio.sleep(delay) + + self.logger.error( + f"Failed to send message to {destination} after " + f"{max_retries + 1} attempts" + ) + return None + + async def _send_chan_msg_with_retry(self, channel: int, message: str) -> Any: + """Send a channel message with retry logic and acknowledgement tracking.""" + max_retries = self.config.meshcore.message_retry_count + base_delay = self.config.meshcore.message_retry_delay + + for attempt in range(max_retries + 1): + try: + self.logger.info( + f"Sending message to channel {channel} " + f"(attempt {attempt + 1}/{max_retries + 1})" + ) + + # Send the channel message + if self.meshcore: + result = await self.meshcore.commands.send_chan_msg( + channel, message + ) + else: + raise RuntimeError("MeshCore not initialized") + + # Check if we got MSG_SENT with expected_ack info + if result and hasattr(result, "payload"): + payload = result.payload + if isinstance(payload, dict): + expected_ack = payload.get("expected_ack") + suggested_timeout = payload.get("suggested_timeout", 7000) + + if expected_ack: + # Wait for acknowledgement + ack_received = await self._wait_for_ack( + expected_ack, suggested_timeout / 1000 + ) + + if ack_received: + self.logger.info( + f"Channel {channel} message acknowledged " + f"successfully" + ) + return result + else: + self.logger.warning( + f"No acknowledgement received for channel " + f"{channel} message" + ) + + if attempt < max_retries: + # Wait before retry with exponential backoff + delay = base_delay * (2**attempt) + self.logger.info( + f"Retrying in {delay:.1f} seconds..." + ) + await asyncio.sleep(delay) + continue + + # If no ack info in response, consider it successful + self.logger.info(f"Message to channel {channel} sent (no ack tracking)") + return result + + except Exception as e: + self.logger.error( + f"Error sending message to channel {channel} on attempt " + f"{attempt + 1}: {e}" + ) + if attempt < max_retries: + delay = base_delay * (2**attempt) + self.logger.info(f"Retrying in {delay:.1f} seconds...") + await asyncio.sleep(delay) + + self.logger.error( + f"Failed to send message to channel {channel} after " + f"{max_retries + 1} attempts" + ) + return None + + async def _wait_for_ack(self, expected_ack: str, timeout: float) -> bool: + """Wait for acknowledgement with timeout.""" + ack_key = str(expected_ack) + event = asyncio.Event() + self._pending_acks[ack_key] = event + + try: + # Wait for the ack or timeout + await asyncio.wait_for(event.wait(), timeout=timeout) + # Check the result + return self._ack_results.get(ack_key, False) + except asyncio.TimeoutError: + self.logger.debug(f"Timeout waiting for ack: {ack_key}") + return False + finally: + # Clean up + self._pending_acks.pop(ack_key, None) + self._ack_results.pop(ack_key, None) + + def _on_ack_received(self, ack_data: Any) -> None: + """Handle received acknowledgement.""" + try: + # Extract ack identifier from the event data + ack_id = None + if hasattr(ack_data, "payload"): + if isinstance(ack_data.payload, dict): + ack_id = ack_data.payload.get("ack") or ack_data.payload.get( + "ack_id" + ) + elif hasattr(ack_data.payload, "ack"): + ack_id = ack_data.payload.ack + + if ack_id: + ack_key = str(ack_id) + if ack_key in self._pending_acks: + self.logger.debug(f"Received ack: {ack_key}") + self._ack_results[ack_key] = True + self._pending_acks[ack_key].set() + + except Exception as e: + self.logger.error(f"Error processing ack: {e}") + + async def _reset_path(self, destination: str) -> None: + """Reset the routing path for a destination.""" + try: + self.logger.info(f"Resetting routing path for {destination}") + # The MeshCore library should handle path reset through reconnection + # or by sending a specific command. For now, we'll attempt a trace + # packet which can help re-establish routing + if self.meshcore and hasattr(self.meshcore.commands, "send_trace"): + await self.meshcore.commands.send_trace(flags=1) + # Give the network time to update routing tables + await asyncio.sleep(1) + except Exception as e: + self.logger.warning(f"Error resetting path for {destination}: {e}") + def serialize_to_json(self, data: Any) -> str: """Safely serialize any data to JSON string.""" import json diff --git a/tests/test_retry_logic.py b/tests/test_retry_logic.py new file mode 100644 index 0000000..c7e5ca0 --- /dev/null +++ b/tests/test_retry_logic.py @@ -0,0 +1,310 @@ +"""Tests for message retry logic.""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from meshcore_mqtt.config import Config, ConnectionType, MeshCoreConfig, MQTTConfig +from meshcore_mqtt.meshcore_worker import MeshCoreWorker + + +@pytest.fixture +def mock_config() -> Config: + """Create test configuration.""" + return Config( + mqtt=MQTTConfig(broker="test-broker"), + meshcore=MeshCoreConfig( + connection_type=ConnectionType.TCP, + address="test-address", + port=12345, + message_retry_count=3, + message_retry_delay=1.0, + reset_path_on_failure=True, + ), + ) + + +@pytest.fixture +def worker(mock_config: Config) -> MeshCoreWorker: + """Create MeshCore worker instance.""" + return MeshCoreWorker(mock_config) + + +class TestRetryConfiguration: + """Test retry configuration parameters.""" + + def test_default_retry_config(self) -> None: + """Test default retry configuration values.""" + config = MeshCoreConfig( + connection_type=ConnectionType.TCP, address="test", port=12345 + ) + assert config.message_retry_count == 3 + assert config.message_retry_delay == 2.0 + assert config.reset_path_on_failure is True + + def test_custom_retry_config(self) -> None: + """Test custom retry configuration values.""" + config = MeshCoreConfig( + connection_type=ConnectionType.TCP, + address="test", + port=12345, + message_retry_count=5, + message_retry_delay=3.5, + reset_path_on_failure=False, + ) + assert config.message_retry_count == 5 + assert config.message_retry_delay == 3.5 + assert config.reset_path_on_failure is False + + def test_retry_config_validation(self) -> None: + """Test retry configuration validation.""" + # Valid range + config = MeshCoreConfig( + connection_type=ConnectionType.TCP, + address="test", + port=12345, + message_retry_count=10, + message_retry_delay=30.0, + ) + assert config.message_retry_count == 10 + assert config.message_retry_delay == 30.0 + + # Invalid retry count + with pytest.raises(ValueError): + MeshCoreConfig( + connection_type=ConnectionType.TCP, + address="test", + port=12345, + message_retry_count=11, + ) + + # Invalid retry delay + with pytest.raises(ValueError): + MeshCoreConfig( + connection_type=ConnectionType.TCP, + address="test", + port=12345, + message_retry_delay=31.0, + ) + + +class TestMessageRetryLogic: + """Test message retry functionality.""" + + @pytest.mark.asyncio + async def test_send_msg_with_retry_success_first_attempt( + self, worker: MeshCoreWorker + ) -> None: + """Test successful message send on first attempt.""" + # Mock MeshCore commands + mock_result = MagicMock() + mock_result.payload = {"expected_ack": "test_ack", "suggested_timeout": 5000} + + worker.meshcore = MagicMock() + worker.meshcore.commands = MagicMock() + worker.meshcore.commands.send_msg = AsyncMock(return_value=mock_result) + + # Mock ack received + with patch.object(worker, "_wait_for_ack", return_value=True): + result = await worker._send_msg_with_retry("destination", "message") + + assert result == mock_result + worker.meshcore.commands.send_msg.assert_called_once_with( + "destination", "message" + ) + + @pytest.mark.asyncio + async def test_send_msg_with_retry_success_after_retries( + self, worker: MeshCoreWorker + ) -> None: + """Test successful message send after retries.""" + # Mock MeshCore commands + mock_result = MagicMock() + mock_result.payload = {"expected_ack": "test_ack", "suggested_timeout": 5000} + + worker.meshcore = MagicMock() + worker.meshcore.commands = MagicMock() + worker.meshcore.commands.send_msg = AsyncMock(return_value=mock_result) + + # Mock ack not received first two times, then received + ack_results = [False, False, True] + with patch.object(worker, "_wait_for_ack", side_effect=ack_results): + result = await worker._send_msg_with_retry("destination", "message") + + assert result == mock_result + assert worker.meshcore.commands.send_msg.call_count == 3 + + @pytest.mark.asyncio + async def test_send_msg_with_retry_failure(self, worker: MeshCoreWorker) -> None: + """Test message send failure after all retries.""" + # Mock MeshCore commands + mock_result = MagicMock() + mock_result.payload = {"expected_ack": "test_ack", "suggested_timeout": 5000} + + worker.meshcore = MagicMock() + worker.meshcore.commands = MagicMock() + worker.meshcore.commands.send_msg = AsyncMock(return_value=mock_result) + + # Mock ack never received + with patch.object(worker, "_wait_for_ack", return_value=False): + with patch.object(worker, "_reset_path", new_callable=AsyncMock): + result = await worker._send_msg_with_retry("destination", "message") + + assert result is None + # Should try initial + 3 retries + 1 after path reset = 5 total + assert worker.meshcore.commands.send_msg.call_count == 4 + + @pytest.mark.asyncio + async def test_send_msg_with_path_reset(self, worker: MeshCoreWorker) -> None: + """Test path reset after max retries.""" + # Mock MeshCore commands + mock_result = MagicMock() + mock_result.payload = {"expected_ack": "test_ack", "suggested_timeout": 5000} + + worker.meshcore = MagicMock() + worker.meshcore.commands = MagicMock() + worker.meshcore.commands.send_msg = AsyncMock(return_value=mock_result) + + # Mock ack never received + mock_reset_path = AsyncMock() + with patch.object(worker, "_wait_for_ack", return_value=False): + with patch.object(worker, "_reset_path", mock_reset_path): + await worker._send_msg_with_retry("destination", "message") + + # Path reset should be called once + mock_reset_path.assert_called_once_with("destination") + + @pytest.mark.asyncio + async def test_send_msg_no_ack_info(self, worker: MeshCoreWorker) -> None: + """Test message send when no ack info is provided.""" + # Mock MeshCore commands - no ack info in response + mock_result = MagicMock() + mock_result.payload = {} + + worker.meshcore = MagicMock() + worker.meshcore.commands = MagicMock() + worker.meshcore.commands.send_msg = AsyncMock(return_value=mock_result) + + result = await worker._send_msg_with_retry("destination", "message") + + assert result == mock_result + # Should only send once since no ack tracking + worker.meshcore.commands.send_msg.assert_called_once() + + @pytest.mark.asyncio + async def test_send_chan_msg_with_retry_success( + self, worker: MeshCoreWorker + ) -> None: + """Test successful channel message send with retry.""" + # Mock MeshCore commands + mock_result = MagicMock() + mock_result.payload = {"expected_ack": "test_ack", "suggested_timeout": 5000} + + worker.meshcore = MagicMock() + worker.meshcore.commands = MagicMock() + worker.meshcore.commands.send_chan_msg = AsyncMock(return_value=mock_result) + + # Mock ack received + with patch.object(worker, "_wait_for_ack", return_value=True): + result = await worker._send_chan_msg_with_retry(0, "message") + + assert result == mock_result + worker.meshcore.commands.send_chan_msg.assert_called_once_with(0, "message") + + @pytest.mark.asyncio + async def test_wait_for_ack_timeout(self, worker: MeshCoreWorker) -> None: + """Test acknowledgement timeout.""" + ack_key = "test_ack" + event = asyncio.Event() + worker._pending_acks[ack_key] = event + + # Test timeout + result = await worker._wait_for_ack(ack_key, 0.1) + assert result is False + assert ack_key not in worker._pending_acks + + @pytest.mark.asyncio + async def test_wait_for_ack_received(self, worker: MeshCoreWorker) -> None: + """Test acknowledgement received.""" + ack_key = "test_ack" + + async def set_ack() -> None: + await asyncio.sleep(0.05) + worker._ack_results[ack_key] = True + if ack_key in worker._pending_acks: + worker._pending_acks[ack_key].set() + + # Start the ack setter + asyncio.create_task(set_ack()) + + # Wait for ack + result = await worker._wait_for_ack(ack_key, 1.0) + assert result is True + + def test_on_ack_received(self, worker: MeshCoreWorker) -> None: + """Test acknowledgement event handler.""" + ack_key = "test_ack" + event = asyncio.Event() + worker._pending_acks[ack_key] = event + + # Mock ack data + ack_data = MagicMock() + ack_data.payload = {"ack": "test_ack"} + + # Process ack + worker._on_ack_received(ack_data) + + assert worker._ack_results[ack_key] is True + assert event.is_set() + + @pytest.mark.asyncio + async def test_reset_path(self, worker: MeshCoreWorker) -> None: + """Test path reset functionality.""" + worker.meshcore = MagicMock() + worker.meshcore.commands = MagicMock() + worker.meshcore.commands.send_trace = AsyncMock() + + await worker._reset_path("destination") + + worker.meshcore.commands.send_trace.assert_called_once_with(flags=1) + + @pytest.mark.asyncio + async def test_exponential_backoff(self, worker: MeshCoreWorker) -> None: + """Test exponential backoff timing.""" + # Set shorter delays for testing + worker.config.meshcore.message_retry_delay = 0.1 + + # Mock MeshCore commands + mock_result = MagicMock() + mock_result.payload = {"expected_ack": "test_ack", "suggested_timeout": 5000} + + worker.meshcore = MagicMock() + worker.meshcore.commands = MagicMock() + worker.meshcore.commands.send_msg = AsyncMock(return_value=mock_result) + + # Track timing + call_times = [] + + async def mock_send_msg(*args: str) -> MagicMock: + call_times.append(asyncio.get_event_loop().time()) + return mock_result + + worker.meshcore.commands.send_msg = mock_send_msg + + # Mock ack never received + with patch.object(worker, "_wait_for_ack", return_value=False): + with patch.object(worker, "_reset_path", new_callable=AsyncMock): + await worker._send_msg_with_retry("destination", "message") + + # Check exponential backoff timing + assert len(call_times) == 4 # Initial + 3 retries + if len(call_times) > 1: + # First retry after base_delay (0.1s) + assert 0.05 < (call_times[1] - call_times[0]) < 0.2 + if len(call_times) > 2: + # Second retry after base_delay * 2 (0.2s) + assert 0.15 < (call_times[2] - call_times[1]) < 0.3 + if len(call_times) > 3: + # Third attempt happens immediately after path reset (no delay) + assert (call_times[3] - call_times[2]) < 0.1