Ensure header heights are consecutive.

In a `Multiwatcher`, the individual watchers only make progress when a
threshold has delivered a value. Each watcher is required to deliver
consecutive and strictly monotonic block heights, otherwise the watcher
is dropped and re-connects after some time, starting at the current
lower bound of heights.

Author: twittner

robusta/src/lib.rs

@@ -303,7 +303,7 @@ mod tests {
         let clt = Client::new(cfg.clone());
         let height = clt.height().await.unwrap();
         let mut watcher = Watcher::new(cfg, height, None);
-        let header = watcher.next().await;
+        let header = watcher.next().await.unwrap_right();
         assert_eq!(u64::from(height), header.height());
     }
 }

robusta/src/multiwatcher.rs

@@ -4,15 +4,18 @@ use std::{
         Arc,
         atomic::{AtomicU64, Ordering},
     },
+    time::Duration,
 };
 
 use crate::{Config, Height, Watcher};
+use either::Either;
 use espresso_types::{Header, NamespaceId};
 use futures::{StreamExt, stream::SelectAll};
 use tokio::{
     spawn,
     sync::{Barrier, mpsc},
     task::JoinHandle,
+    time::sleep,
 };
 use tokio_stream::wrappers::ReceiverStream;
 use tracing::{debug, warn};
@@ -66,16 +69,47 @@ impl Multiwatcher {
             let lower_bound = lower_bound.clone();
             watchers.push(spawn(async move {
                 let i = Id(i);
-                let mut w = Watcher::new(c, height, nsid);
                 loop {
-                    let h = w.next().await;
-                    if h.height() <= lower_bound.load(Ordering::Relaxed) {
-                        continue;
+                    let height = lower_bound.load(Ordering::Relaxed);
+                    let mut w = Watcher::new(c.clone(), height, nsid);
+                    let mut expected = height + 1;
+                    loop {
+                        match w.next().await {
+                            Either::Right(hdr) => {
+                                if hdr.height() > expected {
+                                    warn!(
+                                        url      = %c.wss_base_url,
+                                        height   = %hdr.height(),
+                                        expected = %expected,
+                                        "unexpected block height"
+                                    );
+                                    break;
+                                }
+                                expected += 1;
+                                if hdr.height() <= lower_bound.load(Ordering::Relaxed) {
+                                    continue;
+                                }
+                                if tx.send((i, hdr)).await.is_err() {
+                                    return;
+                                }
+                            }
+                            Either::Left(height) => {
+                                if *height > expected {
+                                    warn!(
+                                        url      = %c.wss_base_url,
+                                        height   = %height,
+                                        expected = %expected,
+                                        "unexpected block height"
+                                    );
+                                    break;
+                                }
+                                expected += 1;
+                            }
+                        }
+                        barrier.wait().await;
                     }
-                    if tx.send((i, h)).await.is_err() {
-                        break;
-                    }
-                    barrier.wait().await;
+                    drop(w);
+                    sleep(Duration::from_secs(3)).await // wait a little before re-connecting
                 }
             }));
         }

robusta/src/types.rs

@@ -1,4 +1,8 @@
-use std::{borrow::Cow, fmt};
+use std::{
+    borrow::Cow,
+    fmt,
+    ops::{Add, AddAssign, Deref},
+};
 
 use bon::Builder;
 use data_encoding::BASE64URL_NOPAD;
@@ -42,6 +46,42 @@ macro_rules! Primitive {
                 self.0.fmt(f)
             }
         }
+
+        impl Deref for $name {
+            type Target = $t;
+
+            fn deref(&self) -> &$t {
+                &self.0
+            }
+        }
+
+        impl Add for $name {
+            type Output = Self;
+
+            fn add(self, rhs: Self) -> Self::Output {
+                Self(self.0 + rhs.0)
+            }
+        }
+
+        impl Add<$t> for $name {
+            type Output = Self;
+
+            fn add(self, rhs: $t) -> Self {
+                Self(self.0 + rhs)
+            }
+        }
+
+        impl AddAssign for $name {
+            fn add_assign(&mut self, rhs: Self) {
+                *self = *self + rhs
+            }
+        }
+
+        impl AddAssign<$t> for $name {
+            fn add_assign(&mut self, rhs: $t) {
+                *self = *self + rhs
+            }
+        }
     };
 }
 

robusta/src/watcher.rs

@@ -1,6 +1,7 @@
 use std::str::from_utf8;
 
 use bytes::Bytes;
+use either::Either;
 use espresso_types::{Header, NamespaceId};
 use futures::{SinkExt, StreamExt};
 use reqwest::header::LOCATION;
@@ -37,7 +38,7 @@ impl Watcher {
         }
     }
 
-    pub async fn next(&mut self) -> Header {
+    pub async fn next(&mut self) -> Either<Height, Header> {
         'main: loop {
             let ws = if let Some(w) = &mut self.websocket {
                 w
@@ -74,16 +75,16 @@ impl Watcher {
                     Some(Ok(Message::Text(text))) => {
                         match serde_json::from_str::<Header>(text.as_str()) {
                             Ok(hdr) => {
+                                self.height = hdr.height().into();
                                 if let Some(id) = &self.namespace {
                                     if hdr.ns_table().find_ns_id(id).is_some() {
-                                        self.height = hdr.height().into();
-                                        return hdr;
+                                        return Either::Right(hdr);
                                     } else {
                                         debug!(height = %hdr.height(), "namespace id not found");
+                                        return Either::Left(self.height);
                                     }
                                 } else {
-                                    self.height = hdr.height().into();
-                                    return hdr;
+                                    return Either::Right(hdr);
                                 }
                             }
                             Err(err) => {