From dfb0889955e80534b5b28a7acacc7c75416fa7ec Mon Sep 17 00:00:00 2001 From: racequite Date: Sun, 2 Aug 2026 01:50:10 +0800 Subject: [PATCH] fix(syncing): make Syncer shutdown idempotent after start failure --- block/internal/syncing/syncer.go | 30 +++++++- .../internal/syncing/syncer_benchmark_test.go | 1 + block/internal/syncing/syncer_test.go | 73 +++++++++++++++++++ 3 files changed, 101 insertions(+), 3 deletions(-) diff --git a/block/internal/syncing/syncer.go b/block/internal/syncing/syncer.go index aba3d8e44a..7352ae002e 100644 --- a/block/internal/syncing/syncer.go +++ b/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 { diff --git a/block/internal/syncing/syncer_benchmark_test.go b/block/internal/syncing/syncer_benchmark_test.go index b066f2ec71..e332c7e416 100644 --- a/block/internal/syncing/syncer_benchmark_test.go +++ b/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) diff --git a/block/internal/syncing/syncer_test.go b/block/internal/syncing/syncer_test.go index e78a8771c6..82f03c616e 100644 --- a/block/internal/syncing/syncer_test.go +++ b/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()