fix, refactor(sequencer): refactor ics20 logic (#1495)

## Summary
Fixes incorrect addresses in ics20 refunds, and fixes/adds source vs
sink zone detection of received/refunded asets.

## Background
This PR started with the goal of using
`Ics20WithdrawalFromRollup.rollup_return_address` when emitting ics20
deposit events instead of the rollup's bridge address (which for the
rollup was quite meaningless).

However, because the original logic was overly convoluted, this patch
evolved into a refactor which subsequently revealed 2 more bugs.

## Changes
- Refactor ics20 logic: split up `execute_ics20_transfer` into two
functions `receive_tokens` and `refund_tokens`
- Fix deposit events being emitted with the packet sender/bridge address
as the receiving address on the rollup: instead use
`Ics20WithdrawalFromRollup.rollup_return_address` which is the address
originally supplied and understood by the rollup.
- Fix bridge lock events of source tokens being emitted with an extra
(port, channel) pair added (this resulting token is likely completely
meaningless and not understood by the rollup): instead strip the leading
(port, channel) pair to get the original token on sequencer.
- Fix refunds to a rollup of failed ics20 transfers not checking if
sequencer is their source or sink zone: instead perform the check and
decrease the ibc escrow account is necessary.

## Testing

Renamed and expanded tests to also check for sink vs source zone assets
being received, the correct asset being emitted in bridge lock events,
and the correct rollup return address being emitted in refunds.
Specifically these tests were added or renamed:


- `receive_source_zone_asset_on_sequencer_account`
- `receive_sink_zone_asset_on_sequencer_account`
- `receive_source_zone_asset_on_bridge_account_and_emit_to_rollup`
- `receive_sink_zone_asset_on_bridge_account_and_emit_to_rollup`
- `transfer_to_bridge_is_rejected_due_to_invalid_memo`
- `transfer_to_bridge_account_is_rejected_due_to_not_permitted_token`
- `refund_sequencer_account_with_source_zone_asset`
- `refund_sequencer_account_with_sink_zone_asset`
- `refund_rollup_with_sink_zone_asset`
- `refund_rollup_with_source_zone_asset`
- `refund_rollup_with_source_zone_asset_compat_prefix`

## Related Issues
Closes #1439
Closes #1496
Closes #1514

---------

Co-authored-by: noot <[email protected]>

crates/astria-core/src/primitive/v1/asset/denom.rs

@@ -207,90 +207,50 @@ impl TracePrefixed {
         self.trace.is_empty()
     }
 
-    /// Checks if the trace prefixed denom starts with `s`.
+    /// Checks if the trace prefixed denom has `port` in left-most position.
     ///
     /// # Examples
     ///
     /// ```
     /// use astria_core::primitive::v1::asset::denom::TracePrefixed;
     /// let denom = "four/segments/of/a/denom".parse::<TracePrefixed>().unwrap();
+    /// assert!(denom.has_leading_port("four"));
+    /// assert!(!denom.has_leading_port("segments"));
+    /// assert!(!denom.has_leading_port("of"));
+    /// assert!(!denom.has_leading_port("a"));
+    /// assert!(!denom.has_leading_port("denom"));
+    /// assert!(!denom.has_leading_port(""));
+    /// ```
+    #[must_use]
+    pub fn has_leading_port<T: AsRef<str>>(&self, port: T) -> bool {
+        self.trace.leading_port() == Some(port.as_ref())
+    }
+
+    /// Checks if the trace prefixed denom has `channel` in left-most position.
     ///
-    /// // Empty string is always true:
-    /// assert!(denom.starts_with_str(""));
-    /// // Single slash is always false:
-    /// assert!(!denom.starts_with_str("/"));
-    /// // Emptry strings are false:
-    /// assert!(!denom.starts_with_str(" "));
-    ///
-    /// // In general, whitespace is not trimmed and leads to false
-    /// assert!(!denom.starts_with_str("four/segments /"));
-    ///
-    /// // Trailing slashes don't change the result if they are part of the trace prefix:
-    /// assert!(denom.starts_with_str("four/segments"));
-    /// assert!(denom.starts_with_str("four/segments/"));
-    ///
-    /// // Trailing slashes on the full trace prefix denom however return false:
-    /// assert!(!denom.starts_with_str("four/segments/of/a/denom/"));
-    ///
-    /// // Providing only a port is true
-    /// assert!(denom.starts_with_str("four"));
-    /// // Providing a full port/channel pair followed by just a port is also true
-    /// assert!(denom.starts_with_str("four/segments/of"));
-    ///
-    /// // Half of a port or channel is false
-    /// assert!(!denom.starts_with_str("four/segm"));
+    /// # Examples
     ///
-    /// // The full trace prefixed denom is true:
-    /// assert!(denom.starts_with_str("four/segments/of/a/denom"));
+    /// ```
+    /// use astria_core::primitive::v1::asset::denom::TracePrefixed;
+    /// let denom = "four/segments/of/a/denom".parse::<TracePrefixed>().unwrap();
+    /// assert!(!denom.has_leading_channel("four"));
+    /// assert!(denom.has_leading_channel("segments"));
+    /// assert!(!denom.has_leading_channel("of"));
+    /// assert!(!denom.has_leading_channel("a"));
+    /// assert!(!denom.has_leading_channel("denom"));
+    /// assert!(!denom.has_leading_channel(""));
     /// ```
     #[must_use]
-    pub fn starts_with_str(&self, s: &str) -> bool {
-        if s.is_empty() {
-            return true;
-        }
-        let mut had_trailing_slash = false;
-        let s = s
-            .strip_suffix('/')
-            .inspect(|_| had_trailing_slash = true)
-            .unwrap_or(s);
-        if s.is_empty() {
-            return false;
-        }
-        let mut parts = s.split('/');
-        for segment in self.trace.iter() {
-            // first iteration: we know that s is not empty after stripping the /
-            // so that this is not wrongly returning true.
-            let Some(port) = parts.next() else {
-                return true;
-            };
-            if segment.port() != port {
-                return false;
-            }
-            let Some(channel) = parts.next() else {
-                return true;
-            };
-            if segment.channel() != channel {
-                return false;
-            }
-        }
-        let Some(base_denom) = parts.next() else {
-            return true;
-        };
-        if base_denom != self.base_denom {
-            return false;
-        }
-        if had_trailing_slash {
-            return false;
-        }
-        parts.next().is_none()
+    pub fn has_leading_channel<T: AsRef<str>>(&self, channel: T) -> bool {
+        self.trace.leading_channel() == Some(channel.as_ref())
     }
 
     #[must_use]
     pub fn last_channel(&self) -> Option<&str> {
         self.trace.last_channel()
     }
 
-    pub fn pop_trace_segment(&mut self) -> Option<PortAndChannel> {
+    pub fn pop_leading_port_and_channel(&mut self) -> Option<PortAndChannel> {
         self.trace.pop()
     }
 
@@ -311,6 +271,14 @@ impl TraceSegments {
         }
     }
 
+    fn leading_port(&self) -> Option<&str> {
+        self.inner.front().map(|segment| &*segment.port)
+    }
+
+    fn leading_channel(&self) -> Option<&str> {
+        self.inner.front().map(|segment| &*segment.channel)
+    }
+
     fn push(&mut self, seg: PortAndChannel) {
         self.inner.push_back(seg);
     }
@@ -326,10 +294,6 @@ impl TraceSegments {
     fn is_empty(&self) -> bool {
         self.inner.is_empty()
     }
-
-    fn iter(&self) -> impl Iterator<Item = &PortAndChannel> {
-        self.inner.iter()
-    }
 }
 
 impl FromStr for TraceSegments {
@@ -717,37 +681,14 @@ mod tests {
     #[test]
     fn pop_path() {
         let mut denom = "a/long/path/to/denom".parse::<TracePrefixed>().unwrap();
-        let port_and_channel = denom.pop_trace_segment().unwrap();
+        let port_and_channel = denom.pop_leading_port_and_channel().unwrap();
         assert_eq!("a", port_and_channel.port());
         assert_eq!("long", port_and_channel.channel());
 
-        let port_and_channel = denom.pop_trace_segment().unwrap();
+        let port_and_channel = denom.pop_leading_port_and_channel().unwrap();
         assert_eq!("path", port_and_channel.port());
         assert_eq!("to", port_and_channel.channel());
 
-        assert_eq!(None, denom.pop_trace_segment());
-    }
-
-    #[test]
-    fn start_prefixes() {
-        let denom = "four/segments/of/a/denom".parse::<TracePrefixed>().unwrap();
-
-        assert!(denom.starts_with_str(""));
-        assert!(!denom.starts_with_str("/"));
-        assert!(!denom.starts_with_str(" "));
-
-        assert!(!denom.starts_with_str("four/segments /"));
-
-        assert!(denom.starts_with_str("four/segments"));
-        assert!(denom.starts_with_str("four/segments/"));
-
-        assert!(!denom.starts_with_str("four/segments/of/a/denom/"));
-
-        assert!(denom.starts_with_str("four"));
-        assert!(denom.starts_with_str("four/segments/of"));
-
-        assert!(!denom.starts_with_str("four/segm"));
-
-        assert!(denom.starts_with_str("four/segments/of/a/denom"));
+        assert_eq!(None, denom.pop_leading_port_and_channel());
     }
 }

crates/astria-core/src/protocol/test_utils.rs

@@ -116,10 +116,10 @@ impl ConfigureSequencerBlock {
 
         let mut deposits_map: HashMap<RollupId, Vec<Deposit>> = HashMap::new();
         for deposit in deposits {
-            if let Some(entry) = deposits_map.get_mut(deposit.rollup_id()) {
+            if let Some(entry) = deposits_map.get_mut(&deposit.rollup_id) {
                 entry.push(deposit);
             } else {
-                deposits_map.insert(*deposit.rollup_id(), vec![deposit]);
+                deposits_map.insert(deposit.rollup_id, vec![deposit]);
             }
         }
 

crates/astria-core/src/sequencerblock/v1alpha1/block.rs

@@ -1291,85 +1291,23 @@ impl FilteredSequencerBlockError {
 )]
 pub struct Deposit {
     // the address on the sequencer to which the funds were sent to.
-    bridge_address: Address,
+    pub bridge_address: Address,
     // the rollup ID registered to the `bridge_address`
-    rollup_id: RollupId,
+    pub rollup_id: RollupId,
     // the amount that was transferred to `bridge_address`
-    amount: u128,
+    pub amount: u128,
     // the IBC ICS20 denom of the asset that was transferred
-    asset: asset::Denom,
+    pub asset: asset::Denom,
     // the address on the destination chain (rollup) which to send the bridged funds to
-    destination_chain_address: String,
+    pub destination_chain_address: String,
     // the transaction ID of the source action for the deposit, consisting
     // of the transaction hash.
-    source_transaction_id: TransactionId,
+    pub source_transaction_id: TransactionId,
     // index of the deposit's source action within its transaction
-    source_action_index: u64,
-}
-
-impl From<Deposit> for crate::generated::sequencerblock::v1alpha1::Deposit {
-    fn from(deposit: Deposit) -> Self {
-        deposit.into_raw()
-    }
+    pub source_action_index: u64,
 }
 
 impl Deposit {
-    #[must_use]
-    pub fn new(
-        bridge_address: Address,
-        rollup_id: RollupId,
-        amount: u128,
-        asset: asset::Denom,
-        destination_chain_address: String,
-        source_transaction_id: TransactionId,
-        source_action_index: u64,
-    ) -> Self {
-        Self {
-            bridge_address,
-            rollup_id,
-            amount,
-            asset,
-            destination_chain_address,
-            source_transaction_id,
-            source_action_index,
-        }
-    }
-
-    #[must_use]
-    pub fn bridge_address(&self) -> &Address {
-        &self.bridge_address
-    }
-
-    #[must_use]
-    pub fn rollup_id(&self) -> &RollupId {
-        &self.rollup_id
-    }
-
-    #[must_use]
-    pub fn amount(&self) -> u128 {
-        self.amount
-    }
-
-    #[must_use]
-    pub fn asset(&self) -> &asset::Denom {
-        &self.asset
-    }
-
-    #[must_use]
-    pub fn destination_chain_address(&self) -> &str {
-        &self.destination_chain_address
-    }
-
-    #[must_use]
-    pub fn source_transaction_id(&self) -> &TransactionId {
-        &self.source_transaction_id
-    }
-
-    #[must_use]
-    pub fn source_action_index(&self) -> u64 {
-        self.source_action_index
-    }
-
     #[must_use]
     pub fn into_raw(self) -> raw::Deposit {
         let Self {
@@ -1439,6 +1377,12 @@ impl Deposit {
     }
 }
 
+impl From<Deposit> for crate::generated::sequencerblock::v1alpha1::Deposit {
+    fn from(deposit: Deposit) -> Self {
+        deposit.into_raw()
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 #[error(transparent)]
 pub struct DepositError(DepositErrorKind);

crates/astria-sequencer-utils/src/blob_parser.rs

@@ -537,11 +537,11 @@ impl TryFrom<&RawDeposit> for PrintableDeposit {
     fn try_from(raw_deposit: &RawDeposit) -> Result<Self, Self::Error> {
         let deposit = Deposit::try_from_raw(raw_deposit.clone())?;
         Ok(PrintableDeposit {
-            bridge_address: deposit.bridge_address().to_string(),
-            rollup_id: deposit.rollup_id().to_string(),
-            amount: deposit.amount(),
-            asset: deposit.asset().to_string(),
-            destination_chain_address: deposit.destination_chain_address().to_string(),
+            bridge_address: deposit.bridge_address.to_string(),
+            rollup_id: deposit.rollup_id.to_string(),
+            amount: deposit.amount,
+            asset: deposit.asset.to_string(),
+            destination_chain_address: deposit.destination_chain_address.to_string(),
         })
     }
 }

crates/astria-sequencer/src/accounts/mod.rs

@@ -22,12 +22,20 @@ pub(crate) use state_ext::{
 
 pub(crate) trait AddressBytes: Send + Sync {
     fn address_bytes(&self) -> [u8; ADDRESS_LEN];
+
+    fn display_address(&self) -> impl std::fmt::Display {
+        telemetry::display::base64(self.address_bytes())
+    }
 }
 
 impl AddressBytes for Address {
     fn address_bytes(&self) -> [u8; ADDRESS_LEN] {
         self.bytes()
     }
+
+    fn display_address(&self) -> impl std::fmt::Display {
+        self
+    }
 }
 
 impl AddressBytes for [u8; ADDRESS_LEN] {

crates/astria-sequencer/src/accounts/state_ext.rs

@@ -186,7 +186,9 @@ pub(crate) trait StateReadExt: StateRead + crate::assets::StateReadExt {
         }
     }
 
-    #[instrument(skip_all)]
+    // allow: false positive due to proc macro; fixed with rust/clippy 1.81
+    #[allow(clippy::blocks_in_conditions)]
+    #[instrument(skip_all, fields(address = %address.display_address(), %asset), err)]
     async fn get_account_balance<'a, TAddress, TAsset>(
         &self,
         address: TAddress,
@@ -244,7 +246,7 @@ impl<T: StateRead + ?Sized> StateReadExt for T {}
 
 #[async_trait]
 pub(crate) trait StateWriteExt: StateWrite {
-    #[instrument(skip_all)]
+    #[instrument(skip_all, fields(address = %address.display_address(), %asset, balance), err)]
     fn put_account_balance<TAddress, TAsset>(
         &mut self,
         address: TAddress,
@@ -267,7 +269,9 @@ pub(crate) trait StateWriteExt: StateWrite {
         Ok(())
     }
 
-    #[instrument(skip_all)]
+    // allow: false positive due to proc macro; fixed with rust/clippy 1.81
+    #[allow(clippy::blocks_in_conditions)]
+    #[instrument(skip_all, fields(address = %address.display_address(), %asset, amount), err)]
     async fn increase_balance<TAddress, TAsset>(
         &mut self,
         address: TAddress,
@@ -294,7 +298,7 @@ pub(crate) trait StateWriteExt: StateWrite {
         Ok(())
     }
 
-    #[instrument(skip_all)]
+    #[instrument(skip_all, fields(address = %address.display_address(), %asset, amount))]
     async fn decrease_balance<TAddress, TAsset>(
         &mut self,
         address: TAddress,