feat(sync): decouple code syncer from state syncer and run concurrently

scripts/lint.sh

@@ -148,9 +148,17 @@ function test_require_error_is_no_funcs_as_params {
 }
 
 function test_require_no_error_inline_func {
-  if grep -R -zo -P '\t+err :?= ((?!require|if).|\n)*require\.NoError\((t, )?err\)' "${DEFAULT_FILES[@]}"; then
+  # Flag only when a single variable whose name contains "err" or "Err"
+  # (e.g., err, myErr, parseError) is assigned from a call (:= or =), and later
+  # that same variable is passed to require.NoError(...). We explicitly require
+  # no commas on the LHS to avoid flagging multi-return assignments like
+  # "val, err := f()" or "err, val := f()".
+  #
+  # Capture the variable name and enforce it matches in the subsequent require.NoError.
+  local -r pattern='^\s*([A-Za-z_][A-Za-z0-9_]*[eE]rr[A-Za-z0-9_]*)\s*:?=\s*[^,\n]*\([^)]*\).*\n(?:(?!^\s*(?:if|require)).*\n)*^\s*require\.NoError\((?:t,\s*)?\1\)'
+  if grep -R -zo -P "$pattern" "${DEFAULT_FILES[@]}"; then
     echo ""
-    echo "Checking that a function with a single error return doesn't error should be done in-line."
+    echo "Checking that a function with a single error return doesn't error should be done in-line (single LHS var containing 'err')."
     echo ""
     return 1
   fi

sync/statesync/code_syncer.go

@@ -28,14 +28,14 @@ const (
 )
 
 var (
-	_ synccommon.Syncer = (*codeSyncer)(nil)
+	_ synccommon.Syncer = (*CodeSyncer)(nil)
 
 	errFailedToAddCodeHashesToQueue = errors.New("failed to add code hashes to queue")
 )
 
-// codeSyncer syncs code bytes from the network in a seprate thread.
+// CodeSyncer syncs code bytes from the network in a seprate thread.
 // Tracks outstanding requests in the DB, so that it will still fulfill them if interrupted.
-type codeSyncer struct {
+type CodeSyncer struct {
 	db                    ethdb.Database
 	client                statesyncclient.Client
 	config                Config
@@ -47,16 +47,17 @@ type codeSyncer struct {
 	done                  <-chan struct{}
 }
 
-// newCodeSyncer returns a code syncer that will sync code bytes from the network in a separate thread.
-func newCodeSyncer(client statesyncclient.Client, db ethdb.Database, config Config) (*codeSyncer, error) {
+// NewCodeSyncer allows external packages (e.g., registry wiring) to create a code syncer as a separate task and
+// inject it as a [CodeFetcher] into the state syncer.
+func NewCodeSyncer(client statesyncclient.Client, db ethdb.Database, config Config) (*CodeSyncer, error) {
 	cfg := config.WithUnsetDefaults()
 
 	dbCodeHashes, err := getCodeToFetchFromDB(db)
 	if err != nil {
 		return nil, fmt.Errorf("failed to add code hashes to queue: %w", err)
 	}
 
-	return &codeSyncer{
+	return &CodeSyncer{
 		db:           db,
 		client:       client,
 		config:       cfg,
@@ -66,10 +67,10 @@ func newCodeSyncer(client statesyncclient.Client, db ethdb.Database, config Conf
 	}, nil
 }
 
-// Start the worker thread and populate the code hashes queue with active work.
+// Sync starts the worker thread and populates the code hashes queue with active work.
 // Blocks until all outstanding code requests from a previous sync have been
 // fetched and the code channel has been closed, or the context is cancelled.
-func (c *codeSyncer) Sync(ctx context.Context) error {
+func (c *CodeSyncer) Sync(ctx context.Context) error {
 	eg, egCtx := errgroup.WithContext(ctx)
 	c.done = egCtx.Done()
 
@@ -80,7 +81,7 @@ func (c *codeSyncer) Sync(ctx context.Context) error {
 
 	// Queue the code hashes from the previous sync
 	if err := c.addCode(c.dbCodeHashes); err != nil {
-		return fmt.Errorf("Unable to resume previous sync: %w", err)
+		return fmt.Errorf("unable to resume previous sync: %w", err)
 	}
 	c.dbCodeHashes = nil
 	close(c.open)
@@ -128,7 +129,7 @@ func getCodeToFetchFromDB(db ethdb.Database) ([]common.Hash, error) {
 
 // work fulfills any incoming requests from the producer channel by fetching code bytes from the network
 // and fulfilling them by updating the database.
-func (c *codeSyncer) work(ctx context.Context) error {
+func (c *CodeSyncer) work(ctx context.Context) error {
 	codeHashes := make([]common.Hash, 0, message.MaxCodeHashesPerRequest)
 
 	for {
@@ -165,7 +166,7 @@ func (c *codeSyncer) work(ctx context.Context) error {
 // marks the work as complete.
 // codeHashes should not be empty or contain duplicate hashes.
 // Returns an error if one is encountered, signaling the worker thread to terminate.
-func (c *codeSyncer) fulfillCodeRequest(ctx context.Context, codeHashes []common.Hash) error {
+func (c *CodeSyncer) fulfillCodeRequest(ctx context.Context, codeHashes []common.Hash) error {
 	codeByteSlices, err := c.client.GetCode(ctx, codeHashes)
 	if err != nil {
 		return err
@@ -190,12 +191,12 @@ func (c *codeSyncer) fulfillCodeRequest(ctx context.Context, codeHashes []common
 // AddCode checks if [codeHashes] need to be fetched from the network and adds them to the queue if so.
 // assumes that [codeHashes] are valid non-empty code hashes.
 // This blocks until the code syncer is open and ready to accept requests.
-func (c *codeSyncer) AddCode(codeHashes []common.Hash) error {
+func (c *CodeSyncer) AddCode(codeHashes []common.Hash) error {
 	<-c.open
 	return c.addCode(codeHashes)
 }
 
-func (c *codeSyncer) addCode(codeHashes []common.Hash) error {
+func (c *CodeSyncer) addCode(codeHashes []common.Hash) error {
 	batch := c.db.NewBatch()
 
 	c.lock.Lock()
@@ -217,18 +218,21 @@ func (c *codeSyncer) addCode(codeHashes []common.Hash) error {
 	return c.addHashesToQueue(selectedCodeHashes)
 }
 
-// notifyAccountTrieCompleted notifies the code syncer that there will be no more incoming
-// code hashes from syncing the account trie, so it only needs to compelete its outstanding
-// work.
+// Finalize implements CodeFetcher by signaling no further code hashes will be added.
+// Notifies the code syncer that there will be no more incoming code hashes from syncing the account trie,
+// so it only needs to complete its outstanding work.
 // Note: this allows the worker threads to exit and return a nil error.
-func (c *codeSyncer) notifyAccountTrieCompleted() {
-	<-c.open // The code syncer must queue the previous code from the db first
+func (c *CodeSyncer) Finalize() {
+	<-c.open // The code syncer must queue the previous code from the db first.
 	close(c.codeHashes)
 }
 
+// Ready returns a channel that is closed when the code syncer is ready to accept code hashes.
+func (c *CodeSyncer) Ready() <-chan struct{} { return c.open }
+
 // addHashesToQueue adds [codeHashes] to the queue and blocks until it is able to do so.
 // This should be called after all other operation to add code hashes to the queue has been completed.
-func (c *codeSyncer) addHashesToQueue(codeHashes []common.Hash) error {
+func (c *CodeSyncer) addHashesToQueue(codeHashes []common.Hash) error {
 	for _, codeHash := range codeHashes {
 		select {
 		case c.codeHashes <- codeHash:

sync/statesync/code_syncer_test.go

@@ -4,9 +4,9 @@
 package statesync
 
 import (
-	"context"
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/ava-labs/avalanchego/utils"
 	"github.com/ava-labs/libevm/common"
@@ -19,14 +19,15 @@ import (
 	"github.com/ava-labs/coreth/plugin/evm/customrawdb"
 	"github.com/ava-labs/coreth/plugin/evm/message"
 	"github.com/ava-labs/coreth/sync/handlers"
+	"github.com/ava-labs/coreth/utils/utilstest"
 
 	statesyncclient "github.com/ava-labs/coreth/sync/client"
 	handlerstats "github.com/ava-labs/coreth/sync/handlers/stats"
 )
 
 type codeSyncerTest struct {
 	clientDB          ethdb.Database
-	setupCodeSyncer   func(*codeSyncer)
+	setupCodeSyncer   func(*CodeSyncer)
 	codeRequestHashes [][]common.Hash
 	codeByteSlices    [][]byte
 	getCodeIntercept  func(hashes []common.Hash, codeBytes [][]byte) ([][]byte, error)
@@ -54,10 +55,11 @@ func testCodeSyncer(t *testing.T, test codeSyncerTest) {
 		clientDB = rawdb.NewMemoryDatabase()
 	}
 
-	codeSyncer, err := newCodeSyncer(
+	cfg := NewDefaultConfig(testRequestSize)
+	codeSyncer, err := NewCodeSyncer(
 		mockClient,
 		clientDB,
-		NewDefaultConfig(testRequestSize),
+		cfg,
 	)
 	require.NoError(t, err)
 	if test.setupCodeSyncer != nil {
@@ -69,10 +71,14 @@ func testCodeSyncer(t *testing.T, test codeSyncerTest) {
 				require.ErrorIs(t, err, test.err)
 			}
 		}
-		codeSyncer.notifyAccountTrieCompleted()
+		codeSyncer.Finalize()
 	}()
 
-	err = codeSyncer.Sync(context.Background())
+	ctx, cancel := utilstest.NewTestContext(t)
+	defer cancel()
+
+	// Run the sync and handle expected error.
+	err = codeSyncer.Sync(ctx)
 	require.ErrorIs(t, err, test.err)
 	if err != nil {
 		return // don't check the state
@@ -106,7 +112,7 @@ func TestCodeSyncerManyCodeHashes(t *testing.T) {
 	}
 
 	testCodeSyncer(t, codeSyncerTest{
-		setupCodeSyncer: func(c *codeSyncer) {
+		setupCodeSyncer: func(c *CodeSyncer) {
 			c.codeHashes = make(chan common.Hash, 10)
 		},
 		codeRequestHashes: [][]common.Hash{codeHashes[0:100], codeHashes[100:2000], codeHashes[2000:2005], codeHashes[2005:]},
@@ -162,3 +168,39 @@ func TestCodeSyncerAddsMoreInProgressThanQueueSize(t *testing.T) {
 		codeByteSlices:    codeByteSlices,
 	})
 }
+
+func TestCodeSyncerReady(t *testing.T) {
+	serverDB := memorydb.New()
+	codeRequestHandler := handlers.NewCodeRequestHandler(serverDB, message.Codec, handlerstats.NewNoopHandlerStats())
+	mockClient := statesyncclient.NewTestClient(message.Codec, nil, codeRequestHandler, nil)
+
+	clientDB := rawdb.NewMemoryDatabase()
+	cfg := NewDefaultConfig(testRequestSize)
+	codeSyncer, err := NewCodeSyncer(mockClient, clientDB, cfg)
+	require.NoError(t, err)
+
+	select {
+	case <-codeSyncer.Ready():
+		// If already closed, this would mean ready before Sync which shouldn't happen.
+		t.Fatalf("Ready should not be closed before Sync starts")
+	default:
+	}
+
+	ctx, cancel := utilstest.NewTestContext(t)
+
+	doneCh := make(chan error, 1)
+	go func() { doneCh <- codeSyncer.Sync(ctx) }()
+
+	// Wait briefly for Sync to initialize and close the ready channel.
+	utilstest.SleepWithContext(ctx, 50*time.Millisecond)
+
+	select {
+	case <-codeSyncer.Ready():
+		// ok
+	default:
+		t.Fatalf("Ready should be closed after Sync initialization")
+	}
+
+	cancel()
+	require.Error(t, <-doneCh)
+}

sync/statesync/state_syncer.go

@@ -5,6 +5,7 @@ package statesync
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 
@@ -27,7 +28,10 @@ const (
 	defaultNumWorkers      = 8
 )
 
-var _ synccommon.Syncer = (*stateSync)(nil)
+var (
+	_                      synccommon.Syncer = (*stateSync)(nil)
+	errCodeFetcherRequired                   = errors.New("code fetcher is required")
+)
 
 type Config struct {
 	BatchSize uint
@@ -76,10 +80,10 @@ type stateSync struct {
 	snapshot  snapshot.SnapshotIterable // used to access the database we are syncing as a snapshot.
 	batchSize uint                      // write batches when they reach this size
 
-	segments   chan syncclient.LeafSyncTask   // channel of tasks to sync
-	syncer     *syncclient.CallbackLeafSyncer // performs the sync, looping over each task's range and invoking specified callbacks
-	codeSyncer *codeSyncer                    // manages the asynchronous download and batching of code hashes
-	trieQueue  *trieQueue                     // manages a persistent list of storage tries we need to sync and any segments that are created for them
+	segments    chan syncclient.LeafSyncTask   // channel of tasks to sync
+	syncer      *syncclient.CallbackLeafSyncer // performs the sync, looping over each task's range and invoking specified callbacks
+	codeFetcher synccommon.CodeFetcher         // fetcher that manages the asynchronous download and batching of code hashes
+	trieQueue   *trieQueue                     // manages a persistent list of storage tries we need to sync and any segments that are created for them
 
 	// track the main account trie specifically to commit its root at the end of the operation
 	mainTrie *trieToSync
@@ -95,7 +99,7 @@ type stateSync struct {
 	stats              *trieSyncStats
 }
 
-func NewSyncer(client syncclient.Client, db ethdb.Database, root common.Hash, config Config) (synccommon.Syncer, error) {
+func NewSyncer(client syncclient.Client, db ethdb.Database, root common.Hash, fetcher synccommon.CodeFetcher, config Config) (synccommon.Syncer, error) {
 	cfg := config.WithUnsetDefaults()
 
 	ss := &stateSync{
@@ -125,12 +129,13 @@ func NewSyncer(client syncclient.Client, db ethdb.Database, root common.Hash, co
 		OnFailure:   ss.onSyncFailure,
 	})
 
-	var err error
-	ss.codeSyncer, err = newCodeSyncer(client, db, cfg)
-	if err != nil {
-		return nil, err
+	if fetcher == nil {
+		return nil, errCodeFetcherRequired
 	}
 
+	ss.codeFetcher = fetcher
+
+	var err error
 	ss.trieQueue = NewTrieQueue(db)
 	if err := ss.trieQueue.clearIfRootDoesNotMatch(ss.root); err != nil {
 		return nil, err
@@ -171,7 +176,7 @@ func (t *stateSync) onStorageTrieFinished(root common.Hash) error {
 
 // onMainTrieFinished is called after the main trie finishes syncing.
 func (t *stateSync) onMainTrieFinished() error {
-	t.codeSyncer.notifyAccountTrieCompleted()
+	t.codeFetcher.Finalize()
 
 	// count the number of storage tries we need to sync for eta purposes.
 	numStorageTries, err := t.trieQueue.countTries()
@@ -255,7 +260,14 @@ func (t *stateSync) storageTrieProducer(ctx context.Context) error {
 }
 
 func (t *stateSync) Sync(ctx context.Context) error {
-	// Start the code syncer and leaf syncer.
+	// Wait for the code fetcher to be ready before starting.
+	select {
+	case <-t.codeFetcher.Ready():
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	// Start the leaf syncer and storage trie producer.
 	eg, egCtx := errgroup.WithContext(ctx)
 
 	eg.Go(func() error {
@@ -264,9 +276,7 @@ func (t *stateSync) Sync(ctx context.Context) error {
 		}
 		return t.onSyncComplete()
 	})
-	eg.Go(func() error {
-		return t.codeSyncer.Sync(egCtx)
-	})
+	// Note: code syncer is no longer started here. It should be run by the [SyncerRegistry] and registered separately.
 	eg.Go(func() error {
 		return t.storageTrieProducer(egCtx)
 	})

sync/statesync/sync_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/ava-labs/libevm/trie"
 	"github.com/ava-labs/libevm/triedb"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/ava-labs/coreth/core/state/snapshot"
 	"github.com/ava-labs/coreth/plugin/evm/customrawdb"
@@ -57,13 +58,21 @@ func testSync(t *testing.T, test syncTest) {
 	mockClient.GetLeafsIntercept = test.GetLeafsIntercept
 	mockClient.GetCodeIntercept = test.GetCodeIntercept
 
-	s, err := NewSyncer(mockClient, clientDB, root, Config{
+	cfg := Config{
 		BatchSize:   1000, // Use a lower batch size in order to get test coverage of batches being written early.
 		RequestSize: testRequestSize,
-	})
+	}
+	codeSyncer, err := NewCodeSyncer(mockClient, clientDB, cfg)
+	require.NoError(t, err, "failed to create code syncer")
+	stateSyncer, err := NewSyncer(mockClient, clientDB, root, codeSyncer, cfg)
 	require.NoError(t, err, "failed to create state syncer")
 
-	err = s.Sync(ctx)
+	// Run both syncers concurrently and wait for the first error.
+	eg, egCtx := errgroup.WithContext(ctx)
+	eg.Go(func() error { return codeSyncer.Sync(egCtx) })
+	eg.Go(func() error { return stateSyncer.Sync(egCtx) })
+
+	err = eg.Wait()
 	require.ErrorIs(t, err, test.expectedError, "unexpected error during sync")
 
 	// Only assert database consistency if the sync was expected to succeed.

sync/statesync/trie_sync_tasks.go

@@ -85,8 +85,8 @@ func (m *mainTrieTask) OnLeafs(db ethdb.KeyValueWriter, keys, vals [][]byte) err
 			codeHashes = append(codeHashes, codeHash)
 		}
 	}
-	// Add collected code hashes to the code syncer.
-	return m.sync.codeSyncer.AddCode(codeHashes)
+	// Add collected code hashes to the code fetcher.
+	return m.sync.codeFetcher.AddCode(codeHashes)
 }
 
 type storageTrieTask struct {