Minor updates

Author: alpe

node/failover.go

@@ -166,6 +166,7 @@ func setupFailoverState(
 
 func (f *failoverState) Run(pCtx context.Context) (multiErr error) {
 	stopService := func(stoppable func(context.Context) error, name string) {
+		// parent context is cancelled already, so we need to create a new one
 		shutdownCtx, done := context.WithTimeout(context.Background(), 3*time.Second)
 		defer done()
 

pkg/raft/election.go

@@ -63,6 +63,7 @@ func (d *DynamicLeaderElection) Run(ctx context.Context) error {
 	}()
 
 	startWorker := func(name string, workerFunc func(ctx context.Context) error) {
+		workerCancel()
 		workerCtx, cancel := context.WithCancel(ctx)
 		workerCancel = cancel
 		wg.Add(1)
@@ -89,12 +90,20 @@ func (d *DynamicLeaderElection) Run(ctx context.Context) error {
 			d.logger.Info().Msg("Raft leader changed notification")
 			if becameLeader && !isCurrentlyLeader { // new leader
 				if isStarted {
+					var synced bool
 					d.logger.Info().Msg("became leader, stopping follower operations")
 					// wait for in flight raft msgs to land
-					time.Sleep(d.node.Config().SendTimeout)
-					if !runnable.IsSynced(d.node.GetState()) {
+				awaitSyncLoop:
+					for end := time.Now().Add(d.node.Config().SendTimeout); time.Now().Before(end); {
+						if synced = runnable.IsSynced(d.node.GetState()); synced {
+							break awaitSyncLoop
+						}
+						time.Sleep(d.node.Config().SendTimeout / 10)
+					}
+					if !synced && !runnable.IsSynced(d.node.GetState()) {
 						d.logger.Info().Msg("became leader, but not synced. Pass on leadership")
 						if err := d.node.leadershipTransfer(); err != nil && !errors.Is(err, raft.ErrNotLeader) {
+							// the leadership transfer can fail due to no suitable leader. Better stop than double sign on old state
 							return err
 						}
 						continue
@@ -127,10 +136,12 @@ func (d *DynamicLeaderElection) Run(ctx context.Context) error {
 		case <-ticker.C: // LeaderCh fires only when leader changes not on initial election
 			if isStarted {
 				ticker.Stop()
+				ticker.C = nil
 				continue
 			}
 			if leaderID := d.node.leaderID(); leaderID != "" && leaderID != d.node.NodeID() {
 				ticker.Stop()
+				ticker.C = nil
 				d.logger.Info().Msg("starting follower operations")
 				isStarted = true
 				var err error

pkg/raft/node.go

@@ -112,6 +112,11 @@ func (n *Node) Start(_ context.Context) error {
 		return fmt.Errorf("raft cluster requires bootstrap mode")
 	}
 
+	if future := n.raft.GetConfiguration(); future.Error() == nil && len(future.Configuration().Servers) > 0 {
+		n.logger.Info().Msg("cluster already bootstrapped, skipping")
+		return nil
+	}
+
 	n.logger.Info().Msg("Boostrap raft cluster")
 	thisNode := raft.Server{ID: raft.ServerID(n.config.NodeID), Address: raft.ServerAddress(n.config.RaftAddr)}
 	cfg := raft.Configuration{
@@ -264,6 +269,7 @@ func (f *FSM) Apply(log *raft.Log) interface{} {
 		select {
 		case f.applyCh <- RaftApplyMsg{Index: log.Index, State: &state}:
 		default:
+			// on a slow consumer, the raft cluster should not be blocked. Followers can sync from DA or other peers, too.
 			f.logger.Warn().Msg("apply channel full, dropping message")
 		}
 	}
@@ -326,6 +332,7 @@ func splitPeerAddr(peer string) (raft.Server, error) {
 	if address == "" {
 		return raft.Server{}, errors.New("address cannot be empty")
 	}
+	// we can skip address validation as they come from a local configuration
 
 	return raft.Server{
 		ID:      raft.ServerID(nodeID),

pkg/raft/types.go

@@ -2,8 +2,7 @@ package raft
 
 import "fmt"
 
-// todo: refactor to use proto
-// RaftBlockState represents replicated block state
+// RaftBlockState represents a replicated block state
 type RaftBlockState struct {
 	Height    uint64
 	Hash      []byte