diff --git a/repeater/handler_helpers/room_server.py b/repeater/handler_helpers/room_server.py index 3680dbc..eda2e66 100644 --- a/repeater/handler_helpers/room_server.py +++ b/repeater/handler_helpers/room_server.py @@ -67,23 +67,35 @@ class GlobalRateLimiter: def __init__(self, min_gap_seconds: float = 0.1): self.min_gap = min_gap_seconds # Minimum gap between consecutive messages self.lock = asyncio.Lock() # Only one transmission at a time - self.last_release_time = 0 + self.last_release_time = 0.0 async def acquire(self): + """Acquire the global transmit lock and enforce the inter-message gap. - async with self.lock: - # Enforce minimum gap between consecutive transmissions - now = time.time() - time_since_last = now - self.last_release_time + The lock is **held on return** and must be released with ``release()`` + (call it in a ``finally``). The previous implementation released the + lock inside an ``async with`` before returning, so the caller's radio + transmission ran with no lock held and multiple room loops could push + concurrently. A monotonic clock is used so a wall-clock jump cannot + skip or extend the gap. + """ + await self.lock.acquire() + try: + time_since_last = time.monotonic() - self.last_release_time if time_since_last < self.min_gap: wait_time = self.min_gap - time_since_last logger.debug(f"Global rate limiter: waiting {wait_time * 1000:.0f}ms") await asyncio.sleep(wait_time) - # Lock is now held - caller can transmit - # Will be released when context exits + except BaseException: + # Never leak the lock if the gap wait is cancelled. + self.lock.release() + raise def release(self): - self.last_release_time = time.time() + """Record the transmission time and release the transmit lock.""" + self.last_release_time = time.monotonic() + if self.lock.locked(): + self.lock.release() class RoomServer: @@ -338,10 +350,6 @@ class RoomServer: async def push_post_to_client(self, client_info, post: Dict) -> bool: try: - # SAFETY: Global transmission lock - only ONE message on radio at a time - # This is critical because LoRa is serial (0.5-9s airtime per message) - await self.global_limiter.acquire() - # SAFETY: Check client failure backoff sync_state = self.db.get_client_sync( room_hash=f"0x{self.room_hash:02X}", @@ -436,34 +444,47 @@ class RoomServer: PUSH_TIMEOUT_BASE_MS + PUSH_ACK_TIMEOUT_FACTOR_MS * (path_len + 1) ) / 1000.0 - # Update client sync state with pending ACK current_sync_since = ( sync_state.get("sync_since", 0) if sync_state else getattr(client_info, "sync_since", 0) ) - self.db.upsert_client_sync( - room_hash=f"0x{self.room_hash:02X}", - client_pubkey=client_info.id.get_public_key().hex(), - sync_since=current_sync_since, - pending_ack_crc=expected_ack_crc, - push_post_timestamp=post["post_timestamp"], - ack_timeout_time=time.time() + ack_timeout, - last_activity=time.time(), - ) - # Send and wait for the client's delivery ACK. The injector must be - # told the crypto ACK CRC we computed above — its default - # (packet.get_crc()) is a packet-hash CRC no client ever sends. - # This blocks for the entire transmission duration (0.5-9 seconds) - success = await self.packet_injector( - packet, - wait_for_ack=True, - expected_crc=expected_ack_crc, - ack_timeout_s=ack_timeout, - ) - # SAFETY: Release transmission lock AFTER send completes - self.global_limiter.release() + # SAFETY: Global transmission lock - only ONE message on the radio at + # a time, enforced across the entire send. LoRa is serial (0.5-9s + # airtime per message), and every room has its own sync loop sharing + # this limiter, so the lock must be held from before we start the ACK + # timeout clock through the blocking send. Released in `finally` so + # backoff/exception paths above (which returned before acquiring) do + # not, and the send path always does, free it exactly once. + await self.global_limiter.acquire() + try: + # Update client sync state with pending ACK. Done under the lock + # so the ACK-timeout clock starts when the send actually begins, + # not before the inter-message gap has elapsed. + self.db.upsert_client_sync( + room_hash=f"0x{self.room_hash:02X}", + client_pubkey=client_info.id.get_public_key().hex(), + sync_since=current_sync_since, + pending_ack_crc=expected_ack_crc, + push_post_timestamp=post["post_timestamp"], + ack_timeout_time=time.time() + ack_timeout, + last_activity=time.time(), + ) + # Send and wait for the client's delivery ACK. The injector must + # be told the crypto ACK CRC we computed above — its default + # (packet.get_crc()) is a packet-hash CRC no client ever sends. + # This blocks for the entire transmission duration (0.5-9 seconds) + success = await self.packet_injector( + packet, + wait_for_ack=True, + expected_crc=expected_ack_crc, + ack_timeout_s=ack_timeout, + ) + finally: + # SAFETY: Release the transmission lock on every path (ACK, + # timeout, or exception during the send). + self.global_limiter.release() if success: # ACK received! Update sync state diff --git a/tests/test_handler_helpers_room_server.py b/tests/test_handler_helpers_room_server.py index e1ee422..7908783 100644 --- a/tests/test_handler_helpers_room_server.py +++ b/tests/test_handler_helpers_room_server.py @@ -1,3 +1,4 @@ +import asyncio import time from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -9,6 +10,7 @@ from repeater.handler_helpers.room_server import ( MAX_UNSYNCED_POSTS, TXT_TYPE_PLAIN, TXT_TYPE_SIGNED_PLAIN, + GlobalRateLimiter, RoomServer, _truncate_utf8, ) @@ -509,3 +511,104 @@ async def test_room_server_start_and_stop_are_idempotent(): # Ensure task is cleaned up. if first_task: assert first_task.cancelled() or first_task.done() + + +def _push_post(author=b"F" * 32): + return { + "author_pubkey": author.hex(), + "message_text": "payload", + "post_timestamp": 1234.5, + } + + +@pytest.mark.asyncio +async def test_global_rate_limiter_serializes_concurrent_room_pushes(): + """Two room loops sharing the limiter must not transmit concurrently, and + the inter-message gap must be enforced between their sends. + + Regression: acquire() used to release the lock before returning, so + both pushes ran their radio send with no lock held. + """ + limiter = GlobalRateLimiter(min_gap_seconds=0.05) + + active = 0 + max_active = 0 + starts = [] + + async def injector(packet, wait_for_ack=True, expected_crc=0, ack_timeout_s=0.0): + nonlocal active, max_active + starts.append(time.monotonic()) + active += 1 + max_active = max(max_active, active) + try: + await asyncio.sleep(0.02) # simulate airtime + finally: + active -= 1 + return True + + rooms = [] + for i in range(2): + db = _FakeDB() + db.get_client_sync.return_value = {"push_failures": 0} + rs = _make_room_server(db=db, injector=injector) + rs.global_limiter = limiter # share one real limiter across both rooms + rs._handle_ack_received = AsyncMock() + rooms.append(rs) + + client = _FakeClient(pubkey=b"E" * 32) + packet = SimpleNamespace(path=bytearray(), path_len=0) + with ( + patch( + "repeater.handler_helpers.room_server.CryptoUtils.sha256", + return_value=b"\x01\x02\x03\x04abcd", + ), + patch( + "repeater.handler_helpers.room_server.PacketBuilder.create_datagram", + return_value=packet, + ), + ): + results = await asyncio.gather( + *(rs.push_post_to_client(client, _push_post()) for rs in rooms) + ) + + assert results == [True, True] + assert max_active == 1 # never two transmissions in flight at once + assert len(starts) == 2 + # The second send started at least min_gap after the first began. + assert starts[1] - starts[0] >= 0.05 + assert not limiter.lock.locked() # fully released at the end + + +@pytest.mark.asyncio +async def test_global_rate_limiter_releases_lock_on_send_exception(): + """If the radio send raises, the limiter lock must still be released so the + next push is not deadlocked.""" + limiter = GlobalRateLimiter(min_gap_seconds=0.0) + + async def failing_injector(packet, **kwargs): + raise RuntimeError("radio boom") + + db = _FakeDB() + db.get_client_sync.return_value = {"push_failures": 0} + rs = _make_room_server(db=db, injector=failing_injector) + rs.global_limiter = limiter + + client = _FakeClient(pubkey=b"E" * 32) + packet = SimpleNamespace(path=bytearray(), path_len=0) + with ( + patch( + "repeater.handler_helpers.room_server.CryptoUtils.sha256", + return_value=b"\x01\x02\x03\x04abcd", + ), + patch( + "repeater.handler_helpers.room_server.PacketBuilder.create_datagram", + return_value=packet, + ), + ): + ok = await rs.push_post_to_client(client, _push_post()) + + assert ok is False + assert not limiter.lock.locked() + # The lock is reusable immediately afterwards. + await limiter.acquire() + limiter.release()