feat(host): apply reqactor,reqpool

Cargo.toml

@@ -15,6 +15,9 @@ members = [
     "pipeline",
     "core",
     "taskdb",
+    "redis-derive",
+    "reqpool",
+    "reqactor",
 ]
 
 # Always optimize; building and running the guest takes much longer without optimization.
@@ -37,6 +40,9 @@ opt-level = 3
 raiko-lib = { path = "./lib", features = ["std"] }
 raiko-core = { path = "./core" }
 raiko-tasks = { path = "./taskdb" }
+raiko-redis-derive = { path = "./redis-derive" }
+raiko-reqpool = { path = "./reqpool" }
+raiko-reqactor = { path = "./reqactor" }
 
 # reth
 reth-primitives = { git = "https://github.com/taikoxyz/taiko-reth.git", branch = "v1.0.0-rc.2-taiko", default-features = false, features = [
@@ -186,6 +192,8 @@ dirs = "5.0.1"
 pathdiff = "0.2.1"
 dotenv = "0.15.0"
 backoff = "0.4.0"
+derive-getters = "0.5.0"
+test-log = "0.2.16"
 
 [patch.crates-io]
 revm = { git = "https://github.com/taikoxyz/revm.git", branch = "v36-taiko" }

host/Cargo.toml

@@ -15,6 +15,8 @@ sgx-prover = { path = "../provers/sgx/prover", optional = true }
 raiko-lib = { workspace = true }
 raiko-core = { workspace = true }
 raiko-tasks = { workspace = true }
+raiko-reqpool = { workspace = true }
+raiko-reqactor = { workspace = true }
 
 # alloy
 alloy-rlp = { workspace = true }
@@ -82,13 +84,15 @@ assert_cmd = { workspace = true }
 rstest = { workspace = true }
 ethers-core = { workspace = true }
 rand = { workspace = true }
+test-log = { workspace = true }
 
 [features]
 default = []
 sp1 = ["raiko-core/sp1"]
 risc0 = ["raiko-core/risc0"]
 sgx = ["raiko-core/sgx"]
 integration = []
+test-utils = ["raiko-core/test-utils"]
 
 [[bin]]
 name = "raiko-host"

host/config/chain_spec_list_default.json

@@ -25,7 +25,7 @@
         },
         "l1_contract": null,
         "l2_contract": null,
-        "rpc": "https://ethereum-rpc.publicnode.com",
+        "rpc": "https://eth.drpc.org",
         "beacon_rpc": "https://ethereum-beacon-api.publicnode.com",
         "verifier_address_forks": {
             "FRONTIER": {
@@ -152,4 +152,4 @@
         "seconds_per_slot": 1,
         "is_taiko": true
     }
-]
+]

host/src/bin/main.rs

@@ -1,5 +1,6 @@
 #![allow(incomplete_features)]
-use raiko_host::{interfaces::HostResult, server::serve, ProverState};
+use raiko_host::{interfaces::HostResult, parse_chain_specs, parse_opts, server::serve};
+use raiko_reqpool::RedisPoolConfig;
 use std::path::PathBuf;
 use tracing::{debug, info};
 use tracing_appender::{
@@ -14,20 +15,35 @@ async fn main() -> HostResult<()> {
     env_logger::Builder::from_default_env()
         .target(env_logger::Target::Stdout)
         .init();
-    let state = ProverState::init()?;
-    let _guard = subscribe_log(
-        &state.opts.log_path,
-        &state.opts.log_level,
-        state.opts.max_log,
-    );
-    debug!("Start config:\n{:#?}", state.opts.proof_request_opt);
-    debug!("Args:\n{:#?}", state.opts);
+    let opts = parse_opts()?;
+    let chain_specs = parse_chain_specs(&opts);
+    let default_request_config = opts.proof_request_opt.clone();
+    let max_proving_concurrency = opts.concurrency_limit;
 
-    info!("Supported chains: {:?}", state.chain_specs);
-    info!("Start config:\n{:#?}", state.opts.proof_request_opt);
-    info!("Args:\n{:#?}", state.opts);
+    let pool = raiko_reqpool::Pool::open(RedisPoolConfig {
+        redis_url: opts.redis_url.clone(),
+        redis_ttl: opts.redis_ttl,
+        enable_memory_backend: opts.use_memory_backend,
+    })
+    .map_err(|e| anyhow::anyhow!(e))?;
 
-    serve(state).await?;
+    let actor = raiko_reqactor::start_actor(
+        pool,
+        chain_specs.clone(),
+        default_request_config.clone(),
+        max_proving_concurrency,
+    )
+    .await;
+
+    let _guard = subscribe_log(&opts.log_path, &opts.log_level, opts.max_log);
+    debug!("Start config:\n{:#?}", default_request_config);
+    debug!("Args:\n{:#?}", opts);
+    info!("Supported chains: {:?}", chain_specs);
+
+    let address = opts.address.as_str();
+    let concurrency = opts.concurrency_limit;
+    let jwt_secret = opts.jwt_secret.clone();
+    serve(actor, address, concurrency, jwt_secret).await?;
     Ok(())
 }
 

host/src/lib.rs

@@ -85,6 +85,9 @@ pub struct Opts {
 
     #[arg(long, require_equals = true, default_value = "3600")]
     pub redis_ttl: u64,
+
+    #[arg(long, default_value = "false")]
+    pub use_memory_backend: bool,
 }
 
 impl Opts {
@@ -168,23 +171,11 @@ pub enum Message {
 
 impl ProverState {
     pub fn init() -> HostResult<Self> {
-        // Read the command line arguments;
-        let mut opts = Opts::parse();
-        // Read env supported options.
-        opts.merge_from_env();
-        // Read the config file.
-        opts.merge_from_file()?;
-
+        let opts = parse_opts()?;
         Self::init_with_opts(opts)
     }
 
     pub fn init_with_opts(opts: Opts) -> HostResult<Self> {
-        let chain_specs = if let Some(cs_path) = &opts.chain_spec_path {
-            SupportedChainSpecs::merge_from_file(cs_path.clone()).unwrap_or_default()
-        } else {
-            SupportedChainSpecs::default()
-        };
-
         // Check if the cache path exists and create it if it doesn't.
         if let Some(cache_path) = &opts.cache_path {
             if !cache_path.exists() {
@@ -196,6 +187,7 @@ impl ProverState {
         let pause_flag = Arc::new(AtomicBool::new(false));
 
         let opts_clone = opts.clone();
+        let chain_specs = parse_chain_specs(&opts);
         let chain_specs_clone = chain_specs.clone();
         let sender = task_channel.clone();
         tokio::spawn(async move {
@@ -245,6 +237,25 @@ impl ProverState {
     }
 }
 
+pub fn parse_opts() -> HostResult<Opts> {
+    // Read the command line arguments;
+    let mut opts = Opts::parse();
+    // Read env supported options.
+    opts.merge_from_env();
+    // Read the config file.
+    opts.merge_from_file()?;
+
+    Ok(opts)
+}
+
+pub fn parse_chain_specs(opts: &Opts) -> SupportedChainSpecs {
+    if let Some(cs_path) = &opts.chain_spec_path {
+        SupportedChainSpecs::merge_from_file(cs_path.clone()).expect("Failed to parse chain specs")
+    } else {
+        SupportedChainSpecs::default()
+    }
+}
+
 #[global_allocator]
 static ALLOCATOR: Cap<alloc::System> = Cap::new(alloc::System, usize::MAX);