Retrieve the block results that include the event with the exact attr…

…ibutes in query (#2870)

* Debug packet stuck on apollo

* Get the last block if multiple contain event

* Remove debugs

* Search for the most recent 10 blocks which match the packet query

There should only be a single match for this query, but due to
the fact that the indexer treat the query as a disjunction over
all events in a block rather than a conjunction over a single event,
we may end up with partial matches and therefore have to account for
that by fetching multiple results and filter it down after the fact.

* Order events by begin block, tx, end block

* Update comments

* Add changelog entries

Co-authored-by: Anca Zamfir <anca@informal.systems>

.changelog/unreleased/bug-fixes/2867-retrieve-begin-end-block.md

@@ -0,0 +1,3 @@
+- Fix an issue where Hermes would sometimes fails to retrieve the
+ begin/end block events because of a quirk of the Tendermint event indexer
+ ([#2867](https://github.com/informalsystems/ibc-rs/issues/2867))

.changelog/unreleased/bug-fixes/2868-clear-begin-end.md

@@ -0,0 +1,3 @@
+- Hermes tx CLIs that use the `packet_data_query_height` option now also clear
+ begin/end block events ([#2868](https://github.com/informalsystems/ibc-
+ rs/issues/2868))

crates/relayer/src/chain/cosmos.rs

@@ -573,6 +573,7 @@ impl CosmosSdkChain {
  .map(|ev| IbcEventWithHeight::new(ev, response_height))
  .collect(),
  );
+
  Ok((begin_block_events, end_block_events))
  }
 
@@ -586,22 +587,29 @@ impl CosmosSdkChain {
  let mut begin_block_events = vec![];
  let mut end_block_events = vec![];
 
- for seq in request.sequences.iter() {
+ for seq in request.sequences.iter().copied() {
  let response = self
  .block_on(self.rpc_client.block_search(
- packet_query(request, *seq),
+ packet_query(request, seq),
+ // We only need the first page
  1,
- 1, // there should only be a single match for this query
- Order::Ascending,
+ // There should only be a single match for this query, but due to
+ // the fact that the indexer treat the query as a disjunction over
+ // all events in a block rather than a conjunction over a single event,
+ // we may end up with partial matches and therefore have to account for
+ // that by fetching multiple results and filter it down after the fact.
+ // In the worst case we get N blocks where N is the number of channels,
+ // but 10 seems to work well enough in practice while keeping the response
+ // size, and therefore pressure on the node, fairly low.
+ 10,
+ // We could pick either ordering here, since matching blocks may be at pretty
+ // much any height relative to the target blocks, so we went with most recent
+ // blocks first.
+ Order::Descending,
  ))
  .map_err(|e| Error::rpc(self.config.rpc_addr.clone(), e))?;
 
- assert!(
- response.blocks.len() <= 1,
- "block_search: unexpected number of blocks"
- );
-
- if let Some(block) = response.blocks.first().map(|first| &first.block) {
+ for block in response.blocks.into_iter().map(|response| response.block) {
  let response_height =
  ICSHeight::new(self.id().version(), u64::from(block.header.height))
  .map_err(|_| Error::invalid_height_no_source())?;
@@ -612,13 +620,16 @@ impl CosmosSdkChain {
  }
  }
 
+ // `query_packet_from_block` retrieves the begin and end block events
+ // and filter them to retain only those matching the query
  let (new_begin_block_events, new_end_block_events) =
- self.query_packet_from_block(request, &[*seq], &response_height)?;
+ self.query_packet_from_block(request, &[seq], &response_height)?;
 
  begin_block_events.extend(new_begin_block_events);
  end_block_events.extend(new_end_block_events);
  }
  }
+
  Ok((begin_block_events, end_block_events))
  }
 }

crates/relayer/src/chain/cosmos/query/tx.rs

@@ -151,8 +151,6 @@ pub async fn query_packets_from_block(
  crate::time!("query_packets_from_block");
  crate::telemetry!(query, chain_id, "query_packets_from_block");
 
- let mut result: Vec<IbcEventWithHeight> = vec![];
-
  let tm_height = match request.height.get() {
  QueryHeight::Latest => tendermint::block::Height::default(),
  QueryHeight::Specific(h) => {
@@ -163,17 +161,19 @@ pub async fn query_packets_from_block(
  let height = Height::new(chain_id.version(), u64::from(tm_height))
  .map_err(|_| Error::invalid_height_no_source())?;
 
- let exact_tx_block_results = rpc_client
+ let block_results = rpc_client
  .block_results(tm_height)
  .await
- .map_err(|e| Error::rpc(rpc_address.clone(), e))?
- .txs_results;
-
- if let Some(txs) = exact_tx_block_results {
- for tx in txs.iter() {
- let tx_copy = tx.clone();
- result.append(
- &mut tx_copy
+ .map_err(|e| Error::rpc(rpc_address.clone(), e))?;
+
+ let mut tx_events = vec![];
+ let mut begin_block_events = vec![];
+ let mut end_block_events = vec![];
+
+ if let Some(txs) = block_results.txs_results {
+ for tx in txs {
+ tx_events.append(
+ &mut tx
  .events
  .into_iter()
  .filter_map(|e| filter_matching_event(e, request, &request.sequences))
@@ -183,7 +183,31 @@ pub async fn query_packets_from_block(
  }
  }
 
- Ok(result)
+ begin_block_events.append(
+ &mut block_results
+ .begin_block_events
+ .unwrap_or_default()
+ .into_iter()
+ .filter_map(|ev| filter_matching_event(ev, request, &request.sequences))
+ .map(|ev| IbcEventWithHeight::new(ev, height))
+ .collect(),
+ );
+
+ end_block_events.append(
+ &mut block_results
+ .end_block_events
+ .unwrap_or_default()
+ .into_iter()
+ .filter_map(|ev| filter_matching_event(ev, request, &request.sequences))
+ .map(|ev| IbcEventWithHeight::new(ev, height))
+ .collect(),
+ );
+
+ let mut events = begin_block_events;
+ events.append(&mut tx_events);
+ events.append(&mut end_block_events);
+
+ Ok(events)
 }
 
 // Extracts from the Tx the update client event for the requested client and height.