dex: redesign value circuit breaker (#4086)

Closes #4025

This commit sketches a new mechanism but does not fix the existing
tests; it needs to be picked up and pushed over the finish line. (We
should be testing this).

---------

Co-authored-by: Chris Czub <[email protected]>

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

@@ -5,7 +5,7 @@ use cnidarium_component::ActionHandler;
 use penumbra_proto::StateWriteProto as _;
 
 use crate::{
-    component::{PositionManager, PositionRead},
+    component::{PositionManager, PositionRead, ValueCircuitBreaker},
     event,
     lp::{action::PositionOpen, position},
 };
@@ -33,6 +33,13 @@ impl ActionHandler for PositionOpen {
     async fn check_and_execute<S: StateWrite>(&self, mut state: S) -> Result<()> {
         // Validate that the position ID doesn't collide
         state.check_position_id_unused(&self.position.id()).await?;
+
+        // Credit the DEX for the inflows from this position.
+        // TODO: in a future PR, split current PositionManager to PositionManagerInner
+        // and fold this into a position open method
+        state.vcb_credit(self.position.reserves_1()).await?;
+        state.vcb_credit(self.position.reserves_2()).await?;
+
         state.put_position(self.position.clone()).await?;
         state.record_proto(event::position_open(self));
         Ok(())

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

@@ -7,7 +7,7 @@ use decaf377::Fr;
 use penumbra_proto::StateWriteProto;
 
 use crate::{
-    component::{PositionManager, PositionRead},
+    component::{PositionManager, PositionRead, ValueCircuitBreaker},
     event,
     lp::{action::PositionWithdraw, position, Reserves},
 };
@@ -90,6 +90,12 @@ impl ActionHandler for PositionWithdraw {
         // the current reserves.
         state.record_proto(event::position_withdraw(self, &metadata));
 
+        // Debit the DEX for the outflows from this position.
+        // TODO: in a future PR, split current PositionManager to PositionManagerInner
+        // and fold this into a position open method
+        state.vcb_debit(metadata.reserves_1()).await?;
+        state.vcb_debit(metadata.reserves_2()).await?;
+
         // Finally, update the position. This has two steps:
         // - update the state with the correct sequence number;
         // - zero out the reserves, to prevent double-withdrawals.

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

@@ -47,7 +47,9 @@ impl ActionHandler for Swap {
         swap_flow.1 += swap.body.delta_2_i;
 
         // Set the batch swap flow for the trading pair.
-        state.put_swap_flow(&swap.body.trading_pair, swap_flow);
+        state
+            .put_swap_flow(&swap.body.trading_pair, swap_flow)
+            .await?;
 
         // Record the swap commitment in the state.
         let source = state.get_current_source().expect("source is set");

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

@@ -8,7 +8,7 @@ use penumbra_proto::StateWriteProto as _;
 use penumbra_sct::component::clock::EpochRead;
 use tracing::instrument;
 
-use crate::{event, ExecutionCircuitBreaker, SwapExecution};
+use crate::{component::ExecutionCircuitBreaker, event, SwapExecution};
 
 use super::{
     router::{RouteAndFill, RoutingParams},

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

@@ -0,0 +1,5 @@
+mod execution;
+mod value;
+
+pub use execution::ExecutionCircuitBreaker;
+pub use value::ValueCircuitBreaker;