feat: cloud prover many fixs and feats

Cargo.toml

@@ -4,27 +4,24 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-anyhow = "1.0"
-log = "0.4"
+eyre = "0.6"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1.0"
 ethers-core = { git = "https://github.com/scroll-tech/ethers-rs.git", branch = "v2.0.7" }
-ethers-providers = { git = "https://github.com/scroll-tech/ethers-rs.git", branch = "v2.0.7" }
-reqwest = { version = "0.12.4", features = ["gzip"] }
-reqwest-middleware = "0.3"
-reqwest-retry = "0.5"
-hex = "0.4.3"
-tiny-keccak = { version = "2.0.0", features = ["sha3", "keccak"] }
-rand = "0.8.5"
-rlp = "0.5.2"
-tokio = { version = "1.37.0", features = ["full"] }
+reqwest = { version = "0.12", features = ["gzip"] }
+reqwest-middleware = "0.4"
+reqwest-retry = "0.7"
+hex = "0.4"
+tiny-keccak = { version = "2.0", features = ["sha3", "keccak"] }
+rand = "0.9"
+rlp = "0.6"
+tokio = { version = "1.48", features = ["net", "sync"] }
 async-trait = "0.1"
-http = "1.1.0"
+http = "1.4"
 clap = { version = "4.5", features = ["derive"] }
-tracing = "0.1.40"
-tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
-axum = "0.6.0"
-dotenv = "0.15"
-rocksdb = "0.23.0"
-url = "2.5.4"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+axum = { version = "0.8", default-features = false, features = ["tokio", "http1"] }
+dotenvy = "0.15"
+rocksdb = "0.24"
 

examples/cloud.rs

@@ -1,4 +1,5 @@
-use anyhow::{anyhow, Result};
+#![allow(dead_code)]
+use eyre::{eyre, Result};
 use async_trait::async_trait;
 use clap::Parser;
 use reqwest::Url;
@@ -37,7 +38,7 @@ impl CloudProverConfig {
     where
         R: std::io::Read,
     {
-        serde_json::from_reader(reader).map_err(|e| anyhow!(e))
+        serde_json::from_reader(reader).map_err(|e| eyre!(e))
     }
 
     pub fn from_file(file_name: String) -> Result<Self> {
@@ -49,7 +50,7 @@ impl CloudProverConfig {
         std::env::var_os(key)
             .map(|val| {
                 val.to_str()
-                    .ok_or_else(|| anyhow!("{key} env var is not valid UTF-8"))
+                    .ok_or_else(|| eyre!("{key} env var is not valid UTF-8"))
                     .map(String::from)
             })
             .transpose()
@@ -80,13 +81,13 @@ impl ProvingService for CloudProver {
     fn is_local(&self) -> bool {
         false
     }
-    async fn get_vks(&self, req: GetVkRequest) -> GetVkResponse {
+    async fn get_vks(&self, _req: GetVkRequest) -> GetVkResponse {
         todo!()
     }
-    async fn prove(&mut self, req: ProveRequest) -> ProveResponse {
+    async fn prove(&mut self, _req: ProveRequest) -> ProveResponse {
         todo!()
     }
-    async fn query_task(&mut self, req: QueryTaskRequest) -> QueryTaskResponse {
+    async fn query_task(&mut self, _req: QueryTaskRequest) -> QueryTaskResponse {
         todo!()
     }
 }
@@ -101,8 +102,8 @@ impl CloudProver {
     }
 }
 
-#[tokio::main]
-async fn main() -> anyhow::Result<()> {
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<()> {
     init_tracing();
 
     let args = Args::parse();

examples/local.rs

@@ -1,4 +1,4 @@
-use anyhow::{anyhow, Result};
+use eyre::{eyre, Result};
 use async_trait::async_trait;
 use clap::Parser;
 use scroll_proving_sdk::{
@@ -35,7 +35,7 @@ impl LocalProverConfig {
     where
         R: std::io::Read,
     {
-        serde_json::from_reader(reader).map_err(|e| anyhow!(e))
+        serde_json::from_reader(reader).map_err(|e| eyre!(e))
     }
 
     pub fn from_file(file_name: String) -> Result<Self> {
@@ -56,25 +56,25 @@ impl ProvingService for LocalProver {
     fn is_local(&self) -> bool {
         true
     }
-    async fn get_vks(&self, req: GetVkRequest) -> GetVkResponse {
+    async fn get_vks(&self, _req: GetVkRequest) -> GetVkResponse {
         todo!()
     }
-    async fn prove(&mut self, req: ProveRequest) -> ProveResponse {
+    async fn prove(&mut self, _req: ProveRequest) -> ProveResponse {
         todo!()
     }
-    async fn query_task(&mut self, req: QueryTaskRequest) -> QueryTaskResponse {
+    async fn query_task(&mut self, _req: QueryTaskRequest) -> QueryTaskResponse {
         todo!()
     }
 }
 
 impl LocalProver {
-    pub fn new(cfg: LocalProverConfig) -> Self {
+    pub fn new(_cfg: LocalProverConfig) -> Self {
         Self {}
     }
 }
 
-#[tokio::main]
-async fn main() -> anyhow::Result<()> {
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<()> {
     init_tracing();
 
     let args = Args::parse();

src/config.rs

@@ -1,6 +1,6 @@
 use crate::{coordinator_handler::ProverType, prover::ProofType};
-use anyhow::{anyhow, Result};
-use dotenv::dotenv;
+use eyre::{eyre, Result};
+use dotenvy::dotenv;
 use serde::{Deserialize, Serialize};
 use serde_json;
 use std::fs::File;
@@ -47,7 +47,7 @@ impl Config {
     where
         R: std::io::Read,
     {
-        serde_json::from_reader(reader).map_err(|e| anyhow!(e))
+        serde_json::from_reader(reader).map_err(|e| eyre!(e))
     }
 
     pub fn from_file(file_name: String) -> Result<Self> {
@@ -65,7 +65,7 @@ impl Config {
         std::env::var_os(key)
             .map(|val| {
                 val.to_str()
-                    .ok_or_else(|| anyhow!("{key} env var is not valid UTF-8"))
+                    .ok_or_else(|| eyre!("{key} env var is not valid UTF-8"))
                     .map(String::from)
             })
             .transpose()

src/coordinator_handler/api.rs

@@ -8,6 +8,7 @@ use reqwest::{header::CONTENT_TYPE, Url};
 use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
 use reqwest_retry::{policies::ExponentialBackoff, RetryTransientMiddleware};
 use serde::Serialize;
+use tracing::Level;
 
 pub struct Api {
     pub base_url: Url,
@@ -16,7 +17,7 @@ pub struct Api {
 }
 
 impl Api {
-    pub fn new(cfg: CoordinatorConfig) -> anyhow::Result<Self> {
+    pub fn new(cfg: CoordinatorConfig) -> eyre::Result<Self> {
         let retry_wait_duration = Duration::from_secs(cfg.retry_wait_time_sec);
         let retry_policy = ExponentialBackoff::builder()
             .retry_bounds(retry_wait_duration / 2, retry_wait_duration)
@@ -33,16 +34,17 @@ impl Api {
         })
     }
 
-    fn build_url(&self, method: &str) -> anyhow::Result<Url> {
-        self.base_url.join(method).map_err(|e| anyhow::anyhow!(e))
+    fn build_url(&self, method: &str) -> eyre::Result<Url> {
+        self.base_url.join(method).map_err(|e| eyre::eyre!(e))
     }
 
+    #[instrument(target = "coordinator_client", skip(self, req, token), level = Level::DEBUG)]
     async fn post_with_token<Req, Resp>(
         &self,
         method: &str,
         req: &Req,
         token: &String,
-    ) -> anyhow::Result<Resp>
+    ) -> eyre::Result<Resp>
     where
         Req: ?Sized + Serialize,
         Resp: serde::de::DeserializeOwned,
@@ -51,8 +53,8 @@ impl Api {
         let request_body = serde_json::to_string(req)?;
         let size = request_body.len();
 
-        log::info!("[coordinator client], {method}, sent request");
-        log::debug!("[coordinator client], {method}, request: {request_body}, token: {token}, request size: {size}");
+        info!("sent request");
+        trace!(token = %token, request_body = %request_body, size = %size);
         let response = self
             .client
             .post(url)
@@ -64,20 +66,20 @@ impl Api {
             .await?;
 
         if response.status() != http::status::StatusCode::OK {
-            anyhow::bail!(
+            eyre::bail!(
                 "[coordinator client], {method}, status not ok: {}",
                 response.status()
             )
         }
 
         let response_body = response.text().await?;
 
-        log::info!("[coordinator client], {method}, received response");
-        log::debug!("[coordinator client], {method}, response: {response_body}");
-        serde_json::from_str(&response_body).map_err(|e| anyhow::anyhow!(e))
+        info!("received response");
+        trace!(response_body = %response_body);
+        serde_json::from_str(&response_body).map_err(|e| eyre::eyre!(e))
     }
 
-    pub async fn challenge(&self) -> anyhow::Result<Response<ChallengeResponseData>> {
+    pub async fn challenge(&self) -> eyre::Result<Response<ChallengeResponseData>> {
         let method = "/coordinator/v1/challenge";
         let url = self.build_url(method)?;
 
@@ -91,14 +93,14 @@ impl Api {
 
         let response_body = response.text().await?;
 
-        serde_json::from_str(&response_body).map_err(|e| anyhow::anyhow!(e))
+        serde_json::from_str(&response_body).map_err(|e| eyre::eyre!(e))
     }
 
     pub async fn login(
         &self,
         req: &LoginRequest,
         token: &String,
-    ) -> anyhow::Result<Response<LoginResponseData>> {
+    ) -> eyre::Result<Response<LoginResponseData>> {
         let method = "/coordinator/v1/login";
         self.post_with_token(method, req, token).await
     }
@@ -107,9 +109,9 @@ impl Api {
         &self,
         req: &GetTaskRequest,
         token: &String,
-    ) -> anyhow::Result<Response<GetTaskResponseData>> {
+    ) -> eyre::Result<Response<GetTaskResponseData>> {
         let method = "/coordinator/v1/get_task";
-        if self.send_timeout < core::time::Duration::from_secs(600) {
+        if self.send_timeout < Duration::from_secs(600) {
             tracing::warn!(
                 "get_task API is time-consuming, timeout setting is too low ({}), set it to more than 600s",
                 self.send_timeout.as_secs(),
@@ -123,7 +125,7 @@ impl Api {
         &self,
         req: &SubmitProofRequest,
         token: &String,
-    ) -> anyhow::Result<Response<SubmitProofResponseData>> {
+    ) -> eyre::Result<Response<SubmitProofResponseData>> {
         let method = "/coordinator/v1/submit_proof";
         self.post_with_token(method, req, token).await
     }

src/coordinator_handler/coordinator_client.rs

@@ -23,7 +23,7 @@ impl CoordinatorClient {
         prover_name: String,
         prover_provider_type: ProverProviderType,
         key_signer: KeySigner,
-    ) -> anyhow::Result<Self> {
+    ) -> eyre::Result<Self> {
         let api = Api::new(cfg)?;
         let client = Self {
             prover_types,
@@ -40,7 +40,7 @@ impl CoordinatorClient {
     pub async fn get_task(
         &self,
         req: &GetTaskRequest,
-    ) -> anyhow::Result<Response<GetTaskResponseData>> {
+    ) -> eyre::Result<Response<GetTaskResponseData>> {
         let token = self.get_token(false).await?;
         let response = self.api.get_task(req, &token).await?;
 
@@ -55,7 +55,7 @@ impl CoordinatorClient {
     pub async fn submit_proof(
         &self,
         req: &SubmitProofRequest,
-    ) -> anyhow::Result<Response<SubmitProofResponseData>> {
+    ) -> eyre::Result<Response<SubmitProofResponseData>> {
         let token = self.get_token(false).await?;
         let response = self.api.submit_proof(req, &token).await?;
 
@@ -73,7 +73,7 @@ impl CoordinatorClient {
     ///
     /// If the token is expired, `force_relogin` is set to `true`, or a login was never performed
     /// before, it will authenticate and fetch a new token.
-    pub async fn get_token(&self, force_relogin: bool) -> anyhow::Result<String> {
+    pub async fn get_token(&self, force_relogin: bool) -> eyre::Result<String> {
         let token_guard = self.token.lock().await;
 
         match *token_guard {
@@ -87,15 +87,15 @@ impl CoordinatorClient {
     async fn login(
         &self,
         mut token_guard: MutexGuard<'_, Option<String>>,
-    ) -> anyhow::Result<String> {
+    ) -> eyre::Result<String> {
         let challenge_response = self
             .api
             .challenge()
             .await
-            .map_err(|e| anyhow::anyhow!("Failed to request a challenge: {e}"))?;
+            .map_err(|e| eyre::eyre!("Failed to request a challenge: {e}"))?;
 
         if challenge_response.errcode != ErrorCode::Success {
-            anyhow::bail!(
+            eyre::bail!(
                 "Challenge request failed with {:?} {}",
                 challenge_response.errcode,
                 challenge_response.errmsg
@@ -105,7 +105,7 @@ impl CoordinatorClient {
         let login_response_data = challenge_response
             .data
             .as_ref()
-            .ok_or_else(|| anyhow::anyhow!("Missing challenge token"))?;
+            .ok_or_else(|| eyre::eyre!("Missing challenge token"))?;
 
         let login_message = LoginMessage {
             challenge: login_response_data.token.clone(),
@@ -120,7 +120,7 @@ impl CoordinatorClient {
         let signature = self
             .key_signer
             .sign_buffer(&buffer)
-            .map_err(|e| anyhow::anyhow!("Failed to sign the login message: {e}"))?;
+            .map_err(|e| eyre::eyre!("Failed to sign the login message: {e}"))?;
 
         let login_request = LoginRequest {
             message: login_message,
@@ -131,10 +131,10 @@ impl CoordinatorClient {
             .api
             .login(&login_request, &login_response_data.token)
             .await
-            .map_err(|e| anyhow::anyhow!("Failed to login: {e}"))?;
+            .map_err(|e| eyre::eyre!("Failed to login: {e}"))?;
 
         if login_response.errcode != ErrorCode::Success {
-            anyhow::bail!(
+            eyre::bail!(
                 "Login request failed with {:?} {}",
                 login_response.errcode,
                 login_response.errmsg
@@ -143,7 +143,7 @@ impl CoordinatorClient {
         let token = login_response
             .data
             .map(|r| r.token)
-            .ok_or_else(|| anyhow::anyhow!("Empty data in response, lack of login"))?;
+            .ok_or_else(|| eyre::eyre!("Empty data in response, lack of login"))?;
 
         *token_guard = Some(token.clone());
 

src/coordinator_handler/error.rs

@@ -36,7 +36,7 @@ impl ErrorCode {
             50000 => ErrorCode::ErrJWTCommonErr,
             50001 => ErrorCode::ErrJWTTokenExpired,
             _ => {
-                log::error!("get unexpected error code from coordinator: {v}");
+                error!("get unexpected error code from coordinator: {v}");
                 ErrorCode::Undefined(v)
             }
         }