fix: indexer getting stuck when syncing from genesis (#3647)

Author: gilcu3

crates/chain-gateway/src/chain_gateway.rs

@@ -1,3 +1,4 @@
+use std::future::Future;
 use std::path::Path;
 
 use near_account_id::AccountId;
@@ -153,7 +154,7 @@ fn run_node(
     ready_sender: tokio::sync::oneshot::Sender<RunNodeResult>,
     near_config: nearcore::NearConfig,
     home_dir: &Path,
-    shutdown_receiver: tokio::sync::oneshot::Receiver<()>,
+    mut shutdown_receiver: tokio::sync::oneshot::Receiver<()>,
     streamer_setup: Option<StreamerSetup>,
 ) {
     let rt = tokio::runtime::Builder::new_multi_thread()
@@ -185,6 +186,24 @@ fn run_node(
         let rpc_handler = NearRpcActorHandle::new(near_node.rpc_handler);
 
         let stream = if let Some((indexer, streamer_config)) = indexer_and_params {
+            // Don't start the streamer until synced, or its `LatestSynced`
+            // cursor pins at genesis. Raced against shutdown to stay responsive.
+            tracing::info!(
+                "chain-gateway waiting for node to finish syncing before streaming blocks."
+            );
+            let synced = await_sync_or_shutdown(client.wait_for_full_sync(), async {
+                let _ = (&mut shutdown_receiver).await;
+            })
+            .await;
+            if !synced {
+                tracing::info!("shutdown requested before sync completed; exiting startup.");
+                let _ = ready_sender.send(Err(ChainGatewayError::StartupFailed {
+                    msg: "shutdown requested before node finished syncing".to_string(),
+                }));
+                actor_system.stop();
+                return;
+            }
+
             let raw_stream: Receiver<StreamerMessage> = indexer.streamer();
             match event_subscriber::streamer::start(
                 streamer_config,
@@ -232,3 +251,48 @@ struct StreamerSetup {
     indexer_config: near_indexer::IndexerConfig,
     near_config: NearConfig,
 }
+
+async fn await_sync_or_shutdown(
+    sync: impl Future<Output = ()>,
+    shutdown: impl Future<Output = ()>,
+) -> bool {
+    tokio::select! {
+        _ = sync => true,
+        _ = shutdown => false,
+    }
+}
+
+#[cfg(test)]
+#[expect(non_snake_case)]
+mod tests {
+    use super::await_sync_or_shutdown;
+    use std::future::pending;
+
+    #[tokio::test]
+    async fn await_sync_or_shutdown__should_return_true_when_sync_completes_first() {
+        // Given
+        let sync = async {};
+        let shutdown = pending::<()>();
+
+        // When
+        let synced = await_sync_or_shutdown(sync, shutdown).await;
+
+        // Then
+        assert!(synced);
+    }
+
+    /// A shutdown (or abandoned startup) arriving while the node is still
+    /// syncing must win the race rather than block on sync.
+    #[tokio::test]
+    async fn await_sync_or_shutdown__should_return_false_when_shutdown_first() {
+        // Given
+        let sync = pending::<()>();
+        let shutdown = async {};
+
+        // When
+        let synced = await_sync_or_shutdown(sync, shutdown).await;
+
+        // Then
+        assert!(!synced);
+    }
+}

crates/node/src/indexer/real.rs

@@ -19,6 +19,7 @@ use near_account_id::AccountId;
 use near_async::ActorSystem;
 use near_indexer::Indexer;
 use near_mpc_contract_interface::types::ProtocolContractState;
+use std::future::Future;
 use std::path::PathBuf;
 use std::sync::Arc;
 #[cfg(feature = "network-hardship-simulation")]
@@ -110,15 +111,41 @@ pub fn spawn_real_indexer(
 
             let indexer = Indexer::from_near_node(near_indexer_config, near_config, &near_node);
 
-            let stream = indexer.streamer();
-
             let indexer_state = Arc::new(IndexerState::new(
                 near_node.view_client,
                 near_node.client,
                 near_node.rpc_handler,
                 mpc_indexer_config.mpc_contract_id.clone(),
             ));
 
+            tracing::info!("Indexer waiting for node to finish syncing before streaming blocks.");
+
+            // Defer all indexing work until the node has finished syncing. On a
+            // node that state-syncs from scratch, neard sits at genesis until
+            // sync completes; starting the streamer then pins its cursor at
+            // genesis (`LatestSynced`), below the node's block tail, so it can
+            // never reach the chain tip. (#3623)
+            //
+            // The wait is raced against `shutdown_token` so a SIGTERM during the
+            // (possibly long) initial state sync still tears the thread down
+            // cleanly instead of blocking until sync finishes.
+            if !await_sync_or_shutdown(
+                indexer_state.client.wait_for_full_sync(),
+                &shutdown_token,
+            )
+            .await
+            {
+                tracing::info!(
+                    "Indexer thread received shutdown signal before sync completed; exiting."
+                );
+                let _ = indexer_exit_sender.send(Ok(()));
+                return;
+            }
+
+            // The node is fully synced by this point, so `LatestSynced` resolves
+            // to the chain tip rather than genesis.
+            let stream = indexer.streamer();
+
             let txn_sender_result = TransactionProcessorHandle::start_transaction_processor(
                 my_near_account_id_clone,
                 account_secret_key.clone(),
@@ -289,3 +316,49 @@ pub fn spawn_real_indexer(
         foreign_chain_policy_reader,
     }
 }
+
+async fn await_sync_or_shutdown(
+    sync: impl Future<Output = ()>,
+    shutdown: &CancellationToken,
+) -> bool {
+    tokio::select! {
+        _ = sync => true,
+        _ = shutdown.cancelled() => false,
+    }
+}
+
+#[cfg(test)]
+#[expect(non_snake_case)]
+mod tests {
+    use super::await_sync_or_shutdown;
+    use std::future::pending;
+    use tokio_util::sync::CancellationToken;
+
+    #[tokio::test]
+    async fn await_sync_or_shutdown__should_return_true_when_sync_completes_first() {
+        // Given
+        let shutdown = CancellationToken::new();
+
+        // When
+        let synced = await_sync_or_shutdown(async {}, &shutdown).await;
+
+        // Then
+        assert!(synced);
+    }
+
+    /// The dominant production path: a SIGTERM arrives while the node is still
+    /// syncing, so the wait must yield to shutdown rather than block on sync.
+    #[tokio::test]
+    async fn await_sync_or_shutdown__should_return_false_when_shutdown_during_sync() {
+        // Given
+        let shutdown = CancellationToken::new();
+        let shutdown_clone = shutdown.clone();
+        tokio::spawn(async move { shutdown_clone.cancel() });
+
+        // When
+        let synced = await_sync_or_shutdown(pending::<()>(), &shutdown).await;
+
+        // Then
+        assert!(!synced);
+    }
+}