From 847be34e7d28bb620f515fa46649a8edc13d11b9 Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 9 Jul 2026 19:06:11 -0700 Subject: [PATCH] =?UTF-8?q?fix(mqtt):=20keep=20throttled=20nodes=20adminis?= =?UTF-8?q?trable=20=E2=80=94=20priority-aware=20shed=20+=20stale=20expiry?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- MQTT_INTERNALS.md | 15 ++++-- RESTORE_UPSTREAM_NOTES.md | 13 +++-- src/helpers/RxReservePacketManager.h | 76 ++++++++++++++++++++++++---- 3 files changed, 84 insertions(+), 20 deletions(-) diff --git a/MQTT_INTERNALS.md b/MQTT_INTERNALS.md index 1da44129f..fbe30d706 100644 --- a/MQTT_INTERNALS.md +++ b/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 diff --git a/RESTORE_UPSTREAM_NOTES.md b/RESTORE_UPSTREAM_NOTES.md index 1b05bd3af..25030f25f 100644 --- a/RESTORE_UPSTREAM_NOTES.md +++ b/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) diff --git a/src/helpers/RxReservePacketManager.h b/src/helpers/RxReservePacketManager.h index 61865e95b..96d668270 100644 --- a/src/helpers/RxReservePacketManager.h +++ b/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