diff --git a/packages/rinku-node/src/checkpoint.rs b/packages/rinku-node/src/checkpoint.rs index cc959f8..b0b2124 100644 --- a/packages/rinku-node/src/checkpoint.rs +++ b/packages/rinku-node/src/checkpoint.rs @@ -1927,15 +1927,7 @@ impl CheckpointService { .process_batch_special_txs_with_skip(&all_special_txs, &fast_path_already_finalized) .await; - { - let state = self.state.inner.read().await; - let mut rewards = self.state.rewards.write().await; - for (addr, account) in &state.accounts { - if account.staked > 0 { - rewards.sync_stake_amount(addr, account.staked); - } - } - } + self.state.reconcile_rewards_stakes_from_accounts().await; self.state .process_batch_reward_infos(&contract_lane_txs, &batch_result.executed_hashes) diff --git a/packages/rinku-node/src/main.rs b/packages/rinku-node/src/main.rs index d1954a7..856e7a4 100644 --- a/packages/rinku-node/src/main.rs +++ b/packages/rinku-node/src/main.rs @@ -454,29 +454,13 @@ async fn main() -> Result<()> { let genesis_addresses: std::collections::HashSet = genesis_seed.iter().map(|(a, _)| a.clone()).collect(); + // Register genesis validators in RewardsService if missing. + // Do NOT blanket-remove non-genesis stakes — that confiscated user / + // delegator stakes on every genesis restart. Ghost redeploy validators + // are handled by reconcile_stakes_after_genesis_replace below. { use crate::validator_identity::GENESIS_VALIDATOR_STAKE; let mut rewards = state.rewards.write().await; - if config.is_genesis_node { - let existing_stakes: Vec = rewards - .get_all_stakes() - .iter() - .map(|s| s.staker.clone()) - .collect(); - let mut removed = 0; - for staker in &existing_stakes { - if !genesis_addresses.contains(staker) { - rewards.remove_stake(staker); - removed += 1; - } - } - if removed > 0 { - info!( - "Removed {} stale stake(s) from rewards service (not in genesis set)", - removed - ); - } - } let mut registered = 0; for (address, _) in &genesis_seed { if rewards.get_stake(address).is_none() { @@ -572,11 +556,13 @@ async fn main() -> Result<()> { } } } - // Clean up ghost accounts on ALL nodes (not just genesis) - state.cleanup_stale_accounts(&genesis_addresses).await; + // Preserve user stakes; only clear redeploy-ghost validators (+ empty accounts). + state + .reconcile_stakes_after_genesis_replace(&genesis_addresses) + .await; } else { let empty_set = std::collections::HashSet::new(); - state.cleanup_stale_accounts(&empty_set).await; + state.cleanup_empty_ghost_accounts(&empty_set).await; } // Sync stakes to accounts AFTER genesis validator replacement diff --git a/packages/rinku-node/src/rewards.rs b/packages/rinku-node/src/rewards.rs index 494a877..eb7a503 100644 --- a/packages/rinku-node/src/rewards.rs +++ b/packages/rinku-node/src/rewards.rs @@ -415,7 +415,22 @@ impl RewardsService { } } + /// Align `rewards.stakes` with the canonical `account.staked` value. + /// + /// `canonical_amount == 0` removes the rewards entry entirely so ghost + /// stakes cannot linger after account state has already been cleared + /// (previously we only synced when account.staked > 0, leaving orphans). pub fn sync_stake_amount(&mut self, staker: &str, canonical_amount: u64) { + if canonical_amount == 0 { + if self.stakes.remove(staker).is_some() { + tracing::info!( + "STAKE SYNC REMOVE: {} rewards.stakes cleared (canonical account.staked=0)", + &staker[..16.min(staker.len())] + ); + } + return; + } + if let Some(existing) = self.stakes.get_mut(staker) { if existing.amount != canonical_amount { tracing::info!( @@ -426,7 +441,7 @@ impl RewardsService { ); existing.amount = canonical_amount; } - } else if canonical_amount > 0 { + } else { tracing::info!( "STAKE SYNC CREATE: {} creating rewards.stakes entry with amount {} (canonical account.staked, no prior entry)", &staker[..16.min(staker.len())], @@ -846,7 +861,7 @@ impl RewardsService { } } - fn add_pending_reward(&mut self, address: &str, amount: u64) { + pub(crate) fn add_pending_reward(&mut self, address: &str, amount: u64) { let pending = self.pending_rewards.entry(address.to_string()).or_insert(0); *pending += amount; } @@ -1150,6 +1165,63 @@ mod tests { assert!(unstake_result.is_err()); } + #[test] + fn sync_stake_amount_zero_removes_entry() { + let mut service = RewardsService::new(RewardConfig::default()); + service.stake("user", to_micro_units(199.0), "tx1").unwrap(); + assert!(service.get_stake("user").is_some()); + assert_eq!(service.get_total_staked(), to_micro_units(199.0)); + + service.sync_stake_amount("user", 0); + assert!( + service.get_stake("user").is_none(), + "canonical account.staked=0 must remove rewards.stakes entry" + ); + assert_eq!(service.get_total_staked(), 0); + assert_eq!(service.get_active_validators().len(), 0); + } + + #[test] + fn sync_stake_amount_zero_is_noop_when_absent() { + let mut service = RewardsService::new(RewardConfig::default()); + service.sync_stake_amount("nobody", 0); + assert_eq!(service.get_total_staked(), 0); + } + + #[test] + fn sync_stake_amount_updates_and_creates() { + let mut service = RewardsService::new(RewardConfig::default()); + service.stake("user", to_micro_units(100.0), "tx1").unwrap(); + service.sync_stake_amount("user", to_micro_units(250.0)); + assert_eq!( + service.get_stake("user").unwrap().amount, + to_micro_units(250.0) + ); + + service.sync_stake_amount("other", to_micro_units(100.0)); + assert_eq!( + service.get_stake("other").unwrap().amount, + to_micro_units(100.0) + ); + assert_eq!(service.get_total_staked(), to_micro_units(350.0)); + } + + #[test] + fn pending_rewards_survive_stake_removal() { + let mut service = RewardsService::new(RewardConfig::default()); + service.stake("user", to_micro_units(200.0), "tx1").unwrap(); + service.add_pending_reward("user", to_micro_units(11.0)); + assert!(service.get_pending_rewards("user") > 0); + + service.sync_stake_amount("user", 0); + assert!(service.get_stake("user").is_none()); + assert_eq!( + service.get_pending_rewards("user"), + to_micro_units(11.0), + "pending rewards must not be wiped when stake entry is removed" + ); + } + #[test] fn test_tip_rewards() { let mut service = RewardsService::new(RewardConfig::default()); diff --git a/packages/rinku-node/src/state/accounts.rs b/packages/rinku-node/src/state/accounts.rs index 8959f7e..4cf4f72 100644 --- a/packages/rinku-node/src/state/accounts.rs +++ b/packages/rinku-node/src/state/accounts.rs @@ -1,44 +1,134 @@ use super::*; impl NodeState { - /// Remove ghost accounts from old deployments. - /// First zeroes out stale stakes on accounts that are NOT in the current - /// genesis validator set (they retain staked amounts from a previous deployment). - /// Then removes any non-allowed accounts with 0 balance + 0 staked. + /// Post-genesis-seed stake reconciliation. + /// + /// Preserves all real user / delegator stakes. Only clears *redeploy ghost + /// validators* (prior `GENESIS_VALIDATORS` leftovers) and empty accounts. + /// Never confiscates stake without crediting `account.balance`. + pub async fn reconcile_stakes_after_genesis_replace( + &self, + genesis_addresses: &std::collections::HashSet, + ) { + self.cleanup_redeploy_ghost_validators(genesis_addresses) + .await; + self.cleanup_empty_ghost_accounts(genesis_addresses).await; + } + + /// Legacy entry point — safe wrapper (does **not** zero user stakes). pub async fn cleanup_stale_accounts( &self, allowed_addresses: &std::collections::HashSet, ) { - let mut state = self.inner.write().await; + self.reconcile_stakes_after_genesis_replace(allowed_addresses) + .await; + } - let stale_stakers: Vec = state - .accounts - .iter() - .filter(|(addr, account)| { - *addr != "faucet" - && *addr != "genesis" - && !allowed_addresses.contains(*addr) - && account.staked > 0 - }) - .map(|(addr, _)| addr.clone()) - .collect(); + /// Clear leftover genesis-validator accounts from a previous deploy. + /// + /// A redeploy ghost is an account that is: + /// - not in the current genesis validator set + /// - staked at exactly `GENESIS_VALIDATOR_STAKE` + /// - `nonce == 0` (never sent a user tx — real stakers always bump nonce) + /// + /// Stake is credited back to balance before clearing. Matching rewards + /// entries are removed. Arbitrary user stakes are never touched. + pub async fn cleanup_redeploy_ghost_validators( + &self, + genesis_addresses: &std::collections::HashSet, + ) { + use crate::validator_identity::GENESIS_VALIDATOR_STAKE; - for addr in &stale_stakers { - if let Some(account) = state.accounts.get_mut(addr) { + let ghosts: Vec<(String, u64)> = { + let state = self.inner.read().await; + state + .accounts + .iter() + .filter(|(addr, account)| { + *addr != "faucet" + && *addr != "genesis" + && !genesis_addresses.contains(*addr) + && account.staked == GENESIS_VALIDATOR_STAKE + && account.nonce == 0 + }) + .map(|(addr, account)| (addr.clone(), account.staked)) + .collect() + }; + + if !ghosts.is_empty() { + let mut state = self.inner.write().await; + let mut credited = Vec::with_capacity(ghosts.len()); + for (addr, staked) in &ghosts { + if let Some(account) = state.accounts.get_mut(addr) { + account.balance = account.balance.saturating_add(*staked); + account.staked = 0; + info!( + "Credited redeploy-ghost validator stake to balance for {}: +{} µRKU (stake cleared)", + &addr[..16.min(addr.len())], + staked + ); + credited.push(addr.clone()); + } + } + if !credited.is_empty() { + state.update_state_trie_accounts(&credited); info!( - "Zeroing stale stake on non-validator account {}: {:.4} RKU", - &addr[..16.min(addr.len())], - account.staked + "Cleared {} redeploy-ghost validator account stake(s) with balance credit", + credited.len() ); - account.staked = 0; } } - if !stale_stakers.is_empty() { - info!( - "Zeroed stale stakes on {} ghost validator account(s)", - stale_stakers.len() - ); + + // Scrub rewards.stakes for the same ghost heuristic (and rewards-only + // orphans that look like old genesis validators with no live user account). + { + let account_meta: std::collections::HashMap = { + let state = self.inner.read().await; + state + .accounts + .iter() + .map(|(addr, acc)| (addr.clone(), (acc.staked, acc.nonce))) + .collect() + }; + let mut rewards = self.rewards.write().await; + let reward_stakers: Vec<(String, u64)> = rewards + .get_all_stakes() + .iter() + .map(|s| (s.staker.clone(), s.amount)) + .collect(); + let mut removed = 0u32; + for (staker, amount) in reward_stakers { + if genesis_addresses.contains(&staker) { + continue; + } + if amount != GENESIS_VALIDATOR_STAKE { + continue; + } + let is_user_stake = account_meta.get(&staker).is_some_and(|(staked, nonce)| { + *nonce > 0 || (*staked > 0 && *staked != GENESIS_VALIDATOR_STAKE) + }); + if is_user_stake { + continue; + } + // Ghost: missing account, or nonce==0 with stake already cleared / still genesis-sized. + rewards.remove_stake(&staker); + removed += 1; + } + if removed > 0 { + info!( + "Removed {} redeploy-ghost stake(s) from rewards service", + removed + ); + } } + } + + /// Remove non-system accounts with zero balance and zero stake. + pub async fn cleanup_empty_ghost_accounts( + &self, + allowed_addresses: &std::collections::HashSet, + ) { + let mut state = self.inner.write().await; let stale: Vec = state .accounts @@ -58,12 +148,44 @@ impl NodeState { state.accounts.remove(addr); } info!( - "Cleaned up {} ghost account(s) from old snapshot", + "Cleaned up {} empty ghost account(s) from old snapshot", stale.len() ); } } + /// Push canonical `account.staked` into `RewardsService` for every known + /// staker — including zeros, which remove orphaned rewards entries. + /// + /// Call after checkpoint apply so account state and rewards.stakes converge. + pub async fn reconcile_rewards_stakes_from_accounts(&self) { + let account_stakes: std::collections::HashMap = { + let state = self.inner.read().await; + state + .accounts + .iter() + .map(|(addr, acc)| (addr.clone(), acc.staked)) + .collect() + }; + + let mut rewards = self.rewards.write().await; + let reward_stakers: Vec = rewards + .get_all_stakes() + .iter() + .map(|s| s.staker.clone()) + .collect(); + + for staker in &reward_stakers { + let canonical = account_stakes.get(staker).copied().unwrap_or(0); + rewards.sync_stake_amount(staker, canonical); + } + for (addr, amount) in &account_stakes { + if *amount > 0 { + rewards.sync_stake_amount(addr, *amount); + } + } + } + /// Sync all stakes from RewardsService to account.staked fields /// Must be called AFTER replace_validators_with_genesis to avoid ghost accounts pub async fn sync_stakes_to_accounts(&self) { @@ -526,3 +648,319 @@ impl NodeState { .collect() } } + +#[cfg(test)] +mod stake_cleanup_tests { + use super::*; + use crate::config::NodeConfig; + use crate::validator_identity::GENESIS_VALIDATOR_STAKE; + use rinku_core::types::to_micro_units; + + fn acct(addr: &str, bal: u64, nonce: u64, staked: u64) -> Account { + Account { + address: addr.to_string(), + balance: bal, + nonce, + first_seen: 0, + staked, + unbonding: 0, + unbonding_release: None, + latest_balance_proof: None, + partition_violations: 0, + reputation_penalty: 0.0, + penalty_decay_checkpoint: None, + partition_budget: None, + partition_budget_spent: 0, + ecdsa_public_key: None, + } + } + + async fn fresh_state() -> (tempfile::TempDir, NodeState) { + let dir = tempfile::tempdir().unwrap(); + let config = NodeConfig { + data_dir: dir.path().to_string_lossy().to_string(), + is_genesis_node: true, + ..NodeConfig::default() + }; + let state = NodeState::new(config).await.expect("NodeState"); + (dir, state) + } + + async fn seed(state: &NodeState, accounts: &[Account]) { + let mut inner = state.inner.write().await; + inner.accounts.clear(); + for a in accounts { + inner.accounts.insert(a.address.clone(), a.clone()); + } + inner.state_trie = StateInner::build_state_trie_from_accounts(&inner.accounts); + } + + /// Regression: user stake must survive genesis-set cleanup that previously + /// zeroed every non-GENESIS_VALIDATORS account.staked on restart. + #[tokio::test] + async fn genesis_cleanup_preserves_user_stake_and_balance() { + let (_dir, state) = fresh_state().await; + let user = "923cc639b27a6d07cd1f2879166b8a53e01bb8ef"; + let val_a = "01332180dd6879ed3e4fa853f5c15dbd0b903e97"; + let user_stake = to_micro_units(199.0); + let user_bal = to_micro_units(101.0); + + seed( + &state, + &[ + acct(val_a, 0, 0, GENESIS_VALIDATOR_STAKE), + acct(user, user_bal, 2, user_stake), + ], + ) + .await; + { + let mut rewards = state.rewards.write().await; + rewards.stake(val_a, GENESIS_VALIDATOR_STAKE, "").unwrap(); + rewards.stake(user, user_stake, "user-stake-tx").unwrap(); + rewards.add_pending_reward(user, to_micro_units(11.6)); + } + + let mut genesis = std::collections::HashSet::new(); + genesis.insert(val_a.to_string()); + + state.reconcile_stakes_after_genesis_replace(&genesis).await; + + let (staked, balance, nonce) = { + let inner = state.inner.read().await; + let a = inner.accounts.get(user).expect("user account"); + (a.staked, a.balance, a.nonce) + }; + assert_eq!(staked, user_stake, "user stake must not be confiscated"); + assert_eq!(balance, user_bal, "user balance must be unchanged"); + assert_eq!(nonce, 2); + + let rewards = state.rewards.read().await; + assert_eq!( + rewards.get_stake(user).map(|s| s.amount), + Some(user_stake), + "rewards.stakes for user must survive genesis cleanup" + ); + assert!( + (rewards.get_pending_rewards(user) as i64 - to_micro_units(11.6) as i64).abs() < 2, + "pending rewards must survive" + ); + assert_eq!( + rewards.get_stake(val_a).map(|s| s.amount), + Some(GENESIS_VALIDATOR_STAKE) + ); + } + + /// Redeploy ghost validators (exact genesis stake, nonce 0, not in set) + /// are cleared, but principal is credited back to balance. + #[tokio::test] + async fn redeploy_ghost_validator_stake_credited_not_burned() { + let (_dir, state) = fresh_state().await; + let live = "livevalidator000000000000000000000000001"; + let ghost = "oldvalidator000000000000000000000000001"; + + seed( + &state, + &[ + acct(live, 0, 0, GENESIS_VALIDATOR_STAKE), + acct(ghost, 0, 0, GENESIS_VALIDATOR_STAKE), + ], + ) + .await; + { + let mut rewards = state.rewards.write().await; + let _ = rewards.stake(live, GENESIS_VALIDATOR_STAKE, ""); + let _ = rewards.stake(ghost, GENESIS_VALIDATOR_STAKE, ""); + } + + let mut genesis = std::collections::HashSet::new(); + genesis.insert(live.to_string()); + state.reconcile_stakes_after_genesis_replace(&genesis).await; + + let inner = state.inner.read().await; + let g = inner + .accounts + .get(ghost) + .expect("ghost account kept (has balance now)"); + assert_eq!(g.staked, 0, "ghost stake cleared"); + assert_eq!( + g.balance, GENESIS_VALIDATOR_STAKE, + "ghost stake must be credited to balance, not burned" + ); + let l = inner.accounts.get(live).unwrap(); + assert_eq!(l.staked, GENESIS_VALIDATOR_STAKE); + drop(inner); + + let rewards = state.rewards.read().await; + assert!(rewards.get_stake(ghost).is_none()); + assert_eq!( + rewards.get_stake(live).map(|s| s.amount), + Some(GENESIS_VALIDATOR_STAKE) + ); + } + + /// A wallet that happens to hold GENESIS_VALIDATOR_STAKE but has nonce > 0 + /// is a real user, not a redeploy ghost — must be preserved. + #[tokio::test] + async fn large_user_stake_with_nonce_is_not_treated_as_ghost() { + let (_dir, state) = fresh_state().await; + let user = "bigstaker00000000000000000000000000001"; + seed( + &state, + &[acct(user, to_micro_units(1.0), 1, GENESIS_VALIDATOR_STAKE)], + ) + .await; + { + let mut rewards = state.rewards.write().await; + let _ = rewards.stake(user, GENESIS_VALIDATOR_STAKE, "user-tx"); + } + + let genesis = std::collections::HashSet::new(); // empty — user not a genesis val + state.reconcile_stakes_after_genesis_replace(&genesis).await; + + let inner = state.inner.read().await; + let a = inner.accounts.get(user).unwrap(); + assert_eq!(a.staked, GENESIS_VALIDATOR_STAKE); + assert_eq!(a.balance, to_micro_units(1.0)); + drop(inner); + assert!(state.rewards.read().await.get_stake(user).is_some()); + } + + /// account.staked=0 with a leftover rewards.stakes entry must converge to + /// remove the orphan (the live val-1/val-2 199 RKU ghost symptom). + #[tokio::test] + async fn reconcile_removes_orphaned_rewards_stake_when_account_staked_zero() { + let (_dir, state) = fresh_state().await; + let user = "923cc639b27a6d07cd1f2879166b8a53e01bb8ef"; + seed(&state, &[acct(user, to_micro_units(101.0), 2, 0)]).await; + { + let mut rewards = state.rewards.write().await; + rewards + .stake(user, to_micro_units(199.0), "orphan-tx") + .unwrap(); + } + assert_eq!( + state.rewards.read().await.get_total_staked(), + to_micro_units(199.0) + ); + + state.reconcile_rewards_stakes_from_accounts().await; + + assert!( + state.rewards.read().await.get_stake(user).is_none(), + "orphan rewards.stakes must be removed when account.staked=0" + ); + assert_eq!(state.rewards.read().await.get_total_staked(), 0); + } + + #[tokio::test] + async fn reconcile_creates_rewards_entry_from_account_stake() { + let (_dir, state) = fresh_state().await; + let user = "newstaker0000000000000000000000000001"; + let amount = to_micro_units(150.0); + seed(&state, &[acct(user, 0, 1, amount)]).await; + + state.reconcile_rewards_stakes_from_accounts().await; + + assert_eq!( + state.rewards.read().await.get_stake(user).map(|s| s.amount), + Some(amount) + ); + } + + #[tokio::test] + async fn empty_zero_balance_accounts_are_removed_user_stakers_are_not() { + let (_dir, state) = fresh_state().await; + let empty = "emptyacct0000000000000000000000000001"; + let user = "useracct00000000000000000000000000001"; + seed( + &state, + &[ + acct(empty, 0, 0, 0), + acct(user, to_micro_units(10.0), 1, to_micro_units(100.0)), + ], + ) + .await; + { + let mut rewards = state.rewards.write().await; + let _ = rewards.stake(user, to_micro_units(100.0), "tx"); + } + + let genesis = std::collections::HashSet::new(); + state.reconcile_stakes_after_genesis_replace(&genesis).await; + + let inner = state.inner.read().await; + assert!(inner.accounts.get(empty).is_none()); + assert!(inner.accounts.get(user).is_some()); + assert_eq!( + inner.accounts.get(user).unwrap().staked, + to_micro_units(100.0) + ); + } + + /// End-to-end conservation: after simulated genesis restart cleanup, + /// user (balance + staked) is unchanged and equals pre-cleanup total. + #[tokio::test] + async fn user_funds_conserved_across_simulated_genesis_restart_cleanup() { + let (_dir, state) = fresh_state().await; + let user = "923cc639b27a6d07cd1f2879166b8a53e01bb8ef"; + let v1 = "val10000000000000000000000000000000001"; + let v2 = "val20000000000000000000000000000000002"; + let v3 = "val30000000000000000000000000000000003"; + let user_stake = to_micro_units(199.0); + let user_bal = to_micro_units(101.2324536); + + seed( + &state, + &[ + acct(v1, 0, 0, GENESIS_VALIDATOR_STAKE), + acct(v2, 0, 0, GENESIS_VALIDATOR_STAKE), + acct(v3, 0, 0, GENESIS_VALIDATOR_STAKE), + acct(user, user_bal, 2, user_stake), + ], + ) + .await; + { + let mut rewards = state.rewards.write().await; + for v in [&v1, &v2, &v3] { + let _ = rewards.stake(v, GENESIS_VALIDATOR_STAKE, ""); + } + rewards.stake(user, user_stake, "stake-tx").unwrap(); + } + + let before_total = user_bal + user_stake; + let mut genesis = std::collections::HashSet::new(); + genesis.insert(v1.to_string()); + genesis.insert(v2.to_string()); + genesis.insert(v3.to_string()); + + // Simulate the exact startup path: register genesis stakes (no wipe), + // then reconcile. + { + let mut rewards = state.rewards.write().await; + for v in [&v1, &v2, &v3] { + if rewards.get_stake(v).is_none() { + let _ = rewards.stake(v, GENESIS_VALIDATOR_STAKE, ""); + } + } + } + state.reconcile_stakes_after_genesis_replace(&genesis).await; + state.sync_stakes_to_accounts().await; + + let (bal, staked) = { + let inner = state.inner.read().await; + let a = inner.accounts.get(user).unwrap(); + (a.balance, a.staked) + }; + assert_eq!(bal + staked, before_total, "user funds must be conserved"); + assert_eq!(staked, user_stake); + assert_eq!(bal, user_bal); + + let rewards = state.rewards.read().await; + assert_eq!(rewards.get_stake(user).unwrap().amount, user_stake); + assert_eq!( + rewards.get_total_staked(), + GENESIS_VALIDATOR_STAKE * 3 + user_stake + ); + assert_eq!(rewards.get_active_validators().len(), 4); + } +} diff --git a/packages/rinku-node/src/state/checkpoints.rs b/packages/rinku-node/src/state/checkpoints.rs index 256ed8a..e4b8741 100644 --- a/packages/rinku-node/src/state/checkpoints.rs +++ b/packages/rinku-node/src/state/checkpoints.rs @@ -637,15 +637,7 @@ impl NodeState { self.process_batch_special_txs_with_skip(&all_special_txs, &fast_path_already_finalized) .await; - { - let state = self.inner.read().await; - let mut rewards = self.rewards.write().await; - for (addr, account) in &state.accounts { - if account.staked > 0 { - rewards.sync_stake_amount(addr, account.staked); - } - } - } + self.reconcile_rewards_stakes_from_accounts().await; if has_special && finalized_count > 0 { let mut state = self.inner.write().await; @@ -966,15 +958,7 @@ impl NodeState { self.record_finalized_batch(finalized_count as u64).await; } - { - let state = self.inner.read().await; - let mut rewards = self.rewards.write().await; - for (addr, account) in &state.accounts { - if account.staked > 0 { - rewards.sync_stake_amount(addr, account.staked); - } - } - } + self.reconcile_rewards_stakes_from_accounts().await; { let mut wal = self.wal.lock().await; @@ -1390,15 +1374,7 @@ impl NodeState { self.process_batch_special_txs_with_skip(&all_special_txs, &fast_path_already_finalized) .await; - { - let state = self.inner.read().await; - let mut rewards = self.rewards.write().await; - for (addr, account) in &state.accounts { - if account.staked > 0 { - rewards.sync_stake_amount(addr, account.staked); - } - } - } + self.reconcile_rewards_stakes_from_accounts().await; if has_special && missing_tx_count == 0 { let mut state = self.inner.write().await;