feature: Introduce a limit to transaction pool size

- Add enum for different transaction outcomes
- Introduce transaction pool size accounting

chain/chunks/src/client.rs

@@ -30,27 +30,55 @@ pub enum ShardsManagerResponse {
     ChunkHeaderReadyForInclusion { chunk_header: ShardChunkHeader, chunk_producer: AccountId },
 }
 
+#[derive(Debug)]
+pub enum InsertTransactionResult {
+    /// Transaction was successfully inserted.
+    Success,
+    /// Transaction is already in the pool.
+    Duplicate,
+    /// Not enough space to fit the transaction.
+    NoSpaceLeft,
+}
+
 pub struct ShardedTransactionPool {
     tx_pools: HashMap<ShardId, TransactionPool>,
 
     /// Useful to make tests deterministic and reproducible,
     /// while keeping the security of randomization of transactions in pool
     rng_seed: RngSeed,
+
+    /// If set, new transactions that bring the size of the pool over this limit will be rejected.
+    /// The size is tracked and enforced separately for each shard.
+    pool_size_limit: Option<u64>,
 }
 
 impl ShardedTransactionPool {
-    pub fn new(rng_seed: RngSeed) -> Self {
+    pub fn new(rng_seed: RngSeed, pool_size_limit: Option<u64>) -> Self {
         TransactionPool::init_metrics();
-        Self { tx_pools: HashMap::new(), rng_seed }
+        Self { tx_pools: HashMap::new(), rng_seed, pool_size_limit }
     }
 
     pub fn get_pool_iterator(&mut self, shard_id: ShardId) -> Option<PoolIteratorWrapper<'_>> {
         self.tx_pools.get_mut(&shard_id).map(|pool| pool.pool_iterator())
     }
 
-    /// Returns true if transaction is not in the pool before call
-    pub fn insert_transaction(&mut self, shard_id: ShardId, tx: SignedTransaction) -> bool {
-        self.pool_for_shard(shard_id).insert_transaction(tx)
+    /// Tries to insert the transaction into the pool for a given shard.
+    pub fn insert_transaction(
+        &mut self,
+        shard_id: ShardId,
+        tx: SignedTransaction,
+    ) -> InsertTransactionResult {
+        if let Some(limit) = self.pool_size_limit {
+            if self.pool_for_shard(shard_id).transaction_size() > limit {
+                return InsertTransactionResult::NoSpaceLeft;
+            }
+        }
+
+        if !self.pool_for_shard(shard_id).insert_transaction(tx) {
+            return InsertTransactionResult::Duplicate;
+        }
+
+        InsertTransactionResult::Success
     }
 
     pub fn remove_transactions(&mut self, shard_id: ShardId, transactions: &[SignedTransaction]) {
@@ -81,6 +109,7 @@ impl ShardedTransactionPool {
         shard_id: ShardId,
         transactions: &[SignedTransaction],
     ) {
+        // TODO(akashin): Enforce size limit here as well.
         self.pool_for_shard(shard_id).reintroduce_transactions(transactions.to_vec());
     }
 }

chain/client/src/client.rs

@@ -217,7 +217,8 @@ impl Client {
             chain.store(),
             chain_config.background_migration_threads,
         )?;
-        let sharded_tx_pool = ShardedTransactionPool::new(rng_seed);
+        let sharded_tx_pool =
+            ShardedTransactionPool::new(rng_seed, config.transaction_pool_size_limit);
         let sync_status = SyncStatus::AwaitingPeers;
         let genesis_block = chain.genesis_block();
         let epoch_sync = EpochSync::new(
@@ -2010,8 +2011,19 @@ impl Client {
             } else {
                 // Transactions only need to be recorded if the node is a validator.
                 if me.is_some() {
-                    self.sharded_tx_pool.insert_transaction(shard_id, tx.clone());
-                    trace!(target: "client", shard_id, "Recorded a transaction.");
+                    match self.sharded_tx_pool.insert_transaction(shard_id, tx.clone()) {
+                        near_chunks::client::InsertTransactionResult::Success => {
+                            trace!(target: "client", shard_id, "Recorded a transaction.");
+                        }
+                        near_chunks::client::InsertTransactionResult::Duplicate => {
+                            // TODO(akashin): Decide what to do in this case.
+                            return Err(Error::Other("Duplicate transaction inserted".to_string()));
+                        }
+                        near_chunks::client::InsertTransactionResult::NoSpaceLeft => {
+                            // TODO(akashin): Introduce a dedicated error code.
+                            return Err(Error::Other("No space left".to_string()));
+                        }
+                    }
                 }
 
                 // Active validator:

core/chain-configs/src/client_config.rs

@@ -251,6 +251,9 @@ pub struct ClientConfig {
     pub state_sync_enabled: bool,
     /// Options for syncing state.
     pub state_sync: StateSyncConfig,
+    /// Limit of the size of per-shard transaction pool measured in bytes. If not set, the size
+    /// will be unbounded.
+    pub transaction_pool_size_limit: Option<u64>,
 }
 
 impl ClientConfig {
@@ -323,6 +326,7 @@ impl ClientConfig {
             flat_storage_creation_period: Duration::from_secs(1),
             state_sync_enabled: false,
             state_sync: StateSyncConfig::default(),
+            transaction_pool_size_limit: None,
         }
     }
 }

nearcore/src/config.rs

@@ -680,6 +680,7 @@ impl NearConfig {
                 flat_storage_creation_period: config.store.flat_storage_creation_period,
                 state_sync_enabled: config.state_sync_enabled.unwrap_or(false),
                 state_sync: config.state_sync.unwrap_or_default(),
+                transaction_pool_size_limit: None,
             },
             network_config: NetworkConfig::new(
                 config.network,