node: add more tests for refs relay

Signed-off-by: Alexis Sellier <alexis@radicle.xyz>

radicle-node/src/service.rs

@@ -643,8 +643,9 @@ where
                         return Ok(false);
                     }
                     // TODO: Check refs to see if we should try to fetch or not.
-                    // FIXME: This code is wrong: we shouldn't be fetching from the connected peer,
-                    // we should fetch from the origin.
+                    // Refs are only supposed to be relayed by peers who are tracking
+                    // the resource. Therefore, it's safe to fetch from the remote
+                    // peer, even though it isn't the announcer.
                     let updated = self.storage.fetch(message.id, git).unwrap();
                     let is_updated = !updated.is_empty();
 

radicle-node/src/service/message.rs

@@ -122,6 +122,16 @@ pub struct Subscribe {
     pub until: Timestamp,
 }
 
+impl Subscribe {
+    pub fn all() -> Self {
+        Self {
+            filter: Filter::default(),
+            since: Timestamp::MIN,
+            until: Timestamp::MAX,
+        }
+    }
+}
+
 /// Node announcing itself to the network.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct NodeAnnouncement {

radicle-node/src/test/peer.rs

@@ -7,8 +7,10 @@ use log::*;
 
 use crate::address;
 use crate::clock::{RefClock, Timestamp};
+use crate::crypto::Signer;
 use crate::git;
 use crate::git::Url;
+use crate::identity::Id;
 use crate::node;
 use crate::prelude::NodeId;
 use crate::service;
@@ -23,12 +25,12 @@ use crate::test::simulator;
 use crate::{Link, LocalTime};
 
 /// Service instantiation used for testing.
-pub type Service<S> = service::Service<routing::Table, address::Book, S, MockSigner>;
+pub type Service<S, G> = service::Service<routing::Table, address::Book, S, G>;
 
 #[derive(Debug)]
-pub struct Peer<S> {
+pub struct Peer<S, G> {
     pub name: &'static str,
-    pub service: Service<S>,
+    pub service: Service<S, G>,
     pub ip: net::IpAddr,
     pub rng: fastrand::Rng,
     pub local_time: LocalTime,
@@ -37,9 +39,10 @@ pub struct Peer<S> {
     initialized: bool,
 }
 
-impl<S> simulator::Peer<S> for Peer<S>
+impl<S, G> simulator::Peer<S, G> for Peer<S, G>
 where
     S: WriteStorage + 'static,
+    G: Signer + 'static,
 {
     fn init(&mut self) {
         self.initialize()
@@ -50,21 +53,21 @@ where
     }
 }
 
-impl<S> Deref for Peer<S> {
-    type Target = Service<S>;
+impl<S, G> Deref for Peer<S, G> {
+    type Target = Service<S, G>;
 
     fn deref(&self) -> &Self::Target {
         &self.service
     }
 }
 
-impl<S> DerefMut for Peer<S> {
+impl<S, G> DerefMut for Peer<S, G> {
     fn deref_mut(&mut self) -> &mut Self::Target {
         &mut self.service
     }
 }
 
-impl<S> Peer<S>
+impl<S> Peer<S, MockSigner>
 where
     S: WriteStorage + 'static,
 {
@@ -75,6 +78,8 @@ where
 
             ..git::Url::default()
         };
+        let mut rng = fastrand::Rng::new();
+        let signer = MockSigner::new(&mut rng);
 
         Self::config(
             name,
@@ -84,20 +89,27 @@ where
             },
             ip,
             storage,
-            fastrand::Rng::new(),
+            signer,
+            rng,
         )
     }
+}
 
+impl<S, G> Peer<S, G>
+where
+    S: WriteStorage + 'static,
+    G: Signer + 'static,
+{
     pub fn config(
         name: &'static str,
         config: Config,
         ip: impl Into<net::IpAddr>,
         storage: S,
-        mut rng: fastrand::Rng,
+        signer: G,
+        rng: fastrand::Rng,
     ) -> Self {
         let local_time = LocalTime::now();
         let clock = RefClock::from(local_time);
-        let signer = MockSigner::new(&mut rng);
         let routing = routing::Table::memory().unwrap();
         let addrs = address::Book::memory().unwrap();
         let service = Service::new(config, clock, routing, storage, addrs, signer, rng.clone());
@@ -170,9 +182,7 @@ where
         )
     }
 
-    pub fn refs_announcement(&self) -> Message {
-        let inv = self.inventory().unwrap();
-        let id = inv[self.rng.usize(..inv.len())];
+    pub fn refs_announcement(&self, id: Id) -> Message {
         let refs = BTreeMap::new().into();
         let ann = AnnouncementMessage::from(RefsAnnouncement {
             id,
@@ -185,7 +195,7 @@ where
     }
 
     pub fn connect_from(&mut self, peer: &Self) {
-        let remote = simulator::Peer::<S>::addr(peer);
+        let remote = simulator::Peer::<S, G>::addr(peer);
         let local = net::SocketAddr::new(self.ip, self.rng.u16(..));
         let git = format!("file:///{}.git", remote.ip());
         let git = Url::from_bytes(git.as_bytes()).unwrap();
@@ -213,7 +223,7 @@ where
     }
 
     pub fn connect_to(&mut self, peer: &Self) {
-        let remote = simulator::Peer::<S>::addr(peer);
+        let remote = simulator::Peer::<S, G>::addr(peer);
 
         self.initialize();
         self.service.attempted(&remote);

radicle-node/src/test/simulator.rs

@@ -14,6 +14,7 @@ use log::*;
 use nakamoto_net as nakamoto;
 use nakamoto_net::{Link, LocalDuration, LocalTime};
 
+use crate::crypto::Signer;
 use crate::service::reactor::Io;
 use crate::service::{DisconnectReason, Envelope, Event};
 use crate::storage::WriteStorage;
@@ -29,7 +30,9 @@ pub const MAX_EVENTS: usize = 2048;
 type NodeId = std::net::IpAddr;
 
 /// A simulated peer. Service instances have to be wrapped in this type to be simulated.
-pub trait Peer<S>: Deref<Target = Service<S>> + DerefMut<Target = Service<S>> + 'static {
+pub trait Peer<S, G>:
+    Deref<Target = Service<S, G>> + DerefMut<Target = Service<S, G>> + 'static
+{
     /// Initialize the peer. This should at minimum initialize the service with the
     /// current time.
     fn init(&mut self);
@@ -166,7 +169,7 @@ impl Default for Options {
 }
 
 /// A peer-to-peer node simulation.
-pub struct Simulation<S> {
+pub struct Simulation<S, G> {
     /// Inbox of inputs to be delivered by the simulation.
     inbox: Inbox,
     /// Events emitted during simulation.
@@ -191,9 +194,11 @@ pub struct Simulation<S> {
     rng: fastrand::Rng,
     /// Storage type.
     storage: PhantomData<S>,
+    /// Signer type.
+    signer: PhantomData<G>,
 }
 
-impl<S: WriteStorage + 'static> Simulation<S> {
+impl<S: WriteStorage + 'static, G: Signer> Simulation<S, G> {
     /// Create a new simulation.
     pub fn new(time: LocalTime, rng: fastrand::Rng, opts: Options) -> Self {
         Self {
@@ -211,6 +216,7 @@ impl<S: WriteStorage + 'static> Simulation<S> {
             time,
             rng,
             storage: PhantomData,
+            signer: PhantomData,
         }
     }
 
@@ -267,7 +273,7 @@ impl<S: WriteStorage + 'static> Simulation<S> {
     /// Initialize peers.
     pub fn initialize<'a, P>(self, peers: impl IntoIterator<Item = &'a mut P>) -> Self
     where
-        P: Peer<S>,
+        P: Peer<S, G>,
     {
         for peer in peers.into_iter() {
             peer.init();
@@ -281,7 +287,7 @@ impl<S: WriteStorage + 'static> Simulation<S> {
         peers: impl IntoIterator<Item = &'a mut P>,
         pred: impl Fn(&Self) -> bool,
     ) where
-        P: Peer<S>,
+        P: Peer<S, G>,
     {
         let mut nodes: BTreeMap<_, _> = peers.into_iter().map(|p| (p.addr().ip(), p)).collect();
 
@@ -295,12 +301,12 @@ impl<S: WriteStorage + 'static> Simulation<S> {
     /// Process one scheduled input from the inbox, using the provided peers.
     /// This function should be called until it returns `false`, or some desired state is reached.
     /// Returns `true` if there are more messages to process.
-    pub fn step<'a, P: Peer<S>>(&mut self, peers: impl IntoIterator<Item = &'a mut P>) -> bool {
+    pub fn step<'a, P: Peer<S, G>>(&mut self, peers: impl IntoIterator<Item = &'a mut P>) -> bool {
         let mut nodes: BTreeMap<_, _> = peers.into_iter().map(|p| (p.addr().ip(), p)).collect();
         self.step_(&mut nodes)
     }
 
-    fn step_<P: Peer<S>>(&mut self, nodes: &mut BTreeMap<NodeId, &mut P>) -> bool {
+    fn step_<P: Peer<S, G>>(&mut self, nodes: &mut BTreeMap<NodeId, &mut P>) -> bool {
         if !self.opts.latency.is_empty() {
             // Configure latencies.
             for (i, from) in nodes.keys().enumerate() {

radicle-node/src/test/tests.rs

@@ -15,6 +15,7 @@ use crate::service::ServiceState as _;
 use crate::service::*;
 use crate::storage::git::Storage;
 use crate::storage::ReadStorage;
+use crate::test::arbitrary;
 use crate::test::assert_matches;
 use crate::test::fixtures;
 #[allow(unused)]
@@ -77,14 +78,21 @@ fn test_inbound_connection() {
 
 #[test]
 fn test_persistent_peer_connect() {
-    let rng = fastrand::Rng::new();
+    let mut rng = fastrand::Rng::new();
     let bob = Peer::new("bob", [8, 8, 8, 8], MockStorage::empty());
     let eve = Peer::new("eve", [9, 9, 9, 9], MockStorage::empty());
     let config = Config {
         connect: vec![bob.address(), eve.address()],
         ..Config::default()
     };
-    let mut alice = Peer::config("alice", config, [7, 7, 7, 7], MockStorage::empty(), rng);
+    let mut alice = Peer::config(
+        "alice",
+        config,
+        [7, 7, 7, 7],
+        MockStorage::empty(),
+        MockSigner::new(&mut rng),
+        rng,
+    );
 
     alice.initialize();
 
@@ -158,6 +166,7 @@ fn test_tracking() {
         },
         [7, 7, 7, 7],
         MockStorage::empty(),
+        MockSigner::default(),
         fastrand::Rng::new(),
     );
     let proj_id: identity::Id = test::arbitrary::gen(1);
@@ -334,6 +343,94 @@ fn test_announcement_relay() {
     );
 }
 
+#[test]
+fn test_refs_announcement_relay() {
+    let tmp = tempfile::tempdir().unwrap();
+    let mut alice = Peer::new(
+        "alice",
+        [7, 7, 7, 7],
+        Storage::open(tmp.path().join("alice")).unwrap(),
+    );
+    let eve = Peer::new(
+        "eve",
+        [8, 8, 8, 8],
+        Storage::open(tmp.path().join("eve")).unwrap(),
+    );
+
+    let bob = {
+        let mut rng = fastrand::Rng::new();
+        let signer = MockSigner::new(&mut rng);
+        let storage = fixtures::storage(tmp.path().join("bob"), &signer).unwrap();
+
+        Peer::config(
+            "bob",
+            Config {
+                git_url: git::Url {
+                    scheme: git::url::Scheme::File,
+                    path: storage.path().to_string_lossy().to_string().into(),
+
+                    ..git::Url::default()
+                },
+                ..Config::default()
+            },
+            [9, 9, 9, 9],
+            storage,
+            signer,
+            rng,
+        )
+    };
+    let bob_inv = bob.inventory().unwrap();
+
+    alice.track(bob_inv[0]);
+    alice.track(bob_inv[1]);
+    alice.track(bob_inv[2]);
+    alice.connect_to(&bob);
+    alice.connect_to(&eve);
+    alice.receive(&eve.addr(), Message::Subscribe(Subscribe::all()));
+
+    alice.receive(&bob.addr(), bob.refs_announcement(bob_inv[0]));
+    assert_matches!(
+        alice.messages(&eve.addr()).next(),
+        Some(Message::Announcement(_)),
+        "A refs announcement from Bob is relayed to Eve"
+    );
+
+    alice.receive(&bob.addr(), bob.refs_announcement(bob_inv[0]));
+    assert!(
+        alice.messages(&eve.addr()).next().is_none(),
+        "The same ref announement is not relayed"
+    );
+
+    alice.receive(&bob.addr(), bob.refs_announcement(bob_inv[1]));
+    assert_matches!(
+        alice.messages(&eve.addr()).next(),
+        Some(Message::Announcement(_)),
+        "But a different one is"
+    );
+
+    alice.receive(&bob.addr(), bob.refs_announcement(bob_inv[2]));
+    assert_matches!(
+        alice.messages(&eve.addr()).next(),
+        Some(Message::Announcement(_)),
+        "And a third one is as well"
+    );
+}
+
+#[test]
+fn test_refs_gossip_no_subscribe() {
+    let mut alice = Peer::new("alice", [7, 7, 7, 7], MockStorage::empty());
+    let bob = Peer::new("bob", [8, 8, 8, 8], MockStorage::empty());
+    let eve = Peer::new("eve", [9, 9, 9, 9], MockStorage::empty());
+    let id = arbitrary::gen(1);
+
+    alice.track(id);
+    alice.connect_to(&bob);
+    alice.connect_to(&eve);
+    alice.receive(&bob.addr(), bob.refs_announcement(id));
+
+    assert!(alice.messages(&eve.addr()).next().is_none());
+}
+
 #[test]
 fn test_inventory_relay() {
     // Topology is eve <-> alice <-> bob
@@ -442,6 +539,7 @@ fn test_persistent_peer_reconnect() {
         },
         [7, 7, 7, 7],
         MockStorage::empty(),
+        MockSigner::default(),
         fastrand::Rng::new(),
     );