Peer Discovery with Kademlia

[anilaltuner]

Hello,

I am new to the project. I am working on peer discovery with Kademlia. I have three different nodes;

Node A and Node B -> worker nodes on a different server

Bootstrap Node -> My node running separately, with an open port.

Node A and Node B are listening on port *4001 and I give the bootstrap node to both nodes.

The situation is as follows; the messages I send from Node A or B are forwarded to each other. In the same way, Bootstrap node finds both nodes as peers and adds them to the Routing Table. But Node A and Node B cannot add each other to the routing table.

Swarm::poll: libp2p_swarm: Connection attempt to peer failed with WrongPeerId { obtained: PeerId("12D3KooWJQbiXjjknhcCNXFedYa3xp9NjUnYYd4WBkGspeNy9hb4"), endpoint: Dialer { address: "/ip4/<addr>/tcp/4001/p2p/12D3KooWEeAy4CrXswPoS9s3uPNTtkr46FjjvEkFAwxXSbGRMXvF", role_override: Dialer } } }. peer=12D3KooWEeAy4CrXswPoS9s3uPNTtkr46FjjjvEkFAwxXSbGRMXvF

Since the bootstrap node is given as multiaddr, it dials from the public ID of the bootstrap node, but when dialing to the node to which Bootstrap is connected, is dialed from the local address.

What could be the problem?

Here is my code;

use std::collections::hash_map::DefaultHasher;
use std::error::Error;
use std::hash::{Hash, Hasher};
use std::str::FromStr;
use std::time::Duration;

use futures::stream::StreamExt;
use libp2p::{gossipsub, identify, identity, kad, mdns, noise, PeerId, StreamProtocol, swarm::NetworkBehaviour, swarm::SwarmEvent, tcp, yamux};
use libp2p::gossipsub::TopicHash;
use libp2p::identity::{Keypair, PublicKey};
use libp2p::kad::{GetClosestPeersError, GetClosestPeersOk, Mode, PROTOCOL_NAME, QueryResult};
use libp2p::kad::store::MemoryStore;
use libp2p::Multiaddr;
use tokio::{io, io::AsyncBufReadExt, select};
use tracing_subscriber::EnvFilter;

const DRIA_PROTO_NAME: StreamProtocol = StreamProtocol::new("/ipfs/kad/1.0.0");

#[derive(NetworkBehaviour)]
struct MyBehaviour {
    gossipsub: gossipsub::Behaviour,
    kademlia: kad::Behaviour<MemoryStore>,
    identify: identify::Behaviour,
}

fn create_kademlia_behavior(local_peer_id: PeerId) -> kad::Behaviour<MemoryStore> {
    // Create a Kademlia behaviour.
    let mut cfg = kad::Config::default();
    println!("{:?}", DRIA_PROTO_NAME.to_string());
    cfg.set_protocol_names(vec![DRIA_PROTO_NAME]);
    cfg.set_query_timeout(Duration::from_secs(5 * 60));
    let store = MemoryStore::new(local_peer_id);
    kad::Behaviour::with_config(local_peer_id, store, cfg)
}

fn create_identify_behavior(local_public_key: PublicKey) -> identify::Behaviour {
    let id_cfg = identify::Config::new( DRIA_PROTO_NAME.to_string(), local_public_key);
    println!("{:?}", DRIA_PROTO_NAME.to_string());
    identify::Behaviour::new(id_cfg)
}

fn create_gossipsub_behavior(id_keys: Keypair) -> gossipsub::Behaviour {
    // To content-address message, we can take the hash of message and use it as an ID.
    let message_id_fn = |message: &gossipsub::Message| {
        let mut s = DefaultHasher::new();
        message.data.hash(&mut s);
        gossipsub::MessageId::from(s.finish().to_string())
    };
    // Set a custom gossipsub
    let gossipsub_config = gossipsub::ConfigBuilder::default()
        .heartbeat_interval(Duration::from_secs(10)) // This is set to aid debugging by not cluttering the log space
        .validation_mode(gossipsub::ValidationMode::Strict) // This sets the kind of message validation. The default is Strict (enforce message signing)
        .message_id_fn(message_id_fn) // content-address messages. No two messages of the same content will be propagated.
        .build()
        .map_err(|msg| io::Error::new(io::ErrorKind::Other, msg)).unwrap(); // Temporary hack because `build` does not return a proper `std::error::Error`.

    gossipsub::Behaviour::new(
        gossipsub::MessageAuthenticity::Signed(id_keys),
        gossipsub_config,
    ).unwrap()
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let _ = tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::from_default_env())
        .try_init();

    if let Some(local_ip) = public_ip::addr().await {
        println!("public ip address: {:?}", local_ip);
    } else {
        println!("couldn't get an IP address");
    }

    let mut swarm = libp2p::SwarmBuilder::with_new_identity()
        .with_tokio()
        .with_tcp(
            tcp::Config::default(),
            noise::Config::new,
            yamux::Config::default,
        )?
        .with_quic()
        .with_behaviour(|key| {
            println!("key: {:?}", key.public());
            println!("peer id: {:?}", key.public().to_peer_id());
            Ok(MyBehaviour {
                gossipsub: create_gossipsub_behavior(key.clone()),
                kademlia: create_kademlia_behavior(key.public().to_peer_id()),
                identify: create_identify_behavior(key.public()),
            })
        })?
        .with_swarm_config(|c| c.with_idle_connection_timeout(Duration::from_secs(60)))
        .build();
    println!("My peer addr is {:?}", swarm.local_peer_id());
    swarm.behaviour_mut().kademlia.set_mode(Some(Mode::Server));


    let topic = gossipsub::IdentTopic::new("foo");
    // subscribes to our topic
    swarm.behaviour_mut().gossipsub.subscribe(&topic)?;

    let bootstrap_nodes = vec![
        <BOOTSTRAP_NODE>,
    ];

    for addr in bootstrap_nodes {
        match addr.parse::<libp2p::Multiaddr>() {
            Ok(addr) => {
                if let Some(peer_id) = addr.iter().find_map(|p| match p {
                    libp2p::multiaddr::Protocol::P2p(peer_id) => Some(peer_id),
                    _ => None,
                }) {
                    println!("dialing peer {}", addr);
                    swarm.dial(addr.clone())?;
                    println!("adding initial peer addr to kademli: routing table: {addr}");
                    // Add the bootnodes to the local routing table.
                    swarm
                        .behaviour_mut()
                        .kademlia
                        .add_address(&peer_id, addr.clone());
                    //doesn't seem to be completely necessary
                    swarm
                        .behaviour_mut()
                        .identify
                        .push(std::iter::once(peer_id));
                } else {
                    println!("Missing peer ID in address: {}", addr);
                }
            }
            Err(err) => println!("Failed to parse address {}: {}", addr, err),
        }
    }


    // Order Kademlia to search for peers.
    let to_search: PeerId = Keypair::generate_ed25519().public().into();
    println!("Searching for the closest peers to {:?}", to_search);
    swarm.behaviour_mut().kademlia.get_closest_peers(to_search);
    swarm.behaviour_mut().kademlia.bootstrap()?;

    // Listen on all interfaces and whatever port the OS assigns
    swarm.listen_on("/ip4/0.0.0.0/tcp/4001".parse()?)?;

    // Read full lines from stdin
    let mut stdin = io::BufReader::new(io::stdin()).lines();
    println!("Enter messages via STDIN and they will be sent to connected peers using Gossipsub");

    // Kick it off
    loop {
        select! {
            Ok(Some(line)) = stdin.next_line() => {
                if line.starts_with("/connect ") {
                    let addr: libp2p::Multiaddr = line[9..].parse()?;
                    swarm.dial(addr)?;
                } else {
                    if let Err(e) = swarm
                        .behaviour_mut().gossipsub
                        .publish(topic.clone(), line.as_bytes()) {
                        println!("Publish error: {e:?}");
                    }
                }
            }
            event = swarm.select_next_some() => match event {
                SwarmEvent::Behaviour(MyBehaviourEvent::Kademlia(kad::Event::OutboundQueryProgressed {
                    result: QueryResult::GetClosestPeers(result),
                    ..
                })) => {
                    match result {
                        Ok(GetClosestPeersOk { key, peers }) => {
                            if !peers.is_empty() {
                                println!("Query finished with closest peers: {:#?}", peers);
                                for peer in peers {
                                    println!("gossipsub adding peer {peer}");
                                    swarm.behaviour_mut().gossipsub.add_explicit_peer(&peer);
                                }
                            } else {
                                println!("Query finished with no closest peers.")
                            }
                        }

                        Err(GetClosestPeersError::Timeout { peers, .. }) => {
                            if !peers.is_empty() {
                                println!("Query timed out with closest peers: {:#?}", peers);
                                for peer in peers {
                                    println!("gossipsub adding peer {peer}");
                                    swarm.behaviour_mut().gossipsub.add_explicit_peer(&peer);
                                }
                            } else {
                                println!("Query timed out with no closest peers.");
                            }
                        }
                    };
                }
                SwarmEvent::Behaviour(MyBehaviourEvent::Identify(identify::Event::Received {
                    peer_id,
                    info:
                        identify::Info {
                            listen_addrs,
                            protocols,
                            ..
                        },
                })) => {
                    println!("Protocols {:?}", protocols);
                    if protocols.iter().any(|p| *p == DRIA_PROTO_NAME) {
                        for addr in listen_addrs {
                            println!("received addr {addr} through identify. Peer is {:?}", peer_id);

                            swarm.behaviour_mut().kademlia.add_address(&peer_id, addr);
                        }
                    } else {
                        println!("something funky happened, investigate it");
                    }
                }
                SwarmEvent::Behaviour(MyBehaviourEvent::Gossipsub(gossipsub::Event::Message {
                    propagation_source: peer_id,
                    message_id: id,
                    message,
                })) => {println!(
                    "Got message: '{}'with id: {id} from peer: {peer_id}",
                    String::from_utf8_lossy(&message.data),
                );
                let random_peer_id = PeerId::random();
                println!("Searching for peers close to {:?}", random_peer_id);
                swarm.behaviour_mut().kademlia.get_closest_peers(random_peer_id);
                }
                SwarmEvent::NewListenAddr { address, .. } => {
                    println!("Local node is listening on {address}");
                }
                _ => println!("Swarm Event {:?}", event),
            }
        }
    }
}

[guillaumemichel]

Welcome to libp2p @anilaltuner!

The solution is certainly to put the Kademlia nodes in server modes, so that they add each other to their routing tables. See this similar issue and let me know if the suggested solution doesn't work.

[anilaltuner]

Thanks @guillaumemichel!

I did but one thing for sure, I saw they add each other but cannot dial themselves and they simply can't peered each other probably cause of ports are closed. Am I need implement nat traversal for it?

[guillaumemichel]

Sorry, I missed it.

What is your network setup? Do you run all nodes within the same network? Are the nodes hosted in a public could/home network/other? Is there a firewall in place?

Swarm::poll: libp2p_swarm: Connection attempt to peer failed with WrongPeerId { obtained: PeerId("12D3KooWJQbiXjjknhcCNXFedYa3xp9NjUnYYd4WBkGspeNy9hb4"), endpoint: Dialer { address: "/ip4/<addr>/tcp/4001/p2p/12D3KooWEeAy4CrXswPoS9s3uPNTtkr46FjjvEkFAwxXSbGRMXvF", role_override: Dialer } } }. peer=12D3KooWEeAy4CrXswPoS9s3uPNTtkr46FjjjvEkFAwxXSbGRMXvF

Is the <addr> a private or a public IP? Also it seems that the error comes from the wrong peer id being used rather than a network issue. Could it be possible?

Could you also share the full output (incl. showing that the nodes A and B don't connect to each other)?

[anilaltuner]

Bootstrap Node: running on AWS, port 4001 (listening port) is open
Node A: Running on local computer, port 4001 closed
Node B: Running on the local computer but on a different network to A. Also 4001 is down

Both local computers are using a home network, not that of an office or organization.

The point is that both Node A and Node B connect to bootstrap, and bootstrap can add them both as peers.

When a connection request comes from the local node to the bootstrap node, it logs;

Swarm Event IncomingConnection { connection_id: ConnectionId(6), local_addr: "/ip4/172.31.8.99/tcp/4001", send_back_addr: "/ip4/212.98.218.10/tcp/49510" }
Swarm Event ConnectionEstablished { peer_id: PeerId("12D3KooWCG7ZNayazLcn5iJhrdBBy7EMztRMkTptTVZwBNgqpwMc"), connection_id: ConnectionId(6), endpoint: Listener { local_addr: "/ip4/212.98.218.10/tcp/4001", send_back_addr: "/ip4/212.98.218.10/tcp/49510" }, num_established: 1, concurrent_dial_errors: None, established_in: 271.696204ms }
Swarm Event Behavior(MyBehaviourEvent: Sent { peer_id: PeerId("12D3KooWCG7ZNayazLcn5iJhrdBBy7EMztRMkTptTVZwBNgqpwMc") })
Swarm Event Behavior(MyBehaviourEvent: UnroutablePeer { peer: PeerId("12D3KooWCG7ZNayazLcn5iJhrdBBy7EMztRMkTptTVZwBNgqpwMc") })
Swarm Event Behavior(MyBehaviourEvent: Subscribed { peer_id: PeerId("12D3KooWCG7ZNayazLcn5iJhrdBBy7EMztRMkTptTVZwBNgqpwMc"), topic: TopicHash { hash: "foo" } })

Of course, in this case there is a direct dialing between bootstrap node and local node. Because I give the local node the public multiaddr of bootstrap

And the other situation between local nodes, here are the logs;

Swarm Event Dialing { peer_id: Some(PeerId("12D3KooWCG7ZNayazLcn5iJhrdBBy7EMztRMkTptTVZwBNgqpwMc")), connection_id: ConnectionId(7) }

Swarm Event OutgoingConnectionError { connection_id: ConnectionId(7), peer_id: Some(PeerId("12D3KooWCG7ZNayazLcn5iJhrdBBy7EMztRMkTptTVZwBNgqpwMc"))), error: Transport([("/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWCG7ZNayazLcn5iJhrdBBy7EMztRMkTptTVZwBNgqpwMc", Other(Custom { kind: Other, error: Other(Left(Left(Left(Os { code: 22, kind: InvalidInput, message: "Invalid argument" })))) })), ("/ip4/192.168.182.229/tcp/4001/p2p/12D3KooWCG7ZNayazLcn5iJhrdBBy7EMztRMkTptTVZwBNgqpwMc", Other(Custom { kind: Other, error: Timeout }))]) }

In this case, when node A adds the Bootstrap node, it wants to call other nodes from the DHT. But when it tries to connect, it uses different ip than the bootsrap node.

One extra note, when I open the 4001 ports of Node A, it can match with Node B.

[guillaumemichel]

One extra note, when I open the 4001 ports of Node A, it can match with Node B.

Do you mean that you don't have connectivity issues when you open the port 4001 for Node A? If that is the case, then the problem doesn't come from Kademlia but from your network connectivity.

If you don't want to open the port, I suggest you look into UPnp and DCUtR, you can learn more about libp2p NAT Traversal here.

[anilaltuner]

Great! I needed the direction of peer routing without open the port and it seems the answer for rn