Feat/prover db

rln-prover/Cargo.toml

@@ -5,6 +5,10 @@ members = [
     "prover",
     "prover_cli",
     "prover_client",
+    "prover_db_migration",
+    "prover_db_entity",
+    "prover_pmtree",
+    "prover_pmtree_db_impl",
 ]
 resolver = "2"
 
@@ -41,11 +45,21 @@ prost = "0.14.1"
 tonic-prost = "0.14.2"
 tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
 tracing = "0.1.41"
+serde = { version = "1.0.228", features = ["derive"] }
+sea-orm = { version = "2.0.0-rc.19", default-features = false, features = [
+    "runtime-tokio-native-tls",
+    "sqlx-postgres",
+    # "sqlx-sqlite",
+]}
+sea-orm-migration = { version = "2.0.0-rc.19", features = [
+    "runtime-tokio-native-tls",
+    "sqlx-postgres",
+    # "sqlx-sqlite"
+]}
 
-#[build-dependencies]
+# for build
 tonic-prost-build = "0.14.2"
-
-#[dev.dependencies]
+# for becnhmark
 criterion = { version = "0.7.0", features = ["async_tokio"] }
 
 [profile.release]

rln-prover/Readme.md

@@ -44,4 +44,5 @@ RUST_LOG=debug cargo run -p prover_cli -- --ip 127.0.0.1 --metrics-ip 127.0.0.1
 ### Unit tests
 
 * cargo test 
-* cargo test --features anvil 
+* cargo test --features anvil 
+

rln-prover/prover/Cargo.toml

@@ -26,6 +26,7 @@ prost.workspace = true
 tonic-prost.workspace = true
 tracing-subscriber.workspace = true
 tracing.workspace = true
+serde.workspace = true
 tower-http = { version = "0.6.6", features = ["cors"] }
 futures = "0.3.31"
 bytesize = "2.1.0"
@@ -35,7 +36,6 @@ http = "1.3.1"
 async-channel = "2.3.1"
 # rand = "0.9.2"
 num-bigint = "0.4.6"
-serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.145"
 rocksdb = { git = "https://github.com/tillrohrmann/rust-rocksdb", branch = "issues/836" }
 nom = "8.0.0"
@@ -45,6 +45,16 @@ metrics = "0.24.2"
 metrics-exporter-prometheus = "0.17.2"
 rayon = "1.11"
 
+# user db 2
+prover_db_entity = { path = "../prover_db_entity" }
+prover_merkle_tree = { path = "../prover_pmtree_db_impl" }
+prover_pmtree = { path = "../prover_pmtree" }
+sea-orm = { version = "2.0.0-rc.18", features = [
+    "runtime-tokio-native-tls",
+    "sqlx-postgres",
+    "debug-print"
+]}
+
 [build-dependencies]
 tonic-prost-build.workspace = true
 
@@ -54,6 +64,17 @@ ark-groth16.workspace = true
 tempfile = "3.21"
 tracing-test = "0.2.5"
 lazy_static = "1.5.0"
+prover_db_migration = { path = "../prover_db_migration" }
+function_name = "0.3.0"
+
+[dev-dependencies.sea-orm]
+workspace = true
+features = [
+    "runtime-tokio-native-tls",
+    "sqlx-postgres",
+    "sqlx-sqlite",
+    "debug-print"
+]
 
 [[bench]]
 name = "prover_bench"
@@ -62,3 +83,10 @@ harness = false
 [[bench]]
 name = "prover_many_subscribers"
 harness = false
+
+[features]
+postgres = []
+
+[lints.rust]
+dead_code = "allow"
+unused = "allow"

rln-prover/prover/benches/prover_bench.rs

@@ -17,6 +17,7 @@ use tokio::task::JoinSet;
 use tonic::Response;
 // internal
 use prover::{AppArgs, MockUser, run_prover};
+use prover_db_migration::{Migrator as MigratorCreate, MigratorTrait};
 
 // grpc
 pub mod prover_proto {
@@ -29,6 +30,7 @@ use prover_proto::{
 };
 
 use lazy_static::lazy_static;
+use sea_orm::{ConnectionTrait, Database, DatabaseConnection, DbErr, Statement};
 use std::sync::Once;
 
 lazy_static! {
@@ -51,6 +53,52 @@ pub fn setup_tracing() {
     });
 }
 
+async fn create_database_connection(
+    f_name: &str,
+    test_name: &str,
+) -> Result<(String, DatabaseConnection), DbErr> {
+    // Drop / Create db_name then return a connection to it
+
+    let db_name = format!(
+        "{}_{}",
+        std::path::Path::new(f_name)
+            .file_stem()
+            .unwrap()
+            .to_str()
+            .unwrap(),
+        test_name
+    );
+
+    println!("db_name: {}", db_name);
+
+    let db_url_base = "postgres://myuser:mysecretpassword@localhost";
+    let db_url = format!("{}/{}", db_url_base, "mydatabase");
+    let db = Database::connect(db_url)
+        .await
+        .expect("Database connection 0 failed");
+
+    db.execute_raw(Statement::from_string(
+        db.get_database_backend(),
+        format!("DROP DATABASE IF EXISTS \"{}\";", db_name),
+    ))
+    .await?;
+    db.execute_raw(Statement::from_string(
+        db.get_database_backend(),
+        format!("CREATE DATABASE \"{}\";", db_name),
+    ))
+    .await?;
+
+    db.close().await?;
+
+    let db_url_final = format!("{}/{}", db_url_base, db_name);
+    let db = Database::connect(&db_url_final)
+        .await
+        .expect("Database connection failed");
+    MigratorCreate::up(&db, None).await?;
+
+    Ok((db_url_final, db))
+}
+
 async fn proof_sender(port: u16, addresses: Vec<Address>, proof_count: usize) {
     let chain_id = GrpcU256 {
         // FIXME: LE or BE?
@@ -165,15 +213,22 @@ fn proof_generation_bench(c: &mut Criterion) {
     temp_file.flush().unwrap();
 
     let port = 50051;
-    let temp_folder = tempfile::tempdir().unwrap();
-    let temp_folder_tree = tempfile::tempdir().unwrap();
+    // let temp_folder = tempfile::tempdir().unwrap();
+    // let temp_folder_tree = tempfile::tempdir().unwrap();
+
+    //     create_database_connection("prover_benches", "prover_bench")
+    //     .await
+    //     .unwrap();
+    // End Setup db
+
     // let proof_service_count = 4;
-    let app_args = AppArgs {
+    let mut app_args = AppArgs {
         ip: IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
         port,
         ws_rpc_url: None,
-        db_path: temp_folder.path().to_path_buf(),
-        merkle_tree_folder: temp_folder_tree.path().to_path_buf(),
+        db_url: None,
+        // db_path: temp_folder.path().to_path_buf(),
+        // merkle_tree_folder: temp_folder_tree.path().to_path_buf(),
         merkle_tree_count: 1,
         merkle_tree_max_count: 1,
         ksc_address: None,
@@ -203,6 +258,12 @@ fn proof_generation_bench(c: &mut Criterion) {
     // Spawn prover
     let notify_start_1 = notify_start.clone();
     rt.spawn(async move {
+        // Setup db
+        let (db_url, _db_conn) = create_database_connection("prover_benches", "prover_bench")
+            .await
+            .unwrap();
+        app_args.db_url = Some(db_url);
+
         tokio::spawn(run_prover(app_args));
         tokio::time::sleep(Duration::from_secs(10)).await;
         println!("Prover is ready, notifying it...");

rln-prover/prover/benches/prover_many_subscribers.rs

@@ -11,12 +11,14 @@ use std::time::Duration;
 use alloy::primitives::{Address, U256};
 use futures::FutureExt;
 use parking_lot::RwLock;
+use sea_orm::{ConnectionTrait, Database, DatabaseConnection, DbErr, Statement};
 use tempfile::NamedTempFile;
 use tokio::sync::Notify;
 use tokio::task::JoinSet;
 use tonic::Response;
 // internal
 use prover::{AppArgs, MockUser, run_prover};
+use prover_db_migration::{Migrator as MigratorCreate, MigratorTrait};
 
 // grpc
 pub mod prover_proto {
@@ -28,6 +30,52 @@ use prover_proto::{
     SendTransactionRequest, U256 as GrpcU256, Wei as GrpcWei, rln_prover_client::RlnProverClient,
 };
 
+async fn create_database_connection(
+    f_name: &str,
+    test_name: &str,
+) -> Result<(String, DatabaseConnection), DbErr> {
+    // Drop / Create db_name then return a connection to it
+
+    let db_name = format!(
+        "{}_{}",
+        std::path::Path::new(f_name)
+            .file_stem()
+            .unwrap()
+            .to_str()
+            .unwrap(),
+        test_name
+    );
+
+    println!("db_name: {}", db_name);
+
+    let db_url_base = "postgres://myuser:mysecretpassword@localhost";
+    let db_url = format!("{}/{}", db_url_base, "mydatabase");
+    let db = Database::connect(db_url)
+        .await
+        .expect("Database connection 0 failed");
+
+    db.execute_raw(Statement::from_string(
+        db.get_database_backend(),
+        format!("DROP DATABASE IF EXISTS \"{}\";", db_name),
+    ))
+    .await?;
+    db.execute_raw(Statement::from_string(
+        db.get_database_backend(),
+        format!("CREATE DATABASE \"{}\";", db_name),
+    ))
+    .await?;
+
+    db.close().await?;
+
+    let db_url_final = format!("{}/{}", db_url_base, db_name);
+    let db = Database::connect(&db_url_final)
+        .await
+        .expect("Database connection failed");
+    MigratorCreate::up(&db, None).await?;
+
+    Ok((db_url_final, db))
+}
+
 async fn proof_sender(ip: IpAddr, port: u16, addresses: Vec<Address>, proof_count: usize) {
     let chain_id = GrpcU256 {
         // FIXME: LE or BE?
@@ -132,12 +180,13 @@ fn proof_generation_bench(c: &mut Criterion) {
     let temp_folder = tempfile::tempdir().unwrap();
     let temp_folder_tree = tempfile::tempdir().unwrap();
     // let proof_service_count = 4;
-    let app_args = AppArgs {
+    let mut app_args = AppArgs {
         ip: IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
         port,
         ws_rpc_url: None,
-        db_path: temp_folder.path().to_path_buf(),
-        merkle_tree_folder: temp_folder_tree.path().to_path_buf(),
+        db_url: None,
+        // db_path: temp_folder.path().to_path_buf(),
+        // merkle_tree_folder: temp_folder_tree.path().to_path_buf(),
         merkle_tree_count: 1,
         merkle_tree_max_count: 1,
         ksc_address: None,
@@ -167,6 +216,12 @@ fn proof_generation_bench(c: &mut Criterion) {
     // Spawn prover
     let notify_start_1 = notify_start.clone();
     rt.spawn(async move {
+        // Setup db
+        let (db_url, _db_conn) = create_database_connection("prover_benches", "prover_bench")
+            .await
+            .unwrap();
+        app_args.db_url = Some(db_url);
+
         tokio::spawn(run_prover(app_args));
         tokio::time::sleep(Duration::from_secs(10)).await;
         println!("Prover is ready, notifying it...");

rln-prover/prover/src/args.rs

@@ -70,14 +70,16 @@ pub struct AppArgs {
         help = "Websocket rpc url (e.g. wss://eth-mainnet.g.alchemy.com/v2/your-api-key)"
     )]
     pub ws_rpc_url: Option<Url>,
-    #[arg(long = "db", help = "Db path", default_value = "./storage/db")]
-    pub db_path: PathBuf,
-    #[arg(
-        long = "tree",
-        help = "Merkle tree folder",
-        default_value = "./storage/trees"
-    )]
-    pub merkle_tree_folder: PathBuf,
+    // #[arg(long = "db", help = "Db path", default_value = "./storage/db")]
+    // pub db_path: PathBuf,
+    // #[arg(
+    //     long = "tree",
+    //     help = "Merkle tree folder",
+    //     default_value = "./storage/trees"
+    // )]
+    // pub merkle_tree_folder: PathBuf,
+    #[arg(long = "db", help = "Db url")]
+    pub db_url: Option<String>,
     #[arg(long = "tree-count", help = "Merkle tree count", default_value = "1")]
     pub merkle_tree_count: u64,
     #[arg(
@@ -274,6 +276,7 @@ mod tests {
         let config = AppArgsConfig {
             ip: None,
             port: Some(config_port),
+            db_url: None,
             mock_sc: Some(true),
             ..Default::default()
         };

rln-prover/prover/src/epoch_service.rs

@@ -9,7 +9,7 @@ use parking_lot::RwLock;
 use tokio::sync::Notify;
 use tracing::{debug, error};
 // internal
-use crate::error::AppError;
+use crate::error::AppError2;
 use crate::metrics::{
     EPOCH_SERVICE_CURRENT_EPOCH, EPOCH_SERVICE_CURRENT_EPOCH_SLICE, EPOCH_SERVICE_DRIFT_MILLIS,
 };
@@ -44,7 +44,7 @@ impl EpochService {
     // Note: listen_for_new_epoch never ends so no log will happen with #[instrument]
     //       + metrics already tracks the current epoch / epoch_slice
     // #[instrument(skip(self), fields(self.epoch_slice_duration, self.genesis, self.current_epoch))]
-    pub(crate) async fn listen_for_new_epoch(&self) -> Result<(), AppError> {
+    pub(crate) async fn listen_for_new_epoch(&self) -> Result<(), AppError2> {
         let epoch_slice_count =
             Self::compute_epoch_slice_count(EPOCH_DURATION, self.epoch_slice_duration);
         debug!("epoch slice in an epoch: {}", epoch_slice_count);
@@ -70,14 +70,14 @@ impl EpochService {
                         error!(
                             "Too many errors while computing the initial wait until, aborting..."
                         );
-                        return Err(AppError::EpochError(WaitUntilError::TooLow(d1, d2)));
+                        return Err(AppError2::EpochError(WaitUntilError::TooLow(d1, d2)));
                     }
                 }
                 Err(e) => {
                     // Another error (like OutOfRange) - exiting...
 
                     error!("Error computing the initial wait until: {}", e);
-                    return Err(AppError::EpochError(e));
+                    return Err(AppError2::EpochError(e));
                 }
             };
         };

rln-prover/prover/src/epoch_service_tests.rs

@@ -13,12 +13,12 @@ mod tests {
     use tracing_test::traced_test;
     // internal
     use crate::epoch_service::{EpochService, WAIT_UNTIL_MIN_DURATION};
-    use crate::error::AppError;
+    use crate::error::AppError2;
 
     #[derive(thiserror::Error, Debug)]
     enum AppErrorExt {
         #[error("AppError: {0}")]
-        AppError(#[from] AppError),
+        AppError(#[from] AppError2),
         #[error("Future timeout")]
         Elapsed,
     }