From 164539597cd7fb3bafe38b628c2a528c55a8e76f Mon Sep 17 00:00:00 2001 From: agessaman Date: Tue, 14 Jul 2026 17:55:52 -0700 Subject: [PATCH] fix(companion): keep restarted queue rows single-owner --- repeater/main.py | 57 ++-------------------- tests/test_companion_bridge_frame_utils.py | 34 +++++++++++++ tests/test_companion_state_load.py | 49 ++----------------- 3 files changed, 42 insertions(+), 98 deletions(-) diff --git a/repeater/main.py b/repeater/main.py index f573ef8..3bcd9c8 100644 --- a/repeater/main.py +++ b/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: diff --git a/tests/test_companion_bridge_frame_utils.py b/tests/test_companion_bridge_frame_utils.py index 2077fd8..d951a13 100644 --- a/tests/test_companion_bridge_frame_utils.py +++ b/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) diff --git a/tests/test_companion_state_load.py b/tests/test_companion_state_load.py index f6f3c06..815e7cc 100644 --- a/tests/test_companion_state_load.py +++ b/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