Tracking Block Number for Consensus Messages,

SurfingNerd · SurfingNerd · commit 1f05e5008847 · 2025-07-29T16:21:13.000+02:00
outdated messages for consensus are now ignored in the resend loop. DMDcoin#261
diff --git a/crates/ethcore/src/client/chain_notify.rs b/crates/ethcore/src/client/chain_notify.rs
@@ -22,7 +22,7 @@ use std::{collections::HashMap, time::Duration};
 /// Messages to broadcast via chain
 pub enum ChainMessageType {
     /// Consensus message
-    Consensus(Vec<u8>),
+    Consensus(u64, Vec<u8>),
 }
 
 /// Route type to indicate whether it is enacted or retracted.
diff --git a/crates/ethcore/src/client/client.rs b/crates/ethcore/src/client/client.rs
@@ -3288,18 +3288,23 @@ impl super::traits::EngineClient for Client {
         }
     }
 
-    fn broadcast_consensus_message(&self, message: Bytes) {
+    fn broadcast_consensus_message(&self, future_block_id: u64, message: Bytes) {
         self.statistics
             .broadcasted_consensus_messages
             .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
         self.statistics
             .broadcasted_consensus_messages_bytes
             .fetch_add(message.len() as u64, std::sync::atomic::Ordering::Relaxed);
 
-        self.notify(|notify| notify.broadcast(ChainMessageType::Consensus(message.clone())));
+        self.notify(|notify| {
+            notify.broadcast(ChainMessageType::Consensus(
+                future_block_id,
+                message.clone(),
+            ))
+        });
     }
 
-    fn send_consensus_message(&self, message: Bytes, node_id: Option<H512>) {
+    fn send_consensus_message(&self, future_block_id: u64, message: Bytes, node_id: Option<H512>) {
         self.statistics
             .sent_consensus_messages
             .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
@@ -3308,7 +3313,12 @@ impl super::traits::EngineClient for Client {
             .fetch_add(message.len() as u64, std::sync::atomic::Ordering::Relaxed);
 
         if let Some(n) = node_id {
-            self.notify(|notify| notify.send(ChainMessageType::Consensus(message.clone()), &n));
+            self.notify(|notify| {
+                notify.send(
+                    ChainMessageType::Consensus(future_block_id, message.clone()),
+                    &n,
+                )
+            });
         }
     }
 
diff --git a/crates/ethcore/src/client/test_client.rs b/crates/ethcore/src/client/test_client.rs
@@ -1232,9 +1232,14 @@ impl super::traits::EngineClient for TestBlockChainClient {
         }
     }
 
-    fn broadcast_consensus_message(&self, _message: Bytes) {}
+    fn broadcast_consensus_message(&self, _future_block_id: u64, _message: Bytes) {}
 
-    fn send_consensus_message(&self, _message: Bytes, _node_id: Option<H512>) {
+    fn send_consensus_message(
+        &self,
+        _future_block_id: u64,
+        _message: Bytes,
+        _node_id: Option<H512>,
+    ) {
         // TODO: allow test to intercept the message to relay it to other test clients
     }
 
diff --git a/crates/ethcore/src/client/traits.rs b/crates/ethcore/src/client/traits.rs
@@ -668,10 +668,10 @@ pub trait EngineClient: Sync + Send + ChainInfo {
     fn submit_seal(&self, block_hash: H256, seal: Vec<Bytes>);
 
     /// Broadcast a consensus message to the network.
-    fn broadcast_consensus_message(&self, message: Bytes);
+    fn broadcast_consensus_message(&self, future_block_id: u64, message: Bytes);
 
     /// Send a consensus message to the specified peer
-    fn send_consensus_message(&self, message: Bytes, node_id: Option<H512>);
+    fn send_consensus_message(&self, future_block_id: u64, message: Bytes, node_id: Option<H512>);
 
     /// Get the transition to the epoch the given parent hash is part of
     /// or transitions to.
diff --git a/crates/ethcore/src/engines/authority_round/mod.rs b/crates/ethcore/src/engines/authority_round/mod.rs
@@ -1158,7 +1158,7 @@ impl AuthorityRound {
         self.empty_steps.lock().insert(empty_step);
     }
 
-    fn generate_empty_step(&self, parent_hash: &H256) {
+    fn generate_empty_step(&self, future_block_id: u64, parent_hash: &H256) {
         let step = self.step.inner.load();
         let empty_step_rlp = empty_step_rlp(step, parent_hash);
 
@@ -1173,16 +1173,16 @@ impl AuthorityRound {
             };
 
             trace!(target: "engine", "broadcasting empty step message: {:?}", empty_step);
-            self.broadcast_message(message_rlp);
+            self.broadcast_message(future_block_id, message_rlp);
             self.handle_empty_step_message(empty_step);
         } else {
             warn!(target: "engine", "generate_empty_step: FAIL: accounts secret key unavailable");
         }
     }
 
-    fn broadcast_message(&self, message: Vec<u8>) {
+    fn broadcast_message(&self, future_block_id: u64, message: Vec<u8>) {
         if let Ok(c) = self.upgrade_client_or(None) {
-            c.broadcast_consensus_message(message);
+            c.broadcast_consensus_message(future_block_id, message);
         }
     }
 
@@ -1759,7 +1759,7 @@ impl Engine<EthereumMachine> for AuthorityRound {
                     .compare_exchange(true, false, AtomicOrdering::SeqCst, AtomicOrdering::SeqCst)
                     .is_ok()
                 {
-                    self.generate_empty_step(header.parent_hash());
+                    self.generate_empty_step(header.number(), header.parent_hash());
                 }
 
                 return Seal::None;
diff --git a/crates/ethcore/src/engines/hbbft/hbbft_engine.rs b/crates/ethcore/src/engines/hbbft/hbbft_engine.rs
@@ -72,6 +72,16 @@ enum Message {
     Sealing(BlockNumber, sealing::Message),
 }
 
+impl Message {
+    /// Returns the epoch (block number) of the message.
+    pub fn block_number(&self) -> BlockNumber {
+        match self {
+            Message::HoneyBadger(_, msg) => msg.epoch(),
+            Message::Sealing(block_num, _) => *block_num,
+        }
+    }
+}
+
 /// The Honey Badger BFT Engine.
 pub struct HoneyBadgerBFT {
     transition_service: IoService<()>,
@@ -727,7 +737,11 @@ impl HoneyBadgerBFT {
                     trace!(target: "consensus", "Dispatching message {:?} to {:?}", m.message, set);
                     for node_id in set.into_iter().filter(|p| p != net_info.our_id()) {
                         trace!(target: "consensus", "Sending message to {}", node_id.0);
-                        client.send_consensus_message(ser.clone(), Some(node_id.0));
+                        client.send_consensus_message(
+                            m.message.block_number(),
+                            ser.clone(),
+                            Some(node_id.0),
+                        );
                     }
                 }
                 Target::AllExcept(set) => {
@@ -737,7 +751,11 @@ impl HoneyBadgerBFT {
                         .filter(|p| (p != &net_info.our_id() && !set.contains(p)))
                     {
                         trace!(target: "consensus", "Sending exclusive message to {}", node_id.0);
-                        client.send_consensus_message(ser.clone(), Some(node_id.0));
+                        client.send_consensus_message(
+                            m.message.block_number(),
+                            ser.clone(),
+                            Some(node_id.0),
+                        );
                     }
                 }
             }
@@ -780,6 +798,7 @@ impl HoneyBadgerBFT {
                 message: Message::HoneyBadger(*message_counter, msg.message),
             }
         });
+
         self.dispatch_messages(&client, messages, network_info);
         std::mem::drop(message_counter);
         self.process_output(client, step.output, network_info);
@@ -813,7 +832,7 @@ impl HoneyBadgerBFT {
 
         let step = match self
             .hbbft_state
-            .try_write_for(std::time::Duration::from_millis(10))
+            .try_write_for(std::time::Duration::from_millis(100))
         {
             Some(mut state_lock) => state_lock.try_send_contribution(client.clone(), &self.signer),
             None => {
diff --git a/crates/ethcore/sync/src/api.rs b/crates/ethcore/sync/src/api.rs
@@ -506,6 +506,11 @@ impl SyncProtocolHandler {
     }
 
     fn send_cached_consensus_messages_for(&self, sync_io: &mut dyn SyncIo, node_id: &NodeId) {
+        let last_interesting_block = self
+            .chain
+            .block_number(types::ids::BlockId::Latest)
+            .unwrap_or(0);
+
         // now since we are connected, lets send any cached messages
         if let Some(vec_msg) = self.message_cache.write().remove(&Some(*node_id)) {
             trace!(target: "consensus", "Cached Messages: Trying to send cached messages to {:?}", node_id);
@@ -514,7 +519,11 @@ impl SyncProtocolHandler {
 
             for msg in vec_msg {
                 match msg {
-                    ChainMessageType::Consensus(message) => {
+                    ChainMessageType::Consensus(block, message) => {
+                        if block < last_interesting_block {
+                            // https://github.com/DMDcoin/diamond-node/issues/261
+                            continue;
+                        }
                         let send_consensus_result = self.sync.write().send_consensus_packet(
                             sync_io,
                             message.clone(),
@@ -525,7 +534,7 @@ impl SyncProtocolHandler {
                             Ok(_) => {}
                             Err(e) => {
                                 info!(target: "consensus", "Error sending cached consensus message to peer (re-adding) {:?}: {:?}", node_id, e);
-                                failed_messages.push(ChainMessageType::Consensus(message));
+                                failed_messages.push(ChainMessageType::Consensus(block, message));
                             }
                         }
                     }
@@ -534,8 +543,9 @@ impl SyncProtocolHandler {
 
             if !failed_messages.is_empty() {
                 // If we failed to send some messages, cache them for later
-                let mut lock = self.message_cache.write();
-                lock.entry(Some(*node_id))
+                self.message_cache
+                    .write()
+                    .entry(Some(*node_id))
                     .or_default()
                     .extend(failed_messages);
             } else {
@@ -729,7 +739,7 @@ impl ChainNotify for EthSync {
                 &self.eth_handler.overlay,
             );
             match message_type {
-                ChainMessageType::Consensus(message) => self
+                ChainMessageType::Consensus(_block, message) => self
                     .eth_handler
                     .sync
                     .write()
@@ -746,13 +756,13 @@ impl ChainNotify for EthSync {
                                              &self.eth_handler.overlay);
 
             match message_type {
-                ChainMessageType::Consensus(message) => {
+                ChainMessageType::Consensus(block, message) => {
                     let send_result = self.eth_handler.sync.write().send_consensus_packet(&mut sync_io, message.clone(), node_id);
                     if let Err(e) = send_result {
                         info!(target: "consensus", "Error sending consensus message to peer - caching message {:?}: {:?}", node_id, e);
                         // If we failed to send the message, cache it for later
                         let mut lock = self.eth_handler.message_cache.write();
-                        lock.entry(Some(node_id.clone())).or_default().push(ChainMessageType::Consensus(message));
+                        lock.entry(Some(node_id.clone())).or_default().push(ChainMessageType::Consensus(block, message));
                     }
                 },
             }
diff --git a/crates/ethcore/sync/src/chain/propagator.rs b/crates/ethcore/sync/src/chain/propagator.rs
@@ -405,7 +405,6 @@ impl ChainSync {
         for peer_id in lucky_peers {
             let send_result =
                 ChainSync::send_packet(io, peer_id, ConsensusDataPacket, packet.clone());
-
             if let Err(e) = send_result {
                 info!(target: "sync", "Error broadcast consensus packet to peer {}: {:?}", peer_id, e);
             } else {
diff --git a/crates/net/network-devp2p/src/session_container.rs b/crates/net/network-devp2p/src/session_container.rs
@@ -433,7 +433,7 @@ impl SessionContainer {
     pub(crate) fn deregister_session_stream<Host: mio::deprecated::Handler>(
         &self,
         stream: usize,
-        
+
         event_loop: &mut mio::deprecated::EventLoop<Host>,
     ) {
         let connections = if stream < self.last_handshake {

Original file line number	Diff line number	Diff line change
`@@ -22,7 +22,7 @@ use std::{collections::HashMap, time::Duration};`
`22`	`22`	`/// Messages to broadcast via chain`
`23`	`23`	`pub enum ChainMessageType {`
`24`	`24`	`/// Consensus message`
`25`		`- Consensus(Vec<u8>),`
	`25`	`+ Consensus(u64, Vec<u8>),`
`26`	`26`	`}`
`27`	`27`
`28`	`28`	`/// Route type to indicate whether it is enacted or retracted.`
Original file line number	Diff line number	Diff line change
`@@ -1232,9 +1232,14 @@ impl super::traits::EngineClient for TestBlockChainClient {`
`1232`	`1232`	`}`
`1233`	`1233`	`}`
`1234`	`1234`
`1235`		`- fn broadcast_consensus_message(&self, _message: Bytes) {}`
	`1235`	`+ fn broadcast_consensus_message(&self, _future_block_id: u64, _message: Bytes) {}`
`1236`	`1236`
`1237`		`- fn send_consensus_message(&self, _message: Bytes, _node_id: Option<H512>) {`
	`1237`	`+ fn send_consensus_message(`
	`1238`	`+ &self,`
	`1239`	`+ _future_block_id: u64,`
	`1240`	`+ _message: Bytes,`
	`1241`	`+ _node_id: Option<H512>,`
	`1242`	`+ ) {`
`1238`	`1243`	`// TODO: allow test to intercept the message to relay it to other test clients`
`1239`	`1244`	`}`
`1240`	`1245`
Original file line number	Diff line number	Diff line change
`@@ -1158,7 +1158,7 @@ impl AuthorityRound {`
`1158`	`1158`	`self.empty_steps.lock().insert(empty_step);`
`1159`	`1159`	`}`
`1160`	`1160`
`1161`		`- fn generate_empty_step(&self, parent_hash: &H256) {`
	`1161`	`+ fn generate_empty_step(&self, future_block_id: u64, parent_hash: &H256) {`
`1162`	`1162`	`let step = self.step.inner.load();`
`1163`	`1163`	`let empty_step_rlp = empty_step_rlp(step, parent_hash);`
`1164`	`1164`
`@@ -1173,16 +1173,16 @@ impl AuthorityRound {`
`1173`	`1173`	`};`
`1174`	`1174`
`1175`	`1175`	`trace!(target: "engine", "broadcasting empty step message: {:?}", empty_step);`
`1176`		`- self.broadcast_message(message_rlp);`
	`1176`	`+ self.broadcast_message(future_block_id, message_rlp);`
`1177`	`1177`	`self.handle_empty_step_message(empty_step);`
`1178`	`1178`	`} else {`
`1179`	`1179`	`warn!(target: "engine", "generate_empty_step: FAIL: accounts secret key unavailable");`
`1180`	`1180`	`}`
`1181`	`1181`	`}`
`1182`	`1182`
`1183`		`- fn broadcast_message(&self, message: Vec<u8>) {`
	`1183`	`+ fn broadcast_message(&self, future_block_id: u64, message: Vec<u8>) {`
`1184`	`1184`	`if let Ok(c) = self.upgrade_client_or(None) {`
`1185`		`- c.broadcast_consensus_message(message);`
	`1185`	`+ c.broadcast_consensus_message(future_block_id, message);`
`1186`	`1186`	`}`
`1187`	`1187`	`}`
`1188`	`1188`
`@@ -1759,7 +1759,7 @@ impl Engine<EthereumMachine> for AuthorityRound {`
`1759`	`1759`	`.compare_exchange(true, false, AtomicOrdering::SeqCst, AtomicOrdering::SeqCst)`
`1760`	`1760`	`.is_ok()`
`1761`	`1761`	`{`
`1762`		`- self.generate_empty_step(header.parent_hash());`
	`1762`	`+ self.generate_empty_step(header.number(), header.parent_hash());`
`1763`	`1763`	`}`
`1764`	`1764`
`1765`	`1765`	`return Seal::None;`