Migrate Events Processor to Use New Version Tracker Impl

rust/integration-tests/src/models/queryable_models.rs

@@ -149,8 +149,8 @@ pub struct FungibleAssetMetadataModel {
     pub supply_aggregator_table_handle_v1: Option<String>,
     pub supply_aggregator_table_key_v1: Option<String>,
     pub token_standard: String,
-    pub is_token_v2: Option<bool>,
     pub inserted_at: chrono::NaiveDateTime,
+    pub is_token_v2: Option<bool>,
     pub supply_v2: Option<BigDecimal>,
     pub maximum_v2: Option<BigDecimal>,
 }

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

@@ -34,6 +34,7 @@ pub async fn setup_events_processor_config(
             processor_config,
             transaction_stream_config,
             db_config,
+            backfill_config: None,
         },
         processor_name,
     )

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

@@ -35,6 +35,7 @@ pub fn setup_fa_processor_config(
             processor_config,
             transaction_stream_config,
             db_config,
+            backfill_config: None,
         },
         processor_name,
     )

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

@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+DROP TABLE IF EXISTS backfill_processor_status;

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

@@ -0,0 +1,11 @@
+-- Your SQL goes here
+CREATE TABLE backfill_processor_status (
+    backfill_alias VARCHAR(50) NOT NULL,
+    backfill_status VARCHAR(50) NOT NULL,
+    last_success_version BIGINT NOT NULL,
+    last_updated TIMESTAMP NOT NULL DEFAULT NOW(),
+    last_transaction_timestamp TIMESTAMP NULL,
+    backfill_start_version BIGINT NOT NULL,
+    backfill_end_version BIGINT NULL,
+    PRIMARY KEY (backfill_alias)
+);

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

@@ -84,6 +84,20 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    backfill_processor_status (backfill_alias) {
+        #[max_length = 50]
+        backfill_alias -> Varchar,
+        #[max_length = 50]
+        backfill_status -> Varchar,
+        last_success_version -> Int8,
+        last_updated -> Timestamp,
+        last_transaction_timestamp -> Nullable<Timestamp>,
+        backfill_start_version -> Int8,
+        backfill_end_version -> Nullable<Int8>,
+    }
+}
+
 diesel::table! {
     block_metadata_transactions (version) {
         version -> Int8,
@@ -837,8 +851,8 @@ diesel::table! {
         supply_aggregator_table_key_v1 -> Nullable<Text>,
         #[max_length = 10]
         token_standard -> Varchar,
-        is_token_v2 -> Nullable<Bool>,
         inserted_at -> Timestamp,
+        is_token_v2 -> Nullable<Bool>,
         supply_v2 -> Nullable<Numeric>,
         maximum_v2 -> Nullable<Numeric>,
     }
@@ -1285,6 +1299,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     ans_lookup_v2,
     ans_primary_name,
     ans_primary_name_v2,
+    backfill_processor_status,
     block_metadata_transactions,
     coin_activities,
     coin_balances,

rust/sdk-processor/Cargo.toml

@@ -9,6 +9,7 @@ ahash = { workspace = true }
 anyhow = { workspace = true }
 aptos-indexer-processor-sdk = { workspace = true }
 aptos-indexer-processor-sdk-server-framework = { workspace = true }
+aptos-indexer-testing-framework = { workspace = true }
 async-trait = { workspace = true }
 bcs = { workspace = true }
 bigdecimal = { workspace = true }

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

@@ -19,6 +19,7 @@ pub struct IndexerProcessorConfig {
     pub processor_config: ProcessorConfig,
     pub transaction_stream_config: TransactionStreamConfig,
     pub db_config: DbConfig,
+    pub backfill_config: Option<BackfillConfig>,
 }
 
 #[async_trait::async_trait]
@@ -47,3 +48,9 @@ impl RunnableConfig for IndexerProcessorConfig {
         before_underscore[..before_underscore.len().min(12)].to_string()
     }
 }
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct BackfillConfig {
+    pub backfill_alias: String,
+}

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

@@ -0,0 +1,89 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+#![allow(clippy::extra_unused_lifetimes)]
+
+use crate::utils::database::DbPoolConnection;
+use diesel::{
+    deserialize,
+    deserialize::{FromSql, FromSqlRow},
+    expression::AsExpression,
+    pg::{Pg, PgValue},
+    serialize,
+    serialize::{IsNull, Output, ToSql},
+    sql_types::Text,
+    AsChangeset, ExpressionMethods, Insertable, OptionalExtension, QueryDsl, Queryable,
+};
+use diesel_async::RunQueryDsl;
+use processor::schema::backfill_processor_status;
+use std::io::Write;
+
+const IN_PROGRESS: &[u8] = b"in_progress";
+const COMPLETE: &[u8] = b"complete";
+
+#[derive(Debug, PartialEq, FromSqlRow, AsExpression, Eq)]
+#[diesel(sql_type = Text)]
+pub enum BackfillStatus {
+    // #[diesel(rename = "in_progress")]
+    InProgress,
+    // #[diesel(rename = "complete")]
+    Complete,
+}
+
+impl ToSql<Text, Pg> for BackfillStatus {
+    fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Pg>) -> serialize::Result {
+        match *self {
+            BackfillStatus::InProgress => out.write_all(IN_PROGRESS)?,
+            BackfillStatus::Complete => out.write_all(COMPLETE)?,
+        }
+        Ok(IsNull::No)
+    }
+}
+
+impl FromSql<Text, Pg> for BackfillStatus {
+    fn from_sql(bytes: PgValue<'_>) -> deserialize::Result<Self> {
+        match bytes.as_bytes() {
+            b"in_progress" => Ok(BackfillStatus::InProgress),
+            b"complete" => Ok(BackfillStatus::Complete),
+            _ => Err("Unrecognized enum variant".into()),
+        }
+    }
+}
+
+#[derive(AsChangeset, Debug, Insertable)]
+#[diesel(table_name = backfill_processor_status)]
+/// Only tracking the latest version successfully processed
+pub struct BackfillProcessorStatus {
+    pub backfill_alias: String,
+    pub backfill_status: BackfillStatus,
+    pub last_success_version: i64,
+    pub last_transaction_timestamp: Option<chrono::NaiveDateTime>,
+    pub backfill_start_version: i64,
+    pub backfill_end_version: Option<i64>,
+}
+
+#[derive(AsChangeset, Debug, Queryable)]
+#[diesel(table_name = backfill_processor_status)]
+/// Only tracking the latest version successfully processed
+pub struct BackfillProcessorStatusQuery {
+    pub backfill_alias: String,
+    pub backfill_status: BackfillStatus,
+    pub last_success_version: i64,
+    pub last_updated: chrono::NaiveDateTime,
+    pub last_transaction_timestamp: Option<chrono::NaiveDateTime>,
+    pub backfill_start_version: i64,
+    pub backfill_end_version: Option<i64>,
+}
+
+impl BackfillProcessorStatusQuery {
+    pub async fn get_by_processor(
+        backfill_alias: &str,
+        conn: &mut DbPoolConnection<'_>,
+    ) -> diesel::QueryResult<Option<Self>> {
+        backfill_processor_status::table
+            .filter(backfill_processor_status::backfill_alias.eq(backfill_alias))
+            .first::<Self>(conn)
+            .await
+            .optional()
+    }
+}

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

@@ -1,2 +1,3 @@
+pub mod backfill_processor_status;
 pub mod events_models;
 pub mod processor_status;

rust/sdk-processor/src/processors/events_processor.rs

@@ -4,9 +4,7 @@ use crate::{
         processor_config::ProcessorConfig,
     },
     steps::{
-        common::latest_processed_version_tracker::{
-            LatestVersionProcessedTracker, UPDATE_PROCESSOR_STATUS_SECS,
-        },
+        common::get_processor_status_saver,
         events_processor::{EventsExtractor, EventsStorer},
     },
     utils::{
@@ -19,10 +17,11 @@ use anyhow::Result;
 use aptos_indexer_processor_sdk::{
     aptos_indexer_transaction_stream::{TransactionStream, TransactionStreamConfig},
     builder::ProcessorBuilder,
-    common_steps::{OrderByVersionStep, TransactionStreamStep},
+    common_steps::{
+        TransactionStreamStep, VersionTrackerStep, DEFAULT_UPDATE_PROCESSOR_STATUS_SECS,
+    },
     traits::{processor_trait::ProcessorTrait, IntoRunnableStep},
 };
-use std::time::Duration;
 use tracing::{debug, info};
 
 pub struct EventsProcessor {
@@ -62,8 +61,6 @@ impl ProcessorTrait for EventsProcessor {
     }
 
     async fn run_processor(&self) -> Result<()> {
-        let processor_name = self.config.processor_config.name();
-
         // Run migrations
         match self.config.db_config {
             DbConfig::PostgresConfig(ref postgres_config) => {
@@ -104,20 +101,17 @@ impl ProcessorTrait for EventsProcessor {
         .await?;
         let events_extractor = EventsExtractor {};
         let events_storer = EventsStorer::new(self.db_pool.clone(), processor_config);
-        let order_step = OrderByVersionStep::new(
-            starting_version,
-            Duration::from_secs(UPDATE_PROCESSOR_STATUS_SECS),
+        let version_tracker = VersionTrackerStep::new(
+            get_processor_status_saver(self.db_pool.clone(), self.config.clone()),
+            DEFAULT_UPDATE_PROCESSOR_STATUS_SECS,
         );
-        let version_tracker =
-            LatestVersionProcessedTracker::new(self.db_pool.clone(), processor_name.to_string());
 
         // Connect processor steps together
         let (_, buffer_receiver) = ProcessorBuilder::new_with_inputless_first_step(
             transaction_stream.into_runnable_step(),
         )
         .connect_to(events_extractor.into_runnable_step(), channel_size)
         .connect_to(events_storer.into_runnable_step(), channel_size)
-        .connect_to(order_step.into_runnable_step(), channel_size)
         .connect_to(version_tracker.into_runnable_step(), channel_size)
         .end_and_return_output_receiver(channel_size);
 

rust/sdk-processor/src/processors/fungible_asset_processor.rs

@@ -4,9 +4,7 @@ use crate::{
         processor_config::ProcessorConfig,
     },
     steps::{
-        common::latest_processed_version_tracker::{
-            LatestVersionProcessedTracker, UPDATE_PROCESSOR_STATUS_SECS,
-        },
+        common::get_processor_status_saver,
         fungible_asset_processor::{
             fungible_asset_extractor::FungibleAssetExtractor,
             fungible_asset_storer::FungibleAssetStorer,
@@ -22,11 +20,12 @@ use anyhow::Result;
 use aptos_indexer_processor_sdk::{
     aptos_indexer_transaction_stream::{TransactionStream, TransactionStreamConfig},
     builder::ProcessorBuilder,
-    common_steps::{OrderByVersionStep, TransactionStreamStep},
+    common_steps::{
+        TransactionStreamStep, VersionTrackerStep, DEFAULT_UPDATE_PROCESSOR_STATUS_SECS,
+    },
     traits::{processor_trait::ProcessorTrait, IntoRunnableStep},
 };
 use processor::worker::TableFlags;
-use std::time::Duration;
 use tracing::{debug, info};
 
 pub struct FungibleAssetProcessor {
@@ -66,8 +65,6 @@ impl ProcessorTrait for FungibleAssetProcessor {
     }
 
     async fn run_processor(&self) -> Result<()> {
-        let processor_name = self.config.processor_config.name();
-
         //  Run migrations
         match self.config.db_config {
             DbConfig::PostgresConfig(ref postgres_config) => {
@@ -108,20 +105,16 @@ impl ProcessorTrait for FungibleAssetProcessor {
             processor_config.clone(),
             deprecated_table_flags,
         );
-        let order_step = OrderByVersionStep::new(
-            starting_version,
-            Duration::from_secs(UPDATE_PROCESSOR_STATUS_SECS),
+        let version_tracker = VersionTrackerStep::new(
+            get_processor_status_saver(self.db_pool.clone(), self.config.clone()),
+            DEFAULT_UPDATE_PROCESSOR_STATUS_SECS,
         );
-        let version_tracker =
-            LatestVersionProcessedTracker::new(self.db_pool.clone(), processor_name.to_string());
-
         // Connect processor steps together
         let (_, buffer_receiver) = ProcessorBuilder::new_with_inputless_first_step(
             transaction_stream.into_runnable_step(),
         )
         .connect_to(fa_extractor.into_runnable_step(), channel_size)
         .connect_to(fa_storer.into_runnable_step(), channel_size)
-        .connect_to(order_step.into_runnable_step(), channel_size)
         .connect_to(version_tracker.into_runnable_step(), channel_size)
         .end_and_return_output_receiver(channel_size);
 

rust/sdk-processor/src/steps/common/mod.rs

@@ -1 +1,4 @@
 pub mod latest_processed_version_tracker;
+pub mod processor_status_saver;
+
+pub use processor_status_saver::get_processor_status_saver;