diff --git a/meshchatx.rsm b/meshchatx.rsm index 259c26d4..48ea0950 100644 Binary files a/meshchatx.rsm and b/meshchatx.rsm differ diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py index e7e50242..019ec8f8 100644 --- a/meshchatx/meshchat.py +++ b/meshchatx/meshchat.py @@ -183,8 +183,10 @@ from meshchatx.src.backend.auto_resend_guard import ( RECENT_SAME_CONTENT_SECONDS, AutoResendCoordinator, cooldown_until, + fields_have_attachments, fields_with_auto_resend_count, next_attempt_count, + parse_fields_dict, should_skip_for_budget, ) from meshchatx.src.backend.local_message_retention import ( @@ -23244,6 +23246,16 @@ class ReticulumMeshChat: ): continue + fields = parse_fields_dict(failed_message.get("fields")) + allow_attachments = ( + ctx.config.allow_auto_resending_failed_messages_with_attachments.get() + ) + if not allow_attachments and fields_have_attachments(fields): + print( + "Not resending failed message with attachments, as setting is disabled", + ) + continue + claimed = ( ctx.database.messages.try_claim_failed_message_for_auto_resend( message_hash, @@ -23257,14 +23269,9 @@ class ReticulumMeshChat: if not claimed: continue - # parse fields as json - fields = json.loads(failed_message["fields"] or "{}") - if not isinstance(fields, dict): - fields = {} - # parse image field image_field = None - if "image" in fields: + if "image" in fields and isinstance(fields.get("image"), dict): image_field = LxmfImageField( fields["image"]["image_type"], base64.b64decode(fields["image"]["image_bytes"]), @@ -23272,7 +23279,7 @@ class ReticulumMeshChat: # parse audio field audio_field = None - if "audio" in fields: + if "audio" in fields and isinstance(fields.get("audio"), dict): audio_field = LxmfAudioField( fields["audio"]["audio_mode"], base64.b64decode(fields["audio"]["audio_bytes"]), @@ -23280,30 +23287,22 @@ class ReticulumMeshChat: # parse file attachments field file_attachments_field = None - if "file_attachments" in fields: + if "file_attachments" in fields and isinstance( + fields.get("file_attachments"), + list, + ): file_attachments = [ LxmfFileAttachment( file_attachment["file_name"], base64.b64decode(file_attachment["file_bytes"]), ) for file_attachment in fields["file_attachments"] + if isinstance(file_attachment, dict) ] file_attachments_field = LxmfFileAttachmentsField( file_attachments, ) - # don't resend message with attachments if not allowed - if not ctx.config.allow_auto_resending_failed_messages_with_attachments.get(): - if ( - image_field is not None - or audio_field is not None - or file_attachments_field is not None - ): - print( - "Not resending failed message with attachments, as setting is disabled", - ) - continue - attempt = next_attempt_count(failed_message.get("fields")) ctx.database.messages.set_message_fields_json( message_hash, diff --git a/meshchatx/src/backend/auto_resend_guard.py b/meshchatx/src/backend/auto_resend_guard.py index 611035d3..9ce2e91e 100644 --- a/meshchatx/src/backend/auto_resend_guard.py +++ b/meshchatx/src/backend/auto_resend_guard.py @@ -67,6 +67,20 @@ class AutoResendCoordinator: return lock +def fields_have_attachments(fields_raw: Any) -> bool: + fields = _parse_fields(fields_raw) + if not fields: + return False + if fields.get("image") or fields.get("audio"): + return True + files = fields.get("file_attachments") + return isinstance(files, list) and len(files) > 0 + + +def parse_fields_dict(fields_raw: Any) -> dict: + return _parse_fields(fields_raw) + + def should_skip_for_budget( fields_raw: Any, *, max_attempts: int = MAX_AUTO_RESEND_ATTEMPTS ) -> bool: diff --git a/meshchatx/src/backend/database/messages.py b/meshchatx/src/backend/database/messages.py index 76185a7e..4f650826 100644 --- a/meshchatx/src/backend/database/messages.py +++ b/meshchatx/src/backend/database/messages.py @@ -852,26 +852,38 @@ class MessageDAO: """Hashes of duplicate rows (same peer, direction, and text), excluding the oldest keep. Empty or whitespace-only content is ignored so attachment-only or blank - rows are not collapsed together. + rows are not collapsed together. Oldest is by timestamp then id. """ - rows = self.provider.fetchall( + groups = self.provider.fetchall( """ - SELECT m.hash AS hash - FROM lxmf_messages m - INNER JOIN ( - SELECT peer_hash, is_incoming, content, MIN(id) AS keep_id - FROM lxmf_messages - WHERE content IS NOT NULL AND TRIM(content) != '' - GROUP BY peer_hash, is_incoming, content - HAVING COUNT(*) > 1 - ) d - ON m.peer_hash = d.peer_hash - AND m.is_incoming = d.is_incoming - AND m.content = d.content - WHERE m.id != d.keep_id + SELECT peer_hash, is_incoming, content + FROM lxmf_messages + WHERE content IS NOT NULL AND TRIM(content) != '' + GROUP BY peer_hash, is_incoming, content + HAVING COUNT(*) > 1 """, ) - return [r["hash"] for r in rows if r.get("hash")] + to_delete: list[str] = [] + for group in groups: + rows = self.provider.fetchall( + """ + SELECT hash + FROM lxmf_messages + WHERE peer_hash = ? + AND is_incoming = ? + AND content = ? + ORDER BY + CASE WHEN timestamp IS NULL THEN 1 ELSE 0 END, + timestamp ASC, + id ASC + """, + (group["peer_hash"], group["is_incoming"], group["content"]), + ) + # Keep the first (oldest); delete the rest. + for row in rows[1:]: + if row.get("hash"): + to_delete.append(row["hash"]) + return to_delete def count_duplicate_lxmf_messages_by_content(self) -> int: return len(self.list_duplicate_lxmf_message_hashes_by_content()) diff --git a/tests/backend/test_auto_resend_regression.py b/tests/backend/test_auto_resend_regression.py new file mode 100644 index 00000000..b81bfb50 --- /dev/null +++ b/tests/backend/test_auto_resend_regression.py @@ -0,0 +1,338 @@ +# SPDX-License-Identifier: 0BSD + +"""Regression tests for auto-resend flooding guards and duplicate cleanup.""" + +from __future__ import annotations + +import asyncio +import json +import time +import types +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from meshchatx.meshchat import ReticulumMeshChat +from meshchatx.src.backend import auto_resend_guard as guard +from meshchatx.src.backend.database import Database +from meshchatx.src.backend.database.provider import DatabaseProvider +from meshchatx.src.backend.database.schema import DatabaseSchema + + +@pytest.fixture +def db(tmp_path): + path = str(tmp_path / "resend_reg.db") + provider = DatabaseProvider(path) + DatabaseSchema(provider).initialize() + database = Database(path) + yield database + database.close_all() + provider.close_all() + + +def _insert_failed(db, *, msg_hash, peer, content, fields="{}", ts=None, next_at=None): + now = ts if ts is not None else time.time() + db.messages.upsert_lxmf_message( + { + "hash": msg_hash, + "source_hash": peer, + "destination_hash": peer, + "peer_hash": peer, + "state": "failed", + "progress": 0, + "is_incoming": 0, + "method": "opportunistic", + "delivery_attempts": 0, + "next_delivery_attempt_at": next_at, + "title": "", + "content": content, + "fields": fields, + "rssi": None, + "snr": None, + "quality": None, + "is_spam": 0, + "reply_to_hash": None, + "attachments_stripped": 0, + "timestamp": now, + }, + ) + + +def _insert_outbound(db, *, msg_hash, peer, content, state="delivered", ts=None): + now = ts if ts is not None else time.time() + db.messages.upsert_lxmf_message( + { + "hash": msg_hash, + "source_hash": peer, + "destination_hash": peer, + "peer_hash": peer, + "state": state, + "progress": 1, + "is_incoming": 0, + "method": "direct", + "delivery_attempts": 1, + "next_delivery_attempt_at": None, + "title": "", + "content": content, + "fields": "{}", + "rssi": None, + "snr": None, + "quality": None, + "is_spam": 0, + "reply_to_hash": None, + "attachments_stripped": 0, + "timestamp": now, + }, + ) + + +def _bind_resend_app(db): + app = MagicMock() + app._auto_resend_coordinator = guard.AutoResendCoordinator() + app.websocket_broadcast = AsyncMock() + app.send_message = AsyncMock() + app.resend_failed_messages_for_destination = types.MethodType( + ReticulumMeshChat.resend_failed_messages_for_destination, + app, + ) + ctx = MagicMock() + ctx.identity.hash.hex.return_value = "aa" * 16 + ctx.database = db + ctx.config.allow_auto_resending_failed_messages_with_attachments.get.return_value = ( + True + ) + return app, ctx + + +def test_fields_have_attachments_and_invalid_json(): + assert not guard.fields_have_attachments("{}") + assert not guard.fields_have_attachments("not-json") + assert guard.fields_have_attachments('{"image":{"image_type":"png"}}') + assert guard.fields_have_attachments('{"file_attachments":[{"file_name":"a"}]}') + assert guard.parse_fields_dict("not-json") == {} + assert guard.should_skip_for_budget( + '{"_mcx_auto_resend_count": 3}', + max_attempts=3, + ) + + +@pytest.mark.asyncio +async def test_resend_skips_when_budget_exhausted(db): + peer = "b" * 32 + msg = "c" * 32 + _insert_failed( + db, + msg_hash=msg, + peer=peer, + content="help", + fields=json.dumps({"_mcx_auto_resend_count": 3}), + ) + app, ctx = _bind_resend_app(db) + await app.resend_failed_messages_for_destination(peer, context=ctx) + app.send_message.assert_not_called() + + +@pytest.mark.asyncio +async def test_resend_skips_when_recent_same_content_exists(db): + peer = "d" * 32 + now = time.time() + _insert_outbound( + db, + msg_hash="1" * 32, + peer=peer, + content="help text", + ts=now - 30, + ) + _insert_failed( + db, + msg_hash="2" * 32, + peer=peer, + content="help text", + ts=now - 5, + ) + app, ctx = _bind_resend_app(db) + await app.resend_failed_messages_for_destination(peer, context=ctx) + app.send_message.assert_not_called() + + +@pytest.mark.asyncio +async def test_resend_skips_attachments_before_claim(db): + peer = "e" * 32 + msg = "f" * 32 + fields = json.dumps( + { + "image": { + "image_type": "png", + "image_bytes": "aa", + }, + }, + ) + _insert_failed(db, msg_hash=msg, peer=peer, content="pic", fields=fields) + app, ctx = _bind_resend_app(db) + ctx.config.allow_auto_resending_failed_messages_with_attachments.get.return_value = ( + False + ) + await app.resend_failed_messages_for_destination(peer, context=ctx) + app.send_message.assert_not_called() + row = db.provider.fetchone( + "SELECT next_delivery_attempt_at FROM lxmf_messages WHERE hash = ?", + (msg,), + ) + # Must not burn cooldown when attachments are disallowed. + assert row["next_delivery_attempt_at"] is None + + +@pytest.mark.asyncio +async def test_resend_claims_once_and_deletes_old_on_success(db): + peer = "1" * 32 + old_hash = "2" * 32 + new_hash_bytes = bytes.fromhex("3" * 32) + _insert_failed(db, msg_hash=old_hash, peer=peer, content="retry me") + app, ctx = _bind_resend_app(db) + + async def send_and_materialize(*args, **kwargs): + new_msg = MagicMock() + new_msg.hash = new_hash_bytes + _insert_outbound( + db, + msg_hash=new_hash_bytes.hex(), + peer=peer, + content="retry me", + state="outbound", + ) + return new_msg + + app.send_message.side_effect = send_and_materialize + + await app.resend_failed_messages_for_destination(peer, context=ctx) + assert app.send_message.await_count == 1 + assert ( + db.provider.fetchone( + "SELECT 1 FROM lxmf_messages WHERE hash = ?", + (old_hash,), + ) + is None + ) + new_row = db.provider.fetchone( + "SELECT fields FROM lxmf_messages WHERE hash = ?", + (new_hash_bytes.hex(),), + ) + assert guard.read_auto_resend_count(new_row["fields"]) == 1 + app.websocket_broadcast.assert_awaited() + + +@pytest.mark.asyncio +async def test_resend_concurrent_calls_send_only_once(db): + peer = "4" * 32 + old_hash = "5" * 32 + _insert_failed(db, msg_hash=old_hash, peer=peer, content="race") + app, ctx = _bind_resend_app(db) + + async def slow_send(*args, **kwargs): + await asyncio.sleep(0.05) + m = MagicMock() + m.hash = bytes.fromhex("6" * 32) + _insert_outbound( + db, + msg_hash=m.hash.hex(), + peer=peer, + content="race", + state="outbound", + ) + return m + + app.send_message.side_effect = slow_send + await asyncio.gather( + app.resend_failed_messages_for_destination(peer, context=ctx), + app.resend_failed_messages_for_destination(peer, context=ctx), + ) + assert app.send_message.await_count == 1 + + +@pytest.mark.asyncio +async def test_resend_does_not_delete_old_when_send_returns_none(db): + peer = "7" * 32 + old_hash = "8" * 32 + _insert_failed(db, msg_hash=old_hash, peer=peer, content="keep") + app, ctx = _bind_resend_app(db) + app.send_message.return_value = None + await app.resend_failed_messages_for_destination(peer, context=ctx) + assert db.provider.fetchone( + "SELECT state FROM lxmf_messages WHERE hash = ?", + (old_hash,), + )["state"] == "failed" + # Cooldown claimed, attempt counted. + row = db.provider.fetchone( + "SELECT fields, next_delivery_attempt_at FROM lxmf_messages WHERE hash = ?", + (old_hash,), + ) + assert guard.read_auto_resend_count(row["fields"]) == 1 + assert row["next_delivery_attempt_at"] is not None + + +def test_duplicate_cleanup_keeps_oldest_by_timestamp(db): + peer = "9" * 32 + base = time.time() + # Later insert id can still be oldest by timestamp. + _insert_outbound(db, msg_hash="a" * 32, peer=peer, content="dup", ts=base - 5) + _insert_outbound(db, msg_hash="b" * 32, peer=peer, content="dup", ts=base - 50) + _insert_outbound(db, msg_hash="c" * 32, peer=peer, content="dup", ts=base - 1) + deleted = db.messages.delete_duplicate_lxmf_messages_by_content() + assert deleted == 2 + left = db.provider.fetchone( + "SELECT hash FROM lxmf_messages WHERE content = 'dup'", + ) + assert left["hash"] == "b" * 32 + + +def test_source_keeps_auto_resend_and_duplicate_guards(): + from pathlib import Path + + root = Path(__file__).resolve().parents[2] + meshchat = (root / "meshchatx" / "meshchat.py").read_text(encoding="utf-8") + guard_src = ( + root / "meshchatx" / "src" / "backend" / "auto_resend_guard.py" + ).read_text(encoding="utf-8") + messages = ( + root / "meshchatx" / "src" / "backend" / "database" / "messages.py" + ).read_text(encoding="utf-8") + settings = ( + root + / "meshchatx" + / "src" + / "frontend" + / "components" + / "settings" + / "SettingsPage.vue" + ).read_text(encoding="utf-8") + + assert "AutoResendCoordinator" in meshchat + assert "try_claim_failed_message_for_auto_resend" in meshchat + assert "fields_have_attachments" in meshchat + assert "has_path(destination_hash)" in meshchat + assert "MAX_AUTO_RESEND_ATTEMPTS" in guard_src + assert "delete_duplicate_lxmf_messages_by_content" in messages + assert "clearDuplicateMessages" in settings + assert "/api/v1/maintenance/messages/duplicates" in meshchat + + +def test_claim_reopens_after_cooldown_expires(db): + peer = "0" * 32 + msg = "a" * 32 + now = 1_000_000.0 + _insert_failed(db, msg_hash=msg, peer=peer, content="x", next_at=now - 1) + assert db.messages.try_claim_failed_message_for_auto_resend( + msg, + cooldown_until=now + 120, + now=now, + ) + assert not db.messages.try_claim_failed_message_for_auto_resend( + msg, + cooldown_until=now + 240, + now=now + 10, + ) + assert db.messages.try_claim_failed_message_for_auto_resend( + msg, + cooldown_until=now + 360, + now=now + 121, + ) diff --git a/tests/backend/test_duplicate_messages_cleanup.py b/tests/backend/test_duplicate_messages_cleanup.py index 708c502c..9c1800d8 100644 --- a/tests/backend/test_duplicate_messages_cleanup.py +++ b/tests/backend/test_duplicate_messages_cleanup.py @@ -65,6 +65,24 @@ def test_delete_duplicate_messages_by_content_keeps_oldest(tmp_path): provider.close_all() +def test_delete_duplicate_prefers_earlier_timestamp_over_lower_id(tmp_path): + path = str(tmp_path / "d3.db") + provider = DatabaseProvider(path) + DatabaseSchema(provider).initialize() + db = Database(path) + peer = "d" * 32 + base = time.time() + # Insert newer timestamp first (lower id), then older timestamp. + _msg(db, msg_hash="1" * 32, peer=peer, content="z", ts=base - 1) + _msg(db, msg_hash="2" * 32, peer=peer, content="z", ts=base - 100) + deleted = db.messages.delete_duplicate_lxmf_messages_by_content() + assert deleted == 1 + left = db.provider.fetchone("SELECT hash FROM lxmf_messages") + assert left["hash"] == "2" * 32 + db.close_all() + provider.close_all() + + def test_duplicates_are_scoped_by_peer_and_direction(tmp_path): path = str(tmp_path / "d2.db") provider = DatabaseProvider(path) diff --git a/tests/frontend/SettingsPage.config-persistence.test.js b/tests/frontend/SettingsPage.config-persistence.test.js index 4d9993bd..a3f4e14e 100644 --- a/tests/frontend/SettingsPage.config-persistence.test.js +++ b/tests/frontend/SettingsPage.config-persistence.test.js @@ -611,6 +611,14 @@ describe("SettingsPage — maintenance, exports, telemetry trust, RNS reload", ( expect(api.delete).toHaveBeenCalledWith("/api/v1/maintenance/messages/duplicates"); }); + it("maintenance UI exposes duplicate cleanup and age purge", async () => { + const w = await mountSettingsPage(api); + const html = w.html(); + expect(html).toContain("maintenance.clear_duplicates"); + expect(html).toContain("maintenance.purge_old_title"); + expect(html).toContain("maintenance.export_old_archive"); + }); + it("clearAnnounces DELETEs announces", async () => { const w = await mountSettingsPage(api); await w.vm.clearAnnounces();