feat: limit the number of messages in prepare proposal

Makefile

@@ -152,7 +152,7 @@ test-race:
 # TODO: Remove the -skip flag once the following tests no longer contain data races.
 # https://github.com/celestiaorg/celestia-app/issues/1369
 	@echo "--> Running tests in race mode"
-	@go test ./... -v -race -skip "TestPrepareProposalConsistency|TestIntegrationTestSuite|TestBlobstreamRPCQueries|TestSquareSizeIntegrationTest|TestStandardSDKIntegrationTestSuite|TestTxsimCommandFlags|TestTxsimCommandEnvVar|TestMintIntegrationTestSuite|TestBlobstreamCLI|TestUpgrade|TestMaliciousTestNode|TestBigBlobSuite|TestQGBIntegrationSuite|TestSignerTestSuite|TestPriorityTestSuite|TestTimeInPrepareProposalContext|TestBlobstream|TestCLITestSuite|TestLegacyUpgrade|TestSignerTwins|TestConcurrentTxSubmission|TestTxClientTestSuite|Test_testnode|TestEvictions"
+	@go test -timeout 15m ./... -v -race -skip "TestPrepareProposalConsistency|TestIntegrationTestSuite|TestBlobstreamRPCQueries|TestSquareSizeIntegrationTest|TestStandardSDKIntegrationTestSuite|TestTxsimCommandFlags|TestTxsimCommandEnvVar|TestMintIntegrationTestSuite|TestBlobstreamCLI|TestUpgrade|TestMaliciousTestNode|TestBigBlobSuite|TestQGBIntegrationSuite|TestSignerTestSuite|TestPriorityTestSuite|TestTimeInPrepareProposalContext|TestBlobstream|TestCLITestSuite|TestLegacyUpgrade|TestSignerTwins|TestConcurrentTxSubmission|TestTxClientTestSuite|Test_testnode|TestEvictions"
 .PHONY: test-race
 
 ## test-bench: Run unit tests in bench mode.

app/test/prepare_proposal_test.go

@@ -1,10 +1,21 @@
 package app_test
 
 import (
+	"crypto/rand"
 	"strings"
 	"testing"
 	"time"
 
+	blobtypes "github.com/celestiaorg/celestia-app/v3/x/blob/types"
+	blobtx "github.com/celestiaorg/go-square/v2/tx"
+
+	"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"
@@ -17,7 +28,6 @@ import (
 	"github.com/celestiaorg/celestia-app/v3/app"
 	"github.com/celestiaorg/celestia-app/v3/app/encoding"
 	"github.com/celestiaorg/celestia-app/v3/pkg/appconsts"
-	"github.com/celestiaorg/celestia-app/v3/pkg/user"
 	testutil "github.com/celestiaorg/celestia-app/v3/test/util"
 	"github.com/celestiaorg/celestia-app/v3/test/util/blobfactory"
 	"github.com/celestiaorg/celestia-app/v3/test/util/testfactory"
@@ -260,6 +270,148 @@ func TestPrepareProposalFiltering(t *testing.T) {
 	}
 }
 
+func TestPrepareProposalCappingNumberOfMessages(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 := testnode.GenerateAccounts(numberOfAccounts)
+	consensusParams := app.DefaultConsensusParams()
+	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 := appconsts.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++
+	}
+
+	multiPFBsPerTxs := make([][]byte, 0, numberOfPFBs)
+	numberOfMsgsPerTx := 10
+	for i := 0; i < numberOfPFBs; i++ {
+		msgs := make([]sdk.Msg, 0)
+		blobs := make([]*share.Blob, 0)
+		for j := 0; j < numberOfMsgsPerTx; j++ {
+			blob, err := share.NewBlob(share.RandomNamespace(), randomBytes, 1, accs[accountIndex].GetAddress().Bytes())
+			require.NoError(t, err)
+			msg, err := blobtypes.NewMsgPayForBlobs(addrs[accountIndex].String(), appconsts.LatestVersion, blob)
+			require.NoError(t, err)
+			msgs = append(msgs, msg)
+			blobs = append(blobs, blob)
+		}
+		txBytes, err := signers[accountIndex].CreateTx(msgs, user.SetGasLimit(2549760000), user.SetFee(10000))
+		require.NoError(t, err)
+		blobTx, err := blobtx.MarshalBlobTx(txBytes, blobs...)
+		require.NoError(t, err)
+		multiPFBsPerTxs = append(multiPFBsPerTxs, blobTx)
+		accountIndex++
+	}
+
+	numberOfMsgSends := appconsts.NonPFBTransactionCap + 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[:appconsts.PFBTransactionCap+50],
+			expectedTransactions: pfbTxs[:appconsts.PFBTransactionCap],
+		},
+		{
+			name:                 "capping only PFB transactions with multiple messages",
+			inputTransactions:    multiPFBsPerTxs[:appconsts.PFBTransactionCap],
+			expectedTransactions: multiPFBsPerTxs[:appconsts.PFBTransactionCap/numberOfMsgsPerTx],
+		},
+		{
+			name:                 "capping only msg send transactions",
+			inputTransactions:    msgSendTxs[:appconsts.NonPFBTransactionCap+50],
+			expectedTransactions: msgSendTxs[:appconsts.NonPFBTransactionCap],
+		},
+		{
+			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, appconsts.NonPFBTransactionCap+100)
+				expected = append(expected, msgSendTxs[:appconsts.NonPFBTransactionCap]...)
+				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, appconsts.PFBTransactionCap+100)
+				expected = append(expected, msgSendTxs[:100]...)
+				expected = append(expected, pfbTxs[:appconsts.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,6 +1,7 @@
 package app
 
 import (
+	"github.com/celestiaorg/celestia-app/v3/pkg/appconsts"
 	"github.com/celestiaorg/go-square/v2/tx"
 	"github.com/cosmos/cosmos-sdk/client"
 	"github.com/cosmos/cosmos-sdk/telemetry"
@@ -44,6 +45,7 @@ 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
+	nonPFBTransactionsCount := 0
 	for _, tx := range txs {
 		sdkTx, err := dec(tx)
 		if err != nil {
@@ -54,6 +56,13 @@ func filterStdTxs(logger log.Logger, dec sdk.TxDecoder, ctx sdk.Context, handler
 		// Set the tx size on the context before calling the AnteHandler
 		ctx = ctx.WithTxBytes(tx)
 
+		msgTypes := msgTypes(sdkTx)
+		if nonPFBTransactionsCount+len(sdkTx.GetMsgs()) > appconsts.NonPFBTransactionCap {
+			logger.Debug("skipping tx because the sdk message cap was reached", "tx", tmbytes.HexBytes(coretypes.Tx(tx).Hash()))
+			continue
+		}
+		nonPFBTransactionsCount += len(sdkTx.GetMsgs())
+
 		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
@@ -63,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
@@ -81,6 +90,7 @@ 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 {
@@ -91,6 +101,12 @@ func filterBlobTxs(logger log.Logger, dec sdk.TxDecoder, ctx sdk.Context, handle
 		// Set the tx size on the context before calling the AnteHandler
 		ctx = ctx.WithTxBytes(tx.Tx)
 
+		if pfbTransactionCount+len(sdkTx.GetMsgs()) > appconsts.PFBTransactionCap {
+			logger.Debug("skipping tx because the pfb transaction cap was reached", "tx", tmbytes.HexBytes(coretypes.Tx(tx.Tx).Hash()))
+			continue
+		}
+		pfbTransactionCount += len(sdkTx.GetMsgs())
+
 		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

pkg/appconsts/global_consts.go

@@ -67,3 +67,12 @@ func UpgradeHeightDelay() int64 {
 func HashLength() int {
 	return hashLength
 }
+
+// The following consts are not consensus breaking and will be applied straight after this binary is started.
+const (
+	// NonPFBTransactionCap is the maximum number of SDK messages, aside from PFBs, that a block can contain.
+	NonPFBTransactionCap = 200
+
+	// PFBTransactionCap is the maximum number of PFB messages a block can contain.
+	PFBTransactionCap = 600
+)

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.