Merge pull request #363 from agessaman/fix/all-the-things

repeater/handler_helpers/acl.py

@@ -64,7 +64,10 @@ class ACL:
         sync_since: int = None,
     ) -> None:
         now = int(time.time())
-        client.last_timestamp = timestamp
+        # Monotonic: the replay watermark must never move backwards, even if
+        # another accepted request advanced it between the replay check and
+        # this write.
+        client.last_timestamp = max(client.last_timestamp, timestamp)
         client.last_activity = now
         client.last_login_success = now
         client.shared_secret = shared_secret

repeater/handler_helpers/room_server.py

@@ -27,7 +27,10 @@ PUSH_TIMEOUT_BASE_MS = 4000
 PUSH_ACK_TIMEOUT_FACTOR_MS = 2000
 
 # Safety limits and protections
-MAX_MESSAGE_LENGTH = 160  # Match C++ MAX_POST_TEXT_LEN (151 bytes for text)
+# Match C++ MAX_POST_TEXT_LEN from examples/simple_room_server/MyMesh.h:
+# #define MAX_POST_TEXT_LEN (160-9) -- 160-byte encrypted text budget minus the
+# 9-byte prefix (4-byte timestamp + 1-byte flags/attempt + 4-byte author pubkey prefix).
+MAX_POST_TEXT_LEN = 151
 MAX_POSTS_PER_CLIENT_PER_MINUTE = 10  # Prevent spam
 MAX_CLIENTS_PER_ROOM = 50  # From ACL default
 MAX_PUSH_FAILURES = 3  # Evict after this many consecutive failures
@@ -47,27 +50,52 @@ _global_push_lock = asyncio.Lock()
 GLOBAL_MIN_GAP_BETWEEN_MESSAGES = 1.1  # 1.1s  minimum gap between transmissions
 
 
+def _truncate_utf8(text: str, max_bytes: int) -> str:
+    """Truncate ``text`` to at most ``max_bytes`` of its UTF-8 encoding.
+
+    Cuts at a codepoint boundary so a partial multi-byte sequence is never
+    emitted (matches firmware's byte-length text budget, but stays
+    UTF-8-safe rather than the firmware's raw ``strncpy``).
+    """
+    encoded = text.encode("utf-8")
+    if len(encoded) <= max_bytes:
+        return text
+    return encoded[:max_bytes].decode("utf-8", errors="ignore")
+
+
 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:
@@ -234,13 +262,15 @@ class RoomServer:
     ) -> bool:
 
         try:
-            # SAFETY: Validate message length
-            if len(message_text) > MAX_MESSAGE_LENGTH:
+            # SAFETY: Validate message length (byte length, matching firmware's
+            # MAX_POST_TEXT_LEN text budget), cutting at a UTF-8 codepoint boundary.
+            encoded_len = len(message_text.encode("utf-8"))
+            if encoded_len > MAX_POST_TEXT_LEN:
                 logger.warning(
                     f"Room '{self.room_name}': Message from {client_pubkey[:4].hex()} "
-                    f"exceeds max length ({len(message_text)} > {MAX_MESSAGE_LENGTH}), truncating"
+                    f"exceeds max length ({encoded_len} > {MAX_POST_TEXT_LEN} bytes), truncating"
                 )
-                message_text = message_text[:MAX_MESSAGE_LENGTH]
+                message_text = _truncate_utf8(message_text, MAX_POST_TEXT_LEN)
 
             # SAFETY: Rate limit per client
             client_key = client_pubkey.hex()
@@ -320,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}",
@@ -353,16 +379,26 @@ class RoomServer:
                         )
                         return False  # Skip this client for now
 
-            # Build message payload
-            timestamp = int(time.time())
-            flags = TXT_TYPE_SIGNED_PLAIN << 2  # Include author prefix
+            # Build message payload.
+            # Timestamp is the post's own stored timestamp, not the delivery
+            # time (MyMesh.cpp:56 serializes the stored post's timestamp).
+            timestamp = int(post["post_timestamp"])
+            # Low 2 bits carry a random attempt nonce so retries of the same
+            # post get a different packet hash/ACK (MyMesh.cpp:59-60).
+            flags = (TXT_TYPE_SIGNED_PLAIN << 2) | (secrets.randbits(8) & 0x03)
 
             # Author prefix (first 4 bytes of pubkey)
             author_pubkey = bytes.fromhex(post["author_pubkey"])
             author_prefix = author_pubkey[:4]
 
             # Plaintext: timestamp(4) + flags(1) + author_prefix(4) + text
-            message_bytes = post["message_text"].encode("utf-8")
+            # SAFETY: Clamp at push time too, in case a legacy stored post
+            # (from before this limit was enforced on write) exceeds
+            # MAX_POST_TEXT_LEN bytes -- truncate at a UTF-8 codepoint boundary.
+            message_text = post["message_text"]
+            if len(message_text.encode("utf-8")) > MAX_POST_TEXT_LEN:
+                message_text = _truncate_utf8(message_text, MAX_POST_TEXT_LEN)
+            message_bytes = message_text.encode("utf-8")
             plaintext = (
                 timestamp.to_bytes(4, "little") + bytes([flags]) + author_prefix + message_bytes
             )
@@ -412,34 +448,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

repeater/main.py

@@ -377,8 +377,6 @@ class RepeaterDaemon:
             # Load additional identities from config (e.g., room servers)
             await self._load_additional_identities()
 
-            self.dispatcher._is_own_packet = lambda pkt: False
-
             self.repeater_handler = RepeaterHandler(
                 self.config,
                 self.dispatcher,

tests/test_engine.py

@@ -692,8 +692,10 @@ class TestForwardRoutedAck:
 
 # Literal MeshCore-format routed-ACK frames (Packet::writeTo layout).  MeshCore's
 # routeDirectRecvAcks re-emits through createAck: the header is rebuilt from
-# scratch (payload type ACK, route DIRECT, version bits cleared, transport codes
-# dropped) and this node is removed from the path.  Independent wire fixtures —
+# scratch (payload type ACK, route DIRECT, transport codes dropped) and this
+# node is removed from the path.  Non-zero-version inputs are not represented
+# here: firmware's Dispatcher::tryParsePacket (and our Packet.read_from) drops
+# them at parse, so they can never reach the relay.  Independent wire fixtures —
 # do NOT rebuild them through Packet.write_to()/PathUtils.
 FIRMWARE_ROUTED_ACK_VECTORS = (
     # header 0x0E = ACK<<2 | ROUTE_DIRECT; path_len 0x02 = 1-byte/2-hop;
@@ -719,13 +721,6 @@ FIRMWARE_ROUTED_ACK_VECTORS = (
         bytes.fromhex("0E01CC4DABAF95"),
         id="transport-direct-1-byte-2-hop",
     ),
-    # header 0x4E = version bit set on a direct ACK; createAck clears it.
-    pytest.param(
-        bytes([0xAB]),
-        bytes.fromhex("4E02ABCC4DABAF95"),
-        bytes.fromhex("0E01CC4DABAF95"),
-        id="direct-version-bits-cleared",
-    ),
 )
 
 

tests/test_handler_helpers_acl_advert.py

@@ -142,6 +142,29 @@ def test_acl_admin_login_sets_client_state_and_replay_protection():
     assert replay_perms == 0
 
 
+def test_acl_touch_client_session_watermark_is_monotonic():
+    """The replay watermark never moves backwards: if another accepted request
+    advanced it between the replay check and the session touch, an older (but
+    already-validated) timestamp must not regress it."""
+    identity = _FakeIdentity(b"B" * 32)
+    acl = ACL(max_clients=5, admin_password="top-secret", guest_password="guest")
+
+    ok, _ = acl.authenticate_client(
+        client_identity=identity,
+        shared_secret=b"k" * 32,
+        password="top-secret",
+        timestamp=2000,
+    )
+    assert ok is True
+    client = acl.get_client(b"B" * 40)
+
+    acl._touch_client_session(client, b"k" * 32, timestamp=1500)
+    assert client.last_timestamp == 2000
+
+    acl._touch_client_session(client, b"k" * 32, timestamp=2500)
+    assert client.last_timestamp == 2500
+
+
 def test_acl_max_clients_invalid_password_and_remove_client_paths():
     acl = ACL(max_clients=1, admin_password="a", guest_password="g")
     id_a = _FakeIdentity(b"C" * 32)

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
@@ -5,10 +6,13 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from repeater.handler_helpers.room_server import (
+    MAX_POST_TEXT_LEN,
     MAX_UNSYNCED_POSTS,
     TXT_TYPE_PLAIN,
     TXT_TYPE_SIGNED_PLAIN,
+    GlobalRateLimiter,
     RoomServer,
+    _truncate_utf8,
 )
 
 
@@ -91,7 +95,7 @@ async def test_room_server_add_post_truncates_and_rate_limits_client():
     first_ok = await rs.add_post(client_key, long_msg, sender_timestamp=5)
     assert first_ok is True
     args = db.insert_room_message.call_args.kwargs
-    assert len(args["message_text"]) == 160
+    assert len(args["message_text"].encode("utf-8")) == MAX_POST_TEXT_LEN
 
     # Force client to appear at post-per-minute limit.
     rs.client_post_times[client_key.hex()] = [time.time() - 1] * 10
@@ -99,6 +103,71 @@ async def test_room_server_add_post_truncates_and_rate_limits_client():
     assert second_ok is False
 
 
+@pytest.mark.asyncio
+async def test_room_server_add_post_ascii_over_limit_stores_exactly_max_bytes():
+    db = _FakeDB()
+    rs = _make_room_server(db=db)
+
+    long_msg = "a" * 200  # ASCII: 1 byte per char, well over MAX_POST_TEXT_LEN
+    ok = await rs.add_post(b"G" * 32, long_msg, sender_timestamp=1)
+    assert ok is True
+
+    stored = db.insert_room_message.call_args.kwargs["message_text"]
+    assert len(stored.encode("utf-8")) == MAX_POST_TEXT_LEN
+    assert stored == "a" * MAX_POST_TEXT_LEN
+
+
+@pytest.mark.asyncio
+async def test_room_server_add_post_multibyte_utf8_truncates_on_codepoint_boundary():
+    db = _FakeDB()
+    rs = _make_room_server(db=db)
+
+    # Each emoji is 4 bytes in UTF-8; padding forces the cut to land mid-emoji
+    # if truncation were byte-naive instead of codepoint-aware.
+    padding = "a" * (MAX_POST_TEXT_LEN - 2)
+    msg = padding + "\U0001f600\U0001f600\U0001f600"  # grinning face emoji x3
+    assert len(msg.encode("utf-8")) > MAX_POST_TEXT_LEN
+
+    ok = await rs.add_post(b"H" * 32, msg, sender_timestamp=2)
+    assert ok is True
+
+    stored = db.insert_room_message.call_args.kwargs["message_text"]
+    encoded = stored.encode("utf-8")
+    assert len(encoded) <= MAX_POST_TEXT_LEN
+    # Must decode cleanly (no partial multi-byte sequence) and round-trip.
+    assert encoded.decode("utf-8") == stored
+
+
+@pytest.mark.asyncio
+async def test_room_server_add_post_exact_limit_text_untouched():
+    db = _FakeDB()
+    rs = _make_room_server(db=db)
+
+    msg = "y" * MAX_POST_TEXT_LEN
+    ok = await rs.add_post(b"I" * 32, msg, sender_timestamp=3)
+    assert ok is True
+
+    stored = db.insert_room_message.call_args.kwargs["message_text"]
+    assert stored == msg
+    assert len(stored.encode("utf-8")) == MAX_POST_TEXT_LEN
+
+
+def test_truncate_utf8_helper_boundary_cases():
+    # Under the limit: untouched.
+    assert _truncate_utf8("short", 151) == "short"
+
+    # Exactly at the limit: untouched.
+    exact = "z" * 151
+    assert _truncate_utf8(exact, 151) == exact
+
+    # Multi-byte straddling the boundary: cuts cleanly, decodes, stays <= limit.
+    text = ("b" * 149) + "\U0001f600\U0001f600"  # 149 + 4 + 4 = 157 bytes
+    result = _truncate_utf8(text, 151)
+    encoded = result.encode("utf-8")
+    assert len(encoded) <= 151
+    assert encoded.decode("utf-8") == result
+
+
 @pytest.mark.asyncio
 async def test_room_server_add_post_returns_false_on_db_insert_failure():
     db = _FakeDB()
@@ -192,6 +261,47 @@ async def test_room_server_push_post_to_client_success_direct_route_sets_path_an
     rs.global_limiter.release.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_room_server_push_post_to_client_clamps_oversized_legacy_stored_text():
+    """A post stored before the write-time limit was enforced (or otherwise
+    over MAX_POST_TEXT_LEN bytes) must still be clamped at push time so the
+    outgoing frame's text portion never exceeds the firmware's budget."""
+    db = _FakeDB()
+    db.get_client_sync.return_value = {"push_failures": 0}
+    injector = AsyncMock(return_value=True)
+    rs = _make_room_server(db=db, injector=injector)
+    rs.global_limiter = SimpleNamespace(acquire=AsyncMock(), release=MagicMock())
+    rs._handle_ack_received = AsyncMock()
+
+    client = _FakeClient(pubkey=b"E" * 32, out_path=b"\xaa\xbb", out_path_len=2)
+    oversized_text = "q" * (MAX_POST_TEXT_LEN + 50)
+    post = {
+        "author_pubkey": (b"F" * 32).hex(),
+        "message_text": oversized_text,
+        "post_timestamp": 1234.5,
+    }
+
+    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,
+        ) as create_datagram,
+    ):
+        ok = await rs.push_post_to_client(client, post)
+
+    assert ok is True
+    plaintext = create_datagram.call_args.kwargs["plaintext"]
+    # Prefix is timestamp(4) + flags(1) + author_prefix(4) = 9 bytes.
+    text_portion = plaintext[9:]
+    assert len(text_portion) == MAX_POST_TEXT_LEN
+    assert text_portion.decode("utf-8") == "q" * MAX_POST_TEXT_LEN
+
+
 @pytest.mark.asyncio
 async def test_room_server_push_expected_ack_matches_firmware_signed_ack():
     """The pending ACK CRC must match what a signed-plain receiver sends back.
@@ -273,6 +383,123 @@ async def test_room_server_push_expected_ack_matches_firmware_signed_ack():
     assert client_ack_crc == expected_ack_crc
 
 
+@pytest.mark.asyncio
+async def test_push_serializes_stored_post_timestamp_not_walltime():
+    """The pushed frame's timestamp must be the post's own stored timestamp,
+    not the delivery wall-clock time (MyMesh.cpp:56 serializes the stored
+    post's timestamp, not `now`)."""
+    db = _FakeDB()
+    db.get_client_sync.return_value = {"push_failures": 0}
+    injector = AsyncMock(return_value=True)
+    rs = _make_room_server(db=db, injector=injector)
+    rs.global_limiter = SimpleNamespace(acquire=AsyncMock(), release=MagicMock())
+    rs._handle_ack_received = AsyncMock()
+
+    client = _FakeClient(pubkey=b"E" * 32)
+    post = {
+        "author_pubkey": (b"F" * 32).hex(),
+        "message_text": "payload",
+        "post_timestamp": 1000.0,
+    }
+
+    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,
+        ) as create_datagram,
+    ):
+        ok = await rs.push_post_to_client(client, post)
+
+    assert ok is True
+    plaintext = create_datagram.call_args.kwargs["plaintext"]
+    assert int.from_bytes(plaintext[0:4], "little") == 1000
+
+
+@pytest.mark.asyncio
+async def test_push_attempt_bits_randomized_across_retries():
+    """Each push must OR a fresh random byte's low 2 bits into flags so a
+    retried post gets a different packet hash (and ACK), per MyMesh.cpp:59-60
+    ("so packet hash (and ACK) will be different")."""
+    db = _FakeDB()
+    db.get_client_sync.return_value = {"push_failures": 0}
+    injector = AsyncMock(return_value=True)
+    rs = _make_room_server(db=db, injector=injector)
+    rs.global_limiter = SimpleNamespace(acquire=AsyncMock(), release=MagicMock())
+    rs._handle_ack_received = AsyncMock()
+
+    client = _FakeClient(pubkey=b"E" * 32)
+    post = {
+        "author_pubkey": (b"F" * 32).hex(),
+        "message_text": "payload",
+        "post_timestamp": 1234.5,
+    }
+
+    plaintexts = []
+    acks = []
+    with patch("repeater.handler_helpers.room_server.secrets.randbits", side_effect=[1, 2]):
+        for _ in range(2):
+            packet = SimpleNamespace(path=bytearray(), path_len=0)
+            with patch(
+                "repeater.handler_helpers.room_server.PacketBuilder.create_datagram",
+                return_value=packet,
+            ) as create_datagram:
+                ok = await rs.push_post_to_client(client, post)
+            assert ok is True
+            plaintexts.append(create_datagram.call_args.kwargs["plaintext"])
+            acks.append(db.upsert_client_sync.call_args.kwargs["pending_ack_crc"])
+
+    flags = [pt[4] for pt in plaintexts]
+    assert (flags[0] >> 2) & 0x3F == TXT_TYPE_SIGNED_PLAIN
+    assert (flags[1] >> 2) & 0x3F == TXT_TYPE_SIGNED_PLAIN
+    assert flags[0] & 0x03 != flags[1] & 0x03
+    assert plaintexts[0][0:4] == plaintexts[1][0:4]  # timestamp unchanged
+    assert plaintexts[0][5:] == plaintexts[1][5:]  # author prefix + text unchanged
+    assert acks[0] != acks[1]
+
+
+@pytest.mark.asyncio
+async def test_push_watermark_equals_serialized_timestamp():
+    """The db sync watermark update must use the same timestamp value that
+    was serialized into the pushed plaintext."""
+    db = _FakeDB()
+    db.get_client_sync.return_value = {"push_failures": 0}
+    injector = AsyncMock(return_value=True)
+    rs = _make_room_server(db=db, injector=injector)
+    rs.global_limiter = SimpleNamespace(acquire=AsyncMock(), release=MagicMock())
+    rs._handle_ack_received = AsyncMock()
+
+    client = _FakeClient(pubkey=b"E" * 32)
+    post = {
+        "author_pubkey": (b"F" * 32).hex(),
+        "message_text": "payload",
+        "post_timestamp": 555.0,
+    }
+
+    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,
+        ) as create_datagram,
+    ):
+        ok = await rs.push_post_to_client(client, post)
+
+    assert ok is True
+    plaintext = create_datagram.call_args.kwargs["plaintext"]
+    serialized_ts = int.from_bytes(plaintext[0:4], "little")
+    watermark = db.upsert_client_sync.call_args.kwargs["push_post_timestamp"]
+    assert int(watermark) == serialized_ts
+
+
 @pytest.mark.asyncio
 async def test_room_server_push_post_to_client_backoff_skip_and_timeout_path():
     db = _FakeDB()
@@ -401,3 +628,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()

tests/test_path_hash_mode_advert.py

@@ -34,8 +34,8 @@ class MockRadio:
 def _make_advert_packet():
     """Build a 0-hop flood ADVERT packet (same shape as PacketBuilder.create_advert)."""
     pkt = Packet()
-    # Version 1, ROUTE_TYPE_FLOOD, PAYLOAD_TYPE_ADVERT
-    pkt.header = (1 << 6) | (PAYLOAD_TYPE_ADVERT << PH_TYPE_SHIFT) | ROUTE_TYPE_FLOOD
+    # Version 0 (the only version firmware accepts), ROUTE_TYPE_FLOOD, PAYLOAD_TYPE_ADVERT
+    pkt.header = (PAYLOAD_TYPE_ADVERT << PH_TYPE_SHIFT) | ROUTE_TYPE_FLOOD
     pkt.path_len = 0
     pkt.path = bytearray()
     pkt.payload = bytearray(b"minimal_advert_payload")