hack co for baseline

consensus/src/consensus_observer/publisher/consensus_publisher.rs

@@ -17,9 +17,11 @@ use crate::consensus_observer::{
 };
 use aptos_channels::aptos_channel::Receiver;
 use aptos_config::{config::ConsensusObserverConfig, network_id::PeerNetworkId};
+use aptos_crypto::HashValue;
 use aptos_infallible::RwLock;
 use aptos_logger::{error, info, warn};
 use aptos_network::application::interface::NetworkClient;
+use dashmap::DashMap;
 use futures::StreamExt;
 use futures_channel::mpsc;
 use std::{collections::HashSet, sync::Arc, time::Duration};
@@ -41,6 +43,8 @@ pub struct ConsensusPublisher {
 
     // The sender for outbound network messages
     outbound_message_sender: mpsc::Sender<(PeerNetworkId, ConsensusObserverDirectSend)>,
+
+    buffered_ordered_block: DashMap<HashValue, ConsensusObserverDirectSend>,
 }
 
 impl ConsensusPublisher {
@@ -64,6 +68,7 @@ impl ConsensusPublisher {
             consensus_observer_config,
             active_subscribers: Arc::new(RwLock::new(HashSet::new())),
             outbound_message_sender,
+            buffered_ordered_block: DashMap::new(),
         };
 
         // Return the publisher and the outbound message receiver
@@ -231,6 +236,16 @@ impl ConsensusPublisher {
         }
     }
 
+    pub fn buffer_ordered_block(&self, block_id: HashValue, message: ConsensusObserverDirectSend) {
+        self.buffered_ordered_block.insert(block_id, message);
+    }
+
+    pub fn flush_buffered_ordered_block(&self, block_id: HashValue) {
+        if let Some((_, message)) = self.buffered_ordered_block.remove(&block_id) {
+            self.publish_message(message);
+        }
+    }
+
     /// Starts the consensus publisher
     pub async fn start(
         self,

consensus/src/pipeline/buffer_manager.rs

@@ -411,7 +411,9 @@ impl BufferManager {
                 ordered_blocks.clone().into_iter().map(Arc::new).collect(),
                 ordered_proof.clone(),
             );
-            consensus_publisher.publish_message(message);
+            // consensus_publisher.publish_message(message);
+            let block_id = ordered_blocks.last().unwrap().id();
+            consensus_publisher.buffer_ordered_block(block_id, message);
         }
         self.execution_schedule_phase_tx
             .send(request)
@@ -526,6 +528,8 @@ impl BufferManager {
                     self.reset().await;
                 }
                 if let Some(consensus_publisher) = &self.consensus_publisher {
+                    let block_id = block.id();
+                    consensus_publisher.flush_buffered_ordered_block(block_id);
                     let message =
                         ConsensusObserverMessage::new_commit_decision_message(commit_proof.clone());
                     consensus_publisher.publish_message(message);