feat: set sequencer mode by hub and not by a flag (#821)

dymensionxyz · May 12, 2024 · 5848812 · 5848812
1 parent 426015b
commit 5848812
Show file tree

Hide file tree

Showing 14 changed files with 56 additions and 158 deletions.
diff --git a/block/manager.go b/block/manager.go
@@ -144,18 +144,21 @@ func NewManager(
 }
 
 // Start starts the block manager.
-func (m *Manager) Start(ctx context.Context, isAggregator bool) error {
+func (m *Manager) Start(ctx context.Context) error {
 	m.logger.Info("Starting the block manager")
 
-	// TODO (#283): set aggregator mode by proposer addr on the hub
-	if isAggregator {
-		// make sure local signing key is the registered on the hub
-		slProposerKey := m.SLClient.GetProposer().PublicKey.Bytes()
-		localProposerKey, _ := m.ProposerKey.GetPublic().Raw()
-		if !bytes.Equal(slProposerKey, localProposerKey) {
-			return fmt.Errorf("proposer key mismatch: settlement proposer key: %s, block manager proposer key: %s", slProposerKey, m.ProposerKey.GetPublic())
-		}
+	// Check if proposer key matches to the one in the settlement layer
+	var isAggregator bool
+	slProposerKey := m.SLClient.GetProposer().PublicKey.Bytes()
+	localProposerKey, err := m.ProposerKey.GetPublic().Raw()
+	if err != nil {
+		return fmt.Errorf("get local node public key: %w", err)
+	}
+	if bytes.Equal(slProposerKey, localProposerKey) {
 		m.logger.Info("Starting in aggregator mode")
+		isAggregator = true
+	} else {
+		m.logger.Info("Starting in non-aggregator mode")
 	}
 
 	// Check if InitChain flow is needed
@@ -172,7 +175,7 @@ func (m *Manager) Start(ctx context.Context, isAggregator bool) error {
 		go uevent.MustSubscribe(ctx, m.Pubsub, "applyGossipedBlocksLoop", p2p.EventQueryNewNewGossipedBlock, m.onNewGossipedBlock, m.logger)
 	}
 
-	err := m.syncBlockManager()
+	err = m.syncBlockManager()
 	if err != nil {
 		return fmt.Errorf("sync block manager: %w", err)
 	}

diff --git a/block/manager_test.go b/block/manager_test.go
@@ -145,7 +145,7 @@ func TestProduceOnlyAfterSynced(t *testing.T) {
 	// Capture the error returned by manager.Start.
 	errChan := make(chan error, 1)
 	go func() {
-		errChan <- manager.Start(ctx, true)
+		errChan <- manager.Start(ctx)
 		err := <-errChan
 		assert.NoError(t, err)
 	}()
@@ -270,7 +270,7 @@ func TestBlockProductionNodeHealth(t *testing.T) {
 	// Start the manager
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	err = manager.Start(ctx, true)
+	err = manager.Start(ctx)
 	require.NoError(err)
 	time.Sleep(100 * time.Millisecond)
 

diff --git a/config/config.go b/config/config.go
@@ -27,7 +27,6 @@ type NodeConfig struct {
 	P2P     P2PConfig
 	RPC     RPCConfig
 	// parameters below are dymint specific and read from config
-	Aggregator         bool `mapstructure:"aggregator"`
 	BlockManagerConfig `mapstructure:",squash"`
 	DALayer            string                 `mapstructure:"da_layer"`
 	DAConfig           string                 `mapstructure:"da_config"`

diff --git a/config/config_test.go b/config/config_test.go
@@ -24,7 +24,6 @@ func TestViperAndCobra(t *testing.T) {
 	nc := config.DefaultConfig("", "")
 	config.EnsureRoot(dir, nc)
 
-	assert.NoError(cmd.Flags().Set(config.FlagAggregator, "true"))
 	assert.NoError(cmd.Flags().Set(config.FlagDALayer, "foobar"))
 	assert.NoError(cmd.Flags().Set(config.FlagDAConfig, `{"json":true}`))
 	assert.NoError(cmd.Flags().Set(config.FlagBlockTime, "1234s"))
@@ -35,7 +34,6 @@ func TestViperAndCobra(t *testing.T) {
 
 	assert.NoError(nc.GetViperConfig(cmd, dir))
 
-	assert.Equal(true, nc.Aggregator)
 	assert.Equal("foobar", nc.DALayer)
 	assert.Equal(`{"json":true}`, nc.DAConfig)
 	assert.Equal(1234*time.Second, nc.BlockTime)

diff --git a/config/defaults.go b/config/defaults.go
@@ -26,7 +26,6 @@ func DefaultConfig(home, chainId string) *NodeConfig {
 			ListenAddress: DefaultListenAddress,
 			Seeds:         "",
 		},
-		Aggregator: true,
 		BlockManagerConfig: BlockManagerConfig{
 			BlockTime:               200 * time.Millisecond,
 			EmptyBlocksMaxTime:      3600 * time.Second,

diff --git a/config/flags.go b/config/flags.go
@@ -7,7 +7,6 @@ import (
 )
 
 const (
-	FlagAggregator             = "dymint.aggregator"
 	FlagDALayer                = "dymint.da_layer"
 	FlagDAConfig               = "dymint.da_config"
 	FlagBlockTime              = "dymint.block_time"
@@ -39,7 +38,6 @@ func AddNodeFlags(cmd *cobra.Command) {
 
 	def := DefaultNodeConfig
 
-	cmd.Flags().Bool(FlagAggregator, false, "run node in aggregator mode")
 	cmd.Flags().String(FlagDALayer, def.DALayer, "Data Availability Layer Client name (mock or grpc")
 	cmd.Flags().String(FlagDAConfig, def.DAConfig, "Data Availability Layer Client config")
 	cmd.Flags().Duration(FlagBlockTime, def.BlockTime, "block time (for aggregator mode)")
@@ -61,9 +59,6 @@ func AddNodeFlags(cmd *cobra.Command) {
 }
 
 func BindDymintFlags(cmd *cobra.Command, v *viper.Viper) error {
-	if err := v.BindPFlag("aggregator", cmd.Flags().Lookup(FlagAggregator)); err != nil {
-		return err
-	}
 	if err := v.BindPFlag("da_layer", cmd.Flags().Lookup(FlagDALayer)); err != nil {
 		return err
 	}

diff --git a/config/toml.go b/config/toml.go
@@ -65,8 +65,6 @@ const defaultConfigTemplate = `
 #######################################################
 ###       Dymint Configuration Options     ###
 #######################################################
-aggregator = "{{ .Aggregator }}"
-
 # block production interval
 block_time = "{{ .BlockManagerConfig.BlockTime }}"
 # block production interval in case of no transactions ("0s" produces empty blocks)

diff --git a/node/integration_test.go b/node/integration_test.go
diff --git a/node/node.go b/node/node.go
@@ -116,9 +116,8 @@ type Node struct {
 
 	baseLayerHealth baseLayerHealth
 
-	// keep context here only because of API compatibility
-	// - it's used in `OnStart` (defined in service.Service interface)
-	Ctx context.Context
+	// shared context for all dymint components
+	ctx context.Context
 }
 
 // NewNode creates new Dymint node.
@@ -248,7 +247,7 @@ func NewNode(
 		TxIndexer:      txIndexer,
 		IndexerService: indexerService,
 		BlockIndexer:   blockIndexer,
-		Ctx:            ctx,
+		ctx:            ctx,
 	}
 
 	node.BaseService = *service.NewBaseService(logger, "Node", node)
@@ -288,7 +287,7 @@ func (n *Node) initGenesisChunks() error {
 // OnStart is a part of Service interface.
 func (n *Node) OnStart() error {
 	n.Logger.Info("starting P2P client")
-	err := n.P2P.Start(n.Ctx)
+	err := n.P2P.Start(n.ctx)
 	if err != nil {
 		return fmt.Errorf("start P2P client: %w", err)
 	}
@@ -313,7 +312,7 @@ func (n *Node) OnStart() error {
 	n.startEventListener()
 
 	// start the block manager
-	err = n.blockManager.Start(n.Ctx, n.conf.Aggregator)
+	err = n.blockManager.Start(n.ctx)
 	if err != nil {
 		return fmt.Errorf("while starting block manager: %w", err)
 	}
@@ -409,8 +408,8 @@ func createAndStartIndexerService(
 
 // All events listeners should be registered here
 func (n *Node) startEventListener() {
-	go uevent.MustSubscribe(n.Ctx, n.PubsubServer, "settlementHealthStatusHandler", settlement.EventQuerySettlementHealthStatus, n.onBaseLayerHealthUpdate, n.Logger)
-	go uevent.MustSubscribe(n.Ctx, n.PubsubServer, "daHealthStatusHandler", da.EventQueryDAHealthStatus, n.onBaseLayerHealthUpdate, n.Logger)
+	go uevent.MustSubscribe(n.ctx, n.PubsubServer, "settlementHealthStatusHandler", settlement.EventQuerySettlementHealthStatus, n.onBaseLayerHealthUpdate, n.Logger)
+	go uevent.MustSubscribe(n.ctx, n.PubsubServer, "daHealthStatusHandler", da.EventQueryDAHealthStatus, n.onBaseLayerHealthUpdate, n.Logger)
 }
 
 func (n *Node) onBaseLayerHealthUpdate(event pubsub.Message) {
@@ -432,7 +431,7 @@ func (n *Node) onBaseLayerHealthUpdate(event pubsub.Message) {
 		if newStatus != nil {
 			n.Logger.Error("Node is unhealthy: base layer has problem.", "error", newStatus)
 		}
-		uevent.MustPublish(n.Ctx, n.PubsubServer, evt, events.HealthStatusList)
+		uevent.MustPublish(n.ctx, n.PubsubServer, evt, events.HealthStatusList)
 	}
 }
 

diff --git a/node/node_test.go b/node/node_test.go
@@ -64,11 +64,10 @@ func TestMempoolDirectly(t *testing.T) {
 	rollappID := "rollapp_1234-1"
 
 	nodeConfig := config.NodeConfig{
-		RootDir:    "",
-		DBPath:     "",
-		P2P:        config.P2PConfig{},
-		RPC:        config.RPCConfig{},
-		Aggregator: false,
+		RootDir: "",
+		DBPath:  "",
+		P2P:     config.P2PConfig{},
+		RPC:     config.RPCConfig{},
 		BlockManagerConfig: config.BlockManagerConfig{
 			BlockTime:               100 * time.Millisecond,
 			BlockBatchSize:          2,
@@ -181,7 +180,7 @@ func TestHealthStatusEventHandler(t *testing.T) {
 			done := make(chan bool, 1)
 			ready := make(chan bool, 1)
 			go func() {
-				HealthSubscription, err := node.PubsubServer.Subscribe(node.Ctx, c.name, events.QueryHealthStatus)
+				HealthSubscription, err := node.PubsubServer.Subscribe(context.Background(), c.name, events.QueryHealthStatus)
 				ready <- true
 				assert.NoError(err)
 				select {