feat: added sequencer set polling and consensus msgs queue

block/consensus.go

@@ -1,9 +1,60 @@
 package block
 
 import (
+	"fmt"
+	"sync"
+
 	"github.com/gogo/protobuf/proto"
+
+	"github.com/dymensionxyz/dymint/types"
+	rdktypes "github.com/dymensionxyz/dymint/types/pb/rollapp/sequencers/types"
+	protoutils "github.com/dymensionxyz/dymint/utils/proto"
+	"github.com/dymensionxyz/dymint/utils/queue"
 )
 
 type ConsensusMessagesStream interface {
-	GetConsensusMessages() ([]proto.Message, error)
+	Add(...proto.Message)
+	Get() []proto.Message
+}
+
+type ConsensusMessagesQueue struct {
+	mu    sync.Mutex
+	queue *queue.Queue[proto.Message]
+}
+
+func NewConsensusMessagesQueue() *ConsensusMessagesQueue {
+	return &ConsensusMessagesQueue{
+		mu:    sync.Mutex{},
+		queue: queue.New[proto.Message](),
+	}
+}
+
+func (q *ConsensusMessagesQueue) Add(message ...proto.Message) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	q.queue.Enqueue(message...)
+}
+
+func (q *ConsensusMessagesQueue) Get() []proto.Message {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	return q.queue.DequeueAll()
+}
+
+// ConsensusMsgsOnSequencerSetUpdate forms a list of consensus messages to handle the sequencer set update.
+func ConsensusMsgsOnSequencerSetUpdate(newSequencers []types.Sequencer) ([]proto.Message, error) {
+	msgs := make([]proto.Message, 0, len(newSequencers))
+	for _, s := range newSequencers {
+		anyPK, err := s.AnyConsPubKey()
+		if err != nil {
+			return nil, fmt.Errorf("sequencer consensus public key: %w", err)
+		}
+		msgs = append(msgs, &rdktypes.ConsensusMsgUpsertSequencer{
+			Operator:   s.SettlementAddress,
+			ConsPubKey: protoutils.CosmosToGogo(anyPK),
+			RewardAddr: s.RewardAddr,
+			Relayers:   s.WhitelistedRelayers,
+		})
+	}
+	return msgs, 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,7 +103,8 @@ 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,
@@ -115,14 +115,9 @@ func (e *Executor) CreateBlock(
 	maxBlockDataSizeBytes = min(maxBlockDataSizeBytes, uint64(max(minBlockMaxBytes, state.ConsensusParams.Block.MaxBytes)))
 	mempoolTxs := e.mempool.ReapMaxBytesMaxGas(int64(maxBlockDataSizeBytes), state.ConsensusParams.Block.MaxGas)
 
-	var consensusAnyMessages []*proto.Any
+	var consensusMsgs []proto2.Message
 	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 = e.consensusMessagesStream.Get()
 	}
 
 	block := &types.Block{
@@ -145,7 +140,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,
 	}
@@ -300,6 +295,10 @@ func (e *Executor) publishEvents(resp *tmstate.ABCIResponses, block *types.Block
 	return err
 }
 
+func (e *Executor) GetConsensusMessagesStream() ConsensusMessagesStream {
+	return e.consensusMessagesStream
+}
+
 func toDymintTxs(txs tmtypes.Txs) types.Txs {
 	optiTxs := make(types.Txs, len(txs))
 	for i := range txs {
@@ -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/executor_test.go

@@ -117,7 +117,7 @@ func TestCreateBlockWithConsensusMessages(t *testing.T) {
 
 	// Create a mock ConsensusMessagesStream
 	mockStream := &MockConsensusMessagesStream{}
-	mockStream.On("GetConsensusMessages").Return([]proto.Message{
+	mockStream.On("Get").Return([]proto.Message{
 		theMsg1,
 		theMsg2,
 	}, nil)
@@ -172,9 +172,13 @@ type MockConsensusMessagesStream struct {
 	mock.Mock
 }
 
-func (m *MockConsensusMessagesStream) GetConsensusMessages() ([]proto.Message, error) {
+func (m *MockConsensusMessagesStream) Get() []proto.Message {
 	args := m.Called()
-	return args.Get(0).([]proto.Message), args.Error(1)
+	return args.Get(0).([]proto.Message)
+}
+
+func (m *MockConsensusMessagesStream) Add(msgs ...proto.Message) {
+	m.Called(msgs)
 }
 
 func TestApplyBlock(t *testing.T) {

block/manager.go

@@ -112,7 +112,7 @@ func NewManager(
 		mempool,
 		proxyApp,
 		eventBus,
-		nil, // TODO add ConsensusMessagesStream
+		NewConsensusMessagesQueue(), // TODO properly specify ConsensusMessagesStream: https://github.com/dymensionxyz/dymint/issues/1125
 		logger,
 	)
 	if err != nil {
@@ -181,9 +181,6 @@ func (m *Manager) Start(ctx context.Context) error {
 		return m.PruningLoop(ctx)
 	})
 
-	// listen to new bonded sequencers events to add them in the sequencer set
-	go uevent.MustSubscribe(ctx, m.Pubsub, "newBondedSequencer", settlement.EventQueryNewBondedSequencer, m.UpdateSequencerSet, m.logger)
-
 	/* ----------------------------- full node mode ----------------------------- */
 	if !isProposer {
 		// Full-nodes can sync from DA but it is not necessary to wait for it, since it can sync from P2P as well in parallel.
@@ -229,12 +226,18 @@ func (m *Manager) Start(ctx context.Context) error {
 	// channel to signal sequencer rotation started
 	rotateSequencerC := make(chan string, 1)
 
+	// channel for sequencer set updates
+	sequencerSetUpdatesC := make(chan []types.Sequencer)
+
 	uerrors.ErrGroupGoLog(eg, m.logger, func() error {
 		return m.SubmitLoop(ctx, bytesProducedC)
 	})
+	uerrors.ErrGroupGoLog(eg, m.logger, func() error {
+		return m.MonitorSequencerSetUpdates(ctx, sequencerSetUpdatesC)
+	})
 	uerrors.ErrGroupGoLog(eg, m.logger, func() error {
 		bytesProducedC <- m.GetUnsubmittedBytes() // load unsubmitted bytes from previous run
-		return m.ProduceBlockLoop(ctx, bytesProducedC)
+		return m.ProduceBlockLoop(ctx, bytesProducedC, sequencerSetUpdatesC)
 	})
 	uerrors.ErrGroupGoLog(eg, m.logger, func() error {
 		return m.MonitorSequencerRotation(ctx, rotateSequencerC)
@@ -273,6 +276,7 @@ func (m *Manager) NextHeightToSubmit() uint64 {
 }
 
 // syncFromSettlement enforces the node to be synced on initial run from SL and DA.
+// The method modifies the state and is not thread-safe.
 func (m *Manager) syncFromSettlement() error {
 	err := m.UpdateSequencerSetFromSL()
 	if err != nil {

block/produce.go

@@ -7,23 +7,21 @@ 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.
 // A signal will be sent to the bytesProduced channel for each block produced
 // In this way it's possible to pause block production by not consuming the channel
-func (m *Manager) ProduceBlockLoop(ctx context.Context, bytesProducedC chan int) error {
+func (m *Manager) ProduceBlockLoop(ctx context.Context, bytesProducedC chan int, sequencerSetUpdates <-chan []types.Sequencer) error {
 	m.logger.Info("Started block producer loop.")
 
 	ticker := time.NewTicker(m.Conf.BlockTime)
@@ -39,6 +37,15 @@ func (m *Manager) ProduceBlockLoop(ctx context.Context, bytesProducedC chan int)
 		select {
 		case <-ctx.Done():
 			return nil
+
+		case update := <-sequencerSetUpdates:
+			err := m.HandleSequencerSetUpdate(update)
+			if err != nil {
+				// occurs on serialization issues and shouldn't happen in practice
+				uevent.MustPublish(ctx, m.Pubsub, &events.DataHealthStatus{Error: err}, events.HealthStatusList)
+				return err
+			}
+
 		case <-ticker.C:
 
 			// if empty blocks are configured to be enabled, and one is scheduled...
@@ -148,9 +155,11 @@ 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())
+	)
+	// if nextProposerInfo is set, we create a last block
 	if nextProposerHash != nil {
 		maxBlockDataSize = 0
 		proposerHashForBlock = *nextProposerHash