WIP: Fixes / KMS / Telemetry / Permissioned Mode

Cargo.toml

@@ -19,16 +19,20 @@ members = [
 [workspace.dependencies]
 anyhow = "1.0.86"
 async-trait = "0.1.81"
+aws-config = "1.5.15"
+aws-sdk-kms = "1.58.0"
 bincode = "1.3.3"
 bytemuck = "1.12"
 bytes = "1.7.2"
 clap = { version = "4.5.21", features = ["derive", "env"] }
 c-kzg = "=1.0.3"
 foundry-compilers = "0.11.0"
+gcloud-sdk = { version = "0.25.8", features = ["google-cloud-kms-v1"] }
 hashbrown = { version = "0.15.0", features = ["rayon"] }
 hex = "0.4.3"
 lazy_static = "1.5.0"
 lru = "0.12.4"
+paste = "1.0.15"
 pot = "3.0.1"
 rkyv = { version = "0.8.9", features = ["hashbrown-0_15", "std"] }
 rocksdb = "0.22.0"
@@ -40,9 +44,20 @@ spin = { version = "0.9.8", features = ["mutex"] }
 tempfile = "3.10.1"
 thiserror = "2.0.11"
 tokio = { version = "1.39.1", features = ["full"] }
+tokio-retry = "0.3.0"
+url = "2.5.4"
+
+# Telemetry
+opentelemetry = "0.27.1"
+opentelemetry_sdk = { version = "0.27.1", features = ["rt-tokio"] }
+opentelemetry-semantic-conventions = "0.27.0"
+opentelemetry-otlp = { version = "0.27.0", features = ["tonic"] }
+opentelemetry-prometheus = "0.27.0"
+opentelemetry-zipkin = "0.27.0"
 tracing = "0.1.40"
+tracing-opentelemetry = "0.28.0"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
-url = "2.5.4"
+
 
 # Alloy
 alloy = { version = "0.9.2", default-features = false, features = ["json"] }
@@ -86,11 +101,11 @@ kona-std-fpvm = { git = "https://github.com/op-rs/kona", rev = "f912b89be07c452d
 bonsai-sdk = { version = "1.2.1", features = ["non_blocking"] }
 boundless-market = "0.5.0"
 risc0-aggregation = "0.1.1"
-risc0-build = { version = "1.2.1", features = ["unstable"] }
+risc0-build = { version = "1.2.2", features = ["unstable"] }
 risc0-ethereum-contracts = { version = "1.2.1", features = ["unstable"] }
-risc0-groth16 = { version = "1.2.1", default-features = false }
-risc0-zkvm = { version = "1.2.1", features = ["heap-embedded-alloc", "unstable"] }
-risc0-zkvm-platform = { version = "1.2.1", features = ["heap-embedded-alloc"] }
+risc0-groth16 = { version = "1.2.2", default-features = false }
+risc0-zkvm = { version = "1.2.2", features = ["heap-embedded-alloc", "unstable"] }
+risc0-zkvm-platform = { version = "1.2.2", features = ["heap-embedded-alloc"] }
 
 # RISC Zero Zeth
 zeth-core = { git = "https://github.com/risc0/zeth", rev = "cecf5ec" }

bin/cli/Cargo.toml

@@ -13,21 +13,33 @@ categories.workspace = true
 [dependencies]
 anyhow.workspace = true
 async-trait.workspace = true
+aws-config.workspace = true
+aws-sdk-kms.workspace = true
 bincode.workspace = true
 bytemuck.workspace = true
 c-kzg.workspace = true
 clap.workspace = true
+gcloud-sdk.workspace = true
 hex.workspace = true
+paste.workspace = true
 rocksdb.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 sha2.workspace = true
 tempfile.workspace = true
 tokio.workspace = true
+tokio-retry.workspace = true
 tracing-subscriber.workspace = true
 tracing.workspace = true
 
-alloy = { workspace = true, features = ["full", "kzg"] }
+opentelemetry.workspace = true
+opentelemetry_sdk.workspace = true
+opentelemetry-semantic-conventions.workspace = true
+opentelemetry-otlp.workspace = true
+opentelemetry-prometheus.workspace = true
+opentelemetry-zipkin.workspace = true
+
+alloy = { workspace = true, features = ["full", "kzg", "signer-aws", "signer-gcp"] }
 alloy-rpc-types-beacon.workspace = true
 
 maili-protocol.workspace = true

bin/cli/src/bench.rs

@@ -12,8 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::CoreArgs;
+use alloy::primitives::map::{Entry, HashMap};
 use alloy::providers::{Provider, ProviderBuilder};
-use alloy::rpc::types::Block;
+use kailua_client::telemetry::TelemetryArgs;
 use risc0_zkvm::is_dev_mode;
 use std::cmp::Ordering;
 use std::collections::BinaryHeap;
@@ -23,39 +25,30 @@ use tracing::{info, warn};
 
 #[derive(clap::Args, Debug, Clone)]
 pub struct BenchArgs {
-    #[arg(long, short, help = "Verbosity level (0-4)", action = clap::ArgAction::Count)]
-    pub v: u8,
-
-    /// Address of OP-NODE endpoint to use
-    #[clap(long)]
-    pub op_node_address: String,
-    /// Address of L2 JSON-RPC endpoint to use (eth and debug namespace required).
-    #[clap(long)]
-    pub l2_node_address: String,
-    /// Address of L1 JSON-RPC endpoint to use (eth namespace required)
-    #[clap(long)]
-    pub l1_node_address: String,
-    /// Address of the L1 Beacon API endpoint to use.
-    #[clap(long)]
-    pub l1_beacon_address: String,
-    #[clap(long)]
-    pub data_dir: String,
+    #[clap(flatten)]
+    pub core: CoreArgs,
 
     /// The starting L2 block number to scan for blocks from
-    #[clap(long)]
+    #[clap(long, env)]
     pub bench_start: u64,
+    /// The length of the sequence of blocks to benchmark
+    #[clap(long, env)]
+    pub bench_length: u64,
     /// The number of L2 blocks to scan as benchmark candidates
-    #[clap(long)]
+    #[clap(long, env)]
     pub bench_range: u64,
     /// The number of top candidate L2 blocks to benchmark
-    #[clap(long)]
+    #[clap(long, env)]
     pub bench_count: u64,
+
+    #[clap(flatten)]
+    pub telemetry: TelemetryArgs,
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct CandidateBlock {
-    pub txn_count: usize,
-    pub block: Block,
+    pub txn_count: u64,
+    pub block_number: u64,
 }
 
 impl PartialOrd for CandidateBlock {
@@ -72,62 +65,81 @@ impl Ord for CandidateBlock {
 
 pub async fn benchmark(args: BenchArgs) -> anyhow::Result<()> {
     let l2_node_provider =
-        ProviderBuilder::new().on_http(args.l2_node_address.as_str().try_into()?);
+        ProviderBuilder::new().on_http(args.core.op_geth_url.as_str().try_into()?);
+    let mut cache: HashMap<u64, u64> = HashMap::new();
     // Scan L2 blocks for highest transaction counts
     let bench_end = args.bench_start + args.bench_range;
     let mut block_heap = BinaryHeap::new();
     info!("Scanning candidates.");
     for block_number in args.bench_start..bench_end {
-        let Some(block) = l2_node_provider
-            .get_block_by_number(block_number.into(), false)
-            .await?
-        else {
-            warn!("Failed to fetch block #{block_number}");
-            break;
-        };
+        let mut txn_count = 0;
+        for i in 0..args.bench_length {
+            let block_number = block_number + i;
+            txn_count += match cache.entry(block_number) {
+                Entry::Occupied(e) => *e.get(),
+                Entry::Vacant(e) => {
+                    let x = l2_node_provider
+                        .get_block_transaction_count_by_number(block_number.into())
+                        .await?
+                        .unwrap_or_else(|| {
+                            panic!("Failed to fetch transaction count for block {block_number}")
+                        });
+                    *e.insert(x)
+                }
+            }
+        }
         block_heap.push(CandidateBlock {
-            txn_count: block.transactions.len(),
-            block,
+            txn_count,
+            block_number,
         })
     }
     // Benchmark top candidates
     for _ in 0..args.bench_count {
-        let Some(block) = block_heap.pop() else {
+        let Some(CandidateBlock {
+            txn_count,
+            block_number,
+        }) = block_heap.pop()
+        else {
             warn!("Ran out of candidates too early.");
             break;
         };
-        let block_number = block.block.header.number.to_string();
-        let txn_count = block.txn_count;
-        info!("Processing candidate block {block_number} with {txn_count} transactions.");
+        let end = block_number + args.bench_length;
+        info!("Processing blocks {block_number}-{end} with {txn_count} transactions.");
         // Derive output file name
-        let version = risc0_zkvm::get_version().unwrap();
-        let output_file_name = format!("bench-risc0-{version}-{block_number}-{txn_count}.out");
+        let version = risc0_zkvm::get_version()?;
+        let output_file_name =
+            format!("bench-risc0-{version}-{block_number}-{end}-{txn_count}.out");
         let output_file = OpenOptions::new()
             .create(true)
             .append(true)
             .open(&output_file_name)?;
         // Pipe outputs to file
-        let verbosity_level = if args.v > 0 {
-            format!("-{}", "v".repeat(args.v as usize))
+        let verbosity_level = if args.core.v > 0 {
+            format!("-{}", "v".repeat(args.core.v as usize))
         } else {
             String::new()
         };
         let mut cmd = Command::new("just");
         if is_dev_mode() {
             cmd.env("RISC0_DEV_MODE", "1");
         }
+        let block_number = block_number.to_string();
+        let block_count = args.bench_length.to_string();
+        let data_dir = args.core.data_dir.clone().unwrap();
         cmd.args(vec![
-                "prove",
-                &block_number,
-                &args.l1_node_address,
-                &args.l1_beacon_address,
-                &args.l2_node_address,
-                &args.op_node_address,
-                &args.data_dir,
-                &verbosity_level,
-            ])
-            .stdout(output_file)
-            .status()?;
+            "prove",
+            &block_number,
+            &block_count,
+            &args.core.eth_rpc_url,
+            &args.core.beacon_rpc_url,
+            &args.core.op_geth_url,
+            &args.core.op_node_url,
+            data_dir.to_str().unwrap(),
+            "debug",
+            &verbosity_level,
+        ]);
+        println!("Executing: {cmd:?}");
+        cmd.stdout(output_file).status()?;
         info!("Output written to {output_file_name}");
     }
     Ok(())

bin/cli/src/config.rs

@@ -18,9 +18,12 @@ use alloy::primitives::address;
 use alloy::providers::ProviderBuilder;
 use anyhow::Context;
 use kailua_build::KAILUA_FPVM_ID;
+use kailua_client::telemetry::TelemetryArgs;
 use kailua_common::config::{config_hash, BN254_CONTROL_ID, CONTROL_ROOT, SET_BUILDER_ID};
 use kailua_contracts::SystemConfig;
 use kailua_host::config::fetch_rollup_config;
+use opentelemetry::global::tracer;
+use opentelemetry::trace::{FutureExt, Status, TraceContextExt, Tracer};
 use risc0_zkvm::sha::Digest;
 
 #[derive(clap::Args, Debug, Clone)]
@@ -37,17 +40,35 @@ pub struct ConfigArgs {
     /// Address of the ethereum rpc endpoint to use (eth namespace required)
     #[clap(long, env)]
     pub eth_rpc_url: String,
+
+    #[clap(flatten)]
+    pub telemetry: TelemetryArgs,
 }
 
 pub async fn config(args: ConfigArgs) -> anyhow::Result<()> {
+    let tracer = tracer("kailua");
+    let context = opentelemetry::Context::current_with_span(tracer.start("config"));
+
     let config = fetch_rollup_config(&args.op_node_url, &args.op_geth_url, None)
+        .with_context(context.clone())
         .await
         .context("fetch_rollup_config")?;
+
     let eth_rpc_provider = ProviderBuilder::new().on_http(args.eth_rpc_url.as_str().try_into()?);
     // load system config
     let system_config = SystemConfig::new(config.l1_system_config_address, &eth_rpc_provider);
-    let portal_address = system_config.optimismPortal().stall().await.addr_;
-    let dgf_address = system_config.disputeGameFactory().stall().await.addr_;
+    let portal_address = system_config
+        .optimismPortal()
+        .stall()
+        .with_context(context.with_span(tracer.start_with_context("optimismPortal", &context)))
+        .await
+        .addr_;
+    let dgf_address = system_config
+        .disputeGameFactory()
+        .stall()
+        .with_context(context.with_span(tracer.start_with_context("disputeGameFactory", &context)))
+        .await
+        .addr_;
 
     // report risc0 version
     println!("RISC0_VERSION: {}", risc0_zkvm::get_version()?);
@@ -124,5 +145,6 @@ pub async fn config(args: ConfigArgs) -> anyhow::Result<()> {
     // report game type
     println!("KAILUA_GAME_TYPE: {}", KAILUA_GAME_TYPE);
 
+    context.span().set_status(Status::Ok);
     Ok(())
 }