mempool: enforce maxMsgSize limit in CheckTx (#3168)

- fixes #3008
- reactor requires encoded messages are less than maxMsgSize
- requires size of tx + amino-overhead to not exceed maxMsgSize

mempool/mempool.go

@@ -65,6 +65,9 @@ var (
 
 	// ErrMempoolIsFull means Tendermint & an application can't handle that much load
 	ErrMempoolIsFull = errors.New("Mempool is full")
+
+	// ErrTxTooLarge means the tx is too big to be sent in a message to other peers
+	ErrTxTooLarge = fmt.Errorf("Tx too large. Max size is %d", maxTxSize)
 )
 
 // ErrPreCheck is returned when tx is too big
@@ -309,6 +312,13 @@ func (mem *Mempool) CheckTx(tx types.Tx, cb func(*abci.Response)) (err error) {
 		return ErrMempoolIsFull
 	}
 
+	// The size of the corresponding amino-encoded TxMessage
+	// can't be larger than the maxMsgSize, otherwise we can't
+	// relay it to peers.
+	if len(tx) > maxTxSize {
+		return ErrTxTooLarge
+	}
+
 	if mem.preCheck != nil {
 		if err := mem.preCheck(tx); err != nil {
 			return ErrPreCheck{err}

mempool/mempool_test.go

@@ -14,10 +14,12 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/stretchr/testify/require"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/tendermint/abci/example/counter"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	abci "github.com/tendermint/tendermint/abci/types"
 	cfg "github.com/tendermint/tendermint/config"
+	cmn "github.com/tendermint/tendermint/libs/common"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/proxy"
 	"github.com/tendermint/tendermint/types"
@@ -394,6 +396,60 @@ func TestMempoolCloseWAL(t *testing.T) {
 	require.Equal(t, 1, len(m3), "expecting the wal match in")
 }
 
+// Size of the amino encoded TxMessage is the length of the
+// encoded byte array, plus 1 for the struct field, plus 4
+// for the amino prefix.
+func txMessageSize(tx types.Tx) int {
+	return amino.ByteSliceSize(tx) + 1 + 4
+}
+
+func TestMempoolMaxMsgSize(t *testing.T) {
+	app := kvstore.NewKVStoreApplication()
+	cc := proxy.NewLocalClientCreator(app)
+	mempl := newMempoolWithApp(cc)
+
+	testCases := []struct {
+		len int
+		err bool
+	}{
+		// check small txs. no error
+		{10, false},
+		{1000, false},
+		{1000000, false},
+
+		// check around maxTxSize
+		// changes from no error to error
+		{maxTxSize - 2, false},
+		{maxTxSize - 1, false},
+		{maxTxSize, false},
+		{maxTxSize + 1, true},
+		{maxTxSize + 2, true},
+
+		// check around maxMsgSize. all error
+		{maxMsgSize - 1, true},
+		{maxMsgSize, true},
+		{maxMsgSize + 1, true},
+	}
+
+	for i, testCase := range testCases {
+		caseString := fmt.Sprintf("case %d, len %d", i, testCase.len)
+
+		tx := cmn.RandBytes(testCase.len)
+		err := mempl.CheckTx(tx, nil)
+		msg := &TxMessage{tx}
+		encoded := cdc.MustMarshalBinaryBare(msg)
+		require.Equal(t, len(encoded), txMessageSize(tx), caseString)
+		if !testCase.err {
+			require.True(t, len(encoded) <= maxMsgSize, caseString)
+			require.NoError(t, err, caseString)
+		} else {
+			require.True(t, len(encoded) > maxMsgSize, caseString)
+			require.Equal(t, err, ErrTxTooLarge, caseString)
+		}
+	}
+
+}
+
 func checksumIt(data []byte) string {
 	h := md5.New()
 	h.Write(data)

mempool/reactor.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	amino "github.com/tendermint/go-amino"
-	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/libs/clist"
 	"github.com/tendermint/tendermint/libs/log"
 
@@ -18,8 +17,10 @@ import (
 const (
 	MempoolChannel = byte(0x30)
 
-	maxMsgSize                 = 1048576 // 1MB TODO make it configurable
+	maxMsgSize = 1048576        // 1MB TODO make it configurable
-	peerCatchupSleepIntervalMS = 100     // If peer is behind, sleep this amount
+	maxTxSize  = maxMsgSize - 8 // account for amino overhead of TxMessage
+
+	peerCatchupSleepIntervalMS = 100 // If peer is behind, sleep this amount
 )
 
 // MempoolReactor handles mempool tx broadcasting amongst peers.
@@ -98,11 +99,6 @@ func (memR *MempoolReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
 	}
 }
 
-// BroadcastTx is an alias for Mempool.CheckTx. Broadcasting itself happens in peer routines.
-func (memR *MempoolReactor) BroadcastTx(tx types.Tx, cb func(*abci.Response)) error {
-	return memR.Mempool.CheckTx(tx, cb)
-}
-
 // PeerState describes the state of a peer.
 type PeerState interface {
 	GetHeight() int64