Refactor indexer version config

rust/processor/README.md

@@ -26,6 +26,7 @@ Indexer GRPC parser is to indexer data processor that leverages the indexer grpc
         number_concurrent_processing_tasks: 10
         auth_token: AUTH_TOKEN
         starting_version: 0 # optional
+        starting_version_if_nothing_in_db: 0 # optional
         ending_version: 0 # optional
         transaction_filter:
           # Only allow transactions from these contract addresses
@@ -46,7 +47,8 @@ Indexer GRPC parser is to indexer data processor that leverages the indexer grpc
 - `indexer_grpc_http2_ping_interval_in_secs`: client-side grpc HTTP2 ping interval.
 - `indexer_grpc_http2_ping_timeout_in_secs`: client-side grpc HTTP2 ping timeout.
 - `auth_token`: Auth token used for connection.
-- `starting_version`: start processor at starting_version.
+- `starting_version`: start processor at starting_version regardless of the database version.
+- `starting_version_if_nothing_in_db`: start processor at the starting_version if nothing is set in the database.
 - `ending_version`: stop processor after ending_version.
 - `number_concurrent_processing_tasks`: number of tasks to parse and insert; 1 means sequential processing, otherwise,
   transactions are splitted into tasks and inserted with random order.

rust/processor/src/config.rs

@@ -27,6 +27,8 @@ pub struct IndexerGrpcProcessorConfig {
     pub auth_token: String,
     // Version to start indexing from
     pub starting_version: Option<u64>,
+    // Version to start indexing from if nothing in db
+    pub starting_version_if_nothing_in_db: Option<u64>,
     // Version to end indexing at
     pub ending_version: Option<u64>,
     // Number of tasks waiting to pull transaction batches from the channel and process them
@@ -77,6 +79,7 @@ impl RunnableConfig for IndexerGrpcProcessorConfig {
             self.grpc_http2_config.clone(),
             self.auth_token.clone(),
             self.starting_version,
+            self.starting_version_if_nothing_in_db,
             self.ending_version,
             self.number_concurrent_processing_tasks,
             self.db_pool_size,

rust/processor/src/worker.rs

@@ -55,6 +55,7 @@ pub struct Worker {
     pub grpc_http2_config: IndexerGrpcHttp2Config,
     pub auth_token: String,
     pub starting_version: Option<u64>,
+    pub starting_version_if_nothing_in_db: Option<u64>,
     pub ending_version: Option<u64>,
     pub number_concurrent_processing_tasks: usize,
     pub gap_detection_batch_size: u64,
@@ -73,6 +74,7 @@ impl Worker {
         grpc_http2_config: IndexerGrpcHttp2Config,
         auth_token: String,
         starting_version: Option<u64>,
+        starting_version_if_nothing_in_db: Option<u64>,
         ending_version: Option<u64>,
         number_concurrent_processing_tasks: Option<usize>,
         db_pool_size: Option<u32>,
@@ -107,6 +109,7 @@ impl Worker {
             indexer_grpc_data_service_address,
             grpc_http2_config,
             starting_version,
+            starting_version_if_nothing_in_db,
             ending_version,
             auth_token,
             number_concurrent_processing_tasks,
@@ -141,27 +144,37 @@ impl Worker {
             "[Parser] Finished migrations"
         );
 
-        let starting_version_from_db = self
+        let mut starting_version_from_db = None;
+
+        let mut starting_version = match self
             .get_start_version()
             .await
-            .expect("[Parser] Database error when getting starting version")
-            .unwrap_or_else(|| {
+            .expect("[Parser] Database error when getting starting version") {
+            None => {
                 info!(
                     processor_name = processor_name,
                     service_type = PROCESSOR_SERVICE_TYPE,
                     "[Parser] No starting version from db so starting from version 0"
                 );
-                0
-            });
+                self.starting_version_if_nothing_in_db.unwrap_or(0)
+            }
+            Some(version) => {
+                starting_version_from_db = Some(version);
+                version
+            }
+        };
 
-        let starting_version = self.starting_version.unwrap_or(starting_version_from_db);
+        if let Some(force_start_version) = self.starting_version {
+            starting_version = force_start_version
+        }
 
         info!(
             processor_name = processor_name,
             service_type = PROCESSOR_SERVICE_TYPE,
             stream_address = self.indexer_grpc_data_service_address.to_string(),
             final_start_version = starting_version,
             start_version_from_config = self.starting_version,
+            starting_version_if_nothing_in_db = self.starting_version_if_nothing_in_db,
             start_version_from_db = starting_version_from_db,
             "[Parser] Building processor",
         );
@@ -746,4 +759,4 @@ pub fn build_processor(
             UserTransactionProcessor::new(db_pool, per_table_chunk_sizes),
         ),
     }
-}
+}