fix(raft): abdicate leadership when store is significantly behind raft state

When a node wins election but its local store is more than 1 block behind
the raft FSM state, RecoverFromRaft cannot replay the gap (it only handles
the single latest block in the raft snapshot). Previously the node would
log "recovery successful" and start leader operations anyway, then stall
block production while the executor repeatedly failed to sync the missing
blocks from a store that did not have them.

Fix: in DynamicLeaderElection.Run, detect diff < -1 at the
follower→leader transition and immediately transfer leadership to a
better-synced peer. diff == -1 is preserved: RecoverFromRaft can apply
exactly one block from the raft snapshot, so that path is unchanged.

Closes #3255

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: auricom

pkg/raft/election.go

@@ -132,6 +132,23 @@ func (d *DynamicLeaderElection) Run(ctx context.Context) error {
 					if err != nil {
 						return err
 					}
+					if diff < -1 {
+						// Store is more than 1 block behind raft state.
+						// RecoverFromRaft can only apply the single latest block
+						// from the raft snapshot; it cannot replay a larger gap.
+						// Starting leader operations in this state would stall block
+						// production until catch-up completes (potentially minutes or
+						// hours). Abdicate immediately so a better-synced peer can
+						// take leadership.
+						d.logger.Warn().
+							Int("store_lag_blocks", -diff).
+							Uint64("raft_height", raftState.Height).
+							Msg("became leader but store is significantly behind raft state; abdicating to prevent stalled block production")
+						if tErr := d.node.leadershipTransfer(); tErr != nil {
+							d.logger.Error().Err(tErr).Msg("leadership transfer failed after store-lag abdication")
+						}
+						continue
+					}
 					if diff != 0 {
 						d.logger.Info().Msg("became leader but not synced, attempting recovery")
 						if err := runnable.Recover(ctx, raftState); err != nil {

pkg/raft/election_test.go

@@ -221,6 +221,52 @@ func TestDynamicLeaderElectionRun(t *testing.T) {
 				assert.ErrorIs(t, err, context.Canceled)
 			},
 		},
+		"abdicate when store significantly behind raft": {
+			setup: func(t *testing.T) (*DynamicLeaderElection, context.Context, context.CancelFunc) {
+				m := newMocksourceNode(t)
+				leaderCh := make(chan bool, 2)
+				m.EXPECT().leaderCh().Return((<-chan bool)(leaderCh))
+				// GetState called in verifyState (follower start) and in leader sync check
+				m.EXPECT().GetState().Return(&RaftBlockState{Height: 10})
+				m.EXPECT().GetState().Return(&RaftBlockState{Height: 10})
+				m.EXPECT().Config().Return(testCfg()).Times(2)
+				m.EXPECT().waitForMsgsLanded(2 * time.Millisecond).Return(nil)
+				m.EXPECT().NodeID().Return("self")
+				m.EXPECT().leaderID().Return("self")
+				// Abdication must transfer leadership
+				m.EXPECT().leadershipTransfer().Return(nil)
+
+				fStarted := make(chan struct{})
+				follower := &testRunnable{
+					startedCh:  fStarted,
+					isSyncedFn: func(*RaftBlockState) (int, error) { return -5, nil },
+				}
+				// Leader must never start
+				leader := &testRunnable{runFn: func(ctx context.Context) error {
+					t.Fatal("leader should not start when store is significantly behind raft")
+					return nil
+				}}
+
+				logger := zerolog.Nop()
+				d := &DynamicLeaderElection{logger: logger, node: m,
+					leaderFactory:   func() (Runnable, error) { return leader, nil },
+					followerFactory: func() (Runnable, error) { return follower, nil },
+				}
+				ctx, cancel := context.WithCancel(t.Context())
+				go func() {
+					leaderCh <- false
+					<-fStarted
+					leaderCh <- true
+					time.Sleep(10 * time.Millisecond)
+					cancel()
+				}()
+				return d, ctx, cancel
+			},
+			assertF: func(t *testing.T, err error) {
+				require.Error(t, err)
+				assert.ErrorIs(t, err, context.Canceled)
+			},
+		},
 		"lost leadership during sync wait": {
 			setup: func(t *testing.T) (*DynamicLeaderElection, context.Context, context.CancelFunc) {
 				m := newMocksourceNode(t)