fix: sec updates, update select loop

topos-protocol · Apr 24, 2024 · 7fbbb29 · 7fbbb29
1 parent 19da5d9
commit 7fbbb29
Show file tree

Hide file tree

Showing 2 changed files with 39 additions and 43 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/topos-tce-broadcast/src/task_manager/mod.rs b/crates/topos-tce-broadcast/src/task_manager/mod.rs
@@ -108,7 +108,6 @@ impl TaskManager {
 
     pub async fn run(mut self, shutdown_receiver: CancellationToken) {
         let mut pending_certificate_interval = tokio::time::interval(Duration::from_micros(500));
-        let mut message_interval = tokio::time::interval(Duration::from_millis(100));
 
         loop {
             tokio::select! {
@@ -118,29 +117,26 @@ impl TaskManager {
                     self.next_pending_certificate();
                 }
 
-                _ = message_interval.tick() => {
-                    if let Some(msg) = self.message_receiver.recv().await {
-                        match msg {
-                            DoubleEchoCommand::Echo { certificate_id, .. } | DoubleEchoCommand::Ready { certificate_id, .. } => {
-                                if let Some(task_context) = self.tasks.get(&certificate_id) {
-                                    _ = task_context.sink.send(msg).await;
-                                } else {
-                                    self.buffered_messages
-                                        .entry(certificate_id)
-                                        .or_default()
-                                        .push(msg);
-                                };
-                            }
-                            DoubleEchoCommand::Broadcast { ref cert, need_gossip, pending_id } => {
-                                trace!("Received broadcast message for certificate {} ", cert.id);
+                let Some(msg) = self.message_receiver.recv() {
+                    match msg {
+                        DoubleEchoCommand::Echo { certificate_id, .. } | DoubleEchoCommand::Ready { certificate_id, .. } => {
+                            if let Some(task_context) = self.tasks.get(&certificate_id) {
+                                _ = task_context.sink.send(msg).await;
+                            } else {
+                                self.buffered_messages
+                                    .entry(certificate_id)
+                                    .or_default()
+                                    .push(msg);
+                            };
+                        }
+                        DoubleEchoCommand::Broadcast { ref cert, need_gossip, pending_id } => {
+                            trace!("Received broadcast message for certificate {} ", cert.id);
 
-                                self.create_task(cert, need_gossip, pending_id)
-                            }
+                            self.create_task(cert, need_gossip, pending_id)
                         }
                     }
                 }
 
-
                 Some((certificate_id, status)) = self.running_tasks.next() => {
                     if let TaskStatus::Success = status {
                         trace!("Task for certificate {} finished successfully", certificate_id);