fix(companion): keep restarted queue rows single-owner

repeater/main.py

@@ -1,6 +1,5 @@
 import asyncio
 import functools
-import inspect
 import logging
 import os
 import signal
@@ -731,14 +730,14 @@ class RepeaterDaemon:
     async def _restore_companion_state(
         self, sqlite_handler, bridge, companion_hash_str: str, name: str
     ) -> None:
-        """Restore persisted contacts/channels/messages from SQLite into a bridge.
+        """Restore persisted contacts and channels from SQLite into a bridge.
 
         Each load is cross-checked against the table's row count for this
         companion and retried once on mismatch; raises CompanionStateLoadError
         when persisted rows exist but cannot be loaded, so the companion fails
         init loudly instead of starting with an empty store.
         """
-        from openhop_core.companion.models import Channel, QueuedMessage
+        from openhop_core.companion.models import Channel
 
         contact_rows, contact_count = await _load_companion_rows_verified(
             sqlite_handler.companion_load_contacts,
@@ -790,63 +789,15 @@ class RepeaterDaemon:
                     row.get("name", ""),
                 )
 
-        # Preload queued messages, bounded by offline_queue_size (0 disables
-        # offline storage entirely).
-        loaded_messages = 0
-        message_count = 0
-        retention = getattr(bridge.message_queue, "max_size", None)
-        if retention != 0:
-            message_rows, message_count = await _load_companion_rows_verified(
-                sqlite_handler.companion_load_messages,
-                sqlite_handler.companion_count_messages,
-                "messages",
-                companion_hash_str,
-                name,
-                limit=retention or 100,
-            )
-            loaded_messages = len(message_rows)
-            # openhop_core < the sender_prefix change (paired with fd43d86) has no
-            # QueuedMessage.sender_prefix; drop the prefix there instead of failing init.
-            supports_sender_prefix = "sender_prefix" in inspect.signature(QueuedMessage).parameters
-            if message_rows and not supports_sender_prefix:
-                logger.warning(
-                    "Companion %s ('%s'): installed openhop_core QueuedMessage has no "
-                    "sender_prefix field; persisted sender prefixes will be dropped "
-                    "(update openhop_core to restore signed room-post authors)",
-                    companion_hash_str,
-                    name,
-                )
-            for msg_dict in message_rows:
-                sk = msg_dict.get("sender_key", b"")
-                if isinstance(sk, str):
-                    sk = bytes.fromhex(sk)
-                sp = msg_dict.get("sender_prefix", b"")
-                if isinstance(sp, str):
-                    sp = bytes.fromhex(sp) if sp else b""
-                msg_kwargs = dict(
-                    sender_key=sk,
-                    txt_type=msg_dict.get("txt_type", 0),
-                    timestamp=msg_dict.get("timestamp", 0),
-                    text=msg_dict.get("text", ""),
-                    is_channel=bool(msg_dict.get("is_channel", False)),
-                    channel_idx=msg_dict.get("channel_idx", 0),
-                    path_len=msg_dict.get("path_len", 0),
-                )
-                if supports_sender_prefix:
-                    msg_kwargs["sender_prefix"] = sp
-                bridge.message_queue.push(QueuedMessage(**msg_kwargs))
-
         logger.info(
-            "Companion %s ('%s'): restored %d/%d contact(s), %d/%d channel(s), "
-            "%d/%d message(s) from SQLite",
+            "Companion %s ('%s'): restored %d/%d contact(s), %d/%d channel(s); "
+            "queued messages remain in SQLite",
             companion_hash_str,
             name,
             len(contact_rows),
             contact_count,
             len(channel_rows),
             channel_count,
-            loaded_messages,
-            message_count,
         )
 
     async def add_companion_from_config(self, comp_config: dict) -> None:

tests/test_companion_bridge_frame_utils.py

@@ -201,6 +201,40 @@ async def test_frame_server_no_more_messages_response_when_empty():
         assert srv._write_frame.call_args[0][0] == bytes([RESP_CODE_NO_MORE_MESSAGES])
 
 
+@pytest.mark.asyncio
+async def test_restart_queue_rows_are_delivered_once_from_sqlite():
+    sqlite = SimpleNamespace(
+        companion_pop_message=MagicMock(
+            side_effect=[
+                {"sender_key": b"a", "timestamp": 1, "text": "first"},
+                {"sender_key": b"b", "timestamp": 2, "text": "second"},
+                None,
+            ]
+        )
+    )
+    bridge = SimpleNamespace(sync_next_message=lambda: None)
+
+    with patch(
+        "repeater.companion.frame_server._BaseFrameServer.__init__", lambda self, **kwargs: None
+    ):
+        srv = CompanionFrameServer(bridge=bridge, companion_hash="h", sqlite_handler=sqlite)
+        srv.bridge = bridge
+        srv.companion_hash = "h"
+        srv._write_frame = MagicMock()
+        srv._build_message_frame = lambda message: message.text.encode()
+
+        await srv._cmd_sync_next_message(b"")
+        await srv._cmd_sync_next_message(b"")
+        await srv._cmd_sync_next_message(b"")
+
+    assert [call.args[0] for call in srv._write_frame.call_args_list] == [
+        b"first",
+        b"second",
+        bytes([RESP_CODE_NO_MORE_MESSAGES]),
+    ]
+    assert sqlite.companion_pop_message.call_count == 3
+
+
 @pytest.mark.asyncio
 async def test_rejected_queue_callback_skips_sqlite_persistence_but_notifies_client():
     server = object.__new__(CompanionFrameServer)

tests/test_companion_state_load.py

@@ -94,7 +94,7 @@ class TestRestoreCompanionState:
         return RepeaterDaemon({"repeater": {"node_name": "n"}, "logging": {}}, radio=object())
 
     @pytest.mark.asyncio
-    async def test_restores_all_state(self):
+    async def test_restores_contacts_and_channels_without_preloading_messages(self):
         daemon = self._daemon()
         bridge = self._bridge()
         sqlite = self._sqlite(
@@ -106,7 +106,9 @@ class TestRestoreCompanionState:
         bridge.contacts.load_from_dicts.assert_called_once()
         bridge.channels.set.assert_called_once()
         assert bridge.channels.set.call_args[0][0] == 1
-        bridge.message_queue.push.assert_called_once()
+        sqlite.companion_load_messages.assert_not_called()
+        sqlite.companion_count_messages.assert_not_called()
+        bridge.message_queue.push.assert_not_called()
 
     @pytest.mark.asyncio
     async def test_channel_load_failure_raises_before_bridge_touch(self):
@@ -130,49 +132,6 @@ class TestRestoreCompanionState:
             await daemon._restore_companion_state(sqlite, bridge, _HASH, _NAME)
         assert any("rejected persisted channel" in r.message for r in caplog.records)
 
-    @pytest.mark.asyncio
-    async def test_old_core_without_sender_prefix_drops_prefix(self, caplog):
-        # openhop_core releases before the sender_prefix change reject the kwarg;
-        # message restore must degrade (drop the prefix) instead of failing init.
-        from dataclasses import dataclass, field
-
-        @dataclass
-        class LegacyQueuedMessage:
-            sender_key: bytes = b""
-            txt_type: int = 0
-            timestamp: int = 0
-            text: str = ""
-            is_channel: bool = False
-            channel_idx: int = 0
-            path_len: int = 0
-            snr: float = 0.0
-            rssi: int = 0
-            channel_data_payload: bytes = field(default=b"")
-
-        daemon = self._daemon()
-        bridge = self._bridge()
-        sqlite = self._sqlite(
-            messages=[{"sender_key": b"", "text": "hi", "sender_prefix": b"\xab\xcd"}]
-        )
-        with (
-            patch("openhop_core.companion.models.QueuedMessage", LegacyQueuedMessage),
-            caplog.at_level(logging.WARNING),
-        ):
-            await daemon._restore_companion_state(sqlite, bridge, _HASH, _NAME)
-        bridge.message_queue.push.assert_called_once()
-        pushed = bridge.message_queue.push.call_args[0][0]
-        assert isinstance(pushed, LegacyQueuedMessage)
-        assert pushed.text == "hi"
-        assert any("sender prefixes will be dropped" in r.message for r in caplog.records)
-
-    @pytest.mark.asyncio
-    async def test_zero_retention_skips_message_load(self):
-        daemon = self._daemon()
-        bridge = self._bridge(max_size=0)
-        sqlite = self._sqlite()
-        await daemon._restore_companion_state(sqlite, bridge, _HASH, _NAME)
-        sqlite.companion_load_messages.assert_not_called()
-
 
 class TestCompanionInitSurfacesLoadFailure:
     @staticmethod