fix(mqtt): shed retransmissions, not capture, when duty cycle starves the pool

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.

Author: agessaman

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`,

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:

examples/simple_repeater/MyMesh.cpp

@@ -1,6 +1,7 @@
 #include "MyMesh.h"
 #include <algorithm>
 #include <stdlib.h>  // for qsort()
+#include <helpers/RxReservePacketManager.h>
 
 /* ------------------------------ 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),

examples/simple_room_server/MyMesh.cpp

@@ -1,4 +1,5 @@
 #include "MyMesh.h"
+#include <helpers/RxReservePacketManager.h>
 
 #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)

src/helpers/RxReservePacketManager.h

@@ -0,0 +1,43 @@
+#pragma once
+
+#include <MeshCore.h>
+#include <helpers/StaticPoolPacketManager.h>
+
+// 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
+}