From cffdb7dca62257dc3d4076a6906222d940fa72be Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Tue, 16 Sep 2025 19:30:38 +0200 Subject: [PATCH] Prioritize node posts in queued API updates (#107) * Prioritize node posts in queued API updates * run black --- data/mesh.py | 62 ++++++++++++++++++++++++++++++++++++++++++++-- tests/test_mesh.py | 57 +++++++++++++++++++++++++++++++++++------- 2 files changed, 108 insertions(+), 11 deletions(-) diff --git a/data/mesh.py b/data/mesh.py index b737f09..d91a211 100644 --- a/data/mesh.py +++ b/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( diff --git a/tests/test_mesh.py b/tests/test_mesh.py index 5a12ec5..077de78 100644 --- a/tests/test_mesh.py +++ b/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)