Block packet size limit (#6398)

* Block packet size limit

* Update client/network/src/protocol.rs

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

* Add block response limit

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

substrate/client/network/src/block_requests.rs

@@ -119,6 +119,7 @@ pub enum Event<B: Block> {
 #[derive(Debug, Clone)]
 pub struct Config {
 	max_block_data_response: u32,
+	max_block_body_bytes: usize,
 	max_request_len: usize,
 	max_response_len: usize,
 	inactivity_timeout: Duration,
@@ -137,6 +138,7 @@ impl Config {
 	pub fn new(id: &ProtocolId) -> Self {
 		let mut c = Config {
 			max_block_data_response: 128,
+			max_block_body_bytes: 8 * 1024 * 1024,
 			max_request_len: 1024 * 1024,
 			max_response_len: 16 * 1024 * 1024,
 			inactivity_timeout: Duration::from_secs(15),
@@ -171,6 +173,15 @@ impl Config {
 		self
 	}
 
+	/// Set the maximum total bytes of block bodies that are send in the response.
+	/// Note that at least one block is always sent regardless of the limit.
+	/// This should be lower than the value specified in `set_max_response_len`
+	/// accounting for headers, justifications and encoding overhead.
+	pub fn set_max_block_body_bytes(&mut self, v: usize) -> &mut Self {
+		self.max_block_body_bytes = v;
+		self
+	}
+
 	/// Set protocol to use for upgrade negotiation.
 	pub fn set_protocol(&mut self, id: &ProtocolId) -> &mut Self {
 		let mut v = Vec::new();
@@ -385,8 +396,11 @@ where
 
 		let mut blocks = Vec::new();
 		let mut block_id = from_block_id;
+		let mut total_size = 0;
 		while let Some(header) = self.chain.header(block_id).unwrap_or(None) {
-			if blocks.len() >= max_blocks as usize {
+			if blocks.len() >= max_blocks as usize
+				|| (blocks.len() >= 1 && total_size > self.config.max_block_body_bytes)
+			{
 				break
 			}
 
@@ -400,6 +414,20 @@ where
 			};
 			let is_empty_justification = justification.as_ref().map(|j| j.is_empty()).unwrap_or(false);
 
+			let body = if get_body {
+				match self.chain.block_body(&BlockId::Hash(hash))? {
+					Some(mut extrinsics) => extrinsics.iter_mut()
+						.map(|extrinsic| extrinsic.encode())
+						.collect(),
+					None => {
+						log::trace!(target: "sync", "Missing data for block request.");
+						break;
+					}
+				}
+			} else {
+				Vec::new()
+			};
+
 			let block_data = schema::v1::BlockData {
 				hash: hash.encode(),
 				header: if get_header {
@@ -407,21 +435,14 @@ where
 				} else {
 					Vec::new()
 				},
-				body: if get_body {
-					self.chain.block_body(&BlockId::Hash(hash))?
-						.unwrap_or_default()
-						.iter_mut()
-						.map(|extrinsic| extrinsic.encode())
-						.collect()
-				} else {
-					Vec::new()
-				},
+				body,
 				receipt: Vec::new(),
 				message_queue: Vec::new(),
 				justification: justification.unwrap_or_default(),
 				is_empty_justification,
 			};
 
+			total_size += block_data.body.len();
 			blocks.push(block_data);
 
 			match direction {

substrate/client/network/src/protocol.rs

@@ -95,6 +95,9 @@ pub(crate) const MIN_VERSION: u32 = 3;
 
 // Maximum allowed entries in `BlockResponse`
 const MAX_BLOCK_DATA_RESPONSE: u32 = 128;
+// Maximum total bytes allowed for block bodies in `BlockResponse`
+const MAX_BODIES_BYTES: usize = 8 * 1024 * 1024;
+
 /// When light node connects to the full node and the full node is behind light node
 /// for at least `LIGHT_MAXIMAL_BLOCKS_DIFFERENCE` blocks, we consider it not useful
 /// and disconnect to free connection slot.
@@ -756,8 +759,9 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 		let get_justification = request
 			.fields
 			.contains(message::BlockAttributes::JUSTIFICATION);
+		let mut total_size = 0;
 		while let Some(header) = self.context_data.chain.header(id).unwrap_or(None) {
-			if blocks.len() >= max {
+			if blocks.len() >= max || (blocks.len() >= 1 && total_size > MAX_BODIES_BYTES) {
 				break;
 			}
 			let number = *header.number();
@@ -788,6 +792,7 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 				trace!(target: "sync", "Missing data for block request.");
 				break;
 			}
+			total_size += block_data.body.as_ref().map_or(0, |b| b.len());
 			blocks.push(block_data);
 			match request.direction {
 				message::Direction::Ascending => id = BlockId::Number(number + One::one()),