indexer: make block finality configurable (#12685)

Tested this out by using Finality::None in mpc node indexers. Observed
reduction of latency.

chain/indexer/src/lib.rs

@@ -6,7 +6,7 @@ use tokio::sync::mpsc;
 
 use near_chain_configs::GenesisValidationMode;
 pub use near_primitives;
-use near_primitives::types::Gas;
+use near_primitives::types::{Finality, Gas};
 pub use nearcore::{get_default_home, init_configs, NearConfig};
 
 pub use near_indexer_primitives::{
@@ -83,6 +83,8 @@ pub struct IndexerConfig {
     pub sync_mode: SyncModeEnum,
     /// Whether await for node to be synced or not
     pub await_for_node_synced: AwaitForNodeSyncedEnum,
+    /// Finality level at which blocks are streamed
+    pub finality: Finality,
     /// Tells whether to validate the genesis file before starting
     pub validate_genesis: bool,
 }

chain/indexer/src/streamer/fetchers.rs

@@ -29,12 +29,13 @@ pub(crate) async fn fetch_status(
 /// entire block or we already fetched this block.
 pub(crate) async fn fetch_latest_block(
     client: &Addr<near_client::ViewClientActor>,
+    finality: &near_primitives::types::Finality,
 ) -> Result<views::BlockView, FailedToFetchData> {
     tracing::debug!(target: INDEXER, "Fetching latest block");
     client
         .send(
             near_client::GetBlock(near_primitives::types::BlockReference::Finality(
-                near_primitives::types::Finality::Final,
+                finality.clone(),
             ))
             .with_span_context(),
         )

chain/indexer/src/streamer/mod.rs

@@ -37,7 +37,7 @@ static DELAYED_LOCAL_RECEIPTS_CACHE: std::sync::LazyLock<
     Arc<RwLock<HashMap<CryptoHash, views::ReceiptView>>>,
 > = std::sync::LazyLock::new(|| Arc::new(RwLock::new(HashMap::new())));
 
-const INTERVAL: Duration = Duration::from_millis(500);
+const INTERVAL: Duration = Duration::from_millis(250);
 
 /// Blocks #47317863 and #47317864 with restored receipts.
 const PROBLEMATIC_BLOCKS: [CryptoHash; 2] = [
@@ -413,11 +413,12 @@ pub(crate) async fn start(
             AwaitForNodeSyncedEnum::StreamWhileSyncing => {}
         };
 
-        let block = if let Ok(block) = fetch_latest_block(&view_client).await {
-            block
-        } else {
-            continue;
-        };
+        let block =
+            if let Ok(block) = fetch_latest_block(&view_client, &indexer_config.finality).await {
+                block
+            } else {
+                continue;
+            };
 
         let latest_block_height = block.header.height;
         let start_syncing_block_height = if let Some(last_synced_block_height) =

tools/indexer/example/Cargo.toml

@@ -23,3 +23,4 @@ tracing.workspace = true
 near-config-utils.workspace = true
 near-indexer.workspace = true
 near-o11y.workspace = true
+near-primitives.workspace = true

tools/indexer/example/src/main.rs

@@ -275,6 +275,7 @@ fn main() -> Result<()> {
                 home_dir,
                 sync_mode: near_indexer::SyncModeEnum::FromInterruption,
                 await_for_node_synced: near_indexer::AwaitForNodeSyncedEnum::WaitForFullSync,
+                finality: near_primitives::types::Finality::Final,
                 validate_genesis: true,
             };
             let system = actix::System::new();

tools/mirror/src/lib.rs

@@ -1763,6 +1763,7 @@ impl<T: ChainAccess> TxMirror<T> {
             home_dir,
             sync_mode: near_indexer::SyncModeEnum::FromInterruption,
             await_for_node_synced: near_indexer::AwaitForNodeSyncedEnum::StreamWhileSyncing,
+            finality: Finality::Final,
             validate_genesis: false,
         })
         .context("failed to start target chain indexer")?;