feat(executor): added MsgUpsertSequencer consensus message

block/consensus.go

@@ -1,9 +1,55 @@
 package block
 
 import (
+	"fmt"
+
+	codectypes "github.com/cosmos/cosmos-sdk/codec/types"
+	cryptocodec "github.com/cosmos/cosmos-sdk/crypto/codec"
 	"github.com/gogo/protobuf/proto"
+
+	rdktypes "github.com/dymensionxyz/dymint/types/pb/rollapp/sequencers/types"
+	protoutils "github.com/dymensionxyz/dymint/utils/proto"
 )
 
 type ConsensusMessagesStream interface {
 	GetConsensusMessages() ([]proto.Message, error)
 }
+
+// consensusMsgsOnCreateBlock forms a list of consensus messages that need execution on rollapp's BeginBlock.
+// Currently, we need to create a sequencer in the rollapp if it doesn't exist in the following cases:
+//   - On the very first block after the genesis or
+//   - On the last block of the current sequencer (eg, during the rotation).
+func (m *Manager) consensusMsgsOnCreateBlock(
+	nextProposerSettlementAddr string,
+	lastSeqBlock bool, // Indicates that the block is the last for the current seq. True during the rotation.
+) ([]proto.Message, error) {
+	if !m.State.IsGenesis() && !lastSeqBlock {
+		return nil, nil
+	}
+
+	nextSeq := m.State.Sequencers.GetByAddress(nextProposerSettlementAddr)
+	// Sanity check. Must never happen in practice. The sequencer's existence is verified beforehand in Manager.CompleteRotation.
+	if nextSeq == nil {
+		return nil, fmt.Errorf("no sequencer found for address while creating a new block: %s", nextProposerSettlementAddr)
+	}
+
+	// Get proposer's consensus public key and convert it to proto.Any
+	val, err := nextSeq.TMValidator()
+	if err != nil {
+		return nil, fmt.Errorf("convert next squencer to tendermint validator: %w", err)
+	}
+	pubKey, err := cryptocodec.FromTmPubKeyInterface(val.PubKey)
+	if err != nil {
+		return nil, fmt.Errorf("convert tendermint pubkey to cosmos: %w", err)
+	}
+	anyPK, err := codectypes.NewAnyWithValue(pubKey)
+	if err != nil {
+		return nil, fmt.Errorf("convert cosmos pubkey to any: %w", err)
+	}
+
+	return []proto.Message{&rdktypes.ConsensusMsgUpsertSequencer{
+		Operator:   nextProposerSettlementAddr,
+		ConsPubKey: protoutils.CosmosToGogo(anyPK),
+		RewardAddr: nextProposerSettlementAddr,
+	}}, nil
+}

block/executor.go

@@ -6,7 +6,6 @@ import (
 
 	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"
@@ -104,25 +103,25 @@ func (e *Executor) InitChain(genesis *tmtypes.GenesisDoc, valset []*tmtypes.Vali
 	})
 }
 
-// CreateBlock reaps transactions from mempool and builds a 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)
 
-	var consensusAnyMessages []*proto.Any
 	if e.consensusMessagesStream != nil {
 		consensusMessages, err := e.consensusMessagesStream.GetConsensusMessages()
 		if err != nil {
 			e.logger.Error("Failed to get consensus messages", "error", err)
 		}
-
-		consensusAnyMessages = fromProtoMsgSliceToAnySlice(consensusMessages)
+		consensusMsgs = append(consensusMsgs, consensusMessages...)
 	}
 
 	block := &types.Block{
@@ -145,7 +144,7 @@ func (e *Executor) CreateBlock(
 			Txs:                    toDymintTxs(mempoolTxs),
 			IntermediateStateRoots: types.IntermediateStateRoots{RawRootsList: nil},
 			Evidence:               types.EvidenceData{Evidence: nil},
-			ConsensusMessages:      consensusAnyMessages,
+			ConsensusMessages:      fromProtoMsgSliceToAnySlice(consensusMsgs...),
 		},
 		LastCommit: *lastCommit,
 	}
@@ -328,7 +327,7 @@ func fromProtoMsgToAny(msg proto2.Message) *proto.Any {
 	}
 }
 
-func fromProtoMsgSliceToAnySlice(msgs []proto2.Message) []*proto.Any {
+func fromProtoMsgSliceToAnySlice(msgs ...proto2.Message) []*proto.Any {
 	result := make([]*proto.Any, len(msgs))
 	for i, msg := range msgs {
 		result[i] = fromProtoMsgToAny(msg)

block/manager.go

@@ -112,7 +112,7 @@ func NewManager(
 		mempool,
 		proxyApp,
 		eventBus,
-		nil, // TODO add ConsensusMessagesStream
+		nil, // TODO add ConsensusMessagesStream: https://github.com/dymensionxyz/dymint/issues/1125
 		logger,
 	)
 	if err != nil {

block/produce.go

@@ -7,17 +7,15 @@ import (
 	"time"
 
 	"github.com/dymensionxyz/gerr-cosmos/gerrc"
-
-	"github.com/dymensionxyz/dymint/node/events"
-	"github.com/dymensionxyz/dymint/store"
-	uevent "github.com/dymensionxyz/dymint/utils/event"
-
 	tmed25519 "github.com/tendermint/tendermint/crypto/ed25519"
 	cmtproto "github.com/tendermint/tendermint/proto/tendermint/types"
 	tmtypes "github.com/tendermint/tendermint/types"
 	tmtime "github.com/tendermint/tendermint/types/time"
 
+	"github.com/dymensionxyz/dymint/node/events"
+	"github.com/dymensionxyz/dymint/store"
 	"github.com/dymensionxyz/dymint/types"
+	uevent "github.com/dymensionxyz/dymint/utils/event"
 )
 
 // ProduceBlockLoop is calling publishBlock in a loop as long as we're synced.
@@ -148,14 +146,32 @@ func (m *Manager) produceBlock(allowEmpty bool, nextProposerHash *[32]byte) (*ty
 		return nil, nil, fmt.Errorf("load block: height: %d: %w: %w", newHeight, err, ErrNonRecoverable)
 	}
 
-	maxBlockDataSize := uint64(float64(m.Conf.BatchSubmitBytes) * types.MaxBlockSizeAdjustment)
-	proposerHashForBlock := [32]byte(m.State.Sequencers.ProposerHash())
-	// if nextProposerHash is set, we create a last block
+	var (
+		maxBlockDataSize     = uint64(float64(m.Conf.BatchSubmitBytes) * types.MaxBlockSizeAdjustment)
+		proposerHashForBlock = [32]byte(m.State.Sequencers.ProposerHash())
+		nextProposerAddr     = m.State.Sequencers.Proposer.SettlementAddress
+		lastProposerBlock    = false // Indicates that the block is the last for the current seq. True during the rotation.
+	)
+	// if nextProposerInfo is set, we create a last block
 	if nextProposerHash != nil {
+		nextSeq, err := m.State.Sequencers.GetByHash(nextProposerHash[:])
+		if err != nil {
+			return nil, nil, fmt.Errorf("get next sequencer by hash: %w", err)
+		}
 		maxBlockDataSize = 0
 		proposerHashForBlock = *nextProposerHash
+		nextProposerAddr = nextSeq.SettlementAddress
+		lastProposerBlock = true
+	}
+	// TODO: Ideally, there should be only one point for adding consensus messages. Given that they come from
+	// ConsensusMessagesStream, this should send them there instead of having to ways of sending consensusMessages.
+	// There is no implementation of the stream as of now. Unify the approach of adding consensus messages when
+	// the stream is implemented! https://github.com/dymensionxyz/dymint/issues/1125
+	consensusMsgs, err := m.consensusMsgsOnCreateBlock(nextProposerAddr, lastProposerBlock)
+	if err != nil {
+		return nil, nil, fmt.Errorf("create consensus msgs for create block: last proposer block: %v, height: %d, next proposer addr: %s: %w: %w", lastProposerBlock, newHeight, nextProposerAddr, err, ErrNonRecoverable)
 	}
-	block = m.Executor.CreateBlock(newHeight, lastCommit, lastHeaderHash, proposerHashForBlock, m.State, maxBlockDataSize)
+	block = m.Executor.CreateBlock(newHeight, lastCommit, lastHeaderHash, proposerHashForBlock, m.State, maxBlockDataSize, consensusMsgs...)
 	if !allowEmpty && len(block.Data.Txs) == 0 {
 		return nil, nil, fmt.Errorf("%w: %w", types.ErrEmptyBlock, ErrRecoverable)
 	}

block/sequencers.go

@@ -6,9 +6,10 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/tendermint/tendermint/libs/pubsub"
+
 	"github.com/dymensionxyz/dymint/settlement"
 	"github.com/dymensionxyz/dymint/types"
-	"github.com/tendermint/tendermint/libs/pubsub"
 )
 
 func (m *Manager) MonitorSequencerRotation(ctx context.Context, rotateC chan string) error {

p2p/client_test.go

@@ -17,10 +17,11 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/dymensionxyz/dymint/store"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/libs/pubsub"
 
+	"github.com/dymensionxyz/dymint/store"
+
 	"github.com/dymensionxyz/dymint/config"
 	"github.com/dymensionxyz/dymint/p2p"
 	"github.com/dymensionxyz/dymint/testutil"
@@ -49,6 +50,8 @@ func TestClientStartup(t *testing.T) {
 	assert.NoError(err)
 }
 
+// TestBootstrapping TODO: this test is flaky in main. Try running it 100 times to reproduce.
+// https://github.com/dymensionxyz/dymint/issues/869
 func TestBootstrapping(t *testing.T) {
 	assert := assert.New(t)
 	logger := log.TestingLogger()

settlement/grpc/grpc.go

@@ -14,6 +14,7 @@ import (
 	cryptocodec "github.com/cosmos/cosmos-sdk/crypto/codec"
 	"github.com/cosmos/cosmos-sdk/crypto/keys/ed25519"
 	cryptotypes "github.com/cosmos/cosmos-sdk/crypto/types"
+	"github.com/cosmos/cosmos-sdk/types/bech32"
 	"github.com/dymensionxyz/gerr-cosmos/gerrc"
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/tendermint/tendermint/libs/pubsub"
@@ -28,6 +29,10 @@ import (
 	rollapptypes "github.com/dymensionxyz/dymint/types/pb/dymensionxyz/dymension/rollapp"
 )
 
+const (
+	addressPrefix = "dym"
+)
+
 // Client is an extension of the base settlement layer client
 // for usage in tests and local development.
 type Client struct {
@@ -225,7 +230,12 @@ func (c *Client) GetProposer() *types.Sequencer {
 		c.logger.Error("Error converting to tendermint pubkey", "err", err)
 		return nil
 	}
-	return types.NewSequencer(tmPubKey, pubKey.Address().String())
+	settlementAddr, err := bech32.ConvertAndEncode(addressPrefix, pubKeyBytes)
+	if err != nil {
+		c.logger.Error("Error converting pubkey to settlement address", "err", err)
+		return nil
+	}
+	return types.NewSequencer(tmPubKey, settlementAddr)
 }
 
 // GetSequencerByAddress returns all sequencer information by its address. Not implemented since it will not be used in grpc SL

settlement/local/local.go

@@ -15,6 +15,7 @@ import (
 	cryptocodec "github.com/cosmos/cosmos-sdk/crypto/codec"
 	"github.com/cosmos/cosmos-sdk/crypto/keys/ed25519"
 	cryptotypes "github.com/cosmos/cosmos-sdk/crypto/types"
+	"github.com/cosmos/cosmos-sdk/types/bech32"
 	"github.com/dymensionxyz/gerr-cosmos/gerrc"
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/tendermint/tendermint/libs/pubsub"
@@ -28,7 +29,10 @@ import (
 	uevent "github.com/dymensionxyz/dymint/utils/event"
 )
 
-const kvStoreDBName = "settlement"
+const (
+	kvStoreDBName = "settlement"
+	addressPrefix = "dym"
+)
 
 var (
 	settlementKVPrefix = []byte{0}
@@ -203,7 +207,12 @@ func (c *Client) GetProposer() *types.Sequencer {
 		c.logger.Error("Error converting to tendermint pubkey", "err", err)
 		return nil
 	}
-	return types.NewSequencer(tmPubKey, pubKey.Address().String())
+	settlementAddr, err := bech32.ConvertAndEncode(addressPrefix, pubKeyBytes)
+	if err != nil {
+		c.logger.Error("Error converting pubkey to settlement address", "err", err)
+		return nil
+	}
+	return types.NewSequencer(tmPubKey, settlementAddr)
 }
 
 // GetSequencerByAddress returns all sequencer information by its address. Not implemented since it will not be used in mock SL

testutil/types.go

@@ -5,9 +5,7 @@ import (
 	"math/big"
 	"time"
 
-	"github.com/dymensionxyz/dymint/types"
-	"github.com/dymensionxyz/dymint/types/pb/dymint"
-	dymintversion "github.com/dymensionxyz/dymint/version"
+	"github.com/cosmos/cosmos-sdk/types/bech32"
 	"github.com/libp2p/go-libp2p/core/crypto"
 	abci "github.com/tendermint/tendermint/abci/types"
 	tmcrypto "github.com/tendermint/tendermint/crypto"
@@ -16,13 +14,19 @@ import (
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
 	version "github.com/tendermint/tendermint/proto/tendermint/version"
 	tmtypes "github.com/tendermint/tendermint/types"
+
+	"github.com/dymensionxyz/dymint/types"
+	"github.com/dymensionxyz/dymint/types/pb/dymint"
+	dymintversion "github.com/dymensionxyz/dymint/version"
 )
 
 const (
 	// BlockVersion is the default block version for testing
 	BlockVersion = 1
 	// AppVersion is the default app version for testing
 	AppVersion = 2
+
+	SettlementAccountPrefix = "dym"
 )
 
 func createRandomHashes() [][32]byte {
@@ -38,6 +42,15 @@ func createRandomHashes() [][32]byte {
 	return h
 }
 
+func GenerateSettlementAddress() string {
+	addrBytes := ed25519.GenPrivKey().PubKey().Address().Bytes()
+	addr, err := bech32.ConvertAndEncode(SettlementAccountPrefix, addrBytes)
+	if err != nil {
+		panic(err)
+	}
+	return addr
+}
+
 func GetRandomTx() types.Tx {
 	n, _ := rand.Int(rand.Reader, big.NewInt(100))
 	size := uint64(n.Int64()) + 100
@@ -247,7 +260,7 @@ func GenerateStateWithSequencer(initialHeight int64, lastBlockHeight int64, pubk
 			},
 		},
 	}
-	s.Sequencers.SetProposer(types.NewSequencer(pubkey, ""))
+	s.Sequencers.SetProposer(types.NewSequencer(pubkey, GenerateSettlementAddress()))
 	s.SetHeight(uint64(lastBlockHeight))
 	return s
 }

types/sequencer_set.go

@@ -52,6 +52,10 @@ func (s Sequencer) Hash() []byte {
 	return tempProposerSet.Hash()
 }
 
+func (s Sequencer) String() string {
+	return fmt.Sprintf("Sequencer{SettlementAddress: %s Validator: %s}", s.SettlementAddress, s.val.String())
+}
+
 // SequencerSet is a set of rollapp sequencers and a proposer.
 type SequencerSet struct {
 	// Sequencers is the set of sequencers registered in the settlement layer
@@ -93,6 +97,18 @@ func (s *SequencerSet) SetProposerByHash(hash []byte) error {
 	return ErrMissingProposerPubKey
 }
 
+// GetByHash gets the sequencer by hash. It returns an error if the hash is not found in the sequencer set.
+func (s *SequencerSet) GetByHash(hash []byte) (Sequencer, error) {
+	for _, seq := range s.Sequencers {
+		if bytes.Equal(seq.Hash(), hash) {
+			return seq, nil
+		}
+	}
+	// can't find the proposer in the sequencer set
+	// can happen in cases where the node is not synced with the SL and the sequencer array in the set is not updated
+	return Sequencer{}, ErrMissingProposerPubKey
+}
+
 // SetProposer sets the proposer and adds it to the sequencer set if not already present.
 func (s *SequencerSet) SetProposer(proposer *Sequencer) {
 	if proposer == nil {