x

Author: alpe

node/full.go

@@ -146,12 +146,13 @@ func initRaftNode(nodeConfig config.Config, logger zerolog.Logger) (*raftpkg.Nod
 	}
 
 	raftCfg := &raftpkg.Config{
-		NodeID:      nodeConfig.Raft.NodeID,
-		RaftAddr:    nodeConfig.Raft.RaftAddr,
-		RaftDir:     raftDir,
-		Bootstrap:   nodeConfig.Raft.Bootstrap,
-		SnapCount:   nodeConfig.Raft.SnapCount,
-		SendTimeout: nodeConfig.Raft.SendTimeout,
+		NodeID:           nodeConfig.Raft.NodeID,
+		RaftAddr:         nodeConfig.Raft.RaftAddr,
+		RaftDir:          raftDir,
+		Bootstrap:        nodeConfig.Raft.Bootstrap,
+		SnapCount:        nodeConfig.Raft.SnapCount,
+		SendTimeout:      nodeConfig.Raft.SendTimeout,
+		HeartbeatTimeout: nodeConfig.Raft.HeartbeatTimeout,
 	}
 
 	if nodeConfig.Raft.Peers != "" {

pkg/config/config.go

@@ -142,6 +142,8 @@ const (
 	FlagRaftSnapCount = FlagPrefixEvnode + "raft.snap_count"
 	// FlagRaftSendTimeout max time to wait for a message to be sent to a peer 
 	FlagRaftSendTimeout = FlagPrefixEvnode + "raft.send_timeout"
+	// FlagRaftHeartbeatTimeout is a flag for specifying heartbeat timeout
+	FlagRaftHeartbeatTimeout = FlagPrefixEvnode + "raft.heartbeat_timeout"
 )
 
 // Config stores Rollkit configuration.
@@ -252,14 +254,15 @@ type RPCConfig struct {
 
 // RaftConfig contains all Raft consensus configuration parameters
 type RaftConfig struct {
-	Enable      bool          `mapstructure:"enable" yaml:"enable" comment:"Enable Raft consensus for leader election and state replication"`
-	NodeID      string        `mapstructure:"node_id" yaml:"node_id" comment:"Unique identifier for this node in the Raft cluster"`
-	RaftAddr    string        `mapstructure:"raft_addr" yaml:"raft_addr" comment:"Address for Raft communication (host:port)"`
-	RaftDir     string        `mapstructure:"raft_dir" yaml:"raft_dir" comment:"Directory for Raft logs and snapshots"`
-	Bootstrap   bool          `mapstructure:"bootstrap" yaml:"bootstrap" comment:"Bootstrap a new Raft cluster (only for the first node)"`
-	Peers       string        `mapstructure:"peers" yaml:"peers" comment:"Comma-separated list of peer Raft addresses (nodeID@host:port)"`
-	SnapCount   uint64        `mapstructure:"snap_count" yaml:"snap_count" comment:"Number of log entries between snapshots"`
-	SendTimeout time.Duration `mapstructure:"send_timeout" yaml:"send_timeout" comment:"Max duration to wait for a message to be sent to a peer"`
+	Enable           bool          `mapstructure:"enable" yaml:"enable" comment:"Enable Raft consensus for leader election and state replication"`
+	NodeID           string        `mapstructure:"node_id" yaml:"node_id" comment:"Unique identifier for this node in the Raft cluster"`
+	RaftAddr         string        `mapstructure:"raft_addr" yaml:"raft_addr" comment:"Address for Raft communication (host:port)"`
+	RaftDir          string        `mapstructure:"raft_dir" yaml:"raft_dir" comment:"Directory for Raft logs and snapshots"`
+	Bootstrap        bool          `mapstructure:"bootstrap" yaml:"bootstrap" comment:"Bootstrap a new Raft cluster (only for the first node)"`
+	Peers            string        `mapstructure:"peers" yaml:"peers" comment:"Comma-separated list of peer Raft addresses (nodeID@host:port)"`
+	SnapCount        uint64        `mapstructure:"snap_count" yaml:"snap_count" comment:"Number of log entries between snapshots"`
+	SendTimeout      time.Duration `mapstructure:"send_timeout" yaml:"send_timeout" comment:"Max duration to wait for a message to be sent to a peer"`
+	HeartbeatTimeout time.Duration `mapstructure:"heartbeat_timeout" yaml:"heartbeat_timeout" comment:"Time between leader heartbeats to followers"`
 }
 
 // Validate ensures validates the config and ensure that the root directory exists.
@@ -394,6 +397,7 @@ func AddFlags(cmd *cobra.Command) {
 	cmd.Flags().String(FlagRaftPeers, def.Raft.Peers, "comma-separated list of peer Raft addresses (nodeID@host:port)")
 	cmd.Flags().Uint64(FlagRaftSnapCount, def.Raft.SnapCount, "number of log entries between snapshots")
 	cmd.Flags().Duration(FlagRaftSendTimeout, def.Raft.SendTimeout, "max duration to wait for a message to be sent to a peer")
+	cmd.Flags().Duration(FlagRaftHeartbeatTimeout, def.Raft.HeartbeatTimeout, "time between leader heartbeats to followers")
 }
 
 // Load loads the node configuration in the following order of precedence:

pkg/config/defaults.go

@@ -77,6 +77,9 @@ func DefaultConfig() Config {
 			Address:               "127.0.0.1:7331",
 			EnableDAVisualization: false,
 		},
+		Raft: RaftConfig{
+			HeartbeatTimeout: 1000 * time.Millisecond,
+		},
 	}
 }
 

pkg/raft/node.go

@@ -33,13 +33,14 @@ type Node struct {
 
 // Config holds raft node configuration
 type Config struct {
-	NodeID      string
-	RaftAddr    string
-	RaftDir     string
-	Bootstrap   bool
-	Peers       []string
-	SnapCount   uint64
-	SendTimeout time.Duration
+	NodeID           string
+	RaftAddr         string
+	RaftDir          string
+	Bootstrap        bool
+	Peers            []string
+	SnapCount        uint64
+	SendTimeout      time.Duration
+	HeartbeatTimeout time.Duration
 }
 
 // FSM implements raft.FSM for block state
@@ -58,6 +59,8 @@ func NewNode(cfg *Config, clusterClient clusterClient, logger zerolog.Logger) (*
 	raftConfig := raft.DefaultConfig()
 	raftConfig.LocalID = raft.ServerID(cfg.NodeID)
 	raftConfig.LogLevel = "INFO"
+	raftConfig.HeartbeatTimeout = cfg.HeartbeatTimeout
+	raftConfig.LeaderLeaseTimeout = cfg.HeartbeatTimeout / 2
 
 	fsm := &FSM{
 		logger: logger.With().Str("component", "raft-fsm").Logger(),

test/e2e/failover_e2e_test.go

@@ -135,7 +135,8 @@ func TestLeaseFailoverE2E(t *testing.T) {
 	require.EventuallyWithT(t, func(collect *assert.CollectT) {
 		newLeader = clusterNodes.Leader(collect)
 	}, 5*time.Second, 200*time.Millisecond)
-	t.Logf("+++ New leader: %s within %s\n", newLeader, time.Since(leaderElectionStart))
+	leaderFailoverTime := time.Since(leaderElectionStart)
+	t.Logf("+++ New leader: %s within %s\n", newLeader, leaderFailoverTime)
 	require.NotEqual(t, oldLeader, newLeader)
 
 	_, blk2 := submitTxToURL(t, clusterNodes.Details(newLeader).ethClient(t))
@@ -164,6 +165,7 @@ func TestLeaseFailoverE2E(t *testing.T) {
 
 	// Cleanup processes
 	clusterNodes.killAll()
+	t.Logf("Completed leader change in: %s", leaderFailoverTime)
 }
 
 func initChain(t *testing.T, sut *SystemUnderTest, workDir string) {
@@ -222,7 +224,8 @@ func setupRaftSequencerNode(
 		"--evnode.raft.bootstrap="+strconv.FormatBool(bootstrap),
 		"--evnode.raft.peers="+strings.Join(raftPeers, ","),
 		"--evnode.raft.snap_count=10",
-		"--evnode.raft.send_timeout=5s",
+		"--evnode.raft.send_timeout=300ms",
+		"--evnode.raft.heartbeat_timeout=300ms",
 
 		"--rollkit.p2p.peers", p2pPeers,
 		"--rollkit.rpc.address", rpcAddr,
@@ -268,7 +271,11 @@ func submitTxToURL(t *testing.T, client *ethclient.Client) (common.Hash, uint64)
 
 func queryLastDAHeight(t *testing.T, startHeight uint64, jwtSecret string, daAddress string) uint64 {
 	t.Helper()
-	client, err := jsonrpc.NewClient(t.Context(), zerolog.New(zerolog.NewTestWriter(t)), daAddress, jwtSecret, 0, 1, 0)
+	logger := zerolog.Nop()
+	if testing.Verbose() {
+		logger = zerolog.New(zerolog.NewTestWriter(t)).Level(zerolog.DebugLevel)
+	}
+	client, err := jsonrpc.NewClient(t.Context(), logger, daAddress, jwtSecret, 0, 1, 0)
 	require.NoError(t, err)
 	defer client.Close()
 	var lastDABlock = startHeight
@@ -280,7 +287,7 @@ func queryLastDAHeight(t *testing.T, startHeight uint64, jwtSecret string, daAdd
 			}
 			t.Fatal("failed to get IDs:", err)
 		}
-		if len(res.IDs) != 0 {
+		if len(res.IDs) != 0 && testing.Verbose() {
 			t.Log("+++ DA block: ", lastDABlock, " ids: ", len(res.IDs))
 		}
 		lastDABlock++