diff --git a/repeater/companion/frame_server.py b/repeater/companion/frame_server.py index b92e5e5..4f4315a 100644 --- a/repeater/companion/frame_server.py +++ b/repeater/companion/frame_server.py @@ -76,13 +76,19 @@ class CompanionFrameServer(_BaseFrameServer): if retention == 0: self.bridge.message_queue.pop_last() return - await asyncio.to_thread( + persisted = await asyncio.to_thread( self.sqlite_handler.companion_push_message, self.companion_hash, msg_dict, retention, ) - self.bridge.message_queue.pop_last() + if persisted: + self.bridge.message_queue.pop_last() + else: + logger.debug( + "Companion %s: retaining message in memory after SQLite queue rejection", + self.companion_hash, + ) def _sync_next_from_persistence(self) -> Optional[QueuedMessage]: """Retrieve next message from SQLite when bridge queue is empty.""" diff --git a/repeater/data_acquisition/sqlite_handler.py b/repeater/data_acquisition/sqlite_handler.py index 7ac67fa..4a26380 100644 --- a/repeater/data_acquisition/sqlite_handler.py +++ b/repeater/data_acquisition/sqlite_handler.py @@ -3358,12 +3358,17 @@ class SQLiteHandler: previous SELECT + INSERT round-trip (two statements, two SD-card reads) with a single atomic statement. - When ``max_messages`` is set, the oldest rows beyond that retention limit - are trimmed after a successful insert (power-user ``offline_queue_size``). + When ``max_messages`` is set, capacity follows MeshCore's offline queue + policy: evict the oldest channel message first and never displace a + retained direct message. The insert and any eviction share one + transaction. - Returns True if inserted, False if the message was a duplicate (skipped). + Returns True if the message is retained, False if it is a duplicate or + the protected queue cannot make room for it. """ try: + if max_messages is not None and max_messages <= 0: + return False packet_hash = msg.get("packet_hash") or None if isinstance(packet_hash, bytes): packet_hash = packet_hash.decode("utf-8", errors="replace") if packet_hash else None @@ -3372,6 +3377,7 @@ class SQLiteHandler: if not isinstance(sender_prefix, str): sender_prefix = bytes(sender_prefix or b"").hex() with self._connect() as conn: + conn.execute("SAVEPOINT companion_message_push") cursor = conn.execute( """ INSERT OR IGNORE INTO companion_messages @@ -3394,22 +3400,49 @@ class SQLiteHandler: ), ) inserted = cursor.rowcount > 0 - if inserted and max_messages is not None: - # Keep the newest `max_messages` rows; drop older overflow. - conn.execute( - """ - DELETE FROM companion_messages - WHERE companion_hash = ? AND id NOT IN ( + if not inserted: + conn.execute("RELEASE SAVEPOINT companion_message_push") + conn.commit() + return False + if max_messages is not None: + count = conn.execute( + "SELECT COUNT(*) FROM companion_messages WHERE companion_hash = ?", + (companion_hash,), + ).fetchone()[0] + while count > max_messages: + oldest_channel = conn.execute( + """ SELECT id FROM companion_messages - WHERE companion_hash = ? - ORDER BY created_at DESC, id DESC - LIMIT ? + WHERE companion_hash = ? AND is_channel = 1 + ORDER BY created_at ASC, id ASC LIMIT 1 + """, + (companion_hash,), + ).fetchone() + if oldest_channel is None: + # The just-inserted row is not retainable without + # sacrificing a direct message. Keep every prior + # row intact, including any channel rows already + # considered while satisfying a lowered limit. + conn.execute("ROLLBACK TO SAVEPOINT companion_message_push") + conn.execute("RELEASE SAVEPOINT companion_message_push") + conn.commit() + return False + if oldest_channel[0] == cursor.lastrowid: + # A new channel message is itself the only + # evictable row; retain the protected directs and + # report that the incoming message was rejected. + conn.execute("ROLLBACK TO SAVEPOINT companion_message_push") + conn.execute("RELEASE SAVEPOINT companion_message_push") + conn.commit() + return False + conn.execute( + "DELETE FROM companion_messages WHERE id = ?", + (oldest_channel[0],), ) - """, - (companion_hash, companion_hash, max_messages), - ) + count -= 1 + conn.execute("RELEASE SAVEPOINT companion_message_push") conn.commit() - return inserted + return True except Exception as e: logger.error(f"Failed to push companion message: {e}") return False diff --git a/tests/test_companion_bridge_frame_utils.py b/tests/test_companion_bridge_frame_utils.py index c3122dc..2077fd8 100644 --- a/tests/test_companion_bridge_frame_utils.py +++ b/tests/test_companion_bridge_frame_utils.py @@ -5,7 +5,7 @@ from typing import Any, cast from unittest.mock import AsyncMock, MagicMock, patch import pytest -from openhop_core.companion.constants import RESP_CODE_NO_MORE_MESSAGES +from openhop_core.companion.constants import PUSH_CODE_MSG_WAITING, RESP_CODE_NO_MORE_MESSAGES from repeater.companion.bridge import RepeaterCompanionBridge, _to_json_safe from repeater.companion.frame_server import CompanionFrameServer @@ -201,6 +201,24 @@ 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_rejected_queue_callback_skips_sqlite_persistence_but_notifies_client(): + server = object.__new__(CompanionFrameServer) + server._persist_companion_message = AsyncMock() + server._enqueue_frame = MagicMock() + + await server._on_message_received( + b"\x01" * 32, + "rejected", + 1, + 0, + queued=False, + ) + + server._persist_companion_message.assert_not_awaited() + server._enqueue_frame.assert_called_once_with(bytes([PUSH_CODE_MSG_WAITING])) + + def test_companion_utils_validation_and_normalization(): assert normalize_companion_identity_key(" 0xAABB ") == "AABB" assert validate_companion_node_name(" node-1 ") == "node-1" diff --git a/tests/test_companion_settings.py b/tests/test_companion_settings.py index 9c95cd9..070248f 100644 --- a/tests/test_companion_settings.py +++ b/tests/test_companion_settings.py @@ -176,9 +176,54 @@ class TestSqliteRetentionTrim: def test_trims_to_max_messages(self, tmp_path): h = self._handler(tmp_path) - for i in range(5): - self._push(h, "0x01", i, max_messages=3) - assert len(h.companion_load_messages("0x01")) == 3 + results = [self._push(h, "0x01", i, max_messages=3) for i in range(5)] + assert results == [True, True, True, False, False] + assert [m["text"] for m in h.companion_load_messages("0x01")] == ["m0", "m1", "m2"] + + def test_evicts_oldest_channel_message_before_direct_message(self, tmp_path): + h = self._handler(tmp_path) + direct_one = {"text": "direct one", "packet_hash": "d1", "is_channel": False} + channel_one = {"text": "channel one", "packet_hash": "c1", "is_channel": True} + direct_two = {"text": "direct two", "packet_hash": "d2", "is_channel": False} + + assert h.companion_push_message("0x01", direct_one, max_messages=2) + assert h.companion_push_message("0x01", channel_one, max_messages=2) + assert h.companion_push_message("0x01", direct_two, max_messages=2) + + messages = h.companion_load_messages("0x01") + assert [m["text"] for m in messages] == ["direct one", "direct two"] + assert [m["is_channel"] for m in messages] == [0, 0] + + def test_rejects_channel_when_queue_contains_only_direct_messages(self, tmp_path): + h = self._handler(tmp_path) + for packet_hash in ("d1", "d2"): + assert h.companion_push_message( + "0x01", {"text": packet_hash, "packet_hash": packet_hash}, max_messages=2 + ) + + assert not h.companion_push_message( + "0x01", {"text": "channel", "packet_hash": "c1", "is_channel": True}, max_messages=2 + ) + assert [m["text"] for m in h.companion_load_messages("0x01")] == ["d1", "d2"] + + def test_rejected_insert_keeps_existing_channels_when_limit_is_lowered(self, tmp_path): + h = self._handler(tmp_path) + existing = [ + {"text": "direct one", "packet_hash": "d1", "is_channel": False}, + {"text": "channel one", "packet_hash": "c1", "is_channel": True}, + {"text": "direct two", "packet_hash": "d2", "is_channel": False}, + ] + for message in existing: + assert h.companion_push_message("0x01", message) + + assert not h.companion_push_message( + "0x01", {"text": "incoming", "packet_hash": "d3"}, max_messages=2 + ) + assert [m["text"] for m in h.companion_load_messages("0x01")] == [ + "direct one", + "channel one", + "direct two", + ] def test_none_keeps_all(self, tmp_path): h = self._handler(tmp_path) @@ -381,6 +426,14 @@ class TestPersistSkipWhenOff: asyncio.run(fs._persist_companion_message({"text": "x"})) fs.sqlite_handler.companion_push_message.assert_called_once_with("0x01", {"text": "x"}, 7) + def test_keeps_memory_message_when_sqlite_rejects_it(self): + import asyncio + + fs = self._frame_server(7) + fs.sqlite_handler.companion_push_message.return_value = False + asyncio.run(fs._persist_companion_message({"text": "x"})) + fs.bridge.message_queue.pop_last.assert_not_called() + class TestImportRepeaterContactsCap: """The import endpoint must never leave persisted contacts above max_contacts.