Make backfill behavior more explicit

rust/integration-tests/src/sdk_tests/account_restoration_processor_tests.rs

@@ -2,7 +2,7 @@ use ahash::AHashMap;
 use aptos_indexer_testing_framework::sdk_test_context::SdkTestContext;
 use sdk_processor::config::{
     db_config::{DbConfig, PostgresConfig},
-    indexer_processor_config::IndexerProcessorConfig,
+    indexer_processor_config::{IndexerProcessorConfig, ProcessorMode},
     processor_config::{DefaultProcessorConfig, ProcessorConfig},
 };
 use std::collections::HashSet;
@@ -33,6 +33,9 @@ pub fn setup_account_restoration_processor_config(
             transaction_stream_config,
             db_config,
             backfill_config: None,
+            bootstrap_config: None,
+            testing_config: None,
+            mode: ProcessorMode::Default,
         },
         processor_name,
     )

rust/integration-tests/src/sdk_tests/account_transaction_processor_tests.rs

@@ -2,7 +2,7 @@ use ahash::AHashMap;
 use aptos_indexer_testing_framework::sdk_test_context::SdkTestContext;
 use sdk_processor::config::{
     db_config::{DbConfig, PostgresConfig},
-    indexer_processor_config::IndexerProcessorConfig,
+    indexer_processor_config::{IndexerProcessorConfig, ProcessorMode},
     processor_config::{DefaultProcessorConfig, ProcessorConfig},
 };
 use std::collections::HashSet;
@@ -32,6 +32,9 @@ pub fn setup_acc_txn_processor_config(
             transaction_stream_config,
             db_config,
             backfill_config: None,
+            bootstrap_config: None,
+            testing_config: None,
+            mode: ProcessorMode::Default,
         },
         processor_name,
     )

rust/integration-tests/src/sdk_tests/ans_processor_tests.rs

@@ -3,7 +3,7 @@ use aptos_indexer_testing_framework::sdk_test_context::SdkTestContext;
 use sdk_processor::{
     config::{
         db_config::{DbConfig, PostgresConfig},
-        indexer_processor_config::IndexerProcessorConfig,
+        indexer_processor_config::{IndexerProcessorConfig, ProcessorMode},
         processor_config::{DefaultProcessorConfig, ProcessorConfig},
     },
     processors::ans_processor::AnsProcessorConfig,
@@ -44,6 +44,9 @@ pub fn setup_ans_processor_config(
             transaction_stream_config,
             db_config,
             backfill_config: None,
+            bootstrap_config: None,
+            testing_config: None,
+            mode: ProcessorMode::Default,
         },
         processor_name,
     )

rust/integration-tests/src/sdk_tests/default_processor_tests.rs

@@ -2,7 +2,7 @@ use ahash::AHashMap;
 use aptos_indexer_testing_framework::sdk_test_context::SdkTestContext;
 use sdk_processor::config::{
     db_config::{DbConfig, PostgresConfig},
-    indexer_processor_config::IndexerProcessorConfig,
+    indexer_processor_config::{IndexerProcessorConfig, ProcessorMode},
     processor_config::{DefaultProcessorConfig, ProcessorConfig},
 };
 use std::collections::HashSet;
@@ -32,6 +32,9 @@ pub fn setup_default_processor_config(
             transaction_stream_config,
             db_config,
             backfill_config: None,
+            bootstrap_config: None,
+            testing_config: None,
+            mode: ProcessorMode::Default,
         },
         processor_name,
     )

rust/integration-tests/src/sdk_tests/events_processor_tests.rs

@@ -2,7 +2,7 @@ use ahash::AHashMap;
 use aptos_indexer_testing_framework::sdk_test_context::SdkTestContext;
 use sdk_processor::config::{
     db_config::{DbConfig, PostgresConfig},
-    indexer_processor_config::IndexerProcessorConfig,
+    indexer_processor_config::{IndexerProcessorConfig, ProcessorMode},
     processor_config::{DefaultProcessorConfig, ProcessorConfig},
 };
 use std::collections::HashSet;
@@ -32,6 +32,9 @@ pub fn setup_events_processor_config(
             transaction_stream_config,
             db_config,
             backfill_config: None,
+            bootstrap_config: None,
+            testing_config: None,
+            mode: ProcessorMode::Default,
         },
         processor_name,
     )

rust/integration-tests/src/sdk_tests/fungible_asset_processor_tests.rs

@@ -2,7 +2,7 @@ use ahash::AHashMap;
 use aptos_indexer_testing_framework::sdk_test_context::SdkTestContext;
 use sdk_processor::config::{
     db_config::{DbConfig, PostgresConfig},
-    indexer_processor_config::IndexerProcessorConfig,
+    indexer_processor_config::{IndexerProcessorConfig, ProcessorMode},
     processor_config::{DefaultProcessorConfig, ProcessorConfig},
 };
 use std::collections::HashSet;
@@ -33,6 +33,9 @@ pub fn setup_fa_processor_config(
             transaction_stream_config,
             db_config,
             backfill_config: None,
+            bootstrap_config: None,
+            testing_config: None,
+            mode: ProcessorMode::Default,
         },
         processor_name,
     )

rust/integration-tests/src/sdk_tests/objects_processor_tests.rs

@@ -3,7 +3,7 @@ use aptos_indexer_testing_framework::sdk_test_context::SdkTestContext;
 use sdk_processor::{
     config::{
         db_config::{DbConfig, PostgresConfig},
-        indexer_processor_config::IndexerProcessorConfig,
+        indexer_processor_config::{IndexerProcessorConfig, ProcessorMode},
         processor_config::{DefaultProcessorConfig, ProcessorConfig},
     },
     processors::objects_processor::ObjectsProcessorConfig,
@@ -43,6 +43,9 @@ pub fn setup_objects_processor_config(
             transaction_stream_config,
             db_config,
             backfill_config: None,
+            bootstrap_config: None,
+            testing_config: None,
+            mode: ProcessorMode::Default,
         },
         processor_name,
     )

rust/integration-tests/src/sdk_tests/stake_processor_tests.rs

@@ -3,7 +3,7 @@ use aptos_indexer_testing_framework::sdk_test_context::SdkTestContext;
 use sdk_processor::{
     config::{
         db_config::{DbConfig, PostgresConfig},
-        indexer_processor_config::IndexerProcessorConfig,
+        indexer_processor_config::{IndexerProcessorConfig, ProcessorMode},
         processor_config::{DefaultProcessorConfig, ProcessorConfig},
     },
     processors::stake_processor::StakeProcessorConfig,
@@ -40,6 +40,9 @@ pub fn setup_stake_processor_config(
             transaction_stream_config,
             db_config,
             backfill_config: None,
+            bootstrap_config: None,
+            testing_config: None,
+            mode: ProcessorMode::Default,
         },
         processor_name,
     )

rust/integration-tests/src/sdk_tests/token_v2_processor_tests.rs

@@ -3,7 +3,7 @@ use aptos_indexer_testing_framework::sdk_test_context::SdkTestContext;
 use sdk_processor::{
     config::{
         db_config::{DbConfig, PostgresConfig},
-        indexer_processor_config::IndexerProcessorConfig,
+        indexer_processor_config::{IndexerProcessorConfig, ProcessorMode},
         processor_config::{DefaultProcessorConfig, ProcessorConfig},
     },
     processors::token_v2_processor::TokenV2ProcessorConfig,
@@ -42,6 +42,9 @@ pub fn setup_token_v2_processor_config(
             transaction_stream_config,
             db_config,
             backfill_config: None,
+            bootstrap_config: None,
+            testing_config: None,
+            mode: ProcessorMode::Default,
         },
         processor_name,
     )

rust/integration-tests/src/sdk_tests/user_transaction_processor_tests.rs

@@ -2,7 +2,7 @@ use ahash::AHashMap;
 use aptos_indexer_testing_framework::sdk_test_context::SdkTestContext;
 use sdk_processor::config::{
     db_config::{DbConfig, PostgresConfig},
-    indexer_processor_config::IndexerProcessorConfig,
+    indexer_processor_config::{IndexerProcessorConfig, ProcessorMode},
     processor_config::{DefaultProcessorConfig, ProcessorConfig},
 };
 use std::collections::HashSet;
@@ -33,6 +33,9 @@ pub fn setup_user_txn_processor_config(
             transaction_stream_config,
             db_config,
             backfill_config: None,
+            bootstrap_config: None,
+            testing_config: None,
+            mode: ProcessorMode::Default,
         },
         processor_name,
     )

rust/processor/parser.yaml

@@ -5,6 +5,6 @@ health_check_port: 8084
 server_config:
   processor_config:
     type: default_processor
-  postgres_connection_string: postgresql://postgres:@localhost:5432/default_processor
-  indexer_grpc_data_service_address: http://127.0.0.1:50051
-  auth_token: AUTH_TOKEN
+    postgres_connection_string: postgresql://postgres:@localhost:5432/default_processor
+    indexer_grpc_data_service_address: http://127.0.0.1:50051
+    auth_token: AUTH_TOKEN

rust/processor/src/db/postgres/migrations/2024-10-22-214753_backfill_processor_status/up.sql

@@ -6,6 +6,6 @@ CREATE TABLE backfill_processor_status (
     last_updated TIMESTAMP NOT NULL DEFAULT NOW(),
     last_transaction_timestamp TIMESTAMP NULL,
     backfill_start_version BIGINT NOT NULL,
-    backfill_end_version BIGINT NULL,
+    backfill_end_version BIGINT NOT NULL,
     PRIMARY KEY (backfill_alias)
 );

rust/processor/src/db/postgres/schema.rs

@@ -117,7 +117,7 @@ diesel::table! {
         last_updated -> Timestamp,
         last_transaction_timestamp -> Nullable<Timestamp>,
         backfill_start_version -> Int8,
-        backfill_end_version -> Nullable<Int8>,
+        backfill_end_version -> Int8,
     }
 }
 

rust/sdk-processor/src/config/indexer_processor_config.rs

@@ -36,13 +36,84 @@ use serde::{Deserialize, Serialize};
 pub const QUERY_DEFAULT_RETRIES: u32 = 5;
 pub const QUERY_DEFAULT_RETRY_DELAY_MS: u64 = 500;
 
-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize, Default)]
+#[serde(deny_unknown_fields)]
+pub enum ProcessorMode {
+    #[serde(rename = "default")]
+    #[default]
+    Default,
+    #[serde(rename = "backfill")]
+    Backfill,
+    #[serde(rename = "testing")]
+    Testing,
+}
+
+#[derive(Clone, Debug, Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct IndexerProcessorConfig {
     pub processor_config: ProcessorConfig,
     pub transaction_stream_config: TransactionStreamConfig,
     pub db_config: DbConfig,
     pub backfill_config: Option<BackfillConfig>,
+    pub bootstrap_config: Option<BootStrapConfig>,
+    pub testing_config: Option<TestingConfig>,
+    #[serde(default)]
+    pub mode: ProcessorMode,
+}
+
+impl IndexerProcessorConfig {
+    fn validate(&self) -> Result<(), String> {
+        match self.mode {
+            ProcessorMode::Testing => {
+                if self.testing_config.is_none() {
+                    return Err("testing_config must be present when mode is 'testing'".to_string());
+                }
+            },
+            ProcessorMode::Backfill => {
+                if self.backfill_config.is_none() {
+                    return Err(
+                        "backfill_config must be present when mode is 'backfill'".to_string()
+                    );
+                }
+            },
+            ProcessorMode::Default => {},
+        }
+        Ok(())
+    }
+}
+
+impl<'de> Deserialize<'de> for IndexerProcessorConfig {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        #[derive(Deserialize)]
+        #[serde(deny_unknown_fields)]
+        struct Inner {
+            processor_config: ProcessorConfig,
+            transaction_stream_config: TransactionStreamConfig,
+            db_config: DbConfig,
+            backfill_config: Option<BackfillConfig>,
+            bootstrap_config: Option<BootStrapConfig>,
+            testing_config: Option<TestingConfig>,
+            #[serde(default)]
+            mode: ProcessorMode,
+        }
+
+        let inner = Inner::deserialize(deserializer)?;
+        let config = IndexerProcessorConfig {
+            processor_config: inner.processor_config,
+            transaction_stream_config: inner.transaction_stream_config,
+            db_config: inner.db_config,
+            backfill_config: inner.backfill_config,
+            bootstrap_config: inner.bootstrap_config,
+            testing_config: inner.testing_config,
+            mode: inner.mode,
+        };
+
+        config.validate().map_err(serde::de::Error::custom)?;
+        Ok(config)
+    }
 }
 
 #[async_trait::async_trait]
@@ -155,5 +226,21 @@ impl RunnableConfig for IndexerProcessorConfig {
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct BackfillConfig {
-    pub backfill_alias: String,
+    pub backfill_id: String,
+    pub initial_starting_version: u64,
+    pub ending_version: u64,
+    pub overwrite_checkpoint: bool,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct BootStrapConfig {
+    pub initial_starting_version: u64,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct TestingConfig {
+    pub override_starting_version: u64,
+    pub ending_version: u64,
 }

rust/sdk-processor/src/db/common/models/backfill_processor_status.rs

@@ -59,7 +59,7 @@ pub struct BackfillProcessorStatus {
     pub last_success_version: i64,
     pub last_transaction_timestamp: Option<chrono::NaiveDateTime>,
     pub backfill_start_version: i64,
-    pub backfill_end_version: Option<i64>,
+    pub backfill_end_version: i64,
 }
 
 #[derive(AsChangeset, Debug, Queryable)]
@@ -72,14 +72,16 @@ pub struct BackfillProcessorStatusQuery {
     pub last_updated: chrono::NaiveDateTime,
     pub last_transaction_timestamp: Option<chrono::NaiveDateTime>,
     pub backfill_start_version: i64,
-    pub backfill_end_version: Option<i64>,
+    pub backfill_end_version: i64,
 }
 
 impl BackfillProcessorStatusQuery {
     pub async fn get_by_processor(
-        backfill_alias: &str,
+        processor_type: &str,
+        backfill_id: &str,
         conn: &mut DbPoolConnection<'_>,
     ) -> diesel::QueryResult<Option<Self>> {
+        let backfill_alias = format!("{}_{}", processor_type, backfill_id);
         backfill_processor_status::table
             .filter(backfill_processor_status::backfill_alias.eq(backfill_alias))
             .first::<Self>(conn)