distributions: add distributions events (#5044)

## Describe your changes

add LQT pool size increase event to the distributions component, emitted
at the end of each block.

## Issue ticket number and link

references #5041

## Checklist before requesting a review

- [x] I have added guiding text to explain how a reviewer should test
these changes.

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

  > LQT branch

crates/core/component/distributions/src/component.rs

@@ -6,17 +6,18 @@ mod view;
 
 use std::sync::Arc;
 
+use crate::event;
+use crate::genesis;
 use anyhow::{Context, Result};
 use async_trait::async_trait;
 use cnidarium::StateWrite;
 use cnidarium_component::Component;
 use penumbra_sdk_num::Amount;
+use penumbra_sdk_proto::StateWriteProto;
 use penumbra_sdk_sct::{component::clock::EpochRead, epoch::Epoch};
 use tendermint::v0_37::abci;
 use tracing::instrument;
 
-use crate::genesis;
-
 pub struct Distributions {}
 
 #[async_trait]
@@ -40,11 +41,36 @@ impl Component for Distributions {
     ) {
     }
 
-    #[instrument(name = "distributions", skip(_state, _end_block))]
+    #[instrument(name = "distributions", skip(state))]
     async fn end_block<S: StateWrite + 'static>(
-        _state: &mut Arc<S>,
-        _end_block: &abci::request::EndBlock,
+        state: &mut Arc<S>,
+        end_block: &abci::request::EndBlock,
     ) {
+        let state = Arc::get_mut(state).expect("the state should not be shared");
+
+        let current_epoch = state.get_current_epoch().await.expect("msg");
+        let current_block_height = end_block
+            .height
+            .try_into()
+            .expect("block height should not be negative");
+
+        let new_issuance = state
+            .get_distributions_params()
+            .await
+            .expect("distribution parameters should be available")
+            .staking_issuance_per_block as u128;
+
+        let total_new_issuance = state
+            .compute_lqt_issuance_from_blocks(current_epoch, current_block_height)
+            .await
+            .expect("should be able to compute LQT issuance from block");
+
+        // Emit an event for LQT pool size increase at the end of the block.
+        state.record_proto(event::event_lqt_pool_size_increase(
+            current_epoch.index,
+            new_issuance.into(),
+            total_new_issuance,
+        ))
     }
 
     #[instrument(name = "distributions", skip(state))]
@@ -108,9 +134,12 @@ trait DistributionManager: StateWriteExt {
         Ok(self.set_staking_token_issuance_for_epoch(new_issuance))
     }
 
-    /// Computes total LQT reward issuance for the epoch.
-    async fn compute_new_lqt_issuance(&self, current_epoch: Epoch) -> Result<Amount> {
-        let current_block_height = self.get_block_height().await?;
+    /// Helper function that computes the LQT issuance for the current block height in the epoch.
+    async fn compute_lqt_issuance_from_blocks(
+        &self,
+        current_epoch: Epoch,
+        current_block_height: u64,
+    ) -> Result<Amount> {
         let epoch_length = current_block_height
             .checked_sub(current_epoch.start_height)
             .unwrap_or_else(|| panic!("epoch start height is greater than current block height (epoch_start={}, current_height={}", current_epoch.start_height, current_block_height));
@@ -138,17 +167,28 @@ trait DistributionManager: StateWriteExt {
         Ok(Amount::from(total_pool_size_for_epoch))
     }
 
+    /// Computes total LQT reward issuance for the epoch.
+    async fn compute_new_lqt_issuance(&self, current_epoch: Epoch) -> Result<Amount> {
+        let current_block_height = self.get_block_height().await?;
+
+        Ok(self
+            .compute_lqt_issuance_from_blocks(current_epoch, current_block_height)
+            .await?)
+    }
+
     /// Update the nonverifiable storage with the newly issued LQT rewards for the current epoch.
     async fn define_lqt_budget(&mut self) -> Result<()> {
         // Grab the ambient epoch index.
         let current_epoch = self.get_current_epoch().await?;
 
+        // New issuance for the current epoch.
         let new_issuance = self.compute_new_lqt_issuance(current_epoch).await?;
         tracing::debug!(
             ?new_issuance,
-            "computed new lqt reward issuance for epoch {}",
+            "computed new lqt reward issuance for current epoch {}",
             current_epoch.index
         );
+
         Ok(self.set_lqt_reward_issuance_for_epoch(current_epoch.index, new_issuance))
     }
 }

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

@@ -0,0 +1,15 @@
+use penumbra_sdk_num::Amount;
+use penumbra_sdk_proto::penumbra::core::component::distributions::v1 as pb;
+
+/// Event for when LQT pool size increases.
+pub fn event_lqt_pool_size_increase(
+    epoch: u64,
+    increase: Amount,
+    new_total: Amount,
+) -> pb::EventLqtPoolSizeIncrease {
+    pb::EventLqtPoolSizeIncrease {
+        epoch,
+        increase: Some(increase.into()),
+        new_total: Some(new_total.into()),
+    }
+}

crates/core/component/distributions/src/lib.rs

@@ -6,3 +6,4 @@ pub mod component;
 pub mod genesis;
 pub mod params;
 pub use params::DistributionsParameters;
+pub mod event;

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

@@ -105,6 +105,29 @@ impl ::prost::Name for LqtPoolSizeByEpochResponse {
         "/penumbra.core.component.distributions.v1.LqtPoolSizeByEpochResponse".into()
     }
 }
+/// Event emitted when the size of the LQT pool increases.
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
+pub struct EventLqtPoolSizeIncrease {
+    /// The epoch in which the pool size increase occurred.
+    #[prost(uint64, tag = "1")]
+    pub epoch: u64,
+    /// The amount by which the LQT pool size increased in the block.
+    #[prost(message, optional, tag = "2")]
+    pub increase: ::core::option::Option<super::super::super::num::v1::Amount>,
+    /// The new total size of the LQT pool after the increase in the block.
+    #[prost(message, optional, tag = "3")]
+    pub new_total: ::core::option::Option<super::super::super::num::v1::Amount>,
+}
+impl ::prost::Name for EventLqtPoolSizeIncrease {
+    const NAME: &'static str = "EventLqtPoolSizeIncrease";
+    const PACKAGE: &'static str = "penumbra.core.component.distributions.v1";
+    fn full_name() -> ::prost::alloc::string::String {
+        "penumbra.core.component.distributions.v1.EventLqtPoolSizeIncrease".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "/penumbra.core.component.distributions.v1.EventLqtPoolSizeIncrease".into()
+    }
+}
 /// Generated client implementations.
 #[cfg(feature = "rpc")]
 pub mod distributions_service_client {

crates/proto/src/gen/penumbra.core.component.distributions.v1.serde.rs

@@ -310,6 +310,140 @@ impl<'de> serde::Deserialize<'de> for DistributionsParameters {
         deserializer.deserialize_struct("penumbra.core.component.distributions.v1.DistributionsParameters", FIELDS, GeneratedVisitor)
     }
 }
+impl serde::Serialize for EventLqtPoolSizeIncrease {
+    #[allow(deprecated)]
+    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        use serde::ser::SerializeStruct;
+        let mut len = 0;
+        if self.epoch != 0 {
+            len += 1;
+        }
+        if self.increase.is_some() {
+            len += 1;
+        }
+        if self.new_total.is_some() {
+            len += 1;
+        }
+        let mut struct_ser = serializer.serialize_struct("penumbra.core.component.distributions.v1.EventLqtPoolSizeIncrease", len)?;
+        if self.epoch != 0 {
+            #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
+            struct_ser.serialize_field("epoch", ToString::to_string(&self.epoch).as_str())?;
+        }
+        if let Some(v) = self.increase.as_ref() {
+            struct_ser.serialize_field("increase", v)?;
+        }
+        if let Some(v) = self.new_total.as_ref() {
+            struct_ser.serialize_field("newTotal", v)?;
+        }
+        struct_ser.end()
+    }
+}
+impl<'de> serde::Deserialize<'de> for EventLqtPoolSizeIncrease {
+    #[allow(deprecated)]
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        const FIELDS: &[&str] = &[
+            "epoch",
+            "increase",
+            "new_total",
+            "newTotal",
+        ];
+
+        #[allow(clippy::enum_variant_names)]
+        enum GeneratedField {
+            Epoch,
+            Increase,
+            NewTotal,
+            __SkipField__,
+        }
+        impl<'de> serde::Deserialize<'de> for GeneratedField {
+            fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+            where
+                D: serde::Deserializer<'de>,
+            {
+                struct GeneratedVisitor;
+
+                impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+                    type Value = GeneratedField;
+
+                    fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                        write!(formatter, "expected one of: {:?}", &FIELDS)
+                    }
+
+                    #[allow(unused_variables)]
+                    fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+                    where
+                        E: serde::de::Error,
+                    {
+                        match value {
+                            "epoch" => Ok(GeneratedField::Epoch),
+                            "increase" => Ok(GeneratedField::Increase),
+                            "newTotal" | "new_total" => Ok(GeneratedField::NewTotal),
+                            _ => Ok(GeneratedField::__SkipField__),
+                        }
+                    }
+                }
+                deserializer.deserialize_identifier(GeneratedVisitor)
+            }
+        }
+        struct GeneratedVisitor;
+        impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+            type Value = EventLqtPoolSizeIncrease;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                formatter.write_str("struct penumbra.core.component.distributions.v1.EventLqtPoolSizeIncrease")
+            }
+
+            fn visit_map<V>(self, mut map_: V) -> std::result::Result<EventLqtPoolSizeIncrease, V::Error>
+                where
+                    V: serde::de::MapAccess<'de>,
+            {
+                let mut epoch__ = None;
+                let mut increase__ = None;
+                let mut new_total__ = None;
+                while let Some(k) = map_.next_key()? {
+                    match k {
+                        GeneratedField::Epoch => {
+                            if epoch__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("epoch"));
+                            }
+                            epoch__ = 
+                                Some(map_.next_value::<::pbjson::private::NumberDeserialize<_>>()?.0)
+                            ;
+                        }
+                        GeneratedField::Increase => {
+                            if increase__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("increase"));
+                            }
+                            increase__ = map_.next_value()?;
+                        }
+                        GeneratedField::NewTotal => {
+                            if new_total__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("newTotal"));
+                            }
+                            new_total__ = map_.next_value()?;
+                        }
+                        GeneratedField::__SkipField__ => {
+                            let _ = map_.next_value::<serde::de::IgnoredAny>()?;
+                        }
+                    }
+                }
+                Ok(EventLqtPoolSizeIncrease {
+                    epoch: epoch__.unwrap_or_default(),
+                    increase: increase__,
+                    new_total: new_total__,
+                })
+            }
+        }
+        deserializer.deserialize_struct("penumbra.core.component.distributions.v1.EventLqtPoolSizeIncrease", FIELDS, GeneratedVisitor)
+    }
+}
 impl serde::Serialize for GenesisContent {
     #[allow(deprecated)]
     fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>

proto/penumbra/penumbra/core/component/distributions/v1/distributions.proto

@@ -47,3 +47,13 @@ message LqtPoolSizeByEpochResponse {
   // The total LQT pool size for the given epoch.
   core.num.v1.Amount pool_size = 2;
 }
+
+// Event emitted when the size of the LQT pool increases.
+message EventLqtPoolSizeIncrease {
+  // The epoch in which the pool size increase occurred.
+  uint64 epoch = 1;
+  // The amount by which the LQT pool size increased in the block.
+  core.num.v1.Amount increase = 2;
+  // The new total size of the LQT pool after the increase in the block.
+  core.num.v1.Amount new_total = 3;
+}