Merge branch 'main' into lowhung/unify-hash-types

# Conflicts:
#	modules/historical_accounts_state/src/immutable_historical_account_store.rs

Author: lowhung

.github/workflows/run-tests-on-push-to-main.yml

@@ -39,8 +39,10 @@ jobs:
             --package acropolis_module_epochs_state \
             --package acropolis_module_genesis_bootstrapper \
             --package acropolis_module_governance_state \
+            --package acropolis_module_historical_accounts_state \
             --package acropolis_module_mithril_snapshot_fetcher \
             --package acropolis_module_parameters_state \
+            --package acropolis_module_rest_blockfrost \
             --package acropolis_module_snapshot_bootstrapper \
             --package acropolis_module_spdd_state \
             --package acropolis_module_spo_state \

common/src/queries/accounts.rs

@@ -18,7 +18,7 @@ pub enum AccountsStateQuery {
     GetAccountRegistrationHistory { account: StakeAddress },
     GetAccountDelegationHistory { account: StakeAddress },
     GetAccountMIRHistory { account: StakeAddress },
-    GetAccountWithdrawalHistory { stake_key: Vec<u8> },
+    GetAccountWithdrawalHistory { account: StakeAddress },
     GetAccountAssociatedAddresses { stake_key: Vec<u8> },
     GetAccountAssets { stake_key: Vec<u8> },
     GetAccountAssetsTotals { stake_key: Vec<u8> },
@@ -52,7 +52,7 @@ pub enum AccountsStateQueryResponse {
     AccountRegistrationHistory(Vec<RegistrationUpdate>),
     AccountDelegationHistory(Vec<DelegationUpdate>),
     AccountMIRHistory(Vec<AccountWithdrawal>),
-    AccountWithdrawalHistory(AccountWithdrawalHistory),
+    AccountWithdrawalHistory(Vec<AccountWithdrawal>),
     AccountAssociatedAddresses(AccountAssociatedAddresses),
     AccountAssets(AccountAssets),
     AccountAssetsTotals(AccountAssetsTotals),

common/src/stake_addresses.rs

@@ -866,6 +866,8 @@ mod tests {
     }
 
     mod withdrawal_tests {
+        use crate::TxIdentifier;
+
         use super::*;
 
         #[test]
@@ -879,6 +881,7 @@ mod tests {
             let withdrawal = Withdrawal {
                 address: stake_address.clone(),
                 value: 40,
+                tx_identifier: TxIdentifier::default(),
             };
             stake_addresses.process_withdrawal(&withdrawal);
 
@@ -897,6 +900,7 @@ mod tests {
             let withdrawal = Withdrawal {
                 address: stake_address.clone(),
                 value: 24,
+                tx_identifier: TxIdentifier::default(),
             };
             stake_addresses.process_withdrawal(&withdrawal);
             assert_eq!(stake_addresses.get(&stake_address).unwrap().rewards, 12);
@@ -905,6 +909,7 @@ mod tests {
             let withdrawal = Withdrawal {
                 address: stake_address.clone(),
                 value: 2,
+                tx_identifier: TxIdentifier::default(),
             };
             stake_addresses.process_withdrawal(&withdrawal);
             assert_eq!(stake_addresses.get(&stake_address).unwrap().rewards, 10);
@@ -921,6 +926,7 @@ mod tests {
             let withdrawal = Withdrawal {
                 address: stake_address.clone(),
                 value: 0,
+                tx_identifier: TxIdentifier::default(),
             };
 
             stake_addresses.process_withdrawal(&withdrawal);
@@ -935,6 +941,7 @@ mod tests {
             let withdrawal = Withdrawal {
                 address: stake_address.clone(),
                 value: 10,
+                tx_identifier: TxIdentifier::default(),
             };
 
             // Should log error but not panic

common/src/types.rs

@@ -612,6 +612,9 @@ pub struct Withdrawal {
 
     /// Value to withdraw
     pub value: Lovelace,
+
+    // Identifier of withdrawal tx
+    pub tx_identifier: TxIdentifier,
 }
 
 /// Treasury pot account

modules/accounts_state/src/rewards.rs

@@ -14,7 +14,7 @@ use std::sync::Arc;
 use tracing::{debug, info, warn};
 
 /// Type of reward being given
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub enum RewardType {
     Leader,
     Member,
@@ -34,7 +34,7 @@ pub struct RewardDetail {
 }
 
 /// Result of a rewards calculation
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct RewardsResult {
     /// Epoch these rewards were earned in (when blocks produced)
     pub epoch: u64,

modules/accounts_state/src/state.rs

@@ -1,6 +1,6 @@
 //! Acropolis AccountsState: State storage
 use crate::monetary::calculate_monetary_change;
-use crate::rewards::{calculate_rewards, RewardsResult};
+use crate::rewards::{calculate_rewards, RewardType, RewardsResult};
 use crate::snapshot::Snapshot;
 use crate::verifier::Verifier;
 use acropolis_common::queries::accounts::OptimalPoolSizing;
@@ -117,9 +117,6 @@ pub struct State {
     /// Pool refunds to apply next epoch (list of reward accounts to refund to)
     pool_refunds: Vec<StakeAddress>,
 
-    /// Stake address refunds to apply next epoch
-    stake_refunds: Vec<(StakeAddress, Lovelace)>,
-
     /// MIRs to pay next epoch
     mirs: Vec<MoveInstantaneousReward>,
 
@@ -325,7 +322,6 @@ impl State {
 
         // Pay the refunds after snapshot, so they don't appear in active_stake
         reward_deltas.extend(self.pay_pool_refunds());
-        reward_deltas.extend(self.pay_stake_refunds());
 
         // Verify pots state
         verifier.verify_pots(epoch, &self.pots);
@@ -497,33 +493,6 @@ impl State {
         reward_deltas
     }
 
-    /// Pay stake address refunds
-    fn pay_stake_refunds(&mut self) -> Vec<StakeRewardDelta> {
-        let mut reward_deltas = Vec::<StakeRewardDelta>::new();
-
-        let refunds = take(&mut self.stake_refunds);
-        if !refunds.is_empty() {
-            info!(
-                "{} deregistered stake addresses, total refunds {}",
-                refunds.len(),
-                refunds.iter().map(|(_, n)| n).sum::<Lovelace>()
-            );
-        }
-
-        // Send them their deposits back
-        for (stake_address, deposit) in refunds {
-            let mut stake_addresses = self.stake_addresses.lock().unwrap();
-            reward_deltas.push(StakeRewardDelta {
-                stake_address: stake_address.clone(), // Extract hash for delta
-                delta: deposit as i64,
-            });
-            stake_addresses.add_to_reward(&stake_address, deposit);
-            self.pots.deposits -= deposit;
-        }
-
-        reward_deltas
-    }
-
     /// Pay MIRs
     fn pay_mirs(&mut self) -> Vec<StakeRewardDelta> {
         let mut reward_deltas = Vec::<StakeRewardDelta>::new();
@@ -658,36 +627,52 @@ impl State {
 
         // If rewards have been calculated, save the results
         if let Some(task) = task.take() {
-            if let Ok(Ok(reward_result)) = task.await {
-                // Collect rewards to stake addresses reward deltas
-                for rewards in reward_result.rewards.values() {
-                    reward_deltas.extend(
-                        rewards
-                            .iter()
-                            .map(|reward| StakeRewardDelta {
-                                stake_address: reward.account.clone(),
-                                delta: reward.amount as i64,
-                            })
-                            .collect::<Vec<_>>(),
-                    );
-                }
-
-                // Verify them
-                verifier.verify_rewards(reward_result.epoch, &reward_result);
-
+            if let Ok(Ok(rewards_result)) = task.await {
                 // Pay the rewards
                 let mut stake_addresses = self.stake_addresses.lock().unwrap();
-                for (_, rewards) in reward_result.rewards {
+                let mut filtered_rewards_result = rewards_result.clone();
+                for (spo, rewards) in rewards_result.rewards {
                     for reward in rewards {
-                        stake_addresses.add_to_reward(&reward.account, reward.amount);
+                        if stake_addresses.is_registered(&reward.account) {
+                            stake_addresses.add_to_reward(&reward.account, reward.amount);
+                            reward_deltas.push(StakeRewardDelta {
+                                stake_address: reward.account.clone(),
+                                delta: reward.amount as i64,
+                            });
+                        } else {
+                            warn!(
+                                "Reward account {} deregistered - paying reward {} to treasury",
+                                reward.account, reward.amount
+                            );
+                            self.pots.treasury += reward.amount;
+
+                            // Remove from filtered version for comparison and result
+                            if let Some(rewards) = filtered_rewards_result.rewards.get_mut(&spo) {
+                                rewards.retain(|r| r.account != reward.account);
+                            }
+
+                            if let Some((_, spor)) = filtered_rewards_result
+                                .spo_rewards
+                                .iter_mut()
+                                .find(|(fspo, _)| *fspo == spo)
+                            {
+                                spor.total_rewards -= reward.amount;
+                                if reward.rtype == RewardType::Leader {
+                                    spor.operator_rewards -= reward.amount;
+                                }
+                            }
+                        }
                     }
                 }
 
+                // Verify them
+                verifier.verify_rewards(&filtered_rewards_result);
+
                 // save SPO rewards
-                spo_rewards = reward_result.spo_rewards.into_iter().collect();
+                spo_rewards = filtered_rewards_result.spo_rewards.clone();
 
                 // Adjust the reserves for next time with amount actually paid
-                self.pots.reserves -= reward_result.total_paid;
+                self.pots.reserves -= rewards_result.total_paid;
             }
         };
 
@@ -842,8 +827,7 @@ impl State {
                 }
             };
 
-            // Schedule refund
-            self.stake_refunds.push((stake_address.clone(), deposit));
+            self.pots.deposits -= deposit;
 
             // Add to registration changes
             self.current_epoch_registration_changes.lock().unwrap().push(RegistrationChange {
@@ -1331,6 +1315,7 @@ mod tests {
             withdrawals: vec![Withdrawal {
                 address: stake_address.clone(),
                 value: 39,
+                tx_identifier: TxIdentifier::default(),
             }],
         };
 

modules/accounts_state/src/verifier.rs

@@ -119,7 +119,8 @@ impl Verifier {
     }
 
     /// Verify rewards, logging any errors
-    pub fn verify_rewards(&self, epoch: u64, rewards: &RewardsResult) {
+    pub fn verify_rewards(&self, rewards: &RewardsResult) {
+        let epoch = rewards.epoch;
         if let Some(template) = &self.rewards_file_template {
             let path = template.replace("{}", &epoch.to_string());
 

modules/historical_accounts_state/src/historical_accounts_state.rs

@@ -110,7 +110,7 @@ impl HistoricalAccountsState {
                 if let Message::Cardano((ref block_info, CardanoMessage::ProtocolParams(params))) =
                     params_msg.as_ref()
                 {
-                    Self::check_sync(&current_block, &block_info);
+                    Self::check_sync(&current_block, block_info);
                     let mut state = state_mutex.lock().await;
                     state.volatile.start_new_epoch(block_info.number);
                     if let Some(shelley) = &params.params.shelley {
@@ -124,7 +124,7 @@ impl HistoricalAccountsState {
                     CardanoMessage::StakeRewardDeltas(rewards_msg),
                 )) = rewards_msg.as_ref()
                 {
-                    Self::check_sync(&current_block, &block_info);
+                    Self::check_sync(&current_block, block_info);
                     let mut state = state_mutex.lock().await;
                     state
                         .handle_rewards(rewards_msg)
@@ -142,12 +142,9 @@ impl HistoricalAccountsState {
                     );
                     let _entered = span.enter();
 
-                    Self::check_sync(&current_block, &block_info);
+                    Self::check_sync(&current_block, block_info);
                     let mut state = state_mutex.lock().await;
-                    state
-                        .handle_tx_certificates(tx_certs_msg, block_info.epoch as u32)
-                        .inspect_err(|e| error!("TxCertificates handling error: {e:#}"))
-                        .ok();
+                    state.handle_tx_certificates(tx_certs_msg, block_info.epoch as u32);
                 }
 
                 _ => error!("Unexpected message type: {certs_message:?}"),
@@ -163,12 +160,9 @@ impl HistoricalAccountsState {
                     );
                     let _entered = span.enter();
 
-                    Self::check_sync(&current_block, &block_info);
+                    Self::check_sync(&current_block, block_info);
                     let mut state = state_mutex.lock().await;
-                    state
-                        .handle_withdrawals(withdrawals_msg)
-                        .inspect_err(|e| error!("Withdrawals handling error: {e:#}"))
-                        .ok();
+                    state.handle_withdrawals(withdrawals_msg);
                 }
 
                 _ => error!("Unexpected message type: {message:?}"),
@@ -184,7 +178,7 @@ impl HistoricalAccountsState {
                     );
                     let _entered = span.enter();
 
-                    Self::check_sync(&current_block, &block_info);
+                    Self::check_sync(&current_block, block_info);
                     {
                         let mut state = state_mutex.lock().await;
                         state
@@ -206,8 +200,7 @@ impl HistoricalAccountsState {
 
                 if should_prune {
                     let (store, cfg) = {
-                        let mut state: tokio::sync::MutexGuard<'_, State> =
-                            state_mutex.lock().await;
+                        let mut state = state_mutex.lock().await;
                         state.prune_volatile().await;
                         (state.immutable.clone(), state.config.clone())
                     };
@@ -320,7 +313,7 @@ impl HistoricalAccountsState {
 
                 let response = match query {
                     AccountsStateQuery::GetAccountRegistrationHistory { account } => {
-                        match state.lock().await.get_registration_history(&account).await {
+                        match state.lock().await.get_registration_history(account).await {
                             Ok(Some(registrations)) => {
                                 AccountsStateQueryResponse::AccountRegistrationHistory(
                                     registrations,
@@ -331,7 +324,7 @@ impl HistoricalAccountsState {
                         }
                     }
                     AccountsStateQuery::GetAccountDelegationHistory { account } => {
-                        match state.lock().await.get_delegation_history(&account).await {
+                        match state.lock().await.get_delegation_history(account).await {
                             Ok(Some(delegations)) => {
                                 AccountsStateQueryResponse::AccountDelegationHistory(delegations)
                             }
@@ -340,13 +333,21 @@ impl HistoricalAccountsState {
                         }
                     }
                     AccountsStateQuery::GetAccountMIRHistory { account } => {
-                        match state.lock().await.get_mir_history(&account).await {
+                        match state.lock().await.get_mir_history(account).await {
                             Ok(Some(mirs)) => AccountsStateQueryResponse::AccountMIRHistory(mirs),
                             Ok(None) => AccountsStateQueryResponse::NotFound,
                             Err(e) => AccountsStateQueryResponse::Error(e.to_string()),
                         }
                     }
-
+                    AccountsStateQuery::GetAccountWithdrawalHistory { account } => {
+                        match state.lock().await.get_withdrawal_history(account).await {
+                            Ok(Some(withdrawals)) => {
+                                AccountsStateQueryResponse::AccountWithdrawalHistory(withdrawals)
+                            }
+                            Ok(None) => AccountsStateQueryResponse::NotFound,
+                            Err(e) => AccountsStateQueryResponse::Error(e.to_string()),
+                        }
+                    }
                     _ => AccountsStateQueryResponse::Error(format!(
                         "Unimplemented query variant: {:?}",
                         query