fix: indexer wedging on startup

Author: gilcu3

crates/chain-gateway/src/chain_gateway.rs

@@ -16,7 +16,7 @@ use crate::near_internals_wrapper::{
     NearClientActorHandle, NearRpcActorHandle, NearViewClientActorHandle,
 };
 use crate::primitives::{
-    FetchLatestFinalBlockInfo, IsSyncing, QueryViewFunction, SubmitSignedTransaction,
+    FetchLatestFinalBlockInfo, IsSyncing, QueryViewFunction, SubmitSignedTransaction, SyncStatus,
 };
 use crate::types::ObservedState;
 
@@ -32,8 +32,8 @@ pub struct ChainGateway {
 
 impl IsSyncing for ChainGateway {
     type Error = NearClientError;
-    async fn is_syncing(&self) -> Result<bool, Self::Error> {
-        self.client.is_syncing().await
+    async fn sync_status(&self) -> Result<SyncStatus, Self::Error> {
+        self.client.sync_status().await
     }
 }
 

crates/chain-gateway/src/mock.rs

@@ -1,5 +1,5 @@
 use crate::primitives::{
-    FetchLatestFinalBlockInfo, IsSyncing, QueryViewFunction, SubmitSignedTransaction,
+    FetchLatestFinalBlockInfo, IsSyncing, QueryViewFunction, SubmitSignedTransaction, SyncStatus,
 };
 use crate::types::{LatestFinalBlockInfo, ObservedState};
 use near_account_id::AccountId;
@@ -137,8 +137,23 @@ impl MockChainStateBuilder {
 
 impl IsSyncing for MockChainState {
     type Error = MockError;
-    async fn is_syncing(&self) -> Result<bool, Self::Error> {
-        self.sync_response.lock().unwrap().clone()
+    async fn sync_status(&self) -> Result<SyncStatus, Self::Error> {
+        // The mock is driven by a single "is it syncing" bool: `false` maps to a
+        // caught-up head, `true` to a head still behind its peers.
+        let syncing = self.sync_response.lock().unwrap().clone()?;
+        Ok(if syncing {
+            SyncStatus {
+                syncing: true,
+                head_height: 0,
+                max_peer_height: Some(100),
+            }
+        } else {
+            SyncStatus {
+                syncing: false,
+                head_height: 100,
+                max_peer_height: Some(100),
+            }
+        })
     }
 }
 

crates/chain-gateway/src/near_internals_wrapper/client.rs

@@ -2,7 +2,10 @@ use std::sync::Arc;
 
 use near_async::messaging::CanSendAsync as _;
 
-use crate::{errors::NearClientError, primitives::IsSyncing};
+use crate::{
+    errors::NearClientError,
+    primitives::{IsSyncing, SyncStatus},
+};
 
 /// Arc-wrapper around near-internal struct
 #[derive(Clone)]
@@ -23,12 +26,14 @@ impl NearClientActorHandle {
 /// Implement IsSyncing for our near client
 impl IsSyncing for NearClientActorHandle {
     type Error = NearClientError;
-    async fn is_syncing(&self) -> Result<bool, Self::Error> {
+    async fn sync_status(&self) -> Result<SyncStatus, Self::Error> {
+        // `detailed: true` so the response carries connected-peer heights, which
+        // we use to confirm the head has actually caught up.
         let status_request = near_client::Status {
             is_health_check: false,
-            detailed: false,
+            detailed: true,
         };
-        let status = &self
+        let status = self
             .client
             .send_async(
                 near_o11y::span_wrapped_msg::SpanWrappedMessageExt::span_wrap(status_request),
@@ -40,6 +45,18 @@ impl IsSyncing for NearClientActorHandle {
             .map_err(|err| NearClientError::ResponseError {
                 message: err.to_string(),
             })?;
-        Ok(status.sync_info.syncing)
+        let max_peer_height = status.detailed_debug_status.as_ref().and_then(|detailed| {
+            detailed
+                .network_info
+                .connected_peers
+                .iter()
+                .filter_map(|peer| peer.height)
+                .max()
+        });
+        Ok(SyncStatus {
+            syncing: status.sync_info.syncing,
+            head_height: status.sync_info.latest_block_height,
+            max_peer_height,
+        })
     }
 }

crates/chain-gateway/src/primitives.rs

@@ -10,21 +10,60 @@ use near_indexer::near_primitives::transaction::SignedTransaction;
 use std::future::Future;
 use std::time::Duration;
 
+/// Snapshot of the node's sync progress relative to its peers.
+pub(crate) struct SyncStatus {
+    /// Whether the node currently reports itself as syncing.
+    pub syncing: bool,
+    /// Height of the node's local head.
+    pub head_height: u64,
+    /// Highest block height advertised by a connected peer, if any.
+    pub max_peer_height: Option<u64>,
+}
+
+/// How far behind the highest peer the local head may be while still counting
+/// as caught up. Absorbs peers advancing a block or two between polls; the
+/// `syncing` flag carries the steady state.
+const SYNC_HEIGHT_TOLERANCE: u64 = 5;
+
+impl SyncStatus {
+    /// Whether the node is fully synced to the network head.
+    ///
+    /// The `syncing` flag alone is insufficient: on a freshly state-syncing
+    /// node (or one returning from long downtime) it reads `false` during the
+    /// startup window before the node has learned it is behind, which would pin
+    /// the streamer's `LatestSynced` cursor at a stale head it can never reach.
+    /// We also require the head to be within [`SYNC_HEIGHT_TOLERANCE`] of the
+    /// highest connected peer. While no peer reports a height we cannot confirm
+    /// we are caught up, so we keep waiting.
+    fn is_caught_up(&self) -> bool {
+        if self.syncing {
+            return false;
+        }
+        match self.max_peer_height {
+            None => false,
+            Some(peer_height) => {
+                peer_height.saturating_sub(self.head_height) <= SYNC_HEIGHT_TOLERANCE
+            }
+        }
+    }
+}
+
 /// Low-level trait for checking indexer sync status.
 pub(crate) trait IsSyncing: Send + Sync + 'static {
     type Error: std::error::Error + Send + Sync + 'static;
-    /// Returns whether the node is currently syncing.
-    fn is_syncing(&self) -> impl Future<Output = Result<bool, Self::Error>> + Send;
+    /// Returns the node's current sync progress relative to its peers.
+    fn sync_status(&self) -> impl Future<Output = Result<SyncStatus, Self::Error>> + Send;
 
     const INTERVAL: Duration = Duration::from_millis(500);
-    /// Polls [`is_syncing`](Self::is_syncing) until the node is fully synced.
+    /// Polls [`sync_status`](Self::sync_status) until the node has caught up to
+    /// the network head.
     fn wait_for_full_sync(&self) -> impl Future<Output = ()> + Send {
         async {
             let mut attempt = 0u32;
             loop {
-                match self.is_syncing().await {
-                    Ok(false) => return,
-                    Ok(true) => {
+                match self.sync_status().await {
+                    Ok(status) if status.is_caught_up() => return,
+                    Ok(_) => {
                         if attempt.is_multiple_of(120) {
                             tracing::info!("has been syncing for: {} seconds", attempt / 2);
                         }
@@ -65,3 +104,132 @@ pub(crate) trait SubmitSignedTransaction: Send + Sync + 'static {
         transaction: SignedTransaction,
     ) -> impl Future<Output = Result<(), Self::Error>> + Send;
 }
+
+#[cfg(test)]
+#[expect(non_snake_case)]
+mod tests {
+    use super::{SYNC_HEIGHT_TOLERANCE, SyncStatus};
+
+    #[test]
+    fn is_caught_up__should_be_false_while_node_reports_syncing() {
+        // Given
+        let status = SyncStatus {
+            syncing: true,
+            head_height: 257_000_000,
+            max_peer_height: Some(257_000_000),
+        };
+
+        // When
+        let caught_up = status.is_caught_up();
+
+        // Then
+        assert!(!caught_up);
+    }
+
+    /// The #3623 wedge: at fresh boot the node sits at genesis with `syncing`
+    /// transiently `false` before it has learned a peer is far ahead.
+    #[test]
+    fn is_caught_up__should_be_false_at_genesis_before_sync_starts() {
+        // Given
+        let status = SyncStatus {
+            syncing: false,
+            head_height: 42_376_888,
+            max_peer_height: Some(257_409_058),
+        };
+
+        // When
+        let caught_up = status.is_caught_up();
+
+        // Then
+        assert!(!caught_up);
+    }
+
+    /// Same wedge after long downtime: the stale head is far above genesis but
+    /// still far below the peers.
+    #[test]
+    fn is_caught_up__should_be_false_with_stale_head_far_above_genesis() {
+        // Given
+        let status = SyncStatus {
+            syncing: false,
+            head_height: 200_000_000,
+            max_peer_height: Some(257_409_058),
+        };
+
+        // When
+        let caught_up = status.is_caught_up();
+
+        // Then
+        assert!(!caught_up);
+    }
+
+    /// Until a peer advertises a height we cannot confirm we are caught up.
+    #[test]
+    fn is_caught_up__should_be_false_when_no_peer_height_known() {
+        // Given
+        let status = SyncStatus {
+            syncing: false,
+            head_height: 257_409_058,
+            max_peer_height: None,
+        };
+
+        // When
+        let caught_up = status.is_caught_up();
+
+        // Then
+        assert!(!caught_up);
+    }
+
+    #[test]
+    fn is_caught_up__should_be_true_when_head_reaches_peer_height() {
+        // Given
+        let peer_head = 257_409_058;
+        let status = SyncStatus {
+            syncing: false,
+            head_height: peer_head,
+            max_peer_height: Some(peer_head),
+        };
+
+        // When
+        let caught_up = status.is_caught_up();
+
+        // Then
+        assert!(caught_up);
+    }
+
+    /// Peers may advance a few blocks between polls; being within the tolerance
+    /// still counts as caught up.
+    #[test]
+    fn is_caught_up__should_be_true_within_tolerance_of_peer_height() {
+        // Given
+        let peer_head = 257_409_058;
+        let status = SyncStatus {
+            syncing: false,
+            head_height: peer_head - SYNC_HEIGHT_TOLERANCE,
+            max_peer_height: Some(peer_head),
+        };
+
+        // When
+        let caught_up = status.is_caught_up();
+
+        // Then
+        assert!(caught_up);
+    }
+
+    /// A node slightly ahead of the peers it currently sees is caught up.
+    #[test]
+    fn is_caught_up__should_be_true_when_head_above_peer_height() {
+        // Given
+        let peer_head = 257_409_058;
+        let status = SyncStatus {
+            syncing: false,
+            head_height: peer_head + 10,
+            max_peer_height: Some(peer_head),
+        };
+
+        // When
+        let caught_up = status.is_caught_up();
+
+        // Then
+        assert!(caught_up);
+    }
+}