feat: handle txpool message collisions (#524)

fuel-core-interfaces/src/db.rs

@@ -122,6 +122,7 @@ pub mod helpers {
     use fuel_tx::{
         Address, Bytes32, ContractId, Input, Metadata, Output, Transaction, TxId, UtxoId,
     };
+    use fuel_types::MessageId;
     use fuel_vm::prelude::Contract;
     use std::collections::HashMap;
 
@@ -167,7 +168,7 @@ pub mod helpers {
         /// Dummy contracts
         pub contract: HashMap<ContractId, Contract>,
         /// Dummy da messages.
-        pub messages: HashMap<Bytes32, DaMessage>,
+        pub messages: HashMap<MessageId, DaMessage>,
         /// variable for last committed and finalized fuel height
         pub last_committed_finalized_fuel_height: BlockHeight,
     }
@@ -745,24 +746,24 @@ pub mod helpers {
     }
 
     // bridge message. Used by relayer.
-    impl Storage<Bytes32, DaMessage> for DummyDb {
-        type Error = crate::db::KvStoreError;
+    impl Storage<MessageId, DaMessage> for DummyDb {
+        type Error = KvStoreError;
 
         fn insert(
             &mut self,
-            key: &Bytes32,
+            key: &MessageId,
             value: &DaMessage,
         ) -> Result<Option<DaMessage>, Self::Error> {
             Ok(self.data.lock().messages.insert(*key, value.clone()))
         }
 
-        fn remove(&mut self, key: &Bytes32) -> Result<Option<DaMessage>, Self::Error> {
+        fn remove(&mut self, key: &MessageId) -> Result<Option<DaMessage>, Self::Error> {
             Ok(self.data.lock().messages.remove(key))
         }
 
         fn get<'a>(
             &'a self,
-            key: &Bytes32,
+            key: &MessageId,
         ) -> Result<Option<std::borrow::Cow<'a, DaMessage>>, Self::Error> {
             Ok(self
                 .data
@@ -772,7 +773,7 @@ pub mod helpers {
                 .map(|i| Cow::Owned(i.clone())))
         }
 
-        fn contains_key(&self, key: &Bytes32) -> Result<bool, Self::Error> {
+        fn contains_key(&self, key: &MessageId) -> Result<bool, Self::Error> {
             Ok(self.data.lock().messages.contains_key(key))
         }
     }

fuel-core-interfaces/src/model/messages.rs

@@ -2,11 +2,11 @@ use super::BlockHeight;
 use crate::model::DaBlockHeight;
 use core::ops::Deref;
 use fuel_crypto::Hasher;
-use fuel_types::{Address, Bytes32, Word};
+use fuel_types::{Address, MessageId, Word};
 
 /// Message send from Da layer to fuel by bridge
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
 pub struct DaMessage {
     pub sender: Address,
     pub recipient: Address,
@@ -20,15 +20,15 @@ pub struct DaMessage {
 }
 
 impl DaMessage {
-    pub fn id(&self) -> Bytes32 {
+    pub fn id(&self) -> MessageId {
         let mut hasher = Hasher::default();
         hasher.input(self.sender);
         hasher.input(self.recipient);
-        hasher.input(self.owner);
         hasher.input(self.nonce.to_be_bytes());
+        hasher.input(self.owner);
         hasher.input(self.amount.to_be_bytes());
         hasher.input(&self.data);
-        hasher.digest()
+        MessageId::from(*hasher.digest())
     }
 
     pub fn check(self) -> CheckedDaMessage {
@@ -40,11 +40,11 @@ impl DaMessage {
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct CheckedDaMessage {
     message: DaMessage,
-    id: Bytes32,
+    id: MessageId,
 }
 
 impl CheckedDaMessage {
-    pub fn id(&self) -> &Bytes32 {
+    pub fn id(&self) -> &MessageId {
         &self.id
     }
 }

fuel-core-interfaces/src/relayer.rs

@@ -1,7 +1,7 @@
 use async_trait::async_trait;
 use derive_more::{Deref, DerefMut};
 use fuel_storage::Storage;
-use fuel_types::{Address, Bytes32};
+use fuel_types::{Address, MessageId};
 use std::{collections::HashMap, sync::Arc};
 use tokio::sync::{mpsc, oneshot};
 
@@ -41,7 +41,7 @@ impl StakingDiff {
 // But for ValidatorSet, it is little bit different.
 #[async_trait]
 pub trait RelayerDb:
-     Storage<Bytes32, DaMessage, Error = KvStoreError> // bridge messages
+     Storage<MessageId, DaMessage, Error = KvStoreError> // bridge messages
     + Storage<ValidatorId, (ValidatorStake, Option<ConsensusId>), Error = KvStoreError> // validator set
     + Storage<Address, Vec<DaBlockHeight>,Error = KvStoreError> // delegate index
     + Storage<DaBlockHeight, StakingDiff, Error = KvStoreError> // staking diff
@@ -54,7 +54,7 @@ pub trait RelayerDb:
         &mut self,
         message: &CheckedDaMessage,
     ) {
-        let _ = Storage::<Bytes32, DaMessage>::insert(self,message.id(),message.as_ref());
+        let _ = Storage::<MessageId, DaMessage>::insert(self,message.id(),message.as_ref());
     }
 
     /// Insert difference make on staking in this particular DA height.

fuel-core-interfaces/src/txpool.rs

@@ -1,12 +1,13 @@
 use crate::{
     db::{Error as DbStateError, KvStoreError},
-    model::Coin,
     model::TxInfo,
+    model::{Coin, DaMessage},
 };
 use derive_more::{Deref, DerefMut};
 use fuel_storage::Storage;
 use fuel_tx::{ContractId, UtxoId};
 use fuel_tx::{Transaction, TxId};
+use fuel_types::MessageId;
 use fuel_vm::prelude::Contract;
 use std::sync::Arc;
 use thiserror::Error;
@@ -15,6 +16,7 @@ use tokio::sync::{mpsc, oneshot};
 pub trait TxPoolDb:
     Storage<UtxoId, Coin, Error = KvStoreError>
     + Storage<ContractId, Contract, Error = DbStateError>
+    + Storage<MessageId, DaMessage, Error = KvStoreError>
     + Send
     + Sync
 {
@@ -25,6 +27,11 @@ pub trait TxPoolDb:
     fn contract_exist(&self, contract_id: ContractId) -> Result<bool, DbStateError> {
         Storage::<ContractId, Contract>::contains_key(self, &contract_id)
     }
+
+    fn message(&self, message_id: MessageId) -> Result<Option<DaMessage>, KvStoreError> {
+        Storage::<MessageId, DaMessage>::get(self, &message_id)
+            .map(|t| t.map(|t| t.as_ref().clone()))
+    }
 }
 
 #[derive(Clone, Deref, DerefMut)]
@@ -164,6 +171,10 @@ pub enum Error {
         "Transaction is not inserted. More priced tx has created contract with ContractId {0:#x}"
     )]
     NotInsertedCollisionContractId(ContractId),
+    #[error(
+        "Transaction is not inserted. A higher priced tx {0:#x} is already spending this messageId: {1:#x}"
+    )]
+    NotInsertedCollisionMessageId(TxId, MessageId),
     #[error("Transaction is not inserted. Dependent UTXO output is not existing: {0:#x}")]
     NotInsertedOutputNotExisting(UtxoId),
     #[error("Transaction is not inserted. UTXO input contract is not existing: {0:#x}")]
@@ -174,6 +185,10 @@ pub enum Error {
     NotInsertedInputUtxoIdNotExisting(UtxoId),
     #[error("Transaction is not inserted. UTXO is spent: {0:#x}")]
     NotInsertedInputUtxoIdSpent(UtxoId),
+    #[error("Transaction is not inserted. Message is spent: {0:#x}")]
+    NotInsertedInputMessageIdSpent(MessageId),
+    #[error("Transaction is not inserted. Message id {0:#x} does not match any received message from the DA layer.")]
+    NotInsertedInputMessageUnknown(MessageId),
     #[error(
         "Transaction is not inserted. UTXO requires Contract input {0:#x} that is priced lower"
     )]
@@ -184,6 +199,8 @@ pub enum Error {
     NotInsertedIoWrongAmount,
     #[error("Transaction is not inserted. Input output mismatch. Coin output asset_id does not match expected inputs")]
     NotInsertedIoWrongAssetId,
+    #[error("Transaction is not inserted. The computed message id doesn't match the provided message id.")]
+    NotInsertedIoWrongMessageId,
     #[error(
         "Transaction is not inserted. Input output mismatch. Expected coin but output is contract"
     )]

fuel-core/src/database/message.rs

@@ -1,31 +1,31 @@
 use crate::database::{columns, Database, KvStoreError};
 use fuel_core_interfaces::{
-    common::{fuel_storage::Storage, fuel_types::Bytes32},
+    common::{fuel_storage::Storage, fuel_types::MessageId},
     model::DaMessage,
 };
 use std::borrow::Cow;
 
-impl Storage<Bytes32, DaMessage> for Database {
+impl Storage<MessageId, DaMessage> for Database {
     type Error = KvStoreError;
 
     fn insert(
         &mut self,
-        key: &Bytes32,
+        key: &MessageId,
         value: &DaMessage,
     ) -> Result<Option<DaMessage>, KvStoreError> {
         Database::insert(self, key.as_ref(), columns::DA_MESSAGES, value.clone())
             .map_err(Into::into)
     }
 
-    fn remove(&mut self, key: &Bytes32) -> Result<Option<DaMessage>, KvStoreError> {
+    fn remove(&mut self, key: &MessageId) -> Result<Option<DaMessage>, KvStoreError> {
         Database::remove(self, key.as_ref(), columns::DA_MESSAGES).map_err(Into::into)
     }
 
-    fn get(&self, key: &Bytes32) -> Result<Option<Cow<DaMessage>>, KvStoreError> {
+    fn get(&self, key: &MessageId) -> Result<Option<Cow<DaMessage>>, KvStoreError> {
         Database::get(self, key.as_ref(), columns::DA_MESSAGES).map_err(Into::into)
     }
 
-    fn contains_key(&self, key: &Bytes32) -> Result<bool, KvStoreError> {
+    fn contains_key(&self, key: &MessageId) -> Result<bool, KvStoreError> {
         Database::exists(self, key.as_ref(), columns::DA_MESSAGES).map_err(Into::into)
     }
 }

fuel-relayer/src/finalization_queue.rs

@@ -6,7 +6,7 @@ use crate::{
 use ethers_core::types::{Log, H160};
 use ethers_providers::Middleware;
 use fuel_core_interfaces::{
-    common::fuel_tx::{Address, Bytes32},
+    common::{fuel_tx::Address, fuel_types::MessageId},
     model::{
         BlockHeight, CheckedDaMessage, ConsensusId, DaBlockHeight, DaMessage, SealedFuelBlock,
         ValidatorId, ValidatorStake,
@@ -43,7 +43,7 @@ pub struct DaBlockDiff {
     // Delegation diff contains new delegation list, if we did just withdrawal option will be None.
     pub delegations: HashMap<Address, Option<HashMap<ValidatorId, ValidatorStake>>>,
     /// bridge messages (e.g. erc20 or nft assets)
-    pub messages: HashMap<Bytes32, CheckedDaMessage>,
+    pub messages: HashMap<MessageId, CheckedDaMessage>,
 }
 
 impl DaBlockDiff {