Refactor

Author: vutran1710

src/components/block_source/delta/mod.rs

@@ -25,6 +25,7 @@ pub struct DeltaClient {
     ctx: SessionContext,
     start_block: u64,
     query_step: u64,
+    block_per_file: u32,
 }
 
 impl DeltaClient {
@@ -57,6 +58,7 @@ impl DeltaClient {
             ctx,
             start_block,
             query_step: cfg.query_step,
+            block_per_file: cfg.block_per_file,
         })
     }
 
@@ -82,13 +84,12 @@ impl DeltaClient {
         sender: AsyncSender<Vec<BlockDataMessage>>,
         valve: Valve,
     ) -> Result<(), SourceError> {
-        let mut start_block = self.start_block;
+        let mut start_block = self.start_block - (self.start_block % self.block_per_file as u64);
         info!(DeltaClient, "source start collecting data");
 
         loop {
             while !valve.should_continue() {
                 let sleep_tine = valve.get_wait();
-                info!(DeltaClient, "source sleeping"; sleep_time => sleep_tine);
                 tokio::time::sleep(Duration::from_secs(sleep_tine)).await;
             }
 
@@ -135,6 +136,14 @@ mod test {
     use super::*;
     use crate::config::ValveConfig;
 
+    #[test]
+    fn test_adjust_start_block() {
+        let actual_start_block = 10_124_125;
+        let block_per_file = 2000;
+        let adjusted_start_block = actual_start_block - (actual_start_block % block_per_file);
+        assert_eq!(adjusted_start_block, 10_124_000);
+    }
+
     #[tokio::test]
     async fn test_delta() {
         env_logger::try_init().unwrap_or_default();
@@ -143,6 +152,7 @@ mod test {
             table_path: "s3://ethereum/blocks_01/".to_owned(),
             query_step: 10,
             version: None,
+            block_per_file: 2,
         };
 
         let client = DeltaClient::new(cfg, 10_000_000).await.unwrap();

src/components/subgraph/mod.rs

@@ -2,12 +2,10 @@ mod datasource_wasm_instance;
 mod metrics;
 
 use super::ManifestLoader;
-use super::Valve;
 use crate::chain::ethereum::block::EthereumBlockData;
 use crate::common::HandlerTypes;
 use crate::config::Config;
 use crate::database::DatabaseAgent;
-use crate::error;
 use crate::errors::SubgraphError;
 use crate::info;
 use crate::messages::EthereumFilteredEvent;
@@ -18,11 +16,9 @@ use datasource_wasm_instance::DatasourceWasmInstance;
 use metrics::SubgraphMetrics;
 use prometheus::Registry;
 use std::collections::HashMap;
-use std::time::Instant;
 
 pub struct Subgraph {
     // NOTE: using IPFS might lead to subgraph-id using a hex/hash
-    pub id: String,
     pub name: String,
     sources: HashMap<String, DatasourceWasmInstance>,
     metrics: SubgraphMetrics,
@@ -33,7 +29,6 @@ impl Subgraph {
         Self {
             sources: HashMap::new(),
             name: config.subgraph_name.clone(),
-            id: config.get_subgraph_id(),
             metrics: SubgraphMetrics::new(registry),
         }
     }
@@ -109,17 +104,9 @@ impl Subgraph {
         }
     }
 
-    pub async fn process(
-        &mut self,
-        msg: FilteredDataMessage,
-        db_agent: &DatabaseAgent,
-        rpc_agent: &RpcAgent,
-        valve: &Valve,
-    ) -> Result<(), SubgraphError> {
+    pub async fn process(&mut self, msg: FilteredDataMessage) -> Result<(), SubgraphError> {
         let block_ptr = msg.get_block_ptr();
 
-        rpc_agent.set_block_ptr(block_ptr.clone()).await;
-
         self.metrics
             .current_block_number
             .set(block_ptr.number as i64);
@@ -136,29 +123,6 @@ impl Subgraph {
                 block_number => block_ptr.number,
                 block_hash => block_ptr.hash
             );
-            valve.set_finished(block_ptr.number);
-        }
-
-        if block_ptr.number % 2000 == 0 {
-            info!(Subgraph, "commiting data to DB"; block_number => block_ptr.number);
-            let time = Instant::now();
-            db_agent.migrate(block_ptr.clone()).await.map_err(|e| {
-                error!(
-                    Subgraph, "Failed to commit data to DB";
-                    error => e.to_string(),
-                    block_number => block_ptr.number,
-                    block_hash => block_ptr.hash
-                );
-                SubgraphError::MigrateDbError
-            })?;
-
-            info!(Subgraph, "data committed to database"; execution_time => format!("{:?}", time.elapsed()));
-
-            db_agent
-                .clear_in_memory()
-                .await
-                .map_err(|_| SubgraphError::MigrateDbError)?;
-            info!(Subgraph, "flushed entity cache");
         }
 
         Ok(())

src/config.rs

@@ -10,6 +10,7 @@ pub struct DeltaConfig {
     pub table_path: String,
     pub query_step: u64,
     pub version: Option<u64>,
+    pub block_per_file: u32,
 }
 
 #[derive(Deserialize, Clone, Debug)]
@@ -36,7 +37,6 @@ pub struct Config {
     pub chain: Chain,
     pub source: SourceTypes,
     pub subgraph_name: String,
-    pub subgraph_id: Option<String>,
     pub subgraph_dir: String,
     pub database: DatabaseConfig,
     pub reorg_threshold: u16,
@@ -54,13 +54,6 @@ impl Config {
             .extract()
             .expect("Load config failed")
     }
-
-    pub fn get_subgraph_id(&self) -> String {
-        self.subgraph_id
-            .clone()
-            .unwrap_or(self.subgraph_name.to_owned())
-            .to_owned()
-    }
 }
 
 #[cfg(test)]

src/errors.rs

@@ -103,8 +103,6 @@ pub enum SubgraphError {
     InvalidSourceID(String),
     #[error("Invalid handler_name: {0}")]
     InvalidHandlerName(String),
-    #[error("Migrate memory to db error")]
-    MigrateDbError,
     #[error("Create source failed: `{0}`")]
     CreateSourceFail(String),
 }

src/main.rs

@@ -76,6 +76,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             let time = std::time::Instant::now();
             let sorted_blocks = filter.filter_multi(blocks)?;
             let count_blocks = sorted_blocks.len();
+            let last_block = sorted_blocks.last().map(|b| b.get_block_ptr());
 
             info!(
                 MainFlow,
@@ -88,6 +89,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
             for block in sorted_blocks {
                 let block_ptr = block.get_block_ptr();
+                rpc.set_block_ptr(&block_ptr).await;
 
                 match inspector.check_block(block_ptr.clone()) {
                     BlockInspectionResult::UnexpectedBlock
@@ -105,15 +107,21 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 };
 
                 subgraph.create_sources(&manifest, &db, &rpc).await?;
-                subgraph.process(block, &db, &rpc, &valve).await?;
+                subgraph.process(block).await?;
+            }
+
+            if let Some(block_ptr) = last_block {
+                db.commit_data(block_ptr.clone()).await?;
+                db.flush_cache().await?;
+                valve.set_finished(block_ptr.number);
             }
 
             info!(
                 MainFlow,
                 "block batch processed done";
                 exec_time => format!("{:?}", time.elapsed()),
                 number_of_blocks => count_blocks,
-                avg_speed => format!("~{:?} blocks/sec", (count_blocks as u64 / time.elapsed().as_secs()) as u64)
+                avg_speed => format!("~{:?} blocks/sec", { count_blocks as u64 / time.elapsed().as_secs() })
             );
         }