Merge branch 'ax/crypto-update' into ax/reshare-core

Author: alxiong

robusta/Cargo.toml

@@ -8,17 +8,21 @@ rust-version.workspace = true
 [dependencies]
 bincode = { workspace = true }
 bon = { workspace = true }
+bytes = { workspace = true }
 data-encoding = { workspace = true }
+either = { workspace = true }
 espresso-types = { workspace = true }
 futures = { workspace = true }
 hotshot-query-service = { workspace = true }
 hotshot-types = { workspace = true }
+multisig = { path = "../multisig" }
 reqwest = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 timeboost-types = { path = "../timeboost-types" }
 tokio = { workspace = true }
+tokio-stream = { workspace = true }
 tokio-tungstenite = { workspace = true }
 tracing = { workspace = true }
 url = { workspace = true }

robusta/src/config.rs

@@ -1,35 +1,40 @@
-use std::{iter::repeat, time::Duration};
+use std::{iter::repeat, num::NonZeroUsize, time::Duration};
 
 use bon::Builder;
-use url::{ParseError, Url};
+use url::Url;
 
-const NUM_DELAYS: usize = 5;
+const NUM_DELAYS: NonZeroUsize = NonZeroUsize::new(5).expect("5 > 0");
 
 #[derive(Debug, Clone, Builder)]
 pub struct Config {
+    /// Log label.
+    #[builder(into)]
+    pub(crate) label: String,
+
     /// Espresso network base URL.
-    #[builder(with = |s: &str| -> Result<_, ParseError> { Url::parse(s) })]
     pub(crate) base_url: Url,
 
     /// Espresso network websocket base URL.
-    #[builder(with = |s: &str| -> Result<_, ParseError> { Url::parse(s) })]
     pub(crate) wss_base_url: Url,
 
-    #[builder(default = 3)]
-    pub(crate) max_redirects: usize,
-
     /// The sequence of delays between successive requests.
     ///
     /// The last value is repeated forever.
     #[builder(default = [1, 3, 5, 10, 15])]
-    pub(crate) delays: [u8; NUM_DELAYS],
+    pub(crate) delays: [u8; NUM_DELAYS.get()],
 }
 
 impl Config {
+    pub fn with_websocket_base_url(&self, u: Url) -> Self {
+        let mut c = self.clone();
+        c.wss_base_url = u;
+        c
+    }
+
     pub fn delay_iter(&self) -> impl Iterator<Item = Duration> + use<> {
         self.delays
             .into_iter()
-            .chain(repeat(self.delays[NUM_DELAYS - 1]))
+            .chain(repeat(self.delays[NUM_DELAYS.get() - 1]))
             .map(|n| Duration::from_secs(n.into()))
     }
 }

robusta/src/lib.rs

@@ -1,26 +1,32 @@
 mod config;
+mod multiwatcher;
 mod types;
 mod watcher;
 
+use std::iter::empty;
 use std::time::Duration;
 
+use either::Either;
 use espresso_types::{Header, NamespaceId, Transaction};
-use reqwest::{StatusCode, Url, redirect::Policy};
+use multisig::{Unchecked, Validated};
+use reqwest::{StatusCode, Url};
 use serde::{Serialize, de::DeserializeOwned};
 use serde_json as json;
-use timeboost_types::CertifiedBlock;
+use timeboost_types::sailfish::CommitteeVec;
+use timeboost_types::{BlockNumber, CertifiedBlock};
 use tokio::time::sleep;
-use tracing::warn;
+use tracing::{debug, warn};
 
 use crate::types::{TX, TaggedBase64, TransactionsWithProof, VidCommonResponse};
 
+pub use crate::multiwatcher::Multiwatcher;
 pub use crate::types::Height;
-pub use crate::watcher::{WatchError, watch};
+pub use crate::watcher::{WatchError, Watcher};
 pub use config::{Config, ConfigBuilder};
 pub use espresso_types;
 
 /// A client for the Espresso network.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Client {
     config: Config,
     client: reqwest::Client,
@@ -31,7 +37,6 @@ impl Client {
         let r = reqwest::Client::builder()
             .https_only(true)
             .timeout(Duration::from_secs(30))
-            .redirect(Policy::limited(c.max_redirects))
             .build()
             .expect("TLS and DNS resolver work");
         Self {
@@ -40,45 +45,90 @@ impl Client {
         }
     }
 
-    pub async fn height(&mut self) -> Result<Height, Error> {
+    pub fn config(&self) -> &Config {
+        &self.config
+    }
+
+    pub async fn height(&self) -> Result<Height, Error> {
         let u = self.config.base_url.join("status/block-height")?;
         self.get_with_retry(u).await
     }
 
-    pub async fn submit(&mut self, cb: &CertifiedBlock) -> Result<(), Error> {
-        let nid = NamespaceId::from(u64::from(u32::from(cb.data().namespace())));
-        let trx = Transaction::new(nid, serialize(cb)?);
+    pub async fn submit<N>(&self, nsid: N, cb: &CertifiedBlock<Validated>) -> Result<(), Error>
+    where
+        N: Into<NamespaceId>,
+    {
+        let trx = Transaction::new(nsid.into(), serialize(cb)?);
         let url = self.config.base_url.join("submit/submit")?;
         self.post_with_retry::<_, TaggedBase64<TX>>(url, &trx)
             .await?;
         Ok(())
     }
 
-    pub async fn verify(&mut self, h: &Header, cb: &CertifiedBlock) -> Result<(), Error> {
-        let nsid = NamespaceId::from(u64::from(u32::from(cb.data().namespace())));
-
-        let trxs = self.transactions(h.height(), nsid).await?;
+    pub async fn verified<N, const C: usize>(
+        &self,
+        nsid: N,
+        hdr: &Header,
+        cvec: &CommitteeVec<C>,
+    ) -> impl Iterator<Item = BlockNumber>
+    where
+        N: Into<NamespaceId>,
+    {
+        let nsid = nsid.into();
+        debug!(node = %self.config.label, %nsid, height = %hdr.height(), "verifying blocks");
+        let Ok(trxs) = self.transactions(hdr.height(), nsid).await else {
+            debug!(node = %self.config.label, %nsid, height = %hdr.height(), "no transactions");
+            return Either::Left(empty());
+        };
         let Some(proof) = trxs.proof else {
-            return Err(ProofError::NoProof.into());
+            debug!(node = %self.config.label, %nsid, height = %hdr.height(), "no proof");
+            return Either::Left(empty());
         };
-        if !trxs.transactions.iter().any(|t| matches(t.payload(), cb)) {
-            return Err(Error::TransactionNotFound);
-        }
-
-        let vidc = self.vid_common(h.height()).await?;
-
-        let Some((trxs, ns)) = proof.verify(h.ns_table(), &h.payload_commitment(), &vidc.common)
+        let Ok(vidc) = self.vid_common(hdr.height()).await else {
+            debug!(node = %self.config.label, height = %hdr.height(), "no vid common");
+            return Either::Left(empty());
+        };
+        let Some((trxs, ns)) =
+            proof.verify(hdr.ns_table(), &hdr.payload_commitment(), &vidc.common)
         else {
-            return Err(ProofError::InvalidProof.into());
+            warn!(node = %self.config.label, %nsid, height = %hdr.height(), "proof verification failed");
+            return Either::Left(empty());
         };
         if ns != nsid {
-            return Err(ProofError::NamespaceMismatch(ns, nsid).into());
-        }
-        if !trxs.iter().any(|t| matches(t.payload(), cb)) {
-            return Err(ProofError::TransactionNotInProof.into());
+            warn!(node = %self.config.label, a = %nsid, b = %ns, height = %hdr.height(), "namespace mismatch");
+            return Either::Left(empty());
         }
-
-        Ok(())
+        Either::Right(trxs.into_iter().filter_map(move |t| {
+            match deserialize::<CertifiedBlock<Unchecked>>(t.payload()) {
+                Ok(b) => {
+                    let Some(c) = cvec.get(b.committee()) else {
+                        warn!(
+                            node      = %self.config.label,
+                            height    = %hdr.height(),
+                            committee = %b.committee(),
+                            "unknown committee"
+                        );
+                        return None;
+                    };
+                    if let Some(b) = b.validated(c) {
+                        Some(b.cert().data().num())
+                    } else {
+                        warn!(node = %self.config.label, height = %hdr.height(), "invalid block");
+                        None
+                    }
+                }
+                Err(err) => {
+                    warn!(
+                        node   = %self.config.label,
+                        nsid   = %nsid,
+                        height = %hdr.height(),
+                        err    = %err,
+                        "could not deserialize block"
+                    );
+                    None
+                }
+            }
+        }))
     }
 
     async fn transactions<H, N>(&self, height: H, nsid: N) -> Result<TransactionsWithProof, Error>
@@ -116,7 +166,7 @@ impl Client {
             match self.get(url.clone()).await {
                 Ok(a) => return Ok(a),
                 Err(err) => {
-                    warn!(%url, %err, "failed to get response");
+                    warn!(node = %self.config.label, %url, %err, "failed to get response");
                     sleep(delay.next().expect("infinite delay sequence")).await;
                 }
             }
@@ -133,7 +183,7 @@ impl Client {
             match self.post(url.clone(), a).await {
                 Ok(b) => return Ok(b),
                 Err(err) => {
-                    warn!(%url, %err, "failed to post request");
+                    warn!(node = %self.config.label, %url, %err, "failed to post request");
                     sleep(delay.next().expect("infinite delay sequence")).await;
                 }
             }
@@ -168,13 +218,6 @@ impl Client {
     }
 }
 
-fn matches(a: &[u8], b: &CertifiedBlock) -> bool {
-    let Ok(a) = deserialize::<CertifiedBlock>(a) else {
-        return false;
-    };
-    a.data().hash() == b.data().hash() && a.data().hash() == b.cert().data().hash()
-}
-
 /// Errors `Client` can not recover from.
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
@@ -235,7 +278,7 @@ fn deserialize<T: DeserializeOwned>(d: &[u8]) -> Result<T, Error> {
 
 #[cfg(test)]
 mod tests {
-    use super::{Client, Config};
+    use super::{Client, Config, Watcher};
 
     #[tokio::test]
     async fn decaf_smoke() {
@@ -244,15 +287,23 @@ mod tests {
             .try_init();
 
         let cfg = Config::builder()
-            .base_url("https://query.decaf.testnet.espresso.network/v1/")
-            .unwrap()
-            .wss_base_url("wss://query.decaf.testnet.espresso.network/v1/")
-            .unwrap()
+            .base_url(
+                "https://query.decaf.testnet.espresso.network/v1/"
+                    .parse()
+                    .unwrap(),
+            )
+            .wss_base_url(
+                "wss://query.decaf.testnet.espresso.network/v1/"
+                    .parse()
+                    .unwrap(),
+            )
+            .label("decaf_smoke")
             .build();
 
-        let mut clt = Client::new(cfg.clone());
+        let clt = Client::new(cfg.clone());
         let height = clt.height().await.unwrap();
-        let header = super::watch(&cfg, height, None).await.unwrap();
+        let mut watcher = Watcher::new(cfg, height, None);
+        let header = watcher.next().await;
         assert_eq!(u64::from(height), header.height());
     }
 }