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>

client/api/src/backend.rs

@@ -149,6 +149,7 @@ pub trait BlockImportOperation<Block: BlockT> {
 		&mut self,
 		header: Block::Header,
 		body: Option<Vec<Block::Extrinsic>>,
+		indexed_body: Option<Vec<Vec<u8>>>,
 		justifications: Option<Justifications>,
 		state: NewBlockState,
 	) -> sp_blockchain::Result<()>;

client/api/src/in_mem.rs

@@ -572,6 +572,7 @@ impl<Block: BlockT> backend::BlockImportOperation<Block> for BlockImportOperatio
 		&mut self,
 		header: <Block as BlockT>::Header,
 		body: Option<Vec<<Block as BlockT>::Extrinsic>>,
+		_indexed_body: Option<Vec<Vec<u8>>>,
 		justifications: Option<Justifications>,
 		state: NewBlockState,
 	) -> sp_blockchain::Result<()> {

client/cli/src/arg_enums.rs

@@ -250,8 +250,14 @@ impl Into<sc_network::config::SyncMode> for SyncMode {
 	fn into(self) -> sc_network::config::SyncMode {
 		match self {
 			SyncMode::Full => sc_network::config::SyncMode::Full,
-			SyncMode::Fast => sc_network::config::SyncMode::Fast { skip_proofs: false },
-			SyncMode::FastUnsafe => sc_network::config::SyncMode::Fast { skip_proofs: true },
+			SyncMode::Fast => sc_network::config::SyncMode::Fast {
+				skip_proofs: false,
+				storage_chain_mode: false,
+			},
+			SyncMode::FastUnsafe => sc_network::config::SyncMode::Fast {
+				skip_proofs: true,
+				storage_chain_mode: false,
+			},
 		}
 	}
 }

client/db/src/changes_tries_storage.rs

@@ -572,7 +572,7 @@ mod tests {
 		};
 		let mut op = backend.begin_operation().unwrap();
 		backend.begin_state_operation(&mut op, block_id).unwrap();
-		op.set_block_data(header, None, None, NewBlockState::Best).unwrap();
+		op.set_block_data(header, None, None, None, NewBlockState::Best).unwrap();
 		op.update_changes_trie((changes_trie_update, ChangesTrieCacheAction::Clear)).unwrap();
 		backend.commit_operation(op).unwrap();
 
@@ -916,7 +916,7 @@ mod tests {
 		backend.begin_state_operation(&mut op, BlockId::Hash(block2)).unwrap();
 		op.mark_finalized(BlockId::Hash(block1), None).unwrap();
 		op.mark_finalized(BlockId::Hash(block2), None).unwrap();
-		op.set_block_data(header3, None, None, NewBlockState::Final).unwrap();
+		op.set_block_data(header3, None, None, None, NewBlockState::Final).unwrap();
 		backend.commit_operation(op).unwrap();
 
 		// insert more unfinalized headers
@@ -941,7 +941,7 @@ mod tests {
 		op.mark_finalized(BlockId::Hash(block4), None).unwrap();
 		op.mark_finalized(BlockId::Hash(block5), None).unwrap();
 		op.mark_finalized(BlockId::Hash(block6), None).unwrap();
-		op.set_block_data(header7, None, None, NewBlockState::Final).unwrap();
+		op.set_block_data(header7, None, None, None, NewBlockState::Final).unwrap();
 		backend.commit_operation(op).unwrap();
 	}
 

client/db/src/lib.rs

@@ -74,6 +74,7 @@ use sp_arithmetic::traits::Saturating;
 use sp_runtime::{generic::{DigestItem, BlockId}, Justification, Justifications, Storage};
 use sp_runtime::traits::{
 	Block as BlockT, Header as HeaderT, NumberFor, Zero, One, SaturatedConversion, HashFor,
+	Hash,
 };
 use sp_state_machine::{
 	DBValue, ChangesTrieTransaction, ChangesTrieCacheAction, UsageInfo as StateUsageInfo,
@@ -384,6 +385,7 @@ struct PendingBlock<Block: BlockT> {
 	header: Block::Header,
 	justifications: Option<Justifications>,
 	body: Option<Vec<Block::Extrinsic>>,
+	indexed_body: Option<Vec<Vec<u8>>>,
 	leaf_state: NewBlockState,
 }
 
@@ -824,6 +826,7 @@ impl<Block: BlockT> sc_client_api::backend::BlockImportOperation<Block> for Bloc
 		&mut self,
 		header: Block::Header,
 		body: Option<Vec<Block::Extrinsic>>,
+		indexed_body: Option<Vec<Vec<u8>>>,
 		justifications: Option<Justifications>,
 		leaf_state: NewBlockState,
 	) -> ClientResult<()> {
@@ -834,6 +837,7 @@ impl<Block: BlockT> sc_client_api::backend::BlockImportOperation<Block> for Bloc
 		self.pending_block = Some(PendingBlock {
 			header,
 			body,
+			indexed_body,
 			justifications,
 			leaf_state,
 		});
@@ -1068,7 +1072,7 @@ impl<Block: BlockT> Backend<Block> {
 
 	/// Create new memory-backed client backend for tests.
 	#[cfg(any(test, feature = "test-helpers"))]
-	fn new_test_with_tx_storage(
+	pub fn new_test_with_tx_storage(
 		keep_blocks: u32,
 		canonicalization_delay: u64,
 		transaction_storage: TransactionStorageMode,
@@ -1393,6 +1397,16 @@ impl<Block: BlockT> Backend<Block> {
 					},
 				}
 			}
+			if let Some(body) = pending_block.indexed_body {
+				match self.transaction_storage {
+					TransactionStorageMode::BlockBody => {
+						debug!(target: "db", "Commit: ignored indexed block body");
+					},
+					TransactionStorageMode::StorageChain => {
+						apply_indexed_body::<Block>(&mut transaction, body);
+					},
+				}
+			}
 			if let Some(justifications) = pending_block.justifications {
 				transaction.set_from_vec(columns::JUSTIFICATIONS, &lookup_key, justifications.encode());
 			}
@@ -1881,6 +1895,20 @@ fn apply_index_ops<Block: BlockT>(
 	extrinsic_headers.encode()
 }
 
+fn apply_indexed_body<Block: BlockT>(
+	transaction: &mut Transaction<DbHash>,
+	body: Vec<Vec<u8>>,
+) {
+	for extrinsic in body {
+		let hash = sp_runtime::traits::BlakeTwo256::hash(&extrinsic);
+		transaction.store(
+			columns::TRANSACTION,
+			DbHash::from_slice(hash.as_ref()),
+			extrinsic,
+		);
+	}
+}
+
 impl<Block> sc_client_api::backend::AuxStore for Backend<Block> where Block: BlockT {
 	fn insert_aux<
 		'a,
@@ -2439,7 +2467,7 @@ pub(crate) mod tests {
 		};
 		let mut op = backend.begin_operation().unwrap();
 		backend.begin_state_operation(&mut op, block_id).unwrap();
-		op.set_block_data(header, Some(body), None, NewBlockState::Best).unwrap();
+		op.set_block_data(header, Some(body), None, None, NewBlockState::Best).unwrap();
 		if let Some(index) = transaction_index {
 			op.update_transaction_index(index).unwrap();
 		}
@@ -2481,6 +2509,7 @@ pub(crate) mod tests {
 						header,
 						Some(vec![]),
 						None,
+						None,
 						NewBlockState::Best,
 					).unwrap();
 					db.commit_operation(op).unwrap();
@@ -2537,6 +2566,7 @@ pub(crate) mod tests {
 				header.clone(),
 				Some(vec![]),
 				None,
+				None,
 				NewBlockState::Best,
 			).unwrap();
 
@@ -2579,6 +2609,7 @@ pub(crate) mod tests {
 				header,
 				Some(vec![]),
 				None,
+				None,
 				NewBlockState::Best,
 			).unwrap();
 
@@ -2622,6 +2653,7 @@ pub(crate) mod tests {
 				header,
 				Some(vec![]),
 				None,
+				None,
 				NewBlockState::Best,
 			).unwrap();
 
@@ -2659,6 +2691,7 @@ pub(crate) mod tests {
 				header,
 				Some(vec![]),
 				None,
+				None,
 				NewBlockState::Best,
 			).unwrap();
 
@@ -2695,6 +2728,7 @@ pub(crate) mod tests {
 				header,
 				Some(vec![]),
 				None,
+				None,
 				NewBlockState::Best,
 			).unwrap();
 
@@ -2730,6 +2764,7 @@ pub(crate) mod tests {
 				header,
 				Some(vec![]),
 				None,
+				None,
 				NewBlockState::Best,
 			).unwrap();
 
@@ -3067,6 +3102,7 @@ pub(crate) mod tests {
 				header.clone(),
 				Some(vec![]),
 				None,
+				None,
 				NewBlockState::Best,
 			).unwrap();
 
@@ -3106,6 +3142,7 @@ pub(crate) mod tests {
 				header,
 				Some(vec![]),
 				None,
+				None,
 				NewBlockState::Normal,
 			).unwrap();
 
@@ -3118,7 +3155,7 @@ pub(crate) mod tests {
 			let header = backend.blockchain().header(BlockId::Hash(hash1)).unwrap().unwrap();
 			let mut op = backend.begin_operation().unwrap();
 			backend.begin_state_operation(&mut op, BlockId::Hash(hash0)).unwrap();
-			op.set_block_data(header, None, None, NewBlockState::Best).unwrap();
+			op.set_block_data(header, None, None, None, NewBlockState::Best).unwrap();
 			backend.commit_operation(op).unwrap();
 		}
 

client/light/src/backend.rs

@@ -293,6 +293,7 @@ impl<S, Block> BlockImportOperation<Block> for ImportOperation<Block, S>
 		&mut self,
 		header: Block::Header,
 		_body: Option<Vec<Block::Extrinsic>>,
+		_indexed_body: Option<Vec<Vec<u8>>>,
 		_justifications: Option<Justifications>,
 		state: NewBlockState,
 	) -> ClientResult<()> {

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();

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,
 	},
 }
 

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;
 

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,

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.