Fix execute_schedule method leaking operational data

.changelog/unreleased/bug-fixes/ibc-relayer/1153-fix-execute-schedule-leaky-pipeline.md

@@ -0,0 +1,2 @@
+- Fix `execute_schedule` method dropping operational data due to improper
+  handling of errors. ([#2118](https://github.com/informalsystems/ibc-rs/issues/1153))

relayer/src/link/operational_data.rs

@@ -17,9 +17,12 @@ use crate::chain::tracking::TrackingId;
 use crate::link::error::LinkError;
 use crate::link::RelayPath;
 
+/// The chain that the events associated with a piece of [`OperationalData`] are bound for.
 #[derive(Clone, Copy, PartialEq)]
 pub enum OperationalDataTarget {
+    /// The chain which generated the events associated with the `OperationalData`.
     Source,
+    /// The chain receiving the events associated with the `OperationalData``.
     Destination,
 }
 
@@ -75,18 +78,22 @@ pub struct TransitMessage {
     pub msg: Any,
 }
 
-/// Holds all the necessary information for handling a set of in-transit messages.
-///
-/// Each `OperationalData` item is uniquely identified by the combination of two attributes:
-///     - `target`: represents the target of the packet messages, either source or destination chain,
-///     - `proofs_height`: represents the height for the proofs in all the messages.
-///       Note: this is the height at which the proofs are queried. A client consensus state at
-///       `proofs_height + 1` must exist on-chain in order to verify the proofs.
+/// Holds all the necessary information for handling a batch of in-transit messages. This includes
+/// an event received from a chain along with any other packets related to the event (i.e.
+/// 'receive' or 'timeout' packets) that the relayer has to submit in response to the event.
 #[derive(Clone)]
 pub struct OperationalData {
+    /// Represents the height for the proofs in all the messages. Note that this is the height
+    /// at which the proofs are queried. For example, for Tendermint chains, a client consensus
+    /// state at `proofs_height + 1` must exist on-chain in order to verify the proofs.
     pub proofs_height: Height,
+    /// The batch of messages associated with this piece of operational data.
     pub batch: Vec<TransitMessage>,
+    /// Represents the target of the packet messages, either the source or the destination
+    /// chain.
     pub target: OperationalDataTarget,
+    /// A unique ID for tracking this batch of events starting from when they were received
+    /// until the transactions corresponding to those events is submitted.
     pub tracking_id: TrackingId,
     /// Stores `Some(ConnectionDelay)` if the delay is non-zero and `None` otherwise
     connection_delay: Option<ConnectionDelay>,

relayer/src/link/relay_path.rs

@@ -98,8 +98,8 @@ pub struct RelayPath<ChainA: ChainHandle, ChainB: ChainHandle> {
     // mostly timeout packet messages.
     // The operational data targeting the destination chain
     // comprises mostly RecvPacket and Ack msgs.
-    pub(crate) src_operational_data: Queue<OperationalData>,
-    pub(crate) dst_operational_data: Queue<OperationalData>,
+    pub src_operational_data: Queue<OperationalData>,
+    pub dst_operational_data: Queue<OperationalData>,
 
     // Toggle for the transaction confirmation mechanism.
     confirm_txes: bool,
@@ -391,7 +391,7 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
         for i in 1..=MAX_RETRIES {
             let cleared = self
                 .schedule_recv_packet_and_timeout_msgs(height, tracking_id)
-                .and_then(|()| self.schedule_packet_ack_msgs(height, tracking_id));
+                .and_then(|_| self.schedule_packet_ack_msgs(height, tracking_id));
 
             match cleared {
                 Ok(()) => return Ok(()),
@@ -406,6 +406,7 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
     }
 
     /// Clears any packets that were sent before `height`.
+    /// If no height is passed in, then the latest height of the source chain is used.
     pub fn schedule_packet_clearing(&self, height: Option<Height>) -> Result<(), LinkError> {
         let span = span!(Level::DEBUG, "clear");
         let _enter = span.enter();
@@ -1278,27 +1279,120 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
         }
     }
 
-    /// Checks if there are any operational data items ready,
-    /// and if so performs the relaying of corresponding packets
-    /// to the target chain.
+    /// Drives the relaying of elapsed operational data items meant for
+    /// a specified target chain forward.
     ///
-    /// This method performs relaying using the asynchronous sender.
-    /// Retains the operational data as pending, and associates it
-    /// with one or more transaction hash(es).
-    pub fn execute_schedule(&self) -> Result<(), LinkError> {
-        let (src_ods, dst_ods) = self.try_fetch_scheduled_operational_data()?;
+    /// Given an iterator of `OperationalData` elements, this function
+    /// first determines whether the current piece of operational data
+    /// has elapsed.
+    ///
+    /// A piece of operational data is considered 'elapsed' if it has been waiting
+    /// for an amount of time that surpasses both of the following:
+    /// 1. The time duration specified in the connection delay
+    /// 2. The number of blocks specified in the connection delay
+    ///
+    /// If the current piece of operational data has elapsed, then relaying
+    /// is performed using the asynchronous sender. Operational data is
+    /// retained as pending and is associated with one or more transaction
+    /// hash(es).
+    ///
+    /// Should an error occur when attempting to relay a piece of operational
+    /// data, this function returns all subsequent unprocessed pieces of
+    /// operational data back to the caller so that they can be re-queued
+    /// for processing; the operational data that failed to send is dropped.
+    ///
+    /// Note that pieces of operational data that have not elapsed yet are
+    /// also placed in the 'unprocessed' bucket.
+    fn execute_schedule_for_target_chain<I: Iterator<Item = OperationalData>>(
+        &mut self,
+        mut operations: I,
+        target_chain: OperationalDataTarget,
+    ) -> Result<VecDeque<OperationalData>, (VecDeque<OperationalData>, LinkError)> {
+        let mut unprocessed = VecDeque::new();
+
+        while let Some(od) = operations.next() {
+            let elapsed_result = match target_chain {
+                OperationalDataTarget::Source => od.has_conn_delay_elapsed(
+                    &|| self.src_time_latest(),
+                    &|| self.src_max_block_time(),
+                    &|| self.src_latest_height(),
+                ),
+                OperationalDataTarget::Destination => od.has_conn_delay_elapsed(
+                    &|| self.dst_time_latest(),
+                    &|| self.dst_max_block_time(),
+                    &|| self.dst_latest_height(),
+                ),
+            };
 
-        for od in dst_ods {
-            let reply =
-                self.relay_from_operational_data::<relay_sender::AsyncSender>(od.clone())?;
+            match elapsed_result {
+                Ok(elapsed) => {
+                    if elapsed {
+                        // The current piece of operational data has elapsed; we can go ahead and
+                        // attempt to relay it.
+                        match self
+                            .relay_from_operational_data::<relay_sender::AsyncSender>(od.clone())
+                        {
+                            // The operational data was successfully relayed; enqueue the associated tx.
+                            Ok(reply) => self.enqueue_pending_tx(reply, od),
+                            // The relaying process failed; return all of the subsequent pieces of operational
+                            // data along with the underlying error that occurred.
+                            Err(e) => {
+                                unprocessed.extend(operations);
+
+                                return Err((unprocessed, e));
+                            }
+                        }
+                    } else {
+                        // The current piece of operational data has not elapsed; add it to the bucket
+                        // of unprocessed operational data and continue processing subsequent pieces
+                        // of operational data.
+                        unprocessed.push_back(od);
+                    }
+                }
+                Err(e) => {
+                    // An error occurred when attempting to determine whether the current piece of
+                    // operational data has elapsed or not. Add the current piece of data, along with
+                    // all of the subsequent pieces of data, to the unprocessed bucket and return it
+                    // along with the error that resulted.
+                    unprocessed.push_back(od);
+                    unprocessed.extend(operations);
+
+                    return Err((unprocessed, e));
+                }
+            }
+        }
 
-            self.enqueue_pending_tx(reply, od);
+        Ok(unprocessed)
+    }
+
+    /// While there are pending operational data items, this function
+    /// performs the relaying of packets corresponding to those
+    /// operational data items to both the source and destination chains.
+    ///
+    /// Any operational data items that do not get successfully relayed are
+    /// dropped. Subsequent pending operational data items that went unprocessed
+    /// are queued up again for re-submission.
+    pub fn execute_schedule(&mut self) -> Result<(), LinkError> {
+        let src_od_iter = self.src_operational_data.take().into_iter();
+
+        match self.execute_schedule_for_target_chain(src_od_iter, OperationalDataTarget::Source) {
+            Ok(unprocessed_src_data) => self.src_operational_data = unprocessed_src_data.into(),
+            Err((unprocessed_src_data, e)) => {
+                self.src_operational_data = unprocessed_src_data.into();
+                return Err(e);
+            }
         }
 
-        for od in src_ods {
-            let reply =
-                self.relay_from_operational_data::<relay_sender::AsyncSender>(od.clone())?;
-            self.enqueue_pending_tx(reply, od);
+        let dst_od_iter = self.dst_operational_data.take().into_iter();
+
+        match self
+            .execute_schedule_for_target_chain(dst_od_iter, OperationalDataTarget::Destination)
+        {
+            Ok(unprocessed_dst_data) => self.dst_operational_data = unprocessed_dst_data.into(),
+            Err((unprocessed_dst_data, e)) => {
+                self.dst_operational_data = unprocessed_dst_data.into();
+                return Err(e);
+            }
         }
 
         Ok(())

relayer/src/util/queue.rs

@@ -63,3 +63,9 @@ impl<T> Default for Queue<T> {
         Self::new()
     }
 }
+
+impl<T> From<VecDeque<T>> for Queue<T> {
+    fn from(deque: VecDeque<T>) -> Self {
+        Queue(Arc::new(RwLock::new(deque)))
+    }
+}

relayer/src/worker/packet.rs

@@ -73,7 +73,7 @@ pub fn spawn_packet_worker<ChainA: ChainHandle, ChainB: ChainHandle>(
     };
 
     spawn_background_task(span, Some(Duration::from_millis(1000)), move || {
-        let relay_path = &link.lock().unwrap().a_to_b;
+        let relay_path = &mut link.lock().unwrap().a_to_b;
 
         relay_path
             .refresh_schedule()
@@ -120,7 +120,7 @@ pub fn spawn_packet_cmd_worker<ChainA: ChainHandle, ChainB: ChainHandle>(
             retry_with_index(retry_strategy::worker_stubborn_strategy(), |index| {
                 handle_packet_cmd(
                     &mut is_first_run,
-                    &link.lock().unwrap(),
+                    &mut link.lock().unwrap(),
                     clear_on_start,
                     clear_interval,
                     &path,
@@ -145,7 +145,7 @@ pub fn spawn_packet_cmd_worker<ChainA: ChainHandle, ChainB: ChainHandle>(
 /// data that is ready.
 fn handle_packet_cmd<ChainA: ChainHandle, ChainB: ChainHandle>(
     is_first_run: &mut bool,
-    link: &Link<ChainA, ChainB>,
+    link: &mut Link<ChainA, ChainB>,
     clear_on_start: bool,
     clear_interval: u64,
     path: &Packet,

tools/integration-test/src/tests/execute_schedule.rs

@@ -0,0 +1,92 @@
+//! This test ensures that the `RelayPath::execute_schedule` method does not
+//! drop any scheduled `OperationalData` when events associated with a prior
+//! piece of operational data fails to send. Subsequent pieces of operational
+//! data that were scheduled should be re-queued and not dropped.
+//!
+//! In order to test this behavior, the test manually relays a batch (i.e. at least
+//! 2) IBC transfers from chain A to chain B. Chain B is then shut down in order to
+//! force the batch of messages (in the form of their associated pieces of operational
+//! data) to be queued up again for re-submission.
+//!
+//! It is expected that the first message of the batch gets dropped (i.e. it is not
+//! later found in the pending queue), but all of the subsequent messages should
+//! exist in the pending queue.
+
+use ibc_test_framework::prelude::*;
+use ibc_test_framework::util::random::random_u64_range;
+
+use ibc_relayer::link::{Link, LinkParameters};
+
+/// The number of messages to be sent in a batch contained in a piece of operational data.
+const BATCH_SIZE: usize = 10;
+
+#[test]
+fn test_execute_schedule() -> Result<(), Error> {
+    run_binary_channel_test(&ExecuteScheduleTest)
+}
+
+pub struct ExecuteScheduleTest;
+
+impl TestOverrides for ExecuteScheduleTest {
+    fn should_spawn_supervisor(&self) -> bool {
+        false
+    }
+}
+
+impl BinaryChannelTest for ExecuteScheduleTest {
+    fn run<ChainA: ChainHandle, ChainB: ChainHandle>(
+        &self,
+        _config: &TestConfig,
+        _relayer: RelayerDriver,
+        chains: ConnectedChains<ChainA, ChainB>,
+        channel: ConnectedChannel<ChainA, ChainB>,
+    ) -> Result<(), Error> {
+        let amount1 = random_u64_range(1000, 5000);
+
+        let chain_a_link_opts = LinkParameters {
+            src_port_id: channel.port_a.clone().into_value(),
+            src_channel_id: channel.channel_id_a.clone().into_value(),
+        };
+
+        let chain_a_link = Link::new_from_opts(
+            chains.handle_a().clone(),
+            chains.handle_b().clone(),
+            chain_a_link_opts,
+            true,
+        )?;
+
+        let mut relay_path_a_to_b = chain_a_link.a_to_b;
+
+        // Construct `BATCH_SIZE` pieces of operational data and queue them up to be sent to chain B.
+        for i in 0..BATCH_SIZE {
+            chains.node_a.chain_driver().ibc_transfer_token(
+                &channel.port_a.as_ref(),
+                &channel.channel_id_a.as_ref(),
+                &chains.node_a.wallets().user1(),
+                &chains.node_b.wallets().user1().address(),
+                &chains.node_a.denom(),
+                amount1,
+            )?;
+
+            relay_path_a_to_b.schedule_packet_clearing(None)?;
+
+            info!("Performing IBC send packet with a token transfer #{} from chain A to be received by chain B", i);
+        }
+
+        // We should see that all of the events in the batch are queued up to be sent to chain B.
+        assert_eq!(relay_path_a_to_b.dst_operational_data.len(), BATCH_SIZE);
+
+        chains.node_b.value().kill()?;
+
+        // With chain B inactive, if we attempt to send the batch of messages, we expect to see
+        // `BATCH_SIZE` - 1 messages in the batch since the initial event should have failed to
+        // be relayed and was thus dropped. The subsequent messages in the batch should have all
+        // been re-added to the pending queue.
+        match relay_path_a_to_b.execute_schedule() {
+            Ok(_) => panic!("Expected an error when relaying tx from A to B"),
+            Err(_) => assert_eq!(relay_path_a_to_b.dst_operational_data.len(), BATCH_SIZE - 1),
+        }
+
+        Ok(())
+    }
+}

tools/integration-test/src/tests/mod.rs

@@ -9,6 +9,7 @@ pub mod clear_packet;
 pub mod client_expiration;
 mod client_settings;
 pub mod connection_delay;
+pub mod execute_schedule;
 pub mod memo;
 pub mod python;
 mod query_packet;