feat: limit the number of messages in prepare proposal

app/test/prepare_proposal_test.go

@@ -1,9 +1,19 @@
 package app_test
 
 import (
+	"crypto/rand"
+	"fmt"
 	"testing"
 	"time"
 
+	v3consts "github.com/celestiaorg/celestia-app/v3/pkg/appconsts/v3"
+	"github.com/celestiaorg/celestia-app/v3/pkg/user"
+	"github.com/celestiaorg/celestia-app/v3/test/util/testnode"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/cosmos/cosmos-sdk/x/auth/types"
+	banktypes "github.com/cosmos/cosmos-sdk/x/bank/types"
+	"github.com/stretchr/testify/assert"
+
 	tmrand "github.com/tendermint/tendermint/libs/rand"
 
 	"github.com/cosmos/cosmos-sdk/crypto/hd"
@@ -204,6 +214,126 @@ func TestPrepareProposalFiltering(t *testing.T) {
 	}
 }
 
+func TestPrepareProposalCappingNumberOfTransactions(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping prepare proposal capping number of transactions test in short mode.")
+	}
+	// creating a big number of accounts so that every account
+	// only creates a single transaction. This is for transactions
+	// to be skipped without worrying about the sequence number being
+	// sequential.
+	numberOfAccounts := 8000
+	accounts := make([]string, 0, numberOfAccounts)
+	for i := 0; i < numberOfAccounts; i++ {
+		accounts = append(accounts, fmt.Sprintf("account%d", i))
+	}
+	consensusParams := app.DefaultConsensusParams()
+	consensusParams.Block.MaxBytes = 128
+	testApp, kr := testutil.SetupTestAppWithGenesisValSetAndMaxSquareSize(consensusParams, 128, accounts...)
+	enc := encoding.MakeConfig(app.ModuleEncodingRegisters...)
+
+	addrs := make([]sdk.AccAddress, 0, numberOfAccounts)
+	accs := make([]types.AccountI, 0, numberOfAccounts)
+	signers := make([]*user.Signer, 0, numberOfAccounts)
+	for index, account := range accounts {
+		addr := testfactory.GetAddress(kr, account)
+		addrs = append(addrs, addr)
+		acc := testutil.DirectQueryAccount(testApp, addrs[index])
+		accs = append(accs, acc)
+		signer, err := user.NewSigner(kr, enc.TxConfig, testutil.ChainID, appconsts.LatestVersion, user.NewAccount(account, acc.GetAccountNumber(), acc.GetSequence()))
+		require.NoError(t, err)
+		signers = append(signers, signer)
+	}
+
+	numberOfPFBs := v3consts.PFBTransactionCap + 500
+	pfbTxs := make([][]byte, 0, numberOfPFBs)
+	randomBytes := make([]byte, 2000)
+	_, err := rand.Read(randomBytes)
+	require.NoError(t, err)
+	accountIndex := 0
+	for i := 0; i < numberOfPFBs; i++ {
+		blob, err := share.NewBlob(share.RandomNamespace(), randomBytes, 1, accs[accountIndex].GetAddress().Bytes())
+		require.NoError(t, err)
+		tx, _, err := signers[accountIndex].CreatePayForBlobs(accounts[accountIndex], []*share.Blob{blob}, user.SetGasLimit(2549760000), user.SetFee(10000))
+		require.NoError(t, err)
+		pfbTxs = append(pfbTxs, tx)
+		accountIndex++
+	}
+
+	numberOfMsgSends := v3consts.MsgSendTransactionCap + 500
+	msgSendTxs := make([][]byte, 0, numberOfMsgSends)
+	for i := 0; i < numberOfMsgSends; i++ {
+		msg := banktypes.NewMsgSend(
+			addrs[accountIndex],
+			testnode.RandomAddress().(sdk.AccAddress),
+			sdk.NewCoins(sdk.NewInt64Coin(appconsts.BondDenom, 10)),
+		)
+		rawTx, err := signers[accountIndex].CreateTx([]sdk.Msg{msg}, user.SetGasLimit(1000000), user.SetFee(10))
+		require.NoError(t, err)
+		msgSendTxs = append(msgSendTxs, rawTx)
+		accountIndex++
+	}
+
+	testCases := []struct {
+		name                 string
+		inputTransactions    [][]byte
+		expectedTransactions [][]byte
+	}{
+		{
+			name:                 "capping only PFB transactions",
+			inputTransactions:    pfbTxs[:v3consts.PFBTransactionCap+50],
+			expectedTransactions: pfbTxs[:v3consts.PFBTransactionCap],
+		},
+		{
+			name:                 "capping only msg send transactions",
+			inputTransactions:    msgSendTxs[:v3consts.MsgSendTransactionCap+50],
+			expectedTransactions: msgSendTxs[:v3consts.MsgSendTransactionCap],
+		},
+		{
+			name: "capping msg send after pfb transactions",
+			inputTransactions: func() [][]byte {
+				input := make([][]byte, 0, len(msgSendTxs)+100)
+				input = append(input, pfbTxs[:100]...)
+				input = append(input, msgSendTxs...)
+				return input
+			}(),
+			expectedTransactions: func() [][]byte {
+				expected := make([][]byte, 0, v3consts.MsgSendTransactionCap+100)
+				expected = append(expected, msgSendTxs[:v3consts.MsgSendTransactionCap]...)
+				expected = append(expected, pfbTxs[:100]...)
+				return expected
+			}(),
+		},
+		{
+			name: "capping pfb after msg send transactions",
+			inputTransactions: func() [][]byte {
+				input := make([][]byte, 0, len(pfbTxs)+100)
+				input = append(input, msgSendTxs[:100]...)
+				input = append(input, pfbTxs...)
+				return input
+			}(),
+			expectedTransactions: func() [][]byte {
+				expected := make([][]byte, 0, v3consts.PFBTransactionCap+100)
+				expected = append(expected, msgSendTxs[:100]...)
+				expected = append(expected, pfbTxs[:v3consts.PFBTransactionCap]...)
+				return expected
+			}(),
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			resp := testApp.PrepareProposal(abci.RequestPrepareProposal{
+				BlockData: &tmproto.Data{
+					Txs: testCase.inputTransactions,
+				},
+				ChainId: testApp.GetChainID(),
+				Height:  10,
+			})
+			assert.Equal(t, testCase.expectedTransactions, resp.BlockData.Txs)
+		})
+	}
+}
+
 func queryAccountInfo(capp *app.App, accs []string, kr keyring.Keyring) []blobfactory.AccountInfo {
 	infos := make([]blobfactory.AccountInfo, len(accs))
 	for i, acc := range accs {

app/validate_txs.go

@@ -1,10 +1,13 @@
 package app
 
 import (
+	v3consts "github.com/celestiaorg/celestia-app/v3/pkg/appconsts/v3"
+	types2 "github.com/celestiaorg/celestia-app/v3/x/blob/types"
 	"github.com/celestiaorg/go-square/v2/tx"
 	"github.com/cosmos/cosmos-sdk/client"
 	"github.com/cosmos/cosmos-sdk/telemetry"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/cosmos/cosmos-sdk/x/bank/types"
 	tmbytes "github.com/tendermint/tendermint/libs/bytes"
 	"github.com/tendermint/tendermint/libs/log"
 	coretypes "github.com/tendermint/tendermint/types"
@@ -44,12 +47,22 @@ func FilterTxs(logger log.Logger, ctx sdk.Context, handler sdk.AnteHandler, txCo
 // function used to apply the ante handler.
 func filterStdTxs(logger log.Logger, dec sdk.TxDecoder, ctx sdk.Context, handler sdk.AnteHandler, txs [][]byte) ([][]byte, sdk.Context) {
 	n := 0
+	msgSendTransactionCount := 0
 	for _, tx := range txs {
 		sdkTx, err := dec(tx)
 		if err != nil {
 			logger.Error("decoding already checked transaction", "tx", tmbytes.HexBytes(coretypes.Tx(tx).Hash()), "error", err)
 			continue
 		}
+		msgTypes, occurrences := msgTypes(sdkTx)
+		if count := occurrences[sdk.MsgTypeURL(&types.MsgSend{})]; count != 0 {
+			if msgSendTransactionCount+count > v3consts.MsgSendTransactionCap {
+				logger.Debug("skipping tx because the msg send transaction cap was reached", "tx", tmbytes.HexBytes(coretypes.Tx(tx).Hash()))
+				continue
+			}
+			msgSendTransactionCount += count
+		}
+
 		ctx, err = handler(ctx, sdkTx, false)
 		// either the transaction is invalid (ie incorrect nonce) and we
 		// simply want to remove this tx, or we're catching a panic from one
@@ -59,7 +72,7 @@ func filterStdTxs(logger log.Logger, dec sdk.TxDecoder, ctx sdk.Context, handler
 				"filtering already checked transaction",
 				"tx", tmbytes.HexBytes(coretypes.Tx(tx).Hash()),
 				"error", err,
-				"msgs", msgTypes(sdkTx),
+				"msgs", msgTypes,
 			)
 			telemetry.IncrCounter(1, "prepare_proposal", "invalid_std_txs")
 			continue
@@ -77,12 +90,21 @@ func filterStdTxs(logger log.Logger, dec sdk.TxDecoder, ctx sdk.Context, handler
 // function used to apply the ante handler.
 func filterBlobTxs(logger log.Logger, dec sdk.TxDecoder, ctx sdk.Context, handler sdk.AnteHandler, txs []*tx.BlobTx) ([]*tx.BlobTx, sdk.Context) {
 	n := 0
+	pfbTransactionCount := 0
 	for _, tx := range txs {
 		sdkTx, err := dec(tx.Tx)
 		if err != nil {
 			logger.Error("decoding already checked blob transaction", "tx", tmbytes.HexBytes(coretypes.Tx(tx.Tx).Hash()), "error", err)
 			continue
 		}
+		_, occurrences := msgTypes(sdkTx)
+		if count := occurrences[sdk.MsgTypeURL(&types2.MsgPayForBlobs{})]; count != 0 {
+			if pfbTransactionCount+count > v3consts.PFBTransactionCap {
+				logger.Debug("skipping tx because the pfb transaction cap was reached", "tx", tmbytes.HexBytes(coretypes.Tx(tx.Tx).Hash()))
+				continue
+			}
+			pfbTransactionCount += count
+		}
 		ctx, err = handler(ctx, sdkTx, false)
 		// either the transaction is invalid (ie incorrect nonce) and we
 		// simply want to remove this tx, or we're catching a panic from one
@@ -102,13 +124,18 @@ func filterBlobTxs(logger log.Logger, dec sdk.TxDecoder, ctx sdk.Context, handle
 	return txs[:n], ctx
 }
 
-func msgTypes(sdkTx sdk.Tx) []string {
+// msgTypes takes an sdk transaction and returns the types of the messages
+// included in it along with.
+func msgTypes(sdkTx sdk.Tx) ([]string, map[string]int) {
 	msgs := sdkTx.GetMsgs()
-	msgNames := make([]string, len(msgs))
-	for i, msg := range msgs {
-		msgNames[i] = sdk.MsgTypeURL(msg)
+	types := make([]string, 0, len(msgs))
+	occurrences := make(map[string]int)
+	for _, msg := range msgs {
+		msgType := sdk.MsgTypeURL(msg)
+		types = append(types, msgType)
+		occurrences[msgType]++
 	}
-	return msgNames
+	return types, occurrences
 }
 
 func encodeBlobTxs(blobTxs []*tx.BlobTx) [][]byte {

pkg/appconsts/v3/app_consts.go

@@ -6,4 +6,9 @@ const (
 	SubtreeRootThreshold int    = 64
 	TxSizeCostPerByte    uint64 = 10
 	GasPerBlobByte       uint32 = 8
+	// MsgSendTransactionCap maximum number of msg send transactions that a block can contain
+	MsgSendTransactionCap = 3200
+
+	// PFBTransactionCap maximum number of PFB messages a block can contain
+	PFBTransactionCap = 2700
 )

test/util/test_app.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
@@ -63,7 +64,17 @@ func (ao EmptyAppOptions) Get(_ string) interface{} {
 // of the app from first genesis account. A no-op logger is set in app.
 func SetupTestAppWithGenesisValSet(cparams *tmproto.ConsensusParams, genAccounts ...string) (*app.App, keyring.Keyring) {
 	testApp, valSet, kr := NewTestAppWithGenesisSet(cparams, genAccounts...)
+	initialiseTestApp(testApp, valSet, cparams)
+	return testApp, kr
+}
+
+func SetupTestAppWithGenesisValSetAndMaxSquareSize(cparams *tmproto.ConsensusParams, maxSquareSize int, genAccounts ...string) (*app.App, keyring.Keyring) {
+	testApp, valSet, kr := NewTestAppWithGenesisSetAndMaxSquareSize(cparams, maxSquareSize, genAccounts...)
+	initialiseTestApp(testApp, valSet, cparams)
+	return testApp, kr
+}
 
+func initialiseTestApp(testApp *app.App, valSet *tmtypes.ValidatorSet, cparams *tmproto.ConsensusParams) {
 	// commit genesis changes
 	testApp.Commit()
 	testApp.BeginBlock(abci.RequestBeginBlock{Header: tmproto.Header{
@@ -76,8 +87,6 @@ func SetupTestAppWithGenesisValSet(cparams *tmproto.ConsensusParams, genAccounts
 			App: cparams.Version.AppVersion,
 		},
 	}})
-
-	return testApp, kr
 }
 
 // NewTestApp creates a new app instance with an empty memDB and a no-op logger.
@@ -178,7 +187,27 @@ func SetupDeterministicGenesisState(testApp *app.App, pubKeys []cryptotypes.PubK
 func NewTestAppWithGenesisSet(cparams *tmproto.ConsensusParams, genAccounts ...string) (*app.App, *tmtypes.ValidatorSet, keyring.Keyring) {
 	testApp := NewTestApp()
 	genesisState, valSet, kr := GenesisStateWithSingleValidator(testApp, genAccounts...)
+	testApp = InitialiseTestAppWithGenesis(testApp, cparams, genesisState)
+	return testApp, valSet, kr
+}
 
+// NewTestAppWithGenesisSetAndMaxSquareSize initializes a new app with genesis accounts and a specific max square size
+// and returns the testApp, validator set and keyring.
+func NewTestAppWithGenesisSetAndMaxSquareSize(cparams *tmproto.ConsensusParams, maxSquareSize int, genAccounts ...string) (*app.App, *tmtypes.ValidatorSet, keyring.Keyring) {
+	testApp := NewTestApp()
+	genesisState, valSet, kr := GenesisStateWithSingleValidator(testApp, genAccounts...)
+
+	// hacky way of changing the gov max square size without changing the consts
+	blobJSON := string(genesisState["blob"])
+	replace := strings.Replace(blobJSON, fmt.Sprintf("%d", appconsts.DefaultGovMaxSquareSize), fmt.Sprintf("%d", maxSquareSize), 1)
+	genesisState["blob"] = json.RawMessage(replace)
+
+	testApp = InitialiseTestAppWithGenesis(testApp, cparams, genesisState)
+	return testApp, valSet, kr
+}
+
+// InitialiseTestAppWithGenesis initializes the provided app with the provided genesis.
+func InitialiseTestAppWithGenesis(testApp *app.App, cparams *tmproto.ConsensusParams, genesisState app.GenesisState) *app.App {
 	stateBytes, err := json.MarshalIndent(genesisState, "", " ")
 	if err != nil {
 		panic(err)
@@ -208,7 +237,7 @@ func NewTestAppWithGenesisSet(cparams *tmproto.ConsensusParams, genAccounts ...s
 			ChainId:         ChainID,
 		},
 	)
-	return testApp, valSet, kr
+	return testApp
 }
 
 // AddDeterministicValidatorToGenesis adds a set of five validators to the genesis.