Prioritize node posts in queued API updates (#107)

* Prioritize node posts in queued API updates

* run black

data/mesh.py

@@ -23,6 +23,8 @@ entry point that performs these synchronisation tasks.
 """
 
 import dataclasses
+import heapq
+import itertools
 import json, os, time, threading, signal, urllib.request, urllib.error
 from collections.abc import Mapping
 
@@ -40,6 +42,17 @@ INSTANCE = os.environ.get("POTATOMESH_INSTANCE", "").rstrip("/")
 API_TOKEN = os.environ.get("API_TOKEN", "")
 
 
+# --- POST queue ----------------------------------------------------------------
+_POST_QUEUE_LOCK = threading.Lock()
+_POST_QUEUE = []
+_POST_QUEUE_COUNTER = itertools.count()
+_POST_QUEUE_ACTIVE = False
+
+_NODE_POST_PRIORITY = 0
+_MESSAGE_POST_PRIORITY = 10
+_DEFAULT_POST_PRIORITY = 50
+
+
 def _get(obj, key, default=None):
     """Return a key or attribute value from ``obj``.
 
@@ -82,6 +95,51 @@ def _post_json(path: str, payload: dict):
             print(f"[warn] POST {url} failed: {e}")
 
 
+def _enqueue_post_json(path: str, payload: dict, priority: int):
+    """Store a POST request in the priority queue."""
+
+    with _POST_QUEUE_LOCK:
+        heapq.heappush(
+            _POST_QUEUE, (priority, next(_POST_QUEUE_COUNTER), path, payload)
+        )
+
+
+def _drain_post_queue():
+    """Process queued POST requests in priority order."""
+
+    global _POST_QUEUE_ACTIVE
+    while True:
+        with _POST_QUEUE_LOCK:
+            if not _POST_QUEUE:
+                _POST_QUEUE_ACTIVE = False
+                return
+            _priority, _idx, path, payload = heapq.heappop(_POST_QUEUE)
+        _post_json(path, payload)
+
+
+def _queue_post_json(
+    path: str, payload: dict, *, priority: int = _DEFAULT_POST_PRIORITY
+):
+    """Queue a POST request and start processing if idle."""
+
+    global _POST_QUEUE_ACTIVE
+    _enqueue_post_json(path, payload, priority)
+    with _POST_QUEUE_LOCK:
+        if _POST_QUEUE_ACTIVE:
+            return
+        _POST_QUEUE_ACTIVE = True
+    _drain_post_queue()
+
+
+def _clear_post_queue():
+    """Clear the pending POST queue (used by tests)."""
+
+    global _POST_QUEUE_ACTIVE
+    with _POST_QUEUE_LOCK:
+        _POST_QUEUE.clear()
+        _POST_QUEUE_ACTIVE = False
+
+
 # --- Node upsert --------------------------------------------------------------
 def _node_to_dict(n) -> dict:
     """Convert Meshtastic node or user structures into plain dictionaries.
@@ -129,7 +187,7 @@ def upsert_node(node_id, n):
     """
 
     ndict = _node_to_dict(n)
-    _post_json("/api/nodes", {node_id: ndict})
+    _queue_post_json("/api/nodes", {node_id: ndict}, priority=_NODE_POST_PRIORITY)
 
     if DEBUG:
         user = _get(ndict, "user") or {}
@@ -286,7 +344,7 @@ def store_packet_dict(p: dict):
         "rssi": int(rssi) if rssi is not None else None,
         "hop_limit": int(hop) if hop is not None else None,
     }
-    _post_json("/api/messages", msg)
+    _queue_post_json("/api/messages", msg, priority=_MESSAGE_POST_PRIORITY)
 
     if DEBUG:
         print(

tests/test_mesh.py

@@ -85,9 +85,14 @@ def mesh_module(monkeypatch):
     else:
         module = importlib.import_module(module_name)
 
+    if hasattr(module, "_clear_post_queue"):
+        module._clear_post_queue()
+
     yield module
 
     # Ensure a clean import for the next test
+    if hasattr(module, "_clear_post_queue"):
+        module._clear_post_queue()
     sys.modules.pop(module_name, None)
 
 
@@ -125,7 +130,9 @@ def test_store_packet_dict_posts_text_message(mesh_module, monkeypatch):
     mesh = mesh_module
     captured = []
     monkeypatch.setattr(
-        mesh, "_post_json", lambda path, payload: captured.append((path, payload))
+        mesh,
+        "_queue_post_json",
+        lambda path, payload, *, priority: captured.append((path, payload, priority)),
     )
 
     packet = {
@@ -147,7 +154,7 @@ def test_store_packet_dict_posts_text_message(mesh_module, monkeypatch):
     mesh.store_packet_dict(packet)
 
     assert captured, "Expected POST to be triggered for text message"
-    path, payload = captured[0]
+    path, payload, priority = captured[0]
     assert path == "/api/messages"
     assert payload["id"] == 123
     assert payload["channel"] == 4
@@ -160,13 +167,16 @@ def test_store_packet_dict_posts_text_message(mesh_module, monkeypatch):
     assert payload["hop_limit"] == 3
     assert payload["snr"] == pytest.approx(1.25)
     assert payload["rssi"] == -70
+    assert priority == mesh._MESSAGE_POST_PRIORITY
 
 
 def test_store_packet_dict_ignores_non_text(mesh_module, monkeypatch):
     mesh = mesh_module
     captured = []
     monkeypatch.setattr(
-        mesh, "_post_json", lambda *args, **kwargs: captured.append(args)
+        mesh,
+        "_queue_post_json",
+        lambda *args, **kwargs: captured.append((args, kwargs)),
     )
 
     packet = {
@@ -182,7 +192,7 @@ def test_store_packet_dict_ignores_non_text(mesh_module, monkeypatch):
 
     mesh.store_packet_dict(packet)
 
-    assert not captured, "Non-text messages should not be posted"
+    assert not captured, "Non-text messages should not be queued"
 
 
 def test_node_items_snapshot_handles_transient_runtime_error(mesh_module):
@@ -340,7 +350,9 @@ def test_store_packet_dict_uses_top_level_channel(mesh_module, monkeypatch):
     mesh = mesh_module
     captured = []
     monkeypatch.setattr(
-        mesh, "_post_json", lambda path, payload: captured.append(payload)
+        mesh,
+        "_queue_post_json",
+        lambda path, payload, *, priority: captured.append((path, payload, priority)),
     )
 
     packet = {
@@ -355,18 +367,22 @@ def test_store_packet_dict_uses_top_level_channel(mesh_module, monkeypatch):
     mesh.store_packet_dict(packet)
 
     assert captured, "Expected message to be stored"
-    payload = captured[0]
+    path, payload, priority = captured[0]
+    assert path == "/api/messages"
     assert payload["channel"] == 5
     assert payload["portnum"] == "1"
     assert payload["text"] == "hi"
     assert payload["snr"] is None and payload["rssi"] is None
+    assert priority == mesh._MESSAGE_POST_PRIORITY
 
 
 def test_store_packet_dict_handles_invalid_channel(mesh_module, monkeypatch):
     mesh = mesh_module
     captured = []
     monkeypatch.setattr(
-        mesh, "_post_json", lambda path, payload: captured.append(payload)
+        mesh,
+        "_queue_post_json",
+        lambda path, payload, *, priority: captured.append((path, payload, priority)),
     )
 
     packet = {
@@ -383,7 +399,30 @@ def test_store_packet_dict_handles_invalid_channel(mesh_module, monkeypatch):
     mesh.store_packet_dict(packet)
 
     assert captured
-    assert captured[0]["channel"] == 0
+    path, payload, priority = captured[0]
+    assert path == "/api/messages"
+    assert payload["channel"] == 0
+    assert priority == mesh._MESSAGE_POST_PRIORITY
+
+
+def test_post_queue_prioritises_nodes(mesh_module, monkeypatch):
+    mesh = mesh_module
+    mesh._clear_post_queue()
+    calls = []
+
+    def record(path, payload):
+        calls.append((path, payload))
+
+    monkeypatch.setattr(mesh, "_post_json", record)
+
+    mesh._enqueue_post_json("/api/messages", {"id": 1}, mesh._MESSAGE_POST_PRIORITY)
+    mesh._enqueue_post_json(
+        "/api/nodes", {"!node": {"foo": "bar"}}, mesh._NODE_POST_PRIORITY
+    )
+
+    mesh._drain_post_queue()
+
+    assert [path for path, _ in calls] == ["/api/nodes", "/api/messages"]
 
 
 def test_store_packet_dict_requires_id(mesh_module, monkeypatch):
@@ -392,7 +431,7 @@ def test_store_packet_dict_requires_id(mesh_module, monkeypatch):
     def fail_post(*_, **__):
         raise AssertionError("Should not post without an id")
 
-    monkeypatch.setattr(mesh, "_post_json", fail_post)
+    monkeypatch.setattr(mesh, "_queue_post_json", fail_post)
 
     packet = {"decoded": {"payload": {"text": "hello"}, "portnum": "TEXT_MESSAGE_APP"}}
     mesh.store_packet_dict(packet)