fix: account timeout properly for lagging nodes (#3890)

Author: gilcu3

crates/node/src/requests/debug.rs

@@ -1,4 +1,6 @@
-use super::queue::{ComputationProgress, PendingRequests, QueuedRequest};
+use super::queue::{
+    ComputationProgress, EligibleLeadersAndHeights, PendingRequests, QueuedRequest,
+};
 use crate::indexer::types::ChainRespondArgs;
 use crate::primitives::ParticipantId;
 use crate::types::Request;
@@ -178,7 +180,11 @@ impl<RequestType: Request + Clone, ChainRespondArgsType: ChainRespondArgs> Debug
 {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let mut request_lines = Vec::new();
-        let (eligible_leaders, maximum_height) = self.eligible_leaders_and_maximum_height();
+        let EligibleLeadersAndHeights {
+            eligible_leaders,
+            maximum_height,
+            ..
+        } = self.eligible_leaders_and_heights();
         let online_participants = self.network_api.alive_participants();
         let indexer_heights = self.network_api.indexer_heights();
 

crates/node/src/requests/queue.rs

@@ -442,6 +442,12 @@ impl DropReason {
     }
 }
 
+pub(super) struct EligibleLeadersAndHeights {
+    pub eligible_leaders: HashSet<ParticipantId>,
+    pub maximum_height: u64,
+    pub my_indexer_height: u64,
+}
+
 impl<RequestType: Request + Clone, ChainRespondArgsType: ChainRespondArgs>
     PendingRequests<RequestType, ChainRespondArgsType>
 {
@@ -524,8 +530,8 @@ impl<RequestType: Request + Clone, ChainRespondArgsType: ChainRespondArgs>
     }
 
     /// Returns the set of participants that are eligible to be leaders for the requests,
-    /// as well as the maximum height available.
-    pub(super) fn eligible_leaders_and_maximum_height(&self) -> (HashSet<ParticipantId>, u64) {
+    /// the maximum height available across alive participants, and our own indexer height.
+    pub(super) fn eligible_leaders_and_heights(&self) -> EligibleLeadersAndHeights {
         // Collect the indexer heights and alive participants. Calculate maximum available height
         // from the alive nodes. Then, filter out the participants that are not alive or are too
         // stale.
@@ -536,6 +542,10 @@ impl<RequestType: Request + Clone, ChainRespondArgsType: ChainRespondArgs>
             .map(|p| indexer_heights.get(p).copied().unwrap_or(0))
             .max()
             .unwrap_or(0);
+        let my_indexer_height = indexer_heights
+            .get(&self.my_participant_id)
+            .copied()
+            .unwrap_or(0);
         let eligible_leaders = self
             .all_participants
             .iter()
@@ -546,7 +556,11 @@ impl<RequestType: Request + Clone, ChainRespondArgsType: ChainRespondArgs>
             })
             .copied()
             .collect::<HashSet<_>>();
-        (eligible_leaders, maximum_height)
+        EligibleLeadersAndHeights {
+            eligible_leaders,
+            maximum_height,
+            my_indexer_height,
+        }
     }
 
     /// Returns the list of requests that we should attempt to generate a response for,
@@ -583,7 +597,11 @@ impl<RequestType: Request + Clone, ChainRespondArgsType: ChainRespondArgs>
         };
         let now = self.clock.now();
 
-        let (eligible_leaders, maximum_height) = self.eligible_leaders_and_maximum_height();
+        let EligibleLeadersAndHeights {
+            eligible_leaders,
+            my_indexer_height,
+            ..
+        } = self.eligible_leaders_and_heights();
         tracing::debug!(target: "request", "Eligible leaders: {:?}", eligible_leaders);
 
         let mut result = Vec::new();
@@ -601,9 +619,10 @@ impl<RequestType: Request + Clone, ChainRespondArgsType: ChainRespondArgs>
         }
         let mut requests_to_remove: Vec<(RequestId, Removal)> = Vec::new();
 
-        // any request strictly older than `cutoff_block` will be considered expired
+        // Any request strictly older than `cutoff_block` is considered expired. This is measured
+        // against our own indexer height to prevent spurious timeouts for old requests
         let cutoff_block: BlockHeight =
-            (maximum_height.saturating_sub(REQUEST_EXPIRATION_BLOCKS) + 1).into();
+            (my_indexer_height.saturating_sub(REQUEST_EXPIRATION_BLOCKS) + 1).into();
 
         for (id, request) in &mut self.requests {
             let _span = tracing::debug_span!(
@@ -1302,4 +1321,56 @@ mod tests {
             "expired request should be removed from the queue"
         );
     }
+
+    #[test_log::test]
+    #[expect(non_snake_case)]
+    fn get_requests_to_attempt__should_not_time_out_request_when_own_indexer_is_behind() {
+        // Given
+        let (mut pending_requests, mut setup) = TestSetup::new();
+        let req = setup.add_request_follower();
+        let block_height = setup.update(&mut pending_requests);
+        let me = setup.participant_ids[TestSetup::MY_INDEX];
+
+        // When
+        setup.network_api.set_height(me, block_height);
+        for (i, p) in setup.participant_ids.iter().enumerate() {
+            if i != TestSetup::MY_INDEX {
+                setup
+                    .network_api
+                    .set_height(*p, block_height + REQUEST_EXPIRATION_BLOCKS);
+            }
+        }
+
+        // Then
+        assert!(pending_requests.get_requests_to_attempt().is_empty());
+        assert!(
+            pending_requests.requests.contains_key(&req.id),
+            "request must not be dropped as timed out while our own indexer is behind"
+        );
+    }
+
+    #[test_log::test]
+    #[expect(non_snake_case)]
+    fn get_requests_to_attempt__should_time_out_request_when_own_indexer_advanced_past_expiration()
+    {
+        // Given
+        let (mut pending_requests, mut setup) = TestSetup::new();
+        let req = setup.add_request_follower();
+        let block_height = setup.update(&mut pending_requests);
+        let me = setup.participant_ids[TestSetup::MY_INDEX];
+
+        // When
+        setup
+            .network_api
+            .set_height(me, block_height + REQUEST_EXPIRATION_BLOCKS);
+        for (i, p) in setup.participant_ids.iter().enumerate() {
+            if i != TestSetup::MY_INDEX {
+                setup.network_api.set_height(*p, block_height);
+            }
+        }
+
+        // Then
+        assert!(pending_requests.get_requests_to_attempt().is_empty());
+        assert!(!pending_requests.requests.contains_key(&req.id));
+    }
 }