diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index ba0c8680..23cfb183 100644 --- a/radicle-node/src/service.rs +++ b/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() diff --git a/radicle-node/src/test/environment.rs b/radicle-node/src/test/environment.rs index d2439685..90844568 100644 --- a/radicle-node/src/test/environment.rs +++ b/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::>(); + log::debug!(target: "test", "Node has missing routes: {diff:?}"); } true }); diff --git a/radicle-node/src/test/simulator.rs b/radicle-node/src/test/simulator.rs index 8a2f994c..840698c3 100644 --- a/radicle-node/src/test/simulator.rs +++ b/radicle-node/src/test/simulator.rs @@ -176,7 +176,7 @@ pub struct Simulation { /// Events emitted during simulation. events: BTreeMap>, /// Messages received during simulation. - messages: Vec, + messages: Vec<(NodeId, NodeId, Message)>, /// Priority events that should happen immediately. priority: VecDeque, /// Simulated latencies between nodes. @@ -250,7 +250,7 @@ impl Simulation { } /// 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 Simulation { } } 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(); diff --git a/radicle-node/src/tests.rs b/radicle-node/src/tests.rs index d26ea3c7..b6e27233 100644 --- a/radicle-node/src/tests.rs +++ b/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::(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); diff --git a/radicle-node/src/tests/e2e.rs b/radicle-node/src/tests/e2e.rs index 7347eabb..be513f25 100644 --- a/radicle-node/src/tests/e2e.rs +++ b/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); diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index 5d1b677a..1cc02a3a 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/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,