From 63fb574c13368c0b355d080f92d1ec93bdc6845e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Br=C3=A1zio?= Date: Fri, 4 Sep 2026 19:36:34 +0100 Subject: [PATCH] Drop stale packets from the TX queue to prevent late/duplicate delivery Packets stuck past MAX_PACKET_QUEUE_AGE_MS (60s default) due to duty-cycle exhaustion are now dropped at head-of-line and via a per-loop sweep, so they no longer sit in the queue for hours and deliver long after their content stopped being useful. Their dedup entry is cleared through a new onPacketExpired() hook, so an in-flight copy arriving later via another path still gets forwarded instead of being silently suppressed. Outbound packets are stamped with their enqueue time (queued_at) and TX completion is logged for observability. --- src/Dispatcher.cpp | 50 ++++++++++++++++++++++++++++++++++++++++++++++ src/Dispatcher.h | 13 ++++++++++++ src/Mesh.cpp | 8 ++++++++ src/Mesh.h | 1 + src/Packet.cpp | 1 + src/Packet.h | 1 + 6 files changed, 74 insertions(+) diff --git a/src/Dispatcher.cpp b/src/Dispatcher.cpp index c0610b7f8a..537cd09381 100644 --- a/src/Dispatcher.cpp +++ b/src/Dispatcher.cpp @@ -12,6 +12,15 @@ namespace mesh { #define MIN_TX_BUDGET_RESERVE_MS 100 // min budget (ms) required before allowing next TX #define MIN_TX_BUDGET_AIRTIME_DIV 2 // require at least 1/N of estimated airtime as budget before TX +// Drop an outbound packet that has sat in the TX queue longer than this +// (e.g. due to duty-cycle exhaustion). Delivering a packet minutes/hours +// after it was queued is harmful: time-sensitive payloads (ACK/REQ/RESPONSE) +// are useless, and channel messages show up as if the user typed them long +// ago. Must sit above the worst-case budget-refill wait (~10-20s at 10%). +#ifndef MAX_PACKET_QUEUE_AGE_MS + #define MAX_PACKET_QUEUE_AGE_MS 60000 // 60 seconds +#endif + #ifndef NOISE_FLOOR_CALIB_INTERVAL #define NOISE_FLOOR_CALIB_INTERVAL 2000 // 2 seconds #endif @@ -19,6 +28,7 @@ namespace mesh { void Dispatcher::begin() { n_sent_flood = n_sent_direct = 0; n_recv_flood = n_recv_direct = 0; + n_expired = 0; _err_flags = 0; radio_nonrx_start = _ms->getMillis(); @@ -89,6 +99,11 @@ void Dispatcher::loop() { total_air_time += t; //Serial.print(" airtime="); Serial.println(t); + MESH_DEBUG_PRINTLN("%s Dispatcher::loop(): TX complete (len=%d, airtime=%ld ms)", + getLogDateTime(), + 2 + outbound->getPathByteLen() + outbound->payload_len, + t); + updateTxBudget(); if (t > tx_budget_ms) { @@ -143,6 +158,7 @@ void Dispatcher::loop() { } } checkRecv(); + expireAgedOutbound(); // drop stale packets (frees pool slots before head-of-line) checkSend(); } @@ -268,6 +284,7 @@ void Dispatcher::processRecvPacket(Packet* pkt) { uint8_t priority = (action >> 24) - 1; uint32_t _delay = action & 0xFFFFFF; + pkt->queued_at = _ms->getMillis(); _mgr->queueOutbound(pkt, priority, futureMillis(_delay)); } } @@ -305,6 +322,17 @@ void Dispatcher::checkSend() { cad_busy_start = 0; // reset busy state outbound = _mgr->getNextOutbound(_ms->getMillis()); + // Drop packets that have sat in the TX queue too long (e.g. due to duty-cycle + // exhaustion) before transmitting them stale. + while (outbound && (long)(_ms->getMillis() - outbound->queued_at) > (long)MAX_PACKET_QUEUE_AGE_MS) { + MESH_DEBUG_PRINTLN("%s Dispatcher::checkSend(): outbound packet expired in TX queue (age=%lu ms), dropping", + getLogDateTime(), (unsigned long)(_ms->getMillis() - outbound->queued_at)); + onPacketExpired(outbound); // let sub-class clear dedup so a later copy gets another chance + n_expired++; + _err_flags |= ERR_EVENT_PKT_EXPIRED; + releasePacket(outbound); + outbound = _mgr->getNextOutbound(_ms->getMillis()); + } if (outbound) { int len = 0; uint8_t raw[MAX_TRANS_UNIT]; @@ -368,11 +396,33 @@ void Dispatcher::releasePacket(Packet* packet) { _mgr->free(packet); } +// Drop outbound packets that have sat in the TX queue beyond the max age. +// Frees pool slots before head-of-line (reducing pool exhaustion and dedup +// table rollover) and prevents stale packets from being transmitted. +void Dispatcher::expireAgedOutbound() { + unsigned long now = _ms->getMillis(); + int n = _mgr->getOutboundTotal(); + for (int i = n - 1; i >= 0; i--) { // iterate backwards so removal keeps indices valid + Packet* pkt = _mgr->getOutboundByIdx(i); + if (pkt == NULL) continue; + if ((long)(now - pkt->queued_at) > (long)MAX_PACKET_QUEUE_AGE_MS) { + MESH_DEBUG_PRINTLN("%s Dispatcher::expireAgedOutbound(): packet expired in TX queue (age=%lu ms), dropping", + getLogDateTime(), (unsigned long)(now - pkt->queued_at)); + _mgr->removeOutboundByIdx(i); // remove from queue (returns the packet) + onPacketExpired(pkt); // let sub-class clear dedup so a later copy gets another chance + n_expired++; + _err_flags |= ERR_EVENT_PKT_EXPIRED; + releasePacket(pkt); + } + } +} + void Dispatcher::sendPacket(Packet* packet, uint8_t priority, uint32_t delay_millis) { if (!Packet::isValidPathLen(packet->path_len) || packet->payload_len > MAX_PACKET_PAYLOAD) { MESH_DEBUG_PRINTLN("%s Dispatcher::sendPacket(): ERROR: invalid packet... path_len=%d, payload_len=%d", getLogDateTime(), (uint32_t) packet->path_len, (uint32_t) packet->payload_len); _mgr->free(packet); } else { + packet->queued_at = _ms->getMillis(); _mgr->queueOutbound(packet, priority, futureMillis(delay_millis)); } } diff --git a/src/Dispatcher.h b/src/Dispatcher.h index aad6cba3ec..46d84c8ced 100644 --- a/src/Dispatcher.h +++ b/src/Dispatcher.h @@ -110,6 +110,7 @@ typedef uint32_t DispatcherAction; #define ERR_EVENT_FULL (1 << 0) #define ERR_EVENT_CAD_TIMEOUT (1 << 1) #define ERR_EVENT_STARTRX_TIMEOUT (1 << 2) +#define ERR_EVENT_PKT_EXPIRED (1 << 3) /** * \brief The low-level task that manages detecting incoming Packets, and the queueing @@ -125,12 +126,14 @@ class Dispatcher { bool prev_isrecv_mode; uint32_t n_sent_flood, n_sent_direct; uint32_t n_recv_flood, n_recv_direct; + uint32_t n_expired; unsigned long tx_budget_ms; unsigned long last_budget_update; unsigned long duty_cycle_window_ms; void processRecvPacket(Packet* pkt); void updateTxBudget(); + void expireAgedOutbound(); protected: PacketManager* _mgr; @@ -147,6 +150,7 @@ class Dispatcher { cad_busy_start = 0; next_floor_calib_time = next_agc_reset_time = 0; _err_flags = 0; + n_expired = 0; radio_nonrx_start = 0; prev_isrecv_mode = true; tx_budget_ms = 0; @@ -156,6 +160,13 @@ class Dispatcher { virtual DispatcherAction onRecvPacket(Packet* pkt) = 0; + /** + * \brief Hook invoked when an outbound packet is dropped (e.g. because it + * sat in the TX queue beyond the max age). Sub-classes can use this + * to update dedup tables so a later copy gets another chance. + */ + virtual void onPacketExpired(Packet* pkt) { } + virtual void logRxRaw(float snr, float rssi, const uint8_t raw[], int len) { } // custom hook virtual void logRx(Packet* packet, int len, float score) { } // hooks for custom logging @@ -187,8 +198,10 @@ class Dispatcher { uint32_t getNumSentDirect() const { return n_sent_direct; } uint32_t getNumRecvFlood() const { return n_recv_flood; } uint32_t getNumRecvDirect() const { return n_recv_direct; } + uint32_t getNumExpired() const { return n_expired; } void resetStats() { n_sent_flood = n_sent_direct = n_recv_flood = n_recv_direct = 0; + n_expired = 0; _err_flags = 0; } diff --git a/src/Mesh.cpp b/src/Mesh.cpp index c11f37cacf..d6796be339 100644 --- a/src/Mesh.cpp +++ b/src/Mesh.cpp @@ -38,6 +38,14 @@ int Mesh::searchChannelsByHash(const uint8_t* hash, GroupChannel channels[], int return 0; // not found } +void Mesh::onPacketExpired(Packet* pkt) { + // Packet was dropped from the TX queue (e.g. after sitting past the max age due + // to duty-cycle exhaustion). Clear it from the dedup table so that if a fresh + // copy arrives later (still being relayed via another path) it gets another + // chance to be forwarded, rather than being suppressed. + if (_tables) _tables->clear(pkt); +} + DispatcherAction Mesh::onRecvPacket(Packet* pkt) { if (pkt->isRouteDirect() && pkt->getPayloadType() == PAYLOAD_TYPE_TRACE) { if (pkt->path_len < MAX_PATH_SIZE) { diff --git a/src/Mesh.h b/src/Mesh.h index 49a299a6a4..e6f98210d6 100644 --- a/src/Mesh.h +++ b/src/Mesh.h @@ -36,6 +36,7 @@ class Mesh : public Dispatcher { protected: DispatcherAction onRecvPacket(Packet* pkt) override; + void onPacketExpired(Packet* pkt) override; virtual uint32_t getCADFailRetryDelay() const override; diff --git a/src/Packet.cpp b/src/Packet.cpp index aad3e2f48e..3e1999a3ca 100644 --- a/src/Packet.cpp +++ b/src/Packet.cpp @@ -8,6 +8,7 @@ Packet::Packet() { header = 0; path_len = 0; payload_len = 0; + queued_at = 0; } bool Packet::isValidPathLen(uint8_t path_len) { diff --git a/src/Packet.h b/src/Packet.h index c19d9e9d8f..1afe7d9029 100644 --- a/src/Packet.h +++ b/src/Packet.h @@ -49,6 +49,7 @@ class Packet { uint8_t path[MAX_PATH_SIZE]; uint8_t payload[MAX_PACKET_PAYLOAD]; int8_t _snr; + uint32_t queued_at; // (transient, not serialized) millis when enqueued for outbound TX /** * \brief calculate the hash of payload + type