From b87792edb75d236108941c7658df54a7ecbf379a Mon Sep 17 00:00:00 2001 From: Martin Dobrev Date: Sat, 29 Aug 2026 23:40:54 +0100 Subject: [PATCH] Send the announces that outbound() queues Transport::outbound() queues an announce when an interface's announce cap says it may not go out yet, and refuses to transmit any announce on an interface that has one queued. Nothing drained the queue: Interface::process_announce_queue() was an empty body around commented Python, and no caller existed. The first announce an interface ever queued therefore stopped that interface forwarding announces for the rest of the uptime, with no log line and no way back short of a restart. Port the reference implementation: expire entries past QUEUED_ANNOUNCE_LIFE, then send the queued announce with the fewest hops, oldest first, and set announce_allowed_at from the cap. RNS arms a timer for the next one; there are no timers here, so Transport::jobs() polls every interface in its existing interface-jobs block and the cap alone decides whether an announce leaves. The cap's wait moves to Interface::announce_wait_time(), which outbound() and the recursive path request now ask instead of each recomputing it. Both computed it in integers, where the airtime of a packet on anything faster than a slow LoRa channel truncates to zero and the cap never engages; it is a double now, which is what makes the queue reachable on the interfaces the cap was meant for. Tested with four cases added to test_transport, covering hop order, the cap holding the rest of the queue back, the drain once the wait has elapsed, stale expiry and an interface with no bitrate. All four fail against the empty implementation and pass with it; the rest of test_transport is unaffected. Also built into an ESP32-S3 firmware (Arduino core 3 / IDF 5) that registers LoRa, TCP and AutoInterface interfaces. --- src/microReticulum/Interface.cpp | 107 +++++++++------- src/microReticulum/Interface.h | 16 +++ src/microReticulum/Transport.cpp | 30 +++-- test/test_transport/test_transport.cpp | 165 +++++++++++++++++++++++++ 4 files changed, 262 insertions(+), 56 deletions(-) diff --git a/src/microReticulum/Interface.cpp b/src/microReticulum/Interface.cpp index 46cd09d5..940b0bca 100644 --- a/src/microReticulum/Interface.cpp +++ b/src/microReticulum/Interface.cpp @@ -16,9 +16,11 @@ #include "Identity.h" #include "Transport.h" +#include "Utilities/OS.h" using namespace RNS; using namespace RNS::Type::Interface; +using namespace RNS::Utilities; /*static*/ uint8_t Interface::DISCOVER_PATHS_FOR = MODE_ACCESS_POINT | MODE_GATEWAY | MODE_ROAMING; @@ -86,51 +88,68 @@ void Interface::handle_incoming(const Bytes& data) { } } +// Send an announce that Transport::outbound had to queue, as often as the +// interface's announce cap allows one out. +// +// RNS arms a timer for the cap's wait and sends the next announce when it +// fires. There are no timers here, so Transport::jobs calls this on every +// interface and the cap is kept by _announce_allowed_at instead: the interval +// between calls decides only how promptly a queue drains, never how fast +// announces leave. One per call is what the timer does per firing, and is far +// more than the arrival rate a queue forms from. +// +// While this had no body nothing ever drained the queue, and outbound() +// refuses to transmit an announce onto an interface that has any queued: the +// first announce an interface ever queued stopped it forwarding announces for +// the rest of the uptime, silently and with no way back short of a reboot. void Interface::process_announce_queue() { -/* - if not hasattr(self, "announce_cap"): - self.announce_cap = RNS.Reticulum.ANNOUNCE_CAP - - if hasattr(self, "announce_queue"): - try: - now = time.time() - stale = [] - for a in self.announce_queue: - if now > a["time"]+RNS.Reticulum.QUEUED_ANNOUNCE_LIFE: - stale.append(a) - - for s in stale: - if s in self.announce_queue: - self.announce_queue.remove(s) - - if len(self.announce_queue) > 0: - min_hops = min(entry["hops"] for entry in self.announce_queue) - entries = list(filter(lambda e: e["hops"] == min_hops, self.announce_queue)) - entries.sort(key=lambda e: e["time"]) - selected = entries[0] - - double now = OS::time(); - uint32_t wait_time = 0; - if (_impl->_bitrate > 0 && _impl->_announce_cap > 0) { - uint32_t tx_time = (len(selected["raw"])*8) / _impl->_bitrate; - wait_time = (tx_time / _impl->_announce_cap); - } - _impl->_announce_allowed_at = now + wait_time; - - self.on_outgoing(selected["raw"]) - - if selected in self.announce_queue: - self.announce_queue.remove(selected) - - if len(self.announce_queue) > 0: - timer = threading.Timer(wait_time, self.process_announce_queue) - timer.start() - - except Exception as e: - self.announce_queue = [] - RNS.log("Error while processing announce queue on "+str(self)+". The contained exception was: "+str(e), RNS.LOG_ERROR) - RNS.log("The announce queue for this interface has been cleared.", RNS.LOG_ERROR) -*/ + assert(_impl); + if (_impl->_announce_queue.empty()) return; + + try { + double now = OS::time(); + + // An announce that has been waiting this long is not worth the airtime + // any more: whoever sent it has almost certainly announced again since. + size_t held = _impl->_announce_queue.size(); + _impl->_announce_queue.remove_if([now](const AnnounceEntry& entry) { + return now > entry._time + Type::Reticulum::QUEUED_ANNOUNCE_LIFE; + }); + if (_impl->_announce_queue.size() < held) { + DEBUGF("Dropped %u stale queued announce(s) on %s", + (unsigned)(held - _impl->_announce_queue.size()), toString().c_str()); + } + if (_impl->_announce_queue.empty() || now < _impl->_announce_allowed_at) return; + + // Fewest hops first, and among those the one that has waited longest, + // so a queue draining slowly still carries the nearest path onwards + // first and nothing in it can be starved by later arrivals. + auto selected = _impl->_announce_queue.begin(); + for (auto entry = _impl->_announce_queue.begin(); entry != _impl->_announce_queue.end(); ++entry) { + if (entry->_hops < selected->_hops || + (entry->_hops == selected->_hops && entry->_time < selected->_time)) { + selected = entry; + } + } + + double wait_time = announce_wait_time(selected->_raw.size()); + _impl->_announce_allowed_at = now + wait_time; + + Bytes raw(selected->_raw); + _impl->_announce_queue.erase(selected); + DEBUGF("Sending queued announce on %s, %u still queued, next allowed in %.1f s", + toString().c_str(), (unsigned)_impl->_announce_queue.size(), wait_time); + + if (Transport::transmit(*this, raw)) { + sent_announce(); + } + } + catch (const std::exception& e) { + _impl->_announce_queue.clear(); + ERRORF("Error while processing the announce queue on %s. The contained exception was: %s", + toString().c_str(), e.what()); + ERROR("The announce queue for this interface has been cleared."); + } } /* diff --git a/src/microReticulum/Interface.h b/src/microReticulum/Interface.h index 0d939b9b..099a84ca 100644 --- a/src/microReticulum/Interface.h +++ b/src/microReticulum/Interface.h @@ -210,6 +210,22 @@ namespace RNS { inline const Bytes get_hash() const { assert(_impl); return _impl->get_hash(); } void process_announce_queue(); + // How long the announce cap makes this interface wait after sending an + // announce of this size: the packet's airtime divided by the share of + // the link announces are allowed to take. Both the callers that need it + // -- Transport::outbound when it lets an announce through, and + // process_announce_queue when it sends a queued one -- ask here, so the + // two cannot disagree about when the next announce is due. Answered in + // seconds, in floating point: computed in integers, as one caller used + // to, every wait below one second rounds to none at all, which is every + // wait an interface faster than a slow LoRa channel ever has. + inline double announce_wait_time(size_t tx_size) const { + assert(_impl); + if (_impl->_bitrate == 0 || _impl->_announce_cap <= 0.0) return 0.0; + double tx_time = (double)(tx_size * 8) / (double)_impl->_bitrate; + return tx_time / (double)_impl->_announce_cap; + } + // CBA ACCUMULATES inline void add_announce(AnnounceEntry& entry) { assert(_impl); _impl->_announce_queue.push_back(entry); } diff --git a/src/microReticulum/Transport.cpp b/src/microReticulum/Transport.cpp index 89b5238b..6fc094b8 100644 --- a/src/microReticulum/Transport.cpp +++ b/src/microReticulum/Transport.cpp @@ -1007,6 +1007,22 @@ TRACEF("path_request_conditions=%u", path_request_conditions); // Run interface-related jobs if (OS::time() > (_interface_last_jobs + _interface_jobs_interval)) { prioritize_interfaces(); + // Announces that outbound() had to queue are sent from here. + // RNS arms a timer per interface for this; this port has none, + // so the queues are polled and each interface's own announce + // cap decides whether anything actually goes out. Nothing + // called process_announce_queue() at all before, and outbound() + // will not transmit an announce while an interface has one + // queued -- so an interface that queued a single announce + // forwarded no further announce for the rest of the uptime. + try { + for (auto& interface : _interfaces) { + interface.process_announce_queue(); + } + } + catch (const std::exception& e) { + ERRORF("Error while processing queued per-interface announces: %s", e.what()); + } // TODO /* try { @@ -1393,12 +1409,7 @@ TRACEF("path_request_conditions=%u", path_request_conditions); bool queued_announces = (interface.announce_queue().size() > 0); if (!queued_announces && outbound_time > interface.announce_allowed_at()) { - uint16_t wait_time = 0; - if (interface.bitrate() > 0 && interface.announce_cap() > 0) { - uint16_t tx_time = (packet.raw().size() * 8) / interface.bitrate(); - wait_time = (tx_time / interface.announce_cap()); - } - interface.announce_allowed_at(outbound_time + wait_time); + interface.announce_allowed_at(outbound_time + interface.announce_wait_time(packet.raw().size())); } else { should_transmit = false; @@ -3784,12 +3795,7 @@ will announce it. return; } else { - //p tx_time = ((len(path_request_data)+RNS.Reticulum.HEADER_MINSIZE)*8) / on_interface.bitrate - uint32_t wait_time = 0; - if ( on_interface.bitrate() > 0 && on_interface.announce_cap() > 0) { - uint32_t tx_time = ((path_request_data.size() + Type::Reticulum::HEADER_MINSIZE)*8) / on_interface.bitrate(); - wait_time = (tx_time / on_interface.announce_cap()); - } + double wait_time = on_interface.announce_wait_time(path_request_data.size() + Type::Reticulum::HEADER_MINSIZE); const_cast(on_interface).announce_allowed_at(now + wait_time); } } diff --git a/test/test_transport/test_transport.cpp b/test/test_transport/test_transport.cpp index 9d956433..b2df1612 100644 --- a/test/test_transport/test_transport.cpp +++ b/test/test_transport/test_transport.cpp @@ -631,6 +631,165 @@ void test_receipt_timeout_handler_capture() { void setUp(void) {} void tearDown(void) {} +// ============================================================================ +// Announce queue +// ============================================================================ + +// Counts what it is asked to send and keeps the last of it, so a test can say +// which queued announce went out and when. Bitrate and announce cap are what +// decide the wait between two announces, so each test sets the pair its case +// is about. +class QueueInterface : public RNS::InterfaceImpl { +public: + QueueInterface(uint32_t bitrate, float cap, const char* name = "QueueInterface") + : RNS::InterfaceImpl(name) { + _OUT = true; + _IN = false; + _bitrate = bitrate; + _announce_cap = cap; + } + virtual ~QueueInterface() { _name = "(deleted)"; } + virtual bool send_outgoing(const RNS::Bytes& data) { + _sent++; + _last = data; + return true; + } +public: + size_t _sent = 0; + RNS::Bytes _last; +}; + +// A queued announce of a given age and hop count, its raw bytes filled with a +// marker so a test can tell which one was sent. +static RNS::AnnounceEntry test_announce(uint8_t hops, double time, uint8_t marker) { + uint8_t raw[100]; + memset(raw, marker, sizeof(raw)); + uint8_t destination[1] = { marker }; + return RNS::AnnounceEntry(RNS::Bytes(destination, sizeof(destination)), time, hops, 0, + RNS::Bytes(raw, sizeof(raw))); +} + +// The fewest hops leaves first, and the cap holds the rest back. Both halves +// matter: without the first the queue is a queue in name only, and without the +// second an interface can be talked into spending its whole link on announces. +void test_announce_queue_sends_fewest_hops_first() { + + printf("test_announce_queue_sends_fewest_hops_first: BEGIN\n"); + + // 1200 bits/s at a 2 % cap: a 100-byte announce owes about 33 s before the + // next one may go. Computed in integers, as this used to be, the whole wait + // truncates to nothing and the cap never engages at all. + QueueInterface* impl = new QueueInterface(1200, 0.02f); + RNS::Interface iface(impl); + + double now = RNS::Utilities::OS::time(); + RNS::AnnounceEntry far(test_announce(3, now, 'f')); + RNS::AnnounceEntry near(test_announce(1, now, 'n')); + RNS::AnnounceEntry mid(test_announce(2, now, 'm')); + iface.add_announce(far); + iface.add_announce(near); + iface.add_announce(mid); + TEST_ASSERT_EQUAL_UINT32(3, iface.announce_queue().size()); + + iface.process_announce_queue(); + + TEST_ASSERT_EQUAL_UINT32(1, impl->_sent); + TEST_ASSERT_EQUAL_UINT32(2, iface.announce_queue().size()); + TEST_ASSERT_EQUAL_UINT8('n', impl->_last.data()[0]); + // About 33 s of it: 800 bits at 1200 bits/s is two thirds of a second of + // airtime, and a 2 % cap is fifty times that. Unity's double assertions + // are compiled out here, so the comparison is a plain one. + double wait = iface.announce_allowed_at() - now; + TEST_ASSERT_TRUE(wait > 30.0 && wait < 36.0); + + // Called again straight away, the cap has not elapsed and nothing else goes. + iface.process_announce_queue(); + TEST_ASSERT_EQUAL_UINT32(1, impl->_sent); + TEST_ASSERT_EQUAL_UINT32(2, iface.announce_queue().size()); + + printf("test_announce_queue_sends_fewest_hops_first: END\n"); +} + +// The queue drains once the cap's wait has passed. Nothing drained it at all +// before -- and because Transport::outbound refuses to transmit an announce on +// an interface that has one queued, a single queued announce used to stop that +// interface forwarding announces for the rest of the uptime. +void test_announce_queue_drains_when_the_cap_allows() { + + printf("test_announce_queue_drains_when_the_cap_allows: BEGIN\n"); + + // 10 Mbit/s at 2 %: a 100-byte announce owes 4 ms. + QueueInterface* impl = new QueueInterface(10000000, 0.02f); + RNS::Interface iface(impl); + + double now = RNS::Utilities::OS::time(); + RNS::AnnounceEntry first(test_announce(1, now, 'a')); + RNS::AnnounceEntry second(test_announce(1, now + 0.001, 'b')); + iface.add_announce(first); + iface.add_announce(second); + + iface.process_announce_queue(); + TEST_ASSERT_EQUAL_UINT32(1, impl->_sent); + TEST_ASSERT_EQUAL_UINT8('a', impl->_last.data()[0]); // same hops: oldest first + + RNS::Utilities::OS::sleep(0.02); + iface.process_announce_queue(); + TEST_ASSERT_EQUAL_UINT32(2, impl->_sent); + TEST_ASSERT_EQUAL_UINT8('b', impl->_last.data()[0]); + TEST_ASSERT_EQUAL_UINT32(0, iface.announce_queue().size()); + + // An empty queue is not an error, and asks nothing of the interface. + iface.process_announce_queue(); + TEST_ASSERT_EQUAL_UINT32(2, impl->_sent); + + printf("test_announce_queue_drains_when_the_cap_allows: END\n"); +} + +// An announce that has waited longer than a queued announce may live is not +// worth the airtime: whoever sent it has announced again several times since. +void test_stale_queued_announces_are_dropped() { + + printf("test_stale_queued_announces_are_dropped: BEGIN\n"); + + QueueInterface* impl = new QueueInterface(10000000, 0.02f); + RNS::Interface iface(impl); + + double now = RNS::Utilities::OS::time(); + RNS::AnnounceEntry stale(test_announce(1, now - (double)RNS::Type::Reticulum::QUEUED_ANNOUNCE_LIFE - 1, 's')); + RNS::AnnounceEntry fresh(test_announce(1, now, 'f')); + iface.add_announce(stale); + iface.add_announce(fresh); + + iface.process_announce_queue(); + + TEST_ASSERT_EQUAL_UINT32(1, impl->_sent); + TEST_ASSERT_EQUAL_UINT8('f', impl->_last.data()[0]); + TEST_ASSERT_EQUAL_UINT32(0, iface.announce_queue().size()); + + printf("test_stale_queued_announces_are_dropped: END\n"); +} + +// An interface that has not said how fast it is owes no wait, and must still +// drain rather than sit on the queue forever. +void test_announce_queue_drains_without_a_bitrate() { + + printf("test_announce_queue_drains_without_a_bitrate: BEGIN\n"); + + QueueInterface* impl = new QueueInterface(0, 0.02f); + RNS::Interface iface(impl); + + double now = RNS::Utilities::OS::time(); + RNS::AnnounceEntry only(test_announce(1, now, 'o')); + iface.add_announce(only); + + iface.process_announce_queue(); + + TEST_ASSERT_EQUAL_UINT32(1, impl->_sent); + TEST_ASSERT_EQUAL_UINT32(0, iface.announce_queue().size()); + + printf("test_announce_queue_drains_without_a_bitrate: END\n"); +} + int runUnityTests(void) { UNITY_BEGIN(); @@ -658,6 +817,12 @@ int runUnityTests(void) { */ RUN_TEST(test_prioritize_interfaces); RUN_TEST(test_incoming_announce_over_limit); + + // Announce queue + RUN_TEST(test_announce_queue_sends_fewest_hops_first); + RUN_TEST(test_announce_queue_drains_when_the_cap_allows); + RUN_TEST(test_stale_queued_announces_are_dropped); + RUN_TEST(test_announce_queue_drains_without_a_bitrate); //RUN_TEST(test_incoming_announce_stress); #if RNS_NEIGHBOR_PROBING