pink: Support for batch HTTP request

crates/phactory/src/contracts/pink.rs

@@ -12,7 +12,10 @@ use phala_types::contract::messaging::ResourceType;
 use pink::{
     capi::v1::{
         ecall::{ECalls, ECallsRo},
-        ocall::{ExecContext, HttpRequest, HttpRequestError, HttpResponse, OCalls, StorageChanges},
+        ocall::{
+            BatchHttpResult, ExecContext, HttpRequest, HttpRequestError, HttpResponse, OCalls,
+            StorageChanges,
+        },
     },
     local_cache::{self, StorageQuotaExceeded},
     runtimes::v1::{get_runtime, using_ocalls},
@@ -372,6 +375,29 @@ impl OCalls for RuntimeHandle<'_> {
         }
         result
     }
+
+    fn batch_http_request(
+        &self,
+        contract: AccountId,
+        requests: Vec<HttpRequest>,
+        timeout_ms: u64,
+    ) -> BatchHttpResult {
+        let results = pink_extension_runtime::batch_http_request(
+            requests,
+            context::time_remaining().min(timeout_ms),
+        )?;
+        for result in &results {
+            match result {
+                Ok(r) => {
+                    http_counters::add(contract.clone(), r.status_code);
+                }
+                Err(_) => {
+                    http_counters::add(contract.clone(), 0);
+                }
+            }
+        }
+        Ok(results)
+    }
 }
 
 impl OCalls for RuntimeHandleMut<'_> {
@@ -436,6 +462,16 @@ impl OCalls for RuntimeHandleMut<'_> {
     ) -> Result<HttpResponse, HttpRequestError> {
         self.readonly().http_request(contract, request)
     }
+
+    fn batch_http_request(
+        &self,
+        contract: AccountId,
+        requests: Vec<HttpRequest>,
+        timeout_ms: u64,
+    ) -> BatchHttpResult {
+        self.readonly()
+            .batch_http_request(contract, requests, timeout_ms)
+    }
 }
 
 impl v1::CrossCall for RuntimeHandle<'_> {

crates/pink/capi/src/v1/mod.rs

@@ -127,7 +127,7 @@ pub mod ocall {
     use scale::{Decode, Encode};
 
     pub use pink_extension::chain_extension::{
-        HttpRequest, HttpRequestError, HttpResponse, StorageQuotaExceeded,
+        BatchHttpResult, HttpRequest, HttpRequestError, HttpResponse, StorageQuotaExceeded,
     };
     pub type StorageChanges = Vec<(Vec<u8>, (Vec<u8>, i32))>;
 
@@ -192,5 +192,12 @@ pub mod ocall {
             contract: AccountId,
             request: HttpRequest,
         ) -> Result<HttpResponse, HttpRequestError>;
+        #[xcall(id = 15)]
+        fn batch_http_request(
+            &self,
+            contract: AccountId,
+            requests: Vec<HttpRequest>,
+            timeout_ms: u64,
+        ) -> BatchHttpResult;
     }
 }

crates/pink/pink-extension-runtime/Cargo.toml

@@ -11,9 +11,11 @@ pink-extension = { version = "0.4.2", path = "../pink-extension" }
 reqwest-env-proxy = { version = "0.1", path = "../../reqwest-env-proxy" }
 sp-core = { version = "7", features = ["full_crypto"] }
 sp-runtime-interface = { version = "7", features = ["disable_target_static_assertions"] }
-reqwest = { version = "0.11", default-features = false, features = ["rustls-tls", "socks", "blocking"] }
+reqwest = { version = "0.11", default-features = false, features = ["rustls-tls", "socks"] }
 log = "0.4"
 ring = "0.16"
 getrandom = "0.2"
 once_cell = "1.10.0"
 hex_fmt = "0.3.0"
+futures = "0.3"
+tokio = { version = "1", features = ["full"] }

crates/pink/pink-extension-runtime/src/lib.rs

@@ -1,4 +1,5 @@
 use std::borrow::Cow;
+use std::io::Write;
 use std::{
     fmt::Display,
     str::FromStr,
@@ -42,16 +43,43 @@ impl<'a, T, E> DefaultPinkExtension<'a, T, E> {
     }
 }
 
+fn block_on<F: core::future::Future>(f: F) -> F::Output {
+    match tokio::runtime::Handle::try_current() {
+        Ok(handle) => handle.block_on(f),
+        Err(_) => tokio::runtime::Runtime::new()
+            .expect("Failed to create tokio runtime")
+            .block_on(f),
+    }
+}
+
+pub fn batch_http_request(requests: Vec<HttpRequest>, timeout_ms: u64) -> ext::BatchHttpResult {
+    const MAX_CONCURRENT_REQUESTS: usize = 5;
+    if requests.len() > MAX_CONCURRENT_REQUESTS {
+        return Err(ext::HttpRequestError::TooManyRequests);
+    }
+    let futs = requests
+        .into_iter()
+        .map(|request| async_http_request(request, timeout_ms));
+    Ok(block_on(futures::future::join_all(futs)))
+}
+
 pub fn http_request(
     request: HttpRequest,
     timeout_ms: u64,
+) -> Result<HttpResponse, HttpRequestError> {
+    block_on(async_http_request(request, timeout_ms))
+}
+
+async fn async_http_request(
+    request: HttpRequest,
+    timeout_ms: u64,
 ) -> Result<HttpResponse, HttpRequestError> {
     if timeout_ms == 0 {
         return Err(HttpRequestError::Timeout);
     }
     let timeout = Duration::from_millis(timeout_ms);
     let url: reqwest::Url = request.url.parse().or(Err(HttpRequestError::InvalidUrl))?;
-    let client = reqwest::blocking::Client::builder()
+    let client = reqwest::Client::builder()
         .timeout(timeout)
         .env_proxy(url.host_str().unwrap_or_default())
         .build()
@@ -71,7 +99,8 @@ pub fn http_request(
         .request(method, url)
         .headers(headers)
         .body(request.body)
-        .send();
+        .send()
+        .await;
 
     let mut response = match result {
         Ok(response) => response,
@@ -93,20 +122,20 @@ pub fn http_request(
         .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or_default().into()))
         .collect();
 
-    const MAX_BODY_SIZE: usize = 1024 * 16; // 16KB
+    const MAX_BODY_SIZE: usize = 1024 * 1024 * 2; // 2MB
 
     let mut body = Vec::new();
     let mut writer = LimitedWriter::new(&mut body, MAX_BODY_SIZE);
 
-    if let Err(err) = response.copy_to(&mut writer) {
-        log::info!("Failed to read HTTP body: {err}");
-        return Ok(HttpResponse {
-            status_code: 524,
-            reason_phrase: "IO Error".into(),
-            body: format!("{err:?}").into_bytes(),
-            headers: vec![],
-        });
-    };
+    while let Some(chunk) = response
+        .chunk()
+        .await
+        .or(Err(HttpRequestError::NetworkError))?
+    {
+        writer
+            .write_all(&chunk)
+            .or(Err(HttpRequestError::ResponseTooLarge))?;
+    }
 
     let response = HttpResponse {
         status_code: response.status().as_u16(),
@@ -127,6 +156,14 @@ impl<T: PinkRuntimeEnv, E: From<&'static str>> PinkExtBackend for DefaultPinkExt
         http_request(request, 10 * 1000).map_err(|err| err.display().into())
     }
 
+    fn batch_http_request(
+        &self,
+        requests: Vec<HttpRequest>,
+        timeout_ms: u64,
+    ) -> Result<ext::BatchHttpResult, Self::Error> {
+        Ok(batch_http_request(requests, timeout_ms))
+    }
+
     fn sign(
         &self,
         sigtype: SigType,

crates/pink/pink-extension-runtime/src/mock_ext.rs

@@ -26,6 +26,14 @@ impl ext::PinkExtBackend for MockExtension {
         super::DefaultPinkExtension::new(self).http_request(request)
     }
 
+    fn batch_http_request(
+        &self,
+        requests: Vec<ext::HttpRequest>,
+        timeout_ms: u64,
+    ) -> Result<ext::BatchHttpResult, Self::Error> {
+        super::DefaultPinkExtension::new(self).batch_http_request(requests, timeout_ms)
+    }
+
     fn sign(
         &self,
         sigtype: SigType,

crates/pink/pink-extension/Cargo.toml

@@ -14,6 +14,7 @@ pink-extension-macro = { version = "0.4.2", path = "./macro" }
 log = "0.4.17"
 dlmalloc = { version = "0.2.4", default-features = false, features = ["global"], optional = true }
 this-crate = { version = "0.1", path = "../../this-crate" }
+num_enum = { version = "0.6", default-features = false }
 
 [dev-dependencies]
 insta = "1.7.2"
@@ -24,6 +25,7 @@ std = [
     "ink/std",
     "scale/std",
     "scale-info/std",
+    "num_enum/std",
 ]
 runtime_utils = ["std"]
 dlmalloc = ["ink/no-allocator", "dep:dlmalloc"]

crates/pink/pink-extension/src/chain_extension.rs

@@ -2,7 +2,7 @@ use alloc::borrow::Cow;
 use alloc::vec::Vec;
 use ink::ChainExtensionInstance;
 
-pub use http_request::{HttpRequest, HttpResponse, HttpRequestError};
+pub use http_request::{HttpRequest, HttpRequestError, HttpResponse};
 pub use ink::primitives::AccountId;
 pub use signing::SigType;
 
@@ -84,7 +84,7 @@ impl<T: scale::Encode> EncodeOutputFallbask for EncodeOutput<T> {
     }
 }
 
-#[derive(scale::Encode, scale::Decode)]
+#[derive(scale::Encode, scale::Decode, Debug)]
 #[cfg_attr(feature = "std", derive(scale_info::TypeInfo))]
 pub struct ErrorCode(u32);
 
@@ -97,6 +97,8 @@ impl ink::env::chain_extension::FromStatusCode for ErrorCode {
     }
 }
 
+pub type BatchHttpResult = Result<Vec<Result<HttpResponse, HttpRequestError>>, HttpRequestError>;
+
 /// Extensions for the ink runtime defined by phat contract.
 #[pink_extension_macro::chain_extension]
 pub trait PinkExt {
@@ -199,6 +201,10 @@ pub trait PinkExt {
     /// Get the version of the current contract runtime in this cluster.
     #[ink(extension = 21, handle_status = false)]
     fn runtime_version() -> (u32, u32);
+
+    /// Batch http request
+    #[ink(extension = 22, handle_status = true)]
+    fn batch_http_request(requests: Vec<HttpRequest>, timeout_ms: u64) -> BatchHttpResult;
 }
 
 pub fn pink_extension_instance() -> <PinkExt as ChainExtensionInstance>::Instance {