Change Submitter to batch blocks.

Author: twittner

robusta/src/lib.rs

@@ -18,7 +18,9 @@ use timeboost_types::{BlockNumber, CertifiedBlock};
 use tokio::time::sleep;
 use tracing::{debug, warn};
 
-use crate::types::{RecvBody, SendBody, TaggedBase64, TransactionsWithProof, VidCommonResponse, TX};
+use crate::types::{
+    RecvBody, SendBody, TX, TaggedBase64, TransactionsWithProof, VidCommonResponse,
+};
 
 pub use crate::multiwatcher::Multiwatcher;
 pub use crate::types::Height;
@@ -58,7 +60,11 @@ impl Client {
         self.get_with_retry(u).await
     }
 
-    pub async fn submit<N>(&self, nsid: N, blocks: &[CertifiedBlock<Validated>]) -> Result<(), Error>
+    pub async fn submit<N>(
+        &self,
+        nsid: N,
+        blocks: &[CertifiedBlock<Validated>],
+    ) -> Result<(), Error>
     where
         N: Into<NamespaceId>,
     {
@@ -104,25 +110,23 @@ impl Client {
         }
         Either::Right(trxs.into_iter().flat_map(move |t| {
             match minicbor::decode::<RecvBody>(t.payload()) {
-                Ok(body) => {
-                    Either::Right(body.blocks.into_iter().filter_map(|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
-                        }
-                    }))
-                }
+                Ok(body) => Either::Right(body.blocks.into_iter().filter_map(|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,

robusta/src/types.rs

@@ -28,14 +28,14 @@ pub(crate) struct VidCommonResponse {
 #[cbor(map)]
 pub(crate) struct RecvBody {
     #[cbor(n(0))]
-    pub(crate) blocks: Vec<CertifiedBlock<Unchecked>>
+    pub(crate) blocks: Vec<CertifiedBlock<Unchecked>>,
 }
 
 #[derive(Debug, Encode)]
 #[cbor(map)]
 pub(crate) struct SendBody<'a> {
     #[cbor(n(0))]
-    pub(crate) blocks: &'a [CertifiedBlock<Validated>]
+    pub(crate) blocks: &'a [CertifiedBlock<Validated>],
 }
 
 macro_rules! Primitive {

timeboost-builder/src/metrics.rs

@@ -1,10 +1,10 @@
-use metrics::{Gauge, Metrics, NoMetrics};
+use metrics::{Counter, Metrics, NoMetrics};
 
 #[derive(Debug)]
 #[non_exhaustive]
 pub struct BuilderMetrics {
-    pub block_submit: Box<dyn Gauge>,
-    pub submit_tasks: Box<dyn Gauge>,
+    pub blocks_submitted: Box<dyn Counter>,
+    pub blocks_verified: Box<dyn Counter>,
 }
 
 impl Default for BuilderMetrics {
@@ -16,8 +16,8 @@ impl Default for BuilderMetrics {
 impl BuilderMetrics {
     pub fn new<M: Metrics>(m: &M) -> Self {
         Self {
-            block_submit: m.create_gauge("block_submit", None),
-            submit_tasks: m.create_gauge("submit_tasks", None),
+            blocks_submitted: m.create_counter("blocks_submitted", None),
+            blocks_verified: m.create_counter("blocks_verified", None),
         }
     }
 }