Sync nuclear subtree from NUClear@2053a375

An unacknowledged reliable send now disconnects the peer rather than
silently dropping the data, so a nuclear_leave is emitted where the
message would previously have gone missing without any indication.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: Trent Houliston

README.md

@@ -8,7 +8,7 @@ Node.js module for interacting with the [NUClear](https://github.com/Fastcode/NU
 
 Version 2 uses the redesigned **NUClearNet** library from [NUClear PR #190](https://github.com/Fastcode/NUClear/pull/190) (wire protocol **0x03**). It is **not** compatible with 1.x clients or NUClear builds that still use the old `NUClearNetwork` stack (protocol 0x02). Upgrade Node clients and NUClear robots together.
 
-The vendored NUClear tree is updated via `git subtree` from the `houliston/nuclearnet-v2` branch (currently [NUClear@62bc83a1](https://github.com/Fastcode/NUClear/commit/62bc83a1)).
+The vendored NUClear tree is updated via `git subtree` from the `houliston/nuclearnet-v2` branch (currently [NUClear@2053a375](https://github.com/Fastcode/NUClear/commit/2053a375)).
 
 Peer join events may arrive slightly later than in 1.x because connection requires both multicast announce and a unicast CONNECT handshake.
 

src/nuclear/src/nuclearnet/NUClearNet.cpp

@@ -385,7 +385,24 @@ namespace {
         discovery->check_timeouts(now);
 
         // Check for retransmissions
-        auto retransmissions = reliability->check_retransmissions(fragmentation->get_packet_mtu());
+        std::vector<Reliability::UndeliverablePacket> undeliverable;
+        auto retransmissions = reliability->check_retransmissions(fragmentation->get_packet_mtu(), undeliverable);
+
+        // A reliable send we can no longer make good on means the connection is broken, not that one message
+        // was unlucky. Drop the peer so the caller sees a disconnect rather than silently losing the data;
+        // they will reconnect off their next announce and the state is sent again from scratch.
+        for (const auto& failure : undeliverable) {
+            if (should_log(LogLevel::Warn)) {
+                std::ostringstream msg;
+                msg << "peer unreachable " << sock_str(failure.target) << ", dropping connection after "
+                    << failure.retransmits << " retransmits of packet_id=" << failure.packet_id
+                    << " hash=" << hash_hex(failure.hash) << " with " << failure.unacked << "/"
+                    << failure.packet_count << " fragments unacknowledged";
+                log(LogLevel::Warn, "reliability", msg.str());
+            }
+            // Takes the same path as a LEAVE packet, which clears our tracking and notifies the caller
+            discovery->process_leave(failure.target);
+        }
         if (should_log(LogLevel::Debug) && !retransmissions.empty()) {
             std::ostringstream msg;
             msg << "retransmit count=" << retransmissions.size();

src/nuclear/src/nuclearnet/Reliability.cpp

@@ -39,6 +39,10 @@
 namespace NUClear {
 namespace network {
 
+    // Definitions for the in class initialised constants, needed until C++17 makes them implicitly inline
+    constexpr uint16_t Reliability::MAX_RETRANSMITS;
+    constexpr uint16_t Reliability::MAX_BACKOFF_SHIFT;
+    constexpr std::chrono::seconds Reliability::MAX_RETRANSMIT_INTERVAL;
 
     void Reliability::track_packet(const sock_t& target,
                                    uint16_t packet_id,
@@ -143,6 +147,14 @@ namespace network {
     std::vector<Reliability::RetransmitRequest> Reliability::check_retransmissions(
         uint16_t packet_mtu,
         std::chrono::steady_clock::time_point now) {
+        std::vector<UndeliverablePacket> ignored;
+        return check_retransmissions(packet_mtu, ignored, now);
+    }
+
+    std::vector<Reliability::RetransmitRequest> Reliability::check_retransmissions(
+        uint16_t packet_mtu,
+        std::vector<UndeliverablePacket>& failed_peers,
+        std::chrono::steady_clock::time_point now) {
         std::vector<RetransmitRequest> retransmissions;
 
         const std::lock_guard<std::mutex> lock(tracking_mutex);
@@ -158,31 +170,34 @@ namespace network {
             }
 
             // Back the timeout off exponentially while the packet goes unacknowledged, so a peer that cannot
-            // acknowledge it is retried at a decaying rate rather than a constant one
-            rto *= 1 << std::min(tp.retransmit_count, MAX_BACKOFF_SHIFT);
+            // acknowledge it is retried at a decaying rate rather than a constant one. Bounded so that a peer
+            // with a slow measured RTT is still declared unreachable in a sensible amount of time, but never
+            // retried faster than its own timeout.
+            const auto backed_off = rto * (1 << std::min(tp.retransmit_count, MAX_BACKOFF_SHIFT));
+            rto                   = std::max(rto, std::min(backed_off, std::chrono::steady_clock::duration(
+                                                                             MAX_RETRANSMIT_INTERVAL)));
 
             // Check if it's time to retransmit
             if (now - tp.last_send < rto) {
                 ++it;
                 continue;
             }
 
-            // Give up rather than retransmitting forever. Without this a single message the peer will never
-            // acknowledge becomes a permanent load, and every further one adds to it.
+            // Out of retries. Report the peer as unreachable rather than dropping the packet and moving on —
+            // a reliable send promised this data arrived, so the caller has to be told the connection failed.
+            // The packet stays tracked; it is cleaned up when the peer is removed.
             if (tp.retransmit_count >= MAX_RETRANSMITS) {
-                if (should_log(LogLevel::Warn)) {
-                    std::size_t unacked = 0;
-                    for (uint16_t i = 0; i < tp.packet_count; ++i) {
-                        unacked += tp.acked[i] ? 0 : 1;
-                    }
-                    log(LogLevel::Warn,
-                        "reliability",
-                        "giving up on packet_id=" + std::to_string(tp.packet_id) + " hash=" + hash_hex(tp.hash)
-                            + " peer=" + sock_str(tp.target) + " after " + std::to_string(tp.retransmit_count)
-                            + " retransmits with " + std::to_string(unacked) + "/"
-                            + std::to_string(tp.packet_count) + " fragments unacknowledged");
+                UndeliverablePacket failure;
+                failure.target       = tp.target;
+                failure.packet_id    = tp.packet_id;
+                failure.hash         = tp.hash;
+                failure.packet_count = tp.packet_count;
+                failure.retransmits  = tp.retransmit_count;
+                for (uint16_t i = 0; i < tp.packet_count; ++i) {
+                    failure.unacked += tp.acked[i] ? 0 : 1;
                 }
-                it = tracked_packets.erase(it);
+                failed_peers.push_back(failure);
+                ++it;
                 continue;
             }
 

src/nuclear/src/nuclearnet/Reliability.hpp

@@ -51,17 +51,37 @@ namespace network {
         using sock_t = util::network::sock_t;
 
         /**
-         * How many times a packet group is retransmitted before it is given up on.
+         * How many times a packet group is retransmitted before the peer is declared unreachable.
          *
-         * A peer that stays connected but never acknowledges a particular packet would otherwise have it
-         * retransmitted forever, so one undeliverable message becomes a permanent load that grows every time
-         * another one joins it.
+         * A reliable send promises the data arrived, so a packet that cannot be delivered is never abandoned in
+         * favour of carrying on with the next one — that would silently break the guarantee the caller asked
+         * for. As TCP does, exhausting the retries means the connection is broken rather than that one message
+         * was unlucky, so the peer is dropped and has to reconnect.
          */
         static constexpr uint16_t MAX_RETRANSMITS = 10;
 
         /// How many times the retransmission timeout is allowed to double while a packet goes unacknowledged
         static constexpr uint16_t MAX_BACKOFF_SHIFT = 6;
 
+        /**
+         * The longest the backoff is allowed to stretch the retransmission timeout to.
+         *
+         * The timeout is already capped at 60s for a genuinely slow peer, and doubling that six times would
+         * take an hour to notice the peer had gone. The backoff never pushes the interval below the peer's own
+         * timeout, so a slow link is still given the time it needs.
+         */
+        static constexpr std::chrono::seconds MAX_RETRANSMIT_INTERVAL{5};
+
+        /// A packet that could not be delivered, and the peer it was destined for
+        struct UndeliverablePacket {
+            sock_t target{};
+            uint16_t packet_id{0};
+            uint64_t hash{0};
+            uint16_t packet_count{0};
+            uint16_t unacked{0};
+            uint16_t retransmits{0};
+        };
+
         /// Information about a fragment that needs retransmitting
         struct RetransmitRequest {
             sock_t target{};
@@ -129,11 +149,22 @@ namespace network {
         /**
          * Check for packets that need retransmission and return them.
          *
-         * @param packet_mtu The MTU to use for fragmenting retransmissions
-         * @param now        The current time (defaults to steady_clock::now())
+         * @param packet_mtu    The MTU to use for fragmenting retransmissions
+         * @param failed_peers  Filled with peers that have exhausted their retries and should be disconnected.
+         *                      Their packets stay tracked until the peer is removed, so the caller must act on
+         *                      this or the same peers will be reported again.
+         * @param now           The current time (defaults to steady_clock::now())
          *
          * @return List of fragments that need to be retransmitted
          */
+        std::vector<RetransmitRequest> check_retransmissions(
+            uint16_t packet_mtu,
+            std::vector<UndeliverablePacket>& failed_peers,
+            std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now());
+
+        /**
+         * Check for packets that need retransmission, ignoring any peers that have exhausted their retries.
+         */
         std::vector<RetransmitRequest> check_retransmissions(
             uint16_t packet_mtu,
             std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now());

src/nuclear/tests/tests/nuclearnet/ProcessPacket.cpp

@@ -26,6 +26,7 @@
 #include <catch2/catch_test_macros.hpp>
 #include <cstdint>
 #include <cstring>
+#include <thread>
 #include <memory>
 #include <string>
 #include <utility>
@@ -529,3 +530,41 @@ SCENARIO("send to named peer that does not exist delivers nothing", "[nuclearnet
     std::vector<uint8_t> payload = {1, 2, 3};
     net->send(0x1234, payload.data(), payload.size(), "NonExistentPeer", false);
 }
+
+SCENARIO("A peer that never acknowledges a reliable send is disconnected",
+         "[nuclearnet][process_packet][.slow]") {
+    if (!test_util::has_ipv4_multicast()) {
+        SKIP("No multicast support");
+    }
+
+    // A long peer timeout, so the ordinary "stopped announcing" path cannot be what disconnects them and
+    // the only thing left to do it is the unacknowledged reliable send
+    auto net = std::make_unique<NUClearNet>();
+    NetworkConfig cfg;
+    cfg.name         = "TestNode";
+    cfg.peer_timeout = std::chrono::seconds(60);
+    net->reset(cfg);
+
+    // An address nothing will ever answer from, so the reliable send can never be acknowledged
+    const sock_t peer = make_addr(0x0A000001, 5000);
+
+    bool left = false;
+    net->set_leave_callback([&](const NUClear::network::PeerInfo& p) { left = (p.name == "Deaf"); });
+
+    establish_peer(*net, peer, "Deaf");
+
+    const std::vector<uint8_t> payload(64, 0xAB);
+    net->send(0x1234, payload.data(), payload.size(), "Deaf", true);
+
+    // Drive the network until it gives up on the peer. The retransmit interval backs off and is capped, so
+    // this settles well inside the timeout below.
+    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(60);
+    while (!left && std::chrono::steady_clock::now() < deadline) {
+        net->process();
+        std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    }
+
+    THEN("the caller is told the peer left rather than the data being dropped silently") {
+        REQUIRE(left);
+    }
+}

src/nuclear/tests/tests/nuclearnet/Reliability.cpp

@@ -126,13 +126,13 @@ SCENARIO("Reliability retransmits until the peer is removed", "[nuclearnet][reli
 
     // Each retransmission waits longer than the last, so step well past the backed off timeout each time
     for (std::size_t i = 0; i < 3; ++i) {
-        t += std::chrono::seconds(60);
+        t += std::chrono::hours(1);
         REQUIRE(rel.check_retransmissions(100, t).size() == 1);
     }
 
     // Removing the peer cleans up all tracked packets
     rel.remove_peer(target);
-    t += std::chrono::seconds(60);
+    t += std::chrono::hours(1);
     REQUIRE(rel.check_retransmissions(100, t).empty());
 }
 
@@ -165,30 +165,48 @@ SCENARIO("Reliability remove_peer removes all tracked state", "[nuclearnet][reli
     REQUIRE(retransmissions.empty());
 }
 
-SCENARIO("Reliability gives up on a packet that is never acknowledged", "[nuclearnet][reliability]") {
+SCENARIO("Reliability reports the peer as unreachable when retries are exhausted", "[nuclearnet][reliability]") {
     Reliability rel;
 
     const sock_t target = make_addr(0x0A000001, 5000);
     const std::vector<uint8_t> payload(100, 0xFF);
 
     auto t = std::chrono::steady_clock::now();
-    rel.track_packet(target, 1, 1, 0x1234, 0x01, payload.data(), payload.size(), t);
+    rel.track_packet(target, 1, 4, 0x1234, 0x01, payload.data(), payload.size(), t);
 
     // Walk far enough forward each time that the backed off timeout has always expired
     std::size_t rounds = 0;
-    for (std::size_t i = 0; i < Reliability::MAX_RETRANSMITS + 5; ++i) {
-        t += std::chrono::seconds(60);
-        if (!rel.check_retransmissions(100, t).empty()) {
+    std::vector<Reliability::UndeliverablePacket> failed;
+    for (std::size_t i = 0; i < Reliability::MAX_RETRANSMITS + 3; ++i) {
+        t += std::chrono::hours(1);
+        if (!rel.check_retransmissions(100, failed, t).empty()) {
             ++rounds;
         }
     }
 
-    THEN("it is retransmitted a bounded number of times and then dropped") {
+    THEN("it is retransmitted a bounded number of times, then the peer is reported") {
         REQUIRE(rounds == Reliability::MAX_RETRANSMITS);
+        REQUIRE_FALSE(failed.empty());
+        REQUIRE(failed.front().target == target);
+        REQUIRE(failed.front().packet_id == 1);
+        REQUIRE(failed.front().hash == 0x1234);
+        REQUIRE(failed.front().unacked == 4);
+        REQUIRE(failed.front().retransmits == Reliability::MAX_RETRANSMITS);
+    }
+
+    THEN("the packet is not silently dropped, it stays tracked until the peer is removed") {
+        // Reported again rather than forgotten, because a reliable send must not be abandoned quietly
+        failed.clear();
+        t += std::chrono::hours(1);
+        rel.check_retransmissions(100, failed, t);
+        REQUIRE_FALSE(failed.empty());
 
-        // Nothing further, no matter how long we wait
+        // Removing the peer is what clears it
+        rel.remove_peer(target);
+        failed.clear();
         t += std::chrono::hours(1);
-        REQUIRE(rel.check_retransmissions(100, t).empty());
+        REQUIRE(rel.check_retransmissions(100, failed, t).empty());
+        REQUIRE(failed.empty());
     }
 }