From 1c9c6292e5c9369120bc5349c52618523ab4092f Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 9 Jul 2026 17:54:38 -0700 Subject: [PATCH] fix(mqtt): shed retransmissions, not capture, when duty cycle starves the pool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Load-testing the restored token bucket at 'set dutycycle 1' showed MQTT capture dropping to exactly the TX rate. Queued retransmissions hold static- pool packets with no expiry, so throttling parks the whole pool in the send queue; Dispatcher::checkRecv() then discards received packets before logRx() ever feeds the bridge — each completed TX frees exactly one packet for exactly one more RX. Observer builds now use RxReservePacketManager (fork-owned header): once the free pool drops below a quarter of the pool, outbound packets are refused and freed, so RX allocation and MQTT capture continue at full rate while the node sheds repeat load it has no TX budget for anyway. Non-observer builds keep upstream pool behavior via the same factory; StaticPoolPacketManager stays byte-identical to upstream. --- MQTT_INTERNALS.md | 15 ++++++++- RESTORE_UPSTREAM_NOTES.md | 12 +++++++ examples/simple_repeater/MyMesh.cpp | 3 +- examples/simple_room_server/MyMesh.cpp | 3 +- src/helpers/RxReservePacketManager.h | 43 ++++++++++++++++++++++++++ 5 files changed, 73 insertions(+), 3 deletions(-) create mode 100644 src/helpers/RxReservePacketManager.h diff --git a/MQTT_INTERNALS.md b/MQTT_INTERNALS.md index ff65c8b6..1da44129 100644 --- a/MQTT_INTERNALS.md +++ b/MQTT_INTERNALS.md @@ -35,10 +35,23 @@ The observer feature is kept out of upstream-tracked files through three mechani Remaining integration points in upstream files: - `examples/simple_repeater/MyMesh.{h,cpp}`, `examples/simple_room_server/MyMesh.{h,cpp}` - - bridge/alerter/SNMP wiring and packet-feed hooks, guarded by `#ifdef WITH_MQTT_BRIDGE` + bridge/alerter/SNMP wiring and packet-feed hooks, guarded by `#ifdef WITH_MQTT_BRIDGE`; + plus the `createObserverPacketManager()` call in each constructor (see below) - `src/helpers/CommonCLI.{h,cpp}` - the three CLI hooks, `MQTTPrefs` load/save/migration - `src/Dispatcher.{h,cpp}` - radio watchdog block, guarded by `#ifdef WITH_MQTT_BRIDGE` +### Capture vs. duty-cycle throttling + +RX processing needs a free packet from the static pool before `logRx()` (and thus the +MQTT uplink) can run — `Dispatcher::checkRecv()` silently discards received data when +the pool is empty. Because the outbound queue holds pool packets with no expiry, +duty-cycle throttling can park the entire pool waiting on TX budget, capping capture at +the TX rate. Observer builds therefore use `RxReservePacketManager` (fork-owned, +`src/helpers/RxReservePacketManager.h`), which refuses to queue retransmissions once +the free pool drops below a reserve (a quarter of the pool) — the node sheds repeat +load it has no TX budget for, and capture continues at full rate. Non-observer builds +keep the upstream pool behavior. + ### `/mqtt_prefs` file format `/mqtt_prefs` is written with an 8-byte `MQTTPrefsHeader` (`magic`, `version`, diff --git a/RESTORE_UPSTREAM_NOTES.md b/RESTORE_UPSTREAM_NOTES.md index 0b56251e..1b05bd3a 100644 --- a/RESTORE_UPSTREAM_NOTES.md +++ b/RESTORE_UPSTREAM_NOTES.md @@ -42,6 +42,18 @@ This touches core TX timing. Confirm on hardware: verify `next_tx_time` spacing under load). - The observer radio watchdog still recovers a stuck radio (`radio_watchdog_minutes`). +### Known interaction: throttling starves MQTT capture (mitigated) + +Found during the on-device duty-cycle load test (`set dutycycle 1`): queued +retransmissions hold static-pool packets with no expiry, so heavy throttling parks the +whole pool in the send queue. `Dispatcher::checkRecv()` then drops received packets +before `logRx()` runs, capping MQTT capture at the TX rate (each TX frees one packet +for one RX). This is inherent upstream behavior — the old fork's `next_tx` spacing had +the same steady-state drain — but it defeats the observer's purpose. Mitigated on +observer builds by `RxReservePacketManager` (`src/helpers/RxReservePacketManager.h`): +retransmissions are shed once the free pool drops below a quarter of the pool, keeping +RX capture at full rate. See MQTT_INTERNALS.md "Capture vs. duty-cycle throttling". + ## Phase 2 — CAD and FEM RX gain (NOT done; needs care + device testing) Still missing at HEAD, also dropped by `22eb9b87`, still present upstream: diff --git a/examples/simple_repeater/MyMesh.cpp b/examples/simple_repeater/MyMesh.cpp index 672e8ea8..e8d598dc 100644 --- a/examples/simple_repeater/MyMesh.cpp +++ b/examples/simple_repeater/MyMesh.cpp @@ -1,6 +1,7 @@ #include "MyMesh.h" #include #include // for qsort() +#include /* ------------------------------ Config -------------------------------- */ @@ -901,7 +902,7 @@ void MyMesh::sendNodeDiscoverReq() { MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondClock &ms, mesh::RNG &rng, mesh::RTCClock &rtc, mesh::MeshTables &tables) - : mesh::Mesh(radio, ms, rng, rtc, *new StaticPoolPacketManager(32), tables), + : mesh::Mesh(radio, ms, rng, rtc, *createObserverPacketManager(32), tables), region_map(key_store), temp_map(key_store), _cli(board, rtc, sensors, region_map, acl, &_prefs, this), telemetry(MAX_PACKET_PAYLOAD - 4), diff --git a/examples/simple_room_server/MyMesh.cpp b/examples/simple_room_server/MyMesh.cpp index 69fd99fb..0132d246 100644 --- a/examples/simple_room_server/MyMesh.cpp +++ b/examples/simple_room_server/MyMesh.cpp @@ -1,4 +1,5 @@ #include "MyMesh.h" +#include #define REPLY_DELAY_MILLIS 1500 #define PUSH_NOTIFY_DELAY_MILLIS 2000 @@ -634,7 +635,7 @@ void MyMesh::onAckRecv(mesh::Packet *packet, uint32_t ack_crc) { MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondClock &ms, mesh::RNG &rng, mesh::RTCClock &rtc, mesh::MeshTables &tables) - : mesh::Mesh(radio, ms, rng, rtc, *new StaticPoolPacketManager(32), tables), + : mesh::Mesh(radio, ms, rng, rtc, *createObserverPacketManager(32), tables), region_map(key_store), temp_map(key_store), _cli(board, rtc, sensors, region_map, acl, &_prefs, this), telemetry(MAX_PACKET_PAYLOAD - 4) diff --git a/src/helpers/RxReservePacketManager.h b/src/helpers/RxReservePacketManager.h new file mode 100644 index 00000000..61865e95 --- /dev/null +++ b/src/helpers/RxReservePacketManager.h @@ -0,0 +1,43 @@ +#pragma once + +#include +#include + +// Fork-owned (not upstream-tracked). Observer builds capture every received packet +// to MQTT, but RX processing needs a free pool packet first — Dispatcher::checkRecv() +// discards the received bytes before logRx() when allocNew() fails. Under duty-cycle +// throttling the outbound queue can park the entire pool waiting on TX budget, which +// starves RX allocation and silently caps MQTT capture at the TX rate (each completed +// TX frees exactly one packet for exactly one more RX). +// +// This manager sheds *retransmissions* instead: once the free pool drops below the +// reserve, outbound packets are refused (freed straight back to the pool) so RX +// allocation — and therefore capture — continues at full rate. The node was already +// dropping traffic in that state; this chooses to drop repeats it has no TX budget +// for anyway, rather than capture. +class RxReservePacketManager : public StaticPoolPacketManager { + int _rx_reserve; +public: + RxReservePacketManager(int pool_size, int rx_reserve) + : StaticPoolPacketManager(pool_size), _rx_reserve(rx_reserve) {} + + void queueOutbound(mesh::Packet* packet, uint8_t priority, uint32_t scheduled_for) override { + if (getFreeCount() < _rx_reserve) { + MESH_DEBUG_PRINTLN("RxReservePacketManager: pool below RX reserve, shedding outbound"); + free(packet); + return; + } + StaticPoolPacketManager::queueOutbound(packet, priority, scheduled_for); + } +}; + +// The packet manager for an app build: observer builds reserve a quarter of the pool +// for RX so MQTT capture survives duty-cycle throttling; non-observer builds keep the +// upstream pool behavior unchanged. +inline mesh::PacketManager* createObserverPacketManager(int pool_size) { +#ifdef WITH_MQTT_BRIDGE + return new RxReservePacketManager(pool_size, pool_size / 4); +#else + return new StaticPoolPacketManager(pool_size); +#endif +}