[Indexer-Grpc-V2] Several fixes/improvements.

grao1991 · grao1991 · commit b62431c0a4d7 · 2025-03-12T22:10:26.000Z
diff --git a/ecosystem/indexer-grpc/indexer-grpc-data-service-v2/src/historical_data_service.rs b/ecosystem/indexer-grpc/indexer-grpc-data-service-v2/src/historical_data_service.rs
@@ -7,7 +7,7 @@ use crate::{
     metrics::{COUNTER, TIMER},
 };
 use aptos_indexer_grpc_utils::file_store_operator_v2::file_store_reader::FileStoreReader;
-use aptos_protos::indexer::v1::{GetTransactionsRequest, TransactionsResponse};
+use aptos_protos::indexer::v1::{GetTransactionsRequest, ProcessedRange, TransactionsResponse};
 use aptos_transaction_filter::BooleanTransactionFilter;
 use futures::executor::block_on;
 use std::{
@@ -167,14 +167,21 @@ impl HistoricalDataService {
                         /*retries=*/ 3,
                         /*max_files=*/ None,
                         filter,
+                        Some(ending_version),
                         tx,
                     )
                     .await;
             });
 
             let mut close_to_latest = false;
-            while let Some((transactions, batch_size_bytes, timestamp)) = rx.recv().await {
-                next_version += transactions.len() as u64;
+            while let Some((
+                transactions,
+                batch_size_bytes,
+                timestamp,
+                (first_processed_version, last_processed_version),
+            )) = rx.recv().await
+            {
+                next_version = last_processed_version + 1;
                 size_bytes += batch_size_bytes as u64;
                 let timestamp_since_epoch =
                     Duration::new(timestamp.seconds as u64, timestamp.nanos as u32);
@@ -186,28 +193,55 @@ impl HistoricalDataService {
                     close_to_latest = true;
                 }
 
-                if !transactions.is_empty() {
-                    let responses =
-                        transactions
-                            .chunks(max_num_transactions_per_batch)
-                            .map(|chunk| TransactionsResponse {
+                let responses = if !transactions.is_empty() {
+                    let mut current_version = first_processed_version;
+                    let mut responses: Vec<_> = transactions
+                        .chunks(max_num_transactions_per_batch)
+                        .map(|chunk| {
+                            let first_version = current_version;
+                            let last_version = chunk.last().unwrap().version;
+                            current_version = last_version + 1;
+                            TransactionsResponse {
                                 transactions: chunk.to_vec(),
                                 chain_id: Some(self.chain_id),
-                            });
-                    for response in responses {
-                        let _timer = TIMER
-                            .with_label_values(&["historical_data_service_send_batch"])
-                            .start_timer();
-                        if response_sender.send(Ok(response)).await.is_err() {
-                            // NOTE: We are not recalculating the version and size_bytes for the stream
-                            // progress since nobody cares about the accurate if client has dropped the
-                            // connection.
-                            info!(stream_id = id, "Client dropped.");
-                            COUNTER
-                                .with_label_values(&["historical_data_service_client_dropped"])
-                                .inc();
-                            break 'out;
-                        }
+                                processed_range: Some(ProcessedRange {
+                                    first_version,
+                                    last_version,
+                                }),
+                            }
+                        })
+                        .collect();
+                    responses
+                        .last_mut()
+                        .unwrap()
+                        .processed_range
+                        .unwrap()
+                        .last_version = last_processed_version;
+                    responses
+                } else {
+                    vec![TransactionsResponse {
+                        transactions: vec![],
+                        chain_id: Some(self.chain_id),
+                        processed_range: Some(ProcessedRange {
+                            first_version: first_processed_version,
+                            last_version: last_processed_version,
+                        }),
+                    }]
+                };
+
+                for response in responses {
+                    let _timer = TIMER
+                        .with_label_values(&["historical_data_service_send_batch"])
+                        .start_timer();
+                    if response_sender.send(Ok(response)).await.is_err() {
+                        // NOTE: We are not recalculating the version and size_bytes for the stream
+                        // progress since nobody cares about the accurate if client has dropped the
+                        // connection.
+                        info!(stream_id = id, "Client dropped.");
+                        COUNTER
+                            .with_label_values(&["historical_data_service_client_dropped"])
+                            .inc();
+                        break 'out;
                     }
                 }
             }
diff --git a/ecosystem/indexer-grpc/indexer-grpc-data-service-v2/src/live_data_service/in_memory_cache.rs b/ecosystem/indexer-grpc/indexer-grpc-data-service-v2/src/live_data_service/in_memory_cache.rs
@@ -44,7 +44,7 @@ impl<'a> InMemoryCache<'a> {
         max_num_transactions_per_batch: usize,
         max_bytes_per_batch: usize,
         filter: &Option<BooleanTransactionFilter>,
-    ) -> Option<(Vec<Transaction>, usize)> {
+    ) -> Option<(Vec<Transaction>, usize, u64)> {
         let _timer = TIMER.with_label_values(&["cache_get_data"]).start_timer();
 
         while starting_version >= self.data_manager.read().await.end_version {
@@ -97,7 +97,7 @@ impl<'a> InMemoryCache<'a> {
                 }
             }
             trace!("Data was sent from cache, last version: {}.", version - 1);
-            return Some((result, total_bytes));
+            return Some((result, total_bytes, version - 1));
         }
     }
 }
diff --git a/ecosystem/indexer-grpc/indexer-grpc-data-service-v2/src/live_data_service/mod.rs b/ecosystem/indexer-grpc/indexer-grpc-data-service-v2/src/live_data_service/mod.rs
@@ -12,7 +12,7 @@ use crate::{
     live_data_service::in_memory_cache::InMemoryCache,
     metrics::{COUNTER, TIMER},
 };
-use aptos_protos::indexer::v1::{GetTransactionsRequest, TransactionsResponse};
+use aptos_protos::indexer::v1::{GetTransactionsRequest, ProcessedRange, TransactionsResponse};
 use aptos_transaction_filter::BooleanTransactionFilter;
 use std::{sync::Arc, time::Duration};
 use tokio::sync::mpsc::{Receiver, Sender};
@@ -175,7 +175,7 @@ impl<'a> LiveDataService<'a> {
                 continue;
             }
 
-            if let Some((transactions, batch_size_bytes)) = self
+            if let Some((transactions, batch_size_bytes, last_processed_version)) = self
                 .in_memory_cache
                 .get_data(
                     next_version,
@@ -189,12 +189,16 @@ impl<'a> LiveDataService<'a> {
                 let _timer = TIMER
                     .with_label_values(&["live_data_service_send_batch"])
                     .start_timer();
-                next_version += transactions.len() as u64;
-                size_bytes += batch_size_bytes as u64;
                 let response = TransactionsResponse {
                     transactions,
                     chain_id: Some(self.chain_id),
+                    processed_range: Some(ProcessedRange {
+                        first_version: next_version,
+                        last_version: last_processed_version,
+                    }),
                 };
+                next_version = last_processed_version + 1;
+                size_bytes += batch_size_bytes as u64;
                 if response_sender.send(Ok(response)).await.is_err() {
                     info!(stream_id = id, "Client dropped.");
                     COUNTER
diff --git a/ecosystem/indexer-grpc/indexer-grpc-data-service/src/service.rs b/ecosystem/indexer-grpc/indexer-grpc-data-service/src/service.rs
@@ -699,6 +699,7 @@ fn get_transactions_responses_builder(
         .map(|chunk| TransactionsResponse {
             chain_id: Some(chain_id as u64),
             transactions: chunk,
+            processed_range: None,
         })
         .collect();
     (responses, num_stripped)
diff --git a/ecosystem/indexer-grpc/indexer-grpc-fullnode/src/localnet_data_service.rs b/ecosystem/indexer-grpc/indexer-grpc-fullnode/src/localnet_data_service.rs
@@ -105,6 +105,7 @@ impl RawData for LocalnetDataService {
                         },
                         _ => panic!("Unexpected response type."),
                     },
+                    processed_range: None,
                 });
                 match external_service_tx.send(response).await {
                     Ok(_) => {},
diff --git a/ecosystem/indexer-grpc/indexer-grpc-manager/src/data_manager.rs b/ecosystem/indexer-grpc/indexer-grpc-manager/src/data_manager.rs
@@ -296,14 +296,15 @@ impl DataManager {
                 /*retries=*/ 3,
                 /*max_files=*/ Some(1),
                 /*filter=*/ None,
+                /*ending_version=*/ None,
                 tx,
             )
             .await;
 
-        if let Some((transactions, _, _)) = rx.recv().await {
+        if let Some((transactions, _, _, range)) = rx.recv().await {
             debug!(
-                "Transactions returned from filestore: [{start_version}, {}).",
-                transactions.last().unwrap().version
+                "Transactions returned from filestore: [{}, {}].",
+                range.0, range.1
             );
             let first_version = transactions.first().unwrap().version;
             ensure!(
diff --git a/ecosystem/indexer-grpc/indexer-grpc-manager/src/service.rs b/ecosystem/indexer-grpc/indexer-grpc-manager/src/service.rs
@@ -140,6 +140,8 @@ impl GrpcManager for GrpcManagerService {
         Ok(Response::new(TransactionsResponse {
             transactions,
             chain_id: Some(self.chain_id),
+            // Not used.
+            processed_range: None,
         }))
     }
 
diff --git a/ecosystem/indexer-grpc/indexer-grpc-utils/src/file_store_operator_v2/file_store_reader.rs b/ecosystem/indexer-grpc/indexer-grpc-utils/src/file_store_operator_v2/file_store_reader.rs
@@ -76,7 +76,8 @@ impl FileStoreReader {
         retries: u8,
         max_files: Option<usize>,
         filter: Option<BooleanTransactionFilter>,
-        tx: Sender<(Vec<Transaction>, usize, Timestamp)>,
+        ending_version: Option<u64>,
+        tx: Sender<(Vec<Transaction>, usize, Timestamp, (u64, u64))>,
     ) {
         trace!(
             "Getting transactions from file store, version: {version}, max_files: {max_files:?}."
@@ -118,13 +119,20 @@ impl FileStoreReader {
                 if num_to_skip > 0 {
                     transactions = transactions.split_off(num_to_skip);
                 }
+                let processed_range = (
+                    transactions.first().unwrap().version,
+                    transactions.last().unwrap().version,
+                );
+                if let Some(ending_version) = ending_version {
+                    transactions.retain(|t| t.version < ending_version);
+                }
                 if let Some(ref filter) = filter {
                     transactions.retain(|t| filter.matches(t));
                 }
                 let size_bytes = transactions.iter().map(|t| t.encoded_len()).sum();
                 trace!("Got {} transactions from file store to send, size: {size_bytes}, first_version: {:?}", transactions.len(), transactions.first().map(|t| t.version));
                 if tx
-                    .send((transactions, size_bytes, timestamp))
+                    .send((transactions, size_bytes, timestamp, processed_range))
                     .await
                     .is_err()
                 {
diff --git a/ecosystem/indexer-grpc/transaction-filter/src/traits.rs b/ecosystem/indexer-grpc/transaction-filter/src/traits.rs
@@ -47,7 +47,7 @@ where
 
     #[inline]
     fn matches_vec(&self, items: &[T]) -> bool {
-        items.iter().all(|item| self.matches(item))
+        items.iter().any(|item| self.matches(item))
     }
 
     #[inline]
diff --git a/protos/proto/aptos/indexer/v1/raw_data.proto b/protos/proto/aptos/indexer/v1/raw_data.proto
@@ -32,13 +32,20 @@ message GetTransactionsRequest {
   optional BooleanTransactionFilter transaction_filter = 4;
 }
 
+message ProcessedRange {
+    uint64 first_version = 1;
+    uint64 last_version = 2;
+}
+
 // TransactionsResponse is a batch of transactions.
 message TransactionsResponse {
   // Required; transactions data.
   repeated aptos.transaction.v1.Transaction transactions = 1;
 
   // Required; chain id.
   optional uint64 chain_id = 2 [jstype = JS_STRING];
+
+  optional ProcessedRange processed_range = 3;
 }
 
 service RawData {
diff --git a/protos/python/aptos_protos/aptos/indexer/v1/raw_data_pb2.py b/protos/python/aptos_protos/aptos/indexer/v1/raw_data_pb2.py
diff --git a/protos/python/aptos_protos/aptos/indexer/v1/raw_data_pb2.pyi b/protos/python/aptos_protos/aptos/indexer/v1/raw_data_pb2.pyi
@@ -53,18 +53,31 @@ class GetTransactionsRequest(_message.Message):
         ] = ...,
     ) -> None: ...
 
+class ProcessedRange(_message.Message):
+    __slots__ = ["first_version", "last_version"]
+    FIRST_VERSION_FIELD_NUMBER: _ClassVar[int]
+    LAST_VERSION_FIELD_NUMBER: _ClassVar[int]
+    first_version: int
+    last_version: int
+    def __init__(
+        self, first_version: _Optional[int] = ..., last_version: _Optional[int] = ...
+    ) -> None: ...
+
 class TransactionsResponse(_message.Message):
-    __slots__ = ["transactions", "chain_id"]
+    __slots__ = ["transactions", "chain_id", "processed_range"]
     TRANSACTIONS_FIELD_NUMBER: _ClassVar[int]
     CHAIN_ID_FIELD_NUMBER: _ClassVar[int]
+    PROCESSED_RANGE_FIELD_NUMBER: _ClassVar[int]
     transactions: _containers.RepeatedCompositeFieldContainer[
         _transaction_pb2.Transaction
     ]
     chain_id: int
+    processed_range: ProcessedRange
     def __init__(
         self,
         transactions: _Optional[
             _Iterable[_Union[_transaction_pb2.Transaction, _Mapping]]
         ] = ...,
         chain_id: _Optional[int] = ...,
+        processed_range: _Optional[_Union[ProcessedRange, _Mapping]] = ...,
     ) -> None: ...
diff --git a/protos/rust/src/pb/aptos.indexer.v1.rs b/protos/rust/src/pb/aptos.indexer.v1.rs
diff --git a/protos/rust/src/pb/aptos.indexer.v1.serde.rs b/protos/rust/src/pb/aptos.indexer.v1.serde.rs
diff --git a/protos/typescript/src/aptos/indexer/v1/raw_data.ts b/protos/typescript/src/aptos/indexer/v1/raw_data.ts

Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ impl<'a> InMemoryCache<'a> {`
`44`	`44`	`max_num_transactions_per_batch: usize,`
`45`	`45`	`max_bytes_per_batch: usize,`
`46`	`46`	`filter: &Option<BooleanTransactionFilter>,`
`47`		`- ) -> Option<(Vec<Transaction>, usize)> {`
	`47`	`+ ) -> Option<(Vec<Transaction>, usize, u64)> {`
`48`	`48`	`let _timer = TIMER.with_label_values(&["cache_get_data"]).start_timer();`
`49`	`49`
`50`	`50`	`while starting_version >= self.data_manager.read().await.end_version {`
`@@ -97,7 +97,7 @@ impl<'a> InMemoryCache<'a> {`
`97`	`97`	`}`
`98`	`98`	`}`
`99`	`99`	`trace!("Data was sent from cache, last version: {}.", version - 1);`
`100`		`- return Some((result, total_bytes));`
	`100`	`+ return Some((result, total_bytes, version - 1));`
`101`	`101`	`}`
`102`	`102`	`}`
`103`	`103`	`}`
Original file line number	Diff line number	Diff line change
`@@ -140,6 +140,8 @@ impl GrpcManager for GrpcManagerService {`
`140`	`140`	`Ok(Response::new(TransactionsResponse {`
`141`	`141`	`transactions,`
`142`	`142`	`chain_id: Some(self.chain_id),`
	`143`	`+ // Not used.`
	`144`	`+ processed_range: None,`
`143`	`145`	`}))`
`144`	`146`	`}`
`145`	`147`
Original file line number	Diff line number	Diff line change
`@@ -47,7 +47,7 @@ where`
`47`	`47`
`48`	`48`	`#[inline]`
`49`	`49`	`fn matches_vec(&self, items: &[T]) -> bool {`
`50`		`- items.iter().all(\|item\| self.matches(item))`
	`50`	`+ items.iter().any(\|item\| self.matches(item))`
`51`	`51`	`}`
`52`	`52`
`53`	`53`	`#[inline]`