Peer address management across behaviors and connection errors during simultaneous peer startup

[tqwewe]

I'm experiencing connection issues when peers start simultaneously, and I'm unclear about the proper way to manage peer addresses across multiple behaviors.

Setup

I have a behavior combining mdns, kad, and request_response. When mDNS discovers peers, I need to make them available to both kademlia and request_response.

API Usage Question

request_response::Behaviour::add_address is deprecated in favor of Swarm::add_peer_address, but kad::Behaviour::add_address is not deprecated. This suggests different address management patterns:

// For mDNS discovered peers, do I need to explicitly add it to the swarm AND kademlia?
match event {
    SwarmEvent::Behaviour(MyBehaviourEvent::Mdns(mdns::Event::Discovered(list))) => {
        for (peer_id, multiaddr) in list {
            swarm.add_peer_address(peer_id, multiaddr.clone()); 
            swarm.behaviour_mut().kad.add_address(&peer_id, multiaddr); // Still needed?
        }
    }
}

My Current Implementation

To avoid manual kademlia address management, I implemented this in my wrapper's NetworkBehaviour::on_swarm_event:

fn on_swarm_event(&mut self, event: FromSwarm<'_>) {
    match event {
        FromSwarm::NewExternalAddrOfPeer(NewExternalAddrOfPeer { peer_id, addr }) => {
            self.kademlia.add_address(&peer_id, addr.clone());
        }
        FromSwarm::DialFailure(DialFailure {
            peer_id: Some(peer_id),
            error: DialError::Transport(errors),
            ..
        }) => {
            for (addr, _) in errors {
                self.kademlia.remove_address(&peer_id, addr);
            }
        }
        _ => {}
    }
    self.kademlia.on_swarm_event(event)
}

This is replicating the code I found here:

rust-libp2p/swarm/src/behaviour/peer_addresses.rs

Lines 24 to 41 in 1aa4337

	pub fn on_swarm_event(&mut self, event: &FromSwarm) -> bool {
	match event {
	FromSwarm::NewExternalAddrOfPeer(NewExternalAddrOfPeer { peer_id, addr }) => {
	self.add(*peer_id, (*addr).clone())
	}
	FromSwarm::DialFailure(DialFailure {
	peer_id: Some(peer_id),
	error: DialError::Transport(errors),
	..
	}) => {
	for (addr, _error) in errors {
	self.remove(peer_id, addr);
	}
	true
	}
	_ => false,
	}
	}

The Problem

This works when peers start at different times, but when two peers start simultaneously, I consistently get:

Other(Custom { kind: Other, error: Other(Left(Right(Apply(Io(Custom { kind: InvalidData, error: Input }))))) })

This InvalidData error causes discovered addresses to be immediately removed via the DialFailure handler, preventing successful connections.

Questions

1. Why doesn't kademlia automatically handle NewExternalAddrOfPeer like request_response does?
2. Is my manual address management approach correct, or should I be calling both swarm.add_peer_address() and kad.add_address() explicitly?
3. What could cause the InvalidData error specifically during simultaneous startup, and how can I handle it properly?

[tqwewe]

A simple workaround I implemented for now is to delay calling self.kademlia.remove_address(&peer_id, addr); until 5 seconds later. And if within those 5 seconds, the a connection is established with that peer, then it wont be removed anymore.

pub struct Behaviour {
    kademlia: kad::Behaviour<kad::store::MemoryStore>,
    pending_peers: HashMap<(PeerId, Multiaddr), Instant>,
}

impl NetworkBehaviour for Behaviour {
    // ...

    fn on_swarm_event(&mut self, event: FromSwarm<'_>) {
        match event {
            FromSwarm::ConnectionEstablished(ConnectionEstablished {
                peer_id,
                failed_addresses,
                ..
            }) => {
                self.pending_peers.retain(|(pending_peer_id, addr), _| {
                    // Keep all entries for different peers
                    pending_peer_id != &peer_id
                    // OR keep same-peer entries that failed
                    || failed_addresses.iter().any(|failed_addr| failed_addr == addr)
                });
            }
            FromSwarm::NewExternalAddrOfPeer(NewExternalAddrOfPeer { peer_id, addr }) => {
                self.kademlia.add_address(&peer_id, addr.clone());
            }
            FromSwarm::DialFailure(DialFailure {
                peer_id: Some(peer_id),
                error: DialError::Transport(errors),
                ..
            }) => {
                let now = Instant::now();
                for (addr, _) in errors {
                    self.pending_peers
                        .entry((peer_id, addr.clone()))
                        .or_insert(now);
                }
            }
            _ => {}
        }

        self.kademlia.on_swarm_event(event)
    }

    fn poll(
        &mut self,
        cx: &mut task::Context<'_>,
    ) -> task::Poll<ToSwarm<Self::ToSwarm, THandlerInEvent<Self>>> {
        let failed_peers = self
            .pending_peers
            .extract_if(|_, failed_at| failed_at.elapsed() > Duration::from_secs(5));
        for ((peer_id, addr), _) in failed_peers {
            self.kademlia.remove_address(&peer_id, &addr);
        }

        // ...
    }
}

Any advice on if this approach is good or has issues, please let me know. I believe the InvalidData error I was getting is in fact a bug, and this is a simple workaround for it.