Distinguish max replayed and max observed vote (#16936)

core/src/consensus.rs

@@ -295,6 +295,7 @@ impl Tower {
                     key,
                     last_landed_voted_slot,
                     get_frozen_hash(last_landed_voted_slot),
+                    true,
                 );
             }
 

core/src/latest_validator_votes_for_frozen_banks.rs

@@ -5,7 +5,8 @@ use std::collections::{hash_map::Entry, HashMap};
 #[derive(Default)]
 pub(crate) struct LatestValidatorVotesForFrozenBanks {
     // TODO: Clean outdated/unstaked pubkeys from this list.
-    max_frozen_votes: HashMap<Pubkey, (Slot, Vec<Hash>)>,
+    max_gossip_frozen_votes: HashMap<Pubkey, (Slot, Vec<Hash>)>,
+    max_replay_frozen_votes: HashMap<Pubkey, (Slot, Vec<Hash>)>,
     // Pubkeys that had their `max_frozen_votes` updated since the last
     // fork choice update
     fork_choice_dirty_set: HashMap<Pubkey, (Slot, Vec<Hash>)>,
@@ -19,26 +20,42 @@ impl LatestValidatorVotesForFrozenBanks {
         vote_pubkey: Pubkey,
         vote_slot: Slot,
         frozen_hash: Option<Hash>,
+        is_replay_vote: bool,
     ) -> (bool, Option<Slot>) {
-        let max_frozen_votes_entry = self.max_frozen_votes.entry(vote_pubkey);
+        let vote_map = if is_replay_vote {
+            &mut self.max_replay_frozen_votes
+        } else {
+            &mut self.max_gossip_frozen_votes
+        };
+        let pubkey_max_frozen_votes = vote_map.entry(vote_pubkey);
         if let Some(frozen_hash) = frozen_hash {
-            match max_frozen_votes_entry {
+            match pubkey_max_frozen_votes {
                 Entry::Occupied(mut occupied_entry) => {
                     let (latest_frozen_vote_slot, latest_frozen_vote_hashes) =
                         occupied_entry.get_mut();
                     if vote_slot > *latest_frozen_vote_slot {
+                        if is_replay_vote {
+                            // Only record votes detected through replaying blocks,
+                            // because votes in gossip are not consistently observable
+                            // if the validator is replacing them.
                             self.fork_choice_dirty_set
                                 .insert(vote_pubkey, (vote_slot, vec![frozen_hash]));
+                        }
                         *latest_frozen_vote_slot = vote_slot;
                         *latest_frozen_vote_hashes = vec![frozen_hash];
                         return (true, Some(vote_slot));
                     } else if vote_slot == *latest_frozen_vote_slot
                         && !latest_frozen_vote_hashes.contains(&frozen_hash)
                     {
+                        if is_replay_vote {
+                            // Only record votes detected through replaying blocks,
+                            // because votes in gossip are not consistently observable
+                            // if the validator is replacing them.
                             let (_, dirty_frozen_hashes) =
                                 self.fork_choice_dirty_set.entry(vote_pubkey).or_default();
                             assert!(!dirty_frozen_hashes.contains(&frozen_hash));
                             dirty_frozen_hashes.push(frozen_hash);
+                        }
                         latest_frozen_vote_hashes.push(frozen_hash);
                         return (true, Some(vote_slot));
                     } else {
@@ -49,8 +66,10 @@ impl LatestValidatorVotesForFrozenBanks {
 
                 Entry::Vacant(vacant_entry) => {
                     vacant_entry.insert((vote_slot, vec![frozen_hash]));
+                    if is_replay_vote {
                         self.fork_choice_dirty_set
                             .insert(vote_pubkey, (vote_slot, vec![frozen_hash]));
+                    }
                     return (true, Some(vote_slot));
                 }
             }
@@ -60,7 +79,7 @@ impl LatestValidatorVotesForFrozenBanks {
         // struct
         (
             false,
-            match max_frozen_votes_entry {
+            match pubkey_max_frozen_votes {
                 Entry::Occupied(occupied_entry) => Some(occupied_entry.get().0),
                 Entry::Vacant(_) => None,
             },
@@ -80,14 +99,23 @@ impl LatestValidatorVotesForFrozenBanks {
             })
             .collect()
     }
+
+    #[cfg(test)]
+    fn latest_vote(&self, pubkey: &Pubkey, is_replay_vote: bool) -> Option<&(Slot, Vec<Hash>)> {
+        let vote_map = if is_replay_vote {
+            &self.max_replay_frozen_votes
+        } else {
+            &self.max_gossip_frozen_votes
+        };
+        vote_map.get(pubkey)
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
-    #[test]
+    fn run_test_latest_validator_votes_for_frozen_banks_check_add_vote(is_replay_vote: bool) {
-    fn test_latest_validator_votes_for_frozen_banks_check_add_vote() {
         let mut latest_validator_votes_for_frozen_banks =
             LatestValidatorVotesForFrozenBanks::default();
 
@@ -100,13 +128,17 @@ mod tests {
                 vote_pubkey,
                 vote_slot,
                 frozen_hash,
+                is_replay_vote,
             ),
             // Non-frozen bank isn't inserted, so should return None for
             // the highest voted frozen slot
             (false, None)
         );
         assert!(latest_validator_votes_for_frozen_banks
-            .max_frozen_votes
+            .max_replay_frozen_votes
+            .is_empty());
+        assert!(latest_validator_votes_for_frozen_banks
+            .max_gossip_frozen_votes
             .is_empty());
         assert!(latest_validator_votes_for_frozen_banks
             .fork_choice_dirty_set
@@ -127,16 +159,17 @@ mod tests {
                     vote_pubkey,
                     vote_slot,
                     frozen_hash,
+                    is_replay_vote,
                 ),
                 expected_result
             );
             assert_eq!(
                 *latest_validator_votes_for_frozen_banks
-                    .max_frozen_votes
+                    .latest_vote(&vote_pubkey, is_replay_vote)
-                    .get(&vote_pubkey)
                     .unwrap(),
                 (vote_slot, vec![frozen_hash.unwrap()])
             );
+            if is_replay_vote {
                 assert_eq!(
                     *latest_validator_votes_for_frozen_banks
                         .fork_choice_dirty_set
@@ -144,6 +177,12 @@ mod tests {
                         .unwrap(),
                     (vote_slot, vec![frozen_hash.unwrap()])
                 );
+            } else {
+                assert!(latest_validator_votes_for_frozen_banks
+                    .fork_choice_dirty_set
+                    .get(&vote_pubkey)
+                    .is_none());
+            }
         }
 
         // Case 3: Adding duplicate vote for same slot should update the state
@@ -154,16 +193,17 @@ mod tests {
                 vote_pubkey,
                 vote_slot,
                 duplicate_frozen_hash,
+                is_replay_vote,
             ),
             (true, Some(vote_slot))
         );
         assert_eq!(
             *latest_validator_votes_for_frozen_banks
-                .max_frozen_votes
+                .latest_vote(&vote_pubkey, is_replay_vote)
-                .get(&vote_pubkey)
                 .unwrap(),
             (vote_slot, all_frozen_hashes.clone())
         );
+        if is_replay_vote {
             assert_eq!(
                 *latest_validator_votes_for_frozen_banks
                     .fork_choice_dirty_set
@@ -171,6 +211,12 @@ mod tests {
                     .unwrap(),
                 (vote_slot, all_frozen_hashes.clone())
             );
+        } else {
+            assert!(latest_validator_votes_for_frozen_banks
+                .fork_choice_dirty_set
+                .get(&vote_pubkey)
+                .is_none());
+        }
 
         // Case 4: Adding duplicate vote that is not frozen should not update the state
         let frozen_hash = None;
@@ -179,16 +225,17 @@ mod tests {
                 vote_pubkey,
                 vote_slot,
                 frozen_hash,
+                is_replay_vote,
             ),
             (false, Some(vote_slot))
         );
         assert_eq!(
             *latest_validator_votes_for_frozen_banks
-                .max_frozen_votes
+                .latest_vote(&vote_pubkey, is_replay_vote)
-                .get(&vote_pubkey)
                 .unwrap(),
             (vote_slot, all_frozen_hashes.clone())
         );
+        if is_replay_vote {
             assert_eq!(
                 *latest_validator_votes_for_frozen_banks
                     .fork_choice_dirty_set
@@ -196,6 +243,12 @@ mod tests {
                     .unwrap(),
                 (vote_slot, all_frozen_hashes.clone())
             );
+        } else {
+            assert!(latest_validator_votes_for_frozen_banks
+                .fork_choice_dirty_set
+                .get(&vote_pubkey)
+                .is_none());
+        }
 
         // Case 5: Adding a vote for a new higher slot that is not yet frozen
         // should not update the state
@@ -207,16 +260,17 @@ mod tests {
                 vote_pubkey,
                 vote_slot,
                 frozen_hash,
+                is_replay_vote,
             ),
             (false, Some(old_vote_slot))
         );
         assert_eq!(
             *latest_validator_votes_for_frozen_banks
-                .max_frozen_votes
+                .latest_vote(&vote_pubkey, is_replay_vote)
-                .get(&vote_pubkey)
                 .unwrap(),
             (old_vote_slot, all_frozen_hashes.clone())
         );
+        if is_replay_vote {
             assert_eq!(
                 *latest_validator_votes_for_frozen_banks
                     .fork_choice_dirty_set
@@ -224,6 +278,12 @@ mod tests {
                     .unwrap(),
                 (old_vote_slot, all_frozen_hashes)
             );
+        } else {
+            assert!(latest_validator_votes_for_frozen_banks
+                .fork_choice_dirty_set
+                .get(&vote_pubkey)
+                .is_none());
+        }
 
         // Case 6: Adding a vote for a new higher slot that *is* frozen
         // should upate the state
@@ -233,16 +293,17 @@ mod tests {
                 vote_pubkey,
                 vote_slot,
                 frozen_hash,
+                is_replay_vote,
             ),
             (true, Some(vote_slot))
         );
         assert_eq!(
             *latest_validator_votes_for_frozen_banks
-                .max_frozen_votes
+                .latest_vote(&vote_pubkey, is_replay_vote)
-                .get(&vote_pubkey)
                 .unwrap(),
             (vote_slot, vec![frozen_hash.unwrap()])
         );
+        if is_replay_vote {
             assert_eq!(
                 *latest_validator_votes_for_frozen_banks
                     .fork_choice_dirty_set
@@ -250,6 +311,12 @@ mod tests {
                     .unwrap(),
                 (vote_slot, vec![frozen_hash.unwrap()])
             );
+        } else {
+            assert!(latest_validator_votes_for_frozen_banks
+                .fork_choice_dirty_set
+                .get(&vote_pubkey)
+                .is_none());
+        }
 
         // Case 7: Adding a vote for a new pubkey should also update the state
         vote_slot += 1;
@@ -260,16 +327,17 @@ mod tests {
                 vote_pubkey,
                 vote_slot,
                 frozen_hash,
+                is_replay_vote,
             ),
             (true, Some(vote_slot))
         );
         assert_eq!(
             *latest_validator_votes_for_frozen_banks
-                .max_frozen_votes
+                .latest_vote(&vote_pubkey, is_replay_vote)
-                .get(&vote_pubkey)
                 .unwrap(),
             (vote_slot, vec![frozen_hash.unwrap()])
         );
+        if is_replay_vote {
             assert_eq!(
                 *latest_validator_votes_for_frozen_banks
                     .fork_choice_dirty_set
@@ -277,10 +345,25 @@ mod tests {
                     .unwrap(),
                 (vote_slot, vec![frozen_hash.unwrap()])
             );
+        } else {
+            assert!(latest_validator_votes_for_frozen_banks
+                .fork_choice_dirty_set
+                .get(&vote_pubkey)
+                .is_none());
+        }
     }
 
     #[test]
-    fn test_latest_validator_votes_for_frozen_banks_take_votes_dirty_set() {
+    fn test_latest_validator_votes_for_frozen_banks_check_add_vote_is_replay() {
+        run_test_latest_validator_votes_for_frozen_banks_check_add_vote(true)
+    }
+
+    #[test]
+    fn test_latest_validator_votes_for_frozen_banks_check_add_vote_is_not_replay() {
+        run_test_latest_validator_votes_for_frozen_banks_check_add_vote(false)
+    }
+
+    fn run_test_latest_validator_votes_for_frozen_banks_take_votes_dirty_set(is_replay: bool) {
         let mut latest_validator_votes_for_frozen_banks =
             LatestValidatorVotesForFrozenBanks::default();
         let num_validators = 10;
@@ -296,6 +379,7 @@ mod tests {
                                 vote_pubkey,
                                 vote_slot,
                                 Some(frozen_hash1),
+                                is_replay
                             ),
                             // This vote slot was frozen, and is the highest slot inserted thus far,
                             // so the highest vote should be Some(vote_slot)
@@ -308,15 +392,21 @@ mod tests {
                                 vote_pubkey,
                                 vote_slot,
                                 Some(frozen_hash2),
+                                is_replay
                             ),
                             // This vote slot was frozen, and is for a duplicate version of the highest slot
                             // inserted thus far, so the highest vote should be Some(vote_slot).
                             (true, Some(vote_slot))
                         );
+                        if is_replay {
+                            // Only replayed vote should modify the dirty set, which is used for fork fork choice.
                             vec![
                                 (vote_pubkey, (vote_slot, frozen_hash1)),
                                 (vote_pubkey, (vote_slot, frozen_hash2)),
                             ]
+                        } else {
+                            vec![]
+                        }
                     })
                     .collect()
             };
@@ -334,11 +424,12 @@ mod tests {
             .take_votes_dirty_set(0)
             .is_empty());
 
-        // Taking all the firty votes >= num_validators - 1 will only return the last vote
+        // Taking all the dirty votes >= num_validators - 1 will only return the last vote
         let root = num_validators - 1;
         let dirty_set = setup_dirty_set(&mut latest_validator_votes_for_frozen_banks);
         let mut expected_dirty_set: Vec<(Pubkey, SlotHashKey)> =
-            dirty_set[dirty_set.len() - 2..dirty_set.len()].to_vec();
+            // dirty_set could be empty if `is_replay == false`, so use saturating_sub
+            dirty_set[dirty_set.len().saturating_sub(2)..dirty_set.len()].to_vec();
         let mut votes_dirty_set_output =
             latest_validator_votes_for_frozen_banks.take_votes_dirty_set(root);
         votes_dirty_set_output.sort();
@@ -348,4 +439,79 @@ mod tests {
             .take_votes_dirty_set(0)
             .is_empty());
     }
+
+    #[test]
+    fn test_latest_validator_votes_for_frozen_banks_take_votes_dirty_set_is_replay() {
+        run_test_latest_validator_votes_for_frozen_banks_take_votes_dirty_set(true)
+    }
+
+    #[test]
+    fn test_latest_validator_votes_for_frozen_banks_take_votes_dirty_set_is_not_replay() {
+        run_test_latest_validator_votes_for_frozen_banks_take_votes_dirty_set(false)
+    }
+
+    #[test]
+    fn test_latest_validator_votes_for_frozen_banks_add_replay_and_gossip_vote() {
+        let mut latest_validator_votes_for_frozen_banks =
+            LatestValidatorVotesForFrozenBanks::default();
+
+        // First simulate vote from gossip
+        let vote_pubkey = Pubkey::new_unique();
+        let vote_slot = 1;
+        let frozen_hash = Hash::new_unique();
+        let mut is_replay_vote = false;
+        assert_eq!(
+            latest_validator_votes_for_frozen_banks.check_add_vote(
+                vote_pubkey,
+                vote_slot,
+                Some(frozen_hash),
+                is_replay_vote,
+            ),
+            (true, Some(vote_slot))
+        );
+
+        // Should find the vote in the gossip votes
+        assert_eq!(
+            *latest_validator_votes_for_frozen_banks
+                .latest_vote(&vote_pubkey, is_replay_vote)
+                .unwrap(),
+            (vote_slot, vec![frozen_hash])
+        );
+        // Shouldn't find the vote in the replayed votes
+        assert!(latest_validator_votes_for_frozen_banks
+            .latest_vote(&vote_pubkey, !is_replay_vote)
+            .is_none());
+        assert!(latest_validator_votes_for_frozen_banks
+            .take_votes_dirty_set(0)
+            .is_empty());
+
+        // Next simulate vote from replay
+        is_replay_vote = true;
+        assert_eq!(
+            latest_validator_votes_for_frozen_banks.check_add_vote(
+                vote_pubkey,
+                vote_slot,
+                Some(frozen_hash),
+                is_replay_vote,
+            ),
+            (true, Some(vote_slot))
+        );
+        // Should find the vote in the gossip and replay votes
+        assert_eq!(
+            *latest_validator_votes_for_frozen_banks
+                .latest_vote(&vote_pubkey, is_replay_vote)
+                .unwrap(),
+            (vote_slot, vec![frozen_hash])
+        );
+        assert_eq!(
+            *latest_validator_votes_for_frozen_banks
+                .latest_vote(&vote_pubkey, !is_replay_vote)
+                .unwrap(),
+            (vote_slot, vec![frozen_hash])
+        );
+        assert_eq!(
+            latest_validator_votes_for_frozen_banks.take_votes_dirty_set(0),
+            vec![(vote_pubkey, (vote_slot, frozen_hash))]
+        );
+    }
 }

core/src/replay_stage.rs

@@ -420,12 +420,12 @@ impl ReplayStage {
                     // included in a block, so we may not have yet observed these votes just
                     // by replaying blocks.
                     let mut process_unfrozen_gossip_verified_vote_hashes_time = Measure::start("process_gossip_duplicate_confirmed_slots");
-                    /*Self::process_gossip_verified_vote_hashes(
+                    Self::process_gossip_verified_vote_hashes(
                         &gossip_verified_vote_hash_receiver,
                         &mut unfrozen_gossip_verified_vote_hashes,
                         &heaviest_subtree_fork_choice,
                         &mut latest_validator_votes_for_frozen_banks,
-                    );*/
+                    );
                     for _ in gossip_verified_vote_hash_receiver.try_iter() {}
                     process_unfrozen_gossip_verified_vote_hashes_time.stop();
 
@@ -934,7 +934,6 @@ impl ReplayStage {
         }
     }
 
-    #[cfg(test)]
     fn process_gossip_verified_vote_hashes(
         gossip_verified_vote_hash_receiver: &GossipVerifiedVoteHashReceiver,
         unfrozen_gossip_verified_vote_hashes: &mut UnfrozenGossipVerifiedVoteHashes,
@@ -1765,6 +1764,7 @@ impl ReplayStage {
                             pubkey,
                             bank.slot(),
                             Some(bank_hash),
+                            false,
                         );
                     }
                 }
@@ -4700,12 +4700,11 @@ pub(crate) mod tests {
     }
 
     #[test]
-    fn test_gossip_vote_for_unrooted_slot() {
+    fn test_gossip_vote_doesnt_affect_fork_choice() {
         let VoteSimulator {
             bank_forks,
             mut heaviest_subtree_fork_choice,
             mut latest_validator_votes_for_frozen_banks,
-            mut progress,
             vote_pubkeys,
             ..
         } = setup_forks();
@@ -4714,6 +4713,9 @@ pub(crate) mod tests {
         let mut unfrozen_gossip_verified_vote_hashes = UnfrozenGossipVerifiedVoteHashes::default();
         let (gossip_verified_vote_hash_sender, gossip_verified_vote_hash_receiver) = unbounded();
 
+        // Best slot is 4
+        assert_eq!(heaviest_subtree_fork_choice.best_overall_slot().0, 4);
+
         // Cast a vote for slot 3 on one fork
         let vote_slot = 3;
         let vote_bank = bank_forks.read().unwrap().get(vote_slot).unwrap().clone();
@@ -4727,87 +4729,15 @@ pub(crate) mod tests {
             &mut latest_validator_votes_for_frozen_banks,
         );
 
-        // Pick the best fork
+        // Pick the best fork. Gossip votes shouldn't affect fork choice
         heaviest_subtree_fork_choice.compute_bank_stats(
             &vote_bank,
             &Tower::default(),
             &mut latest_validator_votes_for_frozen_banks,
         );
-        assert_eq!(heaviest_subtree_fork_choice.best_overall_slot().0, 6);
 
-        // Now send another vote for a frozen bank on the other fork, where the new vote
+        // Best slot is still 4
-        // is bigger than the last vote
+        assert_eq!(heaviest_subtree_fork_choice.best_overall_slot().0, 4);
-        let bigger_vote_slot = 4;
-        let bigger_vote_bank = bank_forks
-            .read()
-            .unwrap()
-            .get(bigger_vote_slot)
-            .unwrap()
-            .clone();
-        assert!(heaviest_subtree_fork_choice
-            .contains_block(&(bigger_vote_slot, bigger_vote_bank.hash())));
-        gossip_verified_vote_hash_sender
-            .send((vote_pubkey, bigger_vote_slot, bigger_vote_bank.hash()))
-            .expect("Send should succeed");
-        ReplayStage::process_gossip_verified_vote_hashes(
-            &gossip_verified_vote_hash_receiver,
-            &mut unfrozen_gossip_verified_vote_hashes,
-            &heaviest_subtree_fork_choice,
-            &mut latest_validator_votes_for_frozen_banks,
-        );
-
-        // Now set a root for a slot on the previously voted fork thats smaller than the new vote
-        let new_root = 3;
-        ReplayStage::handle_new_root(
-            new_root,
-            &bank_forks,
-            &mut progress,
-            &AbsRequestSender::default(),
-            None,
-            &mut heaviest_subtree_fork_choice,
-            &mut GossipDuplicateConfirmedSlots::default(),
-            &mut unfrozen_gossip_verified_vote_hashes,
-            &mut true,
-            &mut vec![],
-        );
-
-        // Add a new bank, freeze it
-        let parent_bank = bank_forks.read().unwrap().get(6).unwrap().clone();
-        let new_bank = Bank::new_from_parent(&parent_bank, &Pubkey::default(), 7);
-        bank_forks.write().unwrap().insert(new_bank);
-        let new_bank = bank_forks.read().unwrap().get(7).unwrap().clone();
-        new_bank.freeze();
-        heaviest_subtree_fork_choice.add_new_leaf_slot(
-            (new_bank.slot(), new_bank.hash()),
-            Some((parent_bank.slot(), parent_bank.hash())),
-        );
-
-        // Compute bank stats on new slot
-        heaviest_subtree_fork_choice.compute_bank_stats(
-            &new_bank,
-            &Tower::default(),
-            &mut latest_validator_votes_for_frozen_banks,
-        );
-        // Even though the `bigger_vote_slot` no longer exists in the fork choice tree,
-        // this vote should remove the previous vote's weight because we know there
-        // was a later vote
-        let old_vote_node = (vote_slot, vote_bank.hash());
-        assert_eq!(
-            heaviest_subtree_fork_choice
-                .stake_voted_at(&old_vote_node)
-                .unwrap(),
-            0
-        );
-        assert_eq!(
-            heaviest_subtree_fork_choice
-                .stake_voted_subtree(&old_vote_node)
-                .unwrap(),
-            0
-        );
-        assert_eq!(
-            heaviest_subtree_fork_choice.best_overall_slot(),
-            (new_bank.slot(), new_bank.hash())
-        );
     }
 
     #[test]

core/src/unfrozen_gossip_verified_vote_hashes.rs

@@ -21,8 +21,8 @@ impl UnfrozenGossipVerifiedVoteHashes {
     ) {
         // If this is a frozen bank, then we need to update the `latest_validator_votes_for_frozen_banks`
         let frozen_hash = if is_frozen { Some(hash) } else { None };
-        let (was_added, latest_frozen_vote_slot) =
+        let (was_added, latest_frozen_vote_slot) = latest_validator_votes_for_frozen_banks
-            latest_validator_votes_for_frozen_banks.check_add_vote(pubkey, vote_slot, frozen_hash);
+            .check_add_vote(pubkey, vote_slot, frozen_hash, false);
 
         if !was_added
             && latest_frozen_vote_slot