Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
racequite wants to merge 1 commit into
base: main
Choose a base branch
from
+101 −3
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 27 additions & 3 deletions block/internal/syncing/syncer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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()
Comment on lines +178 to +190

@coderabbitai coderabbitai Bot Aug 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🔴 Critical | 🏗️ Heavy lift

Serialize startup and shutdown completion.

At Line 189, Start publishes syncerLifecycleStarted before initialization completes. A concurrent Stop can cancel the context, observe an empty s.wg, and close s.heightInCh. Start can then start s.raftRetriever, create s.daFollower, or launch workers after shutdown. This can cause sends to a closed channel and leave workers outside the shutdown wait.

Stop also reads s.fiRetriever and s.daFollower while Start writes them without lifecycleMu. A second Stop returns at Line 264 before the first cleanup completes.

Add explicit starting and stopping states with a completion signal, or use an equivalent lifecycle gate. Do not allow startup after shutdown begins. Make concurrent Stop callers wait for the first cleanup. Add a concurrent lifecycle test and run it with -race.

As per coding guidelines, protect shared state and prevent goroutine leaks.

Also applies to: 261-271

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@block/internal/syncing/syncer.go` around lines 178 - 190, Serialize Start and
Stop through the syncer lifecycle gate: add explicit starting and stopping
states (or an equivalent completion signal), keep initialization and worker
registration under lifecycle protection, and publish started only after
initialization completes. Prevent Start once stopping or stopped, make
concurrent Stop callers wait for the first cleanup instead of returning early,
and protect fiRetriever and daFollower accesses with lifecycleMu; add a
concurrent lifecycle test covering shutdown races and run it with -race.

Source: Coding guidelines


defer func() { //nolint: contextcheck // use new context as parent can be cancelled already
if err != nil {
Expand Down Expand Up @@ -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 {
Expand Down
1 change: 1 addition & 0 deletions block/internal/syncing/syncer_benchmark_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
73 changes: 73 additions & 0 deletions block/internal/syncing/syncer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
Expand Down Expand Up @@ -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()))

Expand All @@ -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)
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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()
Expand Down