TransactionSource: specify maximum number of transactions to be fetched

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ### Added
 - [2131](https://github.com/FuelLabs/fuel-core/pull/2131): Add flow in TxPool in order to ask to newly connected peers to share their transaction pool
+- [2182](https://github.com/FuelLabs/fuel-core/pull/2151): Limit number of transactions that can be fetched via TxSource::next
 
 ### Changed
 

crates/fuel-core/src/executor.rs

@@ -6,7 +6,10 @@ mod tests {
     use crate as fuel_core;
     use fuel_core::database::Database;
     use fuel_core_executor::{
-        executor::OnceTransactionsSource,
+        executor::{
+            OnceTransactionsSource,
+            MAX_TX_COUNT,
+        },
         ports::{
             MaybeCheckedTransaction,
             RelayerPort,
@@ -2983,6 +2986,42 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn block_producer_never_includes_more_than_max_tx_count_transactions() {
+        let block_height = 1u32;
+        let block_da_height = 2u64;
+
+        let mut consensus_parameters = ConsensusParameters::default();
+
+        // Given
+        let transactions_in_tx_source = (MAX_TX_COUNT 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
+        assert_eq!(
+            result.block.transactions().len(),
+            (MAX_TX_COUNT as usize + 1)
+        );
+    }
+
     #[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

@@ -162,6 +162,10 @@ use alloc::{
     vec::Vec,
 };
 
+/// The maximum amount of transactions that can be included in a block,
+/// excluding the mint transaction.
+pub const MAX_TX_COUNT: u16 = u16::MAX.saturating_sub(1);
+
 pub struct OnceTransactionsSource {
     transactions: ParkingMutex<Vec<MaybeCheckedTransaction>>,
 }
@@ -186,9 +190,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,14 +577,19 @@ 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 u16::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 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()
+            .take(remaining_tx_count as usize)
             .peekable();
         while regular_tx_iter.peek().is_some() {
             for transaction in regular_tx_iter {
@@ -597,11 +614,13 @@ 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
                 .next(remaining_gas_limit, remaining_tx_count, remaining_size)
                 .into_iter()
+                .take(remaining_tx_count as usize)
                 .peekable();
         }
 

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

@@ -41,6 +41,7 @@ use fuel_core_types::{
         txpool::{
             ArcPoolTx,
             InsertionResult,
+            PoolTransaction,
             TransactionStatus,
         },
     },
@@ -391,10 +392,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());

tests/test-helpers/src/lib.rs

@@ -3,15 +3,22 @@ use fuel_core_client::client::{
     FuelClient,
 };
 use fuel_core_types::{
+    fuel_asm::{
+        op,
+        RegId,
+    },
     fuel_crypto::SecretKey,
     fuel_tx::{
         Output,
+        Transaction,
         TransactionBuilder,
     },
 };
 use rand::{
-    prelude::StdRng,
+    rngs::StdRng,
+    CryptoRng,
     Rng,
+    RngCore,
 };
 
 pub mod builder;
@@ -26,6 +33,31 @@ pub async fn send_graph_ql_query(url: &str, query: &str) -> String {
     response.text().await.unwrap()
 }
 
+pub fn make_tx(
+    rng: &mut (impl CryptoRng + RngCore),
+    i: u64,
+    max_gas_limit: u64,
+) -> Transaction {
+    TransactionBuilder::script(
+        op::ret(RegId::ONE).to_bytes().into_iter().collect(),
+        vec![],
+    )
+    .script_gas_limit(max_gas_limit / 2)
+    .add_unsigned_coin_input(
+        SecretKey::random(rng),
+        rng.gen(),
+        1000 + i,
+        Default::default(),
+        Default::default(),
+    )
+    .add_output(Output::Change {
+        amount: 0,
+        asset_id: Default::default(),
+        to: rng.gen(),
+    })
+    .finalize_as_transaction()
+}
+
 pub async fn produce_block_with_tx(rng: &mut StdRng, client: &FuelClient) {
     let secret = SecretKey::random(rng);
     let script_tx = TransactionBuilder::script(vec![], vec![])

tests/tests/blocks.rs

@@ -50,6 +50,11 @@ use std::{
 };
 use test_helpers::send_graph_ql_query;
 
+use rand::{
+    rngs::StdRng,
+    SeedableRng,
+};
+
 #[tokio::test]
 async fn block() {
     // setup test data in the node
@@ -360,6 +365,8 @@ mod full_block {
         },
         FuelClient,
     };
+    use fuel_core_executor::executor;
+    use fuel_core_types::fuel_types::BlockHeight;
 
     #[derive(cynic::QueryFragment, Debug)]
     #[cynic(
@@ -423,4 +430,69 @@ mod full_block {
         assert_eq!(block.header.height.0, 1);
         assert_eq!(block.transactions.len(), 2 /* mint + our tx */);
     }
+
+    #[tokio::test]
+    async fn too_many_transactions_are_split_in_blocks() {
+        // Given
+        let max_gas_limit = 50_000_000;
+        let mut rng = StdRng::seed_from_u64(2322);
+
+        let local_node_config = Config::local_node();
+        let txpool = fuel_core_txpool::Config {
+            max_tx: usize::MAX,
+            ..local_node_config.txpool
+        };
+        let chain_config = local_node_config.snapshot_reader.chain_config().clone();
+        let mut consensus_parameters = chain_config.consensus_parameters;
+        consensus_parameters.set_block_gas_limit(u64::MAX);
+        let snapshot_reader = local_node_config.snapshot_reader.with_chain_config(
+            fuel_core::chain_config::ChainConfig {
+                consensus_parameters,
+                ..chain_config
+            },
+        );
+
+        let patched_node_config = Config {
+            block_production: Trigger::Never,
+            txpool,
+            snapshot_reader,
+            ..local_node_config
+        };
+
+        let srv = FuelService::new_node(patched_node_config).await.unwrap();
+        let client = FuelClient::from(srv.bound_address);
+
+        let tx_count: u64 = 66_000;
+        let txs = (1..=tx_count)
+            .map(|i| test_helpers::make_tx(&mut rng, i, max_gas_limit))
+            .collect_vec();
+
+        // When
+        for tx in txs.iter() {
+            let _tx_id = client.submit(tx).await.unwrap();
+        }
+
+        // Then
+        let _last_block_height: u32 =
+            client.produce_blocks(2, None).await.unwrap().into();
+        let second_last_block = client
+            .block_by_height(BlockHeight::from(1))
+            .await
+            .unwrap()
+            .expect("Second last block should be defined");
+        let last_block = client
+            .block_by_height(BlockHeight::from(2))
+            .await
+            .unwrap()
+            .expect("Last Block should be defined");
+
+        assert_eq!(
+            second_last_block.transactions.len(),
+            executor::MAX_TX_COUNT as usize + 1 // Mint transaction for one block
+        );
+        assert_eq!(
+            last_block.transactions.len(),
+            (tx_count as usize - (executor::MAX_TX_COUNT as usize)) + 1 /* Mint transaction for second block */
+        );
+    }
 }