Extend arbitrage routing to route against recently opened positions a…

…nd recently swapped assets

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

@@ -7,7 +7,9 @@ use penumbra_proto::StateWriteProto;
 use penumbra_sct::component::source::SourceContext;
 
 use crate::{
-    component::{metrics, StateReadExt, StateWriteExt, SwapManager},
+    component::{
+        metrics, position_manager::PositionManager as _, StateReadExt, StateWriteExt, SwapManager,
+    },
     event,
     swap::{proof::SwapProofPublic, Swap},
 };
@@ -65,6 +67,10 @@ impl ActionHandler for Swap {
             .add_swap_payload(self.body.payload.clone(), source)
             .await;
 
+        // Mark the assets for the swap's trading pair as accessed during this block.
+        state.add_recently_accessed_asset(swap.body.trading_pair.asset_1());
+        state.add_recently_accessed_asset(swap.body.trading_pair.asset_2());
+
         metrics::histogram!(crate::component::metrics::DEX_SWAP_DURATION)
             .record(swap_start.elapsed());
         state.record_proto(event::swap(self));

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

@@ -17,7 +17,7 @@ use crate::{
 
 use super::{
     router::{HandleBatchSwaps, RoutingParams},
-    Arbitrage, PositionManager, ValueCircuitBreaker,
+    Arbitrage, PositionManager, PositionRead as _, ValueCircuitBreaker,
 };
 
 pub struct Dex {}
@@ -83,12 +83,22 @@ impl Component for Dex {
         // 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
+        // Extend the fixed candidate set to include recently accessed assets, to have
+        // more arbitrage execution against newly opened positions.
+        let fixed_candidates = Arc::new(
+            routing_params
+                .fixed_candidates
+                .iter()
+                .cloned()
+                // Limit the inclusion of recently accessed assets to 10 to avoid
+                // potentially blowing up routing time.
+                .chain(state.recently_accessed_assets().iter().take(10).cloned())
+                .collect::<Vec<_>>(),
+        );
+
         let arb_routing_params = RoutingParams {
             max_hops: routing_params.max_hops + 2,
-            fixed_candidates: routing_params.fixed_candidates.clone(),
+            fixed_candidates,
             price_limit: Some(1u64.into()),
         };
 

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

@@ -142,6 +142,12 @@ pub trait PositionRead: StateRead {
             })
             .boxed()
     }
+
+    /// Fetch the list of assets interacted with during this block.
+    fn recently_accessed_assets(&self) -> im::Vector<asset::Id> {
+        self.object_get(state_key::recently_accessed_assets())
+            .unwrap_or_default()
+    }
 }
 impl<T: StateRead + ?Sized> PositionRead for T {}
 
@@ -244,13 +250,30 @@ pub trait PositionManager: StateWrite + PositionRead {
         self.vcb_credit(position.reserves_1()).await?;
         self.vcb_credit(position.reserves_2()).await?;
 
+        // Add the asset IDs from the new position's trading pair
+        // to the candidate set for this block.
+        self.add_recently_accessed_asset(position.phi.pair.asset_1());
+        self.add_recently_accessed_asset(position.phi.pair.asset_2());
+
         // Finally, record the new position state.
         self.record_proto(event::position_open(&position));
         self.update_position(None, position).await?;
 
         Ok(())
     }
 
+    /// Adds an asset ID to the list of recently accessed assets,
+    /// making it a candidate for the current block's arbitrage routing.
+    ///
+    /// This ensures that assets associated with recently active positions
+    /// will be eligible for arbitrage if mispriced positions are opened.
+    #[tracing::instrument(level = "debug", skip_all)]
+    fn add_recently_accessed_asset(&mut self, asset_id: asset::Id) {
+        let mut assets = self.recently_accessed_assets();
+        assets.push_back(asset_id);
+        self.object_put(state_key::recently_accessed_assets(), assets);
+    }
+
     /// Record execution against an opened position.
     ///
     /// The `context` parameter records the global context of the path in which

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

@@ -67,6 +67,10 @@ pub fn pending_position_closures() -> &'static str {
     "dex/pending_position_closures"
 }
 
+pub fn recently_accessed_assets() -> &'static str {
+    "dex/recently_accessed_assets"
+}
+
 pub fn pending_payloads() -> &'static str {
     "dex/pending_payloads"
 }