Fix execute_schedule method leaking operational data

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 a piece of `OperationalData` is bound for.
 #[derive(Clone, Copy, PartialEq)]
 pub enum OperationalDataTarget {
+    /// The source chain.
     Source,
+    /// The destination chain.
     Destination,
 }
 

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,109 @@ 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 surpassed
+    /// its target chain's:
+    /// 1. Latest timestamp
+    /// 2. Maximum block time
+    /// 3. Latest height
+    ///
+    /// 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 {
+                        match self
+                            .relay_from_operational_data::<relay_sender::AsyncSender>(od.clone())
+                        {
+                            Ok(reply) => self.enqueue_pending_tx(reply, od),
+                            Err(e) => {
+                                unprocessed.extend(operations);
 
-            self.enqueue_pending_tx(reply, od);
+                                return Err((unprocessed, e));
+                            }
+                        }
+                    } else {
+                        unprocessed.push_back(od);
+                    }
+                }
+                Err(e) => {
+                    unprocessed.push_back(od);
+                    unprocessed.extend(operations);
+
+                    return Err((unprocessed, e));
+                }
+            }
         }
 
-        for od in src_ods {
-            let reply =
-                self.relay_from_operational_data::<relay_sender::AsyncSender>(od.clone())?;
-            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);
+            }
+        }
+
+        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,82 @@
+//! This test ensures that the `RelayPath::execute_schedule` method does not
+//! drop any pending scheduled operational data when a prior transaction 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 at least 2 IBC transfers
+//! from chain A to chain B. Chain B is then shut down in order to force the transactions
+//! to be queued up again for re-submission. It is expected that the first transfer
+//! be dropped and not be present in the pending queue, but all of the subsequent
+//! transactions 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};
+
+const NUM_TXS: 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;
+
+        for i in 0..NUM_TXS {
+            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 transfer #{} from chain A to chain B", i);
+        }
+
+        assert_eq!(relay_path_a_to_b.dst_operational_data.len(), NUM_TXS);
+
+        chains.node_b.value().kill()?;
+
+        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(), NUM_TXS - 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;