core/rawdb: simple legacy receipt converter (#24028)

* cmd,core: add simple legacy receipt converter

core/rawdb: use forEach in migrate

core/rawdb: batch reads in forEach

core/rawdb: make forEach anonymous fn

cmd/geth: check for legacy receipts on node startup

fix err msg

Co-authored-by: rjl493456442 <garyrong0905@gmail.com>

fix log

Co-authored-by: rjl493456442 <garyrong0905@gmail.com>

fix some review comments

add warning to cmd

drop isLegacy fn from migrateTable params

add test for windows rename

test replacing in windows case

* minor fix

* sanity check for tail-deletion

* add log before moving files around

* speed-up hack for mainnet

* fix mainnet check, use networkid instead

* check mainnet genesis

* review fixes

* resume previous migration attempt

* core/rawdb: lint fix

Co-authored-by: Martin Holst Swende <martin@swende.se>

core/rawdb/freezer.go

@@ -19,6 +19,7 @@ package rawdb
 import (
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"math"
 	"os"
 	"path/filepath"
@@ -617,3 +618,116 @@ func (f *freezer) freezeRange(nfdb *nofreezedb, number, limit uint64) (hashes []
 
 	return hashes, err
 }
+
+// convertLegacyFn takes a raw freezer entry in an older format and
+// returns it in the new format.
+type convertLegacyFn = func([]byte) ([]byte, error)
+
+// MigrateTable processes the entries in a given table in sequence
+// converting them to a new format if they're of an old format.
+func (f *freezer) MigrateTable(kind string, convert convertLegacyFn) error {
+	if f.readonly {
+		return errReadOnly
+	}
+	f.writeLock.Lock()
+	defer f.writeLock.Unlock()
+
+	table, ok := f.tables[kind]
+	if !ok {
+		return errUnknownTable
+	}
+	// forEach iterates every entry in the table serially and in order, calling `fn`
+	// with the item as argument. If `fn` returns an error the iteration stops
+	// and that error will be returned.
+	forEach := func(t *freezerTable, offset uint64, fn func(uint64, []byte) error) error {
+		var (
+			items     = atomic.LoadUint64(&t.items)
+			batchSize = uint64(1024)
+			maxBytes  = uint64(1024 * 1024)
+		)
+		for i := offset; i < items; {
+			if i+batchSize > items {
+				batchSize = items - i
+			}
+			data, err := t.RetrieveItems(i, batchSize, maxBytes)
+			if err != nil {
+				return err
+			}
+			for j, item := range data {
+				if err := fn(i+uint64(j), item); err != nil {
+					return err
+				}
+			}
+			i += uint64(len(data))
+		}
+		return nil
+	}
+	// TODO(s1na): This is a sanity-check since as of now no process does tail-deletion. But the migration
+	// process assumes no deletion at tail and needs to be modified to account for that.
+	if table.itemOffset > 0 || table.itemHidden > 0 {
+		return fmt.Errorf("migration not supported for tail-deleted freezers")
+	}
+	ancientsPath := filepath.Dir(table.index.Name())
+	// Set up new dir for the migrated table, the content of which
+	// we'll at the end move over to the ancients dir.
+	migrationPath := filepath.Join(ancientsPath, "migration")
+	newTable, err := NewFreezerTable(migrationPath, kind, FreezerNoSnappy[kind], false)
+	if err != nil {
+		return err
+	}
+	var (
+		batch  = newTable.newBatch()
+		out    []byte
+		start  = time.Now()
+		logged = time.Now()
+		offset = newTable.items
+	)
+	if offset > 0 {
+		log.Info("found previous migration attempt", "migrated", offset)
+	}
+	// Iterate through entries and transform them
+	if err := forEach(table, offset, func(i uint64, blob []byte) error {
+		if i%10000 == 0 && time.Since(logged) > 16*time.Second {
+			log.Info("Processing legacy elements", "count", i, "elapsed", common.PrettyDuration(time.Since(start)))
+			logged = time.Now()
+		}
+		out, err = convert(blob)
+		if err != nil {
+			return err
+		}
+		if err := batch.AppendRaw(i, out); err != nil {
+			return err
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	if err := batch.commit(); err != nil {
+		return err
+	}
+	log.Info("Replacing old table files with migrated ones", "elapsed", common.PrettyDuration(time.Since(start)))
+	// Release and delete old table files. Note this won't
+	// delete the index file.
+	table.releaseFilesAfter(0, true)
+
+	if err := newTable.Close(); err != nil {
+		return err
+	}
+	files, err := ioutil.ReadDir(migrationPath)
+	if err != nil {
+		return err
+	}
+	// Move migrated files to ancients dir.
+	for _, f := range files {
+		// This will replace the old index file as a side-effect.
+		if err := os.Rename(filepath.Join(migrationPath, f.Name()), filepath.Join(ancientsPath, f.Name())); err != nil {
+			return err
+		}
+	}
+	// Delete by now empty dir.
+	if err := os.Remove(migrationPath); err != nil {
+		return err
+	}
+
+	return nil
+}