refactor(example): enhance example for public inputs validation

Makefile

@@ -6,7 +6,7 @@ OS := $(shell uname -s)
 CONFIG_FILE?=config-files/config.yaml
 AGG_CONFIG_FILE?=config-files/config-aggregator.yaml
 
-OPERATOR_VERSION=v0.9.0
+OPERATOR_VERSION=v0.9.1
 
 ifeq ($(OS),Linux)
 	BUILD_ALL_FFI = $(MAKE) build_all_ffi_linux

batcher/aligned-batcher/Cargo.toml

@@ -31,3 +31,7 @@ bincode = "1.3.3"
 aligned-sdk = { path = "../aligned-sdk" }
 ciborium = "=0.2.2"
 priority-queue = "2.1.0"
+
+once_cell = "1.20.2"
+warp = "0.3.7"
+prometheus = { version = "0.13.4", features = ["process"] }

batcher/aligned-batcher/src/lib.rs

@@ -37,6 +37,7 @@ mod config;
 mod connection;
 mod eth;
 pub mod gnark;
+pub mod metrics;
 pub mod risc_zero;
 pub mod s3;
 pub mod sp1;
@@ -212,6 +213,7 @@ impl Batcher {
 
         // Let's spawn the handling of each connection in a separate task.
         while let Ok((stream, addr)) = listener.accept().await {
+            metrics::OPEN_CONNECTIONS.inc();
             let batcher = self.clone();
             tokio::spawn(batcher.handle_connection(stream, addr));
         }
@@ -294,6 +296,7 @@ impl Batcher {
             Ok(_) => info!("{} disconnected", &addr),
         }
 
+        metrics::OPEN_CONNECTIONS.dec();
         Ok(())
     }
 
@@ -313,6 +316,7 @@ impl Batcher {
         };
         let msg_nonce = client_msg.verification_data.nonce;
         debug!("Received message with nonce: {msg_nonce:?}",);
+        metrics::RECEIVED_PROOFS.inc();
 
         // * ---------------------------------------------------*
         // *        Perform validations over the message        *
@@ -1012,6 +1016,8 @@ impl Batcher {
 
         let proof_submitters = finalized_batch.iter().map(|entry| entry.sender).collect();
 
+        metrics::GAS_PRICE_USED_ON_LATEST_BATCH.set(gas_price.as_u64() as i64);
+
         match self
             .create_new_task(
                 *batch_merkle_root,
@@ -1023,6 +1029,7 @@ impl Batcher {
         {
             Ok(_) => {
                 info!("Batch verification task created on Aligned contract");
+                metrics::SENT_BATCHES.inc();
                 Ok(())
             }
             Err(e) => {
@@ -1031,6 +1038,7 @@ impl Batcher {
                     e
                 );
 
+                metrics::REVERTED_BATCHES.inc();
                 Err(e)
             }
         }

batcher/aligned-batcher/src/main.rs

@@ -5,7 +5,9 @@ use std::sync::Arc;
 use clap::Parser;
 use env_logger::Env;
 
+use aligned_batcher::metrics;
 use aligned_batcher::{types::errors::BatcherError, Batcher};
+use warp::Filter;
 
 /// Batcher main flow:
 /// There are two main tasks spawned: `listen_connections` and `listen_new_blocks`
@@ -38,6 +40,14 @@ async fn main() -> Result<(), BatcherError> {
 
     env_logger::Builder::from_env(Env::default().default_filter_or("info")).init();
 
+    // Endpoint for Prometheus
+    metrics::init_variables();
+    let metrics_route = warp::path!("metrics").and_then(metrics::metrics_handler);
+    println!("Starting Batcher metrics on port 9093");
+    tokio::task::spawn(async move {
+        warp::serve(metrics_route).run(([0, 0, 0, 0], 9093)).await;
+    }); //TODO read from config
+
     let batcher = Batcher::new(cli.config).await;
     let batcher = Arc::new(batcher);
 
@@ -53,6 +63,8 @@ async fn main() -> Result<(), BatcherError> {
         }
     });
 
+    metrics::batcher_started();
+
     batcher.listen_connections(&addr).await?;
 
     Ok(())