indexer grpc fullnode now consider both indexer ledger version + tabl…

…e info version to have strong consistency.

ecosystem/indexer-grpc/indexer-grpc-fullnode/src/stream_coordinator.rs

@@ -500,7 +500,17 @@ impl IndexerStreamCoordinator {
 
     pub fn set_highest_known_version(&mut self) -> anyhow::Result<()> {
         let info = self.context.get_latest_ledger_info_wrapped()?;
-        self.highest_known_version = info.ledger_version.0;
+        let latest_table_info_version = self
+            .context
+            .indexer_reader
+            .as_ref()
+            .expect("Table info reader not set")
+            .get_latest_table_info_ledger_version()?
+            .expect("Table info ledger version not set");
+
+        self.highest_known_version =
+            std::cmp::min(info.ledger_version.0, latest_table_info_version);
+
         Ok(())
     }
 

storage/indexer/src/indexer_reader.rs

@@ -58,6 +58,13 @@ impl IndexerReader for IndexerReaders {
         anyhow::bail!("DB indexer reader is not available")
     }
 
+    fn get_latest_table_info_ledger_version(&self) -> anyhow::Result<Option<Version>> {
+        if let Some(table_info_reader) = &self.table_info_reader {
+            return Ok(Some(table_info_reader.next_version()));
+        }
+        anyhow::bail!("Table info reader is not available")
+    }
+
     fn get_events(
         &self,
         event_key: &EventKey,

types/src/indexer/indexer_db_reader.rs

@@ -60,6 +60,7 @@ pub trait IndexerReader: Send + Sync {
     ) -> Result<Box<dyn Iterator<Item = Result<(StateKey, StateValue)>> + '_>>;
 
     fn get_latest_internal_indexer_ledger_version(&self) -> Result<Option<Version>>;
+    fn get_latest_table_info_ledger_version(&self) -> Result<Option<Version>>;
 
     #[cfg(any(test, feature = "fuzzing"))]
     fn wait_for_internal_indexer(&self, version: Version) -> Result<()> {