dex: add DexParameters, remove RoutingParams::default()

crates/bin/pcli/src/command/tx/proposal.rs

@@ -106,6 +106,7 @@ impl ProposalKindCmd {
                     sct_params: None,
                     shielded_pool_params: None,
                     stake_params: None,
+                    dex_params: None,
                 }),
             },
             ProposalKindCmd::CommunityPoolSpend { transaction_plan } => {

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

@@ -9,7 +9,8 @@ use jmt::RootHash;
 use penumbra_community_pool::component::{CommunityPool, StateWriteExt as _};
 use penumbra_community_pool::StateReadExt as _;
 use penumbra_compact_block::component::CompactBlockManager;
-use penumbra_dex::component::Dex;
+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_funding::component::Funding;
@@ -117,7 +118,7 @@ impl App {
                 )
                 .await;
                 Ibc::init_chain(&mut state_tx, Some(&genesis.ibc_content)).await;
-                Dex::init_chain(&mut state_tx, Some(&())).await;
+                Dex::init_chain(&mut state_tx, Some(&genesis.dex_content)).await;
                 CommunityPool::init_chain(&mut state_tx, Some(&genesis.community_pool_content))
                     .await;
                 Governance::init_chain(&mut state_tx, Some(&genesis.governance_content)).await;
@@ -241,6 +242,9 @@ impl App {
             if let Some(stake_params) = app_params.new.stake_params {
                 state_tx.put_stake_params(stake_params);
             }
+            if let Some(dex_params) = app_params.new.dex_params {
+                state_tx.put_dex_params(dex_params);
+            }
         }
 
         // Run each of the begin block handlers for each component, in sequence:
@@ -681,6 +685,7 @@ pub trait StateReadExt: StateRead {
         let sct_params = self.get_sct_params().await?;
         let shielded_pool_params = self.get_shielded_pool_params().await?;
         let stake_params = self.get_stake_params().await?;
+        let dex_params = self.get_dex_params().await?;
 
         Ok(AppParameters {
             chain_id,
@@ -693,6 +698,7 @@ pub trait StateReadExt: StateRead {
             sct_params,
             shielded_pool_params,
             stake_params,
+            dex_params,
         })
     }
 

crates/core/app/src/params.rs

@@ -1,4 +1,5 @@
 use penumbra_community_pool::params::CommunityPoolParameters;
+use penumbra_dex::DexParameters;
 use penumbra_distributions::DistributionsParameters;
 use penumbra_fee::FeeParameters;
 use penumbra_funding::FundingParameters;
@@ -20,6 +21,7 @@ pub struct AppParameters {
     pub chain_id: String,
     pub community_pool_params: CommunityPoolParameters,
     pub distributions_params: DistributionsParameters,
+    pub dex_params: DexParameters,
     pub fee_params: FeeParameters,
     pub funding_params: FundingParameters,
     pub governance_params: GovernanceParameters,
@@ -75,6 +77,10 @@ impl TryFrom<pb::AppParameters> for AppParameters {
                 .stake_params
                 .ok_or_else(|| anyhow::anyhow!("proto response missing stake params"))?
                 .try_into()?,
+            dex_params: msg
+                .dex_params
+                .ok_or_else(|| anyhow::anyhow!("proto response missing dex params"))?
+                .try_into()?,
         })
     }
 }
@@ -92,6 +98,7 @@ impl From<AppParameters> for pb::AppParameters {
             sct_params: Some(params.sct_params.into()),
             shielded_pool_params: Some(params.shielded_pool_params.into()),
             stake_params: Some(params.stake_params.into()),
+            dex_params: Some(params.dex_params.into()),
         }
     }
 }

crates/core/app/src/params/change.rs

@@ -2,6 +2,7 @@ use std::fmt::Display;
 
 use anyhow::Result;
 use penumbra_community_pool::params::CommunityPoolParameters;
+use penumbra_dex::DexParameters;
 use penumbra_distributions::params::DistributionsParameters;
 use penumbra_fee::FeeParameters;
 use penumbra_funding::params::FundingParameters;
@@ -69,6 +70,12 @@ impl AppParameters {
                     min_validator_stake: _,
                     unbonding_delay: _,
                 },
+            dex_params:
+                DexParameters {
+                    is_enabled: _,
+                    fixed_candidates: _,
+                    max_hops: _,
+                },
             // IMPORTANT: Don't use `..` here! We want to ensure every single field is verified!
         } = self;
 
@@ -157,6 +164,12 @@ impl AppParameters {
                     min_validator_stake,
                     unbonding_delay,
                 },
+            dex_params:
+                DexParameters {
+                    is_enabled: _,
+                    fixed_candidates: _,
+                    max_hops: _,
+                },
             // IMPORTANT: Don't use `..` here! We want to ensure every single field is verified!
         } = self;
 
@@ -246,6 +259,7 @@ impl AppParameters {
             shielded_pool_params: Some(self.shielded_pool_params.clone()),
             sct_params: Some(self.sct_params.clone()),
             stake_params: Some(self.stake_params.clone()),
+            dex_params: Some(self.dex_params.clone()),
         }
     }
 
@@ -325,6 +339,11 @@ impl AppParameters {
                     .stake_params
                     .clone()
             }),
+            dex_params: new.dex_params.clone().unwrap_or_else(|| {
+                old.expect("old should be set if new has any None values")
+                    .dex_params
+                    .clone()
+            }),
         })
     }
 }

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

@@ -42,7 +42,7 @@ mod tests {
     use std::sync::Arc;
 
     use crate::component::position_manager::Inner as _;
-    use crate::component::router::{HandleBatchSwaps as _, RoutingParams};
+    use crate::component::router::HandleBatchSwaps as _;
     use crate::component::{StateReadExt as _, StateWriteExt as _};
     use crate::{
         component::{router::limit_buy, tests::TempStorageExt, PositionManager as _},
@@ -241,8 +241,9 @@ mod tests {
         state_tx.apply();
 
         // This call should panic due to the outflow of gn not being covered by the circuit breaker.
+        let routing_params = state.routing_params().await.unwrap();
         state
-            .handle_batch_swaps(trading_pair, swap_flow, 0, 0, RoutingParams::default())
+            .handle_batch_swaps(trading_pair, swap_flow, 0, 0, routing_params)
             .await
             .expect("unable to process batch swaps");
     }

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

@@ -19,32 +19,23 @@ use super::{
 pub trait Arbitrage: StateWrite + Sized {
     /// Attempts to extract as much as possible of the `arb_token` from the available
     /// liquidity positions, and returns the amount of `arb_token` extracted.
-    #[instrument(skip(self, arb_token, fixed_candidates))]
+    #[instrument(skip(self, arb_token, routing_params))]
     async fn arbitrage(
         self: &mut Arc<Self>,
         arb_token: asset::Id,
-        fixed_candidates: Vec<asset::Id>,
+        routing_params: RoutingParams,
     ) -> Result<Value>
     where
         Self: 'static,
     {
-        tracing::debug!(?arb_token, ?fixed_candidates, "beginning arb search");
+        tracing::debug!(?arb_token, ?routing_params, "beginning arb search");
         let arb_start = std::time::Instant::now();
 
         // Work in a new `StateDelta`, so we can transactionally apply any state
         // changes, and roll them back if we fail (e.g., if for some reason we
         // discover at the end that the arb wasn't profitable).
         let mut this = Arc::new(StateDelta::new(self.clone()));
 
-        // TODO: Build an extended candidate set with:
-        // - both ends of all trading pairs for which there were swaps in the block
-        // - both ends of all trading pairs for which positions were opened
-        let params = RoutingParams {
-            max_hops: 5,
-            price_limit: Some(1u64.into()),
-            fixed_candidates: Arc::new(fixed_candidates),
-        };
-
         // Create a flash-loan 2^64 of the arb token to ourselves.
         let flash_loan = Value {
             asset_id: arb_token,
@@ -57,7 +48,7 @@ pub trait Arbitrage: StateWrite + Sized {
                 arb_token,
                 arb_token,
                 flash_loan.amount,
-                params,
+                routing_params,
                 execution_circuit_breaker,
             )
             .await?;

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

@@ -12,8 +12,8 @@ use tendermint::v0_37::abci;
 use tracing::instrument;
 
 use crate::{
-    component::flow::SwapFlow, event, state_key, BatchSwapOutputData, DirectedTradingPair,
-    SwapExecution, TradingPair,
+    component::flow::SwapFlow, event, genesis, state_key, BatchSwapOutputData, DexParameters,
+    DirectedTradingPair, SwapExecution, TradingPair,
 };
 
 use super::{
@@ -25,10 +25,19 @@ pub struct Dex {}
 
 #[async_trait]
 impl Component for Dex {
-    type AppState = ();
+    type AppState = genesis::Content;
 
-    #[instrument(name = "dex", skip(_state, _app_state))]
-    async fn init_chain<S: StateWrite>(_state: S, _app_state: Option<&()>) {}
+    #[instrument(name = "dex", skip(state, app_state))]
+    async fn init_chain<S: StateWrite>(mut state: S, app_state: Option<&Self::AppState>) {
+        match app_state {
+            None => {
+                // Checkpoint -- no-op
+            }
+            Some(app_state) => {
+                state.put_dex_params(app_state.dex_params.clone());
+            }
+        }
+    }
 
     #[instrument(name = "dex", skip(_state, _begin_block))]
     async fn begin_block<S: StateWrite + 'static>(
@@ -42,9 +51,14 @@ impl Component for Dex {
         state: &mut Arc<S>,
         end_block: &abci::request::EndBlock,
     ) {
+        // 1. Add all newly opened positions to the DEX.
+        // This has already happened in the action handlers for each `PositionOpen` action.
+
+        // 2. For each batch swap during the block, calculate clearing prices and set in the JMT.
+
         let current_epoch = state.get_current_epoch().await.expect("epoch is set");
+        let routing_params = state.routing_params().await.expect("dex params are set");
 
-        // For each batch swap during the block, calculate clearing prices and set in the JMT.
         for (trading_pair, swap_flows) in state.swap_flows() {
             let batch_start = std::time::Instant::now();
             state
@@ -57,49 +71,32 @@ impl Component for Dex {
                         .expect("height is part of the end block data"),
                     current_epoch.start_height,
                     // Always include both ends of the target pair as fixed candidates.
-                    RoutingParams::default_with_extra_candidates([
-                        trading_pair.asset_1(),
-                        trading_pair.asset_2(),
-                    ]),
+                    routing_params
+                        .clone()
+                        .with_extra_candidates([trading_pair.asset_1(), trading_pair.asset_2()]),
                 )
                 .await
                 .expect("handling batch swaps is infaillible");
             metrics::histogram!(crate::component::metrics::DEX_BATCH_DURATION)
                 .record(batch_start.elapsed());
         }
 
-        // Then, perform arbitrage:
+        // 3. Perform arbitrage to ensure all prices are consistent post-execution:
+
+        // For arbitrage, we extend the path search by 2 hops to allow a path out of the
+        // staking token and back.
+
+        // TODO: Build an extended candidate set with:
+        // - both ends of all trading pairs for which there were swaps in the block
+        // - both ends of all trading pairs for which positions were opened
+        let arb_routing_params = RoutingParams {
+            max_hops: routing_params.max_hops + 2,
+            fixed_candidates: routing_params.fixed_candidates.clone(),
+            price_limit: Some(1u64.into()),
+        };
+
         let arb_burn = match state
-            .arbitrage(
-                *STAKING_TOKEN_ASSET_ID,
-                vec![
-                    *STAKING_TOKEN_ASSET_ID,
-                    asset::Cache::with_known_assets()
-                        .get_unit("gm")
-                        .expect("gm is a known asset")
-                        .id(),
-                    asset::Cache::with_known_assets()
-                        .get_unit("gn")
-                        .expect("gn is a known asset")
-                        .id(),
-                    asset::Cache::with_known_assets()
-                        .get_unit("test_usd")
-                        .expect("test_usd is a known asset")
-                        .id(),
-                    asset::Cache::with_known_assets()
-                        .get_unit("test_btc")
-                        .expect("test_btc is a known asset")
-                        .id(),
-                    asset::Cache::with_known_assets()
-                        .get_unit("test_atom")
-                        .expect("test_atom is a known asset")
-                        .id(),
-                    asset::Cache::with_known_assets()
-                        .get_unit("test_osmo")
-                        .expect("test_osmo is a known asset")
-                        .id(),
-                ],
-            )
+            .arbitrage(*STAKING_TOKEN_ASSET_ID, arb_routing_params)
             .await
         {
             Ok(v) => v,
@@ -120,10 +117,11 @@ impl Component for Dex {
                 .get_unit("penumbra")
                 .expect("penumbra is a known asset");
             let burn = format!("{}{}", unit.format_value(arb_burn.amount), unit);
+            // TODO: this should be an ABCI event
             tracing::info!(%burn, "executed arbitrage opportunity");
         }
 
-        // Next, close all positions queued for closure at the end of the block.
+        // 4. Close all positions queued for closure at the end of the block.
         // It's important to do this after execution, to allow block-scoped JIT liquidity.
         Arc::get_mut(state)
             .expect("state should be uniquely referenced after batch swaps complete")
@@ -176,13 +174,41 @@ pub trait StateReadExt: StateRead {
         self.object_get(state_key::pending_outputs())
             .unwrap_or_default()
     }
+
+    /// Gets the DEX parameters from the state.
+    async fn get_dex_params(&self) -> Result<DexParameters> {
+        self.get(state_key::config::dex_params())
+            .await?
+            .ok_or_else(|| anyhow::anyhow!("Missing DexParameters"))
+    }
+
+    /// Indicates if the DEX parameters have been updated in this block.
+    fn dex_params_updated(&self) -> bool {
+        self.object_get::<()>(state_key::config::dex_params_updated())
+            .is_some()
+    }
+
+    /// Uses the DEX parameters to construct a `RoutingParams` for use in execution or simulation.
+    async fn routing_params(&self) -> Result<RoutingParams> {
+        let dex_params = self.get_dex_params().await?;
+        Ok(RoutingParams {
+            max_hops: dex_params.max_hops as usize,
+            fixed_candidates: Arc::new(dex_params.fixed_candidates),
+            price_limit: None,
+        })
+    }
 }
 
 impl<T: StateRead + ?Sized> StateReadExt for T {}
 
 /// Extension trait providing write access to dex data.
 #[async_trait]
 pub trait StateWriteExt: StateWrite + StateReadExt {
+    fn put_dex_params(&mut self, params: DexParameters) {
+        self.put(state_key::config::dex_params().to_string(), params);
+        self.object_put(state_key::config::dex_params_updated(), ())
+    }
+
     fn set_output_data(
         &mut self,
         output_data: BatchSwapOutputData,