dex: Swap alt fee tokens for native fee token (#4643)

## Describe your changes

Moves fee handling out of the Transaction's ActionHandler and into
the component, which accumulates fees (in any token), and adds hooks to
the DEX
thatadd chain-submitted swaps for any non-native fee tokens into the
native
token. These are then accumulated back into the fee component.

The base fee and tip are tracked separately through this whole process,
in
order to support the intended behaviour (the base fee is burned to
account for
the _chain_'s resource use, the tip is sent to the proposer to encourage
them
to include transactions in fee priority order). However, tip handling is
not
currently implemented, so both are burned. Later, the
funding/distribution
components should extract the tip and send it to the proposer's funding
streams. However, until a robust fee market develops, this form of
proposer
incentivization can be deferred. The accounting will be there for it
when it
arises.

## Issue ticket number and link

#4328 

## Checklist before requesting a review

- [x] If this code contains consensus-breaking changes, I have added the
"consensus-breaking" label. Otherwise, I declare my belief that there
are not consensus-breaking changes, for the following reason:

  > consensus-breaking but not state-breaking, no migrations necessary

---------

Co-authored-by: Erwan Or <[email protected]>

crates/core/app/src/action_handler/transaction.rs

@@ -3,9 +3,10 @@ use std::sync::Arc;
 use anyhow::Result;
 use async_trait::async_trait;
 use cnidarium::{StateRead, StateWrite};
+use penumbra_fee::component::FeePay as _;
 use penumbra_sct::{component::source::SourceContext, CommitmentSource};
 use penumbra_shielded_pool::component::ClueManager;
-use penumbra_transaction::Transaction;
+use penumbra_transaction::{gas::GasCost as _, Transaction};
 use tokio::task::JoinSet;
 use tracing::{instrument, Instrument};
 
@@ -29,7 +30,20 @@ impl AppActionHandler for Transaction {
     // We only instrument the top-level `check_stateless`, so we get one span for each transaction.
     #[instrument(skip(self, _context))]
     async fn check_stateless(&self, _context: ()) -> Result<()> {
+        // This check should be done first, and complete before all other
+        // stateless checks, like proof verification.  In addition to proving
+        // that value balances, the binding signature binds the proofs to the
+        // transaction, as the binding signature can only be created with
+        // knowledge of all of the openings to the commitments the transaction
+        // makes proofs against. (This is where the name binding signature comes
+        // from).
+        //
+        // This allows us to cheaply eliminate a large class of invalid
+        // transactions upfront -- past this point, we can be sure that the user
+        // who submitted the transaction actually formed the proofs, rather than
+        // replaying them from another transaction.
         valid_binding_signature(self)?;
+        // Other checks probably too cheap to be worth splitting into tasks.
         num_clues_equal_to_num_outputs(self)?;
         check_memo_exists_if_outputs_absent_if_not(self)?;
 
@@ -59,8 +73,9 @@ impl AppActionHandler for Transaction {
     async fn check_historical<S: StateRead + 'static>(&self, state: Arc<S>) -> Result<()> {
         let mut action_checks = JoinSet::new();
 
-        // SAFETY: Transaction parameters (chain id, expiry height, fee) against chain state
+        // SAFETY: Transaction parameters (chain id, expiry height) against chain state
         // that cannot change during transaction execution.
+        // The fee is _not_ checked here, but during execution.
         tx_parameters_historical_check(state.clone(), self).await?;
         // SAFETY: anchors are historical data and cannot change during transaction execution.
         claimed_anchor_is_valid(state.clone(), self).await?;
@@ -95,6 +110,12 @@ impl AppActionHandler for Transaction {
         };
         state.put_current_source(Some(source));
 
+        // Check and record the transaction's fee payment,
+        // before doing the rest of execution.
+        let gas_used = self.gas_cost();
+        let fee = self.transaction_body.transaction_parameters.fee;
+        state.pay_fee(gas_used, fee).await?;
+
         for (i, action) in self.actions().enumerate() {
             let span = action.create_span(i);
             action

crates/core/app/src/action_handler/transaction/stateful.rs

@@ -1,11 +1,9 @@
 use anyhow::{ensure, Result};
 use cnidarium::StateRead;
-use penumbra_fee::component::StateReadExt as _;
 use penumbra_sct::component::clock::EpochRead;
 use penumbra_sct::component::tree::VerificationExt;
 use penumbra_shielded_pool::component::StateReadExt as _;
 use penumbra_shielded_pool::fmd;
-use penumbra_transaction::gas::GasCost;
 use penumbra_transaction::{Transaction, TransactionParameters};
 
 use crate::app::StateReadExt;
@@ -17,8 +15,7 @@ pub async fn tx_parameters_historical_check<S: StateRead>(
     let TransactionParameters {
         chain_id,
         expiry_height,
-        // This is checked in `fee_greater_than_base_fee` against the whole
-        // transaction, for convenience.
+        // This is checked during execution.
         fee: _,
         // IMPORTANT: Adding a transaction parameter? Then you **must** add a SAFETY
         // argument here to justify why it is safe to validate against a historical
@@ -31,9 +28,6 @@ pub async fn tx_parameters_historical_check<S: StateRead>(
     // SAFETY: This is safe to do in a **historical** check because the chain's current
     // block height cannot change during transaction processing.
     expiry_height_is_valid(&state, expiry_height).await?;
-    // SAFETY: This is safe to do in a **historical** check as long as the current gas prices
-    // are static, or set in the previous block.
-    fee_greater_than_base_fee(&state, transaction).await?;
 
     Ok(())
 }
@@ -145,53 +139,3 @@ pub async fn claimed_anchor_is_valid<S: StateRead>(
 ) -> Result<()> {
     state.check_claimed_anchor(transaction.anchor).await
 }
-
-pub async fn fee_greater_than_base_fee<S: StateRead>(
-    state: S,
-    transaction: &Transaction,
-) -> Result<()> {
-    // Check whether the user is requesting to pay fees in the native token
-    // or in an alternative fee token.
-    let user_supplied_fee = transaction.transaction_body().transaction_parameters.fee;
-
-    let current_gas_prices =
-        if user_supplied_fee.asset_id() == *penumbra_asset::STAKING_TOKEN_ASSET_ID {
-            state
-                .get_gas_prices()
-                .await
-                .expect("gas prices must be present in state")
-        } else {
-            let alt_gas_prices = state
-                .get_alt_gas_prices()
-                .await
-                .expect("alt gas prices must be present in state");
-            alt_gas_prices
-                .into_iter()
-                .find(|prices| prices.asset_id == user_supplied_fee.asset_id())
-                .ok_or_else(|| {
-                    anyhow::anyhow!(
-                        "fee token {} not recognized by the chain",
-                        user_supplied_fee.asset_id()
-                    )
-                })?
-        };
-
-    // Double check that the gas price assets match.
-    ensure!(
-        current_gas_prices.asset_id == user_supplied_fee.asset_id(),
-        "unexpected mismatch between fee and queried gas prices (expected: {}, found: {})",
-        user_supplied_fee.asset_id(),
-        current_gas_prices.asset_id,
-    );
-
-    let transaction_base_fee = current_gas_prices.fee(&transaction.gas_cost());
-
-    ensure!(
-        user_supplied_fee.amount() >= transaction_base_fee.amount(),
-        "fee must be greater than or equal to the transaction base price (supplied: {}, base: {})",
-        user_supplied_fee.amount(),
-        transaction_base_fee.amount(),
-    );
-
-    Ok(())
-}

crates/core/app/src/app/mod.rs

@@ -15,7 +15,7 @@ use penumbra_compact_block::component::CompactBlockManager;
 use penumbra_dex::component::StateReadExt as _;
 use penumbra_dex::component::{Dex, StateWriteExt as _};
 use penumbra_distributions::component::{Distributions, StateReadExt as _, StateWriteExt as _};
-use penumbra_fee::component::{Fee, StateReadExt as _, StateWriteExt as _};
+use penumbra_fee::component::{FeeComponent, StateReadExt as _, StateWriteExt as _};
 use penumbra_funding::component::Funding;
 use penumbra_funding::component::{StateReadExt as _, StateWriteExt as _};
 use penumbra_governance::component::{Governance, StateReadExt as _, StateWriteExt as _};
@@ -139,7 +139,7 @@ impl App {
                 CommunityPool::init_chain(&mut state_tx, Some(&genesis.community_pool_content))
                     .await;
                 Governance::init_chain(&mut state_tx, Some(&genesis.governance_content)).await;
-                Fee::init_chain(&mut state_tx, Some(&genesis.fee_content)).await;
+                FeeComponent::init_chain(&mut state_tx, Some(&genesis.fee_content)).await;
                 Funding::init_chain(&mut state_tx, Some(&genesis.funding_content)).await;
 
                 state_tx
@@ -155,7 +155,7 @@ impl App {
                 Dex::init_chain(&mut state_tx, None).await;
                 Governance::init_chain(&mut state_tx, None).await;
                 CommunityPool::init_chain(&mut state_tx, None).await;
-                Fee::init_chain(&mut state_tx, None).await;
+                FeeComponent::init_chain(&mut state_tx, None).await;
                 Funding::init_chain(&mut state_tx, None).await;
             }
         };
@@ -340,7 +340,7 @@ impl App {
         CommunityPool::begin_block(&mut arc_state_tx, begin_block).await;
         Governance::begin_block(&mut arc_state_tx, begin_block).await;
         Staking::begin_block(&mut arc_state_tx, begin_block).await;
-        Fee::begin_block(&mut arc_state_tx, begin_block).await;
+        FeeComponent::begin_block(&mut arc_state_tx, begin_block).await;
         Funding::begin_block(&mut arc_state_tx, begin_block).await;
 
         let state_tx = Arc::try_unwrap(arc_state_tx)
@@ -471,7 +471,7 @@ impl App {
         CommunityPool::end_block(&mut arc_state_tx, end_block).await;
         Governance::end_block(&mut arc_state_tx, end_block).await;
         Staking::end_block(&mut arc_state_tx, end_block).await;
-        Fee::end_block(&mut arc_state_tx, end_block).await;
+        FeeComponent::end_block(&mut arc_state_tx, end_block).await;
         Funding::end_block(&mut arc_state_tx, end_block).await;
         let mut state_tx = Arc::try_unwrap(arc_state_tx)
             .expect("components did not retain copies of shared state");
@@ -533,7 +533,7 @@ impl App {
             Staking::end_epoch(&mut arc_state_tx)
                 .await
                 .expect("able to call end_epoch on Staking component");
-            Fee::end_epoch(&mut arc_state_tx)
+            FeeComponent::end_epoch(&mut arc_state_tx)
                 .await
                 .expect("able to call end_epoch on Fee component");
             Funding::end_epoch(&mut arc_state_tx)

crates/core/app/tests/app_check_dex_vcb.rs

@@ -0,0 +1,75 @@
+mod common;
+
+use self::common::TempStorageExt;
+use cnidarium::{ArcStateDeltaExt, StateDelta, TempStorage};
+use cnidarium_component::ActionHandler;
+use penumbra_asset::asset;
+use penumbra_dex::{
+    component::ValueCircuitBreakerRead,
+    swap::{SwapPlaintext, SwapPlan},
+    TradingPair,
+};
+use penumbra_fee::Fee;
+use penumbra_keys::{test_keys, Address};
+use penumbra_num::Amount;
+use penumbra_sct::component::source::SourceContext;
+use rand_core::SeedableRng;
+use std::{ops::Deref, sync::Arc};
+
+#[tokio::test]
+/// Minimal reproduction of a bug in the DEX VCB swap flow tracking.
+///
+/// Overview: The DEX VCB was double-counting swap flows for a same asset
+/// by computing:  `aggregate = (delta + aggregate) + aggregate`, instead
+/// it should compute: `aggregate = delta + aggregate`.
+/// This bug was fixed in #4643.
+async fn dex_vcb_tracks_multiswap() -> anyhow::Result<()> {
+    let mut rng = rand_chacha::ChaChaRng::seed_from_u64(1776);
+    let storage = TempStorage::new().await?.apply_default_genesis().await?;
+    let mut state = Arc::new(StateDelta::new(storage.latest_snapshot()));
+
+    // Create the first swap:
+    let gm = asset::Cache::with_known_assets().get_unit("gm").unwrap();
+    let gn = asset::Cache::with_known_assets().get_unit("gn").unwrap();
+    let trading_pair = TradingPair::new(gm.id(), gn.id());
+
+    let delta_1 = Amount::from(100_000u64);
+    let delta_2 = Amount::from(0u64);
+    let fee = Fee::default();
+    let claim_address: Address = test_keys::ADDRESS_0.deref().clone();
+    let plaintext =
+        SwapPlaintext::new(&mut rng, trading_pair, delta_1, delta_2, fee, claim_address);
+
+    let swap_plan = SwapPlan::new(&mut rng, plaintext.clone());
+    let swap_one = swap_plan.swap(&test_keys::FULL_VIEWING_KEY);
+
+    let mut state_tx = state.try_begin_transaction().unwrap();
+    state_tx.put_mock_source(1u8);
+    swap_one.check_and_execute(&mut state_tx).await?;
+
+    // Observe the DEX VCB has been credited:
+    let gm_vcb_amount = state_tx
+        .get_dex_vcb_for_asset(&gm.id())
+        .await?
+        .expect("we just accumulated a swap");
+    assert_eq!(
+        gm_vcb_amount,
+        100_000u128.into(),
+        "the DEX VCB does not contain swap 1"
+    );
+
+    // Let's add another swap:
+    let swap_two = swap_one.clone();
+    swap_two.check_and_execute(&mut state_tx).await?;
+    let gm_vcb_amount = state_tx
+        .get_dex_vcb_for_asset(&gm.id())
+        .await?
+        .expect("we accumulated two swaps");
+    assert_eq!(
+        gm_vcb_amount,
+        200_000u128.into(),
+        "the DEX VCB does not contain swap 2"
+    );
+
+    Ok(())
+}

crates/core/component/dex/src/component/action_handler/swap.rs

@@ -9,7 +9,7 @@ use penumbra_proto::StateWriteProto;
 use penumbra_sct::component::source::SourceContext;
 
 use crate::{
-    component::{InternalDexWrite, StateReadExt, SwapDataRead, SwapDataWrite, SwapManager},
+    component::{InternalDexWrite, StateReadExt, SwapDataWrite, SwapManager},
     event,
     swap::{proof::SwapProofPublic, Swap},
 };
@@ -46,18 +46,10 @@ impl ActionHandler for Swap {
 
         let swap = self;
 
-        // All swaps will be tallied for the block so the
-        // BatchSwapOutputData for the trading pair/block height can
-        // be set during `end_block`.
-        let mut swap_flow = state.swap_flow(&swap.body.trading_pair);
-
-        // Add the amount of each asset being swapped to the batch swap flow.
-        swap_flow.0 += swap.body.delta_1_i;
-        swap_flow.1 += swap.body.delta_2_i;
-
-        // Set the batch swap flow for the trading pair.
+        // Accumulate the swap's flows, crediting the DEX VCB for the inflows.
+        let flow = (swap.body.delta_1_i, swap.body.delta_2_i);
         state
-            .put_swap_flow(&swap.body.trading_pair, swap_flow)
+            .accumulate_swap_flow(&swap.body.trading_pair, flow.into())
             .await?;
 
         // Record the swap commitment in the state.

crates/core/component/dex/src/component/chandelier.rs

@@ -328,10 +328,11 @@ mod tests {
         swap_flow.0 += 0u32.into();
         swap_flow.1 += gn.value(1u32.into()).amount;
 
-        // Set the batch swap flow for the trading pair.
+        // Accumulate it into the batch swap flow for the trading pair.
+        // Since this is currently empty this is the same as setting it.
         Arc::get_mut(&mut state)
             .unwrap()
-            .put_swap_flow(&trading_pair, swap_flow.clone())
+            .accumulate_swap_flow(&trading_pair, swap_flow.clone())
             .await
             .unwrap();
 
@@ -420,10 +421,11 @@ mod tests {
         // Swap 2 gn into penumbra, meaning each position is filled.
         swap_flow.1 += gn.value(2u32.into()).amount;
 
-        // Set the batch swap flow for the trading pair.
+        // Accumulate it into the batch swap flow for the trading pair.
+        // Since this is currently empty this is the same as setting it.
         Arc::get_mut(&mut state)
             .unwrap()
-            .put_swap_flow(&trading_pair, swap_flow.clone())
+            .accumulate_swap_flow(&trading_pair, swap_flow.clone())
             .await
             .unwrap();
 

crates/core/component/dex/src/component/circuit_breaker/mod.rs

@@ -3,3 +3,4 @@ pub(crate) mod value;
 
 pub(crate) use execution::ExecutionCircuitBreaker;
 pub(crate) use value::ValueCircuitBreaker;
+pub use value::ValueCircuitBreakerRead;

crates/core/component/dex/src/component/circuit_breaker/value.rs

@@ -1,13 +1,23 @@
 use anyhow::{anyhow, Result};
-use cnidarium::StateWrite;
-use penumbra_asset::Value;
+use cnidarium::{StateRead, StateWrite};
+use penumbra_asset::{asset, Value};
 use penumbra_num::Amount;
 use penumbra_proto::{StateReadProto, StateWriteProto};
 use tonic::async_trait;
 use tracing::instrument;
 
 use crate::{event, state_key};
 
+#[async_trait]
+pub trait ValueCircuitBreakerRead: StateRead {
+    /// Fetch the DEX VCB balance for a specified asset id.
+    async fn get_dex_vcb_for_asset(&self, id: &asset::Id) -> Result<Option<Amount>> {
+        Ok(self.get(&state_key::value_balance(&id)).await?)
+    }
+}
+
+impl<T: StateRead + ?Sized> ValueCircuitBreakerRead for T {}
+
 /// Tracks the aggregate value of deposits in the DEX.
 #[async_trait]
 pub(crate) trait ValueCircuitBreaker: StateWrite {
@@ -19,7 +29,7 @@ pub(crate) trait ValueCircuitBreaker: StateWrite {
         }
 
         let prev_balance: Amount = self
-            .get(&state_key::value_balance(&value.asset_id))
+            .get_dex_vcb_for_asset(&value.asset_id)
             .await?
             .unwrap_or_default();
         let new_balance = prev_balance
@@ -41,7 +51,7 @@ pub(crate) trait ValueCircuitBreaker: StateWrite {
         }
 
         let prev_balance: Amount = self
-            .get(&state_key::value_balance(&value.asset_id))
+            .get_dex_vcb_for_asset(&value.asset_id)
             .await?
             .unwrap_or_default();
         let new_balance = prev_balance
@@ -257,7 +267,7 @@ mod tests {
 
         // Set the batch swap flow for the trading pair.
         state_tx
-            .put_swap_flow(&trading_pair, swap_flow.clone())
+            .accumulate_swap_flow(&trading_pair, swap_flow.clone())
             .await
             .unwrap();
         state_tx.apply();