Use Multiwatcher in verify task.

Author: twittner

robusta/src/multiwatcher.rs

@@ -6,11 +6,9 @@ use futures::{StreamExt, stream::SelectAll};
 use tokio::{spawn, sync::mpsc, task::JoinHandle};
 use tokio_stream::wrappers::ReceiverStream;
 use tracing::{debug, warn};
-use url::Url;
 
 #[derive(Debug)]
 pub struct Multiwatcher {
-    height: Height,
     threshold: usize,
     watchers: Vec<JoinHandle<()>>,
     headers: BTreeMap<Height, HashMap<Header, HashSet<Id>>>,
@@ -29,11 +27,10 @@ impl Drop for Multiwatcher {
 }
 
 impl Multiwatcher {
-    pub fn new<C, H, I, N>(configs: C, height: H, nsid: N, threshold: usize) -> Self
+    pub fn new<C, H, N>(configs: C, height: H, nsid: N, threshold: usize) -> Self
     where
         C: IntoIterator<Item = Config>,
         H: Into<Height>,
-        I: IntoIterator<Item = Url>,
         N: Into<NamespaceId>,
     {
         let height = height.into();
@@ -49,52 +46,48 @@ impl Multiwatcher {
                 while tx.send((id, w.next().await)).await.is_ok() {}
             }));
         }
+        assert!(!watchers.is_empty());
         Self {
-            height,
             threshold,
             stream,
             watchers,
             headers: BTreeMap::from_iter([(height, HashMap::new())]),
         }
     }
 
-    pub async fn next(&mut self) -> Option<Header> {
+    pub async fn next(&mut self) -> Header {
         loop {
-            let (i, hdr) = self.stream.next().await?;
+            let (i, hdr) = self.stream.next().await.expect("watchers never terminate");
             let h = Height::from(hdr.height());
             if Some(h) < self.headers.first_entry().map(|e| *e.key()) {
-                debug!(%h, "ignoring header below minimum height");
+                debug!(height = %h, "ignoring header below minimum height");
                 continue;
             }
             if self.has_voted(h, i) {
-                warn!(%h, "source sent multiple headers for same height");
+                warn!(height = %h, "source sent multiple headers for same height");
                 continue;
             }
-            let votes = self.headers.entry(h).or_default();
-            if let Some(ids) = votes.get(&hdr)
-                && ids.len() + 1 >= self.threshold
-            {
-                self.gc(h);
-                return Some(hdr);
+            let counter = self.headers.entry(h).or_default();
+            let votes = counter.get(&hdr).map(|ids| ids.len()).unwrap_or(0) + 1;
+            if votes >= self.threshold {
+                self.headers.retain(|k, _| *k > h);
+                debug!(height = %h, "header available");
+                return hdr;
             }
-            votes.entry(hdr).or_default().insert(i);
+            debug!(height = %h, %votes, "vote added");
+            counter.entry(hdr).or_default().insert(i);
         }
     }
 
     fn has_voted(&self, height: Height, id: Id) -> bool {
         let Some(m) = self.headers.get(&height) else {
             return false;
         };
-        for v in m.values() {
-            if v.contains(&id) {
+        for ids in m.values() {
+            if ids.contains(&id) {
                 return true;
             }
         }
         false
     }
-
-    fn gc(&mut self, height: Height) {
-        self.headers.retain(|h, _| *h >= height);
-        self.height = height;
-    }
 }

timeboost-builder/src/config.rs

@@ -18,6 +18,6 @@ pub struct SubmitterConfig {
     pub(crate) pubkey: PublicKey,
     #[builder(into)]
     pub(crate) namespace: NamespaceId,
-    pub(crate) robusta: robusta::Config,
+    pub(crate) robusta: (robusta::Config, Vec<robusta::Config>),
     pub(crate) committee: Committee,
 }

timeboost-builder/src/submit.rs

@@ -42,7 +42,7 @@ impl Submitter {
     where
         M: ::metrics::Metrics,
     {
-        let client = robusta::Client::new(cfg.robusta.clone());
+        let client = robusta::Client::new(cfg.robusta.0.clone());
         let verified = Arc::new(Mutex::new(BTreeSet::new()));
         let committees = Arc::new(AsyncMutex::new(CommitteeVec::new(cfg.committee.clone())));
         let handler = Handler {
@@ -58,10 +58,12 @@ impl Submitter {
             client: client.clone(),
             verified,
         };
+        let mut configs = vec![cfg.robusta.0.clone()];
+        configs.extend(cfg.robusta.1.iter().cloned());
         Submitter {
             handler,
             config: cfg,
-            verify_task: spawn(verifier.verify()),
+            verify_task: spawn(verifier.verify(configs)),
             submitters: TaskTracker::new(),
             committees,
             task_permits: Arc::new(Semaphore::new(MAX_TASKS)),
@@ -109,7 +111,7 @@ struct Verifier {
 }
 
 impl Verifier {
-    async fn verify(self) -> Empty {
+    async fn verify(self, configs: Vec<robusta::Config>) -> Empty {
         let mut delays = self.client.config().delay_iter();
         let height = loop {
             if let Ok(h) = self.client.height().await {
@@ -118,7 +120,8 @@ impl Verifier {
             let d = delays.next().expect("delay iterator repeats endlessly");
             sleep(d).await;
         };
-        let mut watcher = robusta::Watcher::new(self.client.config().clone(), height, self.nsid);
+        let threshold = 2 * configs.len() / 3 + 1;
+        let mut watcher = robusta::Multiwatcher::new(configs, height, self.nsid, threshold);
         loop {
             let h = watcher.next().await;
             let committees = self.committees.lock().await;
@@ -297,7 +300,7 @@ mod tests {
 
             let scfg = SubmitterConfig::builder()
                 .pubkey(k.public_key())
-                .robusta(rcfg.clone())
+                .robusta((rcfg.clone(), Vec::new()))
                 .namespace(10_101u64)
                 .committee(committee.clone())
                 .build();

timeboost/src/binaries/timeboost.rs

@@ -234,13 +234,14 @@ async fn main() -> Result<()> {
         .internal_api(my_keyset.internal_address.clone())
         .maybe_nitro_addr(my_keyset.nitro_addr.clone())
         .recover(is_recover)
-        .robusta(
+        .robusta((
             robusta::Config::builder()
                 .base_url(cli.espresso_base_url)
                 .wss_base_url(cli.espresso_websocket_url)
                 .label(pubkey.to_string())
                 .build(),
-        )
+            Vec::new(),
+        ))
         .namespace(cli.namespace)
         .build();
 

timeboost/src/config.rs

@@ -55,7 +55,7 @@ pub struct TimeboostConfig {
     pub(crate) leash_len: usize,
 
     /// Configuration of espresso network client.
-    pub(crate) robusta: robusta::Config,
+    pub(crate) robusta: (robusta::Config, Vec<robusta::Config>),
 
     #[builder(into)]
     pub(crate) namespace: NamespaceId,