diff --git a/src/mesh/FloodingRouter.cpp b/src/mesh/FloodingRouter.cpp index 14d17778e..dc211b20a 100644 --- a/src/mesh/FloodingRouter.cpp +++ b/src/mesh/FloodingRouter.cpp @@ -3,6 +3,10 @@ #include "NodeDB.h" #include "configuration.h" #include "mesh-pb-constants.h" +#include "meshUtils.h" +#if !MESHTASTIC_EXCLUDE_TRACEROUTE +#include "modules/TraceRouteModule.h" +#endif FloodingRouter::FloodingRouter() {} @@ -22,7 +26,37 @@ ErrorCode FloodingRouter::send(meshtastic_MeshPacket *p) bool FloodingRouter::shouldFilterReceived(const meshtastic_MeshPacket *p) { - if (wasSeenRecently(p)) { // Note: this will also add a recent packet record + bool wasUpgraded = false; + bool seenRecently = + wasSeenRecently(p, true, nullptr, nullptr, &wasUpgraded); // Updates history; returns false when an upgrade is detected + + // Handle hop_limit upgrade scenario for rebroadcasters + // isRebroadcaster() is duplicated in perhapsRebroadcast(), but this avoids confusing log messages + if (wasUpgraded && isRebroadcaster() && iface && p->hop_limit > 0) { + // wasSeenRecently() reports false in upgrade cases so we handle replacement before the duplicate short-circuit + // If we overhear a duplicate copy of the packet with more hops left than the one we are waiting to + // rebroadcast, then remove the packet currently sitting in the TX queue and use this one instead. + uint8_t dropThreshold = p->hop_limit; // remove queued packets that have fewer hops remaining + if (iface->removePendingTXPacket(getFrom(p), p->id, dropThreshold)) { + LOG_DEBUG("Processing upgraded packet 0x%08x for rebroadcast with hop limit %d (dropping queued < %d)", p->id, + p->hop_limit, dropThreshold); + + if (nodeDB) + nodeDB->updateFrom(*p); +#if !MESHTASTIC_EXCLUDE_TRACEROUTE + if (traceRouteModule && p->which_payload_variant == meshtastic_MeshPacket_decoded_tag && + p->decoded.portnum == meshtastic_PortNum_TRACEROUTE_APP) + traceRouteModule->processUpgradedPacket(*p); +#endif + + perhapsRebroadcast(p); + + // We already enqueued the improved copy, so make sure the incoming packet stops here. + return true; + } + } + + if (seenRecently) { printPacket("Ignore dupe incoming msg", p); rxDupe++; @@ -134,4 +168,4 @@ void FloodingRouter::sniffReceived(const meshtastic_MeshPacket *p, const meshtas // handle the packet as normal Router::sniffReceived(p, c); -} \ No newline at end of file +} diff --git a/src/mesh/MeshPacketQueue.cpp b/src/mesh/MeshPacketQueue.cpp index a64678a7f..ef5380eb8 100644 --- a/src/mesh/MeshPacketQueue.cpp +++ b/src/mesh/MeshPacketQueue.cpp @@ -103,12 +103,26 @@ meshtastic_MeshPacket *MeshPacketQueue::getFront() return p; } -/** Attempt to find and remove a packet from this queue. Returns a pointer to the removed packet, or NULL if not found */ -meshtastic_MeshPacket *MeshPacketQueue::remove(NodeNum from, PacketId id, bool tx_normal, bool tx_late) +/** Get a packet from this queue. Returns a pointer to the packet, or NULL if not found. */ +meshtastic_MeshPacket *MeshPacketQueue::getPacketFromQueue(NodeNum from, PacketId id) { for (auto it = queue.begin(); it != queue.end(); it++) { auto p = (*it); - if (getFrom(p) == from && p->id == id && ((tx_normal && !p->tx_after) || (tx_late && p->tx_after))) { + if (getFrom(p) == from && p->id == id) { + return p; + } + } + + return NULL; +} + +/** Attempt to find and remove a packet from this queue. Returns a pointer to the removed packet, or NULL if not found */ +meshtastic_MeshPacket *MeshPacketQueue::remove(NodeNum from, PacketId id, bool tx_normal, bool tx_late, uint8_t hop_limit_lt) +{ + for (auto it = queue.begin(); it != queue.end(); it++) { + auto p = (*it); + if (getFrom(p) == from && p->id == id && ((tx_normal && !p->tx_after) || (tx_late && p->tx_after)) && + (!hop_limit_lt || p->hop_limit < hop_limit_lt)) { queue.erase(it); return p; } @@ -120,14 +134,7 @@ meshtastic_MeshPacket *MeshPacketQueue::remove(NodeNum from, PacketId id, bool t /* Attempt to find a packet from this queue. Return true if it was found. */ bool MeshPacketQueue::find(const NodeNum from, const PacketId id) { - for (auto it = queue.begin(); it != queue.end(); it++) { - const auto *p = *it; - if (getFrom(p) == from && p->id == id) { - return true; - } - } - - return false; + return getPacketFromQueue(from, id) != NULL; } /** diff --git a/src/mesh/MeshPacketQueue.h b/src/mesh/MeshPacketQueue.h index 1b338f9ed..ea52eb5bf 100644 --- a/src/mesh/MeshPacketQueue.h +++ b/src/mesh/MeshPacketQueue.h @@ -35,8 +35,12 @@ class MeshPacketQueue meshtastic_MeshPacket *getFront(); + /** Get a packet from this queue. Returns a pointer to the packet, or NULL if not found. */ + meshtastic_MeshPacket *getPacketFromQueue(NodeNum from, PacketId id); + /** Attempt to find and remove a packet from this queue. Returns the packet which was removed from the queue */ - meshtastic_MeshPacket *remove(NodeNum from, PacketId id, bool tx_normal = true, bool tx_late = true); + meshtastic_MeshPacket *remove(NodeNum from, PacketId id, bool tx_normal = true, bool tx_late = true, + uint8_t hop_limit_lt = 0); /* Attempt to find a packet from this queue. Return true if it was found. */ bool find(const NodeNum from, const PacketId id); diff --git a/src/mesh/MeshService.cpp b/src/mesh/MeshService.cpp index 96782cda5..a138ad1d1 100644 --- a/src/mesh/MeshService.cpp +++ b/src/mesh/MeshService.cpp @@ -453,4 +453,4 @@ uint32_t MeshService::GetTimeSinceMeshPacket(const meshtastic_MeshPacket *mp) delta = 0; return delta; -} \ No newline at end of file +} diff --git a/src/mesh/MeshService.h b/src/mesh/MeshService.h index 5d074368f..66d9d9679 100644 --- a/src/mesh/MeshService.h +++ b/src/mesh/MeshService.h @@ -190,4 +190,4 @@ class MeshService friend class RoutingModule; }; -extern MeshService *service; \ No newline at end of file +extern MeshService *service; diff --git a/src/mesh/NextHopRouter.cpp b/src/mesh/NextHopRouter.cpp index 791b6a749..a90fe7ad2 100644 --- a/src/mesh/NextHopRouter.cpp +++ b/src/mesh/NextHopRouter.cpp @@ -1,4 +1,9 @@ #include "NextHopRouter.h" +#include "MeshTypes.h" +#include "meshUtils.h" +#if !MESHTASTIC_EXCLUDE_TRACEROUTE +#include "modules/TraceRouteModule.h" +#endif #include "NodeDB.h" NextHopRouter::NextHopRouter() {} @@ -33,7 +38,35 @@ bool NextHopRouter::shouldFilterReceived(const meshtastic_MeshPacket *p) { bool wasFallback = false; bool weWereNextHop = false; - if (wasSeenRecently(p, true, &wasFallback, &weWereNextHop)) { // Note: this will also add a recent packet record + bool wasUpgraded = false; + bool seenRecently = wasSeenRecently(p, true, &wasFallback, &weWereNextHop, + &wasUpgraded); // Updates history; returns false when an upgrade is detected + + // Handle hop_limit upgrade scenario for rebroadcasters + // isRebroadcaster() is duplicated in perhapsRelay(), but this avoids confusing log messages + if (wasUpgraded && isRebroadcaster() && iface && p->hop_limit > 0) { + // Upgrade detection bypasses the duplicate short-circuit so we replace the queued packet before exiting + uint8_t dropThreshold = p->hop_limit; // remove queued packets that have fewer hops remaining + if (iface->removePendingTXPacket(getFrom(p), p->id, dropThreshold)) { + LOG_DEBUG("Processing upgraded packet 0x%08x for relay with hop limit %d (dropping queued < %d)", p->id, p->hop_limit, + dropThreshold); + + if (nodeDB) + nodeDB->updateFrom(*p); +#if !MESHTASTIC_EXCLUDE_TRACEROUTE + if (traceRouteModule && p->which_payload_variant == meshtastic_MeshPacket_decoded_tag && + p->decoded.portnum == meshtastic_PortNum_TRACEROUTE_APP) + traceRouteModule->processUpgradedPacket(*p); +#endif + + perhapsRelay(p); + + // We already enqueued the improved copy, so make sure the incoming packet stops here. + return true; + } + } + + if (seenRecently) { printPacket("Ignore dupe incoming msg", p); if (p->transport_mechanism == meshtastic_MeshPacket_TransportMechanism_TRANSPORT_LORA) { @@ -298,4 +331,4 @@ void NextHopRouter::setNextTx(PendingPacket *pending) LOG_DEBUG("Setting next retransmission in %u msecs: ", d); printPacket("", pending->packet); setReceivedMessage(); // Run ASAP, so we can figure out our correct sleep time -} \ No newline at end of file +} diff --git a/src/mesh/PacketHistory.cpp b/src/mesh/PacketHistory.cpp index 735386d79..6d14856da 100644 --- a/src/mesh/PacketHistory.cpp +++ b/src/mesh/PacketHistory.cpp @@ -45,7 +45,8 @@ PacketHistory::~PacketHistory() } /** Update recentPackets and return true if we have already seen this packet */ -bool PacketHistory::wasSeenRecently(const meshtastic_MeshPacket *p, bool withUpdate, bool *wasFallback, bool *weWereNextHop) +bool PacketHistory::wasSeenRecently(const meshtastic_MeshPacket *p, bool withUpdate, bool *wasFallback, bool *weWereNextHop, + bool *wasUpgraded) { if (!initOk()) { LOG_ERROR("Packet History - Was Seen Recently: NOT INITIALIZED!"); @@ -66,6 +67,7 @@ bool PacketHistory::wasSeenRecently(const meshtastic_MeshPacket *p, bool withUpd r.id = p->id; r.sender = getFrom(p); // If 0 then use our ID r.next_hop = p->next_hop; + r.hop_limit = p->hop_limit; r.relayed_by[0] = p->relay_node; r.rxTimeMsec = millis(); // @@ -81,6 +83,16 @@ bool PacketHistory::wasSeenRecently(const meshtastic_MeshPacket *p, bool withUpd PacketRecord *found = find(r.sender, r.id); // Find the packet record in the recentPackets array bool seenRecently = (found != NULL); // If found -> the packet was seen recently + // Check for hop_limit upgrade scenario + if (seenRecently && wasUpgraded && found->hop_limit < p->hop_limit) { + LOG_DEBUG("Packet History - Hop limit upgrade: packet 0x%08x from hop_limit=%d to hop_limit=%d", p->id, found->hop_limit, + p->hop_limit); + *wasUpgraded = true; + seenRecently = false; // Allow router processing but prevent duplicate app delivery + } else if (wasUpgraded) { + *wasUpgraded = false; // Initialize to false if not an upgrade + } + if (seenRecently) { uint8_t ourRelayID = nodeDB->getLastByteOfNodeNum(nodeDB->getNodeNum()); // Get our relay ID from our node number @@ -126,6 +138,11 @@ bool PacketHistory::wasSeenRecently(const meshtastic_MeshPacket *p, bool withUpd millis() - found->rxTimeMsec); #endif + // Preserve the highest hop_limit we've ever seen for this packet so upgrades aren't lost when a later copy has + // fewer hops remaining. + if (found->hop_limit > r.hop_limit) + r.hop_limit = found->hop_limit; + // Add the existing relayed_by to the new record for (uint8_t i = 0; i < (NUM_RELAYERS - 1); i++) { if (found->relayed_by[i] != 0) diff --git a/src/mesh/PacketHistory.h b/src/mesh/PacketHistory.h index 4b53c8f6a..3ff7a893d 100644 --- a/src/mesh/PacketHistory.h +++ b/src/mesh/PacketHistory.h @@ -16,8 +16,9 @@ class PacketHistory PacketId id; uint32_t rxTimeMsec; // Unix time in msecs - the time we received it, 0 means empty uint8_t next_hop; // The next hop asked for this packet + uint8_t hop_limit; // Highest hop limit observed for this packet uint8_t relayed_by[NUM_RELAYERS]; // Array of nodes that relayed this packet - }; // 4B + 4B + 4B + 1B + 3B = 16B + }; // 4B + 4B + 4B + 1B + 1B + 3B = 17B (will be padded to 20B) uint32_t recentPacketsCapacity = 0; // Can be set in constructor, no need to recompile. Used to allocate memory for mx_recentPackets. @@ -50,9 +51,10 @@ class PacketHistory * @param withUpdate if true and not found we add an entry to recentPackets * @param wasFallback if not nullptr, packet will be checked for fallback to flooding and value will be set to true if so * @param weWereNextHop if not nullptr, packet will be checked for us being the next hop and value will be set to true if so + * @param wasUpgraded if not nullptr, will be set to true if this packet has better hop_limit than previously seen */ bool wasSeenRecently(const meshtastic_MeshPacket *p, bool withUpdate = true, bool *wasFallback = nullptr, - bool *weWereNextHop = nullptr); + bool *weWereNextHop = nullptr, bool *wasUpgraded = nullptr); /* Check if a certain node was a relayer of a packet in the history given an ID and sender * If wasSole is not nullptr, it will be set to true if the relayer was the only relayer of that packet diff --git a/src/mesh/RadioInterface.h b/src/mesh/RadioInterface.h index eff284747..0c5b6cd1a 100644 --- a/src/mesh/RadioInterface.h +++ b/src/mesh/RadioInterface.h @@ -189,6 +189,12 @@ class RadioInterface /** If the packet is not already in the late rebroadcast window, move it there */ virtual void clampToLateRebroadcastWindow(NodeNum from, PacketId id) { return; } + /** + * If there is a packet pending TX in the queue with a worse hop limit, remove it pending replacement with a better version + * @return Whether a pending packet was removed + */ + virtual bool removePendingTXPacket(NodeNum from, PacketId id, uint32_t hop_limit_lt) { return false; } + /** * Calculate airtime per * https://www.rs-online.com/designspark/rel-assets/ds-assets/uploads/knowledge-items/application-notes-for-the-internet-of-things/LoRa%20Design%20Guide.pdf @@ -266,4 +272,4 @@ class RadioInterface }; /// Debug printing for packets -void printPacket(const char *prefix, const meshtastic_MeshPacket *p); \ No newline at end of file +void printPacket(const char *prefix, const meshtastic_MeshPacket *p); diff --git a/src/mesh/RadioLibInterface.cpp b/src/mesh/RadioLibInterface.cpp index 19d0f794a..3717e8780 100644 --- a/src/mesh/RadioLibInterface.cpp +++ b/src/mesh/RadioLibInterface.cpp @@ -362,6 +362,26 @@ void RadioLibInterface::clampToLateRebroadcastWindow(NodeNum from, PacketId id) } } +/** + * If there is a packet pending TX in the queue with a worse hop limit, remove it pending replacement with a better version + * @return Whether a pending packet was removed + */ +bool RadioLibInterface::removePendingTXPacket(NodeNum from, PacketId id, uint32_t hop_limit_lt) +{ + meshtastic_MeshPacket *p = txQueue.remove(from, id, true, true, hop_limit_lt); + if (p) { + LOG_DEBUG("Dropping pending-TX packet 0x%08x with hop limit %d", p->id, p->hop_limit); + packetPool.release(p); + return true; + } + return false; +} + +/** + * Remove a packet that is eligible for replacement from the TX queue + */ +// void RadioLibInterface::removePending + void RadioLibInterface::handleTransmitInterrupt() { // This can be null if we forced the device to enter standby mode. In that case diff --git a/src/mesh/RadioLibInterface.h b/src/mesh/RadioLibInterface.h index 9f497812f..3444b1a2c 100644 --- a/src/mesh/RadioLibInterface.h +++ b/src/mesh/RadioLibInterface.h @@ -215,4 +215,11 @@ class RadioLibInterface : public RadioInterface, protected concurrency::Notified * If the packet is not already in the late rebroadcast window, move it there */ void clampToLateRebroadcastWindow(NodeNum from, PacketId id); -}; \ No newline at end of file + + /** + * If there is a packet pending TX in the queue with a worse hop limit, remove it pending replacement with a better version + * @return Whether a pending packet was removed + */ + + bool removePendingTXPacket(NodeNum from, PacketId id, uint32_t hop_limit_lt) override; +}; diff --git a/src/mesh/Router.h b/src/mesh/Router.h index 075248af9..92a5a06e5 100644 --- a/src/mesh/Router.h +++ b/src/mesh/Router.h @@ -174,4 +174,4 @@ PacketId generatePacketId(); #define BITFIELD_WANT_RESPONSE_SHIFT 1 #define BITFIELD_OK_TO_MQTT_SHIFT 0 #define BITFIELD_WANT_RESPONSE_MASK (1 << BITFIELD_WANT_RESPONSE_SHIFT) -#define BITFIELD_OK_TO_MQTT_MASK (1 << BITFIELD_OK_TO_MQTT_SHIFT) \ No newline at end of file +#define BITFIELD_OK_TO_MQTT_MASK (1 << BITFIELD_OK_TO_MQTT_SHIFT) diff --git a/src/modules/TextMessageModule.cpp b/src/modules/TextMessageModule.cpp index 72df330c5..aee359158 100644 --- a/src/modules/TextMessageModule.cpp +++ b/src/modules/TextMessageModule.cpp @@ -13,6 +13,7 @@ ProcessMessage TextMessageModule::handleReceived(const meshtastic_MeshPacket &mp auto &p = mp.decoded; LOG_INFO("Received text msg from=0x%0x, id=0x%x, msg=%.*s", mp.from, mp.id, p.payload.size, p.payload.bytes); #endif + // We only store/display messages destined for us. // Keep a copy of the most recent text message. devicestate.rx_text_message = mp; @@ -30,4 +31,4 @@ ProcessMessage TextMessageModule::handleReceived(const meshtastic_MeshPacket &mp bool TextMessageModule::wantPacket(const meshtastic_MeshPacket *p) { return MeshService::isTextPayload(p); -} \ No newline at end of file +} diff --git a/src/modules/TraceRouteModule.cpp b/src/modules/TraceRouteModule.cpp index d7df90bb5..8f69c504a 100644 --- a/src/modules/TraceRouteModule.cpp +++ b/src/modules/TraceRouteModule.cpp @@ -153,6 +153,20 @@ void TraceRouteModule::alterReceivedProtobuf(meshtastic_MeshPacket &p, meshtasti } } +void TraceRouteModule::processUpgradedPacket(const meshtastic_MeshPacket &mp) +{ + if (mp.which_payload_variant != meshtastic_MeshPacket_decoded_tag || mp.decoded.portnum != meshtastic_PortNum_TRACEROUTE_APP) + return; + + meshtastic_RouteDiscovery decoded = meshtastic_RouteDiscovery_init_zero; + if (!pb_decode_from_bytes(mp.decoded.payload.bytes, mp.decoded.payload.size, &meshtastic_RouteDiscovery_msg, &decoded)) + return; + + handleReceivedProtobuf(mp, &decoded); + // Intentionally modify the packet in-place so downstream relays see our updates. + alterReceivedProtobuf(const_cast(mp), &decoded); +} + void TraceRouteModule::insertUnknownHops(meshtastic_MeshPacket &p, meshtastic_RouteDiscovery *r, bool isTowardsDestination) { pb_size_t *route_count; @@ -760,4 +774,4 @@ int32_t TraceRouteModule::runOnce() } return INT32_MAX; -} \ No newline at end of file +} diff --git a/src/modules/TraceRouteModule.h b/src/modules/TraceRouteModule.h index 51d98826e..a069f7157 100644 --- a/src/modules/TraceRouteModule.h +++ b/src/modules/TraceRouteModule.h @@ -35,6 +35,8 @@ class TraceRouteModule : public ProtobufModule, virtual bool wantUIFrame() override { return shouldDraw(); } virtual Observable *getUIFrameObservable() override { return this; } + void processUpgradedPacket(const meshtastic_MeshPacket &mp); + protected: bool handleReceivedProtobuf(const meshtastic_MeshPacket &mp, meshtastic_RouteDiscovery *r) override; @@ -70,4 +72,4 @@ class TraceRouteModule : public ProtobufModule, bool initialized = false; }; -extern TraceRouteModule *traceRouteModule; \ No newline at end of file +extern TraceRouteModule *traceRouteModule;