New delta trino (#82)

* Add trino & delta

* up

* up

* up

* up

* up

* up

* clippy fix

* update

* remove not needed

* up

* up

* up

* up

* update

* update

* update

* up

* refactor

* custom table name

* improve error

* refactor logic

* refactor

* naming

* remove not needed serde trait

* format

* refactor

* refactor error

---------

Co-authored-by: quannguyen <[email protected]>

Author: vutran1710

Cargo.lock

@@ -37,6 +37,8 @@ scylla = "0.10.1"
 apollo-parser = "0.7.3"
 prometheus = "0.13.3"
 warp = "0.3.6"
+prusto = "0.5.1"
+tokio-retry = "0.3.0"
 
 [dev-dependencies]
 async-std = { version = "1.12.0", features = ["attributes"] }

Cargo.toml

@@ -8,6 +8,6 @@ mod subgraph_filter;
 pub use manifest_loader::ManifestLoader;
 pub use progress_ctrl::ProgressCtrl;
 pub use serializer::Serializer;
-pub use source::Source;
+pub use source::BlockSource;
 pub use subgraph::Subgraph;
 pub use subgraph_filter::SubgraphFilter;

src/components/mod.rs

@@ -7,6 +7,7 @@ use crate::database::DatabaseAgent;
 use crate::errors::ProgressCtrlError;
 use crate::messages::SerializedDataMessage;
 
+#[derive(Clone)]
 pub struct ProgressCtrl {
     db: DatabaseAgent,
     recent_block_ptrs: Vec<BlockPtr>,
@@ -31,9 +32,15 @@ impl ProgressCtrl {
         Ok(this)
     }
 
-    fn get_min_start_block(&self) -> u64 {
+    pub fn get_min_start_block(&self) -> u64 {
         let min_start_block = self.sources.iter().filter_map(|s| s.startBlock).min();
-        min_start_block.unwrap_or(0)
+        min_start_block.unwrap_or(0).max(
+            self.recent_block_ptrs
+                .last()
+                .cloned()
+                .map(|b| b.number + 1)
+                .unwrap_or_default(),
+        )
     }
 
     pub async fn progress_check(