feat(aggregation-mode): store last fetched block in payments poller

Author: maximopalopoli

aggregation_mode/payments_poller/src/config.rs

@@ -1,4 +1,4 @@
-use std::{fs::File, io::Read};
+use std::{fs::File, fs::OpenOptions, io::Read, io::Write};
 
 use serde::{Deserialize, Serialize};
 
@@ -7,6 +7,12 @@ pub struct Config {
     pub db_connection_url: String,
     pub eth_rpc_url: String,
     pub payment_service_address: String,
+    pub last_block_fetched_filepath: String,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct LastBlockFetched {
+    pub last_block_fetched: u64,
 }
 
 impl Config {
@@ -17,4 +23,30 @@ impl Config {
         let config: Config = serde_yaml::from_str(&contents)?;
         Ok(config)
     }
+
+    pub fn get_last_block_fetched(&self) -> Result<u64, Box<dyn std::error::Error>> {
+        let mut file = File::open(&self.last_block_fetched_filepath)?;
+        let mut contents = String::new();
+        file.read_to_string(&mut contents)?;
+        let lbf_struct: LastBlockFetched = serde_json::from_str(&contents)?;
+        Ok(lbf_struct.last_block_fetched)
+    }
+
+    pub fn update_last_block_fetched(
+        &self,
+        last_block_fetched: u64,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let last_block_fetched_struct = LastBlockFetched { last_block_fetched };
+
+        let mut file = OpenOptions::new()
+            .write(true)
+            .truncate(true)
+            .create(true)
+            .open(&self.last_block_fetched_filepath)?;
+
+        let content = serde_json::to_string(&last_block_fetched_struct)?;
+        file.write_all(content.as_bytes())?;
+
+        Ok(())
+    }
 }

aggregation_mode/payments_poller/src/main.rs

@@ -30,6 +30,13 @@ async fn main() {
         .await
         .expect("db to start");
 
-    let payment_poller = PaymentsPoller::new(db, config);
-    payment_poller.start().await;
+    let payments_poller = match PaymentsPoller::new(db, config) {
+        Ok(poller) => poller,
+        Err(err) => {
+            tracing::error!("Failed to create Payments Poller: {err}");
+            return;
+        }
+    };
+
+    payments_poller.start().await;
 }

aggregation_mode/payments_poller/src/payments.rs

@@ -10,15 +10,17 @@ use alloy::{
     providers::{Provider, ProviderBuilder},
 };
 use sqlx::types::BigDecimal;
+use tracing::info;
 
 pub struct PaymentsPoller {
     db: Db,
     proof_aggregation_service: AggregationModePaymentServiceContract,
     rpc_provider: RpcProvider,
+    config: Config,
 }
 
 impl PaymentsPoller {
-    pub fn new(db: Db, config: Config) -> Self {
+    pub fn new(db: Db, config: Config) -> Result<Self, Box<dyn std::error::Error>> {
         let rpc_url = config.eth_rpc_url.parse().expect("RPC URL should be valid");
         let rpc_provider = ProviderBuilder::new().connect_http(rpc_url);
         let proof_aggregation_service = AggregationModePaymentService::new(
@@ -27,16 +29,30 @@ impl PaymentsPoller {
             rpc_provider.clone(),
         );
 
-        Self {
+        // This check is here to catch early failures on last block fetching
+        let _ = config.get_last_block_fetched()?;
+
+        Ok(Self {
             db,
             proof_aggregation_service,
             rpc_provider,
-        }
+            config,
+        })
     }
 
     pub async fn start(&self) {
         let seconds_to_wait_between_polls = 12;
+
         loop {
+            let Ok(last_block_fetched) = self.config.get_last_block_fetched() else {
+                tracing::warn!("Could not get last block fetched, skipping polling iteration...");
+                tokio::time::sleep(std::time::Duration::from_secs(
+                    seconds_to_wait_between_polls,
+                ))
+                .await;
+                continue;
+            };
+
             let Ok(current_block) = self.rpc_provider.get_block_number().await else {
                 tracing::warn!("Could not get current block skipping polling iteration...");
                 tokio::time::sleep(std::time::Duration::from_secs(
@@ -46,10 +62,13 @@ impl PaymentsPoller {
                 continue;
             };
 
+            let from = last_block_fetched.saturating_sub(5);
+            info!("Fetching logs from block {from} to {current_block}");
+
             let Ok(logs) = self
                 .proof_aggregation_service
                 .UserPayment_filter()
-                .from_block(current_block - 5)
+                .from_block(last_block_fetched - 5)
                 .to_block(current_block)
                 .query()
                 .await
@@ -91,6 +110,11 @@ impl PaymentsPoller {
                 }
             }
 
+            if let Err(err) = self.config.update_last_block_fetched(current_block) {
+                tracing::error!("Failed to update the last aggregated block: {err}");
+                continue;
+            };
+
             tokio::time::sleep(std::time::Duration::from_secs(
                 seconds_to_wait_between_polls,
             ))

config-files/config-agg-mode-gateway.yaml

@@ -4,3 +4,4 @@ eth_rpc_url: "http://localhost:8545"
 payment_service_address: "0x922D6956C99E12DFeB3224DEA977D0939758A1Fe"
 network: "devnet"
 max_daily_proofs_per_user: 4
+last_block_fetched_filepath: "config-files/proof-aggregator.last_block_fetched.json"

config-files/proof-aggregator.last_block_fetched.json

@@ -0,0 +1 @@
+{"last_block_fetched":0}