feat(executor): added MsgUpsertSequencer consensus message

block/consensus.go

@@ -0,0 +1,9 @@
+package block
+
+import (
+	"github.com/gogo/protobuf/proto"
+)
+
+type ConsensusMessagesStream interface {
+	GetConsensusMessages() ([]proto.Message, error)
+}

block/executor.go

@@ -4,6 +4,8 @@ import (
 	"errors"
 	"time"
 
+	proto2 "github.com/gogo/protobuf/proto"
+	proto "github.com/gogo/protobuf/types"
 	abci "github.com/tendermint/tendermint/abci/types"
 	tmcrypto "github.com/tendermint/tendermint/crypto/encoding"
 	tmstate "github.com/tendermint/tendermint/proto/tendermint/state"
@@ -21,11 +23,12 @@ const minBlockMaxBytes = 10000
 
 // Executor creates and applies blocks and maintains state.
 type Executor struct {
-	localAddress          []byte
-	chainID               string
-	proxyAppConsensusConn proxy.AppConnConsensus
-	proxyAppQueryConn     proxy.AppConnQuery
-	mempool               mempool.Mempool
+	localAddress            []byte
+	chainID                 string
+	proxyAppConsensusConn   proxy.AppConnConsensus
+	proxyAppQueryConn       proxy.AppConnQuery
+	mempool                 mempool.Mempool
+	consensusMessagesStream ConsensusMessagesStream
 
 	eventBus *tmtypes.EventBus
 
@@ -34,15 +37,24 @@ type Executor struct {
 
 // NewExecutor creates new instance of BlockExecutor.
 // localAddress will be used in sequencer mode only.
-func NewExecutor(localAddress []byte, chainID string, mempool mempool.Mempool, proxyApp proxy.AppConns, eventBus *tmtypes.EventBus, logger types.Logger) (*Executor, error) {
+func NewExecutor(
+	localAddress []byte,
+	chainID string,
+	mempool mempool.Mempool,
+	proxyApp proxy.AppConns,
+	eventBus *tmtypes.EventBus,
+	consensusMessagesStream ConsensusMessagesStream,
+	logger types.Logger,
+) (*Executor, error) {
 	be := Executor{
-		localAddress:          localAddress,
-		chainID:               chainID,
-		proxyAppConsensusConn: proxyApp.Consensus(),
-		proxyAppQueryConn:     proxyApp.Query(),
-		mempool:               mempool,
-		eventBus:              eventBus,
-		logger:                logger,
+		localAddress:            localAddress,
+		chainID:                 chainID,
+		proxyAppConsensusConn:   proxyApp.Consensus(),
+		proxyAppQueryConn:       proxyApp.Query(),
+		mempool:                 mempool,
+		eventBus:                eventBus,
+		consensusMessagesStream: consensusMessagesStream,
+		logger:                  logger,
 	}
 	return &be, nil
 }
@@ -91,11 +103,27 @@ func (e *Executor) InitChain(genesis *tmtypes.GenesisDoc, valset []*tmtypes.Vali
 	})
 }
 
-// CreateBlock reaps transactions from mempool and builds a block.
-func (e *Executor) CreateBlock(height uint64, lastCommit *types.Commit, lastHeaderHash, nextSeqHash [32]byte, state *types.State, maxBlockDataSizeBytes uint64) *types.Block {
+// CreateBlock reaps transactions from mempool and builds a block. Optionally, executes consensus messages that
+// gets from the consensus messages stream or from the method args.
+func (e *Executor) CreateBlock(
+	height uint64,
+	lastCommit *types.Commit,
+	lastHeaderHash, nextSeqHash [32]byte,
+	state *types.State,
+	maxBlockDataSizeBytes uint64,
+	consensusMsgs ...proto2.Message,
+) *types.Block {
 	maxBlockDataSizeBytes = min(maxBlockDataSizeBytes, uint64(max(minBlockMaxBytes, state.ConsensusParams.Block.MaxBytes)))
 	mempoolTxs := e.mempool.ReapMaxBytesMaxGas(int64(maxBlockDataSizeBytes), state.ConsensusParams.Block.MaxGas)
 
+	if e.consensusMessagesStream != nil {
+		consensusMessages, err := e.consensusMessagesStream.GetConsensusMessages()
+		if err != nil {
+			e.logger.Error("Failed to get consensus messages", "error", err)
+		}
+		consensusMsgs = append(consensusMsgs, consensusMessages...)
+	}
+
 	block := &types.Block{
 		Header: types.Header{
 			Version: types.Version{
@@ -116,6 +144,7 @@ func (e *Executor) CreateBlock(height uint64, lastCommit *types.Commit, lastHead
 			Txs:                    toDymintTxs(mempoolTxs),
 			IntermediateStateRoots: types.IntermediateStateRoots{RawRootsList: nil},
 			Evidence:               types.EvidenceData{Evidence: nil},
+			ConsensusMessages:      fromProtoMsgSliceToAnySlice(consensusMsgs...),
 		},
 		LastCommit: *lastCommit,
 	}
@@ -209,6 +238,7 @@ func (e *Executor) ExecuteBlock(state *types.State, block *types.Block) (*tmstat
 				Votes: nil,
 			},
 			ByzantineValidators: nil,
+			ConsensusMessages:   block.Data.ConsensusMessages,
 		})
 	if err != nil {
 		return nil, err
@@ -284,3 +314,23 @@ func fromDymintTxs(optiTxs types.Txs) tmtypes.Txs {
 	}
 	return txs
 }
+
+func fromProtoMsgToAny(msg proto2.Message) *proto.Any {
+	theType, err := proto2.Marshal(msg)
+	if err != nil {
+		return nil
+	}
+
+	return &proto.Any{
+		TypeUrl: proto2.MessageName(msg),
+		Value:   theType,
+	}
+}
+
+func fromProtoMsgSliceToAnySlice(msgs ...proto2.Message) []*proto.Any {
+	result := make([]*proto.Any, len(msgs))
+	for i, msg := range msgs {
+		result[i] = fromProtoMsgToAny(msg)
+	}
+	return result
+}

block/executor_test.go

@@ -6,6 +6,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/gogo/protobuf/proto"
+	prototypes "github.com/gogo/protobuf/types"
+	"github.com/golang/groupcache/testpb"
+
 	"github.com/dymensionxyz/dymint/block"
 
 	cryptocodec "github.com/cosmos/cosmos-sdk/crypto/codec"
@@ -47,7 +51,7 @@ func TestCreateBlock(t *testing.T) {
 	require.NotNil(abciClient)
 
 	mpool := mempoolv1.NewTxMempool(logger, cfg.DefaultMempoolConfig(), proxy.NewAppConnMempool(abciClient), 0)
-	executor, err := block.NewExecutor([]byte("test address"), "test", mpool, proxy.NewAppConns(clientCreator), nil, logger)
+	executor, err := block.NewExecutor([]byte("test address"), "test", mpool, proxy.NewAppConns(clientCreator), nil, nil, logger)
 	assert.NoError(err)
 
 	maxBytes := uint64(100)
@@ -86,6 +90,93 @@ func TestCreateBlock(t *testing.T) {
 	assert.Len(block.Data.Txs, 2)
 }
 
+func TestCreateBlockWithConsensusMessages(t *testing.T) {
+	assert := assert.New(t)
+	require := require.New(t)
+	logger := log.TestingLogger()
+	app := &tmmocks.MockApplication{}
+	app.On("CheckTx", mock.Anything).Return(abci.ResponseCheckTx{})
+	clientCreator := proxy.NewLocalClientCreator(app)
+	abciClient, err := clientCreator.NewABCIClient()
+	require.NoError(err)
+	require.NotNil(clientCreator)
+	require.NotNil(abciClient)
+	mpool := mempoolv1.NewTxMempool(logger, cfg.DefaultMempoolConfig(), proxy.NewAppConnMempool(abciClient), 0)
+
+	name, city := "test1", ""
+	theMsg1 := &testpb.TestMessage{
+		Name: &name,
+		City: &city,
+	}
+
+	name, city = "test2", ""
+	theMsg2 := &testpb.TestMessage{
+		Name: &name,
+		City: &city,
+	}
+
+	// Create a mock ConsensusMessagesStream
+	mockStream := &MockConsensusMessagesStream{}
+	mockStream.On("GetConsensusMessages").Return([]proto.Message{
+		theMsg1,
+		theMsg2,
+	}, nil)
+
+	executor, err := block.NewExecutor([]byte("test address"), "test", mpool, proxy.NewAppConns(clientCreator), nil, mockStream, logger)
+	assert.NoError(err)
+
+	maxBytes := uint64(1000)
+	proposerKey := ed25519.GenPrivKey()
+	tmPubKey, err := cryptocodec.ToTmPubKeyInterface(proposerKey.PubKey())
+	require.NoError(err)
+
+	state := &types.State{}
+	state.Sequencers.SetProposer(types.NewSequencerFromValidator(*tmtypes.NewValidator(tmPubKey, 1)))
+	state.ConsensusParams.Block.MaxBytes = int64(maxBytes)
+	state.ConsensusParams.Block.MaxGas = 100000
+
+	block := executor.CreateBlock(1, &types.Commit{}, [32]byte{}, [32]byte(state.Sequencers.ProposerHash()[:]), state, maxBytes)
+
+	require.NotNil(block)
+	assert.Empty(block.Data.Txs)
+	assert.Equal(uint64(1), block.Header.Height)
+	assert.Len(block.Data.ConsensusMessages, 2)
+
+	// Verify the content of ConsensusMessages
+	theType, err := proto.Marshal(theMsg1)
+	require.NoError(err)
+
+	anyMsg1 := &prototypes.Any{
+		TypeUrl: proto.MessageName(theMsg1),
+		Value:   theType,
+	}
+	require.NoError(err)
+
+	theType, err = proto.Marshal(theMsg2)
+	require.NoError(err)
+
+	anyMsg2 := &prototypes.Any{
+		TypeUrl: proto.MessageName(theMsg2),
+		Value:   theType,
+	}
+	require.NoError(err)
+
+	assert.True(proto.Equal(anyMsg1, block.Data.ConsensusMessages[0]))
+	assert.True(proto.Equal(anyMsg2, block.Data.ConsensusMessages[1]))
+
+	mockStream.AssertExpectations(t)
+}
+
+// MockConsensusMessagesStream is a mock implementation of ConsensusMessagesStream
+type MockConsensusMessagesStream struct {
+	mock.Mock
+}
+
+func (m *MockConsensusMessagesStream) GetConsensusMessages() ([]proto.Message, error) {
+	args := m.Called()
+	return args.Get(0).([]proto.Message), args.Error(1)
+}
+
 func TestApplyBlock(t *testing.T) {
 	assert := assert.New(t)
 	require := require.New(t)
@@ -138,7 +229,7 @@ func TestApplyBlock(t *testing.T) {
 	appConns := &tmmocksproxy.MockAppConns{}
 	appConns.On("Consensus").Return(abciClient)
 	appConns.On("Query").Return(abciClient)
-	executor, err := block.NewExecutor([]byte("test address"), chainID, mpool, appConns, eventBus, logger)
+	executor, err := block.NewExecutor([]byte("test address"), chainID, mpool, appConns, eventBus, nil, logger)
 	assert.NoError(err)
 
 	// Subscribe to tx events

block/manager.go

@@ -106,7 +106,15 @@ func NewManager(
 	if err != nil {
 		return nil, err
 	}
-	exec, err := NewExecutor(localAddress, genesis.ChainID, mempool, proxyApp, eventBus, logger)
+	exec, err := NewExecutor(
+		localAddress,
+		genesis.ChainID,
+		mempool,
+		proxyApp,
+		eventBus,
+		nil, // TODO add ConsensusMessagesStream
+		logger,
+	)
 	if err != nil {
 		return nil, fmt.Errorf("create block executor: %w", err)
 	}