Merge pull request #503 from EspressoSystems/tw/submit2

Submit batches of certified blocks

Author: twittner

justfile

@@ -171,7 +171,7 @@ test-contract-deploy *ARGS:
   scripts/test-contract-deploy {{ARGS}}
 
 test-all: build_release build-test-utils
-  env RUST_LOG=block_checker=info,warn target/release/run \
+  env RUST_LOG=timeboost_builder::submit=debug,block_checker=info,warn target/release/run \
     --verbose \
     --timeout 120 \
     --spawn "1:anvil --port 8545" \

robusta/src/lib.rs

@@ -9,7 +9,7 @@ use std::time::Duration;
 
 use either::Either;
 use espresso_types::{Header, NamespaceId, Transaction};
-use multisig::{Unchecked, Validated};
+use multisig::Validated;
 use reqwest::{StatusCode, Url};
 use serde::{Serialize, de::DeserializeOwned};
 use serde_json as json;
@@ -18,7 +18,9 @@ use timeboost_types::{BlockNumber, CertifiedBlock};
 use tokio::time::sleep;
 use tracing::{debug, warn};
 
-use crate::types::{TX, TaggedBase64, TransactionsWithProof, VidCommonResponse};
+use crate::types::{
+    RecvBody, SendBody, TX, TaggedBase64, TransactionsWithProof, VidCommonResponse,
+};
 
 pub use crate::multiwatcher::Multiwatcher;
 pub use crate::types::Height;
@@ -58,11 +60,15 @@ impl Client {
         self.get_with_retry(u).await
     }
 
-    pub async fn submit<N>(&self, nsid: N, cb: &CertifiedBlock<Validated>) -> Result<(), Error>
+    pub async fn submit<N>(
+        &self,
+        nsid: N,
+        blocks: &[CertifiedBlock<Validated>],
+    ) -> Result<(), Error>
     where
         N: Into<NamespaceId>,
     {
-        let trx = Transaction::new(nsid.into(), minicbor::to_vec(cb)?);
+        let trx = Transaction::new(nsid.into(), minicbor::to_vec(SendBody { blocks })?);
         let url = self.config.base_url.join("submit/submit")?;
         self.post_with_retry::<_, TaggedBase64<TX>>(url, &trx)
             .await?;
@@ -102,9 +108,9 @@ impl Client {
             warn!(node = %self.config.label, a = %nsid, b = %ns, height = %hdr.height(), "namespace mismatch");
             return Either::Left(empty());
         }
-        Either::Right(trxs.into_iter().filter_map(move |t| {
-            match minicbor::decode::<CertifiedBlock<Unchecked>>(t.payload()) {
-                Ok(b) => {
+        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,
@@ -120,16 +126,16 @@ impl Client {
                         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"
+                        "could not decode transaction payload"
                     );
-                    None
+                    Either::Left(empty())
                 }
             }
         }))

robusta/src/types.rs

@@ -8,7 +8,10 @@ use bon::Builder;
 use data_encoding::BASE64URL_NOPAD;
 use espresso_types::{NsProof, Transaction};
 use hotshot_query_service::VidCommon;
+use minicbor::{Decode, Encode};
+use multisig::{Unchecked, Validated};
 use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
+use timeboost_types::CertifiedBlock;
 
 #[derive(Debug, Deserialize, Serialize, Builder)]
 pub(crate) struct TransactionsWithProof {
@@ -21,6 +24,20 @@ pub(crate) struct VidCommonResponse {
     pub(crate) common: VidCommon,
 }
 
+#[derive(Debug, Decode)]
+#[cbor(map)]
+pub(crate) struct RecvBody {
+    #[cbor(n(0))]
+    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>],
+}
+
 macro_rules! Primitive {
     ($name:ident, $t:ty) => {
         #[derive(

timeboost-builder/src/lib.rs

@@ -8,4 +8,4 @@ pub use robusta;
 pub use certifier::{Certifier, CertifierDown, CertifierError, Handle};
 pub use config::{CertifierConfig, CertifierConfigBuilder};
 pub use config::{SubmitterConfig, SubmitterConfigBuilder};
-pub use submit::Submitter;
+pub use submit::{SenderTaskDown, Submitter};

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),
         }
     }
 }