Race condition when dialing peer and opening a substream

[dmitry-markin]

If we dial an address and then try to open a substream on LiteP2pEvent::ConnectionEstablished, opening a substream fails ~30% of time.

Timeline:
T0: litep2p.dial_address(address).await
T1: Litep2pEvent::ConnectionEstablished received
T2: self.service.open_substream(peer) — fails with error=PeerDoesNotExist
T3: self.service.dial(peer) — reports AlreadyConnected
T4: self.service.open_substream(peer) — fails for a second time with error=PeerDoesNotExist

Discovered while working on #501.

[lexnv]

Could it be that between T3 and T4, the peer had already disconnected from us? However, it seems highly unlikely that this would happen in 30% of cases

[lexnv]

Indeed this looks like a race between the transport manager and the transport service (which is injected into all protocols):

• On one hand, the transport service dial method calls directly into the transport handle:

litep2p/src/protocol/transport_service.rs

Lines 489 to 498 in 493210e

	pub fn dial(&mut self, peer: &PeerId) -> Result<(), ImmediateDialError> {
	tracing::trace!(
	target: LOG_TARGET,
	?peer,
	protocol = %self.protocol,
	"Dial peer requested",
	);
	self.transport_handle.dial(peer)
	}

This contains the up-to-date state since the peers are shared between the transport_handle and transport manager via a RWlock:

litep2p/src/transport/manager/handle.rs

Lines 303 to 310 in 493210e

	let peers = self.peers.read();
	let Some(PeerContext { state, addresses }) = peers.get(peer) else {
	return Err(ImmediateDialError::NoAddressAvailable);
	};
	match state.can_dial() {
	StateDialResult::AlreadyConnected =>
	return Err(ImmediateDialError::AlreadyConnected),

Effectively, this method reads the same underlying data as the transport manager.

• On the other hand, the open_substream is using its own state that is updated via tokio channels:

litep2p/src/protocol/transport_service.rs

Lines 627 to 638 in 493210e

	Some(InnerTransportEvent::ConnectionEstablished {
	peer,
	endpoint,
	sender,
	connection,
	}) => {
	if let Some(event) =
	self.on_connection_established(peer, endpoint, connection, sender)
	{
	return Poll::Ready(Some(event));
	}
	}

This event will then populate the internal structures and make the "connection" available to the transport service:

litep2p/src/protocol/transport_service.rs

Lines 386 to 389 in 493210e

	"primary connection established",
	);
	self.connections.insert(peer, ConnectionContext::new(handle));

Issue

• There is clearly a gap of time between the moment when the transport manager accepts a peer connection (via rwlock), to the moment when it becomes available to the transport service (via events)

• The internals of the transport layers (TCP/Webscoket) are sending the ConnectionEstablished to all installed protocols, which later forwards it to the service layer:

litep2p/src/transport/tcp/connection.rs

Lines 734 to 737 in 493210e

	pub(crate) async fn start(mut self) -> crate::Result<()> {
	self.protocol_set
	.report_connection_established(self.peer, self.endpoint.clone())
	.await?;

• We can aim at minimizing the time between T1 and T2 failure via turning the fn accept into an async fn accept which returns only when all protocols have been informed of the connected peer

litep2p/src/transport/manager/mod.rs

Lines 1273 to 1282 in 493210e

	let _ = self
	.transports
	.get_mut(&transport)
	.expect("transport to exist")
	.accept(endpoint.connection_id());
	return Some(TransportEvent::ConnectionEstablished {
	peer,
	endpoint,
	});

• It might be a good opportunity to try to rethink this architecture to gain instant access to the peers across all services/protocols