chore: adding republish_count

Signed-off-by: Simon Paitrault <[email protected]>

crates/topos-core/proto/topos/tce/v1/double_echo.proto

@@ -9,6 +9,7 @@ import "topos/uci/v1/certification.proto";
 
 message Gossip {
   topos.uci.v1.Certificate certificate = 1;
+  uint64 republish_count = 2;
 }
 
 message Echo {

crates/topos-core/src/api/grpc/generated/topos.tce.v1.rs

@@ -1453,6 +1453,8 @@ pub mod console_service_server {
 pub struct Gossip {
     #[prost(message, optional, tag = "1")]
     pub certificate: ::core::option::Option<super::super::uci::v1::Certificate>,
+    #[prost(uint64, tag = "2")]
+    pub republish_count: u64,
 }
 #[derive(serde::Deserialize, serde::Serialize)]
 #[allow(clippy::derive_partial_eq_without_eq)]

crates/topos-tce-broadcast/src/double_echo/broadcast_state.rs

@@ -31,6 +31,7 @@ pub struct BroadcastState {
     delivery_time: time::Instant,
     readies: HashSet<Ready>,
     pub(crate) expected_position: Option<Position>,
+    republish_count: u64,
 }
 
 impl BroadcastState {
@@ -59,6 +60,7 @@ impl BroadcastState {
             delivery_time: time::Instant::now(),
             readies: HashSet::new(),
             expected_position: None,
+            republish_count: 0,
         };
 
         _ = state.event_sender.try_send(ProtocolEvents::Broadcast {
@@ -70,8 +72,10 @@ impl BroadcastState {
                 "📣 Gossiping the Certificate {} from the source subnet {}",
                 &state.certificate.id, &state.certificate.source_subnet_id
             );
+
             let _ = state.event_sender.try_send(ProtocolEvents::Gossip {
                 cert: state.certificate.clone(),
+                republish_count: state.republish_count,
             });
         }
 
@@ -80,6 +84,21 @@ impl BroadcastState {
         state
     }
 
+    pub(crate) fn republish_certificate(&mut self) {
+        if let Some(count) = self.republish_count.checked_add(1) {
+            self.republish_count = count;
+
+            // TODO: catch error
+            let _ = self.event_sender.try_send(ProtocolEvents::Gossip {
+                cert: self.certificate.clone(),
+                republish_count: self.republish_count,
+            });
+
+            self.status = Status::Pending;
+
+            self.update_status();
+        }
+    }
     pub fn into_delivered(&self) -> CertificateDelivered {
         CertificateDelivered {
             certificate: self.certificate.clone(),
@@ -129,6 +148,7 @@ impl BroadcastState {
             payload.extend_from_slice(self.certificate.id.as_array());
             payload.extend_from_slice(self.validator_id.as_bytes());
 
+            // TODO: catch error
             let _ = self.event_sender.try_send(ProtocolEvents::Echo {
                 certificate_id: self.certificate.id,
                 signature: self.message_signer.sign_message(&payload).ok()?,
@@ -250,4 +270,8 @@ impl BroadcastState {
 
         delivery_threshold
     }
+
+    pub(crate) fn received_echo(&self) -> bool {
+        self.subscriptions_view.echo.len() < self.subscriptions_view.network_size
+    }
 }

crates/topos-tce-broadcast/src/event.rs

@@ -21,6 +21,7 @@ pub enum ProtocolEvents {
     /// Indicates that 'gossip' message broadcasting is required
     Gossip {
         cert: Certificate,
+        republish_count: u64,
     },
     /// Indicates that 'echo' message broadcasting is required
     Echo {

crates/topos-tce-broadcast/src/task_manager/task.rs

@@ -1,6 +1,7 @@
 use std::future::{Future, IntoFuture};
 use std::pin::Pin;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::{broadcast, mpsc};
 
 use topos_core::types::stream::Position;
@@ -96,8 +97,13 @@ impl IntoFuture for Task {
             );
             self.broadcast_state.expected_position = Some(expected_position);
 
+            let mut gossip_timer = tokio::time::interval(Duration::from_secs(5));
+            gossip_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
             loop {
                 tokio::select! {
+                    _tick = gossip_timer.tick(), if !self.broadcast_state.received_echo() => {
+                        self.broadcast_state.republish_certificate();
+                    }
                     Some(msg) = self.message_receiver.recv() => {
                         match msg {
                             DoubleEchoCommand::Echo { validator_id, .. } => {

crates/topos-tce/src/app_context/network.rs

@@ -30,6 +30,7 @@ impl AppContext {
                 match double_echo_request {
                     double_echo_request::Request::Gossip(Gossip {
                         certificate: Some(certificate),
+                        ..
                     }) => match uci::Certificate::try_from(certificate) {
                         Ok(cert) => {
                             if let hash_map::Entry::Vacant(entry) =

crates/topos-tce/src/app_context/protocol.rs

@@ -11,12 +11,16 @@ impl AppContext {
                 info!("Broadcasting certificate {}", certificate_id);
             }
 
-            ProtocolEvents::Gossip { cert } => {
+            ProtocolEvents::Gossip {
+                cert,
+                republish_count,
+            } => {
                 let cert_id = cert.id;
 
                 let request = DoubleEchoRequest {
                     request: Some(double_echo_request::Request::Gossip(Gossip {
                         certificate: Some(cert.into()),
+                        republish_count,
                     })),
                 };
 

crates/topos-tce/src/tests/mod.rs

@@ -37,6 +37,7 @@ async fn non_validator_publish_gossip(
     context
         .on_protocol_event(ProtocolEvents::Gossip {
             cert: certificates[0].certificate.clone(),
+            republish_count: 0,
         })
         .await;
 

crates/topos-tce/src/tests/network.rs

@@ -33,6 +33,7 @@ async fn handle_gossip(
     let msg = DoubleEchoRequest {
         request: Some(double_echo_request::Request::Gossip(Gossip {
             certificate: Some(certificate.into()),
+            republish_count: 0,
         })),
     };
     context
@@ -118,6 +119,7 @@ async fn handle_already_delivered(
     let msg = DoubleEchoRequest {
         request: Some(double_echo_request::Request::Gossip(Gossip {
             certificate: Some(certificate.into()),
+            republish_count: 0,
         })),
     };
     _ = context