fix: raft HA production hardening — leader fencing, log compaction, election timeout, audit log

block/internal/syncing/raft_retriever.go

@@ -74,6 +74,7 @@ func (r *raftRetriever) Stop() {
 	r.mtx.Unlock()
 
 	r.wg.Wait()
+	r.raftNode.SetApplyCallback(nil)
 }
 
 // raftApplyLoop processes blocks received from raft

block/internal/syncing/raft_retriever_test.go

@@ -0,0 +1,61 @@
+package syncing
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+
+	"github.com/evstack/ev-node/pkg/genesis"
+	pkgraft "github.com/evstack/ev-node/pkg/raft"
+)
+
+type stubRaftNode struct {
+	mu        sync.Mutex
+	callbacks []chan<- pkgraft.RaftApplyMsg
+}
+
+func (s *stubRaftNode) IsLeader() bool { return false }
+
+func (s *stubRaftNode) HasQuorum() bool { return false }
+
+func (s *stubRaftNode) GetState() *pkgraft.RaftBlockState { return nil }
+
+func (s *stubRaftNode) Broadcast(context.Context, *pkgraft.RaftBlockState) error { return nil }
+
+func (s *stubRaftNode) SetApplyCallback(ch chan<- pkgraft.RaftApplyMsg) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.callbacks = append(s.callbacks, ch)
+}
+
+func (s *stubRaftNode) recordedCallbacks() []chan<- pkgraft.RaftApplyMsg {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]chan<- pkgraft.RaftApplyMsg, len(s.callbacks))
+	copy(out, s.callbacks)
+	return out
+}
+
+func TestRaftRetrieverStopClearsApplyCallback(t *testing.T) {
+	t.Parallel()
+
+	raftNode := &stubRaftNode{}
+	retriever := newRaftRetriever(
+		raftNode,
+		genesis.Genesis{},
+		zerolog.Nop(),
+		nil,
+		func(context.Context, *pkgraft.RaftBlockState) error { return nil },
+	)
+
+	require.NoError(t, retriever.Start(t.Context()))
+	retriever.Stop()
+
+	callbacks := raftNode.recordedCallbacks()
+	require.Len(t, callbacks, 2)
+	require.NotNil(t, callbacks[0])
+	require.Nil(t, callbacks[1])
+}

block/internal/syncing/syncer.go

@@ -243,6 +243,11 @@ func (s *Syncer) Stop(ctx context.Context) error {
 	if s.daFollower != nil {
 		s.daFollower.Stop()
 	}
+
+	if s.raftRetriever != nil {
+		s.raftRetriever.Stop()
+	}
+
 	s.wg.Wait()
 
 	// Skip draining if we're shutting down due to a critical error (e.g. execution
@@ -1234,7 +1239,26 @@ func (s *Syncer) RecoverFromRaft(ctx context.Context, raftState *raft.RaftBlockS
 	}
 
 	if currentState.LastBlockHeight > raftState.Height {
-		return fmt.Errorf("invalid block height: %d (expected %d)", raftState.Height, currentState.LastBlockHeight+1)
+		// Local EVM is ahead of the raft snapshot. This is expected on restart when
+		// the raft FSM hasn't finished replaying log entries yet (stale snapshot height),
+		// or when log entries were compacted and the FSM is awaiting a new snapshot from
+		// the leader. Verify that our local block at raftState.Height has the same hash
+		// to confirm shared history before skipping recovery.
+		localHeader, err := s.store.GetHeader(ctx, raftState.Height)
+		if err != nil {
+			return fmt.Errorf("local state ahead of raft snapshot (local=%d raft=%d), cannot verify hash: %w",
+				currentState.LastBlockHeight, raftState.Height, err)
+		}
+		localHash := localHeader.Hash()
+		if !bytes.Equal(localHash, raftState.Hash) {
+			return fmt.Errorf("local state diverged from raft at height %d: local hash %x != raft hash %x",
+				raftState.Height, localHash, raftState.Hash)
+		}
+		s.logger.Info().
+			Uint64("local_height", currentState.LastBlockHeight).
+			Uint64("raft_height", raftState.Height).
+			Msg("local state ahead of stale raft snapshot with matching hash; skipping recovery, raft will catch up")
+		return nil
 	}
 
 	return nil

block/internal/syncing/syncer_test.go

@@ -422,6 +422,171 @@ func TestSyncer_RecoverFromRaft_KeepsStrictValidationAfterStateExists(t *testing
 	require.ErrorContains(t, err, "invalid chain ID")
 }
 
+// TestSyncer_RecoverFromRaft_LocalAheadOfStaleSnapshot tests Bug A: when the node
+// restarts and the EVM is ahead of the raft FSM snapshot (stale snapshot due to
+// timing or log compaction), RecoverFromRaft should skip recovery if the local
+// block at raftState.Height has a matching hash, rather than crashing.
+func TestSyncer_RecoverFromRaft_LocalAheadOfStaleSnapshot(t *testing.T) {
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	st := store.New(ds)
+
+	cm, err := cache.NewManager(config.DefaultConfig(), st, zerolog.Nop())
+	require.NoError(t, err)
+
+	addr, pub, signer := buildSyncTestSigner(t)
+	gen := genesis.Genesis{
+		ChainID:         "1234",
+		InitialHeight:   1,
+		StartTime:       time.Now().Add(-time.Second),
+		ProposerAddress: addr,
+	}
+
+	mockExec := testmocks.NewMockExecutor(t)
+	mockHeaderStore := extmocks.NewMockStore[*types.P2PSignedHeader](t)
+	mockDataStore := extmocks.NewMockStore[*types.P2PData](t)
+	s := NewSyncer(
+		st,
+		mockExec,
+		nil,
+		cm,
+		common.NopMetrics(),
+		config.DefaultConfig(),
+		gen,
+		mockHeaderStore,
+		mockDataStore,
+		zerolog.Nop(),
+		common.DefaultBlockOptions(),
+		make(chan error, 1),
+		nil,
+	)
+
+	// Build block at height 1 and persist it (simulates EVM block persisted before SIGTERM).
+	data1 := makeData(gen.ChainID, 1, 0)
+	headerBz1, hdr1 := makeSignedHeaderBytes(t, gen.ChainID, 1, addr, pub, signer, []byte("app1"), data1, nil)
+	dataBz1, err := data1.MarshalBinary()
+	require.NoError(t, err)
+
+	batch, err := st.NewBatch(t.Context())
+	require.NoError(t, err)
+	require.NoError(t, batch.SaveBlockDataFromBytes(hdr1, headerBz1, dataBz1, &hdr1.Signature))
+	require.NoError(t, batch.SetHeight(1))
+	require.NoError(t, batch.UpdateState(types.State{
+		ChainID:         gen.ChainID,
+		InitialHeight:   1,
+		LastBlockHeight: 1,
+		LastHeaderHash:  hdr1.Hash(),
+	}))
+	require.NoError(t, batch.Commit())
+
+	// Simulate EVM at height 1, raft snapshot stale at height 0 — but there is no
+	// block 0 to check, so use height 1 EVM vs stale snapshot at height 0.
+	// More realistic: EVM at height 2, raft snapshot at height 1.
+	// Build a second block and advance the store state to height 2.
+	data2 := makeData(gen.ChainID, 2, 0)
+	headerBz2, hdr2 := makeSignedHeaderBytes(t, gen.ChainID, 2, addr, pub, signer, []byte("app2"), data2, hdr1.Hash())
+	dataBz2, err := data2.MarshalBinary()
+	require.NoError(t, err)
+
+	batch2, err := st.NewBatch(t.Context())
+	require.NoError(t, err)
+	require.NoError(t, batch2.SaveBlockDataFromBytes(hdr2, headerBz2, dataBz2, &hdr2.Signature))
+	require.NoError(t, batch2.SetHeight(2))
+	require.NoError(t, batch2.UpdateState(types.State{
+		ChainID:         gen.ChainID,
+		InitialHeight:   1,
+		LastBlockHeight: 2,
+		LastHeaderHash:  hdr2.Hash(),
+	}))
+	require.NoError(t, batch2.Commit())
+
+	// Set lastState to height 2 (EVM is at 2).
+	s.SetLastState(types.State{
+		ChainID:         gen.ChainID,
+		InitialHeight:   1,
+		LastBlockHeight: 2,
+		LastHeaderHash:  hdr2.Hash(),
+	})
+
+	t.Run("matching hash skips recovery", func(t *testing.T) {
+		// raft snapshot is stale at height 1 (EVM is at 2); hash matches local block 1.
+		err := s.RecoverFromRaft(t.Context(), &raft.RaftBlockState{
+			Height: 1,
+			Hash:   hdr1.Hash(),
+			Header: headerBz1,
+			Data:   dataBz1,
+		})
+		require.NoError(t, err, "local ahead of stale raft snapshot with matching hash should not error")
+	})
+
+	t.Run("diverged hash returns error", func(t *testing.T) {
+		wrongHash := make([]byte, len(hdr1.Hash()))
+		copy(wrongHash, hdr1.Hash())
+		wrongHash[0] ^= 0xFF // flip a byte to produce a different hash
+
+		err := s.RecoverFromRaft(t.Context(), &raft.RaftBlockState{
+			Height: 1,
+			Hash:   wrongHash,
+			Header: headerBz1,
+			Data:   dataBz1,
+		})
+		require.Error(t, err)
+		require.ErrorContains(t, err, "diverged from raft")
+	})
+
+	t.Run("get header fails returns error", func(t *testing.T) {
+		// lastState is at height 2; raft snapshot at height 0.
+		// No block is stored at height 0, so GetHeader fails.
+		err := s.RecoverFromRaft(t.Context(), &raft.RaftBlockState{
+			Height: 0,
+			Hash:   make([]byte, 32),
+			Header: headerBz1,
+			Data:   dataBz1,
+		})
+		require.Error(t, err)
+		require.ErrorContains(t, err, "cannot verify hash")
+	})
+}
+
+func TestSyncer_Stop_CallsRaftRetrieverStop(t *testing.T) {
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	st := store.New(ds)
+
+	cm, err := cache.NewManager(config.DefaultConfig(), st, zerolog.Nop())
+	require.NoError(t, err)
+
+	raftNode := &stubRaftNode{}
+	s := NewSyncer(
+		st,
+		nil,
+		nil,
+		cm,
+		common.NopMetrics(),
+		config.DefaultConfig(),
+		genesis.Genesis{},
+		nil,
+		nil,
+		zerolog.Nop(),
+		common.DefaultBlockOptions(),
+		make(chan error, 1),
+		raftNode,
+	)
+
+	require.NotNil(t, s.raftRetriever, "raftRetriever should be set when raftNode is provided")
+
+	// Manually set cancel so Stop() doesn't bail out early (simulates having been started).
+	ctx, cancel := context.WithCancel(t.Context())
+	s.ctx = ctx
+	s.cancel = cancel
+
+	require.NoError(t, s.Stop(t.Context()))
+
+	// raftRetriever.Stop clears the apply callback (sets it to nil).
+	// The stub records each SetApplyCallback call; the last one should be nil.
+	callbacks := raftNode.recordedCallbacks()
+	require.NotEmpty(t, callbacks, "expected at least one callback registration")
+	assert.Nil(t, callbacks[len(callbacks)-1], "last callback should be nil after Stop")
+}
+
 func TestSyncer_processPendingEvents(t *testing.T) {
 	ds := dssync.MutexWrap(datastore.NewMapDatastore())
 	st := store.New(ds)

node/full.go

@@ -35,6 +35,7 @@ const (
 )
 
 var _ Node = &FullNode{}
+var _ LeaderResigner = &FullNode{}
 
 type leaderElection interface {
 	Run(ctx context.Context) error
@@ -154,8 +155,12 @@ func initRaftNode(nodeConfig config.Config, logger zerolog.Logger) (*raftpkg.Nod
 		Bootstrap:          nodeConfig.Raft.Bootstrap,
 		SnapCount:          nodeConfig.Raft.SnapCount,
 		SendTimeout:        nodeConfig.Raft.SendTimeout,
+		ShutdownTimeout:    nodeConfig.Raft.ShutdownTimeout,
 		HeartbeatTimeout:   nodeConfig.Raft.HeartbeatTimeout,
 		LeaderLeaseTimeout: nodeConfig.Raft.LeaderLeaseTimeout,
+		ElectionTimeout:    nodeConfig.Raft.ElectionTimeout,
+		SnapshotThreshold:  nodeConfig.Raft.SnapshotThreshold,
+		TrailingLogs:       nodeConfig.Raft.TrailingLogs,
 	}
 
 	if nodeConfig.Raft.Peers != "" {
@@ -384,3 +389,12 @@ func (n *FullNode) GetGenesisChunks() ([]string, error) {
 func (n *FullNode) IsRunning() bool {
 	return n.leaderElection.IsRunning()
 }
+
+// ResignLeader transfers raft leadership before the node shuts down.
+// It is a no-op when raft is not enabled or this node is not the leader.
+func (n *FullNode) ResignLeader() error {
+	if n.raftNode == nil {
+		return nil
+	}
+	return n.raftNode.ResignLeader()
+}

node/full_node_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	raftpkg "github.com/evstack/ev-node/pkg/raft"
 	"github.com/evstack/ev-node/pkg/service"
 )
 
@@ -82,3 +83,15 @@ func TestStartInstrumentationServer(t *testing.T) {
 		assert.NoError(err, "Pprof server shutdown should not return error")
 	}
 }
+
+func TestFullNode_ResignLeader_NilRaftNode(t *testing.T) {
+	n := &FullNode{} // raftNode is nil
+	assert.NoError(t, n.ResignLeader())
+}
+
+func TestFullNode_ResignLeader_NonLeaderRaftNode(t *testing.T) {
+	// Empty *raftpkg.Node has nil raft field so IsLeader() returns false;
+	// ResignLeader() is a no-op and returns nil.
+	n := &FullNode{raftNode: &raftpkg.Node{}}
+	assert.NoError(t, n.ResignLeader())
+}

node/node.go

@@ -21,6 +21,13 @@ type Node interface {
 	IsRunning() bool
 }
 
+// LeaderResigner is an optional interface implemented by nodes that participate
+// in Raft leader election. Callers should type-assert to this interface and call
+// ResignLeader before cancelling the node context on graceful shutdown.
+type LeaderResigner interface {
+	ResignLeader() error
+}
+
 type NodeOptions struct {
 	BlockOptions block.BlockOptions
 }

pkg/cmd/run_node.go

@@ -224,6 +224,25 @@ func StartNode(
 	select {
 	case <-quit:
 		logger.Info().Msg("shutting down node...")
+		// Proactively resign Raft leadership before cancelling the worker context.
+		// This gives the cluster a chance to elect a new leader before this node
+		// stops producing blocks, shrinking the unconfirmed-block window.
+		if resigner, ok := rollnode.(node.LeaderResigner); ok {
+			resignCtx, resignCancel := context.WithTimeout(context.Background(), 3*time.Second)
+			defer resignCancel()
+			resignDone := make(chan error, 1)
+			go func() { resignDone <- resigner.ResignLeader() }()
+			select {
+			case err := <-resignDone:
+				if err != nil {
+					logger.Warn().Err(err).Msg("leadership resign on shutdown failed")
+				} else {
+					logger.Info().Msg("leadership resigned before shutdown")
+				}
+			case <-resignCtx.Done():
+				logger.Warn().Msg("leadership resign timed out")
+			}
+		}
 		cancel()
 	case err := <-errCh:
 		if err != nil && !errors.Is(err, context.Canceled) {