cli: add control flow for announcement

To allow `rad sync` to break early from the call to `announce`,
control flow is introduced for the callback.

In `rad sync`, the number of synced nodes is checked against the
`replicas` -- if specified -- and sends a `Break`.

Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett

radicle-cli/examples/rad-sync.md

@@ -63,12 +63,18 @@ $ rad sync --fetch --seed z6Mkt67GdsW7715MEfRuP4pSZxJRJh6kj6Y48WRqVv4N1tRk
 ✓ Fetched repository from 1 seed(s)
 ```
 
-And the `--replicas` flag to sync with a number of nodes:
+And the `--replicas` flag to sync with a number of nodes. First we'll
+create a new issue so that we have something to announce:
 
 ```
-$ rad sync --fetch --replicas 1
+$ rad issue open --title "Test `rad sync --replicas`" --description "Check that the replicas works" -q --no-announce
+```
+
+```
+$ rad sync --replicas 1
 ✓ Fetching rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji from z6Mkt67…v4N1tRk..
 ✓ Fetched repository from 1 seed(s)
+✓ Synced with 1 node(s)
 ```
 
 We can check the sync status again to make sure everything's in sync:
@@ -78,8 +84,13 @@ $ rad sync status --sort-by alias
 ╭─────────────────────────────────────────────────────────────────────────────────────╮
 │ ●   Node                      Address                  Status   Tip       Timestamp │
 ├─────────────────────────────────────────────────────────────────────────────────────┤
-│ ●   alice   (you)             alice.radicle.xyz:8776            9f615f9   [  ...  ] │
+│ ●   alice   (you)             alice.radicle.xyz:8776            9ee4e68   [  ...  ] │
-│ ●   bob     z6Mkt67…v4N1tRk   bob.radicle.xyz:8776     synced   9f615f9   [  ...  ] │
+│ ●   bob     z6Mkt67…v4N1tRk   bob.radicle.xyz:8776     synced   9ee4e68   [  ...  ] │
-│ ●   eve     z6Mkux1…nVhib7Z   eve.radicle.xyz:8776     synced   9f615f9   [  ...  ] │
+│ ●   eve     z6Mkux1…nVhib7Z   eve.radicle.xyz:8776     synced   9ee4e68   [  ...  ] │
 ╰─────────────────────────────────────────────────────────────────────────────────────╯
 ```
+
+Note that both `bob` and `eve` are in sync despite seeing `Synced with
+1 node(s)`. This is because the announcement was still made to `bob`
+and `eve`, but we will only wait for at least `1` replica to be
+synchronized.

radicle-cli/src/commands/sync.rs

@@ -1,5 +1,6 @@
 use std::cmp::Ordering;
 use std::ffi::OsString;
+use std::ops::ControlFlow;
 use std::str::FromStr;
 use std::time;
 
@@ -387,7 +388,7 @@ fn announce_refs(
         let synced = seeds.iter().filter(|s| s.is_synced());
 
         match mode {
-            RepoSync::Seeds(seeds) => {
+            RepoSync::Seeds(ref seeds) => {
                 let synced = synced.map(|s| s.nid).collect::<Vec<_>>();
                 if seeds.iter().all(|s| synced.contains(s)) {
                     term::success!(
@@ -431,10 +432,20 @@ fn announce_refs(
     }
 
     let mut spinner = term::spinner(format!("Syncing with {} node(s)..", unsynced.len()));
-    let result = node.announce(rid, unsynced, timeout, |event| match event {
+    let cutoff = if let RepoSync::Replicas(replicas) = mode {
-        node::AnnounceEvent::Announced => {}
+        Some(replicas)
+    } else {
+        None
+    };
+    let result = node.announce(rid, unsynced, timeout, |event, synced| match event {
+        node::AnnounceEvent::Announced => ControlFlow::Continue(()),
         node::AnnounceEvent::RefsSynced { remote } => {
             spinner.message(format!("Synced with {remote}.."));
+            if Some(synced.len()) == cutoff {
+                ControlFlow::Break(())
+            } else {
+                ControlFlow::Continue(())
+            }
         }
     })?;
 

radicle-cli/src/node.rs

@@ -1,3 +1,4 @@
+use std::ops::ControlFlow;
 use std::time::Duration;
 
 use radicle::identity::Id;
@@ -33,10 +34,11 @@ fn announce_(rid: Id, node: &mut Node) -> Result<(), radicle::node::Error> {
         rid,
         connected,
         Duration::from_secs(9),
-        |event| match event {
+        |event, _| match event {
-            node::AnnounceEvent::Announced => {}
+            node::AnnounceEvent::Announced => ControlFlow::Continue(()),
             node::AnnounceEvent::RefsSynced { remote } => {
                 spinner.message(format!("Synced with {remote}.."));
+                ControlFlow::Continue(())
             }
         },
     )?;

radicle-remote-helper/src/push.rs

@@ -1,5 +1,6 @@
 use std::collections::HashSet;
 use std::io::IsTerminal;
+use std::ops::ControlFlow;
 use std::path::Path;
 use std::str::FromStr;
 use std::time;
@@ -639,12 +640,18 @@ fn sync(rid: Id, mut node: radicle::Node) -> Result<(), radicle::node::Error> {
     } else {
         cli::spinner_to(message, io::stderr(), io::sink())
     };
-    let result = node.announce(rid, connected, DEFAULT_SYNC_TIMEOUT, |event| match event {
+    let result = node.announce(
-        node::AnnounceEvent::Announced => {}
+        rid,
-        node::AnnounceEvent::RefsSynced { remote } => {
+        connected,
-            spinner.message(format!("Synced with {remote}.."));
+        DEFAULT_SYNC_TIMEOUT,
-        }
+        |event, _| match event {
-    })?;
+            node::AnnounceEvent::Announced => ControlFlow::Continue(()),
+            node::AnnounceEvent::RefsSynced { remote } => {
+                spinner.message(format!("Synced with {remote}.."));
+                ControlFlow::Continue(())
+            }
+        },
+    )?;
 
     if result.synced.is_empty() {
         spinner.failed();

radicle/src/node.rs

@@ -10,7 +10,7 @@ pub mod seed;
 
 use std::collections::{BTreeSet, HashMap, HashSet};
 use std::io::{BufRead, BufReader};
-use std::ops::Deref;
+use std::ops::{ControlFlow, Deref};
 use std::os::unix::net::UnixStream;
 use std::path::{Path, PathBuf};
 use std::str::FromStr;
@@ -901,17 +901,17 @@ impl Node {
         rid: Id,
         seeds: impl IntoIterator<Item = NodeId>,
         timeout: time::Duration,
-        mut callback: impl FnMut(AnnounceEvent),
+        mut callback: impl FnMut(AnnounceEvent, &[PublicKey]) -> ControlFlow<()>,
     ) -> Result<AnnounceResult, Error> {
         let events = self.subscribe(timeout)?;
         let mut seeds = seeds.into_iter().collect::<BTreeSet<_>>();
         let refs = self.announce_refs(rid)?;
 
-        callback(AnnounceEvent::Announced);
-
         let mut synced = Vec::new();
         let mut timeout: Vec<NodeId> = Vec::new();
 
+        callback(AnnounceEvent::Announced, &synced);
+
         for e in events {
             match e {
                 Ok(Event::RefsSynced {
@@ -921,7 +921,9 @@ impl Node {
                 }) if rid == rid_ => {
                     if seeds.remove(&remote) && refs.at == at {
                         synced.push(remote);
-                        callback(AnnounceEvent::RefsSynced { remote });
+                        if callback(AnnounceEvent::RefsSynced { remote }, &synced).is_break() {
+                            break;
+                        }
                     }
                 }
                 Ok(_) => {}