Merge pull request #3041 from massalabs/fix_endorsements_propagation

Fix endorsements propagation

docs/technical-doc/openrpc.json

@@ -2,7 +2,7 @@
     "openrpc": "1.2.4",
     "info": {
         "title": "Massa OpenRPC",
-        "version": "TEST.14.5",
+        "version": "TEST.14.6",
         "description": "Massa OpenRPC spec",
         "termsOfService": "https://open-rpc.org",
         "contact": {

docs/testnet/install.rst

@@ -14,9 +14,9 @@ From binaries
 If you just wish to run a Massa node without compiling it yourself, you
 can simply download the latest binary below and go the the next step: :ref:`Running a node <testnet-running>`.
 
-- `Windows executable <https://github.com/massalabs/massa/releases/download/TEST.14.4/massa_TEST.14.4_release_windows.zip>`_
-- `Linux binary <https://github.com/massalabs/massa/releases/download/TEST.14.4/massa_TEST.14.4_release_linux.tar.gz>`_ - only works with libc2.28 at least (for example Ubuntu 20.04 and higher)
-- `MacOS binary <https://github.com/massalabs/massa/releases/download/TEST.14.4/massa_TEST.14.4_release_macos.tar.gz>`_
+- `Windows executable <https://github.com/massalabs/massa/releases/download/TEST.14.6/massa_TEST.14.6_release_windows.zip>`_
+- `Linux binary <https://github.com/massalabs/massa/releases/download/TEST.14.6/massa_TEST.14.6_release_linux.tar.gz>`_ - only works with libc2.28 at least (for example Ubuntu 20.04 and higher)
+- `MacOS binary <https://github.com/massalabs/massa/releases/download/TEST.14.6/massa_TEST.14.6_release_macos.tar.gz>`_
 
 From source code
 ================

massa-models/src/config/constants.rs

@@ -45,7 +45,7 @@ lazy_static::lazy_static! {
                 .saturating_add(MassaTime::from_millis(1000 * 10))
         )
     } else {
-        1663605000000.into()  //  Saturday, September 19, 2022 16:30:00 PM UTC
+        1663678800000.into()  //  Saturday, September 20, 2022 13:00:00 PM UTC
     };
 
     /// TESTNET: time when the blockclique is ended.
@@ -64,7 +64,7 @@ lazy_static::lazy_static! {
         if cfg!(feature = "sandbox") {
             "SAND.0.0"
         } else {
-            "TEST.14.5"
+            "TEST.14.6"
         }
         .parse()
         .unwrap()
@@ -171,7 +171,6 @@ pub const MAX_BOOTSTRAP_ERROR_LENGTH: u32 = 10000;
 pub const PROTOCOL_CONTROLLER_CHANNEL_SIZE: usize = 1024;
 /// Event channel size
 pub const PROTOCOL_EVENT_CHANNEL_SIZE: usize = 1024;
-
 // ***********************
 // Constants used for execution module (injected from ConsensusConfig)
 //

massa-models/src/version.rs

@@ -191,8 +191,8 @@ impl Version {
     pub fn is_compatible(&self, other: &Version) -> bool {
         self.instance == other.instance
             && self.major == other.major
-            && self.minor >= 5
-            && other.minor >= 5
+            && self.minor >= 6
+            && other.minor >= 6
     }
 }
 

massa-node/base_config/config.toml

@@ -75,9 +75,9 @@
     # max cache size for which operations a foreign node knows about
     max_node_known_ops_size = 20000
     # max cache size for which endorsements our node knows about
-    max_known_endorsements_size = 1024
+    max_known_endorsements_size = 2048
     # max cache size for which endorsements a foreign node knows about
-    max_node_known_endorsements_size = 1024
+    max_node_known_endorsements_size = 2048
     # Maximum number of batches in the memory buffer.
     # Dismiss the new batches if overflow
     operation_batch_buffer_capacity = 10024
@@ -87,6 +87,10 @@
     asked_operations_pruning_period = 100000
     # Max number of operation per message, same as network param but can be smaller
     max_operations_per_message = 1024
+    # Time threshold after which operation are not propagated
+    max_operations_propagation_time = 32000
+    # Time threshold after which operation are not propagated
+    max_endorsements_propagation_time = 48000
 
 [network]
     # port on which to listen for protocol communication

massa-node/src/main.rs

@@ -323,6 +323,10 @@ async fn launch(
         max_serialized_operations_size_per_block: MAX_BLOCK_SIZE as usize,
         controller_channel_size: PROTOCOL_CONTROLLER_CHANNEL_SIZE,
         event_channel_size: PROTOCOL_EVENT_CHANNEL_SIZE,
+        genesis_timestamp: *GENESIS_TIMESTAMP,
+        t0: T0,
+        max_operations_propagation_time: SETTINGS.protocol.max_operations_propagation_time,
+        max_endorsements_propagation_time: SETTINGS.protocol.max_endorsements_propagation_time,
     };
     let (protocol_command_sender, protocol_event_receiver, protocol_manager) =
         start_protocol_controller(

massa-node/src/settings.rs

@@ -186,6 +186,10 @@ pub struct ProtocolSettings {
     pub asked_operations_pruning_period: MassaTime,
     /// Maximum of operations sent in one message.
     pub max_operations_per_message: u64,
+    /// Time threshold after which operation are not propagated
+    pub max_operations_propagation_time: MassaTime,
+    /// Time threshold after which operation are not propagated
+    pub max_endorsements_propagation_time: MassaTime,
 }
 
 #[cfg(test)]

massa-protocol-exports/src/settings.rs

@@ -42,4 +42,12 @@ pub struct ProtocolConfig {
     pub controller_channel_size: usize,
     /// Event channel size
     pub event_channel_size: usize,
+    /// t0
+    pub t0: MassaTime,
+    /// genesis_timestamp
+    pub genesis_timestamp: MassaTime,
+    /// max time we propagate operations
+    pub max_operations_propagation_time: MassaTime,
+    /// max time we propagate endorsements
+    pub max_endorsements_propagation_time: MassaTime,
 }

massa-protocol-exports/src/tests/tools.rs

@@ -279,6 +279,10 @@ pub fn create_protocol_config() -> ProtocolConfig {
         max_serialized_operations_size_per_block: 1024,
         controller_channel_size: 1024,
         event_channel_size: 1024,
+        genesis_timestamp: MassaTime::now(0).unwrap(),
+        t0: MassaTime::from_millis(16000),
+        max_operations_propagation_time: MassaTime::from_millis(30000),
+        max_endorsements_propagation_time: MassaTime::from_millis(60000),
     }
 }
 

massa-protocol-worker/src/protocol_network.rs

@@ -119,7 +119,10 @@ impl ProtocolWorker {
             }
             NetworkEvent::ReceivedEndorsements { node, endorsements } => {
                 massa_trace!(ENDORSEMENTS, { "node": node, "endorsements": endorsements});
-                if let Err(err) = self.note_endorsements_from_node(endorsements, &node).await {
+                if let Err(err) = self
+                    .note_endorsements_from_node(endorsements, &node, true)
+                    .await
+                {
                     warn!(
                         "node {} sent us critically incorrect endorsements, \
                         which may be an attack attempt by the remote node or a \

massa-protocol-worker/src/protocol_worker.rs

@@ -5,6 +5,8 @@ use crate::{node_info::NodeInfo, worker_operations_impl::OperationBatchBuffer};
 
 use massa_logging::massa_trace;
 
+use massa_models::slot::Slot;
+use massa_models::timeslots::get_block_slot_timestamp;
 use massa_models::{
     block::{BlockId, WrappedHeader},
     endorsement::{EndorsementId, WrappedEndorsement},
@@ -21,7 +23,7 @@ use massa_protocol_exports::{
 };
 
 use massa_storage::Storage;
-use massa_time::TimeError;
+use massa_time::{MassaTime, TimeError};
 use std::collections::{HashMap, HashSet};
 use tokio::{
     sync::mpsc,
@@ -792,7 +794,7 @@ impl ProtocolWorker {
         }
 
         let (_endorsement_ids, endorsements_reused) = match self
-            .note_endorsements_from_node(header.content.endorsements.clone(), source_node_id)
+            .note_endorsements_from_node(header.content.endorsements.clone(), source_node_id, false)
             .await
         {
             Err(_) => {
@@ -946,8 +948,41 @@ impl ProtocolWorker {
             let mut ops = self.storage.clone_without_refs();
             ops.store_operations(new_operations.into_values().collect());
 
-            // Request propagation
-            self.propagate_ops(&ops).await;
+            // Propagate operations when their expire period isn't `max_operations_propagation_time` old.
+            let mut ops_to_propagate = ops.clone();
+            let operations_to_not_propagate = {
+                let now = MassaTime::now(0)?;
+                let read_operations = ops_to_propagate.read_operations();
+                ops_to_propagate
+                    .get_op_refs()
+                    .iter()
+                    .filter_map(|op_id| {
+                        let expire_period =
+                            read_operations.get(op_id).unwrap().content.expire_period;
+                        let expire_period_timestamp = get_block_slot_timestamp(
+                            self.config.thread_count,
+                            self.config.t0,
+                            self.config.genesis_timestamp,
+                            Slot::new(expire_period, 0),
+                        );
+                        match expire_period_timestamp {
+                            Ok(slot_timestamp) => {
+                                if slot_timestamp
+                                    .saturating_add(self.config.max_endorsements_propagation_time)
+                                    < now
+                                {
+                                    Some(*op_id)
+                                } else {
+                                    None
+                                }
+                            }
+                            Err(_) => Some(*op_id),
+                        }
+                    })
+                    .collect()
+            };
+            ops_to_propagate.drop_operation_refs(&operations_to_not_propagate);
+            self.propagate_ops(&ops_to_propagate).await;
 
             // Add to pool
             self.pool_controller.add_operations(ops);
@@ -969,6 +1004,7 @@ impl ProtocolWorker {
         &mut self,
         endorsements: Vec<WrappedEndorsement>,
         source_node_id: &NodeId,
+        propagate: bool,
     ) -> Result<(PreHashMap<EndorsementId, u32>, bool), ProtocolError> {
         massa_trace!("protocol.protocol_worker.note_endorsements_from_node", { "node": source_node_id, "endorsements": endorsements});
         let length = endorsements.len();
@@ -1007,7 +1043,44 @@ impl ProtocolWorker {
             endorsements.store_endorsements(new_endorsements.into_values().collect());
 
             // Propagate endorsements
-            self.propagate_endorsements(&endorsements).await;
+            if propagate {
+                // Propagate endorsements when the slot of the block they endorse isn't `max_endorsements_propagation_time` old.
+                let mut endorsements_to_propagate = endorsements.clone();
+                let endorsements_to_not_propagate = {
+                    let now = MassaTime::now(0)?;
+                    let read_endorsements = endorsements_to_propagate.read_endorsements();
+                    endorsements_to_propagate
+                        .get_endorsement_refs()
+                        .iter()
+                        .filter_map(|endorsement_id| {
+                            let slot_endorsed_block =
+                                read_endorsements.get(endorsement_id).unwrap().content.slot;
+                            let slot_timestamp = get_block_slot_timestamp(
+                                self.config.thread_count,
+                                self.config.t0,
+                                self.config.genesis_timestamp,
+                                slot_endorsed_block,
+                            );
+                            match slot_timestamp {
+                                Ok(slot_timestamp) => {
+                                    if slot_timestamp.saturating_add(
+                                        self.config.max_endorsements_propagation_time,
+                                    ) < now
+                                    {
+                                        Some(*endorsement_id)
+                                    } else {
+                                        None
+                                    }
+                                }
+                                Err(_) => Some(*endorsement_id),
+                            }
+                        })
+                        .collect()
+                };
+                endorsements_to_propagate.drop_endorsement_refs(&endorsements_to_not_propagate);
+                self.propagate_endorsements(&endorsements_to_propagate)
+                    .await;
+            }
 
             // Add to pool
             self.pool_controller.add_endorsements(endorsements);

massa-protocol-worker/src/tests/endorsements_scenarios.rs

@@ -490,7 +490,7 @@ async fn test_protocol_propagates_endorsements_only_to_nodes_that_dont_know_abou
 
 #[tokio::test]
 #[serial]
-async fn test_protocol_does_propagates_endorsements_when_receiving_those_inside_a_header() {
+async fn test_protocol_does_not_propagates_endorsements_when_receiving_those_inside_a_header() {
     let protocol_config = &tools::PROTOCOL_CONFIG;
     protocol_test(
         protocol_config,
@@ -535,7 +535,7 @@ async fn test_protocol_does_propagates_endorsements_when_receiving_those_inside_
                     }) => {
                         let id = endorsements[0].id;
                         assert_eq!(id, expected_endorsement_id);
-                        break;
+                        panic!("Unexpected propagation of endorsement.");
                     }
                     Some(_) => panic!("Unexpected network command.."),
                     None => break,

massa-time/src/lib.rs

@@ -362,7 +362,7 @@ impl MassaTime {
     /// assert_eq!(res,MassaTime::from(42*7))
     /// ```
     #[must_use]
-    pub fn saturating_mul(self, n: u64) -> MassaTime {
+    pub const fn saturating_mul(self, n: u64) -> MassaTime {
         MassaTime(self.0.saturating_mul(n))
     }