Announcer Improvements — heartwood

While changing the target logic for the Announcer, I also put Claude, our trusty QA engineer, to work to think of edge cases that were not tested.

I broke up the patch into a series of commits that test certain pieces of logic around the Announcer. Some were just adding coverage, while others surfaced edge cases that were not considered and were fixed in the commit.

 }
 fn print_announcer_result(result: &sync::AnnouncerResult, verbose: bool) {
     use sync::announce::SuccessfulOutcome::*;
     match result {
         sync::AnnouncerResult::Success(success) if verbose => {
             // N.b. Printing how many seeds were synced with is printed
             // elsewhere
             match success.outcome() {
                 sync::announce::SuccessfulOutcome::MinReplicationFactor { preferred, synced }
                 | sync::announce::SuccessfulOutcome::MaxReplicationFactor { preferred, synced } => {
                 MinReplicationFactor { preferred, synced }
                 | MaxReplicationFactor { preferred, synced }
                 | PreferredNodes {
                     preferred,
                     total_nodes_synced: synced,
                 } => {
                     if preferred == 0 {
                         term::success!("Synced {} seed(s)", term::format::positive(synced));
                     } else {

     ///   - [`AnnouncerError::Target`]: the target has no preferred seeds and no
     ///     replicas
     pub fn new(mut config: AnnouncerConfig) -> Result<Self, AnnouncerError> {
         // N.b. ensure that local node is none of the sets
         // N.b. ensure that local node is in none of the sets
         config.preferred_seeds.remove(&config.local_node);
         config.synced.remove(&config.local_node);
         config.unsynced.remove(&config.local_node);
         if config.synced.is_empty() && config.unsynced.is_empty() {
         // N.b extend the unsynced set with any preferred seeds that are not yet
         // synced
         let unsynced_preferred = config
             .preferred_seeds
             .difference(&config.synced)
             .copied()
             .collect::<BTreeSet<_>>();
         config.unsynced.extend(unsynced_preferred);
         // Ensure that the unsynced set does not contain any of the synced set –
         // we trust that the synced nodes are already synced with
         let to_sync = config
             .unsynced
             .difference(&config.synced)
             .copied()
             .collect::<BTreeSet<_>>();
         if config.synced.is_empty() && to_sync.is_empty() {
             return Err(AnnouncerError::NoSeeds);
         }
         if config.unsynced.is_empty() {
         if to_sync.is_empty() {
             let preferred = config.synced.intersection(&config.preferred_seeds).count();
             return Err(AlreadySynced {
                 preferred,

             .into());
         }
         // N.b extend the unsynced set with any preferred seeds that are not yet
         // synced
         config
             .unsynced
             .extend(config.preferred_seeds.difference(&config.synced).copied());
         let replicas = config.replicas.min(config.unsynced.len());
         let replicas = config.replicas.min(to_sync.len());
         let announcer = Self {
             local_node: config.local_node,
             target: Target::new(config.preferred_seeds, replicas)

                 .into_iter()
                 .map(|nid| (nid, SyncStatus::AlreadySynced))
                 .collect(),
             to_sync: config.unsynced,
             to_sync,
         };
         match announcer.is_target_reached() {
             None => Ok(announcer),

                 SuccessfulOutcome::MaxReplicationFactor { preferred, synced } => {
                     Err(AlreadySynced { preferred, synced }.into())
                 }
                 SuccessfulOutcome::PreferredNodes {
                     preferred,
                     total_nodes_synced,
                 } => Err(AlreadySynced {
                     preferred,
                     synced: total_nodes_synced,
                 }
                 .into()),
             },
         }
     }

     /// If there are no more nodes, then [`NoNodes`] is returned in the
     /// [`ControlFlow::Break`], otherwise the [`Announcer`] is returned as-is in
     /// the [`ControlFlow::Continue`].
     // TODO(finto): I'm not sure this is needed with the change to the target
     // logic. Since we can reach the replication factor OR the preferred seeds,
     // AND the replication factor is always capped to the maximum number of
     // seeds to sync with, I don't think we can ever reach a case where
     // `can_continue` hits the `Break`.
     pub fn can_continue(self) -> ControlFlow<NoNodes, Self> {
         if self.to_sync.is_empty() {
             ControlFlow::Break(NoNodes {

     /// Get the [`Progress`] of the [`Announcer`].
     pub fn progress(&self) -> Progress {
         let SuccessCounts { preferred, synced } = self.success_counts();
         let unsynced = self.to_sync.len().saturating_sub(synced);
         let unsynced = self.to_sync.len();
         Progress {
             preferred,
             synced,

     }
     fn is_target_reached(&self) -> Option<SuccessfulOutcome> {
         let SuccessCounts { preferred, synced } = self.success_counts();
         let reached_preferred = self.target.preferred_seeds.is_empty()
             || preferred >= self.target.preferred_seeds.len();
         // It should not be possible to construct a target that has no preferred
         // seeds and set the target to 0
         debug_assert!(self.target.has_preferred_seeds() || self.target.has_replication_factor());
         let replicas = self.target.replicas();
         let min = replicas.lower_bound();
         match replicas.upper_bound() {
             None => (reached_preferred && synced >= min)
                 .then_some(SuccessfulOutcome::MinReplicationFactor { preferred, synced }),
             Some(max) => (reached_preferred && synced >= max)
                 .then_some(SuccessfulOutcome::MaxReplicationFactor { preferred, synced }),
         let SuccessCounts { preferred, synced } = self.success_counts();
         if self.target.has_preferred_seeds() && preferred >= self.target.preferred_seeds.len() {
             Some(SuccessfulOutcome::PreferredNodes {
                 preferred: self.target.preferred_seeds.len(),
                 total_nodes_synced: synced,
             })
         } else {
             // The only target to hit is preferred seeds
             if !self.target.has_replication_factor() {
                 return None;
             }
             let replicas = self.target.replicas();
             let min = replicas.lower_bound();
             match replicas.upper_bound() {
                 None => (synced >= min)
                     .then_some(SuccessfulOutcome::MinReplicationFactor { preferred, synced }),
                 Some(max) => (synced >= max)
                     .then_some(SuccessfulOutcome::MaxReplicationFactor { preferred, synced }),
             }
         }
     }

 }
 /// Configuration of the [`Announcer`].
 #[derive(Clone, Debug)]
 pub struct AnnouncerConfig {
     local_node: NodeId,
     replicas: ReplicationFactor,

             local_node: local,
             replicas,
             preferred_seeds: network.allowed.clone(),
             // TODO(finto): we should check if the seeds are synced with instead
             // of assuming they haven't been yet.
             synced: BTreeSet::new(),
             unsynced: network.allowed,
         }

 }
 /// Error in constructing the [`Announcer`].
 #[derive(Debug)]
 #[derive(Debug, PartialEq, Eq)]
 pub enum AnnouncerError {
     /// Both sets of already synchronized and un-synchronized nodes were empty
     /// of nodes were empty.

     }
 }
 #[derive(Debug)]
 #[derive(Debug, PartialEq, Eq)]
 pub struct AlreadySynced {
     /// The number of preferred nodes that are synchronized.
     preferred: usize,

     }
 }
 #[derive(Debug, thiserror::Error)]
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 #[non_exhaustive]
 #[error("a minimum number of replicas or set of preferred seeds must be provided")]
 pub struct TargetError;

     pub fn replicas(&self) -> &ReplicationFactor {
         &self.replicas
     }
     /// Check if the target has preferred seeds
     pub fn has_preferred_seeds(&self) -> bool {
         !self.preferred_seeds.is_empty()
     }
     /// Check that lower bound of the replication is greater than `0`
     pub fn has_replication_factor(&self) -> bool {
         self.replicas.lower_bound() != 0
     }
 }
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum SuccessfulOutcome {
     MinReplicationFactor { preferred: usize, synced: usize },
     MaxReplicationFactor { preferred: usize, synced: usize },
     MinReplicationFactor {
         preferred: usize,
         synced: usize,
     },
     MaxReplicationFactor {
         preferred: usize,
         synced: usize,
     },
     PreferredNodes {
         preferred: usize,
         total_nodes_synced: usize,
     },
 }
 #[allow(clippy::unwrap_used)]
 #[cfg(test)]
 mod test {
     use crate::test::arbitrary;
     use crate::{assert_matches, test::arbitrary};
     use super::*;
     #[test]
     fn all_synced_nodes_are_preferred_seeds() {
         let local = arbitrary::gen::<NodeId>(0);
         let seeds = arbitrary::set::<NodeId>(5..=5);
         // All preferred seeds, no regular seeds in unsynced
         let preferred_seeds = seeds.iter().take(3).copied().collect::<BTreeSet<_>>();
         let unsynced = preferred_seeds.clone(); // Only preferred seeds to sync with
         let config = AnnouncerConfig::public(
             local,
             // High target that we won't reach with preferred alone
             ReplicationFactor::must_reach(5),
             preferred_seeds.clone(),
             BTreeSet::new(),
             unsynced,
         );
         let mut announcer = Announcer::new(config).unwrap();
         // Sync with all preferred seeds
         let mut synced_count = 0;
         let mut result = None;
         for &node in &preferred_seeds {
             let duration = time::Duration::from_secs(1);
             synced_count += 1;
             match announcer.synced_with(node, duration) {
                 ControlFlow::Continue(progress) => {
                     assert_eq!(
                         progress.preferred(),
                         synced_count,
                         "Preferred count should increment for each preferred seed"
                     );
                     assert_eq!(
                         progress.synced(),
                         synced_count,
                         "Total synced should equal preferred since all are preferred"
                     );
                 }
                 ControlFlow::Break(success) => {
                     result = Some(success);
                     break;
                 }
             }
         }
         assert_eq!(
             result.unwrap().outcome(),
             SuccessfulOutcome::PreferredNodes {
                 preferred: preferred_seeds.len(),
                 total_nodes_synced: preferred_seeds.len()
             },
             "Should succeed with PreferredNodes outcome"
         );
     }
     #[test]
     fn preferred_seeds_already_synced() {
         let local = arbitrary::gen::<NodeId>(0);
         let seeds = arbitrary::set::<NodeId>(6..=6);
         let preferred_seeds = seeds.iter().take(2).copied().collect::<BTreeSet<_>>();
         let already_synced = preferred_seeds.clone(); // Preferred seeds already synced
         let regular_unsynced = seeds.iter().skip(2).copied().collect::<BTreeSet<_>>();
         let config = AnnouncerConfig::public(
             local,
             ReplicationFactor::must_reach(4),
             preferred_seeds.clone(),
             already_synced.clone(),
             regular_unsynced.clone(),
         );
         assert_eq!(
             Announcer::new(config).err(),
             Some(AnnouncerError::AlreadySynced(AlreadySynced {
                 preferred: 2,
                 synced: 2
             }))
         );
     }
     #[test]
     fn announcer_reached_min_replication_target() {
         let local = arbitrary::gen::<NodeId>(0);
         let seeds = arbitrary::set::<NodeId>(10..=10);

         let mut success = None;
         let mut successes = 0;
         for node in preferred_seeds.iter() {
         for node in preferred_seeds.iter().take(1) {
             let t = time::Duration::from_secs(1);
             synced_result.insert(*node, SyncStatus::Synced { duration: t });
             successes += 1;

         assert_eq!(
             success.as_ref().unwrap().outcome(),
             SuccessfulOutcome::MaxReplicationFactor {
                 preferred: 2,
                 preferred: 0,
                 synced: 6,
             }
         )
     }
     #[test]
     fn announcer_must_reach_preferred_seeds() {
     fn announcer_preferred_seeds_or_replica_factor() {
         let local = arbitrary::gen::<NodeId>(0);
         let seeds = arbitrary::set::<NodeId>(10..=10);
         let unsynced = seeds.iter().skip(2).copied().collect::<BTreeSet<_>>();

             unsynced.clone(),
         );
         let mut announcer = Announcer::new(config).unwrap();
         let to_sync = announcer.to_sync();
         assert_eq!(to_sync, unsynced.union(&preferred_seeds).copied().collect());
         let mut synced_result = BTreeMap::new();
         let mut success = None;
         let mut successes = 0;
         // Simulate not being able to reach all nodes
         for node in to_sync.iter() {
         // The preferred seeds then sync, allowing us to reach that part of the
         // target
         for node in preferred_seeds.iter() {
             assert_ne!(*node, local);
             let t = time::Duration::from_secs(1);
             synced_result.insert(*node, SyncStatus::Synced { duration: t });

                 announcer_result = Some(announcer.timed_out());
                 break;
             }
             // Simulate not being able to reach the preferred seeds
             if preferred_seeds.contains(node) {
                 continue;
             }
             let t = time::Duration::from_secs(1);
             synced_result.insert(*node, SyncStatus::Synced { duration: t });
             successes += 1;

             Err(AnnouncerError::AlreadySynced { .. })
         ));
     }
     #[test]
     fn invariant_progress_should_match_state() {
         let local = arbitrary::gen::<NodeId>(0);
         let seeds = arbitrary::set::<NodeId>(6..=6);
         // Set up: 2 already synced, 4 unsynced initially
         let already_synced = seeds.iter().take(2).copied().collect::<BTreeSet<_>>();
         let unsynced = seeds.iter().skip(2).copied().collect::<BTreeSet<_>>();
         let config = AnnouncerConfig::public(
             local,
             ReplicationFactor::must_reach(4), // Need 4 total
             BTreeSet::new(),                  // No preferred seeds
             already_synced.clone(),
             unsynced.clone(),
         );
         let mut announcer = Announcer::new(config).unwrap();
         // No progress made, so values should be the same
         assert_eq!(
             announcer.progress().unsynced(),
             announcer.to_sync().len(),
             "Expected unsynced progress to be the same as the number of nodes to sync"
         );
         // Expected: progress.synced() should be the number of already synced nodes
         assert_eq!(
             announcer.progress().synced(),
             already_synced.len(),
             "Initial synced count should equal already synced nodes"
         );
         // Now sync with one node and check progress again
         let first_unsynced = *unsynced.iter().next().unwrap();
         let duration = time::Duration::from_secs(1);
         match announcer.synced_with(first_unsynced, duration) {
             ControlFlow::Continue(progress) => {
                 assert_eq!(
                     progress.synced(),
                     already_synced.len() + 1,
                     "Synced count should increase by 1"
                 );
                 assert_eq!(
                     progress.unsynced(),
                     announcer.to_sync().len(),
                     "Unsynced count should equal remaining to_sync length"
                 );
                 assert_eq!(
                     progress.unsynced(),
                     unsynced.len() - 1,
                     "Unsynced should be original unsynced count minus nodes we've synced"
                 );
             }
             ControlFlow::Break(outcome) => {
                 panic!("Should not have reached target yet: {outcome:?}")
             }
         }
         // Invariant:
         // synced nodes + unsynced nodes = progress.synced() + progress.unsynced()
         let final_progress = announcer.progress();
         let expected_total = already_synced.len() + unsynced.len();
         let actual_total = final_progress.synced() + final_progress.unsynced();
         assert_eq!(
             actual_total, expected_total,
             "synced + unsynced should equal the total nodes we started with"
         );
     }
     #[test]
     fn local_node_in_preferred_seeds() {
         let local = arbitrary::gen::<NodeId>(0);
         let other_seeds = arbitrary::set::<NodeId>(5..=5);
         // Include local node in preferred seeds
         let mut preferred_seeds = other_seeds.iter().take(2).copied().collect::<BTreeSet<_>>();
         preferred_seeds.insert(local);
         let unsynced = other_seeds.iter().skip(2).copied().collect::<BTreeSet<_>>();
         let config = AnnouncerConfig::public(
             local,
             ReplicationFactor::must_reach(3),
             preferred_seeds.clone(),
             BTreeSet::new(),
             unsynced.clone(),
         );
         let announcer = Announcer::new(config).unwrap();
         // Verify local node was removed from target's preferred seeds
         assert!(
             !announcer.target().preferred_seeds().contains(&local),
             "Local node should be removed from preferred seeds in target"
         );
         // Verify local node is not in to_sync
         assert!(
             !announcer.to_sync().contains(&local),
             "Local node should not be in to_sync set"
         );
         // The preferred seeds in the target should be one less than what we passed in
         assert_eq!(
             announcer.target().preferred_seeds().len(),
             preferred_seeds.len() - 1,
             "Target should have local node removed from preferred seeds"
         );
     }
     #[test]
     fn local_node_in_synced_set() {
         let local = arbitrary::gen::<NodeId>(0);
         let other_seeds = arbitrary::set::<NodeId>(5..=5);
         // Include local node in synced set
         let mut synced = other_seeds.iter().take(2).copied().collect::<BTreeSet<_>>();
         synced.insert(local);
         let unsynced = other_seeds.iter().skip(2).copied().collect::<BTreeSet<_>>();
         let config = AnnouncerConfig::public(
             local,
             ReplicationFactor::must_reach(4),
             BTreeSet::new(),
             synced.clone(),
             unsynced.clone(),
         );
         let announcer = Announcer::new(config).unwrap();
         // Verify local node is not counted in synced nodes
         assert!(
             !announcer.synced.contains_key(&local),
             "Local node should not be in internal synced map"
         );
         // Progress should reflect only the non-local synced nodes
         assert_eq!(
             announcer.progress().synced(),
             synced.len() - 1,
             "Progress should not count local node as synced"
         );
     }
     #[test]
     fn local_node_in_unsynced_set() {
         let local = arbitrary::gen::<NodeId>(0);
         let other_seeds = arbitrary::set::<NodeId>(5..=5);
         let synced = other_seeds.iter().take(2).copied().collect::<BTreeSet<_>>();
         // Include local node in unsynced set
         let mut unsynced = other_seeds.iter().skip(2).copied().collect::<BTreeSet<_>>();
         unsynced.insert(local);
         let config = AnnouncerConfig::public(
             local,
             ReplicationFactor::must_reach(4),
             BTreeSet::new(),
             synced.clone(),
             unsynced.clone(),
         );
         let announcer = Announcer::new(config).unwrap();
         // Verify local node is not in to_sync
         assert!(
             !announcer.to_sync().contains(&local),
             "Local node should not be in to_sync set"
         );
         // The internal to_sync should not contain local node
         assert!(
             !announcer.to_sync.contains(&local),
             "Internal to_sync should not contain local node"
         );
         // Progress unsynced count should not include local node
         assert_eq!(
             announcer.progress().unsynced(),
             unsynced.len() - 1,
             "Progress unsynced should not count local node"
         );
     }
     #[test]
     fn local_node_in_multiple_sets() {
         let local = arbitrary::gen::<NodeId>(0);
         let other_seeds = arbitrary::set::<NodeId>(5..=5);
         // Include local node in ALL sets
         let mut preferred_seeds = other_seeds.iter().take(2).copied().collect::<BTreeSet<_>>();
         preferred_seeds.insert(local);
         let mut synced = other_seeds
             .iter()
             .skip(2)
             .take(1)
             .copied()
             .collect::<BTreeSet<_>>();
         synced.insert(local);
         let mut unsynced = other_seeds.iter().skip(3).copied().collect::<BTreeSet<_>>();
         unsynced.insert(local);
         let config = AnnouncerConfig::public(
             local,
             ReplicationFactor::must_reach(3),
             preferred_seeds.clone(),
             synced.clone(),
             unsynced.clone(),
         );
         let announcer = Announcer::new(config).unwrap();
         // Verify local node is completely absent from all internal structures
         assert!(
             !announcer.target().preferred_seeds().contains(&local),
             "Local node should be removed from preferred seeds"
         );
         assert!(
             !announcer.synced.contains_key(&local),
             "Local node should not be in synced map"
         );
         assert!(
             !announcer.to_sync().contains(&local),
             "Local node should not be in to_sync"
         );
         assert!(
             !announcer.to_sync.contains(&local),
             "Local node should not be in internal to_sync"
         );
         // Verify counts are correct (excluding local node from all)
         assert_eq!(
             announcer.target().preferred_seeds().len(),
             preferred_seeds.len() - 1
         );
         assert_eq!(announcer.progress().synced(), synced.len() - 1);
         // The unsynced nodes includes the preferred seeds, since they are not
         // in the synced set, and `- 1` from each for the local node
         assert_eq!(
             announcer.progress().unsynced(),
             (unsynced.len() - 1) + (preferred_seeds.len() - 1)
         );
     }
     #[test]
     fn synced_with_local_node_is_ignored() {
         let local = arbitrary::gen::<NodeId>(0);
         let unsynced = arbitrary::set::<NodeId>(3..=3).into_iter().collect();
         let config = AnnouncerConfig::public(
             local,
             ReplicationFactor::must_reach(2),
             BTreeSet::new(),
             BTreeSet::new(),
             unsynced,
         );
         let mut announcer = Announcer::new(config).unwrap();
         let initial_progress = announcer.progress();
         // Try to sync with the local node - this should be ignored
         let duration = time::Duration::from_secs(1);
         match announcer.synced_with(local, duration) {
             ControlFlow::Continue(progress) => {
                 // Progress should be unchanged
                 assert_eq!(
                     progress.synced(),
                     initial_progress.synced(),
                     "Syncing with local node should not change synced count"
                 );
                 assert_eq!(
                     progress.unsynced(),
                     initial_progress.unsynced(),
                     "Syncing with local node should not change unsynced count"
                 );
             }
             ControlFlow::Break(_) => panic!("Should not reach target by syncing with local node"),
         }
         // Verify local node is still not in synced map
         assert!(
             !announcer.synced.contains_key(&local),
             "Local node should not be added to synced map"
         );
     }
     #[test]
     fn local_node_only_in_all_sets_results_in_no_seeds_error() {
         let local = arbitrary::gen::<NodeId>(0);
         // Create sets that contain ONLY the local node
         let preferred_seeds = [local].iter().copied().collect::<BTreeSet<_>>();
         let synced = [local].iter().copied().collect::<BTreeSet<_>>();
         let unsynced = [local].iter().copied().collect::<BTreeSet<_>>();
         let config = AnnouncerConfig::public(
             local,
             ReplicationFactor::must_reach(1),
             preferred_seeds,
             synced,
             unsynced,
         );
         // After removing local node from all sets, we should get NoSeeds error
         assert_matches!(Announcer::new(config), Err(AnnouncerError::NoSeeds));
     }
 }