forward request messages to connected peers

Context: when a peer requests a document from us we often don't want to
just check our own storage, we also want to ask all other peers we are
connected to if they have the document. This is the case for e.g. a sync
server which is relaying documents between two peers.

Problem: the current immplementation just checks local storage

Solution: when we receive a request, first send a request out to all the
peers we are connected to and once they have all responded only then
send a response to the requestor.

Cargo.toml

@@ -65,3 +65,4 @@ autosurgeon = "0.8.0"
 bolero = { version = "0.10.0", features = ["arbitrary"] }
 arbitrary = { version = "1.3.1", features = ["derive"] }
 bolero-generator = { version = "0.10.0", features = ["arbitrary"] }
+rand = "0.8.5"

src/network_connect.rs

@@ -1,4 +1,4 @@
-use crate::interfaces::{Message, NetworkError, ProtocolVersion, RepoId, RepoMessage};
+use crate::interfaces::{Message, NetworkError, ProtocolVersion, RepoId};
 use crate::repo::RepoHandle;
 use futures::{Sink, SinkExt, Stream, StreamExt};
 
@@ -22,17 +22,17 @@ impl RepoHandle {
         Str: Stream<Item = Result<Message, RecvErr>> + Send + 'static + Unpin,
     {
         let other_id = self.handshake(&mut stream, &mut sink, direction).await?;
-        tracing::trace!(?other_id, repo_id=?self.get_repo_id(), "Handshake complete");
+        tracing::trace!(?other_id, repo_id=?self.get_repo_id(), "handshake complete");
 
         let stream = stream.map({
             let repo_id = self.get_repo_id().clone();
             move |msg| match msg {
                 Ok(Message::Repo(repo_msg)) => {
-                    tracing::trace!(?repo_msg, repo_id=?repo_id, "Received repo message");
+                    tracing::trace!(?repo_msg, repo_id=?repo_id, "received repo message");
                     Ok(repo_msg)
                 }
                 Ok(m) => {
-                    tracing::warn!(?m, repo_id=?repo_id, "Received non-repo message");
+                    tracing::warn!(?m, repo_id=?repo_id, "received non-repo message");
                     Err(NetworkError::Error(
                         "unexpected non-repo message".to_string(),
                     ))
@@ -48,12 +48,9 @@ impl RepoHandle {
         });
 
         let sink = sink
-            .with_flat_map::<RepoMessage, _, _>(|msg| match msg {
-                RepoMessage::Sync { .. } => futures::stream::iter(vec![Ok(Message::Repo(msg))]),
-                _ => futures::stream::iter(vec![]),
-            })
+            .with::<_, _, _, SendErr>(move |msg| futures::future::ready(Ok(Message::Repo(msg))))
             .sink_map_err(|e| {
-                tracing::error!(?e, "Error sending repo message");
+                tracing::error!(?e, "error sending repo message");
                 NetworkError::Error(format!("error sending repo message: {}", e))
             });