distributions: implement rpc services (#5040)

## Describe your changes

- Implements rpc services for the distributions component in
`DistributionService`
- indexes distributions state key by epoch index and writes to NV
storage

@cronokirby had some suggestions (which can be discussed in this PR) for
replacing the historical epoch RPC, so I'm punting on the impl for now.

## Issue ticket number and link

references #5039

## 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/Cargo.toml

@@ -13,6 +13,7 @@ component = [
     "cnidarium",
     "penumbra-sdk-proto/cnidarium",
     "penumbra-sdk-sct/component",
+    "tonic",
 ]
 default = ["component"]
 docsrs = []
@@ -29,6 +30,7 @@ penumbra-sdk-sct = {workspace = true, default-features = false}
 serde = {workspace = true, features = ["derive"]}
 tendermint = {workspace = true}
 tracing = {workspace = true}
+tonic = {workspace = true, optional = true}
 
 [dev-dependencies]
 getrandom = {workspace = true}

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

@@ -1,5 +1,6 @@
 pub mod state_key;
 pub use view::{StateReadExt, StateWriteExt};
+pub mod rpc;
 
 mod view;
 
@@ -10,6 +11,7 @@ use async_trait::async_trait;
 use cnidarium::StateWrite;
 use cnidarium_component::Component;
 use penumbra_sdk_num::Amount;
+use penumbra_sdk_sct::{component::clock::EpochRead, epoch::Epoch};
 use tendermint::v0_37::abci;
 use tracing::instrument;
 
@@ -99,16 +101,16 @@ trait DistributionManager: StateWriteExt {
     /// Update the object store with the new issuance of staking tokens for this epoch.
     async fn define_staking_budget(&mut self) -> Result<()> {
         let new_issuance = self.compute_new_staking_issuance().await?;
-        tracing::debug!(?new_issuance, "computed new staking issuance for epoch");
+        tracing::debug!(
+            ?new_issuance,
+            "computed new staking issuance for current epoch"
+        );
         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) -> Result<Amount> {
-        use penumbra_sdk_sct::component::clock::EpochRead;
-
+    async fn compute_new_lqt_issuance(&self, current_epoch: Epoch) -> Result<Amount> {
         let current_block_height = self.get_block_height().await?;
-        let current_epoch = self.get_current_epoch().await?;
         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,9 +140,16 @@ trait DistributionManager: StateWriteExt {
 
     /// Update the nonverifiable storage with the newly issued LQT rewards for the current epoch.
     async fn define_lqt_budget(&mut self) -> Result<()> {
-        let new_issuance = self.compute_new_lqt_issuance().await?;
-        tracing::debug!(?new_issuance, "computed new lqt reward issuance for epoch");
-        Ok(self.set_lqt_reward_issuance_for_epoch(new_issuance))
+        // Grab the ambient epoch index.
+        let current_epoch = self.get_current_epoch().await?;
+
+        let new_issuance = self.compute_new_lqt_issuance(current_epoch).await?;
+        tracing::debug!(
+            ?new_issuance,
+            "computed new lqt reward issuance for epoch {}",
+            current_epoch.index
+        );
+        Ok(self.set_lqt_reward_issuance_for_epoch(current_epoch.index, new_issuance))
     }
 }
 

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

@@ -0,0 +1,80 @@
+use cnidarium::Storage;
+use penumbra_sdk_num::Amount;
+use penumbra_sdk_proto::core::component::distributions::v1::{
+    self as pb, distributions_service_server::DistributionsService,
+};
+use penumbra_sdk_sct::component::clock::EpochRead;
+
+use crate::component::StateReadExt;
+
+pub struct Server {
+    storage: Storage,
+}
+
+impl Server {
+    pub fn new(storage: Storage) -> Self {
+        Self { storage }
+    }
+}
+
+#[tonic::async_trait]
+impl DistributionsService for Server {
+    async fn current_lqt_pool_size(
+        &self,
+        _request: tonic::Request<pb::CurrentLqtPoolSizeRequest>,
+    ) -> Result<tonic::Response<pb::CurrentLqtPoolSizeResponse>, tonic::Status> {
+        // Retrieve latest state snapshot.
+        let state = self.storage.latest_snapshot();
+
+        let current_block_height = state.get_block_height().await.map_err(|e| {
+            tonic::Status::internal(format!("failed to get current block height: {}", e))
+        })?;
+        let current_epoch = state
+            .get_current_epoch()
+            .await
+            .map_err(|e| tonic::Status::internal(format!("failed to get current epoch: {}", e)))?;
+        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));
+
+        let lqt_block_reward_rate = state
+            .get_distributions_params()
+            .await
+            .map_err(|e| {
+                tonic::Status::internal(format!("failed to get distributions parameters: {}", e))
+            })?
+            .liquidity_tournament_incentive_per_block as u64;
+
+        let current_lqt_pool_size = lqt_block_reward_rate
+            .checked_mul(epoch_length as u64)
+            .expect("infallible unless issuance is pathological");
+
+        Ok(tonic::Response::new(pb::CurrentLqtPoolSizeResponse {
+            epoch_index: current_epoch.index,
+            pool_size: Some(Amount::from(current_lqt_pool_size).into()),
+        }))
+    }
+
+    async fn lqt_pool_size_by_epoch(
+        &self,
+        request: tonic::Request<pb::LqtPoolSizeByEpochRequest>,
+    ) -> Result<tonic::Response<pb::LqtPoolSizeByEpochResponse>, tonic::Status> {
+        // Retrieve latest state snapshot.
+        let state = self.storage.latest_snapshot();
+        let epoch_index = request.into_inner().epoch;
+        let amount = state
+            .get_lqt_reward_issuance_for_epoch(epoch_index)
+            .await
+            .ok_or_else(|| {
+                tonic::Status::not_found(format!(
+                    "failed to retrieve LQT issuance for epoch {} from non-verifiable storage",
+                    epoch_index,
+                ))
+            })?;
+
+        Ok(tonic::Response::new(pb::LqtPoolSizeByEpochResponse {
+            epoch_index,
+            pool_size: Some(amount.into()),
+        }))
+    }
+}

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

@@ -3,9 +3,22 @@ pub fn staking_token_issuance_for_epoch() -> &'static str {
     "distributions/staking_token_issuance_for_epoch"
 }
 
-// The amount of LQT rewards issued for this epoch.
-pub fn lqt_reward_issuance_for_epoch() -> &'static str {
-    "distributions/lqt_reward_issuance_for_epoch"
+pub mod lqt {
+    pub mod v1 {
+        pub mod budget {
+            pub(crate) fn prefix(epoch_index: u64) -> String {
+                format!("distributions/lqt/v1/budget/{epoch_index:020}")
+            }
+
+            /// The amount of LQT rewards issued for this epoch.
+            pub fn for_epoch(epoch_index: u64) -> [u8; 48] {
+                let prefix_bytes = prefix(epoch_index);
+                let mut key = [0u8; 48];
+                key[0..48].copy_from_slice(prefix_bytes.as_bytes());
+                key
+            }
+        }
+    }
 }
 
 pub fn distributions_parameters() -> &'static str {

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

@@ -21,8 +21,13 @@ pub trait StateReadExt: StateRead {
     }
 
     // Get the total amount of LQT rewards issued for this epoch.
-    fn get_lqt_reward_issuance_for_epoch(&self) -> Option<Amount> {
-        self.object_get(&state_key::lqt_reward_issuance_for_epoch())
+    async fn get_lqt_reward_issuance_for_epoch(&self, epoch_index: u64) -> Option<Amount> {
+        let key = state_key::lqt::v1::budget::for_epoch(epoch_index);
+
+        self.nonverifiable_get(&key).await.unwrap_or_else(|_| {
+            tracing::error!("LQT issuance does not exist for epoch");
+            None
+        })
     }
 }
 
@@ -41,8 +46,11 @@ pub trait StateWriteExt: StateWrite + StateReadExt {
     }
 
     /// Set the total amount of LQT rewards issued for this epoch.
-    fn set_lqt_reward_issuance_for_epoch(&mut self, issuance: Amount) {
-        self.object_put(state_key::lqt_reward_issuance_for_epoch(), issuance);
+    fn set_lqt_reward_issuance_for_epoch(&mut self, epoch_index: u64, issuance: Amount) {
+        self.nonverifiable_put(
+            state_key::lqt::v1::budget::for_epoch(epoch_index).into(),
+            issuance,
+        );
     }
 }
 impl<T: StateWrite + ?Sized> StateWriteExt for T {}