feat(node): gas premium distribution

fendermint/app/src/app.rs

@@ -42,12 +42,14 @@ use num_traits::Zero;
 use serde::{Deserialize, Serialize};
 use tendermint::abci::request::CheckTxKind;
 use tendermint::abci::{request, response};
+use tendermint_rpc::Client;
 use tracing::instrument;
 
 use crate::observe::{
     BlockCommitted, BlockProposalEvaluated, BlockProposalReceived, BlockProposalSent, Message,
     MpoolReceived,
 };
+use crate::validators::ValidatorTracker;
 use crate::AppExitCode;
 use crate::BlockHeight;
 use crate::{tmconv::*, VERSION};
@@ -116,10 +118,11 @@ pub struct AppConfig<S: KVStore> {
 
 /// Handle ABCI requests.
 #[derive(Clone)]
-pub struct App<DB, SS, S, I>
+pub struct App<DB, SS, S, I, C>
 where
     SS: Blockstore + Clone + 'static,
     S: KVStore,
+    C: Client,
 {
     /// Database backing all key-value operations.
     db: Arc<DB>,
@@ -160,9 +163,11 @@ where
     ///
     /// Zero means unlimited.
     state_hist_size: u64,
+    /// Tracks the validator
+    validators: ValidatorTracker<C>,
 }
 
-impl<DB, SS, S, I> App<DB, SS, S, I>
+impl<DB, SS, S, I, C> App<DB, SS, S, I, C>
 where
     S: KVStore
         + Codec<AppState>
@@ -171,6 +176,7 @@ where
         + Codec<FvmStateParams>,
     DB: KVWritable<S> + KVReadable<S> + Clone + 'static,
     SS: Blockstore + Clone + 'static,
+    C: Client,
 {
     pub fn new(
         config: AppConfig<S>,
@@ -179,6 +185,7 @@ where
         interpreter: I,
         chain_env: ChainEnv,
         snapshots: Option<SnapshotClient>,
+        client: C,
     ) -> Result<Self> {
         let app = Self {
             db: Arc::new(db),
@@ -193,13 +200,14 @@ where
             snapshots,
             exec_state: Arc::new(tokio::sync::Mutex::new(None)),
             check_state: Arc::new(tokio::sync::Mutex::new(None)),
+            validators: ValidatorTracker::new(client),
         };
         app.init_committed_state()?;
         Ok(app)
     }
 }
 
-impl<DB, SS, S, I> App<DB, SS, S, I>
+impl<DB, SS, S, I, C> App<DB, SS, S, I, C>
 where
     S: KVStore
         + Codec<AppState>
@@ -208,6 +216,7 @@ where
         + Codec<FvmStateParams>,
     DB: KVWritable<S> + KVReadable<S> + 'static + Clone,
     SS: Blockstore + 'static + Clone,
+    C: Client,
 {
     /// Get an owned clone of the state store.
     fn state_store_clone(&self) -> SS {
@@ -393,7 +402,7 @@ where
 // the `tower-abci` library would throw an exception when it tried to convert a
 // `Response::Exception` into a `ConsensusResponse` for example.
 #[async_trait]
-impl<DB, SS, S, I> Application for App<DB, SS, S, I>
+impl<DB, SS, S, I, C> Application for App<DB, SS, S, I, C>
 where
     S: KVStore
         + Codec<AppState>
@@ -421,6 +430,7 @@ where
         Query = BytesMessageQuery,
         Output = BytesMessageQueryRes,
     >,
+    C: Client + Sync,
 {
     /// Provide information about the ABCI application.
     async fn info(&self, _request: request::Info) -> AbciResult<response::Info> {
@@ -727,10 +737,14 @@ where
 
         state_params.timestamp = to_timestamp(request.header.time);
 
+        let validator = self
+            .validators
+            .get_validator(&request.header.proposer_address, block_height)
+            .await?;
         let state = FvmExecState::new(db, self.multi_engine.as_ref(), block_height, state_params)
             .context("error creating new state")?
             .with_block_hash(block_hash)
-            .with_validator_id(request.header.proposer_address);
+            .with_validator(validator);
 
         tracing::debug!("initialized exec state");
 

fendermint/app/src/cmd/run.rs

@@ -294,7 +294,7 @@ async fn run(settings: Settings) -> anyhow::Result<()> {
         None
     };
 
-    let app: App<_, _, AppStore, _> = App::new(
+    let app: App<_, _, AppStore, _, _> = App::new(
         AppConfig {
             app_namespace: ns.app,
             state_hist_namespace: ns.state_hist,
@@ -310,6 +310,7 @@ async fn run(settings: Settings) -> anyhow::Result<()> {
             parent_finality_votes: parent_finality_votes.clone(),
         },
         snapshots,
+        tendermint_client.clone(),
     )?;
 
     if let Some((agent_proxy, config)) = ipc_tuple {

fendermint/app/src/ipc.rs

@@ -15,6 +15,7 @@ use fvm_ipld_blockstore::Blockstore;
 use std::sync::Arc;
 
 use serde::{Deserialize, Serialize};
+use tendermint_rpc::Client;
 
 /// All the things that can be voted on in a subnet.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -24,17 +25,18 @@ pub enum AppVote {
 }
 
 /// Queries the LATEST COMMITTED parent finality from the storage
-pub struct AppParentFinalityQuery<DB, SS, S, I>
+pub struct AppParentFinalityQuery<DB, SS, S, I, C>
 where
     SS: Blockstore + Clone + 'static,
     S: KVStore,
+    C: Client,
 {
     /// The app to get state
-    app: App<DB, SS, S, I>,
+    app: App<DB, SS, S, I, C>,
     gateway_caller: GatewayCaller<ReadOnlyBlockstore<Arc<SS>>>,
 }
 
-impl<DB, SS, S, I> AppParentFinalityQuery<DB, SS, S, I>
+impl<DB, SS, S, I, C> AppParentFinalityQuery<DB, SS, S, I, C>
 where
     S: KVStore
         + Codec<AppState>
@@ -43,8 +45,9 @@ where
         + Codec<FvmStateParams>,
     DB: KVWritable<S> + KVReadable<S> + 'static + Clone,
     SS: Blockstore + 'static + Clone,
+    C: Client,
 {
-    pub fn new(app: App<DB, SS, S, I>) -> Self {
+    pub fn new(app: App<DB, SS, S, I, C>) -> Self {
         Self {
             app,
             gateway_caller: GatewayCaller::default(),
@@ -62,7 +65,7 @@ where
     }
 }
 
-impl<DB, SS, S, I> ParentFinalityStateQuery for AppParentFinalityQuery<DB, SS, S, I>
+impl<DB, SS, S, I, C> ParentFinalityStateQuery for AppParentFinalityQuery<DB, SS, S, I, C>
 where
     S: KVStore
         + Codec<AppState>
@@ -71,6 +74,7 @@ where
         + Codec<FvmStateParams>,
     DB: KVWritable<S> + KVReadable<S> + 'static + Clone,
     SS: Blockstore + 'static + Clone,
+    C: Client,
 {
     fn get_latest_committed_finality(&self) -> anyhow::Result<Option<IPCParentFinality>> {
         self.with_exec_state(|mut exec_state| {

fendermint/app/src/lib.rs

@@ -8,6 +8,7 @@ pub mod metrics;
 pub mod observe;
 mod store;
 mod tmconv;
+mod validators;
 
 pub use app::{App, AppConfig};
 pub use store::{AppStore, BitswapBlockstore};

fendermint/app/src/validators.rs

@@ -0,0 +1,69 @@
+// Copyright 2022-2024 Protocol Labs
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+//! Tracks the validator id from tendermint to their corresponding public key.
+
+use anyhow::anyhow;
+use fendermint_crypto::PublicKey;
+use fvm_shared::clock::ChainEpoch;
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock};
+use tendermint::block::Height;
+use tendermint_rpc::{Client, Paging};
+
+#[derive(Clone)]
+pub(crate) struct ValidatorTracker<C> {
+    client: C,
+    public_keys: Arc<RwLock<HashMap<tendermint::account::Id, PublicKey>>>,
+}
+
+impl<C: Client> ValidatorTracker<C> {
+    pub fn new(client: C) -> Self {
+        Self {
+            client,
+            public_keys: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+}
+
+impl<C: Client + Sync> ValidatorTracker<C> {
+    /// Get the public key of the validator by id. Note that the id is expected to be a validator.
+    pub async fn get_validator(
+        &self,
+        id: &tendermint::account::Id,
+        height: ChainEpoch,
+    ) -> anyhow::Result<PublicKey> {
+        if let Some(key) = self.get_from_cache(id) {
+            return Ok(key);
+        }
+
+        // this means validators have changed, re-pull all validators
+        let height = Height::try_from(height)?;
+        let response = self.client.validators(height, Paging::All).await?;
+
+        let mut new_validators = HashMap::new();
+        let mut pubkey = None;
+        for validator in response.validators {
+            let p = validator.pub_key.secp256k1().unwrap();
+            let compressed = p.to_encoded_point(true);
+            let b = compressed.as_bytes();
+            let key = PublicKey::parse_slice(b, None)?;
+
+            if *id == validator.address {
+                pubkey = Some(key);
+            }
+
+            new_validators.insert(validator.address, key);
+        }
+
+        *self.public_keys.write().unwrap() = new_validators;
+
+        // cannot find the validator, this should not have happened usually
+        pubkey.ok_or_else(|| anyhow!("{} not validator", id))
+    }
+
+    fn get_from_cache(&self, id: &tendermint::account::Id) -> Option<PublicKey> {
+        let keys = self.public_keys.read().unwrap();
+        keys.get(id).copied()
+    }
+}

fendermint/vm/interpreter/src/chain.rs

@@ -383,7 +383,9 @@ where
                     tracing::debug!("chain interpreter applied topdown msgs");
 
                     let local_block_height = state.block_height() as u64;
-                    let proposer = state.validator_id().map(|id| id.to_string());
+                    let proposer = state
+                        .validator_pubkey()
+                        .map(|id| hex::encode(id.serialize_compressed()));
                     let proposer_ref = proposer.as_deref();
 
                     atomically(|| {

fendermint/vm/interpreter/src/fvm/exec.rs

@@ -13,7 +13,7 @@ use ipc_observability::{emit, measure_time, observe::TracingError, Traceable};
 use tendermint_rpc::Client;
 
 use crate::fvm::cometbft::EndBlockUpdate;
-use crate::fvm::gas::GasMarket;
+use crate::fvm::gas::{GasMarket, GasUtilization};
 use crate::ExecInterpreter;
 
 use super::{
@@ -175,7 +175,7 @@ where
 
             state
                 .gas_market_mut()
-                .record_utilization(apply_ret.msg_receipt.gas_used);
+                .record_utilization(GasUtilization::from(&apply_ret));
 
             (apply_ret, emitters, latency)
         };