diff --git a/repeater/data_acquisition/sqlite_handler.py b/repeater/data_acquisition/sqlite_handler.py index a5b4e09..030dd71 100644 --- a/repeater/data_acquisition/sqlite_handler.py +++ b/repeater/data_acquisition/sqlite_handler.py @@ -3376,7 +3376,7 @@ class SQLiteHandler: path_len, sender_prefix, snr, rssi, channel_data_type, channel_data_payload FROM companion_messages WHERE companion_hash = ? - ORDER BY created_at ASC LIMIT ? + ORDER BY id ASC LIMIT ? """, (companion_hash, limit), ) @@ -3454,41 +3454,45 @@ class SQLiteHandler: conn.commit() return False if max_messages is not None: + last_id = cursor.lastrowid count = conn.execute( "SELECT COUNT(*) FROM companion_messages WHERE companion_hash = ?", (companion_hash,), ).fetchone()[0] - while count > max_messages: - oldest_channel = conn.execute( + excess = count - max_messages + if excess > 0: + # Eviction is ordered by id (an AUTOINCREMENT rowid, so + # insertion order) rather than created_at, keeping the + # policy immune to backwards clock steps. The incoming + # row is excluded so it is never evicted to make room + # for itself. + evictable = conn.execute( """ - SELECT id FROM companion_messages - WHERE companion_hash = ? AND is_channel = 1 - ORDER BY created_at ASC, id ASC LIMIT 1 + SELECT COUNT(*) FROM companion_messages + WHERE companion_hash = ? AND is_channel = 1 AND id != ? """, - (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. + (companion_hash, last_id), + ).fetchone()[0] + if evictable < excess: + # Not enough channel rows to make room without + # displacing a retained direct message. Undo the + # insert and every would-be eviction as one unit, + # keeping every prior row intact. 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],), + """ + DELETE FROM companion_messages + WHERE id IN ( + SELECT id FROM companion_messages + WHERE companion_hash = ? AND is_channel = 1 AND id != ? + ORDER BY id ASC LIMIT ? + ) + """, + (companion_hash, last_id, excess), ) - count -= 1 conn.execute("RELEASE SAVEPOINT companion_message_push") conn.commit() return True @@ -3507,7 +3511,7 @@ class SQLiteHandler: path_len, sender_prefix, snr, rssi, channel_data_type, channel_data_payload FROM companion_messages WHERE companion_hash = ? - ORDER BY created_at ASC LIMIT 1 + ORDER BY id ASC LIMIT 1 """, (companion_hash,), ) diff --git a/tests/test_companion_settings.py b/tests/test_companion_settings.py index 62e242b..d6d5dbf 100644 --- a/tests/test_companion_settings.py +++ b/tests/test_companion_settings.py @@ -281,6 +281,51 @@ class TestSqliteRetentionTrim: assert len(h.companion_load_messages("0x01")) == 2 assert len(h.companion_load_messages("0x02")) == 3 + def test_evicts_insertion_oldest_when_clock_steps_backwards(self, tmp_path, monkeypatch): + from repeater.data_acquisition import sqlite_handler + + h = self._handler(tmp_path) + for i in range(3): + assert h.companion_push_message( + "0x01", + {"text": f"c{i}", "packet_hash": f"c{i}", "is_channel": True}, + max_messages=3, + ) + + # The incoming row records a created_at older than every existing row. + # Insertion-order (id) eviction must drop the oldest existing row and + # keep the new push, rather than treating the incoming row as oldest. + monkeypatch.setattr(sqlite_handler.time, "time", lambda: 1.0) + assert h.companion_push_message( + "0x01", + {"text": "c3", "packet_hash": "c3", "is_channel": True}, + max_messages=3, + ) + + assert [m["text"] for m in h.companion_load_messages("0x01")] == ["c1", "c2", "c3"] + + def test_lowered_limit_evicts_multiple_channels_in_one_push(self, tmp_path): + h = self._handler(tmp_path) + seed = [ + {"text": "d1", "packet_hash": "d1", "is_channel": False}, + {"text": "d2", "packet_hash": "d2", "is_channel": False}, + {"text": "c1", "packet_hash": "c1", "is_channel": True}, + {"text": "c2", "packet_hash": "c2", "is_channel": True}, + {"text": "c3", "packet_hash": "c3", "is_channel": True}, + ] + for message in seed: + assert h.companion_push_message("0x01", message) + + assert h.companion_push_message( + "0x01", + {"text": "c4", "packet_hash": "c4", "is_channel": True}, + max_messages=4, + ) + + messages = h.companion_load_messages("0x01") + assert [m["text"] for m in messages] == ["d1", "d2", "c3", "c4"] + assert [m["is_channel"] for m in messages] == [0, 0, 1, 1] + class TestSenderPrefixPersistence: """sender_prefix (signed room-post author prefix) survives the SQLite round-trip."""