Merge #375

375: A fistful of tweaks r=aphelionz a=ljedrz

A pretty wide range of test improvements, small fixes, refactorings and renames. Each commit describes the specific change, but I'll list the more noteworthy ones:
- automatically remove `Protocol::P2p(_)` from the `Multiaddr`s passed to `Ipfs::add_peer`
- only produce a `RepoEvent::RemovedBlock` when a block was actually removed
- add unit tests for the `Node` topologies

Co-authored-by: ljedrz <ljedrz@gmail.com>

Author: bors[bot]

http/src/main.rs

@@ -26,8 +26,6 @@ enum Options {
 
 fn main() {
     if std::env::var_os("RUST_LOG").is_none() {
-        // FIXME: see if tracing could be used as the frontend for log macros
-        // FIXME: use log macros here as well
         std::env::set_var(
             "RUST_LOG",
             "ipfs_http=trace,ipfs=trace,bitswap=trace,ipfs_unixfs=trace",

src/lib.rs

@@ -1542,7 +1542,7 @@ impl<TRepoTypes: RepoTypes> Future for IpfsFuture<TRepoTypes> {
                 match evt {
                     RepoEvent::WantBlock(cid) => self.swarm.want_block(cid),
                     RepoEvent::UnwantBlock(cid) => self.swarm.bitswap().cancel_block(&cid),
-                    RepoEvent::ProvideBlock(cid, ret) => {
+                    RepoEvent::NewBlock(cid, ret) => {
                         // TODO: consider if cancel is applicable in cases where we provide the
                         // associated Block ourselves
                         self.swarm.bitswap().cancel_block(&cid);
@@ -1554,7 +1554,7 @@ impl<TRepoTypes: RepoTypes> Future for IpfsFuture<TRepoTypes> {
                             let _ = ret.send(Err(anyhow!("not actively providing blocks yet")));
                         }
                     }
-                    RepoEvent::UnprovideBlock(cid) => self.swarm.stop_providing_block(&cid),
+                    RepoEvent::RemovedBlock(cid) => self.swarm.stop_providing_block(&cid),
                 }
             }
 
@@ -1665,7 +1665,12 @@ mod node {
         /// Add a known listen address of a peer participating in the DHT to the routing table.
         /// This is mandatory in order for the peer to be discoverable by other members of the
         /// DHT.
-        pub async fn add_peer(&self, peer_id: PeerId, addr: Multiaddr) -> Result<(), Error> {
+        pub async fn add_peer(&self, peer_id: PeerId, mut addr: Multiaddr) -> Result<(), Error> {
+            // Kademlia::add_address requires the address to not contain the PeerId
+            if matches!(addr.iter().last(), Some(Protocol::P2p(_))) {
+                addr.pop();
+            }
+
             self.to_task
                 .clone()
                 .send(IpfsEvent::AddPeer(peer_id, addr))

src/path.rs

@@ -132,30 +132,6 @@ impl<T: Into<PathRoot>> From<T> for IpfsPath {
     }
 }
 
-// FIXME: get rid of this; it would mean that there must be a clone to retain the rest of the path.
-impl TryInto<Cid> for IpfsPath {
-    type Error = Error;
-
-    fn try_into(self) -> Result<Cid, Self::Error> {
-        match self.root().cid() {
-            Some(cid) => Ok(cid.to_owned()),
-            None => Err(anyhow::anyhow!("expected cid")),
-        }
-    }
-}
-
-// FIXME: get rid of this; it would mean that there must be a clone to retain the rest of the path.
-impl TryInto<PeerId> for IpfsPath {
-    type Error = Error;
-
-    fn try_into(self) -> Result<PeerId, Self::Error> {
-        match self.root().peer_id() {
-            Some(peer_id) => Ok(peer_id.to_owned()),
-            None => Err(anyhow::anyhow!("expected peer id")),
-        }
-    }
-}
-
 /// SlashedPath is internal to IpfsPath variants, and basically holds a unixfs-compatible path
 /// where segments do not contain slashes but can pretty much contain all other valid UTF-8.
 ///

src/repo/mod.rs

@@ -219,11 +219,11 @@ pub struct Repo<TRepoTypes: RepoTypes> {
 pub enum RepoEvent {
     WantBlock(Cid),
     UnwantBlock(Cid),
-    ProvideBlock(
+    NewBlock(
         Cid,
         oneshot::Sender<Result<SubscriptionFuture<KadResult, String>, anyhow::Error>>,
     ),
-    UnprovideBlock(Cid),
+    RemovedBlock(Cid),
 }
 
 impl TryFrom<RequestKind> for RepoEvent {
@@ -292,9 +292,8 @@ impl<TRepoTypes: RepoTypes> Repo<TRepoTypes> {
         let (_cid, res) = self.block_store.put(block.clone()).await?;
 
         // FIXME: this doesn't cause actual DHT providing yet, only some
-        // bitswap housekeeping; RepoEvent::ProvideBlock should probably
-        // be renamed to ::NewBlock and we might want to not ignore the
-        // channel errors when we actually start providing on the DHT
+        // bitswap housekeeping; we might want to not ignore the channel
+        // errors when we actually start providing on the DHT
         if let BlockPut::NewBlock = res {
             self.subscriptions
                 .finish_subscription(cid.clone().into(), Ok(block));
@@ -305,7 +304,7 @@ impl<TRepoTypes: RepoTypes> Repo<TRepoTypes> {
 
             self.events
                 .clone()
-                .send(RepoEvent::ProvideBlock(cid.clone(), tx))
+                .send(RepoEvent::NewBlock(cid.clone(), tx))
                 .await
                 .ok();
 
@@ -323,7 +322,7 @@ impl<TRepoTypes: RepoTypes> Repo<TRepoTypes> {
         // FIXME: here's a race: block_store might give Ok(None) and we get to create our
         // subscription after the put has completed. So maybe create the subscription first, then
         // cancel it?
-        if let Some(block) = self.block_store.get(&cid).await? {
+        if let Some(block) = self.get_block_now(&cid).await? {
             Ok(block)
         } else {
             let subscription = self
@@ -342,11 +341,11 @@ impl<TRepoTypes: RepoTypes> Repo<TRepoTypes> {
 
     /// Retrives a block from the block store if it's available locally.
     pub async fn get_block_now(&self, cid: &Cid) -> Result<Option<Block>, Error> {
-        Ok(self.block_store.get(&cid).await?)
+        self.block_store.get(&cid).await
     }
 
     pub async fn list_blocks(&self) -> Result<Vec<Cid>, Error> {
-        Ok(self.block_store.list().await?)
+        self.block_store.list().await
     }
 
     /// Remove block from the block store.
@@ -355,20 +354,21 @@ impl<TRepoTypes: RepoTypes> Repo<TRepoTypes> {
             return Err(anyhow::anyhow!("block to remove is pinned"));
         }
 
-        // sending only fails if the background task has exited
-        self.events
-            .clone()
-            .send(RepoEvent::UnprovideBlock(cid.clone()))
-            .await
-            .ok();
-
         // FIXME: Need to change location of pinning logic.
         // I like this pattern of the repo abstraction being some sort of
         // "clearing house" for the underlying result enums, but this
         // could potentially be pushed out out of here up to Ipfs, idk
         match self.block_store.remove(&cid).await? {
             Ok(success) => match success {
-                BlockRm::Removed(_cid) => Ok(cid.clone()),
+                BlockRm::Removed(_cid) => {
+                    // sending only fails if the background task has exited
+                    self.events
+                        .clone()
+                        .send(RepoEvent::RemovedBlock(cid.clone()))
+                        .await
+                        .ok();
+                    Ok(cid.clone())
+                }
             },
             Err(err) => match err {
                 BlockRmError::NotFound(_cid) => Err(anyhow::anyhow!("block not found")),

src/subscription.rs

@@ -173,7 +173,9 @@ impl<T: Debug + Clone + PartialEq, E: Debug + Clone> SubscriptionRegistry<T, E>
             }
         }
 
-        trace!("Cancelled {} subscriptions", cancelled);
+        if cancelled > 0 {
+            trace!("Cancelled {} subscriptions", cancelled);
+        }
     }
 }
 

tests/block_exchange.rs

@@ -0,0 +1,47 @@
+use cid::{Cid, Codec};
+use ipfs::Block;
+use multihash::Sha2_256;
+use std::time::Duration;
+use tokio::time::timeout;
+
+mod common;
+use common::{spawn_nodes, Topology};
+
+fn create_block() -> Block {
+    let data = b"hello block\n".to_vec().into_boxed_slice();
+    let cid = Cid::new_v1(Codec::Raw, Sha2_256::digest(&data));
+
+    Block { cid, data }
+}
+
+// verify that a put block can be received via get_block and the data matches
+#[tokio::test(max_threads = 1)]
+async fn two_node_put_get() {
+    let nodes = spawn_nodes(2, Topology::Line).await;
+    let block = create_block();
+
+    nodes[0].put_block(block.clone()).await.unwrap();
+    let found_block = timeout(Duration::from_secs(10), nodes[1].get_block(&block.cid))
+        .await
+        .expect("get_block did not complete in time")
+        .unwrap();
+
+    assert_eq!(block.data, found_block.data);
+}
+
+// check that a long line of nodes still works with get_block
+#[tokio::test(max_threads = 1)]
+async fn long_get_block() {
+    // this number could be higher, but it starts hanging above ~24
+    const N: usize = 10;
+    let nodes = spawn_nodes(N, Topology::Line).await;
+    let block = create_block();
+
+    // the first node should get the block from the last one...
+    nodes[N - 1].put_block(block.clone()).await.unwrap();
+    nodes[0].get_block(&block.cid).await.unwrap();
+
+    // ...and the last one from the first one
+    nodes[0].put_block(block.clone()).await.unwrap();
+    nodes[N - 1].get_block(&block.cid).await.unwrap();
+}

tests/common/mod.rs

@@ -61,3 +61,54 @@ pub async fn spawn_nodes(count: usize, topology: Topology) -> Vec<Node> {
 
     nodes
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const N: usize = 5;
+
+    #[tokio::test(max_threads = 1)]
+    async fn check_topology_line() {
+        let nodes = spawn_nodes(N, Topology::Line).await;
+
+        for (i, node) in nodes.iter().enumerate() {
+            if i == 0 || i == N - 1 {
+                assert_eq!(node.peers().await.unwrap().len(), 1);
+            } else {
+                assert_eq!(node.peers().await.unwrap().len(), 2);
+            }
+        }
+    }
+
+    #[tokio::test(max_threads = 1)]
+    async fn check_topology_ring() {
+        let nodes = spawn_nodes(N, Topology::Ring).await;
+
+        for node in &nodes {
+            assert_eq!(node.peers().await.unwrap().len(), 2);
+        }
+    }
+
+    #[tokio::test(max_threads = 1)]
+    async fn check_topology_mesh() {
+        let nodes = spawn_nodes(N, Topology::Mesh).await;
+
+        for node in &nodes {
+            assert_eq!(node.peers().await.unwrap().len(), N - 1);
+        }
+    }
+
+    #[tokio::test(max_threads = 1)]
+    async fn check_topology_star() {
+        let nodes = spawn_nodes(N, Topology::Star).await;
+
+        for (i, node) in nodes.iter().enumerate() {
+            if i == 0 {
+                assert_eq!(node.peers().await.unwrap().len(), N - 1);
+            } else {
+                assert_eq!(node.peers().await.unwrap().len(), 1);
+            }
+        }
+    }
+}