fix(net): Avoid potential concurrency bugs in outbound handshakes

zebra-network/src/peer_set/candidate_set.rs

@@ -125,7 +125,11 @@ mod tests;
 // When we add the Seed state:
 //   * show that seed peers that transition to other never attempted
 //     states are already in the address book
-pub(crate) struct CandidateSet<S> {
+pub(crate) struct CandidateSet<S>
+where
+    S: Service<Request, Response = Response, Error = BoxError> + Send,
+    S::Future: Send + 'static,
+{
     // Correctness: the address book must be private,
     //              so all operations are performed on a blocking thread (see #1976).
     address_book: Arc<std::sync::Mutex<AddressBook>>,
@@ -136,7 +140,7 @@ pub(crate) struct CandidateSet<S> {
 
 impl<S> CandidateSet<S>
 where
-    S: Service<Request, Response = Response, Error = BoxError>,
+    S: Service<Request, Response = Response, Error = BoxError> + Send,
     S::Future: Send + 'static,
 {
     /// Uses `address_book` and `peer_service` to manage a [`CandidateSet`] of peers.
@@ -180,8 +184,6 @@ where
     /// The handshaker sets up the peer message receiver so it also sends a
     /// [`Responded`] peer address update.
     ///
-    /// [`report_failed`][Self::report_failed] puts peers into the [`Failed`] state.
-    ///
     /// [`next`][Self::next] puts peers into the [`AttemptPending`] state.
     ///
     /// ## Security
@@ -411,21 +413,9 @@ where
         Some(next_peer)
     }
 
-    /// Mark `addr` as a failed peer.
-    pub async fn report_failed(&mut self, addr: &MetaAddr) {
-        let addr = MetaAddr::new_errored(addr.addr, addr.services);
-
-        // # Correctness
-        //
-        // Spawn address book accesses on a blocking thread,
-        // to avoid deadlocks (see #1976).
-        let address_book = self.address_book.clone();
-        let span = Span::current();
-        tokio::task::spawn_blocking(move || {
-            span.in_scope(|| address_book.lock().unwrap().update(addr))
-        })
-        .await
-        .expect("panic in peer failure address book update task");
+    /// Returns the address book for this `CandidateSet`.
+    pub async fn address_book(&self) -> Arc<std::sync::Mutex<AddressBook>> {
+        self.address_book.clone()
     }
 }
 

zebra-network/src/peer_set/candidate_set/tests/prop.rs

@@ -139,7 +139,7 @@ proptest! {
 /// - if no reconnection peer is returned at all.
 async fn check_candidates_rate_limiting<S>(candidate_set: &mut CandidateSet<S>, candidates: u32)
 where
-    S: tower::Service<Request, Response = Response, Error = BoxError>,
+    S: tower::Service<Request, Response = Response, Error = BoxError> + Send,
     S::Future: Send + 'static,
 {
     let mut now = Instant::now();