fix(storage): evict offline-queue rows set-based in insertion order

The companion queue's capacity path ran a Python loop of single-row
SELECT+DELETE pairs ordered by created_at. Wall-clock ordering meant a
backwards clock step (NTP correction) could make the just-inserted
channel row sort as oldest and wrongly reject the incoming message while
older channel rows remained evictable.

Replace the loop with one set-based DELETE ordered by id (AUTOINCREMENT,
i.e. insertion order, immune to clock steps), with an evictable-count
pre-check preserving the all-or-nothing rejection rule: never displace a
direct message, never evict the incoming row to make room for itself,
roll back the insert entirely when channel rows cannot make room. The
queue load and pop queries move to id ordering for the same reason.

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,),
                 )

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."""