Compress state response message in state syncing

substrate/client/network/sync/Cargo.toml

@@ -34,6 +34,7 @@ smallvec = { workspace = true, default-features = true }
 thiserror = { workspace = true }
 tokio-stream = { workspace = true }
 tokio = { features = ["macros", "time"], workspace = true, default-features = true }
+zstd = { workspace = true }
 fork-tree = { workspace = true, default-features = true }
 prometheus-endpoint = { workspace = true, default-features = true }
 sc-client-api = { workspace = true, default-features = true }

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

@@ -1207,8 +1207,10 @@ where
 		Ok(request.encode_to_vec())
 	}
 
-	fn decode_state_response(response: &[u8]) -> Result<OpaqueStateResponse, String> {
-		let response = StateResponse::decode(response)
+	fn decode_state_response(compressed_response: &[u8]) -> Result<OpaqueStateResponse, String> {
+		let response = zstd::stream::decode_all(compressed_response)
+			.map_err(|error| format!("Failed to decompress state response: {error}"))?;
+		let response = StateResponse::decode(response.as_slice())
 			.map_err(|error| format!("Failed to decode state response: {error}"))?;
 
 		Ok(OpaqueStateResponse(Box::new(response)))

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

@@ -264,7 +264,11 @@ where
 
 			let mut data = Vec::with_capacity(response.encoded_len());
 			response.encode(&mut data)?;
-			Ok(data)
+
+			let compressed_response = zstd::stream::encode_all(data.as_slice(), 0)
+				.map_err(HandleRequestError::Compress)?;
+
+			Ok(compressed_response)
 		} else {
 			Err(())
 		};
@@ -291,4 +295,7 @@ enum HandleRequestError {
 
 	#[error("Failed to send response.")]
 	SendResponse,
+
+	#[error("Failed to compress response: {0}.")]
+	Compress(std::io::Error),
 }