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

sync/statesync/code_syncer.go

@@ -47,8 +47,9 @@ 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 [CodeHashSink] into the state syncer.
+func NewCodeSyncer(client statesyncclient.Client, db ethdb.Database, config Config) (*codeSyncer, error) {
 	cfg := config.WithUnsetDefaults()
 
 	dbCodeHashes, err := getCodeToFetchFromDB(db)
@@ -217,11 +218,11 @@ 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.
+// AccountTrieCompleted implements CodeHashSink 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() {
+func (c *codeSyncer) AccountTrieCompleted() {
 	<-c.open // The code syncer must queue the previous code from the db first
 	close(c.codeHashes)
 }

sync/statesync/code_syncer_test.go

@@ -54,10 +54,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,7 +70,7 @@ func testCodeSyncer(t *testing.T, test codeSyncerTest) {
 				require.ErrorIs(t, err, test.err)
 			}
 		}
-		codeSyncer.notifyAccountTrieCompleted()
+		codeSyncer.AccountTrieCompleted()
 	}()
 
 	err = codeSyncer.Sync(context.Background())

sync/statesync/state_syncer.go

@@ -5,6 +5,7 @@ package statesync
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 
@@ -68,6 +69,21 @@ func (c Config) WithUnsetDefaults() Config {
 	return out
 }
 
+// CodeHashSink is a minimal interface for accepting discovered code hashes
+// and signaling when no more code hashes will be produced from the account trie.
+// Implemented by codeSyncer to support decoupled wiring.
+type CodeHashSink interface {
+	AddCode(codeHashes []common.Hash) error
+	AccountTrieCompleted()
+}
+
+// CodeSyncerTask is implemented by the concrete code syncer and combines
+// the sink and runnable task behaviors so callers can use a single type.
+type CodeSyncerTask interface {
+	CodeHashSink
+	synccommon.Syncer
+}
+
 // stateSync keeps the state of the entire state sync operation.
 type stateSync struct {
 	db        ethdb.Database            // database we are syncing
@@ -76,10 +92,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
+	codeSink  CodeHashSink                   // sink 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 +111,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, sink CodeHashSink, config Config) (synccommon.Syncer, error) {
 	cfg := config.WithUnsetDefaults()
 
 	ss := &stateSync{
@@ -125,12 +141,12 @@ 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 sink == nil {
+		return nil, errors.New("CodeHashSink must be provided")
 	}
+	ss.codeSink = sink
 
+	var err error
 	ss.trieQueue = NewTrieQueue(db)
 	if err := ss.trieQueue.clearIfRootDoesNotMatch(ss.root); err != nil {
 		return nil, err
@@ -171,7 +187,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.codeSink.AccountTrieCompleted()
 
 	// count the number of storage tries we need to sync for eta purposes.
 	numStorageTries, err := t.trieQueue.countTries()
@@ -255,7 +271,7 @@ func (t *stateSync) storageTrieProducer(ctx context.Context) error {
 }
 
 func (t *stateSync) Sync(ctx context.Context) error {
-	// Start the code syncer and leaf syncer.
+	// Start the leaf syncer and storage trie producer.
 	eg, egCtx := errgroup.WithContext(ctx)
 
 	eg.Go(func() error {
@@ -264,9 +280,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 registry if 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 sink.
+	return m.sync.codeSink.AddCode(codeHashes)
 }
 
 type storageTrieTask struct {

sync/synctest/synctest.go

@@ -0,0 +1,73 @@
+// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+// Package utilstest provides small testing helpers used across the codebase.
+// It includes lightweight syncer adapters and utilities for coordinating
+// goroutines under test.
+package synctest
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+)
+
+// FuncSyncer adapts a function to the simple Syncer shape used in tests. It is
+// useful for defining small, behavior-driven syncers inline.
+type FuncSyncer struct {
+	fn func(ctx context.Context) error
+}
+
+// Sync calls the wrapped function and returns its result.
+func (f FuncSyncer) Sync(ctx context.Context) error { return f.fn(ctx) }
+
+// NewBarrierSyncer returns a syncer that, upon entering Sync, calls wg.Done() to
+// signal it has started, then blocks until either:
+//   - `releaseCh` is closed, returning nil; or
+//   - `ctx` is canceled, returning ctx.Err.
+//
+// This acts as a barrier to coordinate test goroutines.
+func NewBarrierSyncer(wg *sync.WaitGroup, releaseCh <-chan struct{}) FuncSyncer {
+	return FuncSyncer{fn: func(ctx context.Context) error {
+		wg.Done()
+		select {
+		case <-releaseCh:
+			return nil
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}}
+}
+
+// NewErrorSyncer returns a syncer that waits until either `trigger` is closed
+// (then returns `errToReturn`) or `ctx` is canceled (then returns ctx.Err).
+func NewErrorSyncer(trigger <-chan struct{}, errToReturn error) FuncSyncer {
+	return FuncSyncer{fn: func(ctx context.Context) error {
+		select {
+		case <-trigger:
+			return errToReturn
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}}
+}
+
+// NewCancelAwareSyncer closes `started` as soon as Sync begins, then waits for
+// either:
+//   - `ctx` cancellation: closes `canceled` and returns ctx.Err; or
+//   - `timeout` elapsing: returns an error indicating a timeout.
+//
+// Useful for asserting that cancellation propagates to the syncer under test.
+func NewCancelAwareSyncer(started, canceled chan struct{}, timeout time.Duration) FuncSyncer {
+	return FuncSyncer{fn: func(ctx context.Context) error {
+		close(started)
+		select {
+		case <-ctx.Done():
+			close(canceled)
+			return ctx.Err()
+		case <-time.After(timeout):
+			return errors.New("syncer timed out waiting for cancellation")
+		}
+	}}
+}

sync/vm/client.go

@@ -37,6 +37,7 @@ const (
 	atomicStateSyncOperationName = "Atomic State Sync"
 	blockSyncOperationName       = "Block Sync"
 	evmStateSyncOperationName    = "EVM State Sync"
+	codeHashSyncOperationName    = "Code Hash Sync"
 )
 
 var stateSyncSummaryKey = []byte("stateSyncSummary")
@@ -168,34 +169,45 @@ func (client *client) stateSync(ctx context.Context) error {
 }
 
 func (client *client) registerSyncers(ctx context.Context, registry *SyncerRegistry) error {
-	// Register block syncer.
-	syncer, err := client.createBlockSyncer(client.summary.GetBlockHash(), client.summary.Height())
+	blockSyncer, err := client.createBlockSyncer(client.summary.GetBlockHash(), client.summary.Height())
 	if err != nil {
 		return fmt.Errorf("failed to create block syncer: %w", err)
 	}
-	if err := registry.Register(blockSyncOperationName, syncer); err != nil {
-		return fmt.Errorf("failed to register block syncer: %w", err)
-	}
 
-	// Register EVM state syncer.
-	evmSyncer, err := client.createEVMSyncer()
+	cfg := statesync.NewDefaultConfig(client.RequestSize)
+
+	codeSyncer, err := client.createCodeHashSyncer(cfg)
 	if err != nil {
-		return fmt.Errorf("failed to create EVM syncer: %w", err)
+		return fmt.Errorf("failed to create code syncer: %w", err)
 	}
 
-	if err := registry.Register(evmStateSyncOperationName, evmSyncer); err != nil {
-		return fmt.Errorf("failed to register EVM syncer: %w", err)
+	stateSyncer, err := client.createEVMSyncer(cfg, codeSyncer)
+	if err != nil {
+		return fmt.Errorf("failed to create EVM state syncer: %w", err)
 	}
 
-	// Register atomic syncer.
+	var atomicSyncer synccommon.Syncer
 	if client.Extender != nil {
-		atomicSyncer, err := client.createAtomicSyncer(ctx)
+		atomicSyncer, err = client.createAtomicSyncer(ctx)
 		if err != nil {
 			return fmt.Errorf("failed to create atomic syncer: %w", err)
 		}
+	}
+
+	// Register in a deterministic order
+	if err := registry.Register(blockSyncOperationName, blockSyncer); err != nil {
+		return fmt.Errorf("failed to register %s: %w", blockSyncOperationName, err)
+	}
+	if err := registry.Register(codeHashSyncOperationName, codeSyncer); err != nil {
+		return fmt.Errorf("failed to register %s: %w", codeHashSyncOperationName, err)
+	}
+	if err := registry.Register(evmStateSyncOperationName, stateSyncer); err != nil {
+		return fmt.Errorf("failed to register %s: %w", evmStateSyncOperationName, err)
+	}
 
+	if atomicSyncer != nil {
 		if err := registry.Register(atomicStateSyncOperationName, atomicSyncer); err != nil {
-			return fmt.Errorf("failed to register atomic syncer: %w", err)
+			return fmt.Errorf("failed to register %s: %w", atomicStateSyncOperationName, err)
 		}
 	}
 
@@ -210,8 +222,12 @@ func (client *client) createBlockSyncer(fromHash common.Hash, fromHeight uint64)
 	})
 }
 
-func (client *client) createEVMSyncer() (synccommon.Syncer, error) {
-	return statesync.NewSyncer(client.Client, client.ChainDB, client.summary.GetBlockRoot(), statesync.NewDefaultConfig(client.RequestSize))
+func (client *client) createEVMSyncer(cfg statesync.Config, sink statesync.CodeHashSink) (synccommon.Syncer, error) {
+	return statesync.NewSyncer(client.Client, client.ChainDB, client.summary.GetBlockRoot(), sink, cfg)
+}
+
+func (client *client) createCodeHashSyncer(cfg statesync.Config) (statesync.CodeSyncerTask, error) {
+	return statesync.NewCodeSyncer(client.Client, client.ChainDB, cfg)
 }
 
 func (client *client) createAtomicSyncer(ctx context.Context) (synccommon.Syncer, error) {

sync/vm/registry.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 
 	"github.com/ava-labs/libevm/log"
+	"golang.org/x/sync/errgroup"
 
 	synccommon "github.com/ava-labs/coreth/sync"
 )
@@ -50,15 +51,26 @@ func (r *SyncerRegistry) RunSyncerTasks(ctx context.Context, client *client) err
 		return nil
 	}
 
+	g, ctx := errgroup.WithContext(ctx)
+
 	for _, task := range r.syncers {
-		log.Info("starting syncer", "name", task.name, "summary", client.summary)
+		g.Go(func() error {
+			log.Info("starting syncer", "name", task.name, "summary", client.summary)
+			if err := task.syncer.Sync(ctx); err != nil {
+				log.Error("failed syncing", "name", task.name, "summary", client.summary, "err", err)
+				return fmt.Errorf("%s failed: %w", task.name, err)
+			}
+			log.Info("completed successfully", "name", task.name, "summary", client.summary)
+
+			return nil
+		})
+	}
 
-		if err := task.syncer.Sync(ctx); err != nil {
-			log.Error("failed to complete", "name", task.name, "summary", client.summary, "err", err)
-			return fmt.Errorf("%s failed: %w", task.name, err)
-		}
-		log.Info("completed successfully", "name", task.name, "summary", client.summary)
+	if err := g.Wait(); err != nil {
+		return err
 	}
 
+	log.Info("all syncers completed successfully", "count", len(r.syncers), "summary", client.summary)
+
 	return nil
 }