node: Various small fixes

radicle-node/src/service.rs

@@ -654,6 +654,8 @@ where
             self.last_sync = now;
         }
         if now - self.last_announce >= ANNOUNCE_INTERVAL {
+            trace!(target: "service", "Running 'announce' task...");
+
             if let Err(err) = self
                 .storage
                 .inventory()

radicle-node/src/test/environment.rs

@@ -555,8 +555,8 @@ pub fn converge<'a, G: Signer + cyphernet::Ecdh + 'static>(
                 log::debug!(target: "test", "Node {} has converged", node.id);
                 return false;
             } else {
-                log::debug!(target: "test", "Node {} has {:?}", node.id, routes);
-                log::debug!(target: "test", "All routes {:?}", all_routes);
+                let diff = all_routes.symmetric_difference(&routes).collect::<Vec<_>>();
+                log::debug!(target: "test", "Node has missing routes: {diff:?}");
             }
             true
         });

radicle-node/src/test/simulator.rs

@@ -176,7 +176,7 @@ pub struct Simulation<S, G> {
     /// Events emitted during simulation.
     events: BTreeMap<NodeId, VecDeque<Event>>,
     /// Messages received during simulation.
-    messages: Vec<Message>,
+    messages: Vec<(NodeId, NodeId, Message)>,
     /// Priority events that should happen immediately.
     priority: VecDeque<Scheduled>,
     /// Simulated latencies between nodes.
@@ -250,7 +250,7 @@ impl<S: WriteStorage + 'static, G: Signer> Simulation<S, G> {
     }
 
     /// Get all messages received by nodes during the simulation.
-    pub fn messages(&mut self) -> &[Message] {
+    pub fn messages(&mut self) -> &[(NodeId, NodeId, Message)] {
         &self.messages
     }
 
@@ -400,11 +400,12 @@ impl<S: WriteStorage + 'static, G: Signer> Simulation<S, G> {
                         }
                     }
                     Input::Wake => p.wake(),
-                    Input::Received(id, msgs) => {
+                    Input::Received(from, msgs) => {
                         for msg in msgs.clone() {
-                            p.received_message(id, msg);
+                            p.received_message(from, msg);
                         }
-                        self.messages.extend(msgs);
+                        self.messages
+                            .extend(msgs.into_iter().map(|m| (from, p.node_id(), m)));
                     }
                     Input::Fetched(rid, nid, result) => {
                         let result = Rc::try_unwrap(result).unwrap();

radicle-node/src/tests.rs

@@ -477,6 +477,7 @@ fn test_announcement_rebroadcast_duplicates() {
     let eve = Peer::new("eve", [9, 9, 9, 9]);
     let rids = arbitrary::set::<RepoId>(3..=3);
 
+    carol.init();
     alice.connect_to(&bob);
     alice.receive(bob.id, carol.node_announcement());
 
@@ -936,6 +937,8 @@ fn test_inventory_relay() {
     let now = LocalTime::now().as_millis();
 
     // Inventory from Bob relayed to Eve.
+    alice.init();
+    alice.wake(); // Run all periodic tasks now so they don't trigger later.
     alice.connect_to(&bob);
     alice.connect_from(&eve);
     alice.receive(
@@ -1399,7 +1402,7 @@ fn test_queued_fetch_from_ann_same_rid() {
     // The first fetch is initiated.
     assert_matches!(alice.fetches().next(), Some((rid_, nid_, _)) if rid_ == rid && nid_ == bob.id);
     // We shouldn't send out the 2nd, 3rd fetch while we're doing the 1st fetch.
-    assert_matches!(alice.outbox().next(), None);
+    assert_matches!(alice.fetches().next(), None);
 
     // Have enough time pass that Alice sends a "ping" to Bob.
     alice.elapse(KEEP_ALIVE_DELTA);

radicle-node/src/tests/e2e.rs

@@ -60,11 +60,11 @@ fn test_inventory_sync_bridge() {
     eve.project("eve", "");
 
     let mut alice = alice.spawn();
-    let mut bob = bob.spawn();
-    let eve = eve.spawn();
+    let mut eve = eve.spawn();
+    let bob = bob.spawn();
 
     alice.connect(&bob);
-    bob.connect(&eve);
+    eve.connect(&bob);
 
     let routes = converge([&alice, &bob, &eve]);
     assert_eq!(routes.len(), 3);

radicle-node/src/wire/protocol.rs

@@ -837,7 +837,10 @@ where
                     }
                 }
                 Io::Wakeup(d) => {
-                    self.actions.push_back(reactor::Action::SetTimer(d.into()));
+                    self.actions.push_back(reactor::Action::SetTimer(
+                        // TODO: Remove this when `io-reactor` can handle `0` duration timeouts.
+                        d.max(localtime::LocalDuration::from_millis(1)).into(),
+                    ));
                 }
                 Io::Fetch {
                     rid,