feat: limit the number of msg send and PFBs 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.