chore(coprocessor): use DatabaseURL across all services

Author: goshawk-3

coprocessor/fhevm-engine/gw-listener/src/bin/gw_listener.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 use alloy::providers::{ProviderBuilder, WsConnect};
 use alloy::{primitives::Address, transports::http::reqwest::Url};
 use clap::Parser;
-use fhevm_engine_common::{metrics_server, telemetry};
+use fhevm_engine_common::{metrics_server, telemetry, utils::DatabaseURL};
 use gw_listener::aws_s3::AwsS3Client;
 use gw_listener::chain_id_from_env;
 use gw_listener::gw_listener::GatewayListener;
@@ -18,7 +18,7 @@ use tracing::{error, info, Level};
 #[command(version, about, long_about = None)]
 struct Conf {
     #[arg(long)]
-    database_url: Option<String>,
+    database_url: Option<DatabaseURL>,
 
     #[arg(long, default_value_t = 16)]
     database_pool_size: u32,
@@ -105,18 +105,14 @@ async fn main() -> anyhow::Result<()> {
         .with_max_level(conf.log_level)
         .init();
 
-    info!(conf = ?conf, "Starting gw_listener");
-
     if !conf.service_name.is_empty() {
         if let Err(err) = telemetry::setup_otlp(&conf.service_name) {
             error!(error = %err, "Failed to setup OTLP");
         }
     }
 
-    let database_url = conf
-        .database_url
-        .clone()
-        .unwrap_or_else(|| std::env::var("DATABASE_URL").expect("DATABASE_URL is undefined"));
+    info!(gateway_url = %conf.gw_url, max_retries = %conf.provider_max_retries,
+         retry_interval = ?conf.provider_retry_interval, "Connecting to Gateway");
 
     let provider = loop {
         match ProviderBuilder::new()
@@ -152,7 +148,7 @@ async fn main() -> anyhow::Result<()> {
     };
     let config = ConfigSettings {
         host_chain_id,
-        database_url,
+        database_url: conf.database_url.clone().unwrap_or_default(),
         database_pool_size: conf.database_pool_size,
         verify_proof_req_db_channel: conf.verify_proof_req_database_channel,
         gw_url: conf.gw_url,

coprocessor/fhevm-engine/gw-listener/src/gw_listener.rs

@@ -81,7 +81,7 @@ impl<P: Provider<Ethereum> + Clone + 'static, A: AwsS3Interface + Clone + 'stati
         );
         let db_pool = PgPoolOptions::new()
             .max_connections(self.conf.database_pool_size)
-            .connect(&self.conf.database_url)
+            .connect(self.conf.database_url.as_str())
             .await?;
 
         let input_verification_handle = {
@@ -549,7 +549,7 @@ impl<P: Provider<Ethereum> + Clone + 'static, A: AwsS3Interface + Clone + 'stati
         // Check database connection
         let db_pool_result = PgPoolOptions::new()
             .max_connections(self.conf.database_pool_size)
-            .connect(&self.conf.database_url)
+            .connect(self.conf.database_url.as_str())
             .await;
 
         match db_pool_result {

coprocessor/fhevm-engine/gw-listener/src/lib.rs

@@ -1,5 +1,6 @@
 use alloy::primitives::Uint;
 use alloy::transports::http::reqwest::Url;
+use fhevm_engine_common::utils::DatabaseURL;
 use std::time::Duration;
 
 use tracing::error;
@@ -36,7 +37,7 @@ impl TryFrom<u8> for KeyType {
 #[derive(Clone, Debug)]
 pub struct ConfigSettings {
     pub host_chain_id: ChainId,
-    pub database_url: String,
+    pub database_url: DatabaseURL,
     pub database_pool_size: u32,
     pub verify_proof_req_db_channel: String,
 
@@ -67,8 +68,7 @@ impl Default for ConfigSettings {
     fn default() -> Self {
         Self {
             host_chain_id: chain_id_from_env().unwrap_or(12345),
-            database_url: std::env::var("DATABASE_URL")
-                .unwrap_or("postgres://postgres:postgres@localhost/coprocessor".to_owned()),
+            database_url: DatabaseURL::default(),
             database_pool_size: 16,
             verify_proof_req_db_channel: "event_zkpok_new_work".to_owned(),
             gw_url: "ws://127.0.0.1:8546".try_into().expect("Invalid URL"),

coprocessor/fhevm-engine/gw-listener/tests/gw_listener_tests.rs

@@ -132,15 +132,15 @@ impl TestEnvironment {
                 .await
                 .expect("valid db instance");
             eprintln!("New test database on {}", instance.db_url());
-            conf.database_url = instance.db_url().to_owned();
+            conf.database_url = instance.db_url.clone();
             _test_instance = Some(instance);
         };
         conf.error_sleep_initial_secs = 1;
         conf.error_sleep_max_secs = 1;
         let db_pool = PgPoolOptions::new()
             .max_connections(16)
             .acquire_timeout(Duration::from_secs(5))
-            .connect(&conf.database_url)
+            .connect(conf.database_url.as_str())
             .await?;
 
         // Delete all proofs from the database.

coprocessor/fhevm-engine/host-listener/src/cmd/mod.rs

@@ -24,7 +24,7 @@ use tokio_util::sync::CancellationToken;
 
 use fhevm_engine_common::healthz_server::HttpServer as HealthHttpServer;
 use fhevm_engine_common::types::{BlockchainProvider, Handle};
-use fhevm_engine_common::utils::HeartBeat;
+use fhevm_engine_common::utils::{DatabaseURL, HeartBeat};
 
 use crate::contracts::{AclContract, TfheContract};
 use crate::database::tfhe_event_propagate::{
@@ -58,7 +58,7 @@ pub struct Args {
         long,
         default_value = "postgresql://postgres:postgres@localhost:5432/coprocessor"
     )]
-    pub database_url: String,
+    pub database_url: DatabaseURL,
 
     #[arg(long, default_value = None, help = "Can be negative from last block", allow_hyphen_values = true)]
     pub start_at_block: Option<i64>,
@@ -970,7 +970,7 @@ pub async fn main(args: Args) -> anyhow::Result<()> {
     let mut log_iter = InfiniteLogIter::new(&args);
     let chain_id = log_iter.get_chain_id().await?;
     info!(chain_id = chain_id, "Chain ID");
-    if args.database_url.is_empty() {
+    if args.database_url.as_str().is_empty() {
         error!("Database URL is required");
         panic!("Database URL is required");
     };

coprocessor/fhevm-engine/host-listener/src/database/tfhe_event_propagate.rs

@@ -5,6 +5,7 @@ use anyhow::Result;
 use fhevm_engine_common::telemetry;
 use fhevm_engine_common::types::AllowEvents;
 use fhevm_engine_common::types::SupportedFheOperations;
+use fhevm_engine_common::utils::DatabaseURL;
 use fhevm_engine_common::utils::{compact_hex, HeartBeat};
 use sqlx::postgres::PgConnectOptions;
 use sqlx::postgres::PgPoolOptions;
@@ -70,7 +71,7 @@ pub fn retry_on_sqlx_error(err: &SqlxError, retry_count: &mut usize) -> bool {
 
 // A pool of connection with some cached information and automatic reconnection
 pub struct Database {
-    url: String,
+    url: DatabaseURL,
     pub pool: Arc<RwLock<sqlx::Pool<Postgres>>>,
     pub tenant_id: TenantId,
     pub chain_id: ChainId,
@@ -90,7 +91,7 @@ pub type Transaction<'l> = sqlx::Transaction<'l, Postgres>;
 
 impl Database {
     pub async fn new(
-        url: &str,
+        url: &DatabaseURL,
         coprocessor_api_key: &CoprocessorApiKey,
         bucket_cache_size: u16,
     ) -> Result<Self> {
@@ -105,7 +106,7 @@ impl Database {
             .into(),
         ));
         Ok(Database {
-            url: url.into(),
+            url: url.clone(),
             tenant_id,
             chain_id,
             pool: Arc::new(RwLock::new(pool)),
@@ -114,7 +115,7 @@ impl Database {
         })
     }
 
-    async fn new_pool(url: &str) -> PgPool {
+    async fn new_pool(url: &DatabaseURL) -> PgPool {
         let options: PgConnectOptions = url.parse().expect("bad url");
         let options = options.options([
             ("statement_timeout", "10000"), // 5 seconds

coprocessor/fhevm-engine/host-listener/tests/integration_test.rs

@@ -217,7 +217,7 @@ async fn setup(node_chain_id: Option<u64>) -> Result<Setup, anyhow::Error> {
         initial_block_time: 1,
         acl_contract_address: acl_contract.address().to_string(),
         tfhe_contract_address: tfhe_contract.address().to_string(),
-        database_url: test_instance.db_url().to_string(),
+        database_url: test_instance.db_url.clone(),
         coprocessor_api_key: Some(coprocessor_api_key),
         start_at_block: None,
         end_at_block: None,

coprocessor/fhevm-engine/sns-worker/src/lib.rs

@@ -75,17 +75,6 @@ pub struct DBConfig {
     pub lifo: bool,
 }
 
-impl std::fmt::Debug for DBConfig {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        // Custom debug impl to avoid printing sensitive information
-        write!(
-            f,
-            "db_listen_channel: {:?}, db_notify_channel: {}, db_batch_limit: {}, db_gc_batch_limit: {}, db_polling_interval: {}, db_cleanup_interval: {:?}, db_max_connections: {}, db_timeout: {:?}, lifo: {}",
-            self.listen_channels, self.notify_channel, self.batch_limit, self.gc_batch_limit, self.polling_interval, self.cleanup_interval, self.max_connections, self.timeout, self.lifo
-        )
-    }
-}
-
 #[derive(Clone, Default, Debug)]
 pub struct S3Config {
     pub bucket_ct128: String,
@@ -109,7 +98,7 @@ pub struct HealthCheckConfig {
     pub port: u16,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct Config {
     pub tenant_api_key: String,
     pub service_name: String,
@@ -510,7 +499,7 @@ pub async fn run_all(
         mpsc::channel::<UploadJob>(10 * config.s3.max_concurrent_uploads as usize);
 
     let rayon_threads = rayon::current_num_threads();
-    info!(config = ?config, rayon_threads, "Starting SNS worker");
+    info!(config = %config, rayon_threads, "Starting SNS worker");
 
     if !config.service_name.is_empty() {
         if let Err(err) = telemetry::setup_otlp(&config.service_name) {

coprocessor/fhevm-engine/sns-worker/src/tests/mod.rs

@@ -9,7 +9,6 @@ use aws_config::BehaviorVersion;
 use fhevm_engine_common::utils::{compact_hex, DatabaseURL};
 use serde::{Deserialize, Serialize};
 use serial_test::serial;
-use sqlx::Database;
 use std::{
     fs::File,
     io::{Read, Write},

coprocessor/fhevm-engine/stress-test-generator/data/json/minitest_002_erc20.json

@@ -9,9 +9,9 @@
         "user_address": "0xa0534e99d86F081E8D3868A8C4732C8f65dfdB07",
         "scenario": [
             [
-                20.0,
-                120
+                0.1,
+                5
             ]
         ]
     }
-]
+]