Merge latest changes from master and fix conflicts

Signed-off-by: Thane Thomson <connect@thanethomson.com>

internal/blocksync/reactor.go

@@ -45,6 +45,7 @@ func GetChannelDescriptor() *p2p.ChannelDescriptor {
 		SendQueueCapacity:   1000,
 		RecvBufferCapacity:  1024,
 		RecvMessageCapacity: MaxMsgSize,
+		Name:                "blockSync",
 	}
 }
 

internal/consensus/mempool_test.go

@@ -317,9 +317,20 @@ func (app *CounterApplication) Commit() abci.ResponseCommit {
 
 func (app *CounterApplication) PrepareProposal(
 	req abci.RequestPrepareProposal) abci.ResponsePrepareProposal {
-	return abci.ResponsePrepareProposal{
-		ModifiedTxStatus: abci.ResponsePrepareProposal_UNMODIFIED,
+
+	trs := make([]*abci.TxRecord, 0, len(req.Txs))
+	var totalBytes int64
+	for _, tx := range req.Txs {
+		totalBytes += int64(len(tx))
+		if totalBytes > req.MaxTxBytes {
+			break
+		}
+		trs = append(trs, &abci.TxRecord{
+			Action: abci.TxRecord_UNMODIFIED,
+			Tx:     tx,
+		})
 	}
+	return abci.ResponsePrepareProposal{TxRecords: trs}
 }
 
 func (app *CounterApplication) ProcessProposal(

internal/consensus/mocks/cons_sync_reactor.go

@@ -4,6 +4,7 @@ package mocks
 
 import (
 	mock "github.com/stretchr/testify/mock"
+
 	state "github.com/tendermint/tendermint/internal/state"
 )
 

internal/consensus/peer_state_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+
 	"github.com/tendermint/tendermint/libs/log"
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
 	"github.com/tendermint/tendermint/types"

internal/consensus/reactor.go

@@ -38,6 +38,7 @@ func getChannelDescriptors() map[p2p.ChannelID]*p2p.ChannelDescriptor {
 			SendQueueCapacity:   64,
 			RecvMessageCapacity: maxMsgSize,
 			RecvBufferCapacity:  128,
+			Name:                "state",
 		},
 		DataChannel: {
 			// TODO: Consider a split between gossiping current block and catchup
@@ -49,6 +50,7 @@ func getChannelDescriptors() map[p2p.ChannelID]*p2p.ChannelDescriptor {
 			SendQueueCapacity:   64,
 			RecvBufferCapacity:  512,
 			RecvMessageCapacity: maxMsgSize,
+			Name:                "data",
 		},
 		VoteChannel: {
 			ID:                  VoteChannel,
@@ -57,6 +59,7 @@ func getChannelDescriptors() map[p2p.ChannelID]*p2p.ChannelDescriptor {
 			SendQueueCapacity:   64,
 			RecvBufferCapacity:  128,
 			RecvMessageCapacity: maxMsgSize,
+			Name:                "vote",
 		},
 		VoteSetBitsChannel: {
 			ID:                  VoteSetBitsChannel,
@@ -65,6 +68,7 @@ func getChannelDescriptors() map[p2p.ChannelID]*p2p.ChannelDescriptor {
 			SendQueueCapacity:   8,
 			RecvBufferCapacity:  128,
 			RecvMessageCapacity: maxMsgSize,
+			Name:                "voteSet",
 		},
 	}
 }

internal/consensus/state.go

@@ -1390,7 +1390,6 @@ func (cs *State) createProposalBlock(ctx context.Context) (*types.Block, error)
 	}
 
 	var commit *types.Commit
-	var votes []*types.Vote
 	switch {
 	case cs.Height == cs.state.InitialHeight:
 		// We're creating a proposal for the first block.
@@ -1400,7 +1399,6 @@ func (cs *State) createProposalBlock(ctx context.Context) (*types.Block, error)
 	case cs.LastCommit.HasTwoThirdsMajority():
 		// Make the commit from LastCommit
 		commit = cs.LastCommit.MakeCommit()
-		votes = cs.LastCommit.GetVotes()
 
 	default: // This shouldn't happen.
 		cs.logger.Error("propose step; cannot propose anything without commit for the previous block")
@@ -1416,7 +1414,11 @@ func (cs *State) createProposalBlock(ctx context.Context) (*types.Block, error)
 
 	proposerAddr := cs.privValidatorPubKey.Address()
 
-	return cs.blockExec.CreateProposalBlock(ctx, cs.Height, cs.state, commit, proposerAddr, votes)
+	ret, err := cs.blockExec.CreateProposalBlock(ctx, cs.Height, cs.state, commit, proposerAddr, cs.LastCommit.GetVotes())
+	if err != nil {
+		panic(err)
+	}
+	return ret, nil
 }
 
 // Enter: `timeoutPropose` after entering Propose.

internal/consensus/state_test.go

@@ -2184,9 +2184,7 @@ func TestPrepareProposalReceivesVoteExtensions(t *testing.T) {
 	m.On("ExtendVote", mock.Anything).Return(abci.ResponseExtendVote{
 		VoteExtension: voteExtensions[0],
 	})
-	m.On("PrepareProposal", mock.Anything).Return(abci.ResponsePrepareProposal{
-		ModifiedTxStatus: abci.ResponsePrepareProposal_UNMODIFIED,
-	}).Once()
+	m.On("PrepareProposal", mock.Anything).Return(abci.ResponsePrepareProposal{}).Once()
 
 	cs1, vss := makeState(ctx, t, makeStateArgs{config: config, application: m})
 	height, round := cs1.Height, cs1.Round
@@ -2233,7 +2231,7 @@ func TestPrepareProposalReceivesVoteExtensions(t *testing.T) {
 	m.On("PrepareProposal", mock.MatchedBy(func(r abci.RequestPrepareProposal) bool {
 		rpp = r
 		return true
-	})).Return(abci.ResponsePrepareProposal{ModifiedTxStatus: abci.ResponsePrepareProposal_UNMODIFIED})
+	})).Return(abci.ResponsePrepareProposal{})
 
 	signAddVotes(ctx, t, cs1, tmproto.PrecommitType, config.ChainID(), types.BlockID{}, vss[1:]...)
 	ensureNewRound(t, newRoundCh, height, round)

internal/evidence/mocks/block_store.go

@@ -4,6 +4,7 @@ package mocks
 
 import (
 	mock "github.com/stretchr/testify/mock"
+
 	types "github.com/tendermint/tendermint/types"
 )
 

internal/evidence/reactor.go

@@ -38,6 +38,7 @@ func GetChannelDescriptor() *p2p.ChannelDescriptor {
 		Priority:            6,
 		RecvMessageCapacity: maxMsgSize,
 		RecvBufferCapacity:  32,
+		Name:                "evidence",
 	}
 }
 

internal/mempool/mempool_bench_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+
 	abciclient "github.com/tendermint/tendermint/abci/client"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	"github.com/tendermint/tendermint/libs/log"

internal/mempool/reactor.go

@@ -106,6 +106,7 @@ func getChannelDescriptor(cfg *config.MempoolConfig) *p2p.ChannelDescriptor {
 		Priority:            5,
 		RecvMessageCapacity: batchMsg.Size(),
 		RecvBufferCapacity:  128,
+		Name:                "mempool",
 	}
 }
 

internal/p2p/channel.go

@@ -60,6 +60,7 @@ type Channel struct {
 	errCh chan<- PeerError // peer error reporting
 
 	messageType proto.Message // the channel's message type, used for unmarshaling
+	name        string
 }
 
 // NewChannel creates a new channel. It is primarily for internal and test
@@ -102,6 +103,8 @@ func (ch *Channel) SendError(ctx context.Context, pe PeerError) error {
 	}
 }
 
+func (ch *Channel) String() string { return fmt.Sprintf("p2p.Channel<%d:%s>", ch.ID, ch.name) }
+
 // Receive returns a new unbuffered iterator to receive messages from ch.
 // The iterator runs until ctx ends.
 func (ch *Channel) Receive(ctx context.Context) *ChannelIterator {

internal/p2p/conn/connection.go

@@ -616,6 +616,10 @@ type ChannelDescriptor struct {
 	// RecvBufferCapacity defines the max buffer size of inbound messages for a
 	// given p2p Channel queue.
 	RecvBufferCapacity int
+
+	// Human readable name of the channel, used in logging and
+	// diagnostics.
+	Name string
 }
 
 func (chDesc ChannelDescriptor) FillDefaults() (filled ChannelDescriptor) {

internal/p2p/pex/reactor.go

@@ -63,6 +63,7 @@ func ChannelDescriptor() *conn.ChannelDescriptor {
 		SendQueueCapacity:   10,
 		RecvMessageCapacity: maxMsgSize,
 		RecvBufferCapacity:  128,
+		Name:                "pex",
 	}
 }
 

internal/p2p/router.go

@@ -281,6 +281,7 @@ func (r *Router) OpenChannel(ctx context.Context, chDesc *ChannelDescriptor) (*C
 	outCh := make(chan Envelope, chDesc.RecvBufferCapacity)
 	errCh := make(chan PeerError, chDesc.RecvBufferCapacity)
 	channel := NewChannel(id, messageType, queue.dequeue(), outCh, errCh)
+	channel.name = chDesc.Name
 
 	var wrapper Wrapper
 	if w, ok := messageType.(Wrapper); ok {

internal/proxy/client.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/go-kit/kit/metrics"
+
 	abciclient "github.com/tendermint/tendermint/abci/client"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	"github.com/tendermint/tendermint/abci/types"

internal/proxy/client_test.go

@@ -14,6 +14,8 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
+	"gotest.tools/assert"
+
 	abciclient "github.com/tendermint/tendermint/abci/client"
 	abcimocks "github.com/tendermint/tendermint/abci/client/mocks"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
@@ -21,7 +23,6 @@ import (
 	"github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/libs/log"
 	tmrand "github.com/tendermint/tendermint/libs/rand"
-	"gotest.tools/assert"
 )
 
 //----------------------------------------

internal/state/execution.go

@@ -122,22 +122,15 @@ func (blockExec *BlockExecutor) CreateProposalBlock(
 		// transaction causing an error.
 		//
 		// Also, the App can simply skip any transaction that could cause any kind of trouble.
-		// Either way, we can not recover in a meaningful way, unless we skip proposing
-		// this block, repair what caused the error and try again. Hence, we panic on
-		// purpose for now.
-		panic(err)
-	}
-	if rpp.IsTxStatusUnknown() {
-		panic(fmt.Sprintf("PrepareProposal responded with ModifiedTxStatus %s", rpp.ModifiedTxStatus.String()))
-	}
-
-	if !rpp.IsTxStatusModified() {
-		return block, nil
+		// Either way, we cannot recover in a meaningful way, unless we skip proposing
+		// this block, repair what caused the error and try again. Hence, we return an
+		// error for now (the production code calling this function is expected to panic).
+		return nil, err
 	}
 	txrSet := types.NewTxRecordSet(rpp.TxRecords)
 
 	if err := txrSet.Validate(maxDataBytes, block.Txs); err != nil {
-		panic(fmt.Errorf("ResponsePrepareProposal validation: %w", err))
+		return nil, err
 	}
 
 	for _, rtx := range txrSet.RemovedTxs() {

internal/state/execution_test.go

@@ -2,6 +2,7 @@ package state_test
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -11,6 +12,7 @@ import (
 	dbm "github.com/tendermint/tm-db"
 
 	abciclient "github.com/tendermint/tendermint/abci/client"
+	abciclientmocks "github.com/tendermint/tendermint/abci/client/mocks"
 	abci "github.com/tendermint/tendermint/abci/types"
 	abcimocks "github.com/tendermint/tendermint/abci/types/mocks"
 	"github.com/tendermint/tendermint/crypto"
@@ -271,7 +273,7 @@ func TestFinalizeBlockByzantineValidators(t *testing.T) {
 
 func TestProcessProposal(t *testing.T) {
 	const height = 2
-	txs := factory.MakeTenTxs(height)
+	txs := factory.MakeNTxs(height, 10)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -617,9 +619,7 @@ func TestEmptyPrepareProposal(t *testing.T) {
 	require.NoError(t, eventBus.Start(ctx))
 
 	app := abcimocks.NewBaseMock()
-	app.On("PrepareProposal", mock.Anything).Return(abci.ResponsePrepareProposal{
-		ModifiedTxStatus: abci.ResponsePrepareProposal_UNMODIFIED,
-	}, nil)
+	app.On("PrepareProposal", mock.Anything).Return(abci.ResponsePrepareProposal{})
 	cc := abciclient.NewLocalClient(logger, app)
 	proxyApp := proxy.New(cc, logger, proxy.NopMetrics())
 	err := proxyApp.Start(ctx)
@@ -656,9 +656,10 @@ func TestEmptyPrepareProposal(t *testing.T) {
 	require.NoError(t, err)
 }
 
-// TestPrepareProposalPanicOnInvalid tests that the block creation logic panics
-// if the ResponsePrepareProposal returned from the application is invalid.
-func TestPrepareProposalPanicOnInvalid(t *testing.T) {
+// TestPrepareProposalErrorOnNonExistingRemoved tests that the block creation logic returns
+// an error if the ResponsePrepareProposal returned from the application marks
+//  a transaction as REMOVED that was not present in the original proposal.
+func TestPrepareProposalErrorOnNonExistingRemoved(t *testing.T) {
 	const height = 2
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -680,7 +681,6 @@ func TestPrepareProposalPanicOnInvalid(t *testing.T) {
 
 	// create an invalid ResponsePrepareProposal
 	rpp := abci.ResponsePrepareProposal{
-		ModifiedTxStatus: abci.ResponsePrepareProposal_MODIFIED,
 		TxRecords: []*abci.TxRecord{
 			{
 				Action: abci.TxRecord_REMOVED,
@@ -688,7 +688,7 @@ func TestPrepareProposalPanicOnInvalid(t *testing.T) {
 			},
 		},
 	}
-	app.On("PrepareProposal", mock.Anything).Return(rpp, nil)
+	app.On("PrepareProposal", mock.Anything).Return(rpp)
 
 	cc := abciclient.NewLocalClient(logger, app)
 	proxyApp := proxy.New(cc, logger, proxy.NopMetrics())
@@ -707,10 +707,9 @@ func TestPrepareProposalPanicOnInvalid(t *testing.T) {
 	)
 	pa, _ := state.Validators.GetByIndex(0)
 	commit, votes := makeValidCommit(ctx, t, height, types.BlockID{}, state.Validators, privVals)
-	require.Panics(t,
-		func() {
-			blockExec.CreateProposalBlock(ctx, height, state, commit, pa, votes) //nolint:errcheck
-		})
+	block, err := blockExec.CreateProposalBlock(ctx, height, state, commit, pa, votes)
+	require.Nil(t, block)
+	require.ErrorContains(t, err, "new transaction incorrectly marked as removed")
 
 	mp.AssertExpectations(t)
 }
@@ -733,7 +732,7 @@ func TestPrepareProposalRemoveTxs(t *testing.T) {
 	evpool := &mocks.EvidencePool{}
 	evpool.On("PendingEvidence", mock.Anything).Return([]types.Evidence{}, int64(0))
 
-	txs := factory.MakeTenTxs(height)
+	txs := factory.MakeNTxs(height, 10)
 	mp := &mpmocks.Mempool{}
 	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs(txs))
 
@@ -744,9 +743,8 @@ func TestPrepareProposalRemoveTxs(t *testing.T) {
 
 	app := abcimocks.NewBaseMock()
 	app.On("PrepareProposal", mock.Anything).Return(abci.ResponsePrepareProposal{
-		ModifiedTxStatus: abci.ResponsePrepareProposal_MODIFIED,
-		TxRecords:        trs,
-	}, nil)
+		TxRecords: trs,
+	})
 
 	cc := abciclient.NewLocalClient(logger, app)
 	proxyApp := proxy.New(cc, logger, proxy.NopMetrics())
@@ -794,7 +792,7 @@ func TestPrepareProposalAddedTxsIncluded(t *testing.T) {
 	evpool := &mocks.EvidencePool{}
 	evpool.On("PendingEvidence", mock.Anything).Return([]types.Evidence{}, int64(0))
 
-	txs := factory.MakeTenTxs(height)
+	txs := factory.MakeNTxs(height, 10)
 	mp := &mpmocks.Mempool{}
 	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs(txs[2:]))
 
@@ -804,9 +802,8 @@ func TestPrepareProposalAddedTxsIncluded(t *testing.T) {
 
 	app := abcimocks.NewBaseMock()
 	app.On("PrepareProposal", mock.Anything).Return(abci.ResponsePrepareProposal{
-		ModifiedTxStatus: abci.ResponsePrepareProposal_MODIFIED,
-		TxRecords:        trs,
-	}, nil)
+		TxRecords: trs,
+	})
 
 	cc := abciclient.NewLocalClient(logger, app)
 	proxyApp := proxy.New(cc, logger, proxy.NopMetrics())
@@ -851,7 +848,7 @@ func TestPrepareProposalReorderTxs(t *testing.T) {
 	evpool := &mocks.EvidencePool{}
 	evpool.On("PendingEvidence", mock.Anything).Return([]types.Evidence{}, int64(0))
 
-	txs := factory.MakeTenTxs(height)
+	txs := factory.MakeNTxs(height, 10)
 	mp := &mpmocks.Mempool{}
 	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs(txs))
 
@@ -861,9 +858,8 @@ func TestPrepareProposalReorderTxs(t *testing.T) {
 
 	app := abcimocks.NewBaseMock()
 	app.On("PrepareProposal", mock.Anything).Return(abci.ResponsePrepareProposal{
-		ModifiedTxStatus: abci.ResponsePrepareProposal_MODIFIED,
-		TxRecords:        trs,
-	}, nil)
+		TxRecords: trs,
+	})
 
 	cc := abciclient.NewLocalClient(logger, app)
 	proxyApp := proxy.New(cc, logger, proxy.NopMetrics())
@@ -892,10 +888,9 @@ func TestPrepareProposalReorderTxs(t *testing.T) {
 
 }
 
-// TestPrepareProposalModifiedTxStatusFalse tests that CreateBlock correctly ignores
-// the ResponsePrepareProposal TxRecords if ResponsePrepareProposal does not
-// set ModifiedTxStatus to true.
-func TestPrepareProposalModifiedTxStatusFalse(t *testing.T) {
+// TestPrepareProposalErrorOnTooManyTxs tests that the block creation logic returns
+// an error if the ResponsePrepareProposal returned from the application is invalid.
+func TestPrepareProposalErrorOnTooManyTxs(t *testing.T) {
 	const height = 2
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -905,29 +900,26 @@ func TestPrepareProposalModifiedTxStatusFalse(t *testing.T) {
 	require.NoError(t, eventBus.Start(ctx))
 
 	state, stateDB, privVals := makeState(t, 1, height)
+	// limit max block size
+	state.ConsensusParams.Block.MaxBytes = 60 * 1024
 	stateStore := sm.NewStore(stateDB)
 
 	evpool := &mocks.EvidencePool{}
 	evpool.On("PendingEvidence", mock.Anything).Return([]types.Evidence{}, int64(0))
 
-	txs := factory.MakeTenTxs(height)
+	const nValidators = 1
+	var bytesPerTx int64 = 3
+	maxDataBytes := types.MaxDataBytes(state.ConsensusParams.Block.MaxBytes, 0, nValidators)
+	txs := factory.MakeNTxs(height, maxDataBytes/bytesPerTx+2) // +2 so that tx don't fit
 	mp := &mpmocks.Mempool{}
 	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs(txs))
 
 	trs := txsToTxRecords(types.Txs(txs))
-	trs = append(trs[len(trs)/2:], trs[:len(trs)/2]...)
-	trs = trs[1:]
-	trs[0].Action = abci.TxRecord_REMOVED
-	trs[1] = &abci.TxRecord{
-		Tx:     []byte("new"),
-		Action: abci.TxRecord_ADDED,
-	}
 
 	app := abcimocks.NewBaseMock()
 	app.On("PrepareProposal", mock.Anything).Return(abci.ResponsePrepareProposal{
-		ModifiedTxStatus: abci.ResponsePrepareProposal_UNMODIFIED,
-		TxRecords:        trs,
-	}, nil)
+		TxRecords: trs,
+	})
 
 	cc := abciclient.NewLocalClient(logger, app)
 	proxyApp := proxy.New(cc, logger, proxy.NopMetrics())
@@ -946,11 +938,62 @@ func TestPrepareProposalModifiedTxStatusFalse(t *testing.T) {
 	)
 	pa, _ := state.Validators.GetByIndex(0)
 	commit, votes := makeValidCommit(ctx, t, height, types.BlockID{}, state.Validators, privVals)
+
 	block, err := blockExec.CreateProposalBlock(ctx, height, state, commit, pa, votes)
+	require.Nil(t, block)
+	require.ErrorContains(t, err, "transaction data size exceeds maximum")
+
+	mp.AssertExpectations(t)
+}
+
+// TestPrepareProposalErrorOnPrepareProposalError tests when the client returns an error
+// upon calling PrepareProposal on it.
+func TestPrepareProposalErrorOnPrepareProposalError(t *testing.T) {
+	const height = 2
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	logger := log.NewNopLogger()
+	eventBus := eventbus.NewDefault(logger)
+	require.NoError(t, eventBus.Start(ctx))
+
+	state, stateDB, privVals := makeState(t, 1, height)
+	stateStore := sm.NewStore(stateDB)
+
+	evpool := &mocks.EvidencePool{}
+	evpool.On("PendingEvidence", mock.Anything).Return([]types.Evidence{}, int64(0))
+
+	txs := factory.MakeNTxs(height, 10)
+	mp := &mpmocks.Mempool{}
+	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs(txs))
+
+	cm := &abciclientmocks.Client{}
+	cm.On("IsRunning").Return(true)
+	cm.On("Error").Return(nil)
+	cm.On("Start", mock.Anything).Return(nil).Once()
+	cm.On("Wait").Return(nil).Once()
+	cm.On("PrepareProposal", mock.Anything, mock.Anything).Return(nil, errors.New("an injected error")).Once()
+
+	proxyApp := proxy.New(cm, logger, proxy.NopMetrics())
+	err := proxyApp.Start(ctx)
 	require.NoError(t, err)
-	for i, tx := range block.Data.Txs {
-		require.Equal(t, txs[i], tx)
-	}
+
+	blockExec := sm.NewBlockExecutor(
+		stateStore,
+		logger,
+		proxyApp,
+		mp,
+		evpool,
+		nil,
+		eventBus,
+		sm.NopMetrics(),
+	)
+	pa, _ := state.Validators.GetByIndex(0)
+	commit, votes := makeValidCommit(ctx, t, height, types.BlockID{}, state.Validators, privVals)
+
+	block, err := blockExec.CreateProposalBlock(ctx, height, state, commit, pa, votes)
+	require.Nil(t, block)
+	require.ErrorContains(t, err, "an injected error")
 
 	mp.AssertExpectations(t)
 }

internal/state/helpers_test.go

@@ -62,7 +62,7 @@ func makeAndApplyGoodBlock(
 	evidence []types.Evidence,
 ) (sm.State, types.BlockID) {
 	t.Helper()
-	block := state.MakeBlock(height, factory.MakeTenTxs(height), lastCommit, evidence, proposerAddr)
+	block := state.MakeBlock(height, factory.MakeNTxs(height, 10), lastCommit, evidence, proposerAddr)
 	partSet, err := block.MakePartSet(types.BlockPartSizeBytes)
 	require.NoError(t, err)
 

internal/state/indexer/mocks/event_sink.go

@@ -6,6 +6,7 @@ import (
 	context "context"
 
 	mock "github.com/stretchr/testify/mock"
+
 	indexer "github.com/tendermint/tendermint/internal/state/indexer"
 
 	query "github.com/tendermint/tendermint/internal/pubsub/query"

internal/state/mocks/evidence_pool.go

@@ -6,6 +6,7 @@ import (
 	context "context"
 
 	mock "github.com/stretchr/testify/mock"
+
 	state "github.com/tendermint/tendermint/internal/state"
 
 	types "github.com/tendermint/tendermint/types"

internal/state/mocks/store.go

@@ -4,6 +4,7 @@ package mocks
 
 import (
 	mock "github.com/stretchr/testify/mock"
+
 	state "github.com/tendermint/tendermint/internal/state"
 	tendermintstate "github.com/tendermint/tendermint/proto/tendermint/state"
 

internal/state/test/factory/block.go

@@ -45,7 +45,7 @@ func MakeBlocks(ctx context.Context, t *testing.T, n int, state *sm.State, privV
 func MakeBlock(state sm.State, height int64, c *types.Commit) *types.Block {
 	return state.MakeBlock(
 		height,
-		factory.MakeTenTxs(state.LastBlockHeight),
+		factory.MakeNTxs(state.LastBlockHeight, 10),
 		c,
 		nil,
 		state.Validators.GetProposer().Address,

internal/state/validation_test.go

@@ -338,7 +338,7 @@ func TestValidateBlockEvidence(t *testing.T) {
 				evidence = append(evidence, newEv)
 				currentBytes += int64(len(newEv.Bytes()))
 			}
-			block := state.MakeBlock(height, testfactory.MakeTenTxs(height), lastCommit, evidence, proposerAddr)
+			block := state.MakeBlock(height, testfactory.MakeNTxs(height, 10), lastCommit, evidence, proposerAddr)
 
 			err := blockExec.ValidateBlock(ctx, state, block)
 			if assert.Error(t, err) {

internal/statesync/mocks/state_provider.go

@@ -6,6 +6,7 @@ import (
 	context "context"
 
 	mock "github.com/stretchr/testify/mock"
+
 	state "github.com/tendermint/tendermint/internal/state"
 
 	types "github.com/tendermint/tendermint/types"

internal/statesync/reactor.go

@@ -16,6 +16,7 @@ import (
 	"github.com/tendermint/tendermint/config"
 	"github.com/tendermint/tendermint/internal/eventbus"
 	"github.com/tendermint/tendermint/internal/p2p"
+	"github.com/tendermint/tendermint/internal/p2p/conn"
 	sm "github.com/tendermint/tendermint/internal/state"
 	"github.com/tendermint/tendermint/internal/store"
 	"github.com/tendermint/tendermint/libs/log"
@@ -75,13 +76,13 @@ const (
 func getChannelDescriptors() map[p2p.ChannelID]*p2p.ChannelDescriptor {
 	return map[p2p.ChannelID]*p2p.ChannelDescriptor{
 		SnapshotChannel: {
-
 			ID:                  SnapshotChannel,
 			MessageType:         new(ssproto.Message),
 			Priority:            6,
 			SendQueueCapacity:   10,
 			RecvMessageCapacity: snapshotMsgSize,
 			RecvBufferCapacity:  128,
+			Name:                "snapshot",
 		},
 		ChunkChannel: {
 			ID:                  ChunkChannel,
@@ -90,6 +91,7 @@ func getChannelDescriptors() map[p2p.ChannelID]*p2p.ChannelDescriptor {
 			SendQueueCapacity:   4,
 			RecvMessageCapacity: chunkMsgSize,
 			RecvBufferCapacity:  128,
+			Name:                "chunk",
 		},
 		LightBlockChannel: {
 			ID:                  LightBlockChannel,
@@ -98,6 +100,7 @@ func getChannelDescriptors() map[p2p.ChannelID]*p2p.ChannelDescriptor {
 			SendQueueCapacity:   10,
 			RecvMessageCapacity: lightBlockMsgSize,
 			RecvBufferCapacity:  128,
+			Name:                "light-block",
 		},
 		ParamsChannel: {
 			ID:                  ParamsChannel,
@@ -106,6 +109,7 @@ func getChannelDescriptors() map[p2p.ChannelID]*p2p.ChannelDescriptor {
 			SendQueueCapacity:   10,
 			RecvMessageCapacity: paramMsgSize,
 			RecvBufferCapacity:  128,
+			Name:                "params",
 		},
 	}
 
@@ -233,10 +237,7 @@ func NewReactor(
 // The caller must be sure to execute OnStop to ensure the outbound p2p Channels are
 // closed. No error is returned.
 func (r *Reactor) OnStart(ctx context.Context) error {
-	go r.processCh(ctx, r.snapshotCh, "snapshot")
-	go r.processCh(ctx, r.chunkCh, "chunk")
-	go r.processCh(ctx, r.blockCh, "light block")
-	go r.processCh(ctx, r.paramsCh, "consensus params")
+	go r.processChannels(ctx, r.snapshotCh, r.chunkCh, r.blockCh, r.paramsCh)
 	go r.processPeerUpdates(ctx)
 
 	return nil
@@ -291,6 +292,7 @@ func (r *Reactor) Sync(ctx context.Context) (sm.State, error) {
 		r.metrics,
 	)
 	r.mtx.Unlock()
+
 	defer func() {
 		r.mtx.Lock()
 		// reset syncing objects at the close of Sync
@@ -780,6 +782,8 @@ func (r *Reactor) handleParamsMessage(ctx context.Context, envelope *p2p.Envelop
 		if sp, ok := r.stateProvider.(*stateProviderP2P); ok {
 			select {
 			case sp.paramsRecvCh <- cp:
+			case <-ctx.Done():
+				return ctx.Err()
 			case <-time.After(time.Second):
 				return errors.New("failed to send consensus params, stateprovider not ready for response")
 			}
@@ -797,7 +801,7 @@ func (r *Reactor) handleParamsMessage(ctx context.Context, envelope *p2p.Envelop
 // handleMessage handles an Envelope sent from a peer on a specific p2p Channel.
 // It will handle errors and any possible panics gracefully. A caller can handle
 // any error returned by sending a PeerError on the respective channel.
-func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelope *p2p.Envelope) (err error) {
+func (r *Reactor) handleMessage(ctx context.Context, envelope *p2p.Envelope) (err error) {
 	defer func() {
 		if e := recover(); e != nil {
 			err = fmt.Errorf("panic in processing message: %v", e)
@@ -811,7 +815,7 @@ func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelop
 
 	r.logger.Debug("received message", "message", reflect.TypeOf(envelope.Message), "peer", envelope.From)
 
-	switch chID {
+	switch envelope.ChannelID {
 	case SnapshotChannel:
 		err = r.handleSnapshotMessage(ctx, envelope)
 	case ChunkChannel:
@@ -821,7 +825,7 @@ func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelop
 	case ParamsChannel:
 		err = r.handleParamsMessage(ctx, envelope)
 	default:
-		err = fmt.Errorf("unknown channel ID (%d) for envelope (%v)", chID, envelope)
+		err = fmt.Errorf("unknown channel ID (%d) for envelope (%v)", envelope.ChannelID, envelope)
 	}
 
 	return err
@@ -831,15 +835,35 @@ func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelop
 // encountered during message execution will result in a PeerError being sent on
 // the respective channel. When the reactor is stopped, we will catch the signal
 // and close the p2p Channel gracefully.
-func (r *Reactor) processCh(ctx context.Context, ch *p2p.Channel, chName string) {
-	iter := ch.Receive(ctx)
+func (r *Reactor) processChannels(ctx context.Context, chs ...*p2p.Channel) {
+	// make sure that the iterator gets cleaned up in case of error
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	chanTable := make(map[conn.ChannelID]*p2p.Channel, len(chs))
+	for idx := range chs {
+		ch := chs[idx]
+		chanTable[ch.ID] = ch
+	}
+
+	iter := p2p.MergedChannelIterator(ctx, chs...)
 	for iter.Next(ctx) {
 		envelope := iter.Envelope()
-		if err := r.handleMessage(ctx, ch.ID, envelope); err != nil {
+		if err := r.handleMessage(ctx, envelope); err != nil {
+			ch, ok := chanTable[envelope.ChannelID]
+			if !ok {
+				r.logger.Error("received impossible message",
+					"envelope_from", envelope.From,
+					"envelope_ch", envelope.ChannelID,
+					"num_chs", len(chanTable),
+					"err", err,
+				)
+				return
+			}
 			r.logger.Error("failed to process message",
 				"err", err,
-				"channel", chName,
-				"ch_id", ch.ID,
+				"channel", ch.String(),
+				"ch_id", envelope.ChannelID,
 				"envelope", envelope)
 			if serr := ch.SendError(ctx, p2p.PeerError{
 				NodeID: envelope.From,
@@ -875,14 +899,15 @@ func (r *Reactor) processPeerUpdate(ctx context.Context, peerUpdate p2p.PeerUpda
 
 	r.mtx.Lock()
 	defer r.mtx.Unlock()
+
 	if r.syncer == nil {
 		return
 	}
 
 	switch peerUpdate.Status {
 	case p2p.PeerStatusUp:
-
 		newProvider := NewBlockProvider(peerUpdate.NodeID, r.chainID, r.dispatcher)
+
 		r.providers[peerUpdate.NodeID] = newProvider
 		err := r.syncer.AddPeer(ctx, peerUpdate.NodeID)
 		if err != nil {

internal/statesync/reactor_test.go

@@ -257,8 +257,9 @@ func TestReactor_ChunkRequest_InvalidRequest(t *testing.T) {
 	rts := setup(ctx, t, nil, nil, 2)
 
 	rts.chunkInCh <- p2p.Envelope{
-		From:    types.NodeID("aa"),
-		Message: &ssproto.SnapshotsRequest{},
+		From:      types.NodeID("aa"),
+		ChannelID: ChunkChannel,
+		Message:   &ssproto.SnapshotsRequest{},
 	}
 
 	response := <-rts.chunkPeerErrCh
@@ -315,8 +316,9 @@ func TestReactor_ChunkRequest(t *testing.T) {
 			rts := setup(ctx, t, conn, nil, 2)
 
 			rts.chunkInCh <- p2p.Envelope{
-				From:    types.NodeID("aa"),
-				Message: tc.request,
+				From:      types.NodeID("aa"),
+				ChannelID: ChunkChannel,
+				Message:   tc.request,
 			}
 
 			response := <-rts.chunkOutCh
@@ -335,8 +337,9 @@ func TestReactor_SnapshotsRequest_InvalidRequest(t *testing.T) {
 	rts := setup(ctx, t, nil, nil, 2)
 
 	rts.snapshotInCh <- p2p.Envelope{
-		From:    types.NodeID("aa"),
-		Message: &ssproto.ChunkRequest{},
+		From:      types.NodeID("aa"),
+		ChannelID: SnapshotChannel,
+		Message:   &ssproto.ChunkRequest{},
 	}
 
 	response := <-rts.snapshotPeerErrCh
@@ -400,8 +403,9 @@ func TestReactor_SnapshotsRequest(t *testing.T) {
 			rts := setup(ctx, t, conn, nil, 100)
 
 			rts.snapshotInCh <- p2p.Envelope{
-				From:    types.NodeID("aa"),
-				Message: &ssproto.SnapshotsRequest{},
+				From:      types.NodeID("aa"),
+				ChannelID: SnapshotChannel,
+				Message:   &ssproto.SnapshotsRequest{},
 			}
 
 			if len(tc.expectResponses) > 0 {
@@ -457,7 +461,8 @@ func TestReactor_LightBlockResponse(t *testing.T) {
 	rts.stateStore.On("LoadValidators", height).Return(vals, nil)
 
 	rts.blockInCh <- p2p.Envelope{
-		From: types.NodeID("aa"),
+		From:      types.NodeID("aa"),
+		ChannelID: LightBlockChannel,
 		Message: &ssproto.LightBlockRequest{
 			Height: 10,
 		},
@@ -733,7 +738,8 @@ func handleLightBlockRequests(
 					require.NoError(t, err)
 					select {
 					case sending <- p2p.Envelope{
-						From: envelope.To,
+						From:      envelope.To,
+						ChannelID: LightBlockChannel,
 						Message: &ssproto.LightBlockResponse{
 							LightBlock: lb,
 						},
@@ -750,7 +756,8 @@ func handleLightBlockRequests(
 						require.NoError(t, err)
 						select {
 						case sending <- p2p.Envelope{
-							From: envelope.To,
+							From:      envelope.To,
+							ChannelID: LightBlockChannel,
 							Message: &ssproto.LightBlockResponse{
 								LightBlock: differntLB,
 							},
@@ -761,7 +768,8 @@ func handleLightBlockRequests(
 					case 1: // send nil block i.e. pretend we don't have it
 						select {
 						case sending <- p2p.Envelope{
-							From: envelope.To,
+							From:      envelope.To,
+							ChannelID: LightBlockChannel,
 							Message: &ssproto.LightBlockResponse{
 								LightBlock: nil,
 							},
@@ -802,7 +810,8 @@ func handleConsensusParamsRequest(
 			}
 			select {
 			case sending <- p2p.Envelope{
-				From: envelope.To,
+				From:      envelope.To,
+				ChannelID: ParamsChannel,
 				Message: &ssproto.ParamsResponse{
 					Height:          msg.Height,
 					ConsensusParams: paramsProto,
@@ -913,7 +922,8 @@ func handleSnapshotRequests(
 			require.True(t, ok)
 			for _, snapshot := range snapshots {
 				sendingCh <- p2p.Envelope{
-					From: envelope.To,
+					From:      envelope.To,
+					ChannelID: SnapshotChannel,
 					Message: &ssproto.SnapshotsResponse{
 						Height:   snapshot.Height,
 						Format:   snapshot.Format,
@@ -946,7 +956,8 @@ func handleChunkRequests(
 			msg, ok := envelope.Message.(*ssproto.ChunkRequest)
 			require.True(t, ok)
 			sendingCh <- p2p.Envelope{
-				From: envelope.To,
+				From:      envelope.To,
+				ChannelID: ChunkChannel,
 				Message: &ssproto.ChunkResponse{
 					Height:  msg.Height,
 					Format:  msg.Format,

internal/test/factory/tx.go

@@ -2,10 +2,10 @@ package factory
 
 import "github.com/tendermint/tendermint/types"
 
-func MakeTenTxs(height int64) []types.Tx {
-	txs := make([]types.Tx, 10)
+func MakeNTxs(height, n int64) []types.Tx {
+	txs := make([]types.Tx, n)
 	for i := range txs {
-		txs[i] = types.Tx([]byte{byte(height), byte(i)})
+		txs[i] = types.Tx([]byte{byte(height), byte(i / 256), byte(i % 256)})
 	}
 	return txs
 }