Review feedback

Author: alpe

block/internal/common/raft.go

@@ -9,7 +9,7 @@ import (
 // RaftNode interface for raft consensus integration
 type RaftNode interface {
 	IsLeader() bool
-	GetState() *raft.RaftBlockState
+	GetState() raft.RaftBlockState
 
 	Broadcast(ctx context.Context, state *raft.RaftBlockState) error
 

block/internal/executing/executor.go

@@ -215,7 +215,7 @@ func (e *Executor) initializeState() error {
 
 	if e.raftNode != nil {
 		// ensure node is fully synced before producing any blocks
-		if raftState := e.raftNode.GetState(); raftState != nil && raftState.Height != state.LastBlockHeight {
+		if raftState := e.raftNode.GetState(); raftState.Height != 0 && raftState.Height != state.LastBlockHeight {
 			return fmt.Errorf("invalid state: node is not synced with the chain: raft %d != %d state", raftState.Height, state.LastBlockHeight)
 		}
 	}
@@ -591,7 +591,7 @@ func (e *Executor) createBlock(ctx context.Context, height uint64, batchData *Ba
 	}
 
 	for i, tx := range batchData.Transactions {
-		data.Txs[i] = types.Tx(tx)
+		data.Txs[i] = tx
 	}
 
 	// Set data hash

node/failover.go

@@ -200,13 +200,12 @@ func (f *failoverState) Run(ctx context.Context) (multiErr error) {
 		}
 	}()
 
-	defer func() {
-		if err := f.bc.Stop(); err != nil && !errors.Is(err, context.Canceled) {
-			multiErr = errors.Join(multiErr, fmt.Errorf("stopping block components: %w", err))
-		}
-	}()
-
 	go func() {
+		defer func() {
+			if err := f.bc.Stop(); err != nil && !errors.Is(err, context.Canceled) {
+				multiErr = errors.Join(multiErr, fmt.Errorf("stopping block components: %w", err))
+			}
+		}()
 		if err := f.bc.Start(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			select {
 			case errChan <- fmt.Errorf("components started with error: %w", err):
@@ -224,8 +223,8 @@ func (f *failoverState) Run(ctx context.Context) (multiErr error) {
 	return <-errChan
 }
 
-func (f *failoverState) IsSynced(s *raft.RaftBlockState) bool {
-	if s == nil || s.Height == 0 {
+func (f *failoverState) IsSynced(s raft.RaftBlockState) bool {
+	if s.Height == 0 {
 		return true
 	}
 	if f.bc.Syncer != nil {

node/single_sequencer_test.go

@@ -50,4 +50,5 @@ func TestStartup(t *testing.T) {
 
 	// Run the cleanup function from setupTestNodeWithCleanup
 	cleanup()
+	time.Sleep(time.Second) // shutdown takes some time to persist caches
 }

pkg/raft/election.go

@@ -8,6 +8,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/hashicorp/raft"
 	"github.com/rs/zerolog"
 )
 
@@ -18,15 +19,15 @@ var ErrLeadershipLost = fmt.Errorf("leader lock lost")
 // IsSynced checks whether the component is synced with the given RaftBlockState.
 type Runnable interface {
 	Run(ctx context.Context) error
-	IsSynced(*RaftBlockState) bool
+	IsSynced(RaftBlockState) bool
 }
 
 type sourceNode interface {
 	Config() Config
 	leaderCh() <-chan bool
 	leaderID() string
 	NodeID() string
-	GetState() *RaftBlockState
+	GetState() RaftBlockState
 	leadershipTransfer() error
 }
 
@@ -93,7 +94,7 @@ func (d *DynamicLeaderElection) Run(ctx context.Context) error {
 					time.Sleep(d.node.Config().SendTimeout)
 					if !runnable.IsSynced(d.node.GetState()) {
 						d.logger.Info().Msg("became leader, but not synced. Pass on leadership")
-						if err := d.node.leadershipTransfer(); err != nil {
+						if err := d.node.leadershipTransfer(); err != nil && !errors.Is(err, raft.ErrNotLeader) {
 							return err
 						}
 						continue

pkg/raft/election_test.go

@@ -109,11 +109,11 @@ func TestDynamicLeaderElectionRun(t *testing.T) {
 				leaderCh := make(chan bool, 2)
 				m.EXPECT().leaderCh().Return((<-chan bool)(leaderCh))
 				m.EXPECT().Config().Return(testCfg())
-				m.EXPECT().GetState().Return(&RaftBlockState{Height: 1})
+				m.EXPECT().GetState().Return(RaftBlockState{Height: 1})
 				m.EXPECT().leadershipTransfer().Return(nil)
 
 				fStarted := make(chan struct{})
-				follower := &testRunnable{startedCh: fStarted, isSyncedFn: func(*RaftBlockState) bool { return false }}
+				follower := &testRunnable{startedCh: fStarted, isSyncedFn: func(RaftBlockState) bool { return false }}
 				leader := &testRunnable{runFn: func(ctx context.Context) error {
 					t.Fatal("leader should not be running")
 					return nil
@@ -148,7 +148,7 @@ func TestDynamicLeaderElectionRun(t *testing.T) {
 				m.EXPECT().leaderCh().Return((<-chan bool)(leaderCh))
 				// On leadership change to true, election will sleep SendTimeout, then check sync against state
 				m.EXPECT().Config().Return(testCfg())
-				m.EXPECT().GetState().Return(&RaftBlockState{Height: 1})
+				m.EXPECT().GetState().Return(RaftBlockState{Height: 1})
 
 				fStarted := make(chan struct{})
 				lStarted := make(chan struct{})
@@ -211,7 +211,7 @@ func testCfg() Config {
 // These channels are used only by tests to observe state.
 type testRunnable struct {
 	runFn      func(ctx context.Context) error
-	isSyncedFn func(*RaftBlockState) bool
+	isSyncedFn func(RaftBlockState) bool
 	startedCh  chan struct{}
 	doneCh     chan struct{}
 }
@@ -238,7 +238,7 @@ func (t *testRunnable) Run(ctx context.Context) error {
 	return ctx.Err()
 }
 
-func (t *testRunnable) IsSynced(s *RaftBlockState) bool {
+func (t *testRunnable) IsSynced(s RaftBlockState) bool {
 	if t.isSyncedFn != nil {
 		return t.isSyncedFn(s)
 	}

pkg/raft/node.go

@@ -11,6 +11,7 @@ import (
 	"path/filepath"
 	"slices"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/hashicorp/raft"
@@ -47,7 +48,7 @@ type Config struct {
 // FSM implements raft.FSM for block state
 type FSM struct {
 	logger  zerolog.Logger
-	state   *RaftBlockState
+	state   *atomic.Pointer[RaftBlockState]
 	applyCh chan<- RaftApplyMsg
 }
 
@@ -63,9 +64,11 @@ func NewNode(cfg *Config, clusterClient clusterClient, logger zerolog.Logger) (*
 	raftConfig.HeartbeatTimeout = cfg.HeartbeatTimeout
 	raftConfig.LeaderLeaseTimeout = cfg.HeartbeatTimeout / 2
 
+	startPointer := new(atomic.Pointer[RaftBlockState])
+	startPointer.Store(&RaftBlockState{})
 	fsm := &FSM{
 		logger: logger.With().Str("component", "raft-fsm").Logger(),
-		state:  &RaftBlockState{},
+		state:  startPointer,
 	}
 
 	logStore, err := raftboltdb.NewBoltStore(filepath.Join(cfg.RaftDir, "raft-log.db"))
@@ -131,6 +134,7 @@ func (n *Node) Start(ctx context.Context) error {
 				if n.GetState().Height != 0 {
 					return nil
 				}
+				time.Sleep(time.Second / 10)
 			}
 		}
 	}
@@ -158,7 +162,6 @@ func (n *Node) Start(ctx context.Context) error {
 	}
 	n.logger.Info().Msg("bootstrapped raft cluster")
 	return nil
-
 }
 
 func (n *Node) awaitToBeClusterMember(ctx context.Context, nodeID raft.ServerID) error {
@@ -246,8 +249,8 @@ func (n *Node) Broadcast(_ context.Context, state *RaftBlockState) error {
 }
 
 // GetState returns the current replicated state
-func (n *Node) GetState() *RaftBlockState {
-	return n.fsm.state
+func (n *Node) GetState() RaftBlockState {
+	return *n.fsm.state.Load()
 }
 
 // AddPeer adds a peer to the raft cluster
@@ -304,10 +307,10 @@ func (f *FSM) Apply(log *raft.Log) interface{} {
 		f.logger.Error().Err(err).Msg("unmarshal block state")
 		return err
 	}
-	if err := f.state.assertValid(state); err != nil {
+	if err := f.state.Load().assertValid(state); err != nil {
 		return err
 	}
-	f.state = &state
+	f.state.Store(&state)
 	f.logger.Debug().Uint64("height", state.Height).Msg("received block state")
 
 	if f.applyCh != nil {
@@ -323,7 +326,7 @@ func (f *FSM) Apply(log *raft.Log) interface{} {
 
 // Snapshot implements raft.FSM
 func (f *FSM) Snapshot() (raft.FSMSnapshot, error) {
-	return &fsmSnapshot{state: f.state}, nil
+	return &fsmSnapshot{state: *f.state.Load()}, nil
 }
 
 // Restore implements raft.FSM
@@ -335,13 +338,13 @@ func (f *FSM) Restore(rc io.ReadCloser) error {
 		return fmt.Errorf("decode snapshot: %w", err)
 	}
 
-	f.state = &state
+	f.state.Store(&state)
 	f.logger.Info().Uint64("height", state.Height).Msg("restored from snapshot")
 	return nil
 }
 
 type fsmSnapshot struct {
-	state *RaftBlockState
+	state RaftBlockState
 }
 
 func (s *fsmSnapshot) Persist(sink raft.SnapshotSink) error {

pkg/raft/node_mock.go

@@ -12,6 +12,7 @@ type RaftBlockState struct {
 	Data      []byte
 }
 
+// assertValid checks basic constraints but does not ensure that no gaps exist or chain continuity
 func (s RaftBlockState) assertValid(next RaftBlockState) error {
 	if s.Height > next.Height {
 		return fmt.Errorf("invalid height: %d > %d", s.Height, next.Height)

pkg/raft/types.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"time"
 )
 
 // RaftClusterClient is simple http based client for raft cluster management
@@ -70,6 +71,6 @@ func (r RaftClusterClient) broadcast(ctx context.Context, path string, obj any)
 func NewRaftClusterClient(peers ...string) (RaftClusterClient, error) {
 	return RaftClusterClient{
 		peers:      peers,
-		httpClient: &http.Client{},
+		httpClient: &http.Client{Timeout: 3 * time.Second},
 	}, nil
 }