fix(mqtt): keep throttled nodes administrable — priority-aware shed + stale expiry

Device testing at 'set dutycycle 1' on a busy mesh showed the node becoming
un-administrable within ~2 minutes: the shed policy dropped its own CLI
responses along with repeats, and parked retransmissions (which never expire)
absorbed every budget refill.

RxReservePacketManager now sheds by priority below the RX reserve — only
pri > 1 outbound (multi-hop flood repeats, adverts, trace) is refused, so the
node's own responses/ACKs (pri 0) and login/PATH replies (pri 1) still queue;
below an emergency floor (reserve/2) everything is shed to protect capture.
Queued packets untransmitted 30 s past their scheduled time are expired at
dequeue via a pointer-keyed age table (the pool is a fixed set of packets, so
pool_size slots cover every key). Under normal load the queue drains in
milliseconds and neither policy triggers.

Author: agessaman

MQTT_INTERNALS.md

@@ -46,11 +46,16 @@ RX processing needs a free packet from the static pool before `logRx()` (and thu
 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.
+the TX rate — and the parked repeats absorb every budget refill, starving the node's
+own CLI responses and making it un-administrable over the mesh. Observer builds
+therefore use `RxReservePacketManager` (fork-owned,
+`src/helpers/RxReservePacketManager.h`): below the RX reserve (a quarter of the pool)
+it sheds only low-priority outbound (multi-hop flood repeats, adverts, trace), keeping
+the node's own responses/ACKs queueable; below a smaller emergency floor it sheds
+everything to keep capture alive. Queued packets still untransmitted 30 s past their
+scheduled time are expired at dequeue, so under throttle the queue holds only fresh
+traffic and admin responses reach the trickle of TX budget. Non-observer builds keep
+the upstream pool behavior.
 
 ### `/mqtt_prefs` file format
 

RESTORE_UPSTREAM_NOTES.md

@@ -48,11 +48,14 @@ 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".
+for one RX). Parked repeats also absorb every budget refill, starving the node's own
+CLI responses — device-confirmed: a 1%-duty node on a busy mesh became
+un-administrable within ~2 minutes. 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`): priority-aware shedding below a pool
+reserve (own responses/ACKs stay queueable) plus 30 s expiry of stale queued
+outbound. See MQTT_INTERNALS.md "Capture vs. duty-cycle throttling".
 
 ## Phase 2 — CAD and FEM RX gain (NOT done; needs care + device testing)
 

src/helpers/RxReservePacketManager.h

@@ -8,27 +8,83 @@
 // 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).
+// TX frees exactly one packet for exactly one more RX). Parked retransmissions also
+// absorb every budget refill, starving the node's own CLI responses/ACKs and making
+// a heavily-throttled node un-administrable over the mesh.
 //
-// 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.
+// Two policies fix this, both confined to this manager:
+//
+// 1. Priority-aware shedding. Below the RX reserve, only low-priority outbound
+//    (priority > 1: multi-hop flood repeats, adverts, trace) is refused; the node's
+//    own responses/ACKs (pri 0) and login/PATH replies (pri 1) still queue. Below the
+//    smaller emergency floor everything is shed to keep capture alive.
+// 2. Stale-packet expiry. A queued packet still untransmitted STALE_OUTBOUND_MS past
+//    its scheduled time is dropped at the next dequeue — a repeat delayed that long is
+//    noise (the flood has long since propagated), and a CLI response that old has
+//    already timed out at the client. Under normal load the queue drains in
+//    milliseconds and this never triggers; under throttle it frees the pool and lets
+//    fresh traffic (including admin responses) compete for the trickle of TX budget.
 class RxReservePacketManager : public StaticPoolPacketManager {
-  int _rx_reserve;
+  int _rx_reserve, _emergency_floor;
+  int _cap;
+  // scheduled_for per queued packet, keyed by packet pointer. The pool is a fixed set
+  // of _cap Packet objects, so _cap slots cover every possible key with no eviction.
+  struct AgeEntry { mesh::Packet* pkt; uint32_t scheduled_for; };
+  AgeEntry* _ages;
+
+  static const uint32_t STALE_OUTBOUND_MS = 30000;
+  static const uint8_t MAX_PROTECTED_PRI = 1;  // pri 0-1 = own responses/ACKs/replies
+
+  void recordAge(mesh::Packet* packet, uint32_t scheduled_for) {
+    int empty = -1;
+    for (int i = 0; i < _cap; i++) {
+      if (_ages[i].pkt == packet) { _ages[i].scheduled_for = scheduled_for; return; }
+      if (empty < 0 && _ages[i].pkt == NULL) empty = i;
+    }
+    if (empty >= 0) { _ages[empty].pkt = packet; _ages[empty].scheduled_for = scheduled_for; }
+  }
+
+  bool lookupAge(const mesh::Packet* packet, uint32_t* scheduled_for) const {
+    for (int i = 0; i < _cap; i++) {
+      if (_ages[i].pkt == packet) { *scheduled_for = _ages[i].scheduled_for; return true; }
+    }
+    return false;
+  }
+
 public:
   RxReservePacketManager(int pool_size, int rx_reserve)
-    : StaticPoolPacketManager(pool_size), _rx_reserve(rx_reserve) {}
+    : StaticPoolPacketManager(pool_size), _rx_reserve(rx_reserve),
+      _emergency_floor(rx_reserve / 2), _cap(pool_size) {
+    _ages = new AgeEntry[pool_size];
+    for (int i = 0; i < pool_size; i++) { _ages[i].pkt = NULL; _ages[i].scheduled_for = 0; }
+  }
 
   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");
+    int free_count = getFreeCount();
+    if (free_count < _emergency_floor
+        || (free_count < _rx_reserve && priority > MAX_PROTECTED_PRI)) {
+      MESH_DEBUG_PRINTLN("RxReservePacketManager: pool below RX reserve, shedding outbound (pri %d)", (int)priority);
       free(packet);
       return;
     }
+    recordAge(packet, scheduled_for);
     StaticPoolPacketManager::queueOutbound(packet, priority, scheduled_for);
   }
+
+  mesh::Packet* getNextOutbound(uint32_t now) override {
+    // Expire queued packets that have waited too long past their scheduled time.
+    for (int i = getOutboundTotal() - 1; i >= 0; i--) {
+      mesh::Packet* pkt = getOutboundByIdx(i);
+      uint32_t scheduled_for;
+      if (pkt && lookupAge(pkt, &scheduled_for)
+          && (int32_t)(now - scheduled_for) > (int32_t)STALE_OUTBOUND_MS) {
+        MESH_DEBUG_PRINTLN("RxReservePacketManager: dropping stale queued outbound");
+        removeOutboundByIdx(i);
+        free(pkt);
+      }
+    }
+    return StaticPoolPacketManager::getNextOutbound(now);
+  }
 };
 
 // The packet manager for an app build: observer builds reserve a quarter of the pool