req-resp: Replace SubstreamSet with FuturesStream

[lexnv]

This PR replaces the in-house built SubstreamSet with the FuturesStream.

• provide fairness for substreams that produce elements based on the FuturesUnordered implementation
• clone free implementation for peerId and requested
• substream object is readily available once the substream produces a request (ie there's no need to pay for the hashmap element retrieval)

Substreams are no longer removed from the FuturesStream and instead they are left to expire with the ConnectionClosed event. This has the side-effect of emitting a debug log and it simplifies state tracking.

cc @paritytech/networking

[dmitry-markin]

Is there a way to check here if substream points to alive substream over an existing connection, and not process the request if not? May be we can do this check in on_inbound_request.

[lexnv]

If the connection has been closed, we'll remove the peer-associated data, which will lead to no further polling on the substream:

litep2p/src/protocol/request_response/mod.rs

Lines 302 to 307 in 4bc31d0

	/// Connection closed to remote peer.
	async fn on_connection_closed(&mut self, peer: PeerId) {
	tracing::debug!(target: LOG_TARGET, ?peer, protocol = %self.protocol, "connection closed");
	// Remove any pending outbound substreams for this peer.
	self.pending_outbound.retain(|_, context| context.peer != peer);

The inbound request is directly rejected as a first step here:

litep2p/src/protocol/request_response/mod.rs

Lines 477 to 489 in 4bc31d0

	.active_inbound
	.remove(&request_id)
	.ok_or_else(|| {
	tracing::debug!(
	target: LOG_TARGET,
	?peer,
	protocol = %self.protocol,
	?request_id,
	"no active inbound request",
	);
	Error::InvalidState
	})?;

If the connection has been closed and detected by the substream, we'll return a Substream::Closed error which will not propagate the message:

litep2p/src/protocol/request_response/mod.rs

Lines 501 to 511 in 4bc31d0

	let Ok(request) = request else {
	tracing::debug!(
	target: LOG_TARGET,
	?peer,
	%protocol,
	?request_id,
	?request,
	"failed to read request from substream",
	);
	return Err(Error::InvalidData);
	};

[dmitry-markin]

i mean if we received a request and hold it in pending_inbound_requests, but the peer gets disconnected before we poll pending_inbound_requests, we will process the request anyway and only hit the error when try to send the response to a closed substream? Is there an easy way to not process the inbound request (as it may require resources on the client/runtime side etc.) in this case and fail early?

[lexnv]

Had a chat offline: we'll not process the inbound requests because we remove the peer state in on_connection_closed, have added a comment to make the code easier to follow : 58d5742 🙏

[dmitry-markin]

So with the new logic we will handle the request anyway if the connection is closed, hitting an error when we try to send a response?

[lexnv]

Yep exactly, if the connection gets closed, we'll still have the substream in the FuturesUnordered and let them hit an error (either when we send out the request, or read the response) 🙏