upgrade dep

Author: rtso

rust/Cargo.lock

@@ -239,7 +239,7 @@ fn insert_fungible_asset_balances_query(
     )
 }
 
-fn insert_current_fungible_asset_balances_query(
+pub fn insert_current_fungible_asset_balances_query(
     items_to_insert: Vec<CurrentFungibleAssetBalance>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,

rust/processor/src/processors/fungible_asset_processor.rs

@@ -136,8 +136,13 @@ impl EventsProcessor {
                         continue;
                     }
                     debug!(
-                        "Finished processing events from versions [{:?}, {:?}]",
-                        txn_context.start_version, txn_context.end_version,
+                        "Finished processing events from versions {}",
+                        txn_context
+                            .get_versions()
+                            .iter()
+                            .map(|(x, y)| format!("[{}, {}]", x, y))
+                            .collect::<Vec<_>>()
+                            .join(", "),
                     );
                 },
                 Err(e) => {

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

@@ -6,6 +6,10 @@ use crate::{
     steps::{
         common::latest_processed_version_tracker::LatestVersionProcessedTracker,
         events_processor::{EventsExtractor, EventsStorer},
+        fungible_asset_processor::{
+            fungible_asset_extractor::FungibleAssetExtractor,
+            fungible_asset_storer::FungibleAssetStorer,
+        },
     },
     utils::{
         chain_id::check_or_update_chain_id,
@@ -106,8 +110,8 @@ impl FungibleAssetProcessor {
             ..self.config.transaction_stream_config
         })
         .await?;
-        // let events_extractor = EventsExtractor {};
-        // let events_storer = EventsStorer::new(self.db_pool.clone(), events_processor_config);
+        let fa_extractor = FungibleAssetExtractor {};
+        let fa_storer = FungibleAssetStorer::new(self.db_pool.clone(), fa_config);
         let version_tracker = LatestVersionProcessedTracker::new(
             self.db_pool.clone(),
             starting_version,
@@ -118,8 +122,8 @@ impl FungibleAssetProcessor {
         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(fa_extractor.into_runnable_step(), channel_size)
+        .connect_to(fa_storer.into_runnable_step(), channel_size)
         .connect_to(version_tracker.into_runnable_step(), channel_size)
         .end_and_return_output_receiver(channel_size);
 
@@ -131,8 +135,13 @@ impl FungibleAssetProcessor {
                         continue;
                     }
                     debug!(
-                        "Finished processing versions [{:?}, {:?}]",
-                        txn_context.start_version, txn_context.end_version,
+                        "Finished processing versions {}",
+                        txn_context
+                            .get_versions()
+                            .iter()
+                            .map(|(x, y)| format!("[{}, {}]", x, y))
+                            .collect::<Vec<_>>()
+                            .join(", "),
                     );
                 },
                 Err(e) => {

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

@@ -2,6 +2,7 @@ use crate::utils::database::{execute_with_better_error, ArcDbPool};
 use ahash::AHashMap;
 use anyhow::Result;
 use aptos_indexer_processor_sdk::{
+    aptos_protos,
     traits::{
         pollable_async_step::PollableAsyncRunType, NamedStep, PollableAsyncStep, Processable,
     },
@@ -11,10 +12,18 @@ use aptos_indexer_processor_sdk::{
 use async_trait::async_trait;
 use diesel::{upsert::excluded, ExpressionMethods};
 use processor::{db::common::models::processor_status::ProcessorStatus, schema::processor_status};
+use std::marker::PhantomData;
 use tracing::info;
 
 const UPDATE_PROCESSOR_STATUS_SECS: u64 = 1;
 
+pub struct ProcessedBatch {
+    pub start_version: u64,
+    pub end_version: u64,
+    pub start_transaction_timestamp: Option<aptos_protos::util::timestamp::Timestamp>,
+    pub end_transaction_timestamp: Option<aptos_protos::util::timestamp::Timestamp>,
+}
+
 pub struct LatestVersionProcessedTracker<T>
 where
     Self: Sized + Send + 'static,
@@ -25,9 +34,10 @@ where
     // Next version to process that we expect.
     next_version: u64,
     // Last successful batch of sequentially processed transactions. Includes metadata to write to storage.
-    last_success_batch: Option<TransactionContext<T>>,
+    last_success_batch: Option<ProcessedBatch>,
     // Tracks all the versions that have been processed out of order.
-    seen_versions: AHashMap<u64, TransactionContext<T>>,
+    seen_versions: AHashMap<u64, ProcessedBatch>,
+    _marker: PhantomData<T>,
 }
 
 impl<T> LatestVersionProcessedTracker<T>
@@ -42,10 +52,11 @@ where
             next_version: starting_version,
             last_success_batch: None,
             seen_versions: AHashMap::new(),
+            _marker: PhantomData,
         }
     }
 
-    fn update_last_success_batch(&mut self, current_batch: TransactionContext<T>) {
+    fn update_last_success_batch(&mut self, current_batch: ProcessedBatch) {
         let mut new_prev_batch = current_batch;
         // While there are batches in seen_versions that are in order, update the new_prev_batch to the next batch.
         while let Some(next_version) = self.seen_versions.remove(&(new_prev_batch.end_version + 1))
@@ -107,41 +118,33 @@ where
         &mut self,
         current_batch: TransactionContext<T>,
     ) -> Result<Option<TransactionContext<T>>, ProcessorError> {
-        // info!(
-        //     start_version = current_batch.start_version,
-        //     end_version = current_batch.end_version,
-        //     step_name = self.name(),
-        //     "Processing versions"
-        // );
-        // If there's a gap in the next_version and current_version, save the current_version to seen_versions for
-        // later processing.
-        if self.next_version != current_batch.start_version {
-            info!(
-                expected_next_version = self.next_version,
-                step = self.name(),
-                batch_version = current_batch.start_version,
-                "Gap detected",
-            );
-            self.seen_versions
-                .insert(current_batch.start_version, TransactionContext {
-                    data: vec![], // No data is needed for tracking. This is to avoid clone.
-                    start_version: current_batch.start_version,
-                    end_version: current_batch.end_version,
-                    start_transaction_timestamp: current_batch.start_transaction_timestamp.clone(),
-                    end_transaction_timestamp: current_batch.end_transaction_timestamp.clone(),
-                    total_size_in_bytes: current_batch.total_size_in_bytes,
+        for context in current_batch.context.iter() {
+            // If there's a gap in the next_version and current_version, save the current_version to seen_versions for
+            // later processing.
+            if self.next_version != context.start_version {
+                tracing::debug!(
+                    next_version = self.next_version,
+                    step = self.name(),
+                    "Gap detected starting from version: {}",
+                    context.start_version
+                );
+                self.seen_versions
+                    .insert(context.start_version, ProcessedBatch {
+                        start_version: context.start_version,
+                        end_version: context.end_version,
+                        start_transaction_timestamp: context.start_transaction_timestamp.clone(),
+                        end_transaction_timestamp: context.end_transaction_timestamp.clone(),
+                    });
+            } else {
+                tracing::debug!("No gap detected");
+                // If the current_batch is the next expected version, update the last success batch
+                self.update_last_success_batch(ProcessedBatch {
+                    start_version: context.start_version,
+                    end_version: context.end_version,
+                    start_transaction_timestamp: context.start_transaction_timestamp.clone(),
+                    end_transaction_timestamp: context.end_transaction_timestamp.clone(),
                 });
-        } else {
-            // info!("No gap detected");
-            // If the current_batch is the next expected version, update the last success batch
-            self.update_last_success_batch(TransactionContext {
-                data: vec![], // No data is needed for tracking. This is to avoid clone.
-                start_version: current_batch.start_version,
-                end_version: current_batch.end_version,
-                start_transaction_timestamp: current_batch.start_transaction_timestamp.clone(),
-                end_transaction_timestamp: current_batch.end_transaction_timestamp.clone(),
-                total_size_in_bytes: current_batch.total_size_in_bytes,
-            });
+            }
         }
         // Pass through
         Ok(Some(current_batch))

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

@@ -68,11 +68,7 @@ impl Processable for EventsExtractor {
             .collect::<Vec<EventModel>>();
         Ok(Some(TransactionContext {
             data: events,
-            start_version: item.start_version,
-            end_version: item.end_version,
-            start_transaction_timestamp: item.start_transaction_timestamp,
-            end_transaction_timestamp: item.end_transaction_timestamp,
-            total_size_in_bytes: item.total_size_in_bytes,
+            context: item.context,
         }))
     }
 }

rust/sdk-processor/src/steps/events_processor/events_extractor.rs

@@ -84,15 +84,26 @@ impl Processable for EventsStorer {
         match execute_res {
             Ok(_) => {
                 debug!(
-                    "Events version [{}, {}] stored successfully",
-                    events.start_version, events.end_version
+                    "Events version {} stored successfully",
+                    events
+                        .get_versions()
+                        .iter()
+                        .map(|(x, y)| format!("[{}, {}]", x, y))
+                        .collect::<Vec<_>>()
+                        .join(", "),
                 );
                 Ok(Some(events))
             },
             Err(e) => Err(ProcessorError::DBStoreError {
                 message: format!(
-                    "Failed to store events versions {} to {}: {:?}",
-                    events.start_version, events.end_version, e,
+                    "Failed to store events versions {:?}: {:?}",
+                    events
+                        .get_versions()
+                        .iter()
+                        .map(|(x, y)| format!("[{}, {}]", x, y))
+                        .collect::<Vec<_>>()
+                        .join(", "),
+                    e,
                 ),
                 // TODO: fix it with a debug_query.
                 query: None,

rust/sdk-processor/src/steps/events_processor/events_storer.rs

@@ -71,11 +71,7 @@ impl Processable for FungibleAssetExtractor {
                 current_unified_fungible_asset_balances,
                 coin_supply,
             )],
-            start_version: transactions.start_version,
-            end_version: transactions.end_version,
-            start_transaction_timestamp: transactions.start_transaction_timestamp,
-            end_transaction_timestamp: transactions.end_transaction_timestamp,
-            total_size_in_bytes: transactions.total_size_in_bytes,
+            context: transactions.context,
         }))
     }
 }