fix: indexer wedging on startup

crates/node/src/indexer.rs

@@ -21,7 +21,7 @@ use near_async::{
 use near_client::{RpcHandlerActor, Status, ViewClientActor, client_actor::ClientActor};
 use near_indexer::near_primitives::transaction::SignedTransaction;
 use near_indexer_primitives::{
-    types::{BlockReference, Finality},
+    types::{BlockHeight, BlockReference, Finality},
     views::{BlockView, QueryRequest, QueryResponseKind},
 };
 use near_mpc_contract_interface::method_names::{
@@ -421,8 +421,12 @@ struct IndexerClient {
 
 const INTERVAL: Duration = Duration::from_millis(500);
 
+/// Consecutive non-syncing polls with head progress before the node counts as caught up.
+const REQUIRED_STABLE_POLLS: u32 = 4;
+
 impl IndexerClient {
     async fn wait_for_full_sync(&self) {
+        let mut progress = SyncProgress::default();
         loop {
             tokio::time::sleep(INTERVAL).await;
 
@@ -441,13 +445,47 @@ impl IndexerClient {
                 continue;
             };
 
-            if !status.sync_info.syncing {
+            if progress.observe(
+                status.sync_info.syncing,
+                status.sync_info.latest_block_height,
+            ) {
                 return;
             }
         }
     }
 }
 
+/// Detects catch-up from head progress alone: a freshly state-syncing node
+/// briefly reports `syncing == false` at the genesis head, and returning then
+/// pins the streamer's `LatestSynced` cursor at that stale head. So we require a
+/// run of non-syncing polls over which the head actually advances.
+#[derive(Default)]
+struct SyncProgress {
+    run_start_head: Option<BlockHeight>,
+    run_polls: u32,
+}
+
+impl SyncProgress {
+    fn observe(&mut self, syncing: bool, head_height: BlockHeight) -> bool {
+        if syncing {
+            self.run_start_head = None;
+            self.run_polls = 0;
+            return false;
+        }
+        match self.run_start_head {
+            None => {
+                self.run_start_head = Some(head_height);
+                self.run_polls = 1;
+                false
+            }
+            Some(start_head) => {
+                self.run_polls += 1;
+                self.run_polls >= REQUIRED_STABLE_POLLS && head_height > start_head
+            }
+        }
+    }
+}
+
 // #[derive(Debug)]
 struct IndexerRpcHandler {
     rpc_handler: MultithreadRuntimeHandle<RpcHandlerActor>,
@@ -503,3 +541,109 @@ pub struct IndexerAPI<TransactionSender, ForeignChainPolicyReader> {
 
     pub foreign_chain_policy_reader: ForeignChainPolicyReader,
 }
+
+#[cfg(test)]
+#[expect(non_snake_case)]
+mod tests {
+    use super::{BlockHeight, REQUIRED_STABLE_POLLS, SyncProgress};
+
+    fn first_caught_up_poll(samples: &[(bool, BlockHeight)]) -> Option<usize> {
+        let mut progress = SyncProgress::default();
+        samples
+            .iter()
+            .position(|&(syncing, head)| progress.observe(syncing, head))
+    }
+
+    #[test]
+    fn observe__should_never_report_caught_up_while_syncing() {
+        // Given
+        let samples: Vec<_> = (0..10).map(|i| (true, 42_000_000 + i)).collect();
+
+        // When
+        let caught_up_at = first_caught_up_poll(&samples);
+
+        // Then
+        assert_eq!(caught_up_at, None);
+    }
+
+    #[test]
+    fn observe__should_never_report_caught_up_with_static_head_at_genesis() {
+        // Given
+        let genesis = 42_376_888;
+        let samples: Vec<_> = (0..10).map(|_| (false, genesis)).collect();
+
+        // When
+        let caught_up_at = first_caught_up_poll(&samples);
+
+        // Then
+        assert_eq!(caught_up_at, None);
+    }
+
+    #[test]
+    fn observe__should_not_report_caught_up_before_required_polls() {
+        // Given
+        let samples: Vec<_> = (0..REQUIRED_STABLE_POLLS - 1)
+            .map(|i| (false, 257_000_000 + u64::from(i)))
+            .collect();
+
+        // When
+        let caught_up_at = first_caught_up_poll(&samples);
+
+        // Then
+        assert_eq!(caught_up_at, None);
+    }
+
+    #[test]
+    fn observe__should_report_caught_up_after_sustained_progress() {
+        // Given
+        let samples: Vec<_> = (0..REQUIRED_STABLE_POLLS)
+            .map(|i| (false, 257_000_000 + u64::from(i)))
+            .collect();
+
+        // When
+        let caught_up_at = first_caught_up_poll(&samples);
+
+        // Then
+        let expected = usize::try_from(REQUIRED_STABLE_POLLS).unwrap() - 1;
+        assert_eq!(caught_up_at, Some(expected));
+    }
+
+    #[test]
+    fn observe__should_wait_for_head_to_advance_past_run_start() {
+        // Given
+        let head = 257_000_000;
+        let mut samples: Vec<_> = (0..REQUIRED_STABLE_POLLS + 2)
+            .map(|_| (false, head))
+            .collect();
+        samples.push((false, head + 1));
+
+        // When
+        let caught_up_at = first_caught_up_poll(&samples);
+
+        // Then
+        assert_eq!(caught_up_at, Some(samples.len() - 1));
+    }
+
+    #[test]
+    fn observe__should_reset_run_when_syncing_resumes() {
+        // Given
+        let pre = [
+            (false, 42_000_000),
+            (false, 42_000_001),
+            (false, 42_000_002),
+        ];
+        let resync = [(true, 100_000_000)];
+        let post: Vec<_> = (0..REQUIRED_STABLE_POLLS)
+            .map(|i| (false, 257_000_000 + u64::from(i)))
+            .collect();
+        let samples: Vec<_> = pre.iter().chain(&resync).chain(&post).copied().collect();
+
+        // When
+        let caught_up_at = first_caught_up_poll(&samples);
+
+        // Then
+        let expected =
+            pre.len() + resync.len() + usize::try_from(REQUIRED_STABLE_POLLS).unwrap() - 1;
+        assert_eq!(caught_up_at, Some(expected));
+    }
+}

crates/node/src/indexer/real.rs

@@ -143,15 +143,9 @@ pub fn spawn_real_indexer(
 
             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.
+            // Streaming before the node is synced pins the `LatestSynced` cursor
+            // at genesis, below the block tail it can never reach. Raced against
+            // shutdown so a SIGTERM during state sync still tears down cleanly.
             if !await_sync_or_shutdown(
                 indexer_state.client.wait_for_full_sync(),
                 &shutdown_token,