fix: update external address handling for kad

With libp2p 0.52 external addresses are handled differently. A new event
ExternalAddConfirmed has been added and kad waits for that event to
connect to peers. As such we need to ensure we emit that event.

There are two ways to emit the event:
1. Use autonat to confirm the observed address from identity exchanges.
2. Always trust the observed address from identity exchanges.

beetle/iroh-bitswap/src/handler.rs

@@ -141,7 +141,7 @@ impl Debug for BitswapHandler {
 
 impl BitswapHandler {
     /// Builds a new [`BitswapHandler`].
-    // TODO(nathanielc): Remove uses of KeepAlive::Until
+    // TODO(WS1-1291): Remove uses of KeepAlive::Until
     #[allow(deprecated)]
     pub fn new(protocol_config: ProtocolConfig, idle_timeout: Duration) -> Self {
         Self {
@@ -175,7 +175,7 @@ impl ConnectionHandler for BitswapHandler {
         self.keep_alive
     }
 
-    // TODO(nathanielc): Remove uses of KeepAlive::Until
+    // TODO(WS1-1291): Remove uses of KeepAlive::Until
     #[allow(deprecated)]
     fn poll(&mut self, cx: &mut Context<'_>) -> Poll<BitswapConnectionHandlerEvent> {
         inc!(BitswapMetrics::HandlerPollCount);
@@ -276,7 +276,7 @@ impl ConnectionHandler for BitswapHandler {
         }
     }
 
-    // TODO(nathanielc): Remove uses of KeepAlive::Until
+    // TODO(WS1-1291): Remove uses of KeepAlive::Until
     #[allow(deprecated)]
     fn on_behaviour_event(&mut self, event: Self::FromBehaviour) {
         match event {

beetle/iroh-bitswap/src/lib.rs

@@ -610,7 +610,7 @@ mod tests {
     use libp2p::swarm::SwarmEvent;
     use libp2p::tcp::{tokio::Transport as TcpTransport, Config as TcpConfig};
     use libp2p::yamux;
-    use libp2p::{core::muxing::StreamMuxerBox, swarm::SwarmBuilder};
+    use libp2p::{core::muxing::StreamMuxerBox, swarm};
     use libp2p::{noise, PeerId, Swarm, Transport};
     use tokio::sync::{mpsc, RwLock};
     use tracing::{info, trace};
@@ -741,7 +741,8 @@ mod tests {
         let store1 = TestStore::default();
         let bs1 = Bitswap::new(peer1_id, store1.clone(), Config::default()).await;
 
-        let mut swarm1 = SwarmBuilder::with_tokio_executor(trans, bs1, peer1_id).build();
+        let config = swarm::Config::with_tokio_executor();
+        let mut swarm1 = Swarm::new(trans, bs1, peer1_id, config);
         let blocks = (0..N).map(|_| create_random_block_v1()).collect::<Vec<_>>();
 
         for block in &blocks {
@@ -774,7 +775,8 @@ mod tests {
         let store2 = TestStore::default();
         let bs2 = Bitswap::new(peer2_id, store2.clone(), Config::default()).await;
 
-        let mut swarm2 = SwarmBuilder::with_tokio_executor(trans, bs2, peer2_id).build();
+        let config = swarm::Config::with_tokio_executor();
+        let mut swarm2 = Swarm::new(trans, bs2, peer2_id, config);
 
         let swarm2_bs = swarm2.behaviour().clone();
         let peer2 = tokio::task::spawn(async move {
@@ -790,8 +792,8 @@ mod tests {
                         swarm2.behaviour().on_identify(
                             &peer_id,
                             &[
-                                "/ipfs/bitswap/1.2.0".to_string(),
-                                "/ipfs/bitswap/1.1.0".to_string(),
+                                StreamProtocol::new("/ipfs/bitswap/1.2.0"),
+                                StreamProtocol::new("/ipfs/bitswap/1.1.0"),
                             ],
                         );
                     }

beetle/iroh-bitswap/src/protocol.rs

@@ -197,39 +197,9 @@ impl Decoder for BitswapCodec {
 
 #[cfg(test)]
 mod tests {
-    use futures::prelude::*;
-    use libp2p::core::upgrade;
-    use tokio::net::{TcpListener, TcpStream};
-    use tokio_util::compat::*;
 
     use super::*;
 
-    #[tokio::test]
-    async fn test_upgrade() {
-        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-        let listener_addr = listener.local_addr().unwrap();
-
-        let server = async move {
-            let (incoming, _) = listener.accept().await.unwrap();
-            upgrade::apply_inbound(incoming.compat(), ProtocolConfig::default())
-                .await
-                .unwrap();
-        };
-
-        let client = async move {
-            let stream = TcpStream::connect(&listener_addr).await.unwrap();
-            upgrade::apply_outbound(
-                stream.compat(),
-                ProtocolConfig::default(),
-                upgrade::Version::V1Lazy,
-            )
-            .await
-            .unwrap();
-        };
-
-        future::select(Box::pin(server), Box::pin(client)).await;
-    }
-
     #[test]
     fn test_ord() {
         let mut protocols = [

beetle/iroh-car/src/header.rs

@@ -80,8 +80,8 @@ impl From<Vec<Cid>> for CarHeaderV1 {
 
 #[cfg(test)]
 mod tests {
-    use ipld::codec::{Decode, Encode};
-    use ipld_cbor::DagCborCodec;
+    use libipld::codec::{Decode, Encode};
+    use libipld_cbor::DagCborCodec;
     use multihash::MultihashDigest;
 
     use super::*;

beetle/iroh-car/src/reader.rs

@@ -64,7 +64,7 @@ mod tests {
 
     use cid::Cid;
     use futures::TryStreamExt;
-    use ipld_cbor::DagCborCodec;
+    use libipld_cbor::DagCborCodec;
     use multihash::MultihashDigest;
 
     use crate::{header::CarHeaderV1, writer::CarWriter};

beetle/iroh-metrics/Cargo.toml

@@ -40,7 +40,6 @@ features = [
     "yamux",
     "tcp",
     "dns",
-    #"mplex",
     "request-response",
     "websocket",
     "serde",

p2p/Cargo.toml

@@ -77,6 +77,7 @@ features = [
 [dev-dependencies]
 criterion.workspace = true
 rand_chacha.workspace = true
+test-log.workspace = true
 
 [[bench]]
 name = "lru_cache"

p2p/src/behaviour.rs

@@ -158,12 +158,6 @@ where
                     warn!("Could not parse bootstrap addr {}", multiaddr);
                 }
             }
-
-            // Trigger initial bootstrap
-            if let Err(e) = kademlia.bootstrap() {
-                warn!("Kademlia bootstrap failed: {}", e);
-            }
-
             Some(kademlia)
         } else {
             None

p2p/src/config.rs

@@ -64,6 +64,11 @@ pub struct Libp2pConfig {
     pub connection_event_buffer_size: usize,
     pub dial_concurrency_factor: NonZeroU8,
     pub idle_connection_timeout: Duration,
+    /// Trust as a confirmed external address any reported observed address.
+    ///
+    /// NOTE: It is generally not safe to trust observed addresses received from arbitrary peers.
+    /// Only enable this option if its known that all connecting peers can be trusted.
+    pub trust_observed_addrs: bool,
 }
 
 impl Default for Libp2pConfig {
@@ -91,7 +96,8 @@ impl Default for Libp2pConfig {
             notify_handler_buffer_size: NonZeroUsize::new(256).expect("should not be zero"),
             connection_event_buffer_size: 256,
             dial_concurrency_factor: NonZeroU8::new(8).expect("should not be zero"),
-            idle_connection_timeout: Duration::from_secs(10),
+            idle_connection_timeout: Duration::from_secs(30),
+            trust_observed_addrs: false,
         }
     }
 }