[mempool] estimate in-memory size using BCS raw transaction (aptos-la…

…bs#4193)

### Description

The previous implementation does not work; it would just output the size of the static parts of the struct so all transactions had the same estimate. Instead we replace with BCS raw transaction size. While this does not show an exact allocation of bytes it is a good stand-in estimate -- as BCS does not attempt to compress or otherwise reduce data (e.g., RLE).

The BCS size is cached and later used to compute size on broadcast. So this only requires more work for transactions that end up not being broadcasted.

### Test Plan

Added a unit test that shows relative difference in estimated sizes. This previously failed in main because the sizes outputted were the same.

config/src/config/mempool_config.rs

@@ -37,7 +37,7 @@ impl Default for MempoolConfig {
             max_broadcasts_per_peer: 1,
             mempool_snapshot_interval_secs: 180,
             capacity: 2_000_000,
-            capacity_bytes: 2_147_483_648,
+            capacity_bytes: 2 * 1024 * 1024 * 1024,
             capacity_per_user: 100,
             default_failovers: 3,
             system_transaction_timeout_secs: 600,

mempool/src/core_mempool/transaction.rs

@@ -8,8 +8,12 @@ use aptos_types::{
     transaction::SignedTransaction,
 };
 use serde::{Deserialize, Serialize};
+use std::mem::size_of;
 use std::time::Duration;
 
+/// Estimated per-txn size minus the raw transaction
+pub const TXN_FIXED_ESTIMATED_BYTES: usize = size_of::<MempoolTransaction>();
+
 #[derive(Clone, Debug)]
 pub struct MempoolTransaction {
     pub txn: SignedTransaction,
@@ -49,7 +53,7 @@ impl MempoolTransaction {
         self.txn.clone().committed_hash()
     }
     pub(crate) fn get_estimated_bytes(&self) -> usize {
-        std::mem::size_of_val(self) + TXN_INDEX_ESTIMATED_BYTES
+        self.txn.raw_txn_bytes_len() + TXN_FIXED_ESTIMATED_BYTES + TXN_INDEX_ESTIMATED_BYTES
     }
 }
 
@@ -70,3 +74,60 @@ pub struct SequenceInfo {
     pub transaction_sequence_number: u64,
     pub account_sequence_number_type: AccountSequenceInfo,
 }
+
+#[cfg(test)]
+mod test {
+    use crate::core_mempool::{MempoolTransaction, TimelineState};
+    use aptos_crypto::ed25519::Ed25519PrivateKey;
+    use aptos_crypto::PrivateKey;
+    use aptos_crypto::SigningKey;
+    use aptos_crypto::Uniform;
+    use aptos_types::account_address::AccountAddress;
+    use aptos_types::account_config::AccountSequenceInfo;
+    use aptos_types::chain_id::ChainId;
+    use aptos_types::transaction::{RawTransaction, Script, SignedTransaction, TransactionPayload};
+    use std::time::Duration;
+
+    #[test]
+    fn test_estimated_bytes() {
+        let txn1 = create_test_transaction(0, vec![0x1]);
+        let mempool_txn1 = create_test_mempool_transaction(txn1);
+        let txn2 = create_test_transaction(0, vec![0x1, 0x2]);
+        let mempool_txn2 = create_test_mempool_transaction(txn2);
+
+        assert!(mempool_txn1.get_estimated_bytes() < mempool_txn2.get_estimated_bytes());
+    }
+
+    fn create_test_mempool_transaction(signed_txn: SignedTransaction) -> MempoolTransaction {
+        MempoolTransaction::new(
+            signed_txn,
+            Duration::from_secs(1),
+            1,
+            TimelineState::NotReady,
+            AccountSequenceInfo::Sequential(0),
+        )
+    }
+
+    /// Creates a signed transaction
+    fn create_test_transaction(sequence_number: u64, code_bytes: Vec<u8>) -> SignedTransaction {
+        let private_key = Ed25519PrivateKey::generate_for_testing();
+        let public_key = private_key.public_key();
+
+        let transaction_payload =
+            TransactionPayload::Script(Script::new(code_bytes, vec![], vec![]));
+        let raw_transaction = RawTransaction::new(
+            AccountAddress::random(),
+            sequence_number,
+            transaction_payload,
+            0,
+            0,
+            0,
+            ChainId::new(10),
+        );
+        SignedTransaction::new(
+            raw_transaction.clone(),
+            public_key,
+            private_key.sign(&raw_transaction),
+        )
+    }
+}