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.
This commit is contained in:
agessaman committed 2026-07-09 19:06:41 -07:00
1 parent 1c9c6292e5
commit 847be34e7d
3 files changed
+84 -20

No files matched your search

+10 -5
View File
@@ -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
+8 -5
View File
@@ -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)
+66 -10
View File
@@ -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