Add debug payload tracing and ignored packet logging (#437)

data/mesh_ingestor/handlers.py

@@ -21,10 +21,55 @@ import contextlib
 import importlib
 import json
 import sys
+import threading
 import time
 from collections.abc import Mapping
+from datetime import datetime, timezone
+from pathlib import Path
 
 from . import channels, config, queue
+
+_IGNORED_PACKET_LOG_PATH = Path(__file__).resolve().parents[2] / "ingored.txt"
+"""Filesystem path that stores ignored packets when debugging."""
+
+_IGNORED_PACKET_LOCK = threading.Lock()
+"""Lock guarding writes to :data:`_IGNORED_PACKET_LOG_PATH`."""
+
+
+def _ignored_packet_default(value: object) -> object:
+    """Return a JSON-serialisable representation for ignored packet data."""
+
+    if isinstance(value, (list, tuple, set)):
+        return list(value)
+    if isinstance(value, bytes):
+        return base64.b64encode(value).decode("ascii")
+    if isinstance(value, Mapping):
+        return {
+            str(key): _ignored_packet_default(sub_value)
+            for key, sub_value in value.items()
+        }
+    return str(value)
+
+
+def _record_ignored_packet(packet: Mapping | object, *, reason: str) -> None:
+    """Persist packet details to :data:`ingored.txt` during debugging."""
+
+    if not config.DEBUG:
+        return
+
+    timestamp = datetime.now(timezone.utc).isoformat()
+    entry = {
+        "timestamp": timestamp,
+        "reason": reason,
+        "packet": _ignored_packet_default(packet),
+    }
+    payload = json.dumps(entry, ensure_ascii=False, sort_keys=True)
+    with _IGNORED_PACKET_LOCK:
+        _IGNORED_PACKET_LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
+        with _IGNORED_PACKET_LOG_PATH.open("a", encoding="utf-8") as handle:
+            handle.write(f"{payload}\n")
+
+
 from .serialization import (
     _canonical_node_id,
     _coerce_float,
@@ -1087,10 +1132,6 @@ def store_packet_dict(packet: Mapping) -> None:
             if emoji_text:
                 emoji = emoji_text
 
-    encrypted_flag = _is_encrypted_flag(encrypted)
-    if not any([text, encrypted_flag, emoji is not None, reply_id is not None]):
-        return
-
     allowed_port_values = {"1", "TEXT_MESSAGE_APP", "REACTION_APP"}
     allowed_port_ints = {1}
 
@@ -1130,8 +1171,14 @@ def store_packet_dict(packet: Mapping) -> None:
 
     if portnum and portnum not in allowed_port_values:
         if portnum_int not in allowed_port_ints:
+            _record_ignored_packet(packet, reason="unsupported-port")
             return
 
+    encrypted_flag = _is_encrypted_flag(encrypted)
+    if not any([text, encrypted_flag, emoji is not None, reply_id is not None]):
+        _record_ignored_packet(packet, reason="no-message-payload")
+        return
+
     channel = _first(decoded, "channel", default=None)
     if channel is None:
         channel = _first(packet, "channel", default=0)
@@ -1142,6 +1189,7 @@ def store_packet_dict(packet: Mapping) -> None:
 
     pkt_id = _first(packet, "id", "packet_id", "packetId", default=None)
     if pkt_id is None:
+        _record_ignored_packet(packet, reason="missing-packet-id")
         return
     rx_time = int(_first(packet, "rxTime", "rx_time", default=time.time()))
     from_id = _first(packet, "fromId", "from_id", "from", default=None)
@@ -1181,6 +1229,7 @@ def store_packet_dict(packet: Mapping) -> None:
                 to_id=_canonical_node_id(to_id) or to_id,
                 channel=channel,
             )
+        _record_ignored_packet(packet, reason="skipped-direct-message")
         return
 
     message_payload = {

data/mesh_ingestor/queue.py

@@ -22,10 +22,57 @@ import json
 import threading
 import urllib.request
 from dataclasses import dataclass, field
-from typing import Callable, Iterable, Tuple
+from typing import Callable, Iterable, Mapping, Tuple
 
 from . import config
 
+
+def _stringify_payload_value(value: object) -> str:
+    """Return a stable string representation for ``value``."""
+
+    if isinstance(value, Mapping):
+        try:
+            return json.dumps(
+                {
+                    str(key): value[key]
+                    for key in sorted(value, key=lambda item: str(item))
+                },
+                sort_keys=True,
+                ensure_ascii=False,
+                default=str,
+            )
+        except Exception:  # pragma: no cover - defensive guard
+            return str(value)
+    if isinstance(value, (list, tuple)):
+        try:
+            return json.dumps(list(value), ensure_ascii=False, default=str)
+        except Exception:  # pragma: no cover - defensive guard
+            return str(value)
+    if isinstance(value, set):
+        try:
+            return json.dumps(sorted(value, key=str), ensure_ascii=False, default=str)
+        except Exception:  # pragma: no cover - defensive guard
+            return str(value)
+    if isinstance(value, bytes):
+        return json.dumps(value.decode("utf-8", "replace"), ensure_ascii=False)
+    if isinstance(value, str):
+        return json.dumps(value, ensure_ascii=False)
+    return str(value)
+
+
+def _payload_key_value_pairs(payload: Mapping[str, object]) -> str:
+    """Serialise ``payload`` into ``key=value`` pairs for debug logs."""
+
+    pairs: list[str] = []
+    for key in sorted(payload):
+        try:
+            formatted = _stringify_payload_value(payload[key])
+        except Exception:  # pragma: no cover - defensive guard
+            formatted = str(payload[key])
+        pairs.append(f"{key}={formatted}")
+    return " ".join(pairs)
+
+
 _MESSAGE_POST_PRIORITY = 10
 _NEIGHBOR_POST_PRIORITY = 20
 _POSITION_POST_PRIORITY = 30
@@ -173,6 +220,19 @@ def _queue_post_json(
     if send is None:
         send = _post_json
 
+    if config.DEBUG:
+        formatted_payload = (
+            _payload_key_value_pairs(payload)
+            if isinstance(payload, Mapping)
+            else str(payload)
+        )
+        config._debug_log(
+            f"Forwarding payload to API: {formatted_payload}",
+            context="queue.queue_post_json",
+            path=path,
+            priority=priority,
+        )
+
     _enqueue_post_json(path, payload, priority, state=state)
     with state.lock:
         if state.active:

tests/test_mesh.py

@@ -15,6 +15,7 @@
 import base64
 import enum
 import importlib
+import json
 import re
 import sys
 import threading
@@ -2158,6 +2159,24 @@ def test_post_json_logs_failures(mesh_module, monkeypatch, capsys):
     assert "POST request failed" in captured.out
 
 
+def test_queue_post_json_logs_payload_details(mesh_module, monkeypatch, capsys):
+    mesh = mesh_module
+
+    mesh._clear_post_queue()
+    monkeypatch.setattr(mesh, "DEBUG", True)
+
+    mesh._queue_post_json(
+        "/api/test",
+        {"alpha": "beta", "count": 7},
+        send=lambda *_: None,
+    )
+
+    out = capsys.readouterr().out
+    assert "Forwarding payload to API" in out
+    assert 'alpha="beta"' in out
+    assert "count=7" in out
+
+
 def test_queue_post_json_skips_when_active(mesh_module, monkeypatch):
     mesh = mesh_module
 
@@ -2213,6 +2232,25 @@ def test_upsert_node_logs_in_debug(mesh_module, monkeypatch, capsys):
     assert "Queued node upsert payload" in out
 
 
+def test_store_packet_dict_records_ignored_packets(mesh_module, monkeypatch, tmp_path):
+    mesh = mesh_module
+
+    monkeypatch.setattr(mesh, "DEBUG", True)
+    ignored_path = tmp_path / "ingored.txt"
+    monkeypatch.setattr(mesh.handlers, "_IGNORED_PACKET_LOG_PATH", ignored_path)
+    monkeypatch.setattr(mesh.handlers, "_IGNORED_PACKET_LOCK", threading.Lock())
+
+    packet = {"decoded": {"portnum": "UNKNOWN"}}
+    mesh.store_packet_dict(packet)
+
+    assert ignored_path.exists()
+    lines = ignored_path.read_text(encoding="utf-8").strip().splitlines()
+    assert lines
+    payload = json.loads(lines[-1])
+    assert payload["reason"] == "unsupported-port"
+    assert payload["packet"]["decoded"]["portnum"] == "UNKNOWN"
+
+
 def test_coerce_int_and_float_cover_edge_cases(mesh_module):
     mesh = mesh_module