fix: use correct payload length for blob pooled txs (#4972)

crates/net/eth-wire/tests/pooled_transactions.rs

@@ -1,6 +1,6 @@
 //! Decoding tests for [`PooledTransactions`]
-use alloy_rlp::Decodable;
-use reth_eth_wire::PooledTransactions;
+use alloy_rlp::{Decodable, Encodable};
+use reth_eth_wire::{EthVersion, PooledTransactions, ProtocolMessage};
 use reth_primitives::{hex, Bytes, PooledTransactionsElement};
 use std::{fs, path::PathBuf};
 
@@ -10,7 +10,26 @@ fn decode_pooled_transactions_data() {
         PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("testdata/pooled_transactions_with_blob");
     let data = fs::read_to_string(network_data_path).expect("Unable to read file");
     let hex_data = hex::decode(data.trim()).unwrap();
-    let _txs = PooledTransactions::decode(&mut &hex_data[..]).unwrap();
+    let txs = PooledTransactions::decode(&mut &hex_data[..]).unwrap();
+
+    // do a roundtrip test
+    let mut buf = Vec::new();
+    txs.encode(&mut buf);
+    if hex_data != buf {
+        panic!("mixed pooled transaction roundtrip failed");
+    }
+
+    // now do another decoding, on what we encoded - this should succeed
+    let _txs2 = PooledTransactions::decode(&mut &buf[..]).unwrap();
+}
+
+#[test]
+fn decode_request_pair_pooled_blob_transactions() {
+    let network_data_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("testdata/request_pair_pooled_blob_transactions");
+    let data = fs::read_to_string(network_data_path).expect("Unable to read file");
+    let hex_data = hex::decode(data.trim()).unwrap();
+    let _txs = ProtocolMessage::decode_message(EthVersion::Eth68, &mut &hex_data[..]).unwrap();
 }
 
 #[test]

crates/net/network/src/transactions.rs

@@ -267,7 +267,7 @@ where
             return
         }
 
-        trace!(target: "net::tx", "Start propagating transactions");
+        trace!(target: "net::tx", num_hashes=?hashes.len(), "Start propagating transactions");
 
         // This fetches all transaction from the pool, including the blob transactions, which are
         // only ever sent as hashes.
@@ -324,8 +324,9 @@ where
             let mut new_pooled_hashes = hashes.build();
 
             if !new_pooled_hashes.is_empty() {
-                // determine whether to send full tx objects or hashes.
-                if peer_idx > max_num_full {
+                // determine whether to send full tx objects or hashes. If there are no full
+                // transactions, try to send hashes.
+                if peer_idx > max_num_full || full_transactions.is_empty() {
                     // enforce tx soft limit per message for the (unlikely) event the number of
                     // hashes exceeds it
                     new_pooled_hashes.truncate(NEW_POOLED_TRANSACTION_HASHES_SOFT_LIMIT);
@@ -333,6 +334,9 @@ where
                     for hash in new_pooled_hashes.iter_hashes().copied() {
                         propagated.0.entry(hash).or_default().push(PropagateKind::Hash(*peer_id));
                     }
+
+                    trace!(target: "net::tx", ?peer_id, num_txs=?new_pooled_hashes.len(), "Propagating tx hashes to peer");
+
                     // send hashes of transactions
                     self.network.send_transactions_hashes(*peer_id, new_pooled_hashes);
                 } else {
@@ -345,6 +349,9 @@ where
                             .or_default()
                             .push(PropagateKind::Full(*peer_id));
                     }
+
+                    trace!(target: "net::tx", ?peer_id, num_txs=?new_full_transactions.len(), "Propagating full transactions to peer");
+
                     // send full transactions
                     self.network.send_transactions(*peer_id, new_full_transactions);
                 }
@@ -365,9 +372,10 @@ where
         txs: Vec<TxHash>,
         peer_id: PeerId,
     ) -> Option<PropagatedTransactions> {
+        trace!(target: "net::tx", ?peer_id, "Propagating transactions to peer");
+
         let peer = self.peers.get_mut(&peer_id)?;
         let mut propagated = PropagatedTransactions::default();
-        trace!(target: "net::tx", ?peer_id, "Propagating transactions to peer");
 
         // filter all transactions unknown to the peer
         let mut full_transactions = FullTransactionsBuilder::default();
@@ -442,6 +450,7 @@ where
             for hash in new_pooled_hashes.iter_hashes().copied() {
                 propagated.0.entry(hash).or_default().push(PropagateKind::Hash(peer_id));
             }
+
             // send hashes of transactions
             self.network.send_transactions_hashes(peer_id, new_pooled_hashes);
 
@@ -864,6 +873,11 @@ impl FullTransactionsBuilder {
         self.transactions.push(Arc::clone(&transaction.transaction));
     }
 
+    /// Returns whether or not any transactions are in the [FullTransactionsBuilder].
+    fn is_empty(&self) -> bool {
+        self.transactions.is_empty()
+    }
+
     /// returns the list of transactions.
     fn build(self) -> Vec<Arc<TransactionSigned>> {
         self.transactions