Add migrator logic to evm block database

vms/evm/database/blockdb/database.go

@@ -24,6 +24,8 @@ import (
 )
 
 var (
+	errUnexpectedKey        = errors.New("unexpected database key")
+	errNotInitialized       = errors.New("database not initialized")
 	errAlreadyInitialized   = errors.New("database already initialized")
 	errInvalidEncodedLength = errors.New("invalid encoded length")
 )
@@ -47,6 +49,7 @@ type Database struct {
 	dbPath    string
 	minHeight uint64
 
+	migrator       *migrator
 	heightDBsReady bool
 
 	reg    prometheus.Registerer
@@ -157,6 +160,10 @@ func New(
 			// Load existing database min height.
 			return databaseMinHeight(db.metaDB)
 		},
+		func() (uint64, bool, error) {
+			// Use the minimum block height of existing blocks to migrate.
+			return minBlockHeightToMigrate(evmDB)
+		},
 		func() (uint64, bool, error) {
 			// Use min height 1 unless deferring initialization.
 			return 1, !allowDeferredInit, nil
@@ -212,6 +219,15 @@ func (db *Database) InitBlockDBs(minHeight uint64) error {
 	db.bodyDB = bodyDB
 	db.receiptsDB = receiptsDB
 
+	if err := db.initMigrator(); err != nil {
+		return errors.Join(
+			fmt.Errorf("failed to initialize migrator: %w", err),
+			headerDB.Close(),
+			bodyDB.Close(),
+			receiptsDB.Close(),
+		)
+	}
+
 	db.heightDBsReady = true
 	db.minHeight = minHeight
 
@@ -250,6 +266,7 @@ func parseBlockKey(key []byte) (num uint64, hash common.Hash, ok bool) {
 }
 
 type parsedBlockKey struct {
+	key  []byte
 	db   database.HeightIndex
 	num  uint64
 	hash common.Hash
@@ -301,15 +318,19 @@ func (db *Database) parseKey(key []byte) (*parsedBlockKey, bool) {
 	}
 
 	return &parsedBlockKey{
+		key:  key,
 		db:   hdb,
 		num:  num,
 		hash: hash,
 	}, true
 }
 
-func (*Database) readBlock(p *parsedBlockKey) ([]byte, error) {
+func (db *Database) readBlock(p *parsedBlockKey) ([]byte, error) {
 	data, err := p.db.Get(p.num)
 	if err != nil {
+		if errors.Is(err, database.ErrNotFound) && !db.migrator.isCompleted() {
+			return db.Database.Get(p.key)
+		}
 		return nil, err
 	}
 
@@ -376,6 +397,7 @@ func (db *Database) Delete(key []byte) error {
 }
 
 func (db *Database) Close() error {
+	db.migrator.stop()
 	if !db.heightDBsReady {
 		return db.Database.Close()
 	}

vms/evm/database/blockdb/database_test.go

@@ -4,6 +4,7 @@
 package blockdb
 
 import (
+	"slices"
 	"testing"
 
 	"github.com/ava-labs/libevm/core/rawdb"
@@ -275,6 +276,25 @@ func TestDatabaseInitialization(t *testing.T) {
 			wantDBReady:   true,
 			wantMinHeight: 2,
 		},
+		{
+			name:          "non genesis blocks to migrate",
+			evmDBBlocks:   blocks[5:10],
+			wantDBReady:   true,
+			wantMinHeight: 5,
+		},
+		{
+			name:          "blocks to migrate - including genesis",
+			evmDBBlocks:   slices.Concat([]*types.Block{blocks[0]}, blocks[5:10]),
+			wantDBReady:   true,
+			wantMinHeight: 5,
+		},
+		{
+			name:          "blocks to migrate and deferred init",
+			deferInit:     true,
+			evmDBBlocks:   blocks[5:10],
+			wantDBReady:   true,
+			wantMinHeight: 5,
+		},
 	}
 
 	for _, tc := range tests {

vms/evm/database/blockdb/helpers_test.go

@@ -4,8 +4,10 @@
 package blockdb
 
 import (
+	"fmt"
 	"math/big"
 	"testing"
+	"time"
 
 	"github.com/ava-labs/libevm/common"
 	"github.com/ava-labs/libevm/core/rawdb"
@@ -17,13 +19,28 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 
+	"github.com/ava-labs/avalanchego/database"
 	"github.com/ava-labs/avalanchego/database/leveldb"
 	"github.com/ava-labs/avalanchego/utils/logging"
 
 	evmdb "github.com/ava-labs/avalanchego/vms/evm/database"
 	heightindexdb "github.com/ava-labs/avalanchego/x/blockdb"
 )
 
+// blockingDatabase wraps a HeightIndex and blocks indefinitely on Put
+// if shouldBlock returns true.
+type blockingDatabase struct {
+	database.HeightIndex
+	shouldBlock func() bool
+}
+
+func (b *blockingDatabase) Put(num uint64, encodedBlock []byte) error {
+	if b.shouldBlock == nil || b.shouldBlock() {
+		<-make(chan struct{})
+	}
+	return b.HeightIndex.Put(num, encodedBlock)
+}
+
 func newDatabasesFromDir(t *testing.T, dataDir string) (*Database, ethdb.Database) {
 	t.Helper()
 
@@ -109,3 +126,34 @@ func logsFromReceipts(receipts types.Receipts) [][]*types.Log {
 	}
 	return logs
 }
+
+func startPartialMigration(t *testing.T, db *Database, blocksToMigrate uint64) {
+	t.Helper()
+
+	n := uint64(0)
+	db.migrator.headerDB = &blockingDatabase{
+		HeightIndex: db.headerDB,
+		shouldBlock: func() bool {
+			n++
+			return n > blocksToMigrate
+		},
+	}
+	startMigration(t, db, false)
+	require.Eventually(t, func() bool {
+		return db.migrator.processed.Load() >= blocksToMigrate
+	}, 5*time.Second, 100*time.Millisecond)
+}
+
+func startMigration(t *testing.T, db *Database, waitForCompletion bool) {
+	t.Helper()
+
+	db.migrator.completed.Store(false)
+	require.NoError(t, db.StartMigration(t.Context()))
+
+	if waitForCompletion {
+		timeout := 5 * time.Second
+		msg := fmt.Sprintf("Migration did not complete within timeout: %v", timeout)
+		require.True(t, db.migrator.waitMigratorDone(timeout), msg)
+		require.True(t, db.migrator.isCompleted())
+	}
+}