[stable2412] Backport #7011 (#7053)

Backport #7011 into `stable2412` from dmitry-markin.

See the
[documentation](https://github.com/paritytech/polkadot-sdk/blob/master/docs/BACKPORT.md)
on how to use this bot.

<!--
  # To be used by other automation, do not modify:
  original-pr-number: #${pull_number}
-->

---------

Co-authored-by: Dmitry Markin <[email protected]>

prdoc/pr_7053.prdoc

@@ -0,0 +1,16 @@
+title: 'sync: Send already connected peers to new subscribers'
+doc:
+- audience: Node Dev
+  description: |-
+    Introduce `SyncEvent::InitialPeers` message sent to new subscribers to allow them correctly tracking sync peers. This resolves a race condition described in https://github.com/paritytech/polkadot-sdk/issues/6573#issuecomment-2563091343.
+
+    Fixes https://github.com/paritytech/polkadot-sdk/issues/6573.
+crates:
+- name: sc-network-gossip
+  bump: major
+- name: sc-network-statement
+  bump: patch
+- name: sc-network-sync
+  bump: major
+- name: sc-network-transactions
+  bump: patch

substrate/client/network-gossip/src/bridge.rs

@@ -256,10 +256,12 @@ impl<B: BlockT> Future for GossipEngine<B> {
 
 					match sync_event_stream {
 						Poll::Ready(Some(event)) => match event {
-							SyncEvent::PeerConnected(remote) =>
-								this.network.add_set_reserved(remote, this.protocol.clone()),
-							SyncEvent::PeerDisconnected(remote) =>
-								this.network.remove_set_reserved(remote, this.protocol.clone()),
+							SyncEvent::InitialPeers(peer_ids) =>
+								this.network.add_set_reserved(peer_ids, this.protocol.clone()),
+							SyncEvent::PeerConnected(peer_id) =>
+								this.network.add_set_reserved(vec![peer_id], this.protocol.clone()),
+							SyncEvent::PeerDisconnected(peer_id) =>
+								this.network.remove_set_reserved(peer_id, this.protocol.clone()),
 						},
 						// The sync event stream closed. Do the same for [`GossipValidator`].
 						Poll::Ready(None) => {

substrate/client/network-gossip/src/lib.rs

@@ -82,15 +82,18 @@ mod validator;
 
 /// Abstraction over a network.
 pub trait Network<B: BlockT>: NetworkPeers + NetworkEventStream {
-	fn add_set_reserved(&self, who: PeerId, protocol: ProtocolName) {
-		let addr = Multiaddr::empty().with(Protocol::P2p(*who.as_ref()));
-		let result = self.add_peers_to_reserved_set(protocol, iter::once(addr).collect());
+	fn add_set_reserved(&self, peer_ids: Vec<PeerId>, protocol: ProtocolName) {
+		let addrs = peer_ids
+			.into_iter()
+			.map(|peer_id| Multiaddr::empty().with(Protocol::P2p(peer_id.into())))
+			.collect();
+		let result = self.add_peers_to_reserved_set(protocol, addrs);
 		if let Err(err) = result {
 			log::error!(target: "gossip", "add_set_reserved failed: {}", err);
 		}
 	}
-	fn remove_set_reserved(&self, who: PeerId, protocol: ProtocolName) {
-		let result = self.remove_peers_from_reserved_set(protocol, iter::once(who).collect());
+	fn remove_set_reserved(&self, peer_id: PeerId, protocol: ProtocolName) {
+		let result = self.remove_peers_from_reserved_set(protocol, iter::once(peer_id).collect());
 		if let Err(err) = result {
 			log::error!(target: "gossip", "remove_set_reserved failed: {}", err);
 		}

substrate/client/network/statement/src/lib.rs

@@ -33,7 +33,8 @@ use futures::{channel::oneshot, prelude::*, stream::FuturesUnordered, FutureExt}
 use prometheus_endpoint::{register, Counter, PrometheusError, Registry, U64};
 use sc_network::{
 	config::{NonReservedPeerMode, SetConfig},
-	error, multiaddr,
+	error,
+	multiaddr::{Multiaddr, Protocol},
 	peer_store::PeerStoreProvider,
 	service::{
 		traits::{NotificationEvent, NotificationService, ValidationResult},
@@ -296,9 +297,19 @@ where
 
 	fn handle_sync_event(&mut self, event: SyncEvent) {
 		match event {
-			SyncEvent::PeerConnected(remote) => {
-				let addr = iter::once(multiaddr::Protocol::P2p(remote.into()))
-					.collect::<multiaddr::Multiaddr>();
+			SyncEvent::InitialPeers(peer_ids) => {
+				let addrs = peer_ids
+					.into_iter()
+					.map(|peer_id| Multiaddr::empty().with(Protocol::P2p(peer_id.into())))
+					.collect();
+				let result =
+					self.network.add_peers_to_reserved_set(self.protocol_name.clone(), addrs);
+				if let Err(err) = result {
+					log::error!(target: LOG_TARGET, "Add reserved peers failed: {}", err);
+				}
+			},
+			SyncEvent::PeerConnected(peer_id) => {
+				let addr = Multiaddr::empty().with(Protocol::P2p(peer_id.into()));
 				let result = self.network.add_peers_to_reserved_set(
 					self.protocol_name.clone(),
 					iter::once(addr).collect(),
@@ -307,10 +318,10 @@ where
 					log::error!(target: LOG_TARGET, "Add reserved peer failed: {}", err);
 				}
 			},
-			SyncEvent::PeerDisconnected(remote) => {
+			SyncEvent::PeerDisconnected(peer_id) => {
 				let result = self.network.remove_peers_from_reserved_set(
 					self.protocol_name.clone(),
-					iter::once(remote).collect(),
+					iter::once(peer_id).collect(),
 				);
 				if let Err(err) = result {
 					log::error!(target: LOG_TARGET, "Failed to remove reserved peer: {err}");

substrate/client/network/sync/src/engine.rs

@@ -647,7 +647,11 @@ where
 			ToServiceCommand::SetSyncForkRequest(peers, hash, number) => {
 				self.strategy.set_sync_fork_request(peers, &hash, number);
 			},
-			ToServiceCommand::EventStream(tx) => self.event_streams.push(tx),
+			ToServiceCommand::EventStream(tx) => {
+				let _ = tx
+					.unbounded_send(SyncEvent::InitialPeers(self.peers.keys().cloned().collect()));
+				self.event_streams.push(tx);
+			},
 			ToServiceCommand::RequestJustification(hash, number) =>
 				self.strategy.request_justification(&hash, number),
 			ToServiceCommand::ClearJustificationRequests =>

substrate/client/network/sync/src/types.rs

@@ -127,6 +127,10 @@ where
 
 /// Syncing-related events that other protocols can subscribe to.
 pub enum SyncEvent {
+	/// All connected peers that the syncing implementation is tracking.
+	/// Always sent as the first message to the stream.
+	InitialPeers(Vec<PeerId>),
+
 	/// Peer that the syncing implementation is tracking connected.
 	PeerConnected(PeerId),
 

substrate/client/network/transactions/src/lib.rs

@@ -35,7 +35,8 @@ use log::{debug, trace, warn};
 use prometheus_endpoint::{register, Counter, PrometheusError, Registry, U64};
 use sc_network::{
 	config::{NonReservedPeerMode, ProtocolId, SetConfig},
-	error, multiaddr,
+	error,
+	multiaddr::{Multiaddr, Protocol},
 	peer_store::PeerStoreProvider,
 	service::{
 		traits::{NotificationEvent, NotificationService, ValidationResult},
@@ -377,9 +378,19 @@ where
 
 	fn handle_sync_event(&mut self, event: SyncEvent) {
 		match event {
-			SyncEvent::PeerConnected(remote) => {
-				let addr = iter::once(multiaddr::Protocol::P2p(remote.into()))
-					.collect::<multiaddr::Multiaddr>();
+			SyncEvent::InitialPeers(peer_ids) => {
+				let addrs = peer_ids
+					.into_iter()
+					.map(|peer_id| Multiaddr::empty().with(Protocol::P2p(peer_id.into())))
+					.collect();
+				let result =
+					self.network.add_peers_to_reserved_set(self.protocol_name.clone(), addrs);
+				if let Err(err) = result {
+					log::error!(target: LOG_TARGET, "Add reserved peers failed: {}", err);
+				}
+			},
+			SyncEvent::PeerConnected(peer_id) => {
+				let addr = Multiaddr::empty().with(Protocol::P2p(peer_id.into()));
 				let result = self.network.add_peers_to_reserved_set(
 					self.protocol_name.clone(),
 					iter::once(addr).collect(),
@@ -388,10 +399,10 @@ where
 					log::error!(target: LOG_TARGET, "Add reserved peer failed: {}", err);
 				}
 			},
-			SyncEvent::PeerDisconnected(remote) => {
+			SyncEvent::PeerDisconnected(peer_id) => {
 				let result = self.network.remove_peers_from_reserved_set(
 					self.protocol_name.clone(),
-					iter::once(remote).collect(),
+					iter::once(peer_id).collect(),
 				);
 				if let Err(err) = result {
 					log::error!(target: LOG_TARGET, "Remove reserved peer failed: {}", err);