modify is_stable to be indexer progressing and height caught up

chain-signatures/node/src/mesh/connection.rs

@@ -2,6 +2,7 @@ use std::collections::HashMap;
 use std::time::{Duration, Instant};
 
 use cait_sith::protocol::Participant;
+use near_primitives::types::BlockHeight;
 use tokio::sync::RwLock;
 use url::Url;
 
@@ -161,14 +162,61 @@ impl Pool {
         self.potential_connections.read().await.clone()
     }
 
-    pub async fn is_participant_stable(&self, participant: &Participant) -> bool {
+    async fn max_block_height_among_participants(&self) -> BlockHeight {
+        self.status
+            .read()
+            .await
+            .values()
+            .filter_map(|state| {
+                if let StateView::Running {
+                    latest_block_height,
+                    ..
+                } = state
+                {
+                    Some(*latest_block_height)
+                } else {
+                    None
+                }
+            })
+            .max()
+            .unwrap_or(0)
+    }
+
+    pub async fn is_participant_indexer_progressing(&self, participant: &Participant) -> bool {
         self.status
             .read()
             .await
             .get(participant)
             .map_or(false, |state| match state {
-                StateView::Running { is_stable, .. } => *is_stable,
+                StateView::Running {
+                    is_indexer_progressing,
+                    ..
+                } => *is_indexer_progressing,
                 _ => false,
             })
     }
+
+    pub async fn is_participant_indexer_caught_up(&self, participant: &Participant) -> bool {
+        let max_block_height = self.max_block_height_among_participants().await;
+
+        if max_block_height == 0 {
+            return false;
+        }
+
+        let my_block_height = self
+            .status
+            .read()
+            .await
+            .get(participant)
+            .and_then(|state| match state {
+                StateView::Running {
+                    latest_block_height,
+                    ..
+                } => Some(*latest_block_height),
+                _ => None,
+            })
+            .unwrap_or(0);
+
+        (max_block_height - my_block_height) < 50
+    }
 }

chain-signatures/node/src/mesh/mod.rs

@@ -53,7 +53,15 @@ impl Mesh {
     pub async fn stable_participants(&self) -> Participants {
         let mut stable = Participants::default();
         for (participant, info) in self.active_participants().iter() {
-            if self.connections.is_participant_stable(participant).await {
+            if self
+                .connections
+                .is_participant_indexer_progressing(participant)
+                .await
+                && self
+                    .connections
+                    .is_participant_indexer_caught_up(participant)
+                    .await
+            {
                 stable.insert(participant, info.clone());
             }
         }

chain-signatures/node/src/web/mod.rs

@@ -112,13 +112,13 @@ pub enum StateView {
         presignature_mine_count: usize,
         presignature_potential_count: usize,
         latest_block_height: BlockHeight,
-        is_stable: bool,
+        is_indexer_progressing: bool,
     },
     Resharing {
         old_participants: Vec<Participant>,
         new_participants: Vec<Participant>,
         latest_block_height: BlockHeight,
-        is_stable: bool,
+        is_indexer_progressing: bool,
     },
     Joining {
         participants: Vec<Participant>,
@@ -131,7 +131,7 @@ pub enum StateView {
 async fn state(Extension(state): Extension<Arc<AxumState>>) -> Result<Json<StateView>> {
     tracing::debug!("fetching state");
     let latest_block_height = state.indexer.latest_block_height().await;
-    let is_stable = state.indexer.is_on_track().await;
+    let is_indexer_progressing = state.indexer.is_on_track().await;
     let protocol_state = state.protocol_state.read().await;
 
     match &*protocol_state {
@@ -155,7 +155,7 @@ async fn state(Extension(state): Extension<Arc<AxumState>>) -> Result<Json<State
                 presignature_mine_count,
                 presignature_potential_count,
                 latest_block_height,
-                is_stable,
+                is_indexer_progressing,
             }))
         }
         NodeState::Resharing(state) => {
@@ -165,7 +165,7 @@ async fn state(Extension(state): Extension<Arc<AxumState>>) -> Result<Json<State
                 old_participants,
                 new_participants,
                 latest_block_height,
-                is_stable,
+                is_indexer_progressing,
             }))
         }
         NodeState::Joining(state) => {