Move the poller logic to a separate module

maximopalopoli · maximopalopoli · commit fe7ff8c1ef8e · 2025-12-09T16:47:06.000-03:00
diff --git a/Makefile b/Makefile
@@ -315,10 +315,16 @@ agg_mode_run_migrations: agg_mode_docker_up
 	cargo run --manifest-path ./aggregation_mode/Cargo.toml --release --bin migrate -- postgres://postgres:postgres@localhost:5435/
 
 agg_mode_batcher_start_local: agg_mode_run_migrations
-	cargo run --manifest-path ./aggregation_mode/Cargo.toml --release --bin agg_mode_batcher -- config-files/config-agg-mode-batcher.yaml
+	cargo run --manifest-path ./aggregation_mode/Cargo.toml --release --bin batcher_server -- config-files/config-agg-mode-batcher.yaml
 
 agg_mode_batcher_start_ethereum_package: agg_mode_run_migrations
-	cargo run --manifest-path ./aggregation_mode/Cargo.toml --release --bin agg_mode_batcher -- config-files/config-agg-mode-batcher-ethereum-package.yaml
+	cargo run --manifest-path ./aggregation_mode/Cargo.toml --release --bin batcher_server -- config-files/config-agg-mode-batcher-ethereum-package.yaml
+
+agg_mode_batcher_poller_start_local: agg_mode_run_migrations
+	cargo run --manifest-path ./aggregation_mode/Cargo.toml --release --bin agg_mode_batcher_poller -- config-files/config-agg-mode-batcher.yaml
+
+agg_mode_batcher_poller_start_ethereum_package: agg_mode_run_migrations
+	cargo run --manifest-path ./aggregation_mode/Cargo.toml --release --bin agg_mode_batcher_poller -- config-files/config-agg-mode-batcher-ethereum-package.yaml
 
 __AGGREGATOR__: ## ____
 
diff --git a/aggregation_mode/Cargo.lock b/aggregation_mode/Cargo.lock
diff --git a/aggregation_mode/Cargo.toml b/aggregation_mode/Cargo.toml
@@ -1,6 +1,6 @@
 [workspace]
 resolver = "2"
-members = ["./batcher", "./proof_aggregator", "./db"]
+members = ["./batcher_server", "./proof_aggregator", "./db", "./payments_poller"]
 
 [workspace.package]
 version = "0.1.0"
diff --git a/aggregation_mode/batcher/src/lib.rs b/aggregation_mode/batcher/src/lib.rs
diff --git a/aggregation_mode/batcher/src/poller/mod.rs b/aggregation_mode/batcher/src/poller/mod.rs
diff --git a/aggregation_mode/batcher_server/Cargo.toml b/aggregation_mode/batcher_server/Cargo.toml
@@ -1,5 +1,5 @@
 [package]
-name = "agg_mode_batcher"
+name = "batcher_server"
 version = "0.1.0"
 edition = "2021"
 
diff --git a/aggregation_mode/batcher_server/src/config.rs b/aggregation_mode/batcher_server/src/config.rs
@@ -0,0 +1,19 @@
+use std::{fs::File, io::Read};
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct Config {
+    pub port: u16,
+    pub db_connection_url: String,
+}
+
+impl Config {
+    pub fn from_file(file_path: &str) -> Result<Config, Box<dyn std::error::Error>> {
+        let mut file = File::open(file_path)?;
+        let mut contents = String::new();
+        file.read_to_string(&mut contents)?;
+        let config: Config = serde_yaml::from_str(&contents)?;
+        Ok(config)
+    }
+}
diff --git a/aggregation_mode/batcher_server/src/db.rs b/aggregation_mode/batcher_server/src/db.rs
@@ -72,29 +72,6 @@ impl Db {
         .await
     }
 
-    pub async fn insert_payment_event(
-        &self,
-        address: &str,
-        started_at: &BigDecimal,
-        amount: &BigDecimal,
-        valid_until: &BigDecimal,
-        tx_hash: &str,
-    ) -> Result<(), sqlx::Error> {
-        sqlx::query(
-            "INSERT INTO payment_events (address, started_at, amount, valid_until, tx_hash)
-            VALUES ($1, $2, $3, $4, $5)
-            ON CONFLICT (tx_hash) DO NOTHING",
-        )
-        .bind(address.to_lowercase())
-        .bind(started_at)
-        .bind(amount)
-        .bind(valid_until)
-        .bind(tx_hash)
-        .execute(&self.pool)
-        .await
-        .map(|_| ())
-    }
-
     pub async fn has_active_payment_event(
         &self,
         address: &str,
diff --git a/aggregation_mode/batcher_server/src/helpers.rs b/aggregation_mode/batcher_server/src/helpers.rs
diff --git a/aggregation_mode/batcher_server/src/http.rs b/aggregation_mode/batcher_server/src/http.rs
@@ -18,9 +18,7 @@ use super::{
 use crate::{
     config::Config,
     db::Db,
-    server::types::{
-        SubmitProofRequest, SubmitProofRequestMessageRisc0, SubmitProofRequestMessageSP1,
-    },
+    types::{SubmitProofRequest, SubmitProofRequestMessageRisc0, SubmitProofRequestMessageSP1},
 };
 
 #[derive(Clone, Debug)]
diff --git a/aggregation_mode/batcher_server/src/lib.rs b/aggregation_mode/batcher_server/src/lib.rs
@@ -1,3 +1,5 @@
+pub mod config;
+pub mod db;
 mod helpers;
 pub mod http;
 mod types;
diff --git a/aggregation_mode/batcher_server/src/main.rs b/aggregation_mode/batcher_server/src/main.rs
@@ -0,0 +1,38 @@
+use std::env;
+
+use batcher_server::{config::Config, db::Db, http::BatcherServer};
+use tracing_subscriber::{EnvFilter, FmtSubscriber};
+
+fn read_config_filepath_from_args() -> String {
+    let args: Vec<String> = env::args().collect();
+    if args.len() < 2 {
+        panic!(
+            "You must provide a config file. Usage: {} <config-file-path>",
+            args[0]
+        );
+    }
+
+    args[1].clone()
+}
+
+#[tokio::main]
+async fn main() {
+    let filter = EnvFilter::new("info,sp1_cuda=warn");
+    let subscriber = FmtSubscriber::builder().with_env_filter(filter).finish();
+    tracing::subscriber::set_global_default(subscriber).expect("setting default subscriber failed");
+
+    let config_file_path = read_config_filepath_from_args();
+    tracing::info!("Loading config from {}...", config_file_path);
+    let config = Config::from_file(&config_file_path).expect("Config is valid");
+    tracing::info!("Config loaded");
+
+    let db = Db::try_new(&config.db_connection_url)
+        .await
+        .expect("db to start");
+
+    let http_server = BatcherServer::new(db, config.clone());
+
+    let http_server_handle = tokio::spawn(async move { http_server.start().await });
+
+    let _ = tokio::join!(http_server_handle);
+}
diff --git a/aggregation_mode/batcher_server/src/types.rs b/aggregation_mode/batcher_server/src/types.rs
diff --git a/aggregation_mode/payments_poller/Cargo.toml b/aggregation_mode/payments_poller/Cargo.toml
@@ -0,0 +1,18 @@
+[package]
+name = "agg_mode_batcher_poller"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+serde =  { workspace = true }
+serde_json = { workspace = true }
+serde_yaml = { workspace = true }
+aligned-sdk = { workspace = true }
+tracing = { version = "0.1", features = ["log"] }
+tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }
+actix-web = "4"
+alloy = { workspace = true }
+tokio = { version = "1", features = ["time"]}
+# TODO: enable tls
+sqlx = { version = "0.8", features = [ "runtime-tokio", "postgres", "uuid", "bigdecimal" ] }
+hex = "0.4"
diff --git a/aggregation_mode/payments_poller/abi/AggregationModePaymentService.json b/aggregation_mode/payments_poller/abi/AggregationModePaymentService.json
diff --git a/aggregation_mode/payments_poller/src/config.rs b/aggregation_mode/payments_poller/src/config.rs
@@ -4,7 +4,6 @@ use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct Config {
-    pub port: u16,
     pub db_connection_url: String,
     pub eth_rpc_url: String,
     pub payment_service_address: String,
diff --git a/aggregation_mode/payments_poller/src/db.rs b/aggregation_mode/payments_poller/src/db.rs
@@ -0,0 +1,46 @@
+use sqlx::{postgres::PgPoolOptions, types::BigDecimal, Pool, Postgres};
+
+#[derive(Clone, Debug)]
+pub struct Db {
+    pool: Pool<Postgres>,
+}
+
+#[derive(Debug, Clone)]
+pub enum DbError {
+    ConnectError(String),
+}
+
+impl Db {
+    pub async fn try_new(connection_url: &str) -> Result<Self, DbError> {
+        let pool = PgPoolOptions::new()
+            .max_connections(5)
+            .connect(connection_url)
+            .await
+            .map_err(|e| DbError::ConnectError(e.to_string()))?;
+
+        Ok(Self { pool })
+    }
+
+    pub async fn insert_payment_event(
+        &self,
+        address: &str,
+        started_at: &BigDecimal,
+        amount: &BigDecimal,
+        valid_until: &BigDecimal,
+        tx_hash: &str,
+    ) -> Result<(), sqlx::Error> {
+        sqlx::query(
+            "INSERT INTO payment_events (address, started_at, amount, valid_until, tx_hash)
+            VALUES ($1, $2, $3, $4, $5)
+            ON CONFLICT (tx_hash) DO NOTHING",
+        )
+        .bind(address.to_lowercase())
+        .bind(started_at)
+        .bind(amount)
+        .bind(valid_until)
+        .bind(tx_hash)
+        .execute(&self.pool)
+        .await
+        .map(|_| ())
+    }
+}
diff --git a/aggregation_mode/payments_poller/src/lib.rs b/aggregation_mode/payments_poller/src/lib.rs
@@ -0,0 +1,4 @@
+pub mod config;
+pub mod db;
+pub mod payments;
+pub mod types;
diff --git a/aggregation_mode/payments_poller/src/main.rs b/aggregation_mode/payments_poller/src/main.rs
@@ -1,8 +1,6 @@
 use std::env;
 
-use agg_mode_batcher::config::Config;
-use agg_mode_batcher::poller::payments::PaymentsPoller;
-use agg_mode_batcher::{db::Db, server::http::BatcherServer};
+use agg_mode_batcher_poller::{config::Config, db::Db, payments::PaymentsPoller};
 use tracing_subscriber::{EnvFilter, FmtSubscriber};
 
 fn read_config_filepath_from_args() -> String {
@@ -33,13 +31,8 @@ async fn main() {
         .expect("db to start");
 
     let payment_poller = PaymentsPoller::new(db.clone(), config.clone());
-    let http_server = BatcherServer::new(db, config.clone());
 
     let payment_poller_handle = tokio::spawn(async move { payment_poller.start().await });
-    let http_server_handle = tokio::spawn(async move { http_server.start().await });
 
-    // TODO: maybe this could two different processes (started with different commands) instead of being in the same one
-    // TODO: abort the process if one stops instead of waiting for them both
-    // TODO: ctrl + c handler for aborting the process should work
-    let _ = tokio::join!(payment_poller_handle, http_server_handle);
+    let _ = tokio::join!(payment_poller_handle);
 }
diff --git a/aggregation_mode/payments_poller/src/payments.rs b/aggregation_mode/payments_poller/src/payments.rs
@@ -1,16 +1,16 @@
 use std::str::FromStr;
 
-use crate::{config::Config, db::Db};
+use crate::{
+    config::Config,
+    db::Db,
+    types::{AggregationModePaymentService, AggregationModePaymentServiceContract, RpcProvider},
+};
 use alloy::{
     primitives::Address,
     providers::{Provider, ProviderBuilder},
 };
 use sqlx::types::BigDecimal;
 
-use crate::poller::types::{
-    AggregationModePaymentService, AggregationModePaymentServiceContract, RpcProvider,
-};
-
 pub struct PaymentsPoller {
     db: Db,
     proof_aggregation_service: AggregationModePaymentServiceContract,
diff --git a/aggregation_mode/payments_poller/src/types.rs b/aggregation_mode/payments_poller/src/types.rs
