From ef234f143b35f1167f3a5d3b48f438a2ba42c53e Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 16 Jul 2026 20:28:13 -0700 Subject: [PATCH] fix(room_server): hold the global transmit lock across the whole send GlobalRateLimiter.acquire() enforced the inter-message gap inside an async-with that released the lock before returning, so the caller's radio transmission ran with no lock held and every room's sync loop could push concurrently. acquire() now returns with the lock held (released via release() in a finally, including on cancellation during the gap wait), uses a monotonic clock so wall-clock jumps cannot skip or stretch the gap, and push_post_to_client acquires it only around the sync-state write plus the blocking send, so the ACK-timeout clock starts when the transmission actually begins. Regression tests cover two rooms sharing one limiter (no concurrent sends, gap enforced) and lock release when the send raises. --- repeater/handler_helpers/room_server.py | 89 ++++++++++++------- tests/test_handler_helpers_room_server.py | 103 ++++++++++++++++++++++ 2 files changed, 158 insertions(+), 34 deletions(-) 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()