fix(vmsync,statesync): ensure syncers detect cancellation during graceful shutdown and fix registry cancellation logging (#1411)

Author: powerslider

plugin/evm/atomic/sync/syncer.go

@@ -156,7 +156,11 @@ func addZeroes(height uint64) []byte {
 }
 
 // onLeafs is the callback for the leaf syncer, which will insert the key-value pairs into the trie.
-func (s *Syncer) onLeafs(keys [][]byte, values [][]byte) error {
+func (s *Syncer) onLeafs(ctx context.Context, keys [][]byte, values [][]byte) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+
 	for i, key := range keys {
 		if len(key) != atomicstate.TrieKeyLength {
 			return fmt.Errorf("unexpected key len (%d) in atomic trie sync", len(key))
@@ -239,6 +243,6 @@ func (a *syncerLeafTask) OnFinish(context.Context) error { return a.syncer.onFin
 func (*syncerLeafTask) OnStart() (bool, error)           { return false, nil }
 func (a *syncerLeafTask) Root() common.Hash              { return a.syncer.targetRoot }
 func (*syncerLeafTask) Account() common.Hash             { return common.Hash{} }
-func (a *syncerLeafTask) OnLeafs(keys, vals [][]byte) error {
-	return a.syncer.onLeafs(keys, vals)
+func (a *syncerLeafTask) OnLeafs(ctx context.Context, keys, vals [][]byte) error {
+	return a.syncer.onLeafs(ctx, keys, vals)
 }

plugin/evm/vmsync/doubles_test.go

@@ -27,15 +27,13 @@ func (FuncSyncer) ID() string   { return "test_id" }
 
 var _ syncpkg.Syncer = FuncSyncer{}
 
-// 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 {
+// NewBarrierSyncer returns a syncer that signals startedWG.Done() when Sync begins,
+// then blocks until releaseCh is closed (returns nil) or ctx is canceled (returns ctx.Err).
+func NewBarrierSyncer(startedWG *sync.WaitGroup, releaseCh <-chan struct{}) FuncSyncer {
 	return FuncSyncer{fn: func(ctx context.Context) error {
-		wg.Done()
+		if startedWG != nil {
+			startedWG.Done()
+		}
 		select {
 		case <-releaseCh:
 			return nil
@@ -45,10 +43,13 @@ func NewBarrierSyncer(wg *sync.WaitGroup, releaseCh <-chan struct{}) FuncSyncer
 	}}
 }
 
-// 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 {
+// NewErrorSyncer returns a syncer that signals startedWG.Done() when Sync begins,
+// then blocks until trigger is closed (returns errToReturn) or ctx is canceled (returns ctx.Err).
+func NewErrorSyncer(startedWG *sync.WaitGroup, trigger <-chan struct{}, errToReturn error) FuncSyncer {
 	return FuncSyncer{fn: func(ctx context.Context) error {
+		if startedWG != nil {
+			startedWG.Done()
+		}
 		select {
 		case <-trigger:
 			return errToReturn
@@ -58,18 +59,15 @@ func NewErrorSyncer(trigger <-chan struct{}, errToReturn error) FuncSyncer {
 	}}
 }
 
-// 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 {
+// NewCancelAwareSyncer returns a syncer that signals startedWG.Done() when Sync begins,
+// then blocks until ctx is canceled (returns ctx.Err) or timeout elapses (returns timeout error).
+func NewCancelAwareSyncer(startedWG *sync.WaitGroup, timeout time.Duration) FuncSyncer {
 	return FuncSyncer{fn: func(ctx context.Context) error {
-		close(started)
+		if startedWG != nil {
+			startedWG.Done()
+		}
 		select {
 		case <-ctx.Done():
-			close(canceled)
 			return ctx.Err()
 		case <-time.After(timeout):
 			return errors.New("syncer timed out waiting for cancellation")

plugin/evm/vmsync/registry.go

@@ -13,15 +13,15 @@ import (
 
 	"github.com/ava-labs/coreth/plugin/evm/message"
 
-	synccommon "github.com/ava-labs/coreth/sync"
+	syncpkg "github.com/ava-labs/coreth/sync"
 )
 
 var errSyncerAlreadyRegistered = errors.New("syncer already registered")
 
 // SyncerTask represents a single syncer with its name for identification.
 type SyncerTask struct {
 	name   string
-	syncer synccommon.Syncer
+	syncer syncpkg.Syncer
 }
 
 // SyncerRegistry manages a collection of syncers for sequential execution.
@@ -39,7 +39,7 @@ func NewSyncerRegistry() *SyncerRegistry {
 
 // Register adds a syncer to the registry.
 // Returns an error if a syncer with the same name is already registered.
-func (r *SyncerRegistry) Register(syncer synccommon.Syncer) error {
+func (r *SyncerRegistry) Register(syncer syncpkg.Syncer) error {
 	id := syncer.ID()
 	if r.registeredNames[id] {
 		return fmt.Errorf("%w with id '%s'", errSyncerAlreadyRegistered, id)
@@ -51,22 +51,46 @@ func (r *SyncerRegistry) Register(syncer synccommon.Syncer) error {
 	return nil
 }
 
-// RunSyncerTasks executes all registered syncers.
-// The provided summary is used only for logging to decouple from concrete client types.
+// RunSyncerTasks executes all registered syncers synchronously.
 func (r *SyncerRegistry) RunSyncerTasks(ctx context.Context, summary message.Syncable) error {
+	// Early return if context is already canceled (e.g., during shutdown).
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+
+	g := r.StartAsync(ctx, summary)
+
+	if err := g.Wait(); err != nil {
+		return err
+	}
+
+	log.Info("all syncers completed successfully", "count", len(r.syncers), "summary", summary.GetBlockHash().Hex())
+
+	return nil
+}
+
+// StartAsync launches all registered syncers and returns an [errgroup.Group]
+// whose Wait() completes when all syncers exit. The context returned will be
+// cancelled when any syncer fails, propagating shutdown to the others.
+func (r *SyncerRegistry) StartAsync(ctx context.Context, summary message.Syncable) *errgroup.Group {
+	g, egCtx := errgroup.WithContext(ctx)
+
 	if len(r.syncers) == 0 {
-		return nil
+		return g
 	}
 
 	summaryBlockHashHex := summary.GetBlockHash().Hex()
 	blockHeight := summary.Height()
 
-	g, ctx := errgroup.WithContext(ctx)
-
 	for _, task := range r.syncers {
 		g.Go(func() error {
 			log.Info("starting syncer", "name", task.name, "summary", summaryBlockHashHex, "height", blockHeight)
-			if err := task.syncer.Sync(ctx); err != nil {
+			if err := task.syncer.Sync(egCtx); err != nil {
+				// Context cancellation during shutdown is expected.
+				if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+					log.Info("syncer cancelled", "name", task.name, "summary", summaryBlockHashHex, "height", blockHeight)
+					return err
+				}
 				log.Error("failed syncing", "name", task.name, "summary", summaryBlockHashHex, "height", blockHeight, "err", err)
 				return fmt.Errorf("%s failed: %w", task.name, err)
 			}
@@ -76,11 +100,5 @@ func (r *SyncerRegistry) RunSyncerTasks(ctx context.Context, summary message.Syn
 		})
 	}
 
-	if err := g.Wait(); err != nil {
-		return err
-	}
-
-	log.Info("all syncers completed successfully", "count", len(r.syncers), "summary", summaryBlockHashHex)
-
-	return nil
+	return g
 }