Extract warp sync strategy from ChainSync (#2467)

Extract `WarpSync` (and `StateSync` as part of warp sync) from
`ChainSync` as independent syncing strategy called by `SyncingEngine`.
Introduce `SyncingStrategy` enum as a proxy between `SyncingEngine` and
specific syncing strategies.

## Limitations
Gap sync is kept in `ChainSync` for now because it shares the same set
of peers as block syncing implementation in `ChainSync`. Extraction of a
common context responsible for peer management in syncing strategies
able to run in parallel is planned for a follow-up PR.

## Further improvements
A possibility of conversion of `SyncingStartegy` into a trait should be
evaluated. The main stopper for this is that different strategies need
to communicate different actions to `SyncingEngine` and respond to
different events / provide different APIs (e.g., requesting
justifications is only possible via `ChainSync` and not through
`WarpSync`; `SendWarpProofRequest` action is only relevant to
`WarpSync`, etc.)

---------

Co-authored-by: Aaro Altonen <[email protected]>

polkadot/node/service/src/lib.rs

@@ -734,7 +734,7 @@ pub fn new_full<OverseerGenerator: OverseerGen>(
 	}: NewFullParams<OverseerGenerator>,
 ) -> Result<NewFull, Error> {
 	use polkadot_node_network_protocol::request_response::IncomingRequest;
-	use sc_network_sync::warp::WarpSyncParams;
+	use sc_network_sync::WarpSyncParams;
 
 	let is_offchain_indexing_enabled = config.offchain_worker.indexing_enabled;
 	let role = config.role.clone();

prdoc/pr_2467.prdoc

@@ -0,0 +1,15 @@
+# Schema: Polkadot SDK PRDoc Schema (prdoc) v1.0.0
+# See doc at https://raw.githubusercontent.com/paritytech/polkadot-sdk/master/prdoc/schema_user.json
+
+title: Extract warp sync strategy from `ChainSync`
+
+doc:
+  - audience: Node Dev
+    description: |
+      `WarpSync`, and `StateSync` as the logical part of warp sync, are extracted from  `ChainSync`
+      as independent syncing strategies. `SyncingStrategy` enum is introduced as a proxy between
+      `SyncingEngine` and specific strategies. `SyncingStrategy` may be replaced by a trait in a
+      follow-up PRs.
+
+crates: 
+  - name: sc-network-sync

substrate/bin/node/cli/src/service.rs

@@ -30,7 +30,7 @@ use node_primitives::Block;
 use sc_client_api::{Backend, BlockBackend};
 use sc_consensus_babe::{self, SlotProportion};
 use sc_network::{event::Event, NetworkEventStream, NetworkService};
-use sc_network_sync::{warp::WarpSyncParams, SyncingService};
+use sc_network_sync::{strategy::warp::WarpSyncParams, SyncingService};
 use sc_service::{config::Configuration, error::Error as ServiceError, RpcHandlers, TaskManager};
 use sc_statement_store::Store as StatementStore;
 use sc_telemetry::{Telemetry, TelemetryWorker};

substrate/client/consensus/grandpa/src/warp_proof.rs

@@ -23,7 +23,7 @@ use crate::{
 	BlockNumberOps, GrandpaJustification, SharedAuthoritySet,
 };
 use sc_client_api::Backend as ClientBackend;
-use sc_network_sync::warp::{EncodedProof, VerificationResult, WarpSyncProvider};
+use sc_network_sync::strategy::warp::{EncodedProof, VerificationResult, WarpSyncProvider};
 use sp_blockchain::{Backend as BlockchainBackend, HeaderBackend};
 use sp_consensus_grandpa::{AuthorityList, SetId, GRANDPA_ENGINE_ID};
 use sp_runtime::{

substrate/client/informant/src/display.rs

@@ -21,10 +21,7 @@ use ansi_term::Colour;
 use log::info;
 use sc_client_api::ClientInfo;
 use sc_network::NetworkStatus;
-use sc_network_sync::{
-	warp::{WarpSyncPhase, WarpSyncProgress},
-	SyncState, SyncStatus,
-};
+use sc_network_sync::{SyncState, SyncStatus, WarpSyncPhase, WarpSyncProgress};
 use sp_runtime::traits::{Block as BlockT, CheckedDiv, NumberFor, Saturating, Zero};
 use std::{fmt, time::Instant};
 
@@ -130,9 +127,10 @@ impl<B: BlockT> InformantDisplay<B> {
 					),
 				(_, Some(state), _) => (
 					"⚙️ ",
-					"Downloading state".into(),
+					"State sync".into(),
 					format!(
-						", {}%, {:.2} Mib",
+						", {}, {}%, {:.2} Mib",
+						state.phase,
 						state.percentage,
 						(state.size as f32) / (1024f32 * 1024f32)
 					),

substrate/client/network/sync/Cargo.toml

@@ -49,7 +49,7 @@ sp-consensus-grandpa = { path = "../../../primitives/consensus/grandpa" }
 sp-runtime = { path = "../../../primitives/runtime" }
 
 [dev-dependencies]
-tokio = { version = "1.22.0", features = ["macros"] }
+mockall = "0.11.3"
 quickcheck = { version = "1.0.3", default-features = false }
 sc-block-builder = { path = "../../block-builder" }
 sp-test-primitives = { path = "../../../primitives/test-primitives" }

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

@@ -19,7 +19,7 @@
 //! [`BlockAnnounceValidator`] is responsible for async validation of block announcements.
 //! [`Stream`] implemented by [`BlockAnnounceValidator`] never terminates.
 
-use crate::futures_stream::FuturesStream;
+use crate::{futures_stream::FuturesStream, LOG_TARGET};
 use futures::{stream::FusedStream, Future, FutureExt, Stream, StreamExt};
 use libp2p::PeerId;
 use log::{debug, error, trace, warn};
@@ -33,9 +33,6 @@ use std::{
 	task::{Context, Poll},
 };
 
-/// Log target for this file.
-const LOG_TARGET: &str = "sync";
-
 /// Maximum number of concurrent block announce validations.
 ///
 /// If the queue reaches the maximum, we drop any new block

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

@@ -24,6 +24,7 @@ use crate::{
 		BlockResponse as BlockResponseSchema, BlockResponse, Direction,
 	},
 	service::network::NetworkServiceHandle,
+	LOG_TARGET,
 };
 
 use codec::{Decode, DecodeAll, Encode};
@@ -56,7 +57,6 @@ use std::{
 /// Maximum blocks per response.
 pub(crate) const MAX_BLOCKS_IN_RESPONSE: usize = 128;
 
-const LOG_TARGET: &str = "sync";
 const MAX_BODY_BYTES: usize = 8 * 1024 * 1024;
 const MAX_NUMBER_OF_SAME_REQUESTS_PER_PEER: usize = 2;
 

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

@@ -16,6 +16,7 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
+use crate::LOG_TARGET;
 use libp2p::PeerId;
 use log::trace;
 use sc_network_common::sync::message;
@@ -87,10 +88,10 @@ impl<B: BlockT> BlockCollection<B> {
 
 		match self.blocks.get(&start) {
 			Some(&BlockRangeState::Downloading { .. }) => {
-				trace!(target: "sync", "Inserting block data still marked as being downloaded: {}", start);
+				trace!(target: LOG_TARGET, "Inserting block data still marked as being downloaded: {}", start);
 			},
 			Some(BlockRangeState::Complete(existing)) if existing.len() >= blocks.len() => {
-				trace!(target: "sync", "Ignored block data already downloaded: {}", start);
+				trace!(target: LOG_TARGET, "Ignored block data already downloaded: {}", start);
 				return
 			},
 			_ => (),
@@ -162,7 +163,7 @@ impl<B: BlockT> BlockCollection<B> {
 		};
 		// crop to peers best
 		if range.start > peer_best {
-			trace!(target: "sync", "Out of range for peer {} ({} vs {})", who, range.start, peer_best);
+			trace!(target: LOG_TARGET, "Out of range for peer {} ({} vs {})", who, range.start, peer_best);
 			return None
 		}
 		range.end = cmp::min(peer_best + One::one(), range.end);
@@ -173,7 +174,7 @@ impl<B: BlockT> BlockCollection<B> {
 			.next()
 			.map_or(false, |(n, _)| range.start > *n + max_ahead.into())
 		{
-			trace!(target: "sync", "Too far ahead for peer {} ({})", who, range.start);
+			trace!(target: LOG_TARGET, "Too far ahead for peer {} ({})", who, range.start);
 			return None
 		}
 
@@ -224,7 +225,7 @@ impl<B: BlockT> BlockCollection<B> {
 			};
 			*range_data = BlockRangeState::Queued { len };
 		}
-		trace!(target: "sync", "{} blocks ready for import", ready.len());
+		trace!(target: LOG_TARGET, "{} blocks ready for import", ready.len());
 		ready
 	}
 
@@ -235,7 +236,7 @@ impl<B: BlockT> BlockCollection<B> {
 				self.blocks.remove(&block_num);
 				block_num += One::one();
 			}
-			trace!(target: "sync", "Cleared blocks from {:?} to {:?}", from, to);
+			trace!(target: LOG_TARGET, "Cleared blocks from {:?} to {:?}", from, to);
 		}
 	}