feat(statesync): introduce Finalizer interface for syncer cleanup

Add a `Finalizer` interface to provide explicit cleanup operations for
syncers. This ensures cleanup (like flushing batches to disk) is
performed reliably even on cancellation or early returns.

- Add `Finalizer` interface to `sync/types.go` for explicit cleanup.
- Attach `Finalize()` in `CodeQueue` that finalizes code fetching to this new interface.
- Gather finalization logic in a `Finalize()` for StateSyncer to flush in-progress trie batches.
- Implement `Finalize()` for AtomicSyncer to commit pending database changes.
- Add `FinalizeAll()` to SyncerRegistry with defer to ensure cleanup runs.
- Remove `OnFailure` callback mechanism (replaced by `Finalizer`).

resolves #1089

Signed-off-by: Tsvetan Dimitrov ([email protected])

Author: powerslider

plugin/evm/atomic/sync/syncer.go

@@ -33,6 +33,7 @@ const (
 var (
 	_ sync.Syncer             = (*Syncer)(nil)
 	_ syncclient.LeafSyncTask = (*syncerLeafTask)(nil)
+	_ sync.Finalizer          = (*Syncer)(nil)
 
 	errTargetHeightRequired = errors.New("target height must be > 0")
 )
@@ -125,7 +126,6 @@ func NewSyncer(client syncclient.LeafClient, db *versiondb.Database, atomicTrie
 	syncer.syncer = syncclient.NewCallbackLeafSyncer(client, tasks, &syncclient.LeafSyncerConfig{
 		RequestSize: cfg.requestSize,
 		NumWorkers:  cfg.numWorkers,
-		OnFailure:   func() {}, // No-op since we flush progress to disk at the regular commit interval.
 	})
 
 	return syncer, nil
@@ -146,6 +146,16 @@ func (s *Syncer) Sync(ctx context.Context) error {
 	return s.syncer.Sync(ctx)
 }
 
+// Finalize commits any pending database changes to disk.
+// This ensures that even if the sync is cancelled or fails, we preserve
+// the progress up to the last fully synced height.
+func (s *Syncer) Finalize() error {
+	if s.db == nil {
+		return nil
+	}
+	return s.db.Commit()
+}
+
 // addZeroes returns the big-endian representation of `height`, prefixed with [common.HashLength] zeroes.
 func addZeroes(height uint64) []byte {
 	// Key format is [height(8 bytes)][blockchainID(32 bytes)]. Start should be the

plugin/evm/vmsync/registry.go

@@ -53,6 +53,10 @@ func (r *SyncerRegistry) Register(syncer syncpkg.Syncer) error {
 
 // RunSyncerTasks executes all registered syncers synchronously.
 func (r *SyncerRegistry) RunSyncerTasks(ctx context.Context, summary message.Syncable) error {
+	// Ensure finalization runs regardless of how this function exits.
+	// This guarantees cleanup even on early returns or panics.
+	defer r.FinalizeAll(summary)
+
 	// Early return if context is already canceled (e.g., during shutdown).
 	if err := ctx.Err(); err != nil {
 		return err
@@ -102,3 +106,15 @@ func (r *SyncerRegistry) StartAsync(ctx context.Context, summary message.Syncabl
 
 	return g
 }
+
+// FinalizeAll iterates over all registered syncers and calls Finalize on those that implement the Finalizer interface.
+// Errors are logged but not returned to ensure best-effort cleanup of all syncers.
+func (r *SyncerRegistry) FinalizeAll(summary message.Syncable) {
+	for _, task := range r.syncers {
+		if f, ok := task.syncer.(syncpkg.Finalizer); ok {
+			if err := f.Finalize(); err != nil {
+				log.Error("failed to finalize syncer", "syncer", task.name, "err", err, "summary", summary.GetBlockHash().Hex(), "height", summary.Height())
+			}
+		}
+	}
+}

sync/blocksync/syncer.go

@@ -78,11 +78,6 @@ func (s *BlockSyncer) Sync(ctx context.Context) error {
 	// first, check for blocks already available on disk so we don't
 	// request them from peers.
 	for blocksToFetch > 0 {
-		// Check for context cancellation before checking each block.
-		if err := ctx.Err(); err != nil {
-			return err
-		}
-
 		blk := rawdb.ReadBlock(s.db, nextHash, nextHeight)
 		if blk == nil {
 			// block was not found
@@ -99,11 +94,6 @@ func (s *BlockSyncer) Sync(ctx context.Context) error {
 	// them to disk.
 	batch := s.db.NewBatch()
 	for fetched := uint64(0); fetched < blocksToFetch && (nextHash != common.Hash{}); {
-		// Check for context cancellation before making network requests.
-		if err := ctx.Err(); err != nil {
-			return err
-		}
-
 		log.Info("fetching blocks from peer", "fetched", fetched, "total", blocksToFetch)
 		blocks, err := s.client.GetBlocks(ctx, nextHash, nextHeight, blocksPerRequest)
 		if err != nil {

sync/client/leaf_syncer.go

@@ -37,7 +37,6 @@ type LeafSyncTask interface {
 type LeafSyncerConfig struct {
 	RequestSize uint16 // Number of leafs to request from a peer at a time
 	NumWorkers  int    // Number of workers to process leaf sync tasks
-	OnFailure   func() // Callback for handling errors during sync
 }
 
 type CallbackLeafSyncer struct {
@@ -159,9 +158,9 @@ func (c *CallbackLeafSyncer) Sync(ctx context.Context) error {
 		})
 	}
 
-	err := eg.Wait()
-	if err != nil {
-		c.config.OnFailure()
+	if err := eg.Wait(); err != nil {
+		return err
 	}
-	return err
+
+	return nil
 }

sync/statesync/code_queue.go

@@ -15,11 +15,14 @@ import (
 	"github.com/ava-labs/libevm/libevm/options"
 
 	"github.com/ava-labs/coreth/plugin/evm/customrawdb"
+	syncpkg "github.com/ava-labs/coreth/sync"
 )
 
 const defaultQueueCapacity = 5000
 
 var (
+	_ syncpkg.Finalizer = (*CodeQueue)(nil)
+
 	errFailedToAddCodeHashesToQueue = errors.New("failed to add code hashes to queue")
 	errFailedToFinalizeCodeQueue    = errors.New("failed to finalize code queue")
 )
@@ -175,6 +178,8 @@ func (q *CodeQueue) init() error {
 	if err != nil {
 		return fmt.Errorf("unable to recover previous sync state: %w", err)
 	}
+	// Use context.Background() since init() runs during construction before sync starts.
+	// The channel is empty, so sends won't block. Shutdown is handled via q.quit in AddCode.
 	if err := q.AddCode(context.Background(), dbCodeHashes); err != nil {
 		return fmt.Errorf("unable to resume previous sync: %w", err)
 	}

sync/statesync/state_syncer.go

@@ -106,7 +106,6 @@ func NewSyncer(client syncclient.Client, db ethdb.Database, root common.Hash, co
 	ss.syncer = syncclient.NewCallbackLeafSyncer(client, ss.segments, &syncclient.LeafSyncerConfig{
 		RequestSize: leafsRequestSize,
 		NumWorkers:  defaultNumWorkers,
-		OnFailure:   ss.onSyncFailure,
 	})
 
 	if codeQueue == nil {
@@ -301,19 +300,19 @@ func (t *stateSync) removeTrieInProgress(root common.Hash) (int, error) {
 	return len(t.triesInProgress), nil
 }
 
-// onSyncFailure is called if the sync fails, this writes all
-// batches of in-progress trie segments to disk to have maximum
-// progress to restore.
-func (t *stateSync) onSyncFailure() {
+// Finalize checks if there are any in-progress tries and flushes their batches to disk
+// to preserve progress. This is called by the syncer registry on sync failure or cancellation.
+func (t *stateSync) Finalize() error {
 	t.lock.RLock()
 	defer t.lock.RUnlock()
 
 	for _, trie := range t.triesInProgress {
 		for _, segment := range trie.segments {
 			if err := segment.batch.Write(); err != nil {
-				log.Error("failed to write segment batch on sync failure", "err", err)
-				return
+				log.Error("failed to write segment batch on finalize", "err", err)
+				return err
 			}
 		}
 	}
+	return nil
 }

sync/statesync/trie_sync_tasks.go

@@ -143,9 +143,13 @@ func (s *storageTrieTask) OnFinish() error {
 	return s.sync.onStorageTrieFinished(s.root)
 }
 
-func (s *storageTrieTask) OnLeafs(_ context.Context, db ethdb.KeyValueWriter, keys, vals [][]byte) error {
+func (s *storageTrieTask) OnLeafs(ctx context.Context, db ethdb.KeyValueWriter, keys, vals [][]byte) error {
 	// persists the trie leafs to the snapshot for all accounts associated with this root
 	for _, account := range s.accounts {
+		// Check context cancellation before processing each account to allow early exit during shutdown.
+		if err := ctx.Err(); err != nil {
+			return err
+		}
 		for i, key := range keys {
 			rawdb.WriteStorageSnapshot(db, account, common.BytesToHash(key), vals[i])
 		}

sync/types.go

@@ -31,6 +31,13 @@ type Syncer interface {
 	ID() string
 }
 
+// Finalizer provides a mechanism to perform cleanup operations after a sync operation.
+// This is useful for handling inflight requests, flushing to disk, or other cleanup tasks.
+type Finalizer interface {
+	// Finalize performs any necessary cleanup operations.
+	Finalize() error
+}
+
 // SummaryProvider is an interface for providing state summaries.
 type SummaryProvider interface {
 	StateSummaryAtBlock(ethBlock *types.Block) (block.StateSummary, error)