Modify TxSource::next to handle the maximum number of transactions

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - [2151](https://github.com/FuelLabs/fuel-core/pull/2151): Added limitations on gas used during dry_run in API.
 - [2188](https://github.com/FuelLabs/fuel-core/pull/2188): Added the new variant `V2` for the `ConsensusParameters` which contains the new `block_transaction_size_limit` parameter.
 - [2163](https://github.com/FuelLabs/fuel-core/pull/2163): Added runnable task for fetching block committer data.
+- [2182](https://github.com/FuelLabs/fuel-core/pull/2151): Limit number of transactions that can be fetched via TxSource::next
 
 ### Changed
 
@@ -26,6 +27,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - [2158](https://github.com/FuelLabs/fuel-core/pull/2158): Log the public address of the signing key, if it is specified
 - [2188](https://github.com/FuelLabs/fuel-core/pull/2188): Upgraded the `fuel-vm` to `0.57.0`. More information in the [release](https://github.com/FuelLabs/fuel-vm/releases/tag/v0.57.0).
 
+
 ## [Version 0.35.0]
 
 ### Added

crates/fuel-core/src/executor.rs

@@ -2983,6 +2983,41 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn block_producer_never_includes_more_than_u16_max_transactions() {
+        let block_height = 1u32;
+        let block_da_height = 2u64;
+
+        let mut consensus_parameters = ConsensusParameters::default();
+
+        // Given
+        let transactions_in_tx_source = u16::MAX as usize + 10;
+        consensus_parameters.set_block_gas_limit(u64::MAX);
+        let config = Config {
+            consensus_parameters,
+            ..Default::default()
+        };
+
+        // When
+        let block = test_block(
+            block_height.into(),
+            block_da_height.into(),
+            transactions_in_tx_source,
+        );
+        let partial_fuel_block: PartialFuelBlock = block.into();
+
+        let producer = create_executor(Database::default(), config);
+        let (result, _) = producer
+            .produce_without_commit(partial_fuel_block)
+            .unwrap()
+            .into();
+
+        // Then
+        // u16::MAX -1 transactions have been included from the transaction source, plus the mint transaction
+        // In total we expect to see u16::MAX transactions in the block
+        assert_eq!(result.block.transactions().len(), (u16::MAX) as usize);
+    }
+
     #[cfg(feature = "relayer")]
     mod relayer {
         use super::*;

crates/fuel-core/src/service/adapters/executor.rs

@@ -9,11 +9,15 @@ use fuel_core_types::{
 };
 
 impl fuel_core_executor::ports::TransactionsSource for TransactionsSource {
-    // TODO: Use `tx_count_limit` https://github.com/FuelLabs/fuel-core/issues/2114
     // TODO: Use `size_limit` https://github.com/FuelLabs/fuel-core/issues/2133
-    fn next(&self, gas_limit: u64, _: u16, _: u32) -> Vec<MaybeCheckedTransaction> {
+    fn next(
+        &self,
+        gas_limit: u64,
+        transactions_limit: u16,
+        _: u32,
+    ) -> Vec<MaybeCheckedTransaction> {
         self.txpool
-            .select_transactions(gas_limit)
+            .select_transactions(gas_limit, transactions_limit)
             .into_iter()
             .map(|tx| {
                 MaybeCheckedTransaction::CheckedTransaction(

crates/services/executor/src/executor.rs

@@ -186,9 +186,17 @@ impl OnceTransactionsSource {
 }
 
 impl TransactionsSource for OnceTransactionsSource {
-    fn next(&self, _: u64, _: u16, _: u32) -> Vec<MaybeCheckedTransaction> {
+    fn next(
+        &self,
+        _: u64,
+        transactions_limit: u16,
+        _: u32,
+    ) -> Vec<MaybeCheckedTransaction> {
         let mut lock = self.transactions.lock();
-        core::mem::take(lock.as_mut())
+        let transactions: &mut Vec<MaybeCheckedTransaction> = lock.as_mut();
+        // Avoid panicking if we request more transactions than there are in the vector
+        let transactions_limit = (transactions_limit as usize).min(transactions.len());
+        transactions.drain(..transactions_limit).collect()
     }
 }
 
@@ -565,11 +573,16 @@ where
         let block_gas_limit = self.consensus_params.block_gas_limit();
 
         let mut remaining_gas_limit = block_gas_limit.saturating_sub(data.used_gas);
-        // TODO: Handle `remaining_tx_count` https://github.com/FuelLabs/fuel-core/issues/2114
-        let remaining_tx_count = u16::MAX;
         // TODO: Handle `remaining_size` https://github.com/FuelLabs/fuel-core/issues/2133
         let remaining_size = u32::MAX;
 
+        // We allow at most u64::MAX transactions in a block, including the mint transaction.
+        // When processing l2 transactions, we must take into account transactions from the l1
+        // that have been included in the block already (stored in `data.tx_count`), as well
+        // as the final mint transaction.
+        let max_tx_count = u16::MAX.saturating_sub(1);
+        let mut remaining_tx_count = max_tx_count.saturating_sub(data.tx_count);
+
         let mut regular_tx_iter = l2_tx_source
             .next(remaining_gas_limit, remaining_tx_count, remaining_size)
             .into_iter()
@@ -597,6 +610,7 @@ where
                     }
                 }
                 remaining_gas_limit = block_gas_limit.saturating_sub(data.used_gas);
+                remaining_tx_count = max_tx_count.saturating_sub(data.tx_count);
             }
 
             regular_tx_iter = l2_tx_source

crates/services/executor/src/ports.rs

@@ -116,6 +116,8 @@ impl TransactionExt for MaybeCheckedTransaction {
 
 pub trait TransactionsSource {
     /// Returns the next batch of transactions to satisfy the `gas_limit`.
+    /// The returned batch has at most `tx_count_limit` transactions, none
+    /// of which has a size in bytes greater than `size_limit`.
     fn next(
         &self,
         gas_limit: u64,

crates/services/txpool/src/service.rs

@@ -40,6 +40,7 @@ use fuel_core_types::{
         txpool::{
             ArcPoolTx,
             InsertionResult,
+            PoolTransaction,
             TransactionStatus,
         },
     },
@@ -374,10 +375,17 @@ impl<P2P, ViewProvider, WasmChecker, GasPriceProvider, ConsensusProvider, MP>
         self.txpool.lock().find_one(&id)
     }
 
-    pub fn select_transactions(&self, max_gas: u64) -> Vec<ArcPoolTx> {
+    pub fn select_transactions(
+        &self,
+        max_gas: u64,
+        transactions_limit: u16,
+    ) -> Vec<ArcPoolTx> {
         let mut guard = self.txpool.lock();
         let txs = guard.includable();
-        let sorted_txs = select_transactions(txs, max_gas);
+        let sorted_txs: Vec<Arc<PoolTransaction>> = select_transactions(txs, max_gas)
+            .into_iter()
+            .take(transactions_limit as usize)
+            .collect();
 
         for tx in sorted_txs.iter() {
             guard.remove_committed_tx(&tx.id());