Skip to content

go/worker/storage: Refactor state sync worker #6299

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

Draft
martintomazic wants to merge 11 commits into
base: master
Choose a base branch
from
Draft

go/worker/storage: Refactor state sync worker #6299

Changes from 1 commit
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
0b3d6e2
go/worker/storage: Rename committee package to statesync
martintomazic Aug 2, 2025
10b4705
go/worker/storage/statesync: Move pruning to separate file
martintomazic Aug 2, 2025
8bbc085
go/worker/storage/statesync: Move checkpointert to separate file
martintomazic Aug 2, 2025
b030d02
go/worker/storage/statesync: Pass context explicitly
martintomazic Aug 2, 2025
a4e9069
go/worker/storage/statesync: Remove redundant context
martintomazic Aug 19, 2025
a27b967
go/worker/storage/statesync: Do not panic
martintomazic Aug 3, 2025
8dff06f
go/worker/storage/statesync: Move syncing methods at the bottom
martintomazic Aug 3, 2025
213b17f
go/worker/storage/statesync: Refactor the code
martintomazic Aug 3, 2025
1bb8edb
go/worker/storage/statesync: Move diff sync to separate file
martintomazic Aug 6, 2025
67b006e
go/worker/storage/statesync: Prevent deadlock when terminating
martintomazic Aug 19, 2025
dca1f4b
go/worker/storage/statesync: Remove redundant waitgroup
martintomazic Aug 19, 2025
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
8 changes: 5 additions & 3 deletions go/worker/storage/statesync/diff_sync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,8 @@ func (w *Worker) syncDiffs(
fetchPool := workerpool.New("storage_fetch/" + w.commonNode.Runtime.ID().String())
fetchPool.Resize(config.GlobalConfig.Storage.FetcherCount)
defer fetchPool.Stop()
fetchCtx, cancel := context.WithCancel(ctx)

@peternose peternose Aug 21, 2025

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

Shouldn't we just do this on the top ctx, cancel := context.WithCancel(ctx)?

defer cancel()

heartbeat := heartbeat{}
heartbeat.reset()
Expand Down Expand Up @@ -179,7 +181,7 @@ func (w *Worker) syncDiffs(
return fmt.Errorf("failed to fetch missing block headers: %w", err) // Suggestion: databases can fail, consider retrying.
}

w.triggerRoundFetches(ctx, &wg, fetchPool, diffCh, syncingRounds, summaryCache, lastFullyAppliedRound+1, latestBlockRound)
w.triggerRoundFetches(fetchCtx, &wg, fetchPool, diffCh, syncingRounds, summaryCache, lastFullyAppliedRound+1, latestBlockRound)
case item := <-diffCh:
if item.err != nil {
w.logger.Error("error calling getdiff",
Expand All @@ -197,12 +199,12 @@ func (w *Worker) syncDiffs(
// Item was successfully processed, trigger more round fetches.
// This ensures that new rounds are processed as fast as possible
// when we're syncing and are far behind.
w.triggerRoundFetches(ctx, &wg, fetchPool, diffCh, syncingRounds, summaryCache, lastFullyAppliedRound+1, latestBlockRound)
w.triggerRoundFetches(fetchCtx, &wg, fetchPool, diffCh, syncingRounds, summaryCache, lastFullyAppliedRound+1, latestBlockRound)
heartbeat.reset()
case <-heartbeat.C:
if latestBlockRound != w.undefinedRound {
w.logger.Debug("heartbeat", "in_flight_rounds", len(syncingRounds))
w.triggerRoundFetches(ctx, &wg, fetchPool, diffCh, syncingRounds, summaryCache, lastFullyAppliedRound+1, latestBlockRound)
w.triggerRoundFetches(fetchCtx, &wg, fetchPool, diffCh, syncingRounds, summaryCache, lastFullyAppliedRound+1, latestBlockRound)
}
case finalized := <-finalizedCh:
if finalized.err != nil {
Expand Down