feat: add block proposer to addresses_with_balance_change

chain/src/main.rs

@@ -1,3 +1,4 @@
+use std::collections::HashSet;
 use std::convert::identity;
 use std::sync::Arc;
 
@@ -164,16 +165,32 @@ async fn crawling_fn(
         .map(Token::Ibc)
         .collect::<Vec<Token>>();
 
-    let addresses = block.addresses_with_balance_change(native_token);
+    let addresses = block.addresses_with_balance_change(&native_token);
+
+    let block_proposer_address = namada_service::get_block_proposer_address(
+        &client,
+        &block,
+        &native_token,
+    )
+    .await
+    .into_rpc_error()?;
+
+    let all_balance_changed_addresses = addresses
+        .iter()
+        .chain(block_proposer_address.iter())
+        .cloned()
+        .collect::<HashSet<_>>();
 
     let balances =
         namada_service::query_balance(&client, &addresses, block_height)
             .await
             .into_rpc_error()?;
     tracing::debug!(
         block = block_height,
+        addresses = addresses.len(),
+        block_proposer_address = block_proposer_address.len(),
         "Updating balance for {} addresses...",
-        addresses.len()
+        all_balance_changed_addresses.len()
     );
 
     let next_governance_proposal_id =

chain/src/services/namada.rs

@@ -19,7 +19,7 @@ use namada_sdk::state::Key;
 use namada_sdk::token::Amount as NamadaSdkAmount;
 use namada_sdk::{borsh, rpc, token};
 use shared::balance::{Amount, Balance, Balances};
-use shared::block::{BlockHeight, Epoch};
+use shared::block::{Block, BlockHeight, Epoch};
 use shared::bond::{Bond, BondAddresses, Bonds};
 use shared::id::Id;
 use shared::proposal::{GovernanceProposal, TallyType};
@@ -693,6 +693,40 @@ pub async fn get_validator_set_at_epoch(
     Ok(ValidatorSet { validators, epoch })
 }
 
+pub async fn get_block_proposer_address(
+    client: &HttpClient,
+    block: &Block,
+    native_token: &Id,
+) -> anyhow::Result<HashSet<BalanceChange>> {
+    let validator = RPC
+        .vp()
+        .pos()
+        .validator_by_tm_addr(
+            client,
+            &block.header.proposer_address.to_uppercase(),
+        )
+        .await?;
+
+    tracing::debug!(
+        block = block.header.height,
+        native_token = native_token.to_string(),
+        proposer_address = block.header.proposer_address,
+        namada_address = ?validator,
+        "Got block proposer address"
+    );
+
+    match validator {
+        Some(validator) => {
+            let balance_change = BalanceChange {
+                address: Id::from(validator),
+                token: Token::Native(native_token.clone()),
+            };
+            Ok(std::iter::once(balance_change).collect())
+        }
+        None => Ok(HashSet::new()),
+    }
+}
+
 pub async fn query_pipeline_length(client: &HttpClient) -> anyhow::Result<u64> {
     let pos_parameters = rpc::get_pos_params(client)
         .await

shared/src/block.rs

@@ -339,15 +339,15 @@ impl Block {
     // TODO: move this and process_inner_tx_for_balance to a separate module
     pub fn addresses_with_balance_change(
         &self,
-        native_token: Id,
+        native_token: &Id,
     ) -> HashSet<BalanceChange> {
         self.transactions
             .iter()
             .flat_map(|(wrapper_tx, inners_txs)| {
                 let mut balance_changes: Vec<BalanceChange> = inners_txs
                     .iter()
                     .filter_map(|tx| {
-                        self.process_inner_tx_for_balance(tx, &native_token)
+                        self.process_inner_tx_for_balance(tx, native_token)
                     })
                     .flatten()
                     .collect();