Fast sync (#8884)

* State sync

* Importing state fixes

* Bugfixes

* Sync with proof

* Status reporting

* Unsafe sync mode

* Sync test

* Cleanup

* Apply suggestions from code review

Co-authored-by: cheme <emericchevalier.pro@gmail.com>
Co-authored-by: Pierre Krieger <pierre.krieger1708@gmail.com>

* set_genesis_storage

* Extract keys from range proof

* Detect iter completion

* Download and import bodies with fast sync

* Replaced meta updates tuple with a struct

* Fixed reverting finalized state

* Reverted timeout

* Typo

* Doc

* Doc

* Fixed light client test

* Fixed error handling

* Tweaks

* More UpdateMeta changes

* Rename convert_transaction

* Apply suggestions from code review

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Apply suggestions from code review

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Code review suggestions

* Fixed count handling

Co-authored-by: cheme <emericchevalier.pro@gmail.com>
Co-authored-by: Pierre Krieger <pierre.krieger1708@gmail.com>
Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

substrate/client/network/src/behaviour.rs

@@ -79,6 +79,11 @@ pub struct Behaviour<B: BlockT> {
 	/// [`request_responses::RequestResponsesBehaviour`].
 	#[behaviour(ignore)]
 	block_request_protocol_name: String,
+
+	/// Protocol name used to send out state requests via
+	/// [`request_responses::RequestResponsesBehaviour`].
+	#[behaviour(ignore)]
+	state_request_protocol_name: String,
 }
 
 /// Event generated by `Behaviour`.
@@ -186,6 +191,7 @@ impl<B: BlockT> Behaviour<B> {
 		light_client_request_sender: light_client_requests::sender::LightClientRequestSender<B>,
 		disco_config: DiscoveryConfig,
 		block_request_protocol_config: request_responses::ProtocolConfig,
+		state_request_protocol_config: request_responses::ProtocolConfig,
 		bitswap: Option<Bitswap<B>>,
 		light_client_request_protocol_config: request_responses::ProtocolConfig,
 		// All remaining request protocol configs.
@@ -193,7 +199,9 @@ impl<B: BlockT> Behaviour<B> {
 	) -> Result<Self, request_responses::RegisterError> {
 		// Extract protocol name and add to `request_response_protocols`.
 		let block_request_protocol_name = block_request_protocol_config.name.to_string();
+		let state_request_protocol_name = state_request_protocol_config.name.to_string();
 		request_response_protocols.push(block_request_protocol_config);
+		request_response_protocols.push(state_request_protocol_config);
 
 		request_response_protocols.push(light_client_request_protocol_config);
 
@@ -206,8 +214,8 @@ impl<B: BlockT> Behaviour<B> {
 				request_responses::RequestResponsesBehaviour::new(request_response_protocols.into_iter())?,
 			light_client_request_sender,
 			events: VecDeque::new(),
-
 			block_request_protocol_name,
+			state_request_protocol_name,
 		})
 	}
 
@@ -329,6 +337,21 @@ Behaviour<B> {
 					&target, &self.block_request_protocol_name, buf, pending_response, IfDisconnected::ImmediateError,
 				);
 			},
+			CustomMessageOutcome::StateRequest { target, request, pending_response } => {
+				let mut buf = Vec::with_capacity(request.encoded_len());
+				if let Err(err) = request.encode(&mut buf) {
+					log::warn!(
+						target: "sync",
+						"Failed to encode state request {:?}: {:?}",
+						request, err
+					);
+					return
+				}
+
+				self.request_responses.send_request(
+					&target, &self.state_request_protocol_name, buf, pending_response, IfDisconnected::ImmediateError,
+				);
+			},
 			CustomMessageOutcome::NotificationStreamOpened {
 				remote, protocol, negotiated_fallback, roles, notifications_sink
 			} => {

substrate/client/network/src/chain.rs

@@ -21,6 +21,7 @@
 use sp_blockchain::{Error, HeaderBackend, HeaderMetadata};
 use sc_client_api::{BlockBackend, ProofProvider};
 use sp_runtime::traits::{Block as BlockT, BlockIdTo};
+pub use sc_client_api::{StorageKey, StorageData, ImportedState};
 
 /// Local client abstraction for the network.
 pub trait Client<Block: BlockT>: HeaderBackend<Block> + ProofProvider<Block> + BlockIdTo<Block, Error = Error>

substrate/client/network/src/config.rs

@@ -123,6 +123,15 @@ pub struct Params<B: BlockT, H: ExHashT> {
 	/// [`crate::light_client_requests::handler::LightClientRequestHandler::new`] allowing
 	/// both outgoing and incoming requests.
 	pub light_client_request_protocol_config: RequestResponseConfig,
+
+	/// Request response configuration for the state request protocol.
+	///
+	/// Can be constructed either via
+	/// [`crate::state_requests::generate_protocol_config`] allowing outgoing but not
+	/// incoming requests, or constructed via
+	/// [`crate::state_requests::handler::StateRequestHandler::new`] allowing
+	/// both outgoing and incoming requests.
+	pub state_request_protocol_config: RequestResponseConfig,
 }
 
 /// Role of the local node.
@@ -373,6 +382,24 @@ impl From<multiaddr::Error> for ParseErr {
 	}
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+/// Sync operation mode.
+pub enum SyncMode {
+	/// Full block download and verification.
+	Full,
+	/// Download blocks and the latest state.
+	Fast {
+		/// Skip state proof download and verification.
+		skip_proofs: bool
+	},
+}
+
+impl Default for SyncMode {
+	fn default() -> Self {
+		SyncMode::Full
+	}
+}
+
 /// Network service configuration.
 #[derive(Clone, Debug)]
 pub struct NetworkConfiguration {
@@ -400,6 +427,8 @@ pub struct NetworkConfiguration {
 	pub transport: TransportConfig,
 	/// Maximum number of peers to ask the same blocks in parallel.
 	pub max_parallel_downloads: u32,
+	/// Initial syncing mode.
+	pub sync_mode: SyncMode,
 
 	/// True if Kademlia random discovery should be enabled.
 	///
@@ -462,6 +491,7 @@ impl NetworkConfiguration {
 				wasm_external_transport: None,
 			},
 			max_parallel_downloads: 5,
+			sync_mode: SyncMode::Full,
 			enable_dht_random_walk: true,
 			allow_non_globals_in_dht: false,
 			kademlia_disjoint_query_paths: false,

substrate/client/network/src/gossip/tests.rs

@@ -17,6 +17,7 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 use crate::block_request_handler::BlockRequestHandler;
+use crate::state_request_handler::StateRequestHandler;
 use crate::light_client_requests::handler::LightClientRequestHandler;
 use crate::gossip::QueuedSender;
 use crate::{config,  Event, NetworkService, NetworkWorker};
@@ -107,6 +108,16 @@ fn build_test_full_node(network_config: config::NetworkConfiguration)
 		protocol_config
 	};
 
+	let state_request_protocol_config = {
+		let (handler, protocol_config) = StateRequestHandler::new(
+			&protocol_id,
+			client.clone(),
+			50,
+		);
+		async_std::task::spawn(handler.run().boxed());
+		protocol_config
+	};
+
 	let light_client_request_protocol_config = {
 		let (handler, protocol_config) = LightClientRequestHandler::new(
 			&protocol_id,
@@ -131,6 +142,7 @@ fn build_test_full_node(network_config: config::NetworkConfiguration)
 		),
 		metrics_registry: None,
 		block_request_protocol_config,
+		state_request_protocol_config,
 		light_client_request_protocol_config,
 	})
 	.unwrap();

substrate/client/network/src/lib.rs

@@ -260,6 +260,7 @@ mod utils;
 pub mod block_request_handler;
 pub mod bitswap;
 pub mod light_client_requests;
+pub mod state_request_handler;
 pub mod config;
 pub mod error;
 pub mod gossip;
@@ -268,7 +269,8 @@ pub mod transactions;
 
 #[doc(inline)]
 pub use libp2p::{multiaddr, Multiaddr, PeerId};
-pub use protocol::{event::{DhtEvent, Event, ObservedRole}, sync::SyncState, PeerInfo};
+pub use protocol::{event::{DhtEvent, Event, ObservedRole}, PeerInfo};
+pub use protocol::sync::{SyncState, StateDownloadProgress};
 pub use service::{
 	NetworkService, NetworkWorker, RequestFailure, OutboundFailure, NotificationSender,
 	NotificationSenderReady, IfDisconnected,
@@ -321,4 +323,6 @@ pub struct NetworkStatus<B: BlockT> {
 	pub total_bytes_inbound: u64,
 	/// The total number of bytes sent.
 	pub total_bytes_outbound: u64,
+	/// State sync in progress.
+	pub state_sync: Option<protocol::sync::StateDownloadProgress>,
 }

substrate/client/network/src/protocol.rs

@@ -22,6 +22,7 @@ use crate::{
 	error,
 	request_responses::RequestFailure,
 	utils::{interval, LruHashSet},
+	schema::v1::StateResponse,
 };
 
 use bytes::Bytes;
@@ -49,7 +50,7 @@ use sp_runtime::{
 	traits::{Block as BlockT, Header as HeaderT, NumberFor, Zero, CheckedSub},
 };
 use sp_arithmetic::traits::SaturatedConversion;
-use sync::{ChainSync, SyncState};
+use sync::{ChainSync, Status as SyncStatus};
 use std::borrow::Cow;
 use std::convert::TryFrom as _;
 use std::collections::{HashMap, HashSet, VecDeque};
@@ -179,13 +180,19 @@ pub struct Protocol<B: BlockT> {
 	block_announce_data_cache: lru::LruCache<B::Hash, Vec<u8>>,
 }
 
+#[derive(Debug)]
+enum PeerRequest<B: BlockT> {
+	Block(message::BlockRequest<B>),
+	State,
+}
+
 /// Peer information
 #[derive(Debug)]
 struct Peer<B: BlockT> {
 	info: PeerInfo<B>,
-	/// Current block request, if any. Started by emitting [`CustomMessageOutcome::BlockRequest`].
-	block_request: Option<(
-		message::BlockRequest<B>,
+	/// Current request, if any. Started by emitting [`CustomMessageOutcome::BlockRequest`].
+	request: Option<(
+		PeerRequest<B>,
 		oneshot::Receiver<Result<Vec<u8>, RequestFailure>>,
 	)>,
 	/// Holds a set of blocks known to this peer.
@@ -210,6 +217,21 @@ pub struct ProtocolConfig {
 	pub roles: Roles,
 	/// Maximum number of peers to ask the same blocks in parallel.
 	pub max_parallel_downloads: u32,
+	/// Enable state sync.
+	pub sync_mode: config::SyncMode,
+}
+
+impl ProtocolConfig {
+	fn sync_mode(&self) -> sync::SyncMode {
+		if self.roles.is_light() {
+			sync::SyncMode::Light
+		} else {
+			match self.sync_mode {
+				config::SyncMode::Full => sync::SyncMode::Full,
+				config::SyncMode::Fast { skip_proofs } => sync::SyncMode::LightState { skip_proofs },
+			}
+		}
+	}
 }
 
 impl Default for ProtocolConfig {
@@ -217,6 +239,7 @@ impl Default for ProtocolConfig {
 		ProtocolConfig {
 			roles: Roles::FULL,
 			max_parallel_downloads: 5,
+			sync_mode: config::SyncMode::Full,
 		}
 	}
 }
@@ -263,12 +286,11 @@ impl<B: BlockT> Protocol<B> {
 	) -> error::Result<(Protocol<B>, sc_peerset::PeersetHandle, Vec<(PeerId, Multiaddr)>)> {
 		let info = chain.info();
 		let sync = ChainSync::new(
-			config.roles,
+			config.sync_mode(),
 			chain.clone(),
-			&info,
 			block_announce_validator,
 			config.max_parallel_downloads,
-		);
+		).map_err(Box::new)?;
 
 		let boot_node_ids = {
 			let mut list = HashSet::new();
@@ -454,13 +476,13 @@ impl<B: BlockT> Protocol<B> {
 	pub fn num_active_peers(&self) -> usize {
 		self.peers
 			.values()
-			.filter(|p| p.block_request.is_some())
+			.filter(|p| p.request.is_some())
 			.count()
 	}
 
 	/// Current global sync state.
-	pub fn sync_state(&self) -> SyncState {
-		self.sync.status().state
+	pub fn sync_state(&self) -> SyncStatus<B> {
+		self.sync.status()
 	}
 
 	/// Target sync block number.
@@ -656,6 +678,27 @@ impl<B: BlockT> Protocol<B> {
 		}
 	}
 
+	/// Must be called in response to a [`CustomMessageOutcome::StateRequest`] being emitted.
+	/// Must contain the same `PeerId` and request that have been emitted.
+	pub fn on_state_response(
+		&mut self,
+		peer_id: PeerId,
+		response: StateResponse,
+	) -> CustomMessageOutcome<B> {
+		match self.sync.on_state_data(&peer_id, response) {
+			Ok(sync::OnStateData::Import(origin, block)) =>
+				CustomMessageOutcome::BlockImport(origin, vec![block]),
+			Ok(sync::OnStateData::Request(peer, req)) => {
+				prepare_state_request::<B>(&mut self.peers, peer, req)
+			}
+			Err(sync::BadPeer(id, repu)) => {
+				self.behaviour.disconnect_peer(&id, HARDCODED_PEERSETS_SYNC);
+				self.peerset_handle.report_peer(id, repu);
+				CustomMessageOutcome::None
+			}
+		}
+	}
+
 	/// Perform time based maintenance.
 	///
 	/// > **Note**: This method normally doesn't have to be called except for testing purposes.
@@ -736,7 +779,7 @@ impl<B: BlockT> Protocol<B> {
 				best_hash: status.best_hash,
 				best_number: status.best_number
 			},
-			block_request: None,
+			request: None,
 			known_blocks: LruHashSet::new(NonZeroUsize::new(MAX_KNOWN_BLOCKS)
 				.expect("Constant is nonzero")),
 		};
@@ -1137,7 +1180,7 @@ fn prepare_block_request<B: BlockT>(
 	let (tx, rx) = oneshot::channel();
 
 	if let Some(ref mut peer) = peers.get_mut(&who) {
-		peer.block_request = Some((request.clone(), rx));
+		peer.request = Some((PeerRequest::Block(request.clone()), rx));
 	}
 
 	let request = crate::schema::v1::BlockRequest {
@@ -1161,6 +1204,23 @@ fn prepare_block_request<B: BlockT>(
 	}
 }
 
+fn prepare_state_request<B: BlockT>(
+	peers: &mut HashMap<PeerId, Peer<B>>,
+	who: PeerId,
+	request: crate::schema::v1::StateRequest,
+) -> CustomMessageOutcome<B> {
+	let (tx, rx) = oneshot::channel();
+
+	if let Some(ref mut peer) = peers.get_mut(&who) {
+		peer.request = Some((PeerRequest::State, rx));
+	}
+	CustomMessageOutcome::StateRequest {
+		target: who,
+		request: request,
+		pending_response: tx,
+	}
+}
+
 /// Outcome of an incoming custom message.
 #[derive(Debug)]
 #[must_use]
@@ -1192,6 +1252,12 @@ pub enum CustomMessageOutcome<B: BlockT> {
 		request: crate::schema::v1::BlockRequest,
 		pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
 	},
+	/// A new storage request must be emitted.
+	StateRequest {
+		target: PeerId,
+		request: crate::schema::v1::StateRequest,
+		pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
+	},
 	/// Peer has a reported a new head of chain.
 	PeerNewBest(PeerId, NumberFor<B>),
 	/// Now connected to a new peer for syncing purposes.
@@ -1254,27 +1320,54 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
 
 		// Check for finished outgoing requests.
 		let mut finished_block_requests = Vec::new();
+		let mut finished_state_requests = Vec::new();
 		for (id, peer) in self.peers.iter_mut() {
-			if let Peer { block_request: Some((_, pending_response)), .. } = peer {
+			if let Peer { request: Some((_, pending_response)), .. } = peer {
 				match pending_response.poll_unpin(cx) {
 					Poll::Ready(Ok(Ok(resp))) => {
-						let (req, _) = peer.block_request.take().unwrap();
+						let (req, _) = peer.request.take().unwrap();
+						match req {
+							PeerRequest::Block(req) => {
+								let protobuf_response = match crate::schema::v1::BlockResponse::decode(&resp[..]) {
+									Ok(proto) => proto,
+									Err(e) => {
+										debug!(
+											target: "sync",
+											"Failed to decode block response from peer {:?}: {:?}.",
+											id,
+											e
+										);
+										self.peerset_handle.report_peer(id.clone(), rep::BAD_MESSAGE);
+										self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
+										continue;
+									}
+								};
 
-						let protobuf_response = match crate::schema::v1::BlockResponse::decode(&resp[..]) {
-							Ok(proto) => proto,
-							Err(e) => {
-								debug!(target: "sync", "Failed to decode block request to peer {:?}: {:?}.", id, e);
-								self.peerset_handle.report_peer(id.clone(), rep::BAD_MESSAGE);
-								self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
-								continue;
-							}
-						};
+								finished_block_requests.push((id.clone(), req, protobuf_response));
+							},
+							PeerRequest::State => {
+								let protobuf_response = match crate::schema::v1::StateResponse::decode(&resp[..]) {
+									Ok(proto) => proto,
+									Err(e) => {
+										debug!(
+											target: "sync",
+											"Failed to decode state response from peer {:?}: {:?}.",
+											id,
+											e
+										);
+										self.peerset_handle.report_peer(id.clone(), rep::BAD_MESSAGE);
+										self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
+										continue;
+									}
+								};
 
-						finished_block_requests.push((id.clone(), req, protobuf_response));
+								finished_state_requests.push((id.clone(), protobuf_response));
+							},
+						}
 					},
 					Poll::Ready(Ok(Err(e))) => {
-						peer.block_request.take();
-						debug!(target: "sync", "Block request to peer {:?} failed: {:?}.", id, e);
+						peer.request.take();
+						debug!(target: "sync", "Request to peer {:?} failed: {:?}.", id, e);
 
 						match e {
 							RequestFailure::Network(OutboundFailure::Timeout) => {
@@ -1309,10 +1402,10 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
 						}
 					},
 					Poll::Ready(Err(oneshot::Canceled)) => {
-						peer.block_request.take();
+						peer.request.take();
 						trace!(
 							target: "sync",
-							"Block request to peer {:?} failed due to oneshot being canceled.",
+							"Request to peer {:?} failed due to oneshot being canceled.",
 							id,
 						);
 						self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
@@ -1325,6 +1418,10 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
 			let ev = self.on_block_response(id, req, protobuf_response);
 			self.pending_messages.push_back(ev);
 		}
+		for (id, protobuf_response) in finished_state_requests {
+			let ev = self.on_state_response(id, protobuf_response);
+			self.pending_messages.push_back(ev);
+		}
 
 		while let Poll::Ready(Some(())) = self.tick_timeout.poll_next_unpin(cx) {
 			self.tick();
@@ -1334,6 +1431,10 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
 			let event = prepare_block_request(&mut self.peers, id.clone(), request);
 			self.pending_messages.push_back(event);
 		}
+		if let Some((id, request)) = self.sync.state_request() {
+			let event = prepare_state_request(&mut self.peers, id, request);
+			self.pending_messages.push_back(event);
+		}
 		for (id, request) in self.sync.justification_requests() {
 			let event = prepare_block_request(&mut self.peers, id, request);
 			self.pending_messages.push_back(event);

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

@@ -31,14 +31,16 @@
 
 use codec::Encode;
 use blocks::BlockCollection;
-use sp_blockchain::{Error as ClientError, Info as BlockchainInfo, HeaderMetadata};
+use state::StateSync;
+use sp_blockchain::{Error as ClientError, HeaderMetadata};
 use sp_consensus::{BlockOrigin, BlockStatus,
 	block_validation::{BlockAnnounceValidator, Validation},
 	import_queue::{IncomingBlock, BlockImportResult, BlockImportError}
 };
 use crate::protocol::message::{
-	self, BlockAnnounce, BlockAttributes, BlockRequest, BlockResponse, Roles,
+	self, BlockAnnounce, BlockAttributes, BlockRequest, BlockResponse,
 };
+use crate::schema::v1::{StateResponse, StateRequest};
 use either::Either;
 use extra_requests::ExtraRequests;
 use libp2p::PeerId;
@@ -59,6 +61,7 @@ use futures::{task::Poll, Future, stream::FuturesUnordered, FutureExt, StreamExt
 
 mod blocks;
 mod extra_requests;
+mod state;
 
 /// Maximum blocks to request in a single packet.
 const MAX_BLOCKS_TO_REQUEST: usize = 128;
@@ -84,6 +87,9 @@ const MAX_CONCURRENT_BLOCK_ANNOUNCE_VALIDATIONS: usize = 256;
 /// See [`MAX_CONCURRENT_BLOCK_ANNOUNCE_VALIDATIONS`] for more information.
 const MAX_CONCURRENT_BLOCK_ANNOUNCE_VALIDATIONS_PER_PEER: usize = 4;
 
+/// Pick the state to sync as the latest finalized number minus this.
+const STATE_SYNC_FINALITY_THRESHOLD: u32 = 8;
+
 /// We use a heuristic that with a high likelihood, by the time
 /// `MAJOR_SYNC_BLOCKS` have been imported we'll be on the same
 /// chain as (or at least closer to) the peer so we want to delay
@@ -183,11 +189,8 @@ pub struct ChainSync<B: BlockT> {
 	best_queued_number: NumberFor<B>,
 	/// The best block hash in our queue of blocks to import
 	best_queued_hash: B::Hash,
-	/// The role of this node, e.g. light or full
-	role: Roles,
-	/// What block attributes we require for this node, usually derived from
-	/// what role we are, but could be customized
-	required_block_attributes: message::BlockAttributes,
+	/// Current mode (full/light)
+	mode: SyncMode,
 	/// Any extra justification requests.
 	extra_justifications: ExtraRequests<B>,
 	/// A set of hashes of blocks that are being downloaded or have been
@@ -209,6 +212,11 @@ pub struct ChainSync<B: BlockT> {
 	>,
 	/// Stats per peer about the number of concurrent block announce validations.
 	block_announce_validation_per_peer_stats: HashMap<PeerId, usize>,
+	/// State sync in progress, if any.
+	state_sync: Option<StateSync<B>>,
+	/// Enable importing existing blocks. This is used used after the state download to
+	/// catch up to the latest state while re-importing blocks.
+	import_existing: bool,
 }
 
 /// All the data we have about a Peer that we are trying to sync with
@@ -281,6 +289,8 @@ pub enum PeerSyncState<B: BlockT> {
 	DownloadingStale(B::Hash),
 	/// Downloading justification for given block hash.
 	DownloadingJustification(B::Hash),
+	/// Downloading state.
+	DownloadingState,
 }
 
 impl<B: BlockT> PeerSyncState<B> {
@@ -298,6 +308,15 @@ pub enum SyncState {
 	Downloading
 }
 
+/// Reported state download progress.
+#[derive(Clone, Eq, PartialEq, Debug)]
+pub struct StateDownloadProgress {
+	/// Estimated download percentage.
+	pub percentage: u32,
+	/// Total state size in bytes downloaded so far.
+	pub size: u64,
+}
+
 /// Syncing status and statistics.
 #[derive(Clone)]
 pub struct Status<B: BlockT> {
@@ -309,6 +328,8 @@ pub struct Status<B: BlockT> {
 	pub num_peers: u32,
 	/// Number of blocks queued for import
 	pub queued_blocks: u32,
+	/// State sync status in progress, if any.
+	pub state_sync: Option<StateDownloadProgress>,
 }
 
 /// A peer did not behave as expected and should be reported.
@@ -344,6 +365,15 @@ impl<B: BlockT> OnBlockData<B> {
 	}
 }
 
+/// Result of [`ChainSync::on_state_data`].
+#[derive(Debug)]
+pub enum OnStateData<B: BlockT> {
+	/// The block and state that should be imported.
+	Import(BlockOrigin, IncomingBlock<B>),
+	/// A new state request needs to be made to the given peer.
+	Request(PeerId, StateRequest)
+}
+
 /// Result of [`ChainSync::poll_block_announce_validation`].
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum PollBlockAnnounceValidation<H> {
@@ -429,6 +459,20 @@ pub enum OnBlockJustification<B: BlockT> {
 	}
 }
 
+
+/// Operation mode.
+#[derive(Debug, PartialEq, Eq)]
+pub enum SyncMode {
+	// Sync headers only
+	Light,
+	// Sync headers and block bodies
+	Full,
+	// Sync headers and the last finalied state
+	LightState {
+		skip_proofs: bool
+	},
+}
+
 /// Result of [`ChainSync::has_slot_for_block_announce_validation`].
 enum HasSlotForBlockAnnounceValidation {
 	/// Yes, there is a slot for the block announce validation.
@@ -442,27 +486,19 @@ enum HasSlotForBlockAnnounceValidation {
 impl<B: BlockT> ChainSync<B> {
 	/// Create a new instance.
 	pub fn new(
-		role: Roles,
+		mode: SyncMode,
 		client: Arc<dyn crate::chain::Client<B>>,
-		info: &BlockchainInfo<B>,
 		block_announce_validator: Box<dyn BlockAnnounceValidator<B> + Send>,
 		max_parallel_downloads: u32,
-	) -> Self {
-		let mut required_block_attributes = BlockAttributes::HEADER | BlockAttributes::JUSTIFICATION;
-
-		if role.is_full() {
-			required_block_attributes |= BlockAttributes::BODY
-		}
-
-		ChainSync {
+	) -> Result<Self, ClientError> {
+		let mut sync = ChainSync {
 			client,
 			peers: HashMap::new(),
 			blocks: BlockCollection::new(),
-			best_queued_hash: info.best_hash,
-			best_queued_number: info.best_number,
+			best_queued_hash: Default::default(),
+			best_queued_number: Zero::zero(),
 			extra_justifications: ExtraRequests::new("justification"),
-			role,
-			required_block_attributes,
+			mode,
 			queue_blocks: Default::default(),
 			fork_targets: Default::default(),
 			pending_requests: Default::default(),
@@ -471,6 +507,27 @@ impl<B: BlockT> ChainSync<B> {
 			downloaded_blocks: 0,
 			block_announce_validation: Default::default(),
 			block_announce_validation_per_peer_stats: Default::default(),
+			state_sync: None,
+			import_existing: false,
+		};
+		sync.reset_sync_start_point()?;
+		Ok(sync)
+	}
+
+	fn required_block_attributes(&self) -> BlockAttributes {
+		match self.mode {
+			SyncMode::Full => BlockAttributes::HEADER | BlockAttributes::JUSTIFICATION | BlockAttributes::BODY,
+			SyncMode::Light => BlockAttributes::HEADER | BlockAttributes::JUSTIFICATION,
+			SyncMode::LightState { .. } =>
+				BlockAttributes::HEADER | BlockAttributes::JUSTIFICATION | BlockAttributes::BODY,
+		}
+	}
+
+	fn skip_execution(&self) -> bool {
+		match self.mode {
+			SyncMode::Full => false,
+			SyncMode::Light => true,
+			SyncMode::LightState { .. } => true,
 		}
 	}
 
@@ -502,6 +559,7 @@ impl<B: BlockT> ChainSync<B> {
 			best_seen_block: best_seen,
 			num_peers: self.peers.len() as u32,
 			queued_blocks: self.queue_blocks.len() as u32,
+			state_sync: self.state_sync.as_ref().map(|s| s.progress()),
 		}
 	}
 
@@ -607,7 +665,7 @@ impl<B: BlockT> ChainSync<B> {
 				);
 				self.peers.insert(who.clone(), PeerSync {
 					peer_id: who.clone(),
-					common_number: best_number,
+					common_number: std::cmp::min(self.best_queued_number, best_number),
 					best_hash,
 					best_number,
 					state: PeerSyncState::Available,
@@ -718,7 +776,7 @@ impl<B: BlockT> ChainSync<B> {
 
 	/// Get an iterator over all block requests of all peers.
 	pub fn block_requests(&mut self) -> impl Iterator<Item = (&PeerId, BlockRequest<B>)> + '_ {
-		if self.pending_requests.is_empty() {
+		if self.pending_requests.is_empty() || self.state_sync.is_some() {
 			return Either::Left(std::iter::empty())
 		}
 		if self.queue_blocks.len() > MAX_IMPORTING_BLOCKS {
@@ -726,10 +784,10 @@ impl<B: BlockT> ChainSync<B> {
 			return Either::Left(std::iter::empty())
 		}
 		let major_sync = self.status().state == SyncState::Downloading;
+		let attrs = self.required_block_attributes();
 		let blocks = &mut self.blocks;
-		let attrs = &self.required_block_attributes;
 		let fork_targets = &mut self.fork_targets;
-		let last_finalized = self.client.info().finalized_number;
+		let last_finalized = std::cmp::min(self.best_queued_number, self.client.info().finalized_number);
 		let best_queued = self.best_queued_number;
 		let client = &self.client;
 		let queue = &self.queue_blocks;
@@ -804,6 +862,28 @@ impl<B: BlockT> ChainSync<B> {
 		Either::Right(iter)
 	}
 
+	/// Get a state request, if any
+	pub fn state_request(&mut self) -> Option<(PeerId, StateRequest)> {
+		if let Some(sync) = &self.state_sync {
+			if sync.is_complete() {
+				return None;
+			}
+			if self.peers.iter().any(|(_, peer)| peer.state == PeerSyncState::DownloadingState) {
+				// Only one pending state request is allowed.
+				return None;
+			}
+			for (id, peer) in self.peers.iter_mut() {
+				if peer.state.is_available() && peer.common_number >= sync.target_block_num() {
+					trace!(target: "sync", "New StateRequest for {}", id);
+					peer.state = PeerSyncState::DownloadingState;
+					let request = sync.next_request();
+					return Some((id.clone(), request))
+				}
+			}
+		}
+		None
+	}
+
 	/// Handle a response from the remote to a block request that we made.
 	///
 	/// `request` must be the original request that triggered `response`.
@@ -848,7 +928,9 @@ impl<B: BlockT> ChainSync<B> {
 										justifications,
 										origin: block_data.origin,
 										allow_missing_state: true,
-										import_existing: false,
+										import_existing: self.import_existing,
+										skip_execution: self.skip_execution(),
+										state: None,
 									}
 								}).collect()
 						}
@@ -870,7 +952,9 @@ impl<B: BlockT> ChainSync<B> {
 									justifications,
 									origin: Some(who.clone()),
 									allow_missing_state: true,
-									import_existing: false,
+									import_existing: self.import_existing,
+									skip_execution: self.skip_execution(),
+									state: None,
 								}
 							}).collect()
 						}
@@ -963,10 +1047,11 @@ impl<B: BlockT> ChainSync<B> {
 								peer.state = PeerSyncState::Available;
 								Vec::new()
 							}
-						}
-
-						| PeerSyncState::Available
-						| PeerSyncState::DownloadingJustification(..) => Vec::new()
+						},
+						PeerSyncState::Available
+						| PeerSyncState::DownloadingJustification(..)
+						| PeerSyncState::DownloadingState
+							=> Vec::new()
 					}
 				} else {
 					// When request.is_none() this is a block announcement. Just accept blocks.
@@ -983,6 +1068,8 @@ impl<B: BlockT> ChainSync<B> {
 							origin: Some(who.clone()),
 							allow_missing_state: true,
 							import_existing: false,
+							skip_execution: true,
+							state: None,
 						}
 					}).collect()
 				}
@@ -994,6 +1081,60 @@ impl<B: BlockT> ChainSync<B> {
 		Ok(self.validate_and_queue_blocks(new_blocks))
 	}
 
+	/// Handle a response from the remote to a state request that we made.
+	///
+	/// Returns next request if any.
+	pub fn on_state_data(
+		&mut self,
+		who: &PeerId,
+		response: StateResponse,
+	) -> Result<OnStateData<B>, BadPeer> {
+		let import_result = if let Some(sync) = &mut self.state_sync {
+			debug!(
+				target: "sync",
+				"Importing state data from {} with {} keys, {} proof nodes.",
+				who,
+				response.entries.len(),
+				response.proof.len(),
+			);
+			sync.import(response)
+		} else {
+			debug!(target: "sync", "Ignored obsolete state response from {}", who);
+			return Err(BadPeer(who.clone(), rep::NOT_REQUESTED));
+		};
+
+		match import_result {
+			state::ImportResult::Import(hash, header, state) => {
+				let origin = if self.status().state != SyncState::Downloading {
+					BlockOrigin::NetworkBroadcast
+				} else {
+					BlockOrigin::NetworkInitialSync
+				};
+
+				let block = IncomingBlock {
+					hash,
+					header: Some(header),
+					body: None,
+					justifications: None,
+					origin: None,
+					allow_missing_state: true,
+					import_existing: true,
+					skip_execution: self.skip_execution(),
+					state: Some(state),
+				};
+				debug!(target: "sync", "State sync is complete. Import is queued");
+				Ok(OnStateData::Import(origin, block))
+			}
+			state::ImportResult::Continue(request) => {
+				Ok(OnStateData::Request(who.clone(), request))
+			}
+			state::ImportResult::BadResponse => {
+				debug!(target: "sync", "Bad state data received from {}", who);
+				Err(BadPeer(who.clone(), rep::BAD_BLOCK))
+			}
+		}
+	}
+
 	fn validate_and_queue_blocks(
 		&mut self,
 		mut new_blocks: Vec<IncomingBlock<B>>,
@@ -1048,7 +1189,7 @@ impl<B: BlockT> ChainSync<B> {
 			// We only request one justification at a time
 			let justification = if let Some(block) = response.blocks.into_iter().next() {
 				if hash != block.hash {
-					info!(
+					warn!(
 						target: "sync",
 						"💔 Invalid block justification provided by {}: requested: {:?} got: {:?}", who, hash, block.hash
 					);
@@ -1137,7 +1278,7 @@ impl<B: BlockT> ChainSync<B> {
 
 					if aux.bad_justification {
 						if let Some(ref peer) = who {
-							info!("💔 Sent block with bad justification to import");
+							warn!("💔 Sent block with bad justification to import");
 							output.push(Err(BadPeer(peer.clone(), rep::BAD_JUSTIFICATION)));
 						}
 					}
@@ -1145,6 +1286,17 @@ impl<B: BlockT> ChainSync<B> {
 					if let Some(peer) = who.and_then(|p| self.peers.get_mut(&p)) {
 						peer.update_common_number(number);
 					}
+					let state_sync_complete = self.state_sync.as_ref().map_or(false, |s| s.target() == hash);
+					if state_sync_complete {
+						info!(
+							target: "sync",
+							"State sync is complete ({} MiB), restarting block sync.",
+							self.state_sync.as_ref().map_or(0, |s| s.progress().size / (1024 * 1024)),
+						);
+						self.state_sync = None;
+						self.mode = SyncMode::Full;
+						output.extend(self.restart());
+					}
 				},
 				Err(BlockImportError::IncompleteHeader(who)) => {
 					if let Some(peer) = who {
@@ -1171,7 +1323,7 @@ impl<B: BlockT> ChainSync<B> {
 				},
 				Err(BlockImportError::BadBlock(who)) => {
 					if let Some(peer) = who {
-						info!(
+						warn!(
 							target: "sync",
 							"💔 Block {:?} received from peer {} has been blacklisted",
 							hash,
@@ -1189,6 +1341,7 @@ impl<B: BlockT> ChainSync<B> {
 				e @ Err(BlockImportError::UnknownParent) |
 				e @ Err(BlockImportError::Other(_)) => {
 					warn!(target: "sync", "💔 Error importing block {:?}: {:?}", hash, e);
+					self.state_sync = None;
 					output.extend(self.restart());
 				},
 				Err(BlockImportError::Cancelled) => {}
@@ -1214,6 +1367,29 @@ impl<B: BlockT> ChainSync<B> {
 			is_descendent_of(&**client, base, block)
 		});
 
+		if let SyncMode::LightState { skip_proofs } = &self.mode {
+			if self.state_sync.is_none()
+				&& !self.peers.is_empty()
+				&& self.queue_blocks.is_empty()
+			{
+				// Finalized a recent block.
+				let mut heads: Vec<_> = self.peers.iter().map(|(_, peer)| peer.best_number).collect();
+				heads.sort();
+				let median = heads[heads.len() / 2];
+				if number + STATE_SYNC_FINALITY_THRESHOLD.saturated_into() >= median {
+					if let Ok(Some(header)) = self.client.header(BlockId::hash(hash.clone())) {
+						log::debug!(
+							target: "sync",
+							"Starting state sync for #{} ({})",
+							number,
+							hash,
+						);
+						self.state_sync = Some(StateSync::new(self.client.clone(), header, *skip_proofs));
+					}
+				}
+			}
+		}
+
 		if let Err(err) = r {
 			warn!(
 				target: "sync",
@@ -1536,7 +1712,7 @@ impl<B: BlockT> ChainSync<B> {
 			return PollBlockAnnounceValidation::Nothing { is_best, who, announce }
 		}
 
-		let requires_additional_data = !self.role.is_light() || !known_parent;
+		let requires_additional_data = self.mode != SyncMode::Light || !known_parent;
 		if !requires_additional_data {
 			trace!(
 				target: "sync",
@@ -1595,6 +1771,8 @@ impl<B: BlockT> ChainSync<B> {
 					origin: block_data.origin,
 					allow_missing_state: true,
 					import_existing: false,
+					skip_execution: self.skip_execution(),
+					state: None,
 				}
 			}).collect();
 		if !blocks.is_empty() {
@@ -1611,9 +1789,9 @@ impl<B: BlockT> ChainSync<B> {
 		&'a mut self,
 	) -> impl Iterator<Item = Result<(PeerId, BlockRequest<B>), BadPeer>> + 'a {
 		self.blocks.clear();
-		let info = self.client.info();
-		self.best_queued_hash = info.best_hash;
-		self.best_queued_number = info.best_number;
+		if let Err(e) = self.reset_sync_start_point() {
+			warn!(target: "sync", "💔  Unable to restart sync. :{:?}", e);
+		}
 		self.pending_requests.set_all();
 		debug!(target:"sync", "Restarted with {} ({})", self.best_queued_number, self.best_queued_hash);
 		let old_peers = std::mem::take(&mut self.peers);
@@ -1624,7 +1802,7 @@ impl<B: BlockT> ChainSync<B> {
 			match p.state {
 				PeerSyncState::DownloadingJustification(_) => {
 					// We make sure our commmon number is at least something we have.
-					p.common_number = info.best_number;
+					p.common_number = self.best_queued_number;
 					self.peers.insert(id, p);
 					return None;
 				}
@@ -1640,6 +1818,38 @@ impl<B: BlockT> ChainSync<B> {
 		})
 	}
 
+	/// Find a block to start sync from. If we sync with state, that's the latest block we have state for.
+	fn reset_sync_start_point(&mut self) -> Result<(), ClientError> {
+		let info = self.client.info();
+		if matches!(self.mode, SyncMode::LightState {..}) && info.finalized_state.is_some() {
+			log::warn!(
+				target: "sync",
+				"Can't use fast sync mode with a partially synced database. Reverting to full sync mode."
+			);
+			self.mode = SyncMode::Full;
+		}
+		self.import_existing = false;
+		self.best_queued_hash = info.best_hash;
+		self.best_queued_number = info.best_number;
+		if self.mode == SyncMode::Full {
+			if self.client.block_status(&BlockId::hash(info.best_hash))? != BlockStatus::InChainWithState {
+				self.import_existing = true;
+				// Latest state is missing, start with the last finalized state or genesis instead.
+				if let Some((hash, number)) = info.finalized_state {
+					log::debug!(target: "sync", "Starting from finalized state #{}", number);
+					self.best_queued_hash = hash;
+					self.best_queued_number = number;
+				} else {
+					log::debug!(target: "sync", "Restarting from genesis");
+					self.best_queued_hash = Default::default();
+					self.best_queued_number = Zero::zero();
+				}
+			}
+		}
+		log::trace!(target: "sync", "Restarted sync at #{} ({:?})", self.best_queued_number, self.best_queued_hash);
+		Ok(())
+	}
+
 	/// What is the status of the block corresponding to the given hash?
 	fn block_status(&self, hash: &B::Hash) -> Result<BlockStatus, ClientError> {
 		if self.queue_blocks.contains(hash) {
@@ -1764,7 +1974,7 @@ fn peer_block_request<B: BlockT>(
 	id: &PeerId,
 	peer: &PeerSync<B>,
 	blocks: &mut BlockCollection<B>,
-	attrs: &message::BlockAttributes,
+	attrs: message::BlockAttributes,
 	max_parallel_downloads: u32,
 	finalized: NumberFor<B>,
 	best_num: NumberFor<B>,
@@ -1815,7 +2025,7 @@ fn fork_sync_request<B: BlockT>(
 	targets: &mut HashMap<B::Hash, ForkTarget<B>>,
 	best_num: NumberFor<B>,
 	finalized: NumberFor<B>,
-	attributes: &message::BlockAttributes,
+	attributes: message::BlockAttributes,
 	check_block: impl Fn(&B::Hash) -> BlockStatus,
 ) -> Option<(B::Hash, BlockRequest<B>)> {
 	targets.retain(|hash, r| {
@@ -1994,17 +2204,15 @@ mod test {
 		// internally we should process the response as the justification not being available.
 
 		let client = Arc::new(TestClientBuilder::new().build());
-		let info = client.info();
 		let block_announce_validator = Box::new(DefaultBlockAnnounceValidator);
 		let peer_id = PeerId::random();
 
 		let mut sync = ChainSync::new(
-			Roles::AUTHORITY,
+			SyncMode::Full,
 			client.clone(),
-			&info,
 			block_announce_validator,
 			1,
-		);
+		).unwrap();
 
 		let (a1_hash, a1_number) = {
 			let a1 = client.new_block(Default::default()).unwrap().build().unwrap().block;
@@ -2067,15 +2275,12 @@ mod test {
 	#[test]
 	fn restart_doesnt_affect_peers_downloading_finality_data() {
 		let mut client = Arc::new(TestClientBuilder::new().build());
-		let info = client.info();
-
 		let mut sync = ChainSync::new(
-			Roles::AUTHORITY,
+			SyncMode::Full,
 			client.clone(),
-			&info,
 			Box::new(DefaultBlockAnnounceValidator),
 			1,
-		);
+		).unwrap();
 
 		let peer_id1 = PeerId::random();
 		let peer_id2 = PeerId::random();
@@ -2242,15 +2447,13 @@ mod test {
 		sp_tracing::try_init_simple();
 
 		let mut client = Arc::new(TestClientBuilder::new().build());
-		let info = client.info();
 
 		let mut sync = ChainSync::new(
-			Roles::AUTHORITY,
+			SyncMode::Full,
 			client.clone(),
-			&info,
 			Box::new(DefaultBlockAnnounceValidator),
 			5,
-		);
+		).unwrap();
 
 		let peer_id1 = PeerId::random();
 		let peer_id2 = PeerId::random();
@@ -2359,12 +2562,11 @@ mod test {
 		let info = client.info();
 
 		let mut sync = ChainSync::new(
-			Roles::AUTHORITY,
+			SyncMode::Full,
 			client.clone(),
-			&info,
 			Box::new(DefaultBlockAnnounceValidator),
 			5,
-		);
+		).unwrap();
 
 		let peer_id1 = PeerId::random();
 		let peer_id2 = PeerId::random();
@@ -2481,12 +2683,11 @@ mod test {
 		let info = client.info();
 
 		let mut sync = ChainSync::new(
-			Roles::AUTHORITY,
+			SyncMode::Full,
 			client.clone(),
-			&info,
 			Box::new(DefaultBlockAnnounceValidator),
 			5,
-		);
+		).unwrap();
 
 		let finalized_block = blocks[MAX_BLOCKS_TO_LOOK_BACKWARDS as usize * 2 - 1].clone();
 		let just = (*b"TEST", Vec::new());
@@ -2592,15 +2793,12 @@ mod test {
 			.map(|_| build_block(&mut client, None, false))
 			.collect::<Vec<_>>();
 
-		let info = client.info();
-
 		let mut sync = ChainSync::new(
-			Roles::AUTHORITY,
+			SyncMode::Full,
 			client.clone(),
-			&info,
 			Box::new(DefaultBlockAnnounceValidator),
 			1,
-		);
+		).unwrap();
 
 		let peer_id1 = PeerId::random();
 		let common_block = blocks[1].clone();

substrate/client/network/src/protocol/sync/state.rs

@@ -0,0 +1,187 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2017-2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// 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 std::sync::Arc;
+use codec::{Encode, Decode};
+use sp_runtime::traits::{Block as BlockT, Header, NumberFor};
+use sc_client_api::StorageProof;
+use crate::schema::v1::{StateRequest, StateResponse, StateEntry};
+use crate::chain::{Client, ImportedState};
+use super::StateDownloadProgress;
+
+/// State sync support.
+
+/// State sync state machine. Accumulates partial state data until it
+/// is ready to be imported.
+pub struct StateSync<B: BlockT> {
+	target_block: B::Hash,
+	target_header: B::Header,
+	target_root: B::Hash,
+	last_key: Vec<u8>,
+	state: Vec<(Vec<u8>, Vec<u8>)>,
+	complete: bool,
+	client: Arc<dyn Client<B>>,
+	imported_bytes: u64,
+	skip_proof: bool,
+}
+
+/// Import state chunk result.
+pub enum ImportResult<B: BlockT> {
+	/// State is complete and ready for import.
+	Import(B::Hash, B::Header, ImportedState<B>),
+	/// Continue dowloading.
+	Continue(StateRequest),
+	/// Bad state chunk.
+	BadResponse,
+}
+
+impl<B: BlockT> StateSync<B> {
+	///  Create a new instance.
+	pub fn new(client: Arc<dyn Client<B>>, target: B::Header, skip_proof: bool) -> Self {
+		StateSync {
+			client,
+			target_block: target.hash(),
+			target_root: target.state_root().clone(),
+			target_header: target,
+			last_key: Vec::default(),
+			state: Vec::default(),
+			complete: false,
+			imported_bytes: 0,
+			skip_proof,
+		}
+	}
+
+	///  Validate and import a state reponse.
+	pub fn import(&mut self, response: StateResponse) -> ImportResult<B> {
+		if response.entries.is_empty() && response.proof.is_empty() && !response.complete {
+			log::debug!(
+				target: "sync",
+				"Bad state response",
+			);
+			return ImportResult::BadResponse;
+		}
+		if !self.skip_proof && response.proof.is_empty() {
+			log::debug!(
+				target: "sync",
+				"Missing proof",
+			);
+			return ImportResult::BadResponse;
+		}
+		let complete = if !self.skip_proof {
+			log::debug!(
+				target: "sync",
+				"Importing state from {} trie nodes",
+				response.proof.len(),
+			);
+			let proof_size = response.proof.len() as u64;
+			let proof = match StorageProof::decode(&mut response.proof.as_ref()) {
+				Ok(proof) => proof,
+				Err(e) => {
+					log::debug!(target: "sync", "Error decoding proof: {:?}", e);
+					return ImportResult::BadResponse;
+				}
+			};
+			let (values, complete) = match self.client.verify_range_proof(
+				self.target_root,
+				proof,
+				&self.last_key
+			) {
+				Err(e) => {
+					log::debug!(
+						target: "sync",
+						"StateResponse failed proof verification: {:?}",
+						e,
+					);
+					return ImportResult::BadResponse;
+				},
+				Ok(values) => values,
+			};
+			log::debug!(target: "sync", "Imported with {} keys", values.len());
+
+			if let Some(last) = values.last().map(|(k, _)| k) {
+				self.last_key = last.clone();
+			}
+
+			for (key, value) in values {
+				self.imported_bytes += key.len() as u64;
+				self.state.push((key, value))
+			};
+			self.imported_bytes += proof_size;
+			complete
+		} else {
+			log::debug!(
+				target: "sync",
+				"Importing state from {:?} to {:?}",
+				response.entries.last().map(|e| sp_core::hexdisplay::HexDisplay::from(&e.key)),
+				response.entries.first().map(|e| sp_core::hexdisplay::HexDisplay::from(&e.key)),
+			);
+
+			if let Some(e) = response.entries.last() {
+				self.last_key = e.key.clone();
+			}
+			for StateEntry { key, value } in response.entries {
+				self.imported_bytes += (key.len() + value.len()) as u64;
+				self.state.push((key, value))
+			}
+			response.complete
+		};
+		if complete {
+			self.complete = true;
+			ImportResult::Import(self.target_block.clone(), self.target_header.clone(), ImportedState {
+				block: self.target_block.clone(),
+				state: std::mem::take(&mut self.state)
+			})
+		} else {
+			ImportResult::Continue(self.next_request())
+		}
+	}
+
+	/// Produce next state request.
+	pub fn next_request(&self) -> StateRequest {
+		StateRequest {
+			block: self.target_block.encode(),
+			start: self.last_key.clone(),
+			no_proof: self.skip_proof,
+		}
+	}
+
+	/// Check if the state is complete.
+	pub fn is_complete(&self) -> bool {
+		self.complete
+	}
+
+	/// Returns target block number.
+	pub fn target_block_num(&self) -> NumberFor<B> {
+		self.target_header.number().clone()
+	}
+
+	/// Returns target block hash.
+	pub fn target(&self) -> B::Hash {
+		self.target_block.clone()
+	}
+
+	/// Returns state sync estimated progress.
+	pub fn progress(&self) -> StateDownloadProgress {
+		let percent_done = (*self.last_key.get(0).unwrap_or(&0u8) as u32) * 100 / 256;
+		StateDownloadProgress {
+			percentage: percent_done,
+			size: self.imported_bytes,
+		}
+	}
+}
+

substrate/client/network/src/schema/api.v1.proto

@@ -68,3 +68,28 @@ message BlockData {
 	bytes justifications = 8; // optional
 }
 
+// Request storage data from a peer.
+message StateRequest {
+	// Block header hash.
+	bytes block = 1;
+	// Start from this key. Equivalent to <empty bytes> if omitted.
+	bytes start = 2; // optional
+	// if 'true' indicates that response should contain raw key-values, rather than proof.
+	bool no_proof = 3;
+}
+
+message StateResponse {
+	// A collection of keys-values. Only populated if `no_proof` is `true`
+	repeated StateEntry entries = 1;
+	// If `no_proof` is false in request, this contains proof nodes.
+	bytes proof = 2;
+	// Set to true when there are no more keys to return.
+	bool complete = 3;
+}
+
+// A key-value pair
+message StateEntry {
+	bytes key = 1;
+	bytes value = 2;
+}
+

substrate/client/network/src/service.rs

@@ -48,7 +48,7 @@ use crate::{
 		Protocol,
 		Ready,
 		event::Event,
-		sync::SyncState,
+		sync::{SyncState, Status as SyncStatus},
 	},
 	transactions,
 	transport, ReputationChange,
@@ -196,6 +196,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
 			protocol::ProtocolConfig {
 				roles: From::from(&params.role),
 				max_parallel_downloads: params.network_config.max_parallel_downloads,
+				sync_mode: params.network_config.sync_mode.clone(),
 			},
 			params.chain.clone(),
 			params.protocol_id.clone(),
@@ -331,7 +332,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
 			};
 
 			let behaviour = {
-				let bitswap = if params.network_config.ipfs_server { Some(Bitswap::new(client)) } else { None };
+				let bitswap = params.network_config.ipfs_server.then(|| Bitswap::new(client));
 				let result = Behaviour::new(
 					protocol,
 					user_agent,
@@ -339,6 +340,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
 					light_client_request_sender,
 					discovery_config,
 					params.block_request_protocol_config,
+					params.state_request_protocol_config,
 					bitswap,
 					params.light_client_request_protocol_config,
 					params.network_config.request_response_protocols,
@@ -442,14 +444,16 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
 
 	/// High-level network status information.
 	pub fn status(&self) -> NetworkStatus<B> {
+		let status = self.sync_state();
 		NetworkStatus {
-			sync_state: self.sync_state(),
+			sync_state: status.state,
 			best_seen_block: self.best_seen_block(),
 			num_sync_peers: self.num_sync_peers(),
 			num_connected_peers: self.num_connected_peers(),
 			num_active_peers: self.num_active_peers(),
 			total_bytes_inbound: self.total_bytes_inbound(),
 			total_bytes_outbound: self.total_bytes_outbound(),
+			state_sync: status.state_sync,
 		}
 	}
 
@@ -474,7 +478,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
 	}
 
 	/// Current global sync state.
-	pub fn sync_state(&self) -> SyncState {
+	pub fn sync_state(&self) -> SyncStatus<B> {
 		self.network_service.behaviour().user_protocol().sync_state()
 	}
 
@@ -1869,7 +1873,7 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
 			*this.external_addresses.lock() = external_addresses;
 		}
 
-		let is_major_syncing = match this.network_service.behaviour_mut().user_protocol_mut().sync_state() {
+		let is_major_syncing = match this.network_service.behaviour_mut().user_protocol_mut().sync_state().state {
 			SyncState::Idle => false,
 			SyncState::Downloading => true,
 		};

substrate/client/network/src/service/tests.rs

@@ -18,6 +18,7 @@
 
 use crate::{config, Event, NetworkService, NetworkWorker};
 use crate::block_request_handler::BlockRequestHandler;
+use crate::state_request_handler::StateRequestHandler;
 use crate::light_client_requests::handler::LightClientRequestHandler;
 
 use libp2p::PeerId;
@@ -107,6 +108,16 @@ fn build_test_full_node(config: config::NetworkConfiguration)
 		protocol_config
 	};
 
+	let state_request_protocol_config = {
+		let (handler, protocol_config) = StateRequestHandler::new(
+			&protocol_id,
+			client.clone(),
+			50,
+		);
+		async_std::task::spawn(handler.run().boxed());
+		protocol_config
+	};
+
 	let light_client_request_protocol_config = {
 		let (handler, protocol_config) = LightClientRequestHandler::new(
 			&protocol_id,
@@ -131,6 +142,7 @@ fn build_test_full_node(config: config::NetworkConfiguration)
 		),
 		metrics_registry: None,
 		block_request_protocol_config,
+		state_request_protocol_config,
 		light_client_request_protocol_config,
 	})
 	.unwrap();

substrate/client/network/src/state_request_handler.rs

@@ -0,0 +1,246 @@
+// Copyright 2020 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Helper for handling (i.e. answering) state requests from a remote peer via the
+//! [`crate::request_responses::RequestResponsesBehaviour`].
+
+use codec::{Encode, Decode};
+use crate::chain::Client;
+use crate::config::ProtocolId;
+use crate::request_responses::{IncomingRequest, OutgoingResponse, ProtocolConfig};
+use crate::schema::v1::{StateResponse, StateRequest, StateEntry};
+use crate::{PeerId, ReputationChange};
+use futures::channel::{mpsc, oneshot};
+use futures::stream::StreamExt;
+use log::debug;
+use lru::LruCache;
+use prost::Message;
+use sp_runtime::generic::BlockId;
+use sp_runtime::traits::Block as BlockT;
+use std::sync::Arc;
+use std::time::Duration;
+use std::hash::{Hasher, Hash};
+
+const LOG_TARGET: &str = "sync";
+const MAX_RESPONSE_BYTES: usize = 2 * 1024 * 1024; // Actual reponse may be bigger.
+const MAX_NUMBER_OF_SAME_REQUESTS_PER_PEER: usize = 2;
+
+mod rep {
+	use super::ReputationChange as Rep;
+
+	/// Reputation change when a peer sent us the same request multiple times.
+	pub const SAME_REQUEST: Rep = Rep::new(i32::min_value(), "Same state request multiple times");
+}
+
+/// Generates a [`ProtocolConfig`] for the block request protocol, refusing incoming requests.
+pub fn generate_protocol_config(protocol_id: &ProtocolId) -> ProtocolConfig {
+	ProtocolConfig {
+		name: generate_protocol_name(protocol_id).into(),
+		max_request_size: 1024 * 1024,
+		max_response_size: 16 * 1024 * 1024,
+		request_timeout: Duration::from_secs(40),
+		inbound_queue: None,
+	}
+}
+
+/// Generate the state protocol name from chain specific protocol identifier.
+fn generate_protocol_name(protocol_id: &ProtocolId) -> String {
+	let mut s = String::new();
+	s.push_str("/");
+	s.push_str(protocol_id.as_ref());
+	s.push_str("/state/1");
+	s
+}
+
+/// The key of [`BlockRequestHandler::seen_requests`].
+#[derive(Eq, PartialEq, Clone)]
+struct SeenRequestsKey<B: BlockT> {
+	peer: PeerId,
+	block: B::Hash,
+	start: Vec<u8>,
+}
+
+impl<B: BlockT> Hash for SeenRequestsKey<B> {
+	fn hash<H: Hasher>(&self, state: &mut H) {
+		self.peer.hash(state);
+		self.block.hash(state);
+		self.start.hash(state);
+	}
+}
+
+/// The value of [`StateRequestHandler::seen_requests`].
+enum SeenRequestsValue {
+	/// First time we have seen the request.
+	First,
+	/// We have fulfilled the request `n` times.
+	Fulfilled(usize),
+}
+
+/// Handler for incoming block requests from a remote peer.
+pub struct StateRequestHandler<B: BlockT> {
+	client: Arc<dyn Client<B>>,
+	request_receiver: mpsc::Receiver<IncomingRequest>,
+	/// Maps from request to number of times we have seen this request.
+	///
+	/// This is used to check if a peer is spamming us with the same request.
+	seen_requests: LruCache<SeenRequestsKey<B>, SeenRequestsValue>,
+}
+
+impl<B: BlockT> StateRequestHandler<B> {
+	/// Create a new [`StateRequestHandler`].
+	pub fn new(
+		protocol_id: &ProtocolId,
+		client: Arc<dyn Client<B>>,
+		num_peer_hint: usize,
+	) -> (Self, ProtocolConfig) {
+		// Reserve enough request slots for one request per peer when we are at the maximum
+		// number of peers.
+		let (tx, request_receiver) = mpsc::channel(num_peer_hint);
+
+		let mut protocol_config = generate_protocol_config(protocol_id);
+		protocol_config.inbound_queue = Some(tx);
+
+		let seen_requests = LruCache::new(num_peer_hint * 2);
+
+		(Self { client, request_receiver, seen_requests }, protocol_config)
+	}
+
+	/// Run [`StateRequestHandler`].
+	pub async fn run(mut self) {
+		while let Some(request) = self.request_receiver.next().await {
+			let IncomingRequest { peer, payload, pending_response } = request;
+
+			match self.handle_request(payload, pending_response, &peer) {
+				Ok(()) => debug!(target: LOG_TARGET, "Handled block request from {}.", peer),
+				Err(e) => debug!(
+					target: LOG_TARGET,
+					"Failed to handle state request from {}: {}",
+					peer,
+					e,
+				),
+			}
+		}
+	}
+
+	fn handle_request(
+		&mut self,
+		payload: Vec<u8>,
+		pending_response: oneshot::Sender<OutgoingResponse>,
+		peer: &PeerId,
+	) -> Result<(), HandleRequestError> {
+		let request = StateRequest::decode(&payload[..])?;
+		let block: B::Hash = Decode::decode(&mut request.block.as_ref())?;
+
+		let key = SeenRequestsKey {
+			peer: *peer,
+			block: block.clone(),
+			start: request.start.clone(),
+		};
+
+		let mut reputation_changes = Vec::new();
+
+		match self.seen_requests.get_mut(&key) {
+			Some(SeenRequestsValue::First) => {},
+			Some(SeenRequestsValue::Fulfilled(ref mut requests)) => {
+				*requests = requests.saturating_add(1);
+
+				if *requests > MAX_NUMBER_OF_SAME_REQUESTS_PER_PEER {
+					reputation_changes.push(rep::SAME_REQUEST);
+				}
+			},
+			None => {
+				self.seen_requests.put(key.clone(), SeenRequestsValue::First);
+			}
+		}
+
+		log::trace!(
+			target: LOG_TARGET,
+			"Handling state request from {}: Block {:?}, Starting at {:?}, no_proof={}",
+			peer,
+			request.block,
+			sp_core::hexdisplay::HexDisplay::from(&request.start),
+			request.no_proof,
+		);
+
+		let result = if reputation_changes.is_empty() {
+			let mut response = StateResponse::default();
+
+			if !request.no_proof {
+				let (proof, count) = self.client.read_proof_collection(
+					&BlockId::hash(block),
+					&request.start,
+					MAX_RESPONSE_BYTES,
+				)?;
+				response.proof = proof.encode();
+				if count == 0 {
+					response.complete = true;
+				}
+			} else {
+				let entries = self.client.storage_collection(
+					&BlockId::hash(block),
+					&request.start,
+					MAX_RESPONSE_BYTES,
+				)?;
+				response.entries = entries.into_iter().map(|(key, value)| StateEntry { key, value }).collect();
+				if response.entries.is_empty() {
+					response.complete = true;
+				}
+			}
+
+			log::trace!(
+				target: LOG_TARGET,
+				"StateResponse contains {} keys, {}, proof nodes, complete={}, from {:?} to {:?}",
+				response.entries.len(),
+				response.proof.len(),
+				response.complete,
+				response.entries.first().map(|e| sp_core::hexdisplay::HexDisplay::from(&e.key)),
+				response.entries.last().map(|e| sp_core::hexdisplay::HexDisplay::from(&e.key)),
+			);
+			if let Some(value) = self.seen_requests.get_mut(&key) {
+				// If this is the first time we have processed this request, we need to change
+				// it to `Fulfilled`.
+				if let SeenRequestsValue::First = value {
+					*value = SeenRequestsValue::Fulfilled(1);
+				}
+			}
+
+			let mut data = Vec::with_capacity(response.encoded_len());
+			response.encode(&mut data)?;
+			Ok(data)
+		} else {
+			Err(())
+		};
+
+		pending_response.send(OutgoingResponse {
+			result,
+			reputation_changes,
+			sent_feedback: None,
+		}).map_err(|_| HandleRequestError::SendResponse)
+	}
+}
+
+#[derive(derive_more::Display, derive_more::From)]
+enum HandleRequestError {
+	#[display(fmt = "Failed to decode request: {}.", _0)]
+	DecodeProto(prost::DecodeError),
+	#[display(fmt = "Failed to encode response: {}.", _0)]
+	EncodeProto(prost::EncodeError),
+	#[display(fmt = "Failed to decode block hash: {}.", _0)]
+	InvalidHash(codec::Error),
+	Client(sp_blockchain::Error),
+	#[display(fmt = "Failed to send response.")]
+	SendResponse,
+}