auction(vcb): add a value circuit breaker (#4363)

## Describe your changes

This PR adds a value circuit breaker, modeled after the dex component's.

## Issue ticket number and link

#4356 

## 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:

> This is very consensus breaking, but not state breaking because this
state didn't exist and the auction is not previously deployed so there
are no inflows to track.

crates/core/component/auction/src/component/action_handler/dutch/schedule.rs

@@ -126,7 +126,7 @@ impl ActionHandler for ActionDutchAuctionSchedule {
             "the supplied auction id is already known to the chain (id={id})"
         );
 
-        state.schedule_auction(schedule.description.clone()).await;
+        state.schedule_auction(schedule.description.clone()).await?;
         Ok(())
     }
 }

crates/core/component/auction/src/component/action_handler/dutch/withdraw.rs

@@ -49,7 +49,7 @@ impl ActionHandler for ActionDutchAuctionWithdraw {
 
         // Execute the withdrawal, zero-ing out the auction state
         // and increasing its sequence number.
-        let withdrawn_balance = state.withdraw_auction(auction_state);
+        let withdrawn_balance = state.withdraw_auction(auction_state).await?;
 
         // Check that the reported balance commitment, match the recorded reserves.
         let expected_reserve_commitment = withdrawn_balance.commit(Fr::zero());

crates/core/component/auction/src/component/auction.rs

@@ -1,8 +1,12 @@
 use crate::component::dutch_auction::HandleDutchTriggers;
+use crate::event;
 use anyhow::Result;
 use async_trait::async_trait;
 use cnidarium::{StateRead, StateWrite};
 use cnidarium_component::Component;
+use penumbra_asset::asset;
+use penumbra_asset::Value;
+use penumbra_num::Amount;
 use penumbra_proto::StateReadProto;
 use penumbra_proto::StateWriteProto;
 use std::sync::Arc;
@@ -15,8 +19,6 @@ pub struct Auction {}
 
 #[async_trait]
 impl Component for Auction {
-    // Note: this is currently empty, but will make future
-    // addition easy to do.
     type AppState = crate::genesis::Content;
 
     #[instrument(name = "auction", skip(state, app_state))]
@@ -81,5 +83,106 @@ pub trait StateWriteExt: StateWrite {
 
 impl<T: StateWrite + ?Sized> StateWriteExt for T {}
 
+/// Internal trait implementing value flow tracking.
+/// # Overview
+///
+///                                                               
+///                                                  ║            
+///                                                  ║            
+///                                           User initiated      
+///      Auction                                     ║            
+///     component                                    ║            
+///   ┏━━━━━━━━━━━┓                                  ▼            
+///   ┃┌─────────┐┃                     ╔════════════════════════╗
+///   ┃│    A    │┃◀━━value in━━━━━━━━━━║   Schedule auction A   ║
+///   ┃└─────────┘┃                     ╚════════════════════════╝
+///   ┃           ┃                                               
+///   ┃     │     ┃                                               
+///   ┃           ┃                     ■■■■■■■■■■■■■■■■■■■■■■■■■■
+///   ┃  closed   ┃━━━value out by ━━━━▶■■■■■■Dex□black□box■■■■■■■
+///   ┃   then    ┃   creating lp       ■■■■■■■■■■■■■■■■■■■■■■■■■■
+///   ┃withdrawn  ┃                                  ┃            
+///   ┃           ┃                                  ┃            
+///   ┃     │     ┃                value in by       ┃            
+///   ┃           ┃◀━━━━━━━━━━━━━━━withdrawing lp━━━━┛            
+///   ┃     │     ┃                                               
+///   ┃     ▼     ┃                                               
+///   ┃┌ ─ ─ ─ ─ ┐┃                     ╔════════════════════════╗
+///   ┃     A     ┃━━━value out━━━━━━━━▶║   Withdraw auction A   ║
+///   ┃└ ─ ─ ─ ─ ┘┃                     ╚════════════════════════╝
+///   ┗━━━━━━━━━━━┛                                  ▲            
+///                                                  ║            
+///                                                  ║            
+///                                           User initiated      
+///                                              withdrawl        
+///                                                  ║            
+///                                                  ║            
+///                                                  ║            
+///
+pub(crate) trait AuctionCircuitBreaker: StateWrite {
+    /// Fetch the current balance of the auction circuit breaker for a given asset,
+    /// returning zero if no balance is tracked yet.
+    async fn get_auction_value_balance_for(&self, asset_id: &asset::Id) -> Amount {
+        self.get(&state_key::value_balance::for_asset(asset_id))
+            .await
+            .expect("failed to fetch auction value breaker balance")
+            .unwrap_or_else(Amount::zero)
+    }
+
+    /// Credit a deposit into the auction component.
+    async fn auction_vcb_credit(&mut self, value: Value) -> Result<()> {
+        let prev_balance = self.get_auction_value_balance_for(&value.asset_id).await;
+        let new_balance = prev_balance.checked_add(&value.amount).ok_or_else(|| {
+            tracing::error!(
+                ?prev_balance,
+                ?value,
+                "overflowed balance while crediting auction circuit breaker"
+            );
+            anyhow::anyhow!("overflowed balance while crediting auction circuit breaker")
+        })?;
+
+        // Write the new balance to the chain state.
+        self.put(
+            state_key::value_balance::for_asset(&value.asset_id),
+            new_balance,
+        );
+        // And emit an event to trace the value flow.
+        self.record_proto(event::auction_vcb_credit(
+            value.asset_id,
+            prev_balance,
+            new_balance,
+        ));
+        Ok(())
+    }
+
+    /// Debit a balance from the auction component.
+    async fn auction_vcb_debit(&mut self, value: Value) -> Result<()> {
+        let prev_balance = self.get_auction_value_balance_for(&value.asset_id).await;
+        let new_balance = prev_balance.checked_sub(&value.amount).ok_or_else(|| {
+            tracing::error!(
+                ?prev_balance,
+                ?value,
+                "underflowed balance while debiting auction circuit breaker"
+            );
+            anyhow::anyhow!("underflowed balance while debiting auction circuit breaker")
+        })?;
+
+        // Write the new balance to the chain state.
+        self.put(
+            state_key::value_balance::for_asset(&value.asset_id),
+            new_balance,
+        );
+        // And emit an event to trace the value flow out of the component.
+        self.record_proto(event::auction_vcb_debit(
+            value.asset_id,
+            prev_balance,
+            new_balance,
+        ));
+        Ok(())
+    }
+}
+
+impl<T: StateWrite + ?Sized> AuctionCircuitBreaker for T {}
+
 #[cfg(tests)]
 mod tests {}

crates/core/component/auction/src/component/dutch_auction.rs

@@ -4,9 +4,10 @@ use std::pin::Pin;
 use crate::auction::dutch::{DutchAuction, DutchAuctionDescription, DutchAuctionState};
 use crate::auction::AuctionId;
 use crate::component::trigger_data::TriggerData;
+use crate::component::AuctionCircuitBreaker;
 use crate::component::AuctionStoreRead;
 use crate::{event, state_key};
-use anyhow::Result;
+use anyhow::{Context, Result};
 use async_trait::async_trait;
 use cnidarium::{StateRead, StateWrite};
 use futures::StreamExt;
@@ -25,7 +26,7 @@ use prost::{Message, Name};
 pub(crate) trait DutchAuctionManager: StateWrite {
     /// Schedule an auction for the specified [`DutchAuctionDescritpion`], initializing
     /// its state, and registering it for execution by the component.
-    async fn schedule_auction(&mut self, description: DutchAuctionDescription) {
+    async fn schedule_auction(&mut self, description: DutchAuctionDescription) -> Result<()> {
         let auction_id = description.id();
         let DutchAuctionDescription {
             input: _,
@@ -66,12 +67,17 @@ pub(crate) trait DutchAuctionManager: StateWrite {
             state,
         };
 
+        // Deposit into the component's value balance.
+        self.auction_vcb_credit(dutch_auction.description.input)
+            .await
+            .context("failed to schedule auction")?;
         // Set the triggger
         self.set_trigger_for_dutch_id(auction_id, next_trigger);
         // Write position to state
         self.write_dutch_auction_state(dutch_auction);
         // Emit an event
         self.record_proto(event::dutch_auction_schedule_event(auction_id, description));
+        Ok(())
     }
 
     /// Execute the [`DutchAuction`] associated with [`AuctionId`], ticking its
@@ -156,21 +162,52 @@ pub(crate) trait DutchAuctionManager: StateWrite {
             state: old_dutch_auction.state,
         };
 
-        // After consuming the LP, we reset the state, getting ready to either
-        // execute another session, or retire the auction.
+        // First, we reset the state (Lp/trigger tracking), transfer value from the dex
+        // and prepare to either: execute another session, or retire the auction altogether.
+
+        // 1. We untrack the old position.
         new_dutch_auction.state.current_position = None;
-        new_dutch_auction.state.input_reserves += lp_reserves
-            .provided()
-            .filter(|v| v.asset_id == input.asset_id)
-            .map(|v| v.amount)
-            .sum::<Amount>();
-        new_dutch_auction.state.output_reserves += lp_reserves
-            .provided()
-            .filter(|v| v.asset_id == output_id)
-            .map(|v| v.amount)
-            .sum::<Amount>();
+        // 2. We untrack the trigger.
         new_dutch_auction.state.next_trigger = None;
 
+        /* *********** value transfer *************** */
+        // Critically, we need to orchestrate a value transfer from the Dex (lp position)
+        // into the auction component. This is done in three steps:
+        // 1. Compute the LP inflow to the auction's input and output reserves
+        // 2. Credit the auction's value balance with the respective inflows.
+        // 3. Add the inflows to the auction's reserves.
+
+        // 1. We compute the inflow from the LP's reserves.
+        let lp_inflow_input_asset = Value {
+            asset_id: auction_input_id,
+            amount: lp_reserves
+                .provided()
+                .filter(|v| v.asset_id == auction_input_id)
+                .map(|v| v.amount)
+                .sum::<Amount>(),
+        };
+        let lp_inflow_output_asset = Value {
+            asset_id: auction_output_id,
+            amount: lp_reserves
+                .provided()
+                .filter(|v| v.asset_id == auction_output_id)
+                .map(|v| v.amount)
+                .sum::<Amount>(),
+        };
+
+        // 2. We credit the auction's value balance with the inflows.
+        self.auction_vcb_credit(lp_inflow_input_asset)
+            .await
+            .context("failed to absorb LP inflow of input asset into auction value balance")?;
+        self.auction_vcb_credit(lp_inflow_output_asset)
+            .await
+            .context("failed to absorb LP inflow of output asset into auction value balance")?;
+
+        // 3. We add the inflows to the auction's reserves.
+        new_dutch_auction.state.input_reserves += lp_inflow_input_asset.amount;
+        new_dutch_auction.state.output_reserves += lp_inflow_output_asset.amount;
+        /* ***************** end value transfer ************************** */
+
         // Compute the current step index, between 0 and `step_count`.
         let step_index = auction_trigger
             .compute_step_index(trigger_height)
@@ -316,21 +353,31 @@ pub(crate) trait DutchAuctionManager: StateWrite {
             .get_dutch_auction_by_id(id)
             .await?
             .ok_or_else(|| anyhow::anyhow!("auction not found"))?;
-        self.withdraw_auction(auction);
+        self.withdraw_auction(auction).await?;
         Ok(())
     }
 
-    fn withdraw_auction(&mut self, mut auction: DutchAuction) -> Balance {
-        let previous_input_reserves = Balance::from(Value {
+    async fn withdraw_auction(&mut self, mut auction: DutchAuction) -> Result<Balance> {
+        let previous_input_reserves = Value {
             amount: auction.state.input_reserves,
             asset_id: auction.description.input.asset_id,
-        });
-        let previous_output_reserves = Balance::from(Value {
+        };
+        let previous_output_reserves = Value {
             amount: auction.state.output_reserves,
             asset_id: auction.description.output_id,
-        });
+        };
+
+        // We debit the auction's value balance with the outflows, aborting
+        // if the balance underflows.
+        self.auction_vcb_debit(previous_input_reserves)
+            .await
+            .context("couldn't withdraw input reserves from auction")?;
+        self.auction_vcb_debit(previous_output_reserves)
+            .await
+            .context("couldn't withdraw output reserves from auction")?;
 
-        let withdraw_balance = previous_input_reserves + previous_output_reserves;
+        let withdraw_balance =
+            Balance::from(previous_input_reserves) + Balance::from(previous_output_reserves);
 
         auction.state.sequence = auction.state.sequence.saturating_add(1);
         auction.state.current_position = None;
@@ -339,7 +386,7 @@ pub(crate) trait DutchAuctionManager: StateWrite {
         auction.state.output_reserves = Amount::zero();
         self.write_dutch_auction_state(auction);
 
-        withdraw_balance
+        Ok(withdraw_balance)
     }
 }
 
@@ -450,6 +497,12 @@ trait Inner: StateWrite {
                 attempt_counter += 1;
                 continue;
             } else {
+                self.auction_vcb_debit(lp.reserves_1())
+                    .await
+                    .context("failed to debit vcb of r1 during position allocation")?;
+                self.auction_vcb_debit(lp.reserves_2())
+                    .await
+                    .context("failed to debit vcb of r2 during position allocation")?;
                 self.open_position(lp).await.expect("no state incoherence");
                 return Ok(position_id);
             }

crates/core/component/auction/src/component/mod.rs

@@ -7,6 +7,7 @@ pub mod rpc;
 mod trigger_data;
 
 pub use auction::Auction;
+pub(crate) use auction::AuctionCircuitBreaker;
 pub use auction::{StateReadExt, StateWriteExt};
 pub(crate) use auction_store::AuctionStoreRead;
 pub(crate) use dutch_auction::DutchAuctionManager;

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

@@ -1,5 +1,7 @@
 use crate::auction::dutch::{DutchAuctionDescription, DutchAuctionState};
 use crate::auction::AuctionId;
+use penumbra_asset::asset;
+use penumbra_num::Amount;
 use penumbra_proto::penumbra::core::component::auction::v1alpha1 as pb;
 
 /// Event for a Dutch auction that has been scheduled.
@@ -59,3 +61,29 @@ pub fn dutch_auction_exhausted(
         reason: pb::event_dutch_auction_ended::Reason::Filled as i32,
     }
 }
+
+// Event for value flowing *into* the auction component.
+pub fn auction_vcb_credit(
+    asset_id: asset::Id,
+    previous_balance: Amount,
+    new_balance: Amount,
+) -> pb::EventValueCircuitBreakerCredit {
+    pb::EventValueCircuitBreakerCredit {
+        asset_id: Some(asset_id.into()),
+        previous_balance: Some(previous_balance.into()),
+        new_balance: Some(new_balance.into()),
+    }
+}
+
+// Event for value flowing *out of* the auction component.
+pub fn auction_vcb_debit(
+    asset_id: asset::Id,
+    previous_balance: Amount,
+    new_balance: Amount,
+) -> pb::EventValueCircuitBreakerDebit {
+    pb::EventValueCircuitBreakerDebit {
+        asset_id: Some(asset_id.into()),
+        previous_balance: Some(previous_balance.into()),
+        new_balance: Some(new_balance.into()),
+    }
+}

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

@@ -8,6 +8,20 @@ pub mod parameters {
     }
 }
 
+pub(crate) mod value_balance {
+    use penumbra_asset::asset;
+
+    #[allow(dead_code)] // For some reason, this shows up as unused
+    pub(crate) fn prefix() -> &'static str {
+        "auction/value_breaker/"
+    }
+
+    #[allow(dead_code)] // For some reason, this shows up as unused
+    pub(crate) fn for_asset(asset_id: &asset::Id) -> String {
+        format!("{}{asset_id}", prefix())
+    }
+}
+
 pub mod auction_store {
     use crate::auction::id::AuctionId;