Ignore messages with duplicated nonces

CHANGELOG.md

@@ -7,13 +7,17 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ## [Unreleased]
 
 ### Fixed
+- [2375](https://github.com/FuelLabs/fuel-core/pull/2375): Modify executor to ignore messages from the relayer that have a duplicate nonce
 - [2366](https://github.com/FuelLabs/fuel-core/pull/2366): The `importer_gas_price_for_block` metric is properly collected.
 - [2369](https://github.com/FuelLabs/fuel-core/pull/2369): The `transaction_insertion_time_in_thread_pool_milliseconds` metric is properly collected.
 
 ### Added
 - [2321](https://github.com/FuelLabs/fuel-core/pull/2321): New metrics for the txpool: "The size of transactions in the txpool" (`txpool_tx_size`), "The time spent by a transaction in the txpool in seconds" (`txpool_tx_time_in_txpool_seconds`), The number of transactions in the txpool (`txpool_number_of_transactions`), "The number of transactions pending verification before entering the txpool" (`txpool_number_of_transactions_pending_verification`), "The number of executable transactions in the txpool" (`txpool_number_of_executable_transactions`), "The time it took to select transactions for inclusion in a block in nanoseconds" (`txpool_select_transaction_time_nanoseconds`), The time it took to insert a transaction in the txpool in milliseconds (`txpool_insert_transaction_time_milliseconds`).
 - [2362](https://github.com/FuelLabs/fuel-core/pull/2362): Added a new request_response protocol version `/fuel/req_res/0.0.2`. In comparison with `/fuel/req/0.0.1`, which returns an empty response when a request cannot be fulfilled, this version returns more meaningful error codes. Nodes still support the version `0.0.1` of the protocol to guarantee backward compatibility with fuel-core nodes. Empty responses received from nodes using the old protocol `/fuel/req/0.0.1` are automatically converted into an error `ProtocolV1EmptyResponse` with error code 0, which is also the only error code implemented. More specific error codes will be added in the future.
 
+### Breaking
+- [](https://github.com/FuelLabs/fuel-core/pull/2375): Add new `ExecutorEvent` variant for ignored messages
+
 ## [Version 0.40.0]
 
 ### Added

crates/fuel-core/src/executor.rs

@@ -3285,6 +3285,78 @@ mod tests {
             Ok(())
         }
 
+        fn on_chain_db_containing_message(
+            message: &Message,
+            da_block_height: u64,
+        ) -> Database<OnChain> {
+            let mut db = add_consensus_parameters(
+                Database::default(),
+                &ConsensusParameters::default(),
+            );
+            db.storage_as_mut::<Messages>()
+                .insert(message.id(), message)
+                .unwrap();
+
+            let mut block = Block::default();
+            block.header_mut().set_da_height(da_block_height.into());
+            block.header_mut().recalculate_metadata();
+
+            db.storage_as_mut::<FuelBlocks>()
+                .insert(&0.into(), &block)
+                .expect("Should insert genesis block without any problems");
+            db
+        }
+
+        fn relayer_db_containing_message(message: &Message) -> Database<Relayer> {
+            let mut relayer_db = Database::<Relayer>::default();
+            add_message_to_relayer(&mut relayer_db, message.clone());
+            relayer_db
+        }
+
+        #[test]
+        fn produce_without_commit__messages_with_duplicate_nonces_are_ignored() {
+            // given
+            let previous_da_height = 1;
+            let shared_nonce = 1234.into();
+            let mut message = Message::default();
+            message.set_amount(10);
+            message.set_da_height(previous_da_height.into());
+            message.set_nonce(shared_nonce);
+            let on_chain_db =
+                on_chain_db_containing_message(&message, previous_da_height);
+
+            let mut message_with_matching_nonce = message.clone();
+            message_with_matching_nonce.set_amount(20);
+            let new_da_height = previous_da_height + 1;
+            message_with_matching_nonce.set_da_height(new_da_height.into());
+            let relayer_db = relayer_db_containing_message(&message_with_matching_nonce);
+
+            let producer = create_relayer_executor(on_chain_db, relayer_db);
+            let block_height = 1;
+            let block = test_block(block_height.into(), new_da_height.into(), 0);
+
+            // when
+            let (result, changes) = producer
+                .produce_without_commit(block.into())
+                .unwrap()
+                .into();
+
+            // then
+            let no_messages_in_changes = ChangesIterator::<Column>::new(&changes)
+                .iter_all::<Messages>(None)
+                .count()
+                == 0;
+            assert!(no_messages_in_changes);
+            // and
+            let event_contains_ignore = result.events.iter().any(|event| {
+                matches!(event, ExecutorEvent::MessageIgnored {
+                    message,
+                    ..
+                } if message == &message_with_matching_nonce)
+            });
+            assert!(event_contains_ignore);
+        }
+
         #[test]
         fn execute_without_commit__block_producer_includes_correct_inbox_event_merkle_root(
         ) {

crates/fuel-core/src/graphql_api/worker_service.rs

@@ -200,6 +200,7 @@ where
                         &(),
                     )?;
             }
+            Event::MessageIgnored { .. } => {}
             Event::MessageConsumed(message) => {
                 block_st_transaction
                     .storage_as_mut::<OwnedMessageIds>()

crates/services/executor/src/executor.rs

@@ -812,7 +812,7 @@ where
         D: KeyValueInspect<Column = Column>,
     {
         let forced_transactions = if self.relayer.enabled() {
-            self.process_da(
+            self.process_l1_events(
                 block_height,
                 da_block_height,
                 data,
@@ -875,7 +875,7 @@ where
         }
     }
 
-    fn process_da<D>(
+    fn process_l1_events<D>(
         &mut self,
         block_height: BlockHeight,
         da_block_height: DaBlockHeight,
@@ -916,12 +916,21 @@ where
                         if message.da_height() != da_height {
                             return Err(ExecutorError::RelayerGivesIncorrectMessages)
                         }
-                        block_storage_tx
-                            .storage_as_mut::<Messages>()
-                            .insert(message.nonce(), &message)?;
-                        execution_data
-                            .events
-                            .push(ExecutorEvent::MessageImported(message));
+                        let message_nonce = message.nonce();
+                        let message_event = if !block_storage_tx
+                            .storage_as_ref::<Messages>()
+                            .contains_key(message_nonce)?
+                        {
+                            block_storage_tx
+                                .storage_as_mut::<Messages>()
+                                .insert(message_nonce, &message)?;
+                            ExecutorEvent::MessageImported(message)
+                        } else {
+                            let reason =
+                                "Message with the same nonce already exists".to_string();
+                            ExecutorEvent::MessageIgnored { message, reason }
+                        };
+                        execution_data.events.push(message_event);
                     }
                     Event::Transaction(relayed_tx) => {
                         let id = relayed_tx.id();

crates/types/src/services/executor.rs

@@ -1,5 +1,11 @@
 //! Types related to executor service.
 
+#[cfg(feature = "alloc")]
+use alloc::{
+    string::String,
+    vec::Vec,
+};
+
 use crate::{
     blockchain::{
         block::Block,
@@ -34,12 +40,6 @@ use crate::{
     services::Uncommitted,
 };
 
-#[cfg(feature = "alloc")]
-use alloc::{
-    string::String,
-    vec::Vec,
-};
-
 /// The alias for executor result.
 pub type Result<T> = core::result::Result<T, Error>;
 /// The uncommitted result of the block production execution.
@@ -131,6 +131,13 @@ impl<DatabaseTransaction> UncommittedResult<DatabaseTransaction> {
 pub enum Event {
     /// Imported a new spendable message from the relayer.
     MessageImported(Message),
+    /// Ignored a message from the relayer.
+    MessageIgnored {
+        /// Ignored message
+        message: Message,
+        /// The reason for ignoring the message
+        reason: String,
+    },
     /// The message was consumed by the transaction.
     MessageConsumed(Message),
     /// Created a new spendable coin, produced by the transaction.