feat: loop through all txs before returning the tx set to consensus (#…

…1113)

## Description

This modifies the behavior of the `ReapMaxBytesMaxGas` method called by
consensus. Prior we would return the tx set the moment the first tx was
greater than the limit. Now instead of breaking we continue through all
txs. This means that a large tx doesn't block smaller less priority
transactions from getting through.

mempool/cat/pool.go

@@ -415,12 +415,14 @@ func (txmp *TxPool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs {
 	var keep []types.Tx //nolint:prealloc
 	for _, w := range txmp.allEntriesSorted() {
 		// N.B. When computing byte size, we need to include the overhead for
-		// encoding as protobuf to send to the application.
-		totalGas += w.gasWanted
-		totalBytes += types.ComputeProtoSizeForTxs([]types.Tx{w.tx})
-		if (maxGas >= 0 && totalGas > maxGas) || (maxBytes >= 0 && totalBytes > maxBytes) {
-			break
+		// encoding as protobuf to send to the application. This actually overestimates it
+		// as we add the proto overhead to each transaction
+		txBytes := types.ComputeProtoSizeForTxs([]types.Tx{w.tx})
+		if (maxGas >= 0 && totalGas+w.gasWanted > maxGas) || (maxBytes >= 0 && totalBytes+txBytes > maxBytes) {
+			continue
 		}
+		totalBytes += txBytes
+		totalGas += w.gasWanted
 		keep = append(keep, w.tx)
 	}
 	return keep

mempool/cat/pool_test.go

@@ -374,6 +374,28 @@ func TestTxPool_ReapMaxBytesMaxGas(t *testing.T) {
 	require.Len(t, reapedTxs, 25)
 }
 
+func TestTxMempoolTxLargerThanMaxBytes(t *testing.T) {
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	txmp := setup(t, 0)
+	bigPrefix := make([]byte, 100)
+	_, err := rng.Read(bigPrefix)
+	require.NoError(t, err)
+	// large high priority tx
+	bigTx := []byte(fmt.Sprintf("sender-1-1=%X=2", bigPrefix))
+	smallPrefix := make([]byte, 20)
+	_, err = rng.Read(smallPrefix)
+	require.NoError(t, err)
+	// smaller low priority tx with different sender
+	smallTx := []byte(fmt.Sprintf("sender-2-1=%X=1", smallPrefix))
+	require.NoError(t, txmp.CheckTx(bigTx, nil, mempool.TxInfo{SenderID: 1}))
+	require.NoError(t, txmp.CheckTx(smallTx, nil, mempool.TxInfo{SenderID: 1}))
+
+	// reap by max bytes less than the large tx
+	reapedTxs := txmp.ReapMaxBytesMaxGas(100, -1)
+	require.Len(t, reapedTxs, 1)
+	require.Equal(t, types.Tx(smallTx), reapedTxs[0])
+}
+
 func TestTxPool_ReapMaxTxs(t *testing.T) {
 	txmp := setup(t, 0)
 	txs := checkTxs(t, txmp, 100, 0)

mempool/v1/mempool.go

@@ -330,12 +330,14 @@ func (txmp *TxMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs {
 	var keep []types.Tx //nolint:prealloc
 	for _, w := range txmp.allEntriesSorted() {
 		// N.B. When computing byte size, we need to include the overhead for
-		// encoding as protobuf to send to the application.
-		totalGas += w.gasWanted
-		totalBytes += types.ComputeProtoSizeForTxs([]types.Tx{w.tx})
-		if (maxGas >= 0 && totalGas > maxGas) || (maxBytes >= 0 && totalBytes > maxBytes) {
-			break
+		// encoding as protobuf to send to the application. This actually overestimates it
+		// as we add the proto overhead to each transaction
+		txBytes := types.ComputeProtoSizeForTxs([]types.Tx{w.tx})
+		if (maxGas >= 0 && totalGas+w.gasWanted > maxGas) || (maxBytes >= 0 && totalBytes+txBytes > maxBytes) {
+			continue
 		}
+		totalBytes += txBytes
+		totalGas += w.gasWanted
 		keep = append(keep, w.tx)
 	}
 	return keep

mempool/v1/mempool_test.go

@@ -355,6 +355,28 @@ func TestTxMempool_ReapMaxBytesMaxGas(t *testing.T) {
 	require.Len(t, reapedTxs, 25)
 }
 
+func TestTxMempoolTxLargerThanMaxBytes(t *testing.T) {
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	txmp := setup(t, 0)
+	bigPrefix := make([]byte, 100)
+	_, err := rng.Read(bigPrefix)
+	require.NoError(t, err)
+	// large high priority tx
+	bigTx := []byte(fmt.Sprintf("sender-1-1=%X=2", bigPrefix))
+	smallPrefix := make([]byte, 20)
+	_, err = rng.Read(smallPrefix)
+	require.NoError(t, err)
+	// smaller low priority tx with different sender
+	smallTx := []byte(fmt.Sprintf("sender-2-1=%X=1", smallPrefix))
+	require.NoError(t, txmp.CheckTx(bigTx, nil, mempool.TxInfo{SenderID: 1}))
+	require.NoError(t, txmp.CheckTx(smallTx, nil, mempool.TxInfo{SenderID: 1}))
+
+	// reap by max bytes less than the large tx
+	reapedTxs := txmp.ReapMaxBytesMaxGas(100, -1)
+	require.Len(t, reapedTxs, 1)
+	require.Equal(t, types.Tx(smallTx), reapedTxs[0])
+}
+
 func TestTxMempool_ReapMaxTxs(t *testing.T) {
 	txmp := setup(t, 0)
 	tTxs := checkTxs(t, txmp, 100, 0)