Make tests event-based by injecting time into Discovery, Reliability, and Fragmentation

Add an optional 'now' parameter (defaulting to steady_clock::now()) to all
time-dependent methods in Discovery, Reliability, and Fragmentation. This
allows tests to advance time deterministically without sleeping, making them
faster, more reliable, and immune to CI timing variability.

Tests converted from time-based (sleep_for) to event-based:
- Discovery: check_timeouts and touch_peer tests
- Reliability: all retransmission/timeout tests (7 sleeps removed)
- Fragmentation: cleanup_expired test

The only remaining sleep_for is in Integration.cpp's polling loop for
genuine async UDP networking, which is inherently time-dependent.

Author: TrentHouliston

src/nuclearnet/Discovery.cpp

@@ -86,7 +86,10 @@
             return packet;
         }
 
-        void Discovery::process_announce(const sock_t& source, const uint8_t* data, std::size_t length) {
+        void Discovery::process_announce(const sock_t& source,
+                                          const uint8_t* data,
+                                          std::size_t length,
+                                          std::chrono::steady_clock::time_point now) {
             // Minimum size: header(5) + name_length(2) + num_subscriptions(2) = 9
             if (length < sizeof(PacketHeader) + sizeof(uint16_t) + sizeof(uint16_t)) {
                 return;
@@ -148,14 +151,14 @@
                     PeerInfo info;
                     info.name          = name;
                     info.address       = source;
-                    info.last_seen     = std::chrono::steady_clock::now();
+                    info.last_seen     = now;
                     info.subscriptions = std::move(subscriptions);
                     peers.emplace(source, std::move(info));
                     is_new = true;
                 }
                 else {
                     // Existing peer — update last_seen and check for subscription changes
-                    it->second.last_seen = std::chrono::steady_clock::now();
+                    it->second.last_seen = now;
                     if (it->second.subscriptions != subscriptions) {
                         it->second.subscriptions = std::move(subscriptions);
                         subs_changed             = true;
@@ -198,16 +201,15 @@
             }
         }
 
-        void Discovery::touch_peer(const sock_t& source) {
+        void Discovery::touch_peer(const sock_t& source, std::chrono::steady_clock::time_point now) {
             const std::lock_guard<std::mutex> lock(peers_mutex);
             auto it = peers.find(source);
             if (it != peers.end()) {
-                it->second.last_seen = std::chrono::steady_clock::now();
+                it->second.last_seen = now;
             }
         }
 
-        std::vector<PeerInfo> Discovery::check_timeouts() {
-            auto now = std::chrono::steady_clock::now();
+        std::vector<PeerInfo> Discovery::check_timeouts(std::chrono::steady_clock::time_point now) {
             std::vector<PeerInfo> removed;
 
             {

src/nuclearnet/Discovery.hpp

@@ -119,8 +119,12 @@ namespace network {
          * @param source  The UDP source address (IP + port) of the packet
          * @param data    The raw packet data
          * @param length  The length of the packet data
+         * @param now     The current time (defaults to steady_clock::now())
          */
-        void process_announce(const sock_t& source, const uint8_t* data, std::size_t length);
+        void process_announce(const sock_t& source,
+                              const uint8_t* data,
+                              std::size_t length,
+                              std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now());
 
         /**
          * Process a received leave packet from a peer.
@@ -133,15 +137,20 @@ namespace network {
          * Update the last_seen timestamp for a peer (called on any received packet).
          *
          * @param source The UDP source address
+         * @param now    The current time (defaults to steady_clock::now())
          */
-        void touch_peer(const sock_t& source);
+        void touch_peer(const sock_t& source,
+                        std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now());
 
         /**
          * Check for peers that have timed out and remove them.
          *
+         * @param now The current time (defaults to steady_clock::now())
+         *
          * @return List of peers that were removed due to timeout
          */
-        std::vector<PeerInfo> check_timeouts();
+        std::vector<PeerInfo> check_timeouts(
+            std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now());
 
         /**
          * Get the current list of known peers.

src/nuclearnet/Fragmentation.cpp

@@ -71,7 +71,8 @@ namespace network {
                                         uint8_t flags,
                                         const uint8_t* data,
                                         std::size_t data_length,
-                                        AssembledPacket& out_packet) {
+                                        AssembledPacket& out_packet,
+                                        std::chrono::steady_clock::time_point now) {
         if (packet_count == 0 || packet_no >= packet_count) {
             return false;
         }
@@ -92,7 +93,7 @@ namespace network {
         assembly.hash           = hash;
         assembly.flags          = flags;
         assembly.packet_count   = packet_count;
-        assembly.last_update    = std::chrono::steady_clock::now();
+        assembly.last_update    = now;
 
         // Store this fragment
         assembly.fragments[packet_no].assign(data, data + data_length);
@@ -128,8 +129,7 @@ namespace network {
         return false;
     }
 
-    std::size_t Fragmentation::cleanup_expired() {
-        auto now = std::chrono::steady_clock::now();
+    std::size_t Fragmentation::cleanup_expired(std::chrono::steady_clock::time_point now) {
         std::size_t removed = 0;
 
         const std::lock_guard<std::mutex> lock(assembly_mutex);

src/nuclearnet/Fragmentation.hpp

@@ -100,6 +100,7 @@ namespace network {
          * @param data        The fragment payload
          *
          * @param out_packet  Filled with the assembled packet when reassembly completes
+         * @param now         The current time (defaults to steady_clock::now())
          *
          * @return true if all fragments are now received and @p out_packet is valid, false otherwise
          */
@@ -111,14 +112,18 @@ namespace network {
                              uint8_t flags,
                              const uint8_t* data,
                              std::size_t data_length,
-                             AssembledPacket& out_packet);
+                             AssembledPacket& out_packet,
+                             std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now());
 
         /**
          * Clean up assemblies that have timed out.
          *
+         * @param now The current time (defaults to steady_clock::now())
+         *
          * @return Number of assemblies that were discarded
          */
-        std::size_t cleanup_expired();
+        std::size_t cleanup_expired(
+            std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now());
 
         /**
          * Get the packet MTU (max payload per fragment).

src/nuclearnet/Reliability.cpp

@@ -40,7 +40,8 @@
                                        uint64_t hash,
                                        uint8_t flags,
                                        const uint8_t* payload,
-                                       std::size_t payload_len) {
+                                       std::size_t payload_len,
+                                       std::chrono::steady_clock::time_point now) {
             TrackedPacket tp;
             tp.target       = target;
             tp.packet_id    = packet_id;
@@ -49,7 +50,7 @@
             tp.flags        = flags;
             tp.payload.assign(payload, payload + payload_len);
             tp.acked.resize(packet_count, false);
-            tp.last_send = std::chrono::steady_clock::now();
+            tp.last_send = now;
 
             TrackingKey key{target, packet_id};
 
@@ -61,7 +62,8 @@
                                       uint16_t packet_id,
                                       uint16_t packet_count,
                                       const uint8_t* ack_bitset,
-                                      std::size_t bitset_size) {
+                                      std::size_t bitset_size,
+                                      std::chrono::steady_clock::time_point now) {
             TrackingKey key{source, packet_id};
 
             const std::lock_guard<std::mutex> lock(tracking_mutex);
@@ -73,7 +75,6 @@
             auto& tp = it->second;
 
             // Update RTT estimate based on time since last send
-            auto now = std::chrono::steady_clock::now();
             auto rtt = now - tp.last_send;
             {
                 const std::lock_guard<std::mutex> rtt_lock(rtt_mutex);
@@ -165,8 +166,9 @@
             return packet;
         }
 
-        std::vector<Reliability::RetransmitRequest> Reliability::check_retransmissions(uint16_t packet_mtu) {
-            auto now = std::chrono::steady_clock::now();
+        std::vector<Reliability::RetransmitRequest> Reliability::check_retransmissions(
+            uint16_t packet_mtu,
+            std::chrono::steady_clock::time_point now) {
             std::vector<RetransmitRequest> retransmissions;
             std::vector<TrackingKey> expired;
 

src/nuclearnet/Reliability.hpp

@@ -78,14 +78,16 @@ namespace network {
          * @param flags        Packet flags
          * @param payload      Pointer to the full original payload (copied internally for retransmission)
          * @param payload_len  Length of the payload in bytes
+         * @param now          The current time (defaults to steady_clock::now())
          */
         void track_packet(const sock_t& target,
                           uint16_t packet_id,
                           uint16_t packet_count,
                           uint64_t hash,
                           uint8_t flags,
                           const uint8_t* payload,
-                          std::size_t payload_len);
+                          std::size_t payload_len,
+                          std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now());
 
         /**
          * Process a received ACK packet.
@@ -95,12 +97,14 @@ namespace network {
          * @param packet_count Total fragments in the group
          * @param ack_bitset   Bitset of received fragments (1 bit per fragment, LSB first)
          * @param bitset_size  Size of the ack_bitset in bytes
+         * @param now          The current time (defaults to steady_clock::now())
          */
         void process_ack(const sock_t& source,
                          uint16_t packet_id,
                          uint16_t packet_count,
                          const uint8_t* ack_bitset,
-                         std::size_t bitset_size);
+                         std::size_t bitset_size,
+                         std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now());
 
         /**
          * Process a received NACK packet.
@@ -147,10 +151,13 @@ 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())
          *
          * @return List of fragments that need to be retransmitted
          */
-        std::vector<RetransmitRequest> check_retransmissions(uint16_t packet_mtu);
+        std::vector<RetransmitRequest> check_retransmissions(
+            uint16_t packet_mtu,
+            std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now());
 
         /**
          * Remove all tracking for a given peer (e.g., on disconnect).

tests/tests/nuclearnet/Discovery.cpp

@@ -24,7 +24,6 @@
 
 #include <catch2/catch_test_macros.hpp>
 #include <chrono>
-#include <thread>
 #include <vector>
 
 #include "util/platform.hpp"
@@ -142,13 +141,18 @@ SCENARIO("Discovery check_timeouts removes stale peers", "[nuclearnet][discovery
 
     disc.set_leave_callback([&](const PeerInfo&) { leave_called = true; });
 
+    // Add peer at time T
+    auto t = std::chrono::steady_clock::now();
     auto announce = Discovery::build_announce_packet("peer_a", {});
-    disc.process_announce(peer_addr, announce.data(), announce.size());
+    disc.process_announce(peer_addr, announce.data(), announce.size(), t);
 
-    // Wait for timeout
-    std::this_thread::sleep_for(std::chrono::milliseconds(30));
+    // Check at T+10ms (before timeout) — peer should still be there
+    auto removed = disc.check_timeouts(t + std::chrono::milliseconds(10));
+    REQUIRE(removed.empty());
+    REQUIRE(disc.has_peer(peer_addr));
 
-    auto removed = disc.check_timeouts();
+    // Check at T+25ms (after 20ms timeout) — peer should be removed
+    removed = disc.check_timeouts(t + std::chrono::milliseconds(25));
     REQUIRE(removed.size() == 1);
     REQUIRE(removed[0].name == "peer_a");
     REQUIRE(leave_called);
@@ -160,19 +164,24 @@ SCENARIO("Discovery touch_peer resets timeout", "[nuclearnet][discovery]") {
 
     sock_t peer_addr = make_addr(0x0A000001, 5000);
 
+    // Add peer at time T
+    auto t = std::chrono::steady_clock::now();
     auto announce = Discovery::build_announce_packet("peer_a", {});
-    disc.process_announce(peer_addr, announce.data(), announce.size());
+    disc.process_announce(peer_addr, announce.data(), announce.size(), t);
 
-    // Wait part of the timeout, then touch
-    std::this_thread::sleep_for(std::chrono::milliseconds(120));
-    disc.touch_peer(peer_addr);
+    // Touch at T+120ms (before 200ms timeout expires)
+    disc.touch_peer(peer_addr, t + std::chrono::milliseconds(120));
 
-    // Wait another partial timeout (total would have expired without touch)
-    std::this_thread::sleep_for(std::chrono::milliseconds(120));
-
-    auto removed = disc.check_timeouts();
+    // Check at T+240ms — 240ms since announce, but only 120ms since touch
+    // Since timeout is 200ms from last_seen, peer should still be alive
+    auto removed = disc.check_timeouts(t + std::chrono::milliseconds(240));
     REQUIRE(removed.empty());
     REQUIRE(disc.has_peer(peer_addr));
+
+    // Check at T+325ms — 205ms since touch, should now be timed out
+    removed = disc.check_timeouts(t + std::chrono::milliseconds(325));
+    REQUIRE(removed.size() == 1);
+    REQUIRE(removed[0].name == "peer_a");
 }
 
 SCENARIO("Discovery get_peers returns all known peers", "[nuclearnet][discovery]") {

tests/tests/nuclearnet/Fragmentation.cpp

@@ -26,7 +26,6 @@
 #include <chrono>
 #include <cstdint>
 #include <numeric>
-#include <thread>
 #include <vector>
 
 using NUClear::network::Fragmentation;
@@ -191,20 +190,22 @@ SCENARIO("Fragmentation cleanup_expired removes stale assemblies", "[nuclearnet]
     // Use a very short timeout for testing
     Fragmentation frag(100, 64 * 1024 * 1024, std::chrono::milliseconds(1));
 
-    // Submit a partial assembly
+    // Submit a partial assembly at time T
+    auto t = std::chrono::steady_clock::now();
     uint8_t data[50] = {};
     Fragmentation::AssembledPacket result;
-    frag.submit_fragment(1, 1, 0, 3, 0x1234, 0, data, 50, result);
+    frag.submit_fragment(1, 1, 0, 3, 0x1234, 0, data, 50, result, t);
 
-    // Wait for the timeout
-    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    // Cleanup at T (not expired yet) — nothing removed
+    std::size_t removed = frag.cleanup_expired(t);
+    REQUIRE(removed == 0);
 
-    // Cleanup should remove it
-    std::size_t removed = frag.cleanup_expired();
+    // Cleanup at T+10ms (past 1ms timeout) — should remove it
+    removed = frag.cleanup_expired(t + std::chrono::milliseconds(10));
     REQUIRE(removed == 1);
 
     // Second cleanup should find nothing
-    removed = frag.cleanup_expired();
+    removed = frag.cleanup_expired(t + std::chrono::milliseconds(20));
     REQUIRE(removed == 0);
 }