Move import queue from ChainSync to SyncingEngine (#1736)

This PR is part of [Sync 2.0](https://github.com/paritytech/polkadot-sdk/issues/534) refactoring aimed at making `ChainSync` a pure state machine. Resolves https://github.com/paritytech/polkadot-sdk/issues/501.
2026-04-26 20:27:58 +00:00 · 2023-09-29 17:58:16 +03:00
parent d8d90a82a7
commit 0691c91e15
4 changed files with 176 additions and 171 deletions
@@ -28,7 +28,8 @@ use crate::{
 	schema::v1::{StateRequest, StateResponse},
 	service::{self, chain_sync::ToServiceCommand},
 	warp::WarpSyncParams,
-	BlockRequestEvent, ChainSync, ClientError, SyncingService,
+	BlockRequestAction, ChainSync, ClientError, ImportBlocksAction, ImportJustificationsAction,
+	OnBlockResponse, SyncingService,
 };

 use codec::{Decode, Encode};
@@ -41,7 +42,8 @@ use futures_timer::Delay;
 use libp2p::{request_response::OutboundFailure, PeerId};
 use log::{debug, trace};
 use prometheus_endpoint::{
-	register, Gauge, GaugeVec, MetricSource, Opts, PrometheusError, Registry, SourcedGauge, U64,
+	register, Counter, Gauge, GaugeVec, MetricSource, Opts, PrometheusError, Registry,
+	SourcedGauge, U64,
 };
 use prost::Message;
 use schnellru::{ByLength, LruMap};
@@ -135,6 +137,8 @@ struct Metrics {
 	queued_blocks: Gauge<U64>,
 	fork_targets: Gauge<U64>,
 	justifications: GaugeVec<U64>,
+	import_queue_blocks_submitted: Counter<U64>,
+	import_queue_justifications_submitted: Counter<U64>,
 }

 impl Metrics {
@@ -164,6 +168,20 @@ impl Metrics {
 				)?;
 				register(g, r)?
 			},
+			import_queue_blocks_submitted: {
+				let c = Counter::new(
+					"substrate_sync_import_queue_blocks_submitted",
+					"Number of blocks submitted to the import queue.",
+				)?;
+				register(c, r)?
+			},
+			import_queue_justifications_submitted: {
+				let c = Counter::new(
+					"substrate_sync_import_queue_justifications_submitted",
+					"Number of justifications submitted to the import queue.",
+				)?;
+				register(c, r)?
+			},
 		})
 	}
 }
@@ -311,6 +329,9 @@ pub struct SyncingEngine<B: BlockT, Client> {

 	/// Protocol name used to send out warp sync requests
 	warp_sync_protocol_name: Option<ProtocolName>,
+
+	/// Handle to import queue.
+	import_queue: Box<dyn ImportQueueService<B>>,
 }

 impl<B: BlockT, Client> SyncingEngine<B, Client>
@@ -436,9 +457,7 @@ where
 			max_parallel_downloads,
 			max_blocks_per_request,
 			warp_sync_config,
-			metrics_registry,
 			network_service.clone(),
-			import_queue,
 		)?;

 		let (tx, service_rx) = tracing_unbounded("mpsc_chain_sync", 100_000);
@@ -501,6 +520,7 @@ where
 				block_downloader,
 				state_request_protocol_name,
 				warp_sync_protocol_name,
+				import_queue,
 			},
 			SyncingService::new(tx, num_connected, is_major_syncing),
 			block_announce_config,
@@ -728,13 +748,13 @@ where
 				ToServiceCommand::BlocksProcessed(imported, count, results) => {
 					for result in self.chain_sync.on_blocks_processed(imported, count, results) {
 						match result {
-							Ok(event) => match event {
-								BlockRequestEvent::SendRequest { peer_id, request } => {
+							Ok(action) => match action {
+								BlockRequestAction::SendRequest { peer_id, request } => {
 									// drop obsolete pending response first
 									self.pending_responses.remove(&peer_id);
 									self.send_block_request(peer_id, request);
 								},
-								BlockRequestEvent::RemoveStale { peer_id } => {
+								BlockRequestAction::RemoveStale { peer_id } => {
 									self.pending_responses.remove(&peer_id);
 								},
 							},
@@ -922,7 +942,10 @@ where
 				}
 			}

-			self.chain_sync.peer_disconnected(&peer_id);
+			if let Some(import_blocks_action) = self.chain_sync.peer_disconnected(&peer_id) {
+				self.import_blocks(import_blocks_action)
+			}
+
 			self.pending_responses.remove(&peer_id);
 			self.event_streams.retain(|stream| {
 				stream.unbounded_send(SyncEvent::PeerDisconnected(peer_id)).is_ok()
@@ -1181,10 +1204,14 @@ where
 				PeerRequest::Block(req) => {
 					match self.block_downloader.block_response_into_blocks(&req, resp) {
 						Ok(blocks) => {
-							if let Some((peer_id, new_req)) =
-								self.chain_sync.on_block_response(peer_id, req, blocks)
-							{
-								self.send_block_request(peer_id, new_req);
+							match self.chain_sync.on_block_response(peer_id, req, blocks) {
+								OnBlockResponse::SendBlockRequest { peer_id, request } =>
+									self.send_block_request(peer_id, request),
+								OnBlockResponse::ImportBlocks(import_blocks_action) =>
+									self.import_blocks(import_blocks_action),
+								OnBlockResponse::ImportJustifications(action) =>
+									self.import_justifications(action),
+								OnBlockResponse::Nothing => {},
 							}
 						},
 						Err(BlockResponseError::DecodeFailed(e)) => {
@@ -1230,7 +1257,11 @@ where
 						},
 					};

-					self.chain_sync.on_state_response(peer_id, response);
+					if let Some(import_blocks_action) =
+						self.chain_sync.on_state_response(peer_id, response)
+					{
+						self.import_blocks(import_blocks_action);
+					}
 				},
 				PeerRequest::WarpProof => {
 					self.chain_sync.on_warp_sync_response(peer_id, EncodedProof(resp));
@@ -1337,4 +1368,24 @@ where
 			},
 		}
 	}
+
+	/// Import blocks.
+	fn import_blocks(&mut self, ImportBlocksAction { origin, blocks }: ImportBlocksAction<B>) {
+		if let Some(metrics) = &self.metrics {
+			metrics.import_queue_blocks_submitted.inc();
+		}
+
+		self.import_queue.import_blocks(origin, blocks);
+	}
+
+	/// Import justifications.
+	fn import_justifications(&mut self, action: ImportJustificationsAction<B>) {
+		if let Some(metrics) = &self.metrics {
+			metrics.import_queue_justifications_submitted.inc();
+		}
+
+		let ImportJustificationsAction { peer_id, hash, number, justifications } = action;
+
+		self.import_queue.import_justifications(peer_id, hash, number, justifications);
+	}
 }