Implement indexing for DEX events

crates/core/component/dex/src/component/action_handler/position/reward_claim.rs

@@ -29,6 +29,7 @@ impl ActionHandler for PositionRewardClaim {
         // It's important to reject all LP actions for now, to prevent
         // inflation / minting bugs until we implement all required checks
         // (e.g., minting tokens by withdrawing reserves we don't check)
+        // TODO: add record_proto call to log event here
         Err(anyhow::anyhow!("lp rewards not supported yet"))
     }
 }

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

@@ -4,10 +4,11 @@ use anyhow::Result;
 use async_trait::async_trait;
 use cnidarium::{StateDelta, StateWrite};
 use penumbra_asset::{asset, Value};
+use penumbra_proto::StateWriteProto as _;
 use penumbra_sct::component::clock::EpochRead;
 use tracing::instrument;
 
-use crate::{ExecutionCircuitBreaker, SwapExecution};
+use crate::{event, ExecutionCircuitBreaker, SwapExecution};
 
 use super::{
     router::{RouteAndFill, RoutingParams},
@@ -115,20 +116,20 @@ pub trait Arbitrage: StateWrite + Sized {
 
         // Finally, record the arb execution in the state:
         let height = self_mut.get_block_height().await?;
-        self_mut.set_arb_execution(
-            height,
-            SwapExecution {
-                traces: swap_execution.traces,
-                input: Value {
-                    asset_id: arb_token,
-                    amount: filled_input,
-                },
-                output: Value {
-                    amount: arb_profit,
-                    asset_id: arb_token,
-                },
+        let se = SwapExecution {
+            traces: swap_execution.traces,
+            input: Value {
+                asset_id: arb_token,
+                amount: filled_input,
+            },
+            output: Value {
+                amount: arb_profit,
+                asset_id: arb_token,
             },
-        );
+        };
+        self_mut.set_arb_execution(height, se.clone());
+        // Emit an ABCI event detailing the arb execution.
+        self_mut.record_proto(event::arb_execution(height, se));
         metrics::histogram!(crate::component::metrics::DEX_ARB_DURATION)
             .record(arb_start.elapsed());
         return Ok(Value {

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

@@ -11,8 +11,8 @@ use tendermint::v0_37::abci;
 use tracing::instrument;
 
 use crate::{
-    component::flow::SwapFlow, state_key, BatchSwapOutputData, DirectedTradingPair, SwapExecution,
-    TradingPair,
+    component::flow::SwapFlow, event, state_key, BatchSwapOutputData, DirectedTradingPair,
+    SwapExecution, TradingPair,
 };
 
 use super::{
@@ -183,14 +183,14 @@ pub trait StateWriteExt: StateWrite + StateReadExt {
         self.put(state_key::output_data(height, trading_pair), output_data);
 
         // Store the swap executions for both directions in the state as well.
-        if let Some(swap_execution) = swap_execution_1_for_2 {
+        if let Some(swap_execution) = swap_execution_1_for_2.clone() {
             let tp_1_for_2 = DirectedTradingPair::new(trading_pair.asset_1, trading_pair.asset_2);
             self.put(
                 state_key::swap_execution(height, tp_1_for_2),
                 swap_execution,
             );
         }
-        if let Some(swap_execution) = swap_execution_2_for_1 {
+        if let Some(swap_execution) = swap_execution_2_for_1.clone() {
             let tp_2_for_1 = DirectedTradingPair::new(trading_pair.asset_2, trading_pair.asset_1);
             self.put(
                 state_key::swap_execution(height, tp_2_for_1),
@@ -202,6 +202,13 @@ pub trait StateWriteExt: StateWrite + StateReadExt {
         let mut outputs = self.pending_batch_swap_outputs();
         outputs.insert(trading_pair, output_data);
         self.object_put(state_key::pending_outputs(), outputs);
+
+        // Also generate an ABCI event for indexing:
+        self.record_proto(event::batch_swap(
+            output_data,
+            swap_execution_1_for_2,
+            swap_execution_2_for_1,
+        ));
     }
 
     fn set_arb_execution(&mut self, height: u64, execution: SwapExecution) {

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

@@ -12,10 +12,12 @@ use penumbra_num::{
     fixpoint::{Error, U128x128},
     Amount,
 };
+use penumbra_proto::StateWriteProto as _;
 use tracing::instrument;
 
 use crate::{
     component::{metrics, PositionManager, PositionRead},
+    event,
     lp::{
         position::{self, Position},
         Reserves,
@@ -402,6 +404,10 @@ impl<S: StateRead + StateWrite> Frontier<S> {
     async fn save(&mut self) -> Result<()> {
         for position in &self.positions {
             self.state.put_position(position.clone()).await?;
+
+            // Create an ABCI event signaling that the position was executed against
+            self.state
+                .record_proto(event::position_execution(position.clone()));
         }
         Ok(())
     }

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

@@ -5,6 +5,7 @@ use crate::{
     },
     swap::Swap,
     swap_claim::SwapClaim,
+    BatchSwapOutputData, SwapExecution,
 };
 
 use penumbra_proto::penumbra::core::component::dex::v1 as pb;
@@ -33,7 +34,7 @@ pub fn position_open(position_open: &PositionOpen) -> pb::EventPositionOpen {
         trading_pair: Some(position_open.position.phi.pair.into()),
         reserves_1: Some(position_open.position.reserves.r1.into()),
         reserves_2: Some(position_open.position.reserves.r2.into()),
-        trading_fee: position_open.position.phi.component.fee.into(),
+        trading_fee: position_open.position.phi.component.fee,
     }
 }
 
@@ -56,3 +57,31 @@ pub fn position_withdraw(
         reserves_2: Some(final_position_state.reserves.r2.into()),
     }
 }
+
+pub fn position_execution(post_execution_state: Position) -> pb::EventPositionExecution {
+    pb::EventPositionExecution {
+        position_id: Some(post_execution_state.id().into()),
+        trading_pair: Some(post_execution_state.phi.pair.into()),
+        reserves_1: Some(post_execution_state.reserves.r1.into()),
+        reserves_2: Some(post_execution_state.reserves.r2.into()),
+    }
+}
+
+pub fn batch_swap(
+    bsod: BatchSwapOutputData,
+    swap_execution_1_for_2: Option<SwapExecution>,
+    swap_execution_2_for_1: Option<SwapExecution>,
+) -> pb::EventBatchSwap {
+    pb::EventBatchSwap {
+        batch_swap_output_data: Some(bsod.into()),
+        swap_execution_1_for_2: swap_execution_1_for_2.map(Into::into),
+        swap_execution_2_for_1: swap_execution_2_for_1.map(Into::into),
+    }
+}
+
+pub fn arb_execution(height: u64, swap_execution: SwapExecution) -> pb::EventArbExecution {
+    pb::EventArbExecution {
+        height,
+        swap_execution: Some(swap_execution.into()),
+    }
+}

crates/proto/src/gen/penumbra.core.component.dex.v1.rs

@@ -1312,6 +1312,66 @@ impl ::prost::Name for EventPositionWithdraw {
         ::prost::alloc::format!("penumbra.core.component.dex.v1.{}", Self::NAME)
     }
 }
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct EventPositionExecution {
+    /// The ID of the position executed against.
+    #[prost(message, optional, tag = "1")]
+    pub position_id: ::core::option::Option<PositionId>,
+    /// The trading pair of the position executed against.
+    #[prost(message, optional, tag = "2")]
+    pub trading_pair: ::core::option::Option<TradingPair>,
+    /// The reserves of asset 1 of the position after execution.
+    #[prost(message, optional, tag = "3")]
+    pub reserves_1: ::core::option::Option<super::super::super::num::v1::Amount>,
+    /// The reserves of asset 2 of the position after execution.
+    #[prost(message, optional, tag = "4")]
+    pub reserves_2: ::core::option::Option<super::super::super::num::v1::Amount>,
+}
+impl ::prost::Name for EventPositionExecution {
+    const NAME: &'static str = "EventPositionExecution";
+    const PACKAGE: &'static str = "penumbra.core.component.dex.v1";
+    fn full_name() -> ::prost::alloc::string::String {
+        ::prost::alloc::format!("penumbra.core.component.dex.v1.{}", Self::NAME)
+    }
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct EventBatchSwap {
+    /// The BatchSwapOutputData containing the results of the batch swap.
+    #[prost(message, optional, tag = "1")]
+    pub batch_swap_output_data: ::core::option::Option<BatchSwapOutputData>,
+    /// The record of execution for the batch swap in the 1 -> 2 direction.
+    #[prost(message, optional, tag = "2")]
+    pub swap_execution_1_for_2: ::core::option::Option<SwapExecution>,
+    /// The record of execution for the batch swap in the 2 -> 1 direction.
+    #[prost(message, optional, tag = "3")]
+    pub swap_execution_2_for_1: ::core::option::Option<SwapExecution>,
+}
+impl ::prost::Name for EventBatchSwap {
+    const NAME: &'static str = "EventBatchSwap";
+    const PACKAGE: &'static str = "penumbra.core.component.dex.v1";
+    fn full_name() -> ::prost::alloc::string::String {
+        ::prost::alloc::format!("penumbra.core.component.dex.v1.{}", Self::NAME)
+    }
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct EventArbExecution {
+    /// The height at which the arb execution occurred.
+    #[prost(uint64, tag = "1")]
+    pub height: u64,
+    /// The record of execution for the arb execution.
+    #[prost(message, optional, tag = "2")]
+    pub swap_execution: ::core::option::Option<SwapExecution>,
+}
+impl ::prost::Name for EventArbExecution {
+    const NAME: &'static str = "EventArbExecution";
+    const PACKAGE: &'static str = "penumbra.core.component.dex.v1";
+    fn full_name() -> ::prost::alloc::string::String {
+        ::prost::alloc::format!("penumbra.core.component.dex.v1.{}", Self::NAME)
+    }
+}
 /// Generated client implementations.
 #[cfg(feature = "rpc")]
 pub mod query_service_client {