radicle: introduce announcer — heartwood

Similar to the Fetcher sans-IO approach, an Announcer sans-IO approach is introduced. The previous Handler::announce was already quite close to using a sans-IO approach, but using internally held state. This change makes the business logic more reusable – and testable.

Similarly to the Fetcher, the Announcer is configured and then driven by telling it when nodes have been synchronized with. It will continuously check if it has reached its target, yielding to the caller to decide whether to keep going. In contrast, it also provides a way to mark the process as timed out, since announcing relies on waiting for events to come back in a timely manner.

 ``` ~alice (stderr)
 $ git push rad master -f
 ✓ Canonical head updated to 9170c8795d3a78f0381a0ffafb20ea69fb0f5b6b
 ✓ Synced with 1 node(s)
 ✓ Synced with 1 seed(s)
 To rad://z42hL2jL4XNk6K8oHQaSWfMgCL7ji/z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi
  + fb25886...9170c87 master -> master (forced update)
 ```

 warn: could not determine canonical tip for `refs/heads/master`
 warn: no commit found with at least 3 vote(s) (threshold not met)
 warn: it is recommended to find a commit to agree upon
 ✓ Synced with 2 node(s)
 ✓ Synced with 2 seed(s)
 To rad://z42hL2jL4XNk6K8oHQaSWfMgCL7ji/z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi
    d09e634..0f9bd80  master -> master
 ```

 ``` ~bob (stderr)
 $ git push rad
 ✓ Canonical head updated to 3a75f66dd0020c9a0355cc6ec21f15de989e2001
 ✓ Synced with 2 node(s)
 ✓ Synced with 2 seed(s)
 To rad://z42hL2jL4XNk6K8oHQaSWfMgCL7ji/z6Mkt67GdsW7715MEfRuP4pSZxJRJh6kj6Y48WRqVv4N1tRk
 a37862..0f9bd80  master -> master
 ```

 ``` ~eve (stderr)
 $ git push rad
 ✓ Canonical head updated to 0f9bd8035c04b3f73f5408e73e8454879b20800b
 ✓ Synced with 2 node(s)
 ✓ Synced with 2 seed(s)
 To rad://z42hL2jL4XNk6K8oHQaSWfMgCL7ji/z6Mkux1aUQD2voWWukVb5nNUR7thrHveQG4pDQua8nVhib7Z
 a75f66..0f9bd80  master -> master
 ```

 ``` ~alice (stderr)
 $ git push rad
 ✓ Canonical head updated to 319a7dc3b195368ded4b099f8c90bbb80addccd3
 ✓ Synced with 1 node(s)
 ✓ Synced with 1 seed(s)
 To rad://z42hL2jL4XNk6K8oHQaSWfMgCL7ji/z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi
    f2de534..319a7dc  master -> master
 ```

 │ We agreed to add me as a delegate, so I am creating an issue │
 │ to track that work                                           │
 ╰──────────────────────────────────────────────────────────────╯
 ✓ Synced with 1 node(s)
 ✓ Synced with 1 seed(s)
 ```
 and if we inspect Alice's references, then we will see the following:

 ``` ~bob
 $ cd heartwood
 $ rad issue open --title "No license file" --description "..." -q
 ✓ Synced with 1 node(s)
 ✓ Synced with 1 seed(s)
 $ git commit -m "Change copyright" --allow-empty -q
 $ git push rad HEAD:bob/copy
 $ cd ..

 ✗ Error: no seeds found for rad:z2ug5mwNKZB8KGpBDRTrWHAMbvHCu
 ```
 She allows Bob to view the repository. And when she syncs, one node (Bob) gets
 She allows Bob to view the repository. And when she syncs, one seed (Bob) gets
 the refs.
 ``` ~alice
 $ rad id update --title "Allow Bob" --description "" --allow did:key:z6Mkt67GdsW7715MEfRuP4pSZxJRJh6kj6Y48WRqVv4N1tRk -q
 ...
 $ rad sync --announce --timeout 3
 ✓ Synced with 1 node(s)
 ✓ Synced with 1 seed(s)
 ```
 Bob can now fetch the private repo without specifying a seed, because he knows

 ``` ~bob
 $ rad patch review 6c61ef1 --accept -m "LGTM!"
 ✓ Patch 6c61ef1 accepted
 ✓ Synced with 2 node(s)
 ✓ Synced with 2 seed(s)
 $ rad patch show 6c61ef1 -v
 ╭─────────────────────────────────────────────────────────────────────╮
 │ Title    Define LICENSE for project                                 │

 $ git commit --allow-empty -q -m "Changes"
 $ git push rad HEAD:refs/patches
 ✓ Patch acab0ec777a97d013f30be5d5d1aec32562ecb02 opened
 ✓ Synced with 1 node(s)
 ✓ Synced with 1 seed(s)
   https://app.radicle.xyz/nodes/[..]/rad:z3yXbb1sR6UG6ixxV2YF9jUP7ABra/patches/acab0ec777a97d013f30be5d5d1aec32562ecb02

 $ git commit --allow-empty -m "Bob's commit #1" -q
 $ git push rad -o sync -o patch.message="Bob's patch" HEAD:refs/patches
 ✓ Patch 55b9721ed7f6bfec38f43729e9b6631c5dc812fb opened
 ✓ Synced with 1 node(s)
 ✓ Synced with 1 seed(s)
 To rad://zhbMU4DUXrzB8xT6qAJh6yZ7bFMK/z6Mkt67GdsW7715MEfRuP4pSZxJRJh6kj6Y48WRqVv4N1tRk
  * [new reference]   HEAD -> refs/patches
 ```

 ```
 $ rad sync --announce
 ✓ Nothing to announce, already in sync with 2 node(s) (see `rad sync status`)
 ✓ Nothing to announce, already in sync with 2 seed(s) (see `rad sync status`)
 ```
 We can also use the `--fetch` option to only fetch objects:

 ✓ Target met: 2 seed(s)
 🌱 Fetched from z6Mkux1…nVhib7Z
 🌱 Fetched from z6Mkt67…v4N1tRk
 ✓ Nothing to announce, already in sync with 2 node(s) (see `rad sync status`)
 ✓ Nothing to announce, already in sync with 2 seed(s) (see `rad sync status`)
 ```
 It's also possible to use the `--seed` flag to only sync with a specific node:

 ```
 $ rad patch review e4934b6 --revision 9d62420 --accept
 ✓ Patch e4934b6 accepted
 ✓ Synced with 1 node(s)
 ✓ Synced with 1 seed(s)
 $ git checkout master
 Your branch is up to date with 'rad/master'.
 $ git merge patch/e4934b6

 use std::cmp::Ordering;
 use std::collections::BTreeMap;
 use std::collections::BTreeSet;
 use std::collections::HashSet;
 use std::ffi::OsString;

         );
     }
     match options.op {
     match &options.op {
         Operation::Status => {
             let rid = match options.rid {
                 Some(rid) => rid,

             };
             let settings = settings.clone().with_profile(&profile);
             if [SyncDirection::Fetch, SyncDirection::Both].contains(&direction) {
             if [SyncDirection::Fetch, SyncDirection::Both].contains(direction) {
                 if !profile.policies()?.is_seeding(&rid)? {
                     anyhow::bail!("repository {rid} is not seeded");
                 }
                 let result = fetch(rid, settings.clone(), &mut node, &profile)?;
                 display_fetch_result(&result, options.verbose)
             }
             if [SyncDirection::Announce, SyncDirection::Both].contains(&direction) {
                 announce_refs(rid, settings, options.debug, &mut node, &profile)?;
             if [SyncDirection::Announce, SyncDirection::Both].contains(direction) {
                 announce_refs(rid, settings, &mut node, &profile, &options)?;
             }
         }
         Operation::Synchronize(SyncMode::Inventory) => {

 fn announce_refs(
     rid: RepoId,
     settings: SyncSettings,
     debug: bool,
     node: &mut Node,
     profile: &Profile,
     options: &Options,
 ) -> anyhow::Result<()> {
     let Ok(repo) = profile.storage.repository(rid) else {
         return Err(anyhow!(

         }
     }
     crate::node::announce(
     let result = crate::node::announce(
         &repo,
         settings,
         SyncReporting {
             debug,
             debug: options.debug,
             ..SyncReporting::default()
         },
         node,
         profile,
     )?;
     if let Some(result) = result {
         print_announcer_result(&result, options.verbose)
     }
     Ok(())
 }

     let local = profile.id();
     let is_private = profile.storage.repository(rid).ok().and_then(|repo| {
         let doc = repo.identity_doc().ok()?.doc;
         sync::fetch::PrivateNetwork::private_repo(&doc)
         sync::PrivateNetwork::private_repo(&doc)
     });
     let config = match is_private {
         Some(private) => sync::FetcherConfig::private(private, settings.replicas, *local),

         }
     }
 }
 fn print_announcer_result(result: &sync::AnnouncerResult, verbose: bool) {
     match result {
         sync::AnnouncerResult::Success(success) => {
             // N.b. Printing how many seeds were synced with is printed
             // elsewhere
             if verbose {
                 match success.outcome() {
                     sync::announce::SuccessfulOutcome::MinReplicationFactor {
                         preferred,
                         synced,
                     } => {
                         if preferred == 0 {
                             term::success!("Synced {} seed(s)", term::format::positive(synced));
                         } else {
                             term::success!(
                                 "Synced {} preferred seed(s) and {} total seed(s)",
                                 term::format::positive(preferred),
                                 term::format::positive(synced)
                             );
                         }
                     }
                     sync::announce::SuccessfulOutcome::MaxReplicationFactor {
                         preferred,
                         synced,
                     } => {
                         if preferred == 0 {
                             term::success!("Synced {} seed(s)", term::format::positive(synced));
                         } else {
                             term::success!(
                                 "Synced {} preferred seed(s) and {} total seed(s)",
                                 term::format::positive(preferred),
                                 term::format::positive(synced)
                             );
                         }
                     }
                 }
                 print_synced(success.synced());
             }
         }
         sync::AnnouncerResult::TimedOut(result) => {
             if result.synced().is_empty() {
                 term::error("All seeds timed out, use `rad sync -v` to see the list of seeds");
                 return;
             }
             let timed_out = result.timed_out();
             term::warning(format!(
                 "{} seed(s) timed out, use `rad sync -v` to see the list of seeds",
                 timed_out.len(),
             ));
             if verbose {
                 print_synced(result.synced());
                 for node in timed_out {
                     term::warning(format!("{} timed out", term::format::node(node)));
                 }
             }
         }
         sync::AnnouncerResult::NoNodes(result) => {
             term::info!("Announcement could not sync with anymore seeds.");
             if verbose {
                 print_synced(result.synced())
             }
         }
     }
 }
 fn print_synced(synced: &BTreeMap<NodeId, sync::announce::SyncStatus>) {
     for (node, status) in synced.iter() {
         let mut message = format!("🌱 Synced with {}", term::format::node(node));
         match status {
             sync::announce::SyncStatus::AlreadySynced => {
                 message.push_str(&format!("{}", term::format::dim(" (already in sync)")));
             }
             sync::announce::SyncStatus::Synced { duration } => {
                 message.push_str(&format!(
                     "{}",
                     term::format::dim(format!("in {}s", duration.as_secs()))
                 ));
             }
         }
         term::info!("{}", message);
     }
 }

 use std::collections::BTreeSet;
 use std::io;
 use std::io::Write;
 use std::ops::ControlFlow;
 use radicle::node::{self, sync, AnnounceResult};
 use radicle::node::sync;
 use radicle::node::{Handle as _, NodeId};
 use radicle::storage::{ReadRepository, RepositoryError};
 use radicle::{Node, Profile};
 use radicle_term::format;
 use crate::terminal as term;

     Node(#[from] radicle::node::Error),
     #[error("all seeds timed out")]
     AllSeedsTimedOut,
     #[error(transparent)]
     Target(#[from] sync::announce::TargetError),
 }
 impl SyncError {

     reporting: SyncReporting,
     node: &mut Node,
     profile: &Profile,
 ) -> Result<AnnounceResult, SyncError> {
 ) -> Result<Option<sync::AnnouncerResult>, SyncError> {
     match announce_(repo, settings, reporting, node, profile) {
         Ok(result) => Ok(result),
         Err(e) if e.is_connection_err() => {
             term::hint("Node is stopped. To announce changes to the network, start it with `rad node start`.");
             Ok(AnnounceResult::default())
             Ok(None)
         }
         Err(e) => Err(e),
     }
 }
 fn announce_<R: ReadRepository>(
 fn announce_<R>(
     repo: &R,
     settings: SyncSettings,
     mut reporting: SyncReporting,
     node: &mut Node,
     profile: &Profile,
 ) -> Result<AnnounceResult, SyncError> {
 ) -> Result<Option<sync::AnnouncerResult>, SyncError>
 where
     R: ReadRepository,
 {
     let me = profile.id();
     let rid = repo.id();
     let doc = repo.identity_doc()?;
     let mut settings = settings.with_profile(profile);
     let unsynced: Vec<_> = if doc.is_public() {
         // All seeds.
         let all = node.seeds(rid)?;
         if all.is_empty() {
             term::info!(&mut reporting.completion; "No seeds found for {rid}.");
             return Ok(AnnounceResult::default());
         }
         // Seeds in sync with us.
         let synced = all
             .iter()
             .filter(|s| s.is_synced())
             .map(|s| s.nid)
             .collect::<BTreeSet<_>>();
         // Replicas not counting our local replica.
         let replicas = synced.iter().filter(|nid| *nid != profile.id()).count();
         // Maximum replication factor we can achieve.
         let max_replicas = all.iter().filter(|s| &s.nid != profile.id()).count();
         // If the seeds we specified in the sync settings are all synced.
         let is_seeds_synced = settings.seeds.iter().all(|s| synced.contains(s));
         // If we met our desired replica count. Note that this can never exceed the maximum count.
         let is_replicas_synced = replicas >= settings.replicas.lower_bound().min(max_replicas);
         // Nothing to do if we've met our sync state.
         if is_seeds_synced && is_replicas_synced {
             term::success!(
                 &mut reporting.completion;
                 "Nothing to announce, already in sync with {replicas} node(s) (see `rad sync status`)"
     let settings = settings.with_profile(profile);
     let n_preferred_seeds = settings.seeds.len();
     let config = match sync::PrivateNetwork::private_repo(&doc) {
         None => {
             let (synced, unsynced) = node.seeds(rid)?.iter().fold(
                 (BTreeSet::new(), BTreeSet::new()),
                 |(mut synced, mut unsynced), seed| {
                     if seed.is_synced() {
                         synced.insert(seed.nid);
                     } else {
                         unsynced.insert(seed.nid);
                     }
                     (synced, unsynced)
                 },
             );
             return Ok(AnnounceResult::default());
             sync::AnnouncerConfig::public(*me, settings.replicas, settings.seeds, synced, unsynced)
         }
         Some(network) => {
             let sessions = node.sessions()?;
             let network =
                 network.restrict(|nid| sessions.iter().any(|s| s.nid == *nid && s.is_connected()));
             sync::AnnouncerConfig::private(*me, settings.replicas, network)
         }
         // Return nodes we can announce to. They don't have to be connected directly.
         all.iter()
             .filter(|s| !s.is_synced() && &s.nid != profile.id())
             .map(|s| s.nid)
             .collect()
     } else {
         node.sessions()?
             .into_iter()
             .filter(|s| s.state.is_connected() && doc.is_visible_to(&s.nid.into()))
             .map(|s| s.nid)
             .collect()
     };
     if unsynced.is_empty() {
         term::info!(&mut reporting.completion; "No seeds to announce to for {rid}. (see `rad sync status`)");
         return Ok(AnnounceResult::default());
     }
     // Cap the replicas to the maximum achievable.
     // Nb. It's impossible to know if a replica follows our node. This means that if we announce
     // only our refs, and the replica doesn't follow us, it won't fetch from us.
     settings.replicas = settings.replicas.min(unsynced.len());
     let announcer = match sync::Announcer::new(config) {
         Ok(announcer) => announcer,
         Err(err) => match err {
             sync::AnnouncerError::AlreadySynced(result) => {
                 term::success!(
                     &mut reporting.completion;
                     "Nothing to announce, already in sync with {} seed(s) (see `rad sync status`)",
                     term::format::positive(result.synced()),
                 );
                 return Ok(None);
             }
             sync::AnnouncerError::NoSeeds => {
                 term::info!(
                     &mut reporting.completion;
                     "{}",
                     term::format::yellow("No seeds found for {rid}.")
                 );
                 return Ok(None);
             }
             sync::AnnouncerError::Target(err) => return Err(err.into()),
         },
     };
     let target = announcer.target();
     let min_replicas = target.replicas().lower_bound();
     let mut spinner = term::spinner_to(
         format!("Found {} seed(s)..", unsynced.len()),
         format!("Found {} seed(s)..", announcer.progress().unsynced()),
         reporting.completion.clone(),
         reporting.progress.clone(),
     );
     let result = node.announce(
         rid,
         unsynced,
         settings.timeout,
         |event, replicas| match event {
             node::AnnounceEvent::Announced => ControlFlow::Continue(()),
             node::AnnounceEvent::RefsSynced { remote, time } => {
                 spinner.message(format!(
                     "Synced with {} in {}..",
                     format::dim(remote),
                     format::dim(format!("{time:?}"))
                 ));
                 // We're done syncing when both of these conditions are met:
                 //
                 // 1. We've matched or exceeded our target replica count.
                 // 2. We've synced with one of the seeds specified manually.
                 if replicas.len() >= settings.replicas.lower_bound()
                     && (settings.seeds.is_empty()
                         || settings.seeds.iter().any(|s| replicas.contains_key(s)))
                 {
                     ControlFlow::Break(())
                 } else {
                     ControlFlow::Continue(())
                 }
             }
         },
     )?;
     if result.synced.is_empty() {
         spinner.failed();
     } else {
         spinner.message(format!("Synced with {} node(s)", result.synced.len()));
         spinner.finish();
         if reporting.debug {
             for (seed, time) in &result.synced {
                 writeln!(
                     &mut reporting.completion,
                     "  {}",
                     term::format::dim(format!("Synced with {seed} in {time:?}")),
                 )
                 .ok();
             }
     match node.announce(rid, settings.timeout, announcer, |node, progress| {
         spinner.message(format!(
             "Synced with {}, {} of {} preferred seeds, and {} of at least {} replica(s).",
             term::format::node(node),
             term::format::secondary(progress.preferred()),
             term::format::secondary(n_preferred_seeds),
             term::format::secondary(progress.synced()),
             term::format::secondary(min_replicas),
         ));
     }) {
         Ok(result) => {
             spinner.message(format!(
                 "Synced with {} seed(s)",
                 term::format::positive(result.synced().len())
             ));
             spinner.finish();
             Ok(Some(result))
         }
     }
     for seed in &result.timed_out {
         if settings.seeds.contains(seed) {
             term::notice!(&mut reporting.completion; "Seed {seed} timed out..");
         Err(err) => {
             spinner.error("Sync failed: {err}");
             Err(err.into())
         }
     }
     if result.synced.is_empty() {
         return Err(SyncError::AllSeedsTimedOut);
     }
     Ok(result)
 }

         .follow(seed.id, Some(Alias::new("seed")))
         .unwrap();
     alice.connect(&seed);
     bob.connect(&seed).connect(&alice);
     alice.connect(&seed).connect(&bob);
     bob.connect(&seed);
     alice.routes_to(&[(acme, seed.id)]);
     seed.handle.fetch(acme, alice.id, DEFAULT_TIMEOUT).unwrap();

     let mut urls = Vec::new();
     for seed in profile.config.preferred_seeds.iter() {
         if result.synced(&seed.id).is_some() {
             for resource in updated {
                 let url = profile
                     .config
                     .public_explorer
                     .url(seed.addr.host.clone(), repo.id)
                     .resource(resource);
                 urls.push(url);
     if let Some(result) = result {
         for seed in profile.config.preferred_seeds.iter() {
             if result.is_synced(&seed.id) {
                 for resource in updated {
                     let url = profile
                         .config
                         .public_explorer
                         .url(seed.addr.host.clone(), repo.id)
                         .resource(resource);
                     urls.push(url);
                 }
                 break;
             }
             break;
         }
     }

     }
 }
 /// Announcement result returned by [`Node::announce`].
 #[derive(Debug, Default)]
 pub struct AnnounceResult {
     /// Nodes that timed out.
     pub timed_out: Vec<NodeId>,
     /// Nodes that synced.
     pub synced: Vec<(NodeId, time::Duration)>,
 }
 impl AnnounceResult {
     /// Check if a node synced successfully.
     pub fn synced(&self, nid: &NodeId) -> Option<time::Duration> {
         self.synced
             .iter()
             .find(|(id, _)| id == nid)
             .map(|(_, time)| *time)
     }
 }
 /// A sync event, emitted by [`Node::announce`].
 #[derive(Debug)]
 pub enum AnnounceEvent {
     /// Refs were synced with the given node.
     RefsSynced {
         remote: NodeId,
         time: time::Duration,
     },
     /// Refs were announced to all given nodes.
     Announced,
 }
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(tag = "status", rename_all = "camelCase")]
 pub enum FetchResult {

     pub fn announce(
         &mut self,
         rid: RepoId,
         seeds: impl IntoIterator<Item = NodeId>,
         timeout: time::Duration,
         mut callback: impl FnMut(AnnounceEvent, &HashMap<PublicKey, time::Duration>) -> ControlFlow<()>,
     ) -> Result<AnnounceResult, Error> {
         let events = self.subscribe(timeout)?;
         mut announcer: sync::Announcer,
         mut report: impl FnMut(&NodeId, sync::announce::Progress),
     ) -> Result<sync::AnnouncerResult, Error> {
         let mut events = self.subscribe(timeout)?;
         let refs = self.announce_refs(rid)?;
         let mut unsynced = seeds.into_iter().collect::<BTreeSet<_>>();
         let mut synced = HashMap::new();
         let mut timed_out: Vec<NodeId> = Vec::new();
         let started = time::Instant::now();
         callback(AnnounceEvent::Announced, &synced);
         for e in events {
         loop {
             let Some(e) = events.next() else {
                 // Consider the announcement as timed out if there are no more
                 // events
                 return Ok(announcer.timed_out());
             };
             let elapsed = started.elapsed();
             if elapsed >= timeout {
                 timed_out.extend(unsynced.iter());
                 break;
                 return Ok(announcer.timed_out());
             }
             match e {
                 Ok(Event::RefsSynced {

                     at,
                 }) if rid == rid_ && refs.at == at => {
                     log::debug!(target: "radicle", "Received {e:?}");
                     unsynced.remove(&remote);
                     // We can receive synced events from nodes we didn't directly announce to,
                     // and it's possible to receive duplicates as well.
                     if synced.insert(remote, elapsed).is_none() {
                         let event = AnnounceEvent::RefsSynced {
                             remote,
                             time: elapsed,
                         };
                         if callback(event, &synced).is_break() {
                             break;
                     match announcer.synced_with(remote, elapsed) {
                         ControlFlow::Continue(progress) => {
                             report(&remote, progress);
                         }
                         ControlFlow::Break(finished) => {
                             return Ok(finished.into());
                         }
                     }
                 }
                 Ok(_) => {}
                 Err(Error::TimedOut) => {
                     timed_out.extend(unsynced.iter());
                     break;
                     return Ok(announcer.timed_out());
                 }
                 Err(e) => return Err(e),
             }
             if unsynced.is_empty() {
                 break;
             }
             // Ensure that the announcer is still waiting for nodes to be
             // in-sync with
             announcer = match announcer.can_continue() {
                 ControlFlow::Continue(cont) => cont,
                 ControlFlow::Break(finished) => return Ok(finished.into()),
             };
         }
         Ok(AnnounceResult {
             timed_out,
             synced: synced.into_iter().collect(),
         })
     }
 }

 pub mod fetch;
 pub mod announce;
 pub use announce::{Announcer, AnnouncerConfig, AnnouncerError, AnnouncerResult};
 pub mod fetch;
 pub use fetch::{Fetcher, FetcherConfig, FetcherError, FetcherResult};
 use std::collections::BTreeSet;
 use crate::identity::Visibility;
 use crate::prelude::Doc;
 use super::NodeId;
 /// A set of nodes that form a private network for fetching from.
 ///
 /// This could be the set of allowed nodes for a private repository, using
 /// [`PrivateNetwork::private_repo`]
 pub struct PrivateNetwork {
     allowed: BTreeSet<NodeId>,
 }
 impl PrivateNetwork {
     pub fn private_repo(doc: &Doc) -> Option<Self> {
         match doc.visibility() {
             Visibility::Public => None,
             Visibility::Private { allow } => {
                 let allowed = doc
                     .delegates()
                     .iter()
                     .chain(allow.iter())
                     .map(|did| *did.as_key())
                     .collect();
                 Some(Self { allowed })
             }
         }
     }
     /// Restrict the set of allowed nodes based on the `predicate`, where `true`
     /// keeps the `NodeId` in the allowed set.
     ///
     /// For example, this can be useful to restrict the set to only connected
     /// nodes.
     pub fn restrict<P>(self, predicate: P) -> Self
     where
         P: FnMut(&NodeId) -> bool,
     {
         Self {
             allowed: self.allowed.into_iter().filter(predicate).collect(),
         }
     }
 }
 /// The replication factor of a syncing operation.
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
 pub enum ReplicationFactor {

 use std::{
     collections::{BTreeMap, BTreeSet},
     ops::ControlFlow,
     time,
 };
 use crate::node::NodeId;
 use super::{PrivateNetwork, ReplicationFactor};
 pub struct Announcer {
     local_node: NodeId,
     target: Target,
     synced: BTreeMap<NodeId, SyncStatus>,
     to_sync: BTreeSet<NodeId>,
 }
 impl Announcer {
     /// Construct a new [`Announcer`] from the [`AnnouncerConfig`].
     ///
     /// This will ensure that the local [`NodeId`], provided in the
     /// [`AnnouncerConfig`], will be removed from all sets.
     ///
     /// # Errors
     ///
     /// Returns the following errors:
     ///
     ///   - [`AnnouncerError::NoSeeds`]: both sets of already synchronized and
     ///     un-synchronized nodes were empty
     ///     of nodes were empty
     ///   - [`AnnouncerError::AlreadySynced`]: no more nodes are available for
     ///     synchronizing with
     ///   - [`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
         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() {
             return Err(AnnouncerError::NoSeeds);
         }
         if config.unsynced.is_empty() {
             let preferred = config.synced.intersection(&config.preferred_seeds).count();
             return Err(AlreadySynced {
                 preferred,
                 synced: config.synced.len(),
             }
             .into());
         }
         let replicas = config.replicas.min(config.unsynced.len());
         let announcer = Self {
             local_node: config.local_node,
             target: Target::new(config.preferred_seeds, replicas)
                 .map_err(AnnouncerError::Target)?,
             synced: config
                 .synced
                 .into_iter()
                 .map(|nid| (nid, SyncStatus::AlreadySynced))
                 .collect(),
             to_sync: config.unsynced,
         };
         match announcer.is_target_reached() {
             None => Ok(announcer),
             Some(outcome) => match outcome {
                 SuccessfulOutcome::MinReplicationFactor { preferred, synced } => {
                     Err(AlreadySynced { preferred, synced }.into())
                 }
                 SuccessfulOutcome::MaxReplicationFactor { preferred, synced } => {
                     Err(AlreadySynced { preferred, synced }.into())
                 }
             },
         }
     }
     /// Mark the `node` as synchronized, with the given `duration` it took to
     /// synchronize with.
     ///
     /// If the target for the [`Announcer`] has been reached, then a [`Success`] is
     /// returned via [`ControlFlow::Break`]. Otherwise, [`Progress`] is returned
     /// via [`ControlFlow::Continue`].
     ///
     /// The caller decides whether they wish to continue the announcement process.
     pub fn synced_with(
         &mut self,
         node: NodeId,
         duration: time::Duration,
     ) -> ControlFlow<Success, Progress> {
         if node == self.local_node {
             return ControlFlow::Continue(self.progress());
         }
         self.to_sync.remove(&node);
         self.synced.insert(node, SyncStatus::Synced { duration });
         self.finished()
     }
     /// Complete the [`Announcer`] process returning a [`AnnouncerResult`].
     ///
     /// If the target for the [`Announcer`] has been reached, then the result
     /// will be [`AnnouncerResult::Success`], otherwise, it will be
     /// [`AnnouncerResult::TimedOut`].
     pub fn timed_out(self) -> AnnouncerResult {
         match self.is_target_reached() {
             None => TimedOut {
                 synced: self.synced,
                 timed_out: self.to_sync,
             }
             .into(),
             Some(outcome) => Success {
                 outcome,
                 synced: self.synced,
             }
             .into(),
         }
     }
     /// Check if the [`Announcer`] can continue synchronizing with more nodes.
     /// 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`].
     pub fn can_continue(self) -> ControlFlow<NoNodes, Self> {
         if self.to_sync.is_empty() {
             ControlFlow::Break(NoNodes {
                 synced: self.synced,
             })
         } else {
             ControlFlow::Continue(self)
         }
     }
     /// Get all the nodes to be synchronized with.
     pub fn to_sync(&self) -> BTreeSet<NodeId> {
         self.to_sync
             .iter()
             .filter(|node| *node != &self.local_node)
             .copied()
             .collect()
     }
     /// Get the [`Target`] of the [`Announcer`].
     pub fn target(&self) -> &Target {
         &self.target
     }
     /// Get the [`Progress`] of the [`Announcer`].
     pub fn progress(&self) -> Progress {
         let (synced, preferred) = self.success_counts();
         let unsynced = self.to_sync.len().saturating_sub(synced);
         Progress {
             preferred,
             synced,
             unsynced,
         }
     }
     fn finished(&self) -> ControlFlow<Success, Progress> {
         let progress = self.progress();
         self.is_target_reached()
             .map_or(ControlFlow::Continue(progress), |outcome| {
                 ControlFlow::Break(Success {
                     outcome,
                     synced: self.synced.clone(),
                 })
             })
     }
     fn is_target_reached(&self) -> Option<SuccessfulOutcome> {
         let (preferred, synced) = self.success_counts();
         let reached_preferred = self.target.preferred_seeds.is_empty()
             || preferred >= self.target.preferred_seeds.len();
         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 }),
         }
     }
     fn success_counts(&self) -> (usize, usize) {
         self.synced
             .keys()
             .fold((0, 0), |(mut preferred, mut succeeded), nid| {
                 succeeded += 1;
                 if self.target.preferred_seeds.contains(nid) {
                     preferred += 1;
                 }
                 (preferred, succeeded)
             })
     }
 }
 /// Configuration of the [`Announcer`].
 pub struct AnnouncerConfig {
     local_node: NodeId,
     replicas: ReplicationFactor,
     preferred_seeds: BTreeSet<NodeId>,
     synced: BTreeSet<NodeId>,
     unsynced: BTreeSet<NodeId>,
 }
 impl AnnouncerConfig {
     /// Setup a private network `AnnouncerConfig`, populating the
     /// [`AnnouncerConfig`]'s preferred seeds with the allowed set from the
     /// [`PrivateNetwork`].
     ///
     /// `replicas` is the target number of seeds the [`Announcer`] should reach
     /// before stopping.
     ///
     /// `local` is the [`NodeId`] of the local node, to ensure it is
     /// excluded from the [`Announcer`] process.
     pub fn private(local: NodeId, replicas: ReplicationFactor, network: PrivateNetwork) -> Self {
         AnnouncerConfig {
             local_node: local,
             replicas,
             preferred_seeds: network.allowed.clone(),
             synced: BTreeSet::new(),
             unsynced: network.allowed,
         }
     }
     /// Setup a public `AnnouncerConfig`.
     ///
     /// `preferred_seeds` is the target set of preferred seeds that [`Announcer`] should
     /// attempt to synchronize with.
     ///
     /// `synced` and `unsynced` are the set of nodes that are currently
     /// synchronized and un-synchronized with, respectively.
     ///
     /// `replicas` is the target number of seeds the [`Announcer`] should reach
     /// before stopping.
     ///
     /// `local` is the [`NodeId`] of the local node, to ensure it is
     /// excluded from the [`Announcer`] process.
     pub fn public(
         local: NodeId,
         replicas: ReplicationFactor,
         preferred_seeds: BTreeSet<NodeId>,
         synced: BTreeSet<NodeId>,
         unsynced: BTreeSet<NodeId>,
     ) -> Self {
         Self {
             local_node: local,
             replicas,
             preferred_seeds,
             synced,
             unsynced,
         }
     }
 }
 /// Result of running an [`Announcer`] process.
 pub enum AnnouncerResult {
     /// The target of the [`Announcer`] was successfully met.
     Success(Success),
     /// The [`Announcer`] process was timed out, and all un-synchronized nodes
     /// are marked as timed out.
     ///
     /// Note that some nodes still may have synchronized.
     TimedOut(TimedOut),
     /// The [`Announcer`] ran out of nodes to synchronize with.
     ///
     /// Note that some nodes still may have synchronized.
     NoNodes(NoNodes),
 }
 impl AnnouncerResult {
     /// Get the synchronized nodes, regardless of the result.
     pub fn synced(&self) -> &BTreeMap<NodeId, SyncStatus> {
         match self {
             AnnouncerResult::Success(Success { synced, .. }) => synced,
             AnnouncerResult::TimedOut(TimedOut { synced, .. }) => synced,
             AnnouncerResult::NoNodes(NoNodes { synced }) => synced,
         }
     }
     /// Check if a given node is synchronized with.
     pub fn is_synced(&self, node: &NodeId) -> bool {
         let synced = self.synced();
         synced.contains_key(node)
     }
 }
 impl From<Success> for AnnouncerResult {
     fn from(s: Success) -> Self {
         Self::Success(s)
     }
 }
 impl From<TimedOut> for AnnouncerResult {
     fn from(to: TimedOut) -> Self {
         Self::TimedOut(to)
     }
 }
 impl From<NoNodes> for AnnouncerResult {
     fn from(no: NoNodes) -> Self {
         Self::NoNodes(no)
     }
 }
 pub struct NoNodes {
     synced: BTreeMap<NodeId, SyncStatus>,
 }
 impl NoNodes {
     /// Get the set of synchronized nodes
     pub fn synced(&self) -> &BTreeMap<NodeId, SyncStatus> {
         &self.synced
     }
 }
 pub struct TimedOut {
     synced: BTreeMap<NodeId, SyncStatus>,
     timed_out: BTreeSet<NodeId>,
 }
 impl TimedOut {
     /// Get the set of synchronized nodes
     pub fn synced(&self) -> &BTreeMap<NodeId, SyncStatus> {
         &self.synced
     }
     /// Get the set of timed out nodes
     pub fn timed_out(&self) -> &BTreeSet<NodeId> {
         &self.timed_out
     }
 }
 pub struct Success {
     outcome: SuccessfulOutcome,
     synced: BTreeMap<NodeId, SyncStatus>,
 }
 impl Success {
     /// Get the [`SuccessfulOutcome`] of the success.
     pub fn outcome(&self) -> SuccessfulOutcome {
         self.outcome
     }
     /// Get the set of synchronized nodes.
     pub fn synced(&self) -> &BTreeMap<NodeId, SyncStatus> {
         &self.synced
     }
 }
 /// Error in constructing the [`Announcer`].
 pub enum AnnouncerError {
     /// Both sets of already synchronized and un-synchronized nodes were empty
     /// of nodes were empty.
     AlreadySynced(AlreadySynced),
     /// No more nodes are available for synchronizing with.
     NoSeeds,
     /// The target could not be constructed.
     Target(TargetError),
 }
 impl From<AlreadySynced> for AnnouncerError {
     fn from(value: AlreadySynced) -> Self {
         Self::AlreadySynced(value)
     }
 }
 pub struct AlreadySynced {
     preferred: usize,
     synced: usize,
 }
 impl AlreadySynced {
     /// Get the number of preferred nodes that are already synchronized.
     pub fn preferred(&self) -> usize {
         self.preferred
     }
     /// Get the total number of nodes that are already synchronized.
     pub fn synced(&self) -> usize {
         self.synced
     }
 }
 /// The status of the synchronized node.
 #[derive(Clone, Copy, Debug)]
 pub enum SyncStatus {
     /// The node was already synchronized before starting the [`Announcer`]
     /// process.
     AlreadySynced,
     /// The node was synchronized as part of the [`Announcer`] process, marking
     /// the amount of time that passed to synchronize with the node.
     Synced { duration: time::Duration },
 }
 /// Progress of the [`Announcer`] process.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct Progress {
     preferred: usize,
     synced: usize,
     unsynced: usize,
 }
 impl Progress {
     /// The number of preferred seeds that are synchronized.
     pub fn preferred(&self) -> usize {
         self.preferred
     }
     /// The number of seeds that are synchronized.
     pub fn synced(&self) -> usize {
         self.synced
     }
     /// The number of seeds that are un-synchronized.
     pub fn unsynced(&self) -> usize {
         self.unsynced
     }
 }
 #[derive(Debug, thiserror::Error)]
 #[non_exhaustive]
 #[error("a minimum number of replicas or set of preferred seeds must be provided")]
 pub struct TargetError;
 /// The target for the [`Announcer`] to reach.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Target {
     preferred_seeds: BTreeSet<NodeId>,
     replicas: ReplicationFactor,
 }
 impl Target {
     pub fn new(
         preferred_seeds: BTreeSet<NodeId>,
         replicas: ReplicationFactor,
     ) -> Result<Self, TargetError> {
         if replicas.lower_bound() == 0 && preferred_seeds.is_empty() {
             Err(TargetError)
         } else {
             Ok(Self {
                 preferred_seeds,
                 replicas,
             })
         }
     }
     /// Get the set of preferred seeds that are trying to be synchronized with.
     pub fn preferred_seeds(&self) -> &BTreeSet<NodeId> {
         &self.preferred_seeds
     }
     /// Get the number of replicas that is trying to be reached.
     pub fn replicas(&self) -> &ReplicationFactor {
         &self.replicas
     }
 }
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum SuccessfulOutcome {
     MinReplicationFactor { preferred: usize, synced: usize },
     MaxReplicationFactor { preferred: usize, synced: usize },
 }