Update Hyper dependency (#33)

chore: Update hyper dependency to 1.1

Co-authored-by: sinu <65924192+sinui0@users.noreply.github.com>

wasm/prover/Cargo.toml

@@ -12,42 +12,44 @@ crate-type = ["cdylib", "rlib"]
 
 [dependencies]
 chrono = "0.4"
-elliptic-curve = {version = "0.13.5", features = ["pkcs8"]}
+elliptic-curve = { version = "0.13.5", features = ["pkcs8"] }
 futures = "0.3"
 futures-util = "0.3.28"
-getrandom = {version = "0.2", features = ["js"]}
+getrandom = { version = "0.2", features = ["js"] }
 js-sys = "0.3.64"
-p256 = {version = "0.13", features = ["pem", "ecdsa"]}
+p256 = { version = "0.13", features = ["pem", "ecdsa"] }
 rayon = "1.5"
-serde = {version = "1.0.147", features = ["derive"]}
+serde = { version = "1.0.147", features = ["derive"] }
 serde-wasm-bindgen = "0.6.1"
 serde_json = "1.0"
-tokio-util = "0.7"
 tracing = "0.1"
-url = {version = "2.0", features = ["serde"]}
+url = { version = "2.0", features = ["serde"] }
 wasm-bindgen = "0.2.87"
 wasm-bindgen-futures = "0.4.37"
 wasm-bindgen-rayon = "1.0"
+pin-project-lite = "0.2.4"
 
-hyper = {version = "0.14", features = ["client", "http1"]}
-tracing-subscriber = {version = "0.3", features = ["time"]}
+http-body-util = "0.1"
+hyper = { version = "1.1", features = ["client", "http1"] }
+hyper-util = { version = "0.1", features = ["http1"] }
+tracing-subscriber = { version = "0.3", features = ["time"] }
 tracing-web = "0.1.2"
 
-ring = {version = "0.17", features = ["wasm32_unknown_unknown_js"]}
+ring = { version = "0.17", features = ["wasm32_unknown_unknown_js"] }
 
 # time crate: https://crates.io/crates/time
 # NOTE: It is required, otherwise "time not implemented on this platform" error happens right after "!@# 2".
 # Probably due to tokio's time feature is used in tlsn-prover?
-time = {version = "0.3.34", features = ["wasm-bindgen"]}
+time = { version = "0.3.34", features = ["wasm-bindgen"] }
 # Used to calculate elapsed time.
 web-time = "1.0"
 
-tlsn-core = {git = "https://github.com/tlsnotary/tlsn.git", tag = "v0.1.0-alpha.4", package = "tlsn-core"}
-tlsn-prover = {git = "https://github.com/tlsnotary/tlsn.git", tag = "v0.1.0-alpha.4", package = "tlsn-prover", features = [
+tlsn-core = { git = "https://github.com/tlsnotary/tlsn.git", tag = "v0.1.0-alpha.4", package = "tlsn-core" }
+tlsn-prover = { git = "https://github.com/tlsnotary/tlsn.git", tag = "v0.1.0-alpha.4", package = "tlsn-prover", features = [
   "tracing",
-]}
+] }
 
-web-sys = {version = "0.3.4", features = [
+web-sys = { version = "0.3.4", features = [
   "BinaryType",
   "Blob",
   "ErrorEvent",
@@ -66,25 +68,25 @@ web-sys = {version = "0.3.4", features = [
   'RequestInit',
   'RequestMode',
   'Response',
-]}
+] }
 
 # Use the patched ws_stream_wasm to fix the issue https://github.com/najamelan/ws_stream_wasm/issues/12#issuecomment-1711902958
-ws_stream_wasm = {version = "0.7.4", git = "https://github.com/tlsnotary/ws_stream_wasm", rev = "2ed12aad9f0236e5321f577672f309920b2aef51"}
+ws_stream_wasm = { version = "0.7.4", git = "https://github.com/tlsnotary/ws_stream_wasm", rev = "2ed12aad9f0236e5321f577672f309920b2aef51" }
 
 # The `console_error_panic_hook` crate provides better debugging of panics by
 # logging them with `console.error`. This is great for development, but requires
 # all the `std::fmt` and `std::panicking` infrastructure, so isn't great for
 # code size when deploying.
-console_error_panic_hook = {version = "0.1.7"}
+console_error_panic_hook = { version = "0.1.7" }
 
-strum = {version = "0.26.1"}
+strum = { version = "0.26.1" }
 strum_macros = "0.26.1"
 
 [dev-dependencies]
 wasm-bindgen-test = "0.3.34"
 
 [profile.release]
-lto = true # Enable Link Time Optimization
+lto = true      # Enable Link Time Optimization
 opt-level = "z" # Optimize for size
 
 [package.metadata.wasm-pack.profile.release]

wasm/prover/src/hyper_io.rs

@@ -0,0 +1,88 @@
+use core::slice;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use pin_project_lite::pin_project;
+
+pin_project! {
+    #[derive(Debug)]
+    pub(crate) struct FuturesIo<T> {
+        #[pin]
+        inner: T,
+    }
+}
+
+impl<T> FuturesIo<T> {
+    /// Create a new `FuturesIo` wrapping the given I/O object.
+    ///
+    /// # Safety
+    ///
+    /// This wrapper is only safe to use if the inner I/O object does not under any circumstance
+    /// read from the buffer passed to `poll_read` in the `futures::AsyncRead` implementation.
+    pub(crate) unsafe fn new(inner: T) -> Self {
+        Self { inner }
+    }
+
+    pub(crate) fn into_inner(self) -> T {
+        self.inner
+    }
+}
+
+impl<T> hyper::rt::Write for FuturesIo<T>
+where
+    T: futures::AsyncWrite + Unpin,
+{
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        self.project().inner.poll_write(cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
+        self.project().inner.poll_flush(cx)
+    }
+
+    fn poll_shutdown(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        self.project().inner.poll_close(cx)
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        self.project().inner.poll_write_vectored(cx, bufs)
+    }
+}
+
+// Adapted from https://github.com/hyperium/hyper-util/blob/99b77a5a6f75f24bc0bcb4ca74b5f26a07b19c80/src/rt/tokio.rs
+impl<T> hyper::rt::Read for FuturesIo<T>
+where
+    T: futures::AsyncRead + Unpin,
+{
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        mut buf: hyper::rt::ReadBufCursor<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        // Safety: buf_slice should only be written to, so it's safe to convert `&mut [MaybeUninit<u8>]` to `&mut [u8]`.
+        let buf_slice = unsafe {
+            slice::from_raw_parts_mut(buf.as_mut().as_mut_ptr() as *mut u8, buf.as_mut().len())
+        };
+
+        let n = match futures::AsyncRead::poll_read(self.project().inner, cx, buf_slice) {
+            Poll::Ready(Ok(n)) => n,
+            other => return other.map_ok(|_| ()),
+        };
+
+        unsafe {
+            buf.advance(n);
+        }
+        Poll::Ready(Ok(()))
+    }
+}

wasm/prover/src/lib.rs

@@ -1,3 +1,4 @@
+pub(crate) mod hyper_io;
 mod request_opt;
 mod requests;
 

wasm/prover/src/prover.rs

@@ -7,6 +7,7 @@ use web_time::Instant;
 
 use ws_stream_wasm::*;
 
+use crate::hyper_io::FuturesIo;
 use crate::request_opt::RequestOptions;
 use crate::requests::{ClientType, NotarizationSessionRequest, NotarizationSessionResponse};
 
@@ -15,11 +16,12 @@ pub use wasm_bindgen_rayon::init_thread_pool;
 pub use crate::request_opt::VerifyResult;
 use crate::{fetch_as_json_string, log};
 use futures::AsyncWriteExt;
-use hyper::{body::to_bytes, Body, Request, StatusCode};
+use http_body_util::{BodyExt, Full};
+use hyper::{body::Bytes, Request, StatusCode};
+
 use js_sys::Array;
 use strum::EnumMessage;
 use tlsn_core::proof::TlsProof;
-use tokio_util::compat::FuturesAsyncReadCompatExt;
 use url::Url;
 use wasm_bindgen::prelude::*;
 use web_sys::{Headers, RequestInit, RequestMode};
@@ -203,16 +205,14 @@ pub async fn prover(
     let (_, client_ws_stream) = WsMeta::connect(options.websocket_proxy_url, None)
         .await
         .expect_throw("assume the client ws connection succeeds");
-    let client_ws_stream_into = client_ws_stream.into_io();
 
     // Bind the Prover to the server connection.
     // The returned `mpc_tls_connection` is an MPC TLS connection to the Server: all data written
     // to/read from it will be encrypted/decrypted using MPC with the Notary.
     log_phase(ProverPhases::BindProverToConnection);
-    let (mpc_tls_connection, prover_fut) = prover
-        .connect(client_ws_stream_into)
-        .await
-        .map_err(|e| JsValue::from_str(&format!("Could not connect prover: {:?}", e)))?;
+    let (mpc_tls_connection, prover_fut) =
+        prover.connect(client_ws_stream.into_io()).await.unwrap();
+    let mpc_tls_connection = unsafe { FuturesIo::new(mpc_tls_connection) };
 
     let prover_ctrl = prover_fut.control();
 
@@ -227,7 +227,7 @@ pub async fn prover(
     // Attach the hyper HTTP client to the TLS connection
     log_phase(ProverPhases::AttachHttpClient);
     let (mut request_sender, connection) =
-        hyper::client::conn::handshake(mpc_tls_connection.compat())
+        hyper::client::conn::http1::handshake(mpc_tls_connection)
             .await
             .map_err(|e| JsValue::from_str(&format!("Could not handshake: {:?}", e)))?;
 
@@ -253,10 +253,10 @@ pub async fn prover(
 
     let req_with_body = if options.body.is_empty() {
         log!("empty body");
-        req_with_header.body(Body::empty())
+        req_with_header.body(Full::new(Bytes::default()))
     } else {
         log!("added body - {}", options.body.as_str());
-        req_with_header.body(Body::from(options.body))
+        req_with_header.body(Full::from(options.body))
     };
 
     let unwrapped_request = req_with_body
@@ -286,10 +286,12 @@ pub async fn prover(
 
     log_phase(ProverPhases::ParseResponse);
     // Pretty printing :)
-    let payload = to_bytes(response.into_body())
+    let payload = response
+        .into_body()
+        .collect()
         .await
         .map_err(|e| JsValue::from_str(&format!("Could not get response body: {:?}", e)))?
-        .to_vec();
+        .to_bytes();
     let parsed = serde_json::from_str::<serde_json::Value>(&String::from_utf8_lossy(&payload))
         .map_err(|e| JsValue::from_str(&format!("Could not parse response: {:?}", e)))?;
     let response_pretty = serde_json::to_string_pretty(&parsed)

wasm/prover/tests/web.rs

@@ -5,7 +5,6 @@
 extern crate wasm_bindgen_test;
 use serde_json::Value;
 use std::{collections::HashMap, str};
-use wasm_bindgen::JsValue;
 use wasm_bindgen_test::*;
 use web_sys::RequestInit;