Skip to content

fix(node): race on caught up#3133

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

Merged
julienrbrt merged 1 commit into from
Mar 4, 2026
Merged

fix(node): race on caught up #3133

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
5 changes: 5 additions & 0 deletions block/internal/syncing/syncer.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -467,6 +467,11 @@ func (s *Syncer) fetchDAUntilCaughtUp(ctx context.Context) error {
}
}

// PendingCount returns the number of unprocessed height events in the pipeline.
func (s *Syncer) PendingCount() int {
return len(s.heightInCh)
}

func (s *Syncer) pendingWorkerLoop(ctx context.Context) {
defer s.wg.Done()

Expand Down
3 changes: 2 additions & 1 deletion node/failover.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -315,7 +315,8 @@ func (f *failoverState) waitForCatchup(ctx context.Context) (bool, error) {
p2pCaughtUp = true
}

if daCaughtUp && p2pCaughtUp {
pipelineDrained := f.bc.Syncer == nil || f.bc.Syncer.PendingCount() == 0
if daCaughtUp && p2pCaughtUp && pipelineDrained {
Comment on lines +318 to +319

@coderabbitaicoderabbitaiBotMar 4, 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.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

#!/bin/bashset -euo pipefail
echo"=== 1) PendingCount implementation ==="
rg -n -A5 'func \(s \*Syncer\) PendingCount\(\) int' block/internal/syncing/syncer.go
echoecho"=== 2) Syncer struct definition and state fields ==="
rg -n -B2 -A10 'type Syncer struct' block/internal/syncing/syncer.go
echoecho"=== 3) heightInCh and pipeline queue operations ==="
rg -n -B2 -A2 'heightInCh' block/internal/syncing/syncer.go | head -80
echoecho"=== 4) Failover.go catchup logic ==="
rg -n -B5 -A5 'pipelineDrained' node/failover.go

Repository: evstack/ev-node

Length of output: 3017

🏁 Script executed:

#!/bin/bashset -euo pipefail
echo"=== 1) processHeightEvent implementation ==="
rg -n -A20 'func \(s \*Syncer\) processHeightEvent' block/internal/syncing/syncer.go
echoecho"=== 2) Cache operations and pending events ==="
rg -n -B2 -A2 'cache\.GetNextPendingEvent\|SetPendingEvent\|pendingWorkerLoop' block/internal/syncing/syncer.go | head -100
echoecho"=== 3) Check if Syncer has any Done/Idle/Quiescent signals ==="
rg -n 'sync\.WaitGroup\|chan.*done\|chan.*idle\|chan.*quiescent' block/internal/syncing/syncer.go
echoecho"=== 4) Any existing waiters for pipeline quiescence ==="
rg -n 'WaitForIdle\|WaitForCompletion\|WaitForQuiescence' block/internal/syncing/syncer.go

Repository: evstack/ev-node

Length of output: 869

🏁 Script executed:

#!/bin/bashset -euo pipefail
echo"=== 1) Full processHeightEvent implementation ==="
rg -n -A50 'func \(s \*Syncer\) processHeightEvent' block/internal/syncing/syncer.go | head -80
echoecho"=== 2) pendingWorkerLoop - cache to channel pipeline ==="
rg -n -A30 'func \(s \*Syncer\) pendingWorkerLoop' block/internal/syncing/syncer.go
echoecho"=== 3) Check Syncer fields for tracking work state ==="
sed -n '46,150p' block/internal/syncing/syncer.go

Repository: evstack/ev-node

Length of output: 6599

🏁 Script executed:

#!/bin/bashset -euo pipefail
echo"=== 1) processPendingEvents to confirm cache polling ==="
rg -n -A15 'func \(s \*Syncer\) processPendingEvents' block/internal/syncing/syncer.go
echoecho"=== 2) Check cache.GetNextPendingEvent calls ==="
rg -n 'GetNextPendingEvent' block/internal/syncing/syncer.go
echoecho"=== 3) How failover uses/waits for syncer ==="
rg -n -B5 -A10 'bc\.Syncer.*PendingCount\|bc\.Stop\|Syncer.*idle\|Syncer.*done' node/failover.go | head -80

Repository: evstack/ev-node

Length of output: 874

🏁 Script executed:

#!/bin/bashset -euo pipefail
echo"=== 1) Failover.bc.Stop and syncer lifecycle ==="
rg -n -B10 -A10 'bc\.Stop\(\)' node/failover.go
echoecho"=== 2) Check if syncer has any Stop or WaitAll methods ==="
rg -n 'func \(s \*Syncer\) Stop\|func \(s \*Syncer\) Wait' block/internal/syncing/syncer.go
echoecho"=== 3) Full catchup logic context in failover ==="
sed -n '300,330p' node/failover.go
echoecho"=== 4) Verify cache.GetNextPendingEvent can be non-nil when heightInCh empty ==="
rg -n -B5 -A10 'cache.GetNextPendingEvent\|pendingEvents\|cache.*map' block/internal/syncing/syncer.go | head -60

Repository: evstack/ev-node

Length of output: 1013

🏁 Script executed:

#!/bin/bashset -euo pipefail
echo"=== 1) Syncer Stop method ==="
rg -n 'func \(s \*Syncer\) Stop' block/internal/syncing/syncer.go
echoecho"=== 2) All public Syncer methods ==="
rg -n '^func \(s \*Syncer\)' block/internal/syncing/syncer.go | head -30
echoecho"=== 3) Look at sync.WaitGroup usage in syncer ==="
rg -n 'wg\.Add\|wg\.Wait\|wg\.Done' block/internal/syncing/syncer.go
echoecho"=== 4) Full catchup decision logic in failover ==="
sed -n '300,330p' node/failover.go

Repository: evstack/ev-node

Length of output: 2363

🏁 Script executed:

#!/bin/bashset -euo pipefail
echo"=== 1) Syncer.Stop() method ==="
sed -n '209,250p' block/internal/syncing/syncer.go
echoecho"=== 2) Syncer.Start() - understand wg usage ==="
sed -n '166,210p' block/internal/syncing/syncer.go
echoecho"=== 3) Full failover catchup check context ==="
sed -n '295,330p' node/failover.go
echoecho"=== 4) Understanding cache state - pending events structure ==="
rg -n 'pendingEventsMu\|pendingEvents\s*map' block/internal/syncing/ -A2

Repository: evstack/ev-node

Length of output: 3638

Pipeline can complete catchup before pending cache events are processed.

PendingCount() only measures channel buffer length (line 471-472) and misses two important pipeline states:

  1. Pending cache events: When processHeightEvent() encounters an out-of-sequence block, it stores it in cache via SetPendingEvent() (line 599). The pendingWorkerLoop() polls this cache every 10ms (line 481) and re-queues events to heightInCh when the sequence is complete (line 1109-1111). These cache events don't appear in PendingCount().

  2. In-flight processing: Events dequeued from heightInCh (line 354) remain in-flight during processHeightEvent() execution, which performs I/O and state validation (lines 584-602). This work isn't tracked by channel depth.

The evidence: Stop() calls wg.Wait() (line 216) to wait for all worker goroutines, then after goroutine completion, drains heightInCh in a loop (lines 221-237). This pattern shows the channel can be empty while the pipeline hasn't reached true quiescence.

Line 318 in failover.go declares catchup complete as soon as PendingCount() == 0, which creates a race where pending cache events or in-flight processing aren't accounted for.

Implement a dedicated idle signal in Syncer that waits for all three stages: empty channel, no in-flight processing, and empty pending cache.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@node/failover.go` around lines 318 - 319, The current catchup check uses
Syncer.PendingCount() which only looks at channel buffer and misses pending
cache entries and in-flight processing; add a dedicated idle mechanism on Syncer
(e.g., WaitIdle() or IdleChan/IsIdle()) that atomically ensures: heightInCh is
empty, an in-flight counter (incremented/decremented in processHeightEvent and
when dequeuing from heightInCh) is zero, and the pending cache (populated by
SetPendingEvent and polled in pendingWorkerLoop) is empty; update Syncer to
maintain the in-flight counter and a condition or channel that is signaled when
all three are zero, expose a blocking WaitIdle() or boolean IsIdle() used in
failover.go (replace pipelineDrained := f.bc.Syncer == nil ||
f.bc.Syncer.PendingCount() == 0 with a call to the new idle API) so catchup
waits for true quiescence.

f.logger.Info().
Uint64("store_height", storeHeight).
Uint64("max_p2p_height", maxP2PHeight).
Expand Down
Loading