Warp sync part II (#9284)

* Gap sync * Gap epoch test * Simplified network requests * Update client/db/src/utils.rs Co-authored-by: cheme <emericchevalier.pro@gmail.com> * Fixed v1 migration and added some comments * Next epoch is always regular * Removed fork tree change * Apply suggestions from code review Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com> * Added a comment and converted assert to error Co-authored-by: cheme <emericchevalier.pro@gmail.com> Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>
2026-04-26 11:07:56 +00:00 · 2021-10-07 11:31:39 +02:00
parent 9f1c3acb7d
commit e6ff531d0b
29 changed files with 800 additions and 169 deletions
@@ -182,6 +182,12 @@ impl Default for PendingRequests {
 	}
 }

+struct GapSync<B: BlockT> {
+	blocks: BlockCollection<B>,
+	best_queued_number: NumberFor<B>,
+	target: NumberFor<B>,
+}
+
 /// The main data structure which contains all the state for a chains
 /// active syncing strategy.
 pub struct ChainSync<B: BlockT> {
@@ -226,6 +232,8 @@ pub struct ChainSync<B: BlockT> {
 	/// 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,
+	/// Gap download process.
+	gap_sync: Option<GapSync<B>>,
 }

 /// All the data we have about a Peer that we are trying to sync with
@@ -298,6 +306,8 @@ pub enum PeerSyncState<B: BlockT> {
 	DownloadingState,
 	/// Downloading warp proof.
 	DownloadingWarpProof,
+	/// Actively downloading block history after warp sync.
+	DownloadingGap(NumberFor<B>),
 }

 impl<B: BlockT> PeerSyncState<B> {
@@ -326,7 +336,7 @@ pub struct StateDownloadProgress {

 /// Reported warp sync phase.
 #[derive(Clone, Eq, PartialEq, Debug)]
-pub enum WarpSyncPhase {
+pub enum WarpSyncPhase<B: BlockT> {
 	/// Waiting for peers to connect.
 	AwaitingPeers,
 	/// Downloading and verifying grandpa warp proofs.
@@ -335,24 +345,27 @@ pub enum WarpSyncPhase {
 	DownloadingState,
 	/// Importing state.
 	ImportingState,
+	/// Downloading block history.
+	DownloadingBlocks(NumberFor<B>),
 }

-impl fmt::Display for WarpSyncPhase {
+impl<B: BlockT> fmt::Display for WarpSyncPhase<B> {
 	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 		match self {
 			Self::AwaitingPeers => write!(f, "Waiting for peers"),
 			Self::DownloadingWarpProofs => write!(f, "Downloading finality proofs"),
 			Self::DownloadingState => write!(f, "Downloading state"),
 			Self::ImportingState => write!(f, "Importing state"),
+			Self::DownloadingBlocks(n) => write!(f, "Downloading block history (#{})", n),
 		}
 	}
 }

 /// Reported warp sync progress.
 #[derive(Clone, Eq, PartialEq, Debug)]
-pub struct WarpSyncProgress {
+pub struct WarpSyncProgress<B: BlockT> {
 	/// Estimated download percentage.
-	pub phase: WarpSyncPhase,
+	pub phase: WarpSyncPhase<B>,
 	/// Total bytes downloaded so far.
 	pub total_bytes: u64,
 }
@@ -371,7 +384,7 @@ pub struct Status<B: BlockT> {
 	/// State sync status in progress, if any.
 	pub state_sync: Option<StateDownloadProgress>,
 	/// Warp sync in progress, if any.
-	pub warp_sync: Option<WarpSyncProgress>,
+	pub warp_sync: Option<WarpSyncProgress<B>>,
 }

 /// A peer did not behave as expected and should be reported.
@@ -413,16 +426,7 @@ 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::on_warp_sync_data`].
-#[derive(Debug)]
-pub enum OnWarpSyncData<B: BlockT> {
-	/// Warp proof request is issued.
-	WarpProofRequest(PeerId, warp::WarpProofRequest<B>),
-	/// A new state request needs to be made to the given peer.
-	StateRequest(PeerId, StateRequest),
+	Continue,
 }

 /// Result of [`ChainSync::poll_block_announce_validation`].
@@ -555,6 +559,7 @@ impl<B: BlockT> ChainSync<B> {
 			warp_sync: None,
 			warp_sync_provider,
 			import_existing: false,
+			gap_sync: None,
 		};
 		sync.reset_sync_start_point()?;
 		Ok(sync)
@@ -608,10 +613,14 @@ impl<B: BlockT> ChainSync<B> {
 			SyncState::Idle
 		};

-		let warp_sync_progress = match (&self.warp_sync, &self.mode) {
-			(None, SyncMode::Warp) =>
+		let warp_sync_progress = match (&self.warp_sync, &self.mode, &self.gap_sync) {
+			(_, _, Some(gap_sync)) => Some(WarpSyncProgress {
+				phase: WarpSyncPhase::DownloadingBlocks(gap_sync.best_queued_number),
+				total_bytes: 0,
+			}),
+			(None, SyncMode::Warp, _) =>
 				Some(WarpSyncProgress { phase: WarpSyncPhase::AwaitingPeers, total_bytes: 0 }),
-			(Some(sync), _) => Some(sync.progress()),
+			(Some(sync), _, _) => Some(sync.progress()),
 			_ => None,
 		};

@@ -686,17 +695,6 @@ impl<B: BlockT> ChainSync<B> {
 					return Ok(None)
 				}

-				if let SyncMode::Warp = &self.mode {
-					if self.peers.len() >= MIN_PEERS_TO_START_WARP_SYNC && self.warp_sync.is_none()
-					{
-						log::debug!(target: "sync", "Starting warp state sync.");
-						if let Some(provider) = &self.warp_sync_provider {
-							self.warp_sync =
-								Some(WarpSync::new(self.client.clone(), provider.clone()));
-						}
-					}
-				}
-
 				// If we are at genesis, just start downloading.
 				let (state, req) = if self.best_queued_number.is_zero() {
 					debug!(
@@ -739,6 +737,17 @@ impl<B: BlockT> ChainSync<B> {
 					},
 				);

+				if let SyncMode::Warp = &self.mode {
+					if self.peers.len() >= MIN_PEERS_TO_START_WARP_SYNC && self.warp_sync.is_none()
+					{
+						log::debug!(target: "sync", "Starting warp state sync.");
+						if let Some(provider) = &self.warp_sync_provider {
+							self.warp_sync =
+								Some(WarpSync::new(self.client.clone(), provider.clone()));
+						}
+					}
+				}
+
 				Ok(req)
 			},
 			Ok(BlockStatus::Queued) |
@@ -869,10 +878,13 @@ 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() || self.state_sync.is_some() || self.warp_sync.is_some()
+		if self.pending_requests.is_empty() ||
+			self.state_sync.is_some() ||
+			self.mode == SyncMode::Warp
 		{
 			return Either::Left(std::iter::empty())
 		}
+
 		if self.queue_blocks.len() > MAX_IMPORTING_BLOCKS {
 			trace!(target: "sync", "Too many blocks in the queue.");
 			return Either::Left(std::iter::empty())
@@ -888,6 +900,7 @@ impl<B: BlockT> ChainSync<B> {
 		let queue = &self.queue_blocks;
 		let pending_requests = self.pending_requests.take();
 		let max_parallel = if major_sync { 1 } else { self.max_parallel_downloads };
+		let gap_sync = &mut self.gap_sync;
 		let iter = self.peers.iter_mut().filter_map(move |(id, peer)| {
 			if !peer.state.is_available() || !pending_requests.contains(id) {
 				return None
@@ -947,6 +960,26 @@ impl<B: BlockT> ChainSync<B> {
 				trace!(target: "sync", "Downloading fork {:?} from {}", hash, id);
 				peer.state = PeerSyncState::DownloadingStale(hash);
 				Some((id, req))
+			} else if let Some((range, req)) = gap_sync.as_mut().and_then(|sync| {
+				peer_gap_block_request(
+					id,
+					peer,
+					&mut sync.blocks,
+					attrs,
+					sync.target,
+					sync.best_queued_number,
+				)
+			}) {
+				peer.state = PeerSyncState::DownloadingGap(range.start);
+				trace!(
+					target: "sync",
+					"New gap block request for {}, (best:{}, common:{}) {:?}",
+					id,
+					peer.best_number,
+					peer.common_number,
+					req,
+				);
+				Some((id, req))
 			} else {
 				None
 			}
@@ -966,9 +999,9 @@ impl<B: BlockT> ChainSync<B> {
 			}
 			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();
+					trace!(target: "sync", "New StateRequest for {}: {:?}", id, request);
 					return Some((*id, request))
 				}
 			}
@@ -982,7 +1015,7 @@ impl<B: BlockT> ChainSync<B> {
 			{
 				for (id, peer) in self.peers.iter_mut() {
 					if peer.state.is_available() && peer.best_number >= target {
-						trace!(target: "sync", "New StateRequest for {}", id);
+						trace!(target: "sync", "New StateRequest for {}: {:?}", id, request);
 						peer.state = PeerSyncState::DownloadingState;
 						return Some((*id, request))
 					}
@@ -1039,6 +1072,7 @@ impl<B: BlockT> ChainSync<B> {
 		response: BlockResponse<B>,
 	) -> Result<OnBlockData<B>, BadPeer> {
 		self.downloaded_blocks += response.blocks.len();
+		let mut gap = false;
 		let new_blocks: Vec<IncomingBlock<B>> = if let Some(peer) = self.peers.get_mut(who) {
 			let mut blocks = response.blocks;
 			if request
@@ -1061,6 +1095,43 @@ impl<B: BlockT> ChainSync<B> {
 						}
 						self.drain_blocks()
 					},
+					PeerSyncState::DownloadingGap(start_block) => {
+						let start_block = *start_block;
+						peer.state = PeerSyncState::Available;
+						if let Some(gap_sync) = &mut self.gap_sync {
+							gap_sync.blocks.clear_peer_download(who);
+							validate_blocks::<B>(&blocks, who, Some(request))?;
+							gap_sync.blocks.insert(start_block, blocks, who.clone());
+							gap = true;
+							gap_sync
+								.blocks
+								.drain(gap_sync.best_queued_number + One::one())
+								.into_iter()
+								.map(|block_data| {
+									let justifications = block_data.block.justifications.or(
+										legacy_justification_mapping(
+											block_data.block.justification,
+										),
+									);
+									IncomingBlock {
+										hash: block_data.block.hash,
+										header: block_data.block.header,
+										body: block_data.block.body,
+										indexed_body: block_data.block.indexed_body,
+										justifications,
+										origin: block_data.origin,
+										allow_missing_state: true,
+										import_existing: self.import_existing,
+										skip_execution: true,
+										state: None,
+									}
+								})
+								.collect()
+						} else {
+							debug!(target: "sync", "Unexpected gap block response from {}", who);
+							return Err(BadPeer(who.clone(), rep::NO_BLOCK))
+						}
+					},
 					PeerSyncState::DownloadingStale(_) => {
 						peer.state = PeerSyncState::Available;
 						if blocks.is_empty() {
@@ -1212,7 +1283,7 @@ impl<B: BlockT> ChainSync<B> {
 			return Err(BadPeer(*who, rep::NOT_REQUESTED))
 		};

-		Ok(self.validate_and_queue_blocks(new_blocks))
+		Ok(self.validate_and_queue_blocks(new_blocks, gap))
 	}

 	/// Handle a response from the remote to a state request that we made.
@@ -1223,6 +1294,11 @@ impl<B: BlockT> ChainSync<B> {
 		who: &PeerId,
 		response: StateResponse,
 	) -> Result<OnStateData<B>, BadPeer> {
+		if let Some(peer) = self.peers.get_mut(&who) {
+			if let PeerSyncState::DownloadingState = peer.state {
+				peer.state = PeerSyncState::Available;
+			}
+		}
 		let import_result = if let Some(sync) = &mut self.state_sync {
 			debug!(
 				target: "sync",
@@ -1261,11 +1337,10 @@ impl<B: BlockT> ChainSync<B> {
 					skip_execution: self.skip_execution(),
 					state: Some(state),
 				};
-				debug!(target: "sync", "State sync is complete. Import is queued");
+				debug!(target: "sync", "State download is complete. Import is queued");
 				Ok(OnStateData::Import(origin, block))
 			},
-			state::ImportResult::Continue(request) =>
-				Ok(OnStateData::Request(who.clone(), request)),
+			state::ImportResult::Continue => Ok(OnStateData::Continue),
 			state::ImportResult::BadResponse => {
 				debug!(target: "sync", "Bad state data received from {}", who);
 				Err(BadPeer(*who, rep::BAD_BLOCK))
@@ -1280,7 +1355,12 @@ impl<B: BlockT> ChainSync<B> {
 		&mut self,
 		who: &PeerId,
 		response: warp::EncodedProof,
-	) -> Result<OnWarpSyncData<B>, BadPeer> {
+	) -> Result<(), BadPeer> {
+		if let Some(peer) = self.peers.get_mut(&who) {
+			if let PeerSyncState::DownloadingWarpProof = peer.state {
+				peer.state = PeerSyncState::Available;
+			}
+		}
 		let import_result = if let Some(sync) = &mut self.warp_sync {
 			debug!(
 				target: "sync",
@@ -1295,10 +1375,7 @@ impl<B: BlockT> ChainSync<B> {
 		};

 		match import_result {
-			warp::WarpProofImportResult::StateRequest(request) =>
-				Ok(OnWarpSyncData::StateRequest(*who, request)),
-			warp::WarpProofImportResult::WarpProofRequest(request) =>
-				Ok(OnWarpSyncData::WarpProofRequest(*who, request)),
+			warp::WarpProofImportResult::Success => Ok(()),
 			warp::WarpProofImportResult::BadResponse => {
 				debug!(target: "sync", "Bad proof data received from {}", who);
 				Err(BadPeer(*who, rep::BAD_BLOCK))
@@ -1309,6 +1386,7 @@ impl<B: BlockT> ChainSync<B> {
 	fn validate_and_queue_blocks(
 		&mut self,
 		mut new_blocks: Vec<IncomingBlock<B>>,
+		gap: bool,
 	) -> OnBlockData<B> {
 		let orig_len = new_blocks.len();
 		new_blocks.retain(|b| !self.queue_blocks.contains(&b.hash));
@@ -1320,7 +1398,7 @@ impl<B: BlockT> ChainSync<B> {
 			);
 		}

-		let origin = if self.status().state != SyncState::Downloading {
+		let origin = if !gap && self.status().state != SyncState::Downloading {
 			BlockOrigin::NetworkBroadcast
 		} else {
 			BlockOrigin::NetworkInitialSync
@@ -1494,6 +1572,15 @@ impl<B: BlockT> ChainSync<B> {
 						self.mode = SyncMode::Full;
 						output.extend(self.restart());
 					}
+					let gap_sync_complete =
+						self.gap_sync.as_ref().map_or(false, |s| s.target == number);
+					if gap_sync_complete {
+						info!(
+							target: "sync",
+							"Block history download is complete."
+						);
+						self.gap_sync = None;
+					}
 				},
 				Err(BlockImportError::IncompleteHeader(who)) =>
 					if let Some(peer) = who {
@@ -1601,6 +1688,11 @@ impl<B: BlockT> ChainSync<B> {
 		if self.fork_targets.remove(&hash).is_some() {
 			trace!(target: "sync", "Completed fork sync {:?}", hash);
 		}
+		if let Some(gap_sync) = &mut self.gap_sync {
+			if number > gap_sync.best_queued_number && number <= gap_sync.target {
+				gap_sync.best_queued_number = number;
+			}
+		}
 		if number > self.best_queued_number {
 			self.best_queued_number = number;
 			self.best_queued_hash = *hash;
@@ -1954,6 +2046,9 @@ impl<B: BlockT> ChainSync<B> {
 	/// import, so this functions checks for such blocks and returns them.
 	pub fn peer_disconnected(&mut self, who: &PeerId) -> Option<OnBlockData<B>> {
 		self.blocks.clear_peer_download(who);
+		if let Some(gap_sync) = &mut self.gap_sync {
+			gap_sync.blocks.clear_peer_download(who)
+		}
 		self.peers.remove(who);
 		self.extra_justifications.peer_disconnected(who);
 		self.pending_requests.set_all();
@@ -1963,7 +2058,7 @@ impl<B: BlockT> ChainSync<B> {
 		});
 		let blocks = self.drain_blocks();
 		if !blocks.is_empty() {
-			Some(self.validate_and_queue_blocks(blocks))
+			Some(self.validate_and_queue_blocks(blocks, false))
 		} else {
 			None
 		}
@@ -2043,6 +2138,14 @@ impl<B: BlockT> ChainSync<B> {
 				}
 			}
 		}
+		if let Some((start, end)) = info.block_gap {
+			debug!(target: "sync", "Starting gap sync #{} - #{}", start, end);
+			self.gap_sync = Some(GapSync {
+				best_queued_number: start - One::one(),
+				target: end,
+				blocks: BlockCollection::new(),
+			});
+		}
 		trace!(target: "sync", "Restarted sync at #{} ({:?})", self.best_queued_number, self.best_queued_hash);
 		Ok(())
 	}
@@ -2250,6 +2353,39 @@ fn peer_block_request<B: BlockT>(
 	Some((range, request))
 }

+/// Get a new block request for the peer if any.
+fn peer_gap_block_request<B: BlockT>(
+	id: &PeerId,
+	peer: &PeerSync<B>,
+	blocks: &mut BlockCollection<B>,
+	attrs: message::BlockAttributes,
+	target: NumberFor<B>,
+	common_number: NumberFor<B>,
+) -> Option<(Range<NumberFor<B>>, BlockRequest<B>)> {
+	let range = blocks.needed_blocks(
+		id.clone(),
+		MAX_BLOCKS_TO_REQUEST,
+		std::cmp::min(peer.best_number, target),
+		common_number,
+		1,
+		MAX_DOWNLOAD_AHEAD,
+	)?;
+
+	// The end is not part of the range.
+	let last = range.end.saturating_sub(One::one());
+	let from = message::FromBlock::Number(last);
+
+	let request = message::generic::BlockRequest {
+		id: 0,
+		fields: attrs.clone(),
+		from,
+		to: None,
+		direction: message::Direction::Descending,
+		max: Some((range.end - range.start).saturated_into::<u32>()),
+	};
+	Some((range, request))
+}
+
 /// Get pending fork sync targets for a peer.
 fn fork_sync_request<B: BlockT>(
 	id: &PeerId,
@@ -47,8 +47,8 @@ pub struct StateSync<B: BlockT> {
 pub enum ImportResult<B: BlockT> {
 	/// State is complete and ready for import.
 	Import(B::Hash, B::Header, ImportedState<B>),
-	/// Continue dowloading.
-	Continue(StateRequest),
+	/// Continue downloading.
+	Continue,
 	/// Bad state chunk.
 	BadResponse,
 }
@@ -134,7 +134,7 @@ impl<B: BlockT> StateSync<B> {
 				ImportedState { block: self.target_block, state: std::mem::take(&mut self.state) },
 			)
 		} else {
-			ImportResult::Continue(self.next_request())
+			ImportResult::Continue
 		}
 	}

@@ -37,11 +37,9 @@ enum Phase<B: BlockT> {
 }

 /// Import warp proof result.
-pub enum WarpProofImportResult<B: BlockT> {
-	/// Start downloading state data.
-	StateRequest(StateRequest),
-	/// Continue dowloading warp sync proofs.
-	WarpProofRequest(WarpProofRequest<B>),
+pub enum WarpProofImportResult {
+	/// Import was successful.
+	Success,
 	/// Bad proof.
 	BadResponse,
 }
@@ -69,7 +67,7 @@ impl<B: BlockT> WarpSync<B> {
 		Self { client, warp_sync_provider, phase, total_proof_bytes: 0 }
 	}

-	///  Validate and import a state reponse.
+	///  Validate and import a state response.
 	pub fn import_state(&mut self, response: StateResponse) -> ImportResult<B> {
 		match &mut self.phase {
 			Phase::WarpProof { .. } => {
@@ -80,19 +78,15 @@ impl<B: BlockT> WarpSync<B> {
 		}
 	}

-	///  Validate and import a warp proof reponse.
-	pub fn import_warp_proof(&mut self, response: EncodedProof) -> WarpProofImportResult<B> {
+	///  Validate and import a warp proof response.
+	pub fn import_warp_proof(&mut self, response: EncodedProof) -> WarpProofImportResult {
 		match &mut self.phase {
 			Phase::State(_) => {
 				log::debug!(target: "sync", "Unexpected warp proof response");
 				WarpProofImportResult::BadResponse
 			},
 			Phase::WarpProof { set_id, authorities, last_hash } => {
-				match self.warp_sync_provider.verify(
-					&response,
-					*set_id,
-					std::mem::take(authorities),
-				) {
+				match self.warp_sync_provider.verify(&response, *set_id, authorities.clone()) {
 					Err(e) => {
 						log::debug!(target: "sync", "Bad warp proof response: {:?}", e);
 						return WarpProofImportResult::BadResponse
@@ -103,17 +97,14 @@ impl<B: BlockT> WarpSync<B> {
 						*authorities = new_authorities;
 						*last_hash = new_last_hash.clone();
 						self.total_proof_bytes += response.0.len() as u64;
-						WarpProofImportResult::WarpProofRequest(WarpProofRequest {
-							begin: new_last_hash,
-						})
+						WarpProofImportResult::Success
 					},
 					Ok(VerificationResult::Complete(new_set_id, _, header)) => {
 						log::debug!(target: "sync", "Verified complete proof, set_id={:?}", new_set_id);
 						self.total_proof_bytes += response.0.len() as u64;
 						let state_sync = StateSync::new(self.client.clone(), header, false);
-						let request = state_sync.next_request();
 						self.phase = Phase::State(state_sync);
-						WarpProofImportResult::StateRequest(request)
+						WarpProofImportResult::Success
 					},
 				}
 			},
@@ -161,7 +152,7 @@ impl<B: BlockT> WarpSync<B> {
 	}

 	/// Returns state sync estimated progress (percentage, bytes)
-	pub fn progress(&self) -> WarpSyncProgress {
+	pub fn progress(&self) -> WarpSyncProgress<B> {
 		match &self.phase {
 			Phase::WarpProof { .. } => WarpSyncProgress {
 				phase: WarpSyncPhase::DownloadingWarpProofs,