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 03a2e810..88a44fb5 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 d9b93a69..81389ab1 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 +}