Storage chains sync (#9171)

* Sync storage chains

* Test

* Apply suggestions from code review

Co-authored-by: cheme <emericchevalier.pro@gmail.com>

* Separate block body and indexed body

* Update client/db/src/lib.rs

Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

Co-authored-by: cheme <emericchevalier.pro@gmail.com>
Co-authored-by: Shawn Tabrizi <shawntabrizi@gmail.com>
Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

substrate/client/network/src/block_request_handler.rs

@@ -264,6 +264,7 @@ impl<B: BlockT> BlockRequestHandler<B> {
 	) -> Result<BlockResponse, HandleRequestError> {
 		let get_header = attributes.contains(BlockAttributes::HEADER);
 		let get_body = attributes.contains(BlockAttributes::BODY);
+		let get_indexed_body = attributes.contains(BlockAttributes::INDEXED_BODY);
 		let get_justification = attributes.contains(BlockAttributes::JUSTIFICATION);
 
 		let mut blocks = Vec::new();
@@ -321,6 +322,18 @@ impl<B: BlockT> BlockRequestHandler<B> {
 				Vec::new()
 			};
 
+			let indexed_body =  if get_indexed_body {
+				match self.client.block_indexed_body(&BlockId::Hash(hash))? {
+					Some(transactions) => transactions,
+					None => {
+						log::trace!(target: LOG_TARGET, "Missing indexed block data for block request.");
+						break;
+					}
+				}
+			} else {
+				Vec::new()
+			};
+
 			let block_data = crate::schema::v1::BlockData {
 				hash: hash.encode(),
 				header: if get_header {
@@ -334,6 +347,7 @@ impl<B: BlockT> BlockRequestHandler<B> {
 				justification,
 				is_empty_justification,
 				justifications,
+				indexed_body,
 			};
 
 			total_size += block_data.body.len();

substrate/client/network/src/config.rs

@@ -390,7 +390,9 @@ pub enum SyncMode {
 	/// Download blocks and the latest state.
 	Fast {
 		/// Skip state proof download and verification.
-		skip_proofs: bool
+		skip_proofs: bool,
+		/// Download indexed transactions for recent blocks.
+		storage_chain_mode: bool,
 	},
 }
 

substrate/client/network/src/lib.rs

@@ -263,7 +263,6 @@ pub mod light_client_requests;
 pub mod state_request_handler;
 pub mod config;
 pub mod error;
-pub mod gossip;
 pub mod network_state;
 pub mod transactions;
 

substrate/client/network/src/protocol.rs

@@ -228,7 +228,13 @@ impl ProtocolConfig {
 		} else {
 			match self.sync_mode {
 				config::SyncMode::Full => sync::SyncMode::Full,
-				config::SyncMode::Fast { skip_proofs } => sync::SyncMode::LightState { skip_proofs },
+				config::SyncMode::Fast {
+					skip_proofs,
+					storage_chain_mode,
+				} => sync::SyncMode::LightState {
+					skip_proofs,
+					storage_chain_mode
+				},
 			}
 		}
 	}
@@ -597,6 +603,11 @@ impl<B: BlockT> Protocol<B> {
 				} else {
 					None
 				},
+				indexed_body: if request.fields.contains(message::BlockAttributes::INDEXED_BODY) {
+					Some(block_data.indexed_body)
+				} else {
+					None
+				},
 				receipt: if !block_data.message_queue.is_empty() {
 					Some(block_data.receipt)
 				} else {
@@ -965,6 +976,7 @@ impl<B: BlockT> Protocol<B> {
 						hash: header.hash(),
 						header: Some(header),
 						body: None,
+						indexed_body: None,
 						receipt: None,
 						message_queue: None,
 						justification: None,

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

@@ -77,6 +77,8 @@ bitflags! {
 		const MESSAGE_QUEUE = 0b00001000;
 		/// Include a justification for the block.
 		const JUSTIFICATION = 0b00010000;
+		/// Include indexed transactions for a block.
+		const INDEXED_BODY = 0b00100000;
 	}
 }
 
@@ -248,6 +250,8 @@ pub mod generic {
 		pub header: Option<Header>,
 		/// Block body if requested.
 		pub body: Option<Vec<Extrinsic>>,
+		/// Block body indexed transactions if requested.
+		pub indexed_body: Option<Vec<Vec<u8>>>,
 		/// Block receipt if requested.
 		pub receipt: Option<Vec<u8>>,
 		/// Block message queue if requested.

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

@@ -469,7 +469,8 @@ pub enum SyncMode {
 	Full,
 	// Sync headers and the last finalied state
 	LightState {
-		skip_proofs: bool
+		storage_chain_mode: bool,
+		skip_proofs: bool,
 	},
 }
 
@@ -518,8 +519,10 @@ impl<B: BlockT> ChainSync<B> {
 		match self.mode {
 			SyncMode::Full => BlockAttributes::HEADER | BlockAttributes::JUSTIFICATION | BlockAttributes::BODY,
 			SyncMode::Light => BlockAttributes::HEADER | BlockAttributes::JUSTIFICATION,
-			SyncMode::LightState { .. } =>
+			SyncMode::LightState { storage_chain_mode: false, .. } =>
 				BlockAttributes::HEADER | BlockAttributes::JUSTIFICATION | BlockAttributes::BODY,
+			SyncMode::LightState { storage_chain_mode: true, .. } =>
+				BlockAttributes::HEADER | BlockAttributes::JUSTIFICATION | BlockAttributes::INDEXED_BODY,
 		}
 	}
 
@@ -914,25 +917,7 @@ impl<B: BlockT> ChainSync<B> {
 							peer.state = PeerSyncState::Available;
 							validate_blocks::<B>(&blocks, who, Some(request))?;
 							self.blocks.insert(start_block, blocks, who.clone());
-							self.blocks
-								.drain(self.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,
-										justifications,
-										origin: block_data.origin,
-										allow_missing_state: true,
-										import_existing: self.import_existing,
-										skip_execution: self.skip_execution(),
-										state: None,
-									}
-								}).collect()
+							self.drain_blocks()
 						}
 						PeerSyncState::DownloadingStale(_) => {
 							peer.state = PeerSyncState::Available;
@@ -949,6 +934,7 @@ impl<B: BlockT> ChainSync<B> {
 									hash: b.hash,
 									header: b.header,
 									body: b.body,
+									indexed_body: None,
 									justifications,
 									origin: Some(who.clone()),
 									allow_missing_state: true,
@@ -1064,6 +1050,7 @@ impl<B: BlockT> ChainSync<B> {
 							hash: b.hash,
 							header: b.header,
 							body: b.body,
+							indexed_body: None,
 							justifications,
 							origin: Some(who.clone()),
 							allow_missing_state: true,
@@ -1115,6 +1102,7 @@ impl<B: BlockT> ChainSync<B> {
 					hash,
 					header: Some(header),
 					body: None,
+					indexed_body: None,
 					justifications: None,
 					origin: None,
 					allow_missing_state: true,
@@ -1367,7 +1355,7 @@ impl<B: BlockT> ChainSync<B> {
 			is_descendent_of(&**client, base, block)
 		});
 
-		if let SyncMode::LightState { skip_proofs } = &self.mode {
+		if let SyncMode::LightState { skip_proofs, .. } = &self.mode {
 			if self.state_sync.is_none()
 				&& !self.peers.is_empty()
 				&& self.queue_blocks.is_empty()
@@ -1757,24 +1745,7 @@ impl<B: BlockT> ChainSync<B> {
 			target.peers.remove(who);
 			!target.peers.is_empty()
 		});
-		let blocks: Vec<_> = self.blocks
-			.drain(self.best_queued_number + One::one())
-			.into_iter()
-			.map(|block_data| {
-				let justifications =
-					legacy_justification_mapping(block_data.block.justification);
-				IncomingBlock {
-					hash: block_data.block.hash,
-					header: block_data.block.header,
-					body: block_data.block.body,
-					justifications,
-					origin: block_data.origin,
-					allow_missing_state: true,
-					import_existing: false,
-					skip_execution: self.skip_execution(),
-					state: None,
-				}
-			}).collect();
+		let blocks = self.drain_blocks();
 		if !blocks.is_empty() {
 			Some(self.validate_and_queue_blocks(blocks))
 		} else {
@@ -1878,6 +1849,31 @@ impl<B: BlockT> ChainSync<B> {
 			_priv: ()
 		}
 	}
+
+	/// Drain the downloaded block set up to the first gap.
+	fn drain_blocks(&mut self) -> Vec<IncomingBlock<B>> {
+		self.blocks
+			.drain(self.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: self.skip_execution(),
+					state: None,
+				}
+			}).collect()
+	}
+
 }
 
 // This is purely during a backwards compatible transitionary period and should be removed
@@ -2383,6 +2379,7 @@ mod test {
 					hash: b.hash(),
 					header: Some(b.header().clone()),
 					body: Some(b.deconstruct().1),
+					indexed_body: None,
 					receipt: None,
 					message_queue: None,
 					justification: None,

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

@@ -225,6 +225,7 @@ mod test {
 			hash: H256::random(),
 			header: None,
 			body: None,
+			indexed_body: None,
 			message_queue: None,
 			receipt: None,
 			justification: None,

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

@@ -66,6 +66,8 @@ message BlockData {
 	// is because empty justifications, like all justifications, are paired with a non-empty
 	// consensus engine ID.
 	bytes justifications = 8; // optional
+	// Indexed block body if requestd.
+	repeated bytes indexed_body = 9; // optional
 }
 
 // Request storage data from a peer.