fix(syncing): make Syncer shutdown idempotent after start failure

block/internal/syncing/syncer.go

@@ -43,6 +43,14 @@ const (
 	fullnessThreshold = 0.8
 )
 
+type syncerLifecycleState uint8
+
+const (
+	syncerLifecycleNew syncerLifecycleState = iota
+	syncerLifecycleStarted
+	syncerLifecycleStopped
+)
+
 // Syncer handles block synchronization from DA and P2P sources.
 type Syncer struct {
 	// Core components
@@ -91,6 +99,8 @@ type Syncer struct {
 	lastCheckedEpochEnd  uint64              // highest epochEnd fully verified so far
 
 	// Lifecycle
+	lifecycleMu      sync.Mutex
+	lifecycleState   syncerLifecycleState
 	ctx              context.Context
 	cancel           context.CancelFunc
 	wg               sync.WaitGroup
@@ -165,11 +175,19 @@ func (s *Syncer) SetBlockSyncer(bs BlockSyncer) {
 // Start begins the syncing component
 // The component should not be started after being stopped.
 func (s *Syncer) Start(ctx context.Context) (err error) {
-	if s.cancel != nil {
+	s.lifecycleMu.Lock()
+	switch s.lifecycleState {
+	case syncerLifecycleStarted:
+		s.lifecycleMu.Unlock()
 		return errors.New("syncer already started")
+	case syncerLifecycleStopped:
+		s.lifecycleMu.Unlock()
+		return errors.New("syncer cannot be restarted after stopping")
 	}
 	ctx, cancel := context.WithCancel(ctx)
 	s.ctx, s.cancel = ctx, cancel
+	s.lifecycleState = syncerLifecycleStarted
+	s.lifecycleMu.Unlock()
 
 	defer func() { //nolint: contextcheck // use new context as parent can be cancelled already
 		if err != nil {
@@ -240,11 +258,17 @@ func (s *Syncer) Start(ctx context.Context) (err error) {
 
 // Stop shuts down the syncing component
 func (s *Syncer) Stop(ctx context.Context) error {
-	if s.cancel == nil {
+	s.lifecycleMu.Lock()
+	if s.lifecycleState != syncerLifecycleStarted {
+		s.lifecycleMu.Unlock()
 		return nil
 	}
+	cancel := s.cancel
+	s.cancel = nil
+	s.lifecycleState = syncerLifecycleStopped
+	s.lifecycleMu.Unlock()
 
-	s.cancel()
+	cancel()
 	s.cancelP2PWait(0)
 
 	if s.fiRetriever != nil {

block/internal/syncing/syncer_benchmark_test.go

@@ -137,6 +137,7 @@ func newBenchFixture(b *testing.B, totalHeights uint64, shuffledTx bool, daDelay
 	)
 	require.NoError(b, s.initializeState())
 	s.ctx, s.cancel = ctx, cancel
+	s.lifecycleState = syncerLifecycleStarted
 
 	// prepare height events to emit
 	heightEvents := make([]common.DAHeightEvent, totalHeights)

block/internal/syncing/syncer_test.go

@@ -42,6 +42,34 @@ type stubRaftNode struct {
 	callbacks []chan<- raft.RaftApplyMsg
 }
 
+type countingForcedInclusionRetriever struct {
+	stopCalls int
+}
+
+func (*countingForcedInclusionRetriever) RetrieveForcedIncludedTxs(context.Context, uint64) (*da.ForcedInclusionEvent, error) {
+	return nil, nil
+}
+
+func (*countingForcedInclusionRetriever) Start(context.Context) {}
+
+func (r *countingForcedInclusionRetriever) Stop() {
+	r.stopCalls++
+}
+
+type countingDAFollower struct {
+	stopCalls int
+}
+
+func (*countingDAFollower) Start(context.Context) error { return nil }
+
+func (f *countingDAFollower) Stop() {
+	f.stopCalls++
+}
+
+func (*countingDAFollower) HasReachedHead() bool { return false }
+
+func (*countingDAFollower) QueuePriorityHeight(uint64) {}
+
 func (s *stubRaftNode) IsLeader() bool                                        { return false }
 func (s *stubRaftNode) HasQuorum() bool                                       { return false }
 func (s *stubRaftNode) GetState() *raft.RaftBlockState                        { return nil }
@@ -1073,6 +1101,7 @@ func TestSyncer_Stop_CallsRaftRetrieverStop(t *testing.T) {
 	ctx, cancel := context.WithCancel(t.Context())
 	s.ctx = ctx
 	s.cancel = cancel
+	s.lifecycleState = syncerLifecycleStarted
 
 	require.NoError(t, s.Stop(t.Context()))
 
@@ -1083,6 +1112,48 @@ func TestSyncer_Stop_CallsRaftRetrieverStop(t *testing.T) {
 	assert.Nil(t, callbacks[len(callbacks)-1], "last callback should be nil after Stop")
 }
 
+func TestSyncer_Stop_IsIdempotentAfterStartFailure(t *testing.T) {
+	mockStore := testmocks.NewMockStore(t)
+	mockStore.EXPECT().GetState(mock.Anything).Return(types.State{DAHeight: 1}, nil).Once()
+
+	raftNode := &stubRaftNode{}
+	fiRetriever := &countingForcedInclusionRetriever{}
+	daFollower := &countingDAFollower{}
+	s := NewSyncer(
+		mockStore,
+		nil,
+		nil,
+		nil,
+		common.NopMetrics(),
+		config.DefaultConfig(),
+		genesis.Genesis{DAStartHeight: 2},
+		nil,
+		nil,
+		zerolog.Nop(),
+		common.DefaultBlockOptions(),
+		make(chan error, 1),
+		raftNode,
+	)
+	s.fiRetriever = fiRetriever
+	s.daFollower = daFollower
+
+	err := s.Start(t.Context())
+	require.ErrorContains(t, err, "DA height (1) is lower than DA start height (2)")
+	require.NoError(t, s.Stop(t.Context()))
+	require.NoError(t, s.Stop(t.Context()))
+
+	assert.Nil(t, s.cancel)
+	assert.Equal(t, syncerLifecycleStopped, s.lifecycleState)
+	assert.Equal(t, 1, fiRetriever.stopCalls, "forced inclusion retriever should only be stopped once")
+	assert.Equal(t, 1, daFollower.stopCalls, "DA follower should only be stopped once")
+	_, open := <-s.heightInCh
+	assert.False(t, open, "height input channel should be closed")
+
+	callbacks := raftNode.recordedCallbacks()
+	require.Len(t, callbacks, 1, "raft retriever should only be stopped once")
+	assert.Nil(t, callbacks[0])
+}
+
 func TestSyncer_processPendingEvents(t *testing.T) {
 	ds := dssync.MutexWrap(datastore.NewMapDatastore())
 	st := store.New(ds)
@@ -2019,6 +2090,7 @@ func TestSyncer_Stop_SkipsDrainOnCriticalError(t *testing.T) {
 	ctx, cancel := context.WithCancel(t.Context())
 	s.ctx = ctx
 	s.cancel = cancel
+	s.lifecycleState = syncerLifecycleStarted
 
 	// Enqueue events into heightInCh that would trigger ExecuteTxs if drained
 	lastState := s.getLastState()
@@ -2099,6 +2171,7 @@ func TestSyncer_Stop_DrainWorksWithoutCriticalError(t *testing.T) {
 		ctx, cancel := context.WithCancel(t.Context())
 		s.ctx = ctx
 		s.cancel = cancel
+		s.lifecycleState = syncerLifecycleStarted
 
 		// Build a valid height-1 event that will actually reach ExecuteTxs during drain
 		lastState := s.getLastState()