chore(parallel-executor): compressed coin validation

crates/services/parallel-executor/src/coin.rs

@@ -0,0 +1,165 @@
+use fuel_core_types::{
+    entities::coins::coin::{
+        CompressedCoin,
+        CompressedCoinV1,
+    },
+    fuel_tx::{
+        Address,
+        AssetId,
+        Input,
+        UtxoId,
+        Word,
+        input::coin::{
+            CoinPredicate,
+            CoinSigned,
+        },
+    },
+};
+
+/// can either be a predicate coin or a signed coin
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+pub(crate) enum Variant {
+    Predicate,
+    Signed,
+}
+
+#[derive(Debug, Eq)]
+pub(crate) struct CoinInBatch {
+    /// The utxo id
+    utxo_id: UtxoId,
+    /// The index of the transaction using this coin in the batch
+    idx: usize,
+    /// the owner of the coin
+    owner: Address,
+    /// the amount stored in the coin
+    amount: Word,
+    /// the asset the coin stores
+    asset_id: AssetId,
+    /// variant
+    variant: Variant,
+}
+
+impl PartialEq for CoinInBatch {
+    fn eq(&self, other: &Self) -> bool {
+        self.utxo() == other.utxo()
+            && self.owner() == other.owner()
+            && self.amount() == other.amount()
+            && self.asset_id() == other.asset_id()
+            && self.variant() == other.variant()
+        // we don't include the idx here
+    }
+}
+
+impl CoinInBatch {
+    pub(crate) fn utxo(&self) -> &UtxoId {
+        &self.utxo_id
+    }
+
+    pub(crate) fn idx(&self) -> usize {
+        self.idx
+    }
+
+    pub(crate) fn owner(&self) -> &Address {
+        &self.owner
+    }
+
+    pub(crate) fn amount(&self) -> &Word {
+        &self.amount
+    }
+
+    pub(crate) fn asset_id(&self) -> &AssetId {
+        &self.asset_id
+    }
+
+    pub(crate) fn variant(&self) -> Variant {
+        self.variant
+    }
+
+    pub(crate) fn from_signed_coin(signed_coin: &CoinSigned, idx: usize) -> Self {
+        let CoinSigned {
+            utxo_id,
+            owner,
+            amount,
+            asset_id,
+            ..
+        } = signed_coin;
+
+        CoinInBatch {
+            utxo_id: *utxo_id,
+            idx,
+            owner: *owner,
+            amount: *amount,
+            asset_id: *asset_id,
+            variant: Variant::Signed,
+        }
+    }
+
+    pub(crate) fn from_predicate_coin(
+        predicate_coin: &CoinPredicate,
+        idx: usize,
+    ) -> Self {
+        let CoinPredicate {
+            utxo_id,
+            owner,
+            amount,
+            asset_id,
+            ..
+        } = predicate_coin;
+
+        CoinInBatch {
+            utxo_id: *utxo_id,
+            idx,
+            owner: *owner,
+            amount: *amount,
+            asset_id: *asset_id,
+            variant: Variant::Predicate,
+        }
+    }
+}
+
+impl From<CoinInBatch> for CompressedCoin {
+    fn from(value: CoinInBatch) -> Self {
+        let CoinInBatch {
+            owner,
+            amount,
+            asset_id,
+            ..
+        } = value;
+
+        CompressedCoin::V1(CompressedCoinV1 {
+            owner,
+            amount,
+            asset_id,
+            tx_pointer: Default::default(), // purposely left blank
+        })
+    }
+}
+
+impl From<&CoinInBatch> for Input {
+    fn from(value: &CoinInBatch) -> Self {
+        let CoinInBatch {
+            utxo_id,
+            owner,
+            amount,
+            asset_id,
+            ..
+        } = value;
+
+        match value.variant {
+            Variant::Signed => Input::CoinSigned(CoinSigned {
+                utxo_id: *utxo_id,
+                owner: *owner,
+                amount: *amount,
+                asset_id: *asset_id,
+                ..Default::default()
+            }),
+            Variant::Predicate => Input::CoinPredicate(CoinPredicate {
+                utxo_id: *utxo_id,
+                owner: *owner,
+                amount: *amount,
+                asset_id: *asset_id,
+                ..Default::default()
+            }),
+        }
+    }
+}

crates/services/parallel-executor/src/lib.rs

@@ -1,4 +1,5 @@
-pub mod column_adapter;
+pub(crate) mod coin;
+pub(crate) mod column_adapter;
 pub mod config;
 pub mod executor;
 pub mod ports;

crates/services/parallel-executor/src/scheduler.rs

@@ -52,6 +52,7 @@ use fxhash::FxHashMap;
 use tokio::runtime::Runtime;
 
 use crate::{
+    coin::CoinInBatch,
     column_adapter::ContractColumnsIterator,
     ports::{
         Filter,
@@ -156,10 +157,10 @@ struct WorkSessionExecutionResult {
     changes: Changes,
     /// The coins created by the worker used to verify the coin dependency chain at the end of execution
     /// We also store the index of the transaction in the batch in case the usage is in the same batch
-    coins_created: Vec<(UtxoId, usize)>,
+    coins_created: Vec<CoinInBatch>,
     /// The coins used by the worker used to verify the coin dependency chain at the end of execution
     /// We also store the index of the transaction in the batch in case the creation is in the same batch
-    coins_used: Arc<[(UtxoId, usize)]>,
+    coins_used: Arc<[CoinInBatch]>,
     /// Contracts used during the execution of the transactions to save the changes for future usage of
     /// the contracts
     contracts_used: Arc<[ContractId]>,
@@ -175,10 +176,10 @@ struct WorkSessionSavedData {
     changes: Changes,
     /// The coins created by the worker used to verify the coin dependency chain at the end of execution
     /// We also store the index of the transaction in the batch in case the usage is in the same batch
-    coins_created: Vec<(UtxoId, usize)>,
+    coins_created: Vec<CoinInBatch>,
     /// The coins used by the worker used to verify the coin dependency chain at the end of execution
     /// We also store the index of the transaction in the batch in case the creation is in the same batch
-    coins_used: Arc<[(UtxoId, usize)]>,
+    coins_used: Arc<[CoinInBatch]>,
     /// The transactions of the batch
     txs: Vec<CheckedTransaction>,
 }
@@ -566,7 +567,7 @@ where
 }
 
 struct CoinDependencyChainVerifier {
-    coins_registered: FxHashMap<UtxoId, (usize, usize)>,
+    coins_registered: FxHashMap<UtxoId, (usize, CoinInBatch)>,
 }
 
 impl CoinDependencyChainVerifier {
@@ -579,54 +580,73 @@ impl CoinDependencyChainVerifier {
     fn register_coins_created(
         &mut self,
         batch_id: usize,
-        coins_created: Vec<(UtxoId, usize)>,
+        coins_created: Vec<CoinInBatch>,
     ) {
-        for (coin, idx) in coins_created {
-            self.coins_registered.insert(coin, (batch_id, idx));
+        for coin in coins_created {
+            self.coins_registered.insert(*coin.utxo(), (batch_id, coin));
         }
     }
 
     fn verify_coins_used<'a, S>(
         &self,
         batch_id: usize,
-        coins_used: impl Iterator<Item = &'a (UtxoId, usize)>,
+        coins_used: impl Iterator<Item = &'a CoinInBatch>,
         storage: &S,
     ) -> Result<(), SchedulerError>
     where
         S: Storage,
     {
-        // TODO: Maybe we also want to verify the amount
-        for (coin, idx) in coins_used {
-            match storage.get_coin(coin) {
-                Ok(Some(_)) => {
+        for coin in coins_used {
+            match storage.get_coin(coin.utxo()) {
+                Ok(Some(db_coin)) => {
                     // Coin is in the database
+                    match db_coin.matches_input(&coin.into()) {
+                        Some(true) => continue,
+                        Some(false) => {
+                            return Err(SchedulerError::InternalError(format!(
+                                "coin is invalid: {}",
+                                coin.utxo(),
+                            )))
+                        }
+                        None => {
+                            return Err(SchedulerError::InternalError(format!(
+                                "not a coin: {}",
+                                coin.utxo(),
+                            )))
+                        }
+                    }
                 }
                 Ok(None) => {
                     // Coin is not in the database
-                    match self.coins_registered.get(coin) {
-                        Some((coin_creation_batch_id, coin_creation_tx_idx)) => {
+                    match self.coins_registered.get(coin.utxo()) {
+                        Some((coin_creation_batch_id, registered_coin)) => {
                             // Coin is in the block
                             if coin_creation_batch_id <= &batch_id
-                                && coin_creation_tx_idx <= idx
+                                && registered_coin.idx() <= coin.idx()
+                                && registered_coin == coin
                             {
                                 // Coin is created in a batch that is before the current one
+                                continue;
                             } else {
                                 // Coin is created in a batch that is after the current one
                                 return Err(SchedulerError::InternalError(format!(
-                                    "Coin {coin} is created in a batch that is after the current one"
+                                    "Coin {} is created in a batch that is after the current one",
+                                    coin.utxo()
                                 )));
                             }
                         }
                         None => {
                             return Err(SchedulerError::InternalError(format!(
-                                "Coin {coin} is not in the database and not created in the block"
+                                "Coin {} is not in the database and not created in the block",
+                                coin.utxo(),
                             )));
                         }
                     }
                 }
                 Err(e) => {
                     return Err(SchedulerError::InternalError(format!(
-                        "Error while getting coin {coin}: {e}"
+                        "Error while getting coin {}: {e}",
+                        coin.utxo(),
                     )));
                 }
             }
@@ -638,7 +658,7 @@ impl CoinDependencyChainVerifier {
 #[allow(clippy::type_complexity)]
 fn get_contracts_and_coins_used(
     batch: &[CheckedTransaction],
-) -> (Arc<[ContractId]>, Arc<[(UtxoId, usize)]>) {
+) -> (Arc<[ContractId]>, Arc<[CoinInBatch]>) {
     let mut contracts_used = vec![];
     let mut coins_used = vec![];
 
@@ -650,10 +670,10 @@ fn get_contracts_and_coins_used(
                     contracts_used.push(contract.contract_id);
                 }
                 fuel_core_types::fuel_tx::Input::CoinSigned(coin) => {
-                    coins_used.push((coin.utxo_id, idx));
+                    coins_used.push(CoinInBatch::from_signed_coin(coin, idx));
                 }
                 fuel_core_types::fuel_tx::Input::CoinPredicate(coin) => {
-                    coins_used.push((coin.utxo_id, idx));
+                    coins_used.push(CoinInBatch::from_predicate_coin(coin, idx));
                 }
                 _ => {}
             }