fix(sync): add stale request cleanup to prevent sync blocking

Author: qj0r9j0vc2

vendor/malachitebft-sync/src/handle.rs

@@ -95,7 +95,7 @@ where
 pub async fn on_tick<Ctx>(
     co: Co<Ctx>,
     state: &mut State<Ctx>,
-    _metrics: &Metrics,
+    metrics: &Metrics,
 ) -> Result<(), Error<Ctx>>
 where
     Ctx: Context,
@@ -114,6 +114,38 @@ where
             .reset_inactive_peers_scores(inactive_threshold);
     }
 
+    // Clear stale pending requests that have timed out
+    let cleared_count = state.clear_stale_pending_requests();
+    if cleared_count > 0 {
+        warn!(
+            cleared_count,
+            height.sync = %state.sync_height,
+            pending_requests = state.pending_value_requests.len(),
+            "SYNC STALE CLEANUP: Cleared timed-out pending requests"
+        );
+
+        // After clearing stale requests, check if we need to skip ahead
+        // because peers can't serve the heights we're requesting
+        if let Some((skip_to, skip_peer)) = state.find_earliest_syncable_height(state.sync_height) {
+            if skip_to > state.sync_height {
+                warn!(
+                    height.current = %state.sync_height,
+                    height.skip_to = %skip_to,
+                    peer = %skip_peer,
+                    "SYNC SKIP AFTER CLEANUP: Peers cannot serve old heights, jumping to earliest available"
+                );
+
+                state.sync_height = skip_to;
+                state.tip_height = skip_to.decrement().unwrap_or_default();
+                state.pending_value_requests.clear();
+                state.height_per_request_id.clear();
+
+                // Request from the new height
+                request_value_from_peer(&co, state, metrics, skip_to, skip_peer).await?;
+            }
+        }
+    }
+
     debug!("Peer scores: {:#?}", state.peer_scorer.get_scores());
 
     Ok(())

vendor/malachitebft-sync/src/state.rs

@@ -1,4 +1,5 @@
 use std::collections::BTreeMap;
+use std::time::Instant;
 
 use malachitebft_core_types::{Context, Height};
 use malachitebft_peer::PeerId;
@@ -39,7 +40,8 @@ where
     pub sync_height: Ctx::Height,
 
     /// Decided value requests for these heights have been sent out to peers.
-    pub pending_value_requests: BTreeMap<Ctx::Height, (OutboundRequestId, RequestState)>,
+    /// Tuple contains: (request_id, state, timestamp_when_sent)
+    pub pending_value_requests: BTreeMap<Ctx::Height, (OutboundRequestId, RequestState, Instant)>,
 
     /// Maps request ID to height for pending decided value requests.
     pub height_per_request_id: BTreeMap<OutboundRequestId, Ctx::Height>,
@@ -133,14 +135,14 @@ where
             .insert(request_id.clone(), height);
 
         self.pending_value_requests
-            .insert(height, (request_id, RequestState::WaitingResponse));
+            .insert(height, (request_id, RequestState::WaitingResponse, Instant::now()));
     }
 
     /// Mark that a response has been received for a height.
     ///
     /// State transition: WaitingResponse -> WaitingValidation
     pub fn response_received(&mut self, request_id: OutboundRequestId, height: Ctx::Height) {
-        if let Some((req_id, state)) = self.pending_value_requests.get_mut(&height) {
+        if let Some((req_id, state, _)) = self.pending_value_requests.get_mut(&height) {
             if req_id != &request_id {
                 return; // A new request has been made in the meantime, ignore this response.
             }
@@ -155,7 +157,7 @@ where
     /// State transition: WaitingValidation -> Validated
     /// It is also possible to have the following transition: WaitingResponse -> Validated.
     pub fn validate_response(&mut self, height: Ctx::Height) {
-        if let Some((_, state)) = self.pending_value_requests.get_mut(&height) {
+        if let Some((_, state, _)) = self.pending_value_requests.get_mut(&height) {
             *state = RequestState::Validated;
         }
     }
@@ -167,7 +169,7 @@ where
 
     /// Remove the pending decided value request for a given height.
     pub fn remove_pending_request_by_height(&mut self, height: &Ctx::Height) {
-        if let Some((request_id, _)) = self.pending_value_requests.remove(height) {
+        if let Some((request_id, _, _)) = self.pending_value_requests.remove(height) {
             self.height_per_request_id.remove(&request_id);
         }
     }
@@ -191,7 +193,7 @@ where
 
     /// Check if a pending decided value request for a given height is in the `Validated` state.
     pub fn is_pending_value_request_validated_by_height(&self, height: &Ctx::Height) -> bool {
-        if let Some((_, state)) = self.pending_value_requests.get(height) {
+        if let Some((_, state, _)) = self.pending_value_requests.get(height) {
             *state == RequestState::Validated
         } else {
             false
@@ -207,6 +209,32 @@ where
         }
     }
 
+    /// Clear stale pending requests that have been waiting longer than the configured timeout.
+    /// Returns the number of cleared requests.
+    pub fn clear_stale_pending_requests(&mut self) -> usize {
+        let timeout = self.config.request_timeout;
+        let now = Instant::now();
+
+        // Find heights with stale requests (not yet validated and older than timeout)
+        let stale_heights: Vec<Ctx::Height> = self
+            .pending_value_requests
+            .iter()
+            .filter(|(_, (_, state, sent_at))| {
+                *state != RequestState::Validated && now.duration_since(*sent_at) > timeout
+            })
+            .map(|(height, _)| *height)
+            .collect();
+
+        let count = stale_heights.len();
+
+        // Remove stale requests
+        for height in stale_heights {
+            self.remove_pending_request_by_height(&height);
+        }
+
+        count
+    }
+
     /// Find the earliest height that any peer can serve, above the given height.
     /// Returns the height and a peer that can serve it.
     /// This is used when the requested height is not available from any peer.