Legacy receipt converter tool

cmd/geth/dbcmd.go

@@ -34,6 +34,7 @@ import (
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/console/prompt"
 	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/trie"
@@ -69,6 +70,7 @@ Remove blockchain and state databases`,
 			dbDumpFreezerIndex,
 			dbImportCmd,
 			dbExportCmd,
+			dbMigrateFreezer,
 		},
 	}
 	dbInspectCmd = cli.Command{
@@ -232,6 +234,21 @@ WARNING: This is a low-level operation which may cause database corruption!`,
 		},
 		Description: "Exports the specified chain data to an RLP encoded stream, optionally gzip-compressed.",
 	}
+	dbMigrateFreezer = cli.Command{
+		Action:    utils.MigrateFlags(freezerMigrate),
+		Name:      "freezer-migrate",
+		Usage:     "Migrate legacy parts of the freezer. (WARNING: may take a long time)",
+		ArgsUsage: "",
+		Flags: []cli.Flag{
+			utils.DataDirFlag,
+			utils.SyncModeFlag,
+			utils.MainnetFlag,
+			utils.RopstenFlag,
+			utils.RinkebyFlag,
+			utils.GoerliFlag,
+		},
+		Description: "The import command imports the specific chain data from an RLP encoded stream.",
+	}
 )
 
 func removeDB(ctx *cli.Context) error {
@@ -684,3 +701,84 @@ func exportChaindata(ctx *cli.Context) error {
 	db := utils.MakeChainDatabase(ctx, stack, true)
 	return utils.ExportChaindata(ctx.Args().Get(1), kind, exporter(db), stop)
 }
+
+func freezerMigrate(ctx *cli.Context) error {
+	stack, _ := makeConfigNode(ctx)
+	defer stack.Close()
+
+	db := utils.MakeChainDatabase(ctx, stack, false)
+	defer db.Close()
+
+	// Check first block for legacy receipt format
+	numAncients, err := db.Ancients()
+	if err != nil {
+		return err
+	}
+	if numAncients < 1 {
+		log.Info("No blocks in freezer to migrate")
+		return nil
+	}
+
+	// Find first block with non-empty receipt
+	firstIdx := uint64(0)
+	emptyRLPList := []byte{192}
+	for i := uint64(0); i < numAncients; i++ {
+		r, err := db.Ancient("receipts", i)
+		if err != nil {
+			return err
+		}
+		if len(r) == 0 {
+			continue
+		}
+		if !bytes.Equal(r, emptyRLPList) {
+			firstIdx = i
+			break
+		}
+	}
+	// Is first non-empty receipt legacy?
+	first, err := db.Ancient("receipts", firstIdx)
+	if err != nil {
+		return err
+	}
+	isFirstLegacy, err := types.IsLegacyStoredReceipts(first)
+	if err != nil {
+		return err
+	}
+	if !isFirstLegacy {
+		log.Info("No legacy receipts to migrate", "number", firstIdx)
+		return nil
+	}
+	log.Info("First legacy receipt", "number", firstIdx)
+
+	transformer := func(blob []byte) ([]byte, bool, error) {
+		// Stop when first v5 receipt is spotted.
+		// TODO: Combine detecting legacy and converting it
+		// to avoid 2 decoding.
+		legacy, err := types.IsLegacyStoredReceipts(blob)
+		if err != nil {
+			return nil, false, err
+		}
+		if !legacy {
+			return blob, true, nil
+		}
+
+		out, err := types.ConvertLegacyStoredReceipts(blob)
+		if err != nil {
+			return nil, false, err
+		}
+		return out, false, nil
+	}
+
+	log.Info("Starting migration", "ancients", numAncients)
+	start := time.Now()
+	if err := db.MigrateTable("receipts", transformer); err != nil {
+		return err
+	}
+
+	if err := db.Close(); err != nil {
+		return err
+	}
+	log.Info("Migration finished", "duration", time.Since(start))
+
+	return nil
+}

core/rawdb/database.go

@@ -135,6 +135,10 @@ func (db *nofreezedb) ReadAncients(fn func(reader ethdb.AncientReader) error) (e
 	return fn(db)
 }
 
+func (db *nofreezedb) MigrateTable(_ string, _ TransformerFn) error {
+	return errNotSupported
+}
+
 // NewDatabase creates a high level database on top of a given key-value data
 // store without a freezer moving immutable chain segments into cold storage.
 func NewDatabase(db ethdb.KeyValueStore) ethdb.Database {

core/rawdb/freezer.go

@@ -19,6 +19,7 @@ package rawdb
 import (
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"math"
 	"os"
 	"path/filepath"
@@ -64,6 +65,7 @@ const (
 
 	// freezerTableSize defines the maximum size of freezer data files.
 	freezerTableSize = 2 * 1000 * 1000 * 1000
+	//freezerTableSize = 2 * 1024 * 1024
 )
 
 // freezer is an memory mapped append-only database to store immutable chain data
@@ -546,3 +548,168 @@ func (f *freezer) freezeRange(nfdb *nofreezedb, number, limit uint64) (hashes []
 
 	return hashes, err
 }
+
+// TransformerFn takes a freezer entry in an older format and returns
+// the same in a new format. The second return argument determines
+// if the entry is already of the new format.
+type TransformerFn = func([]byte) ([]byte, bool, 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, fn TransformerFn) error {
+	if f.readonly {
+		return errReadOnly
+	}
+	f.writeLock.Lock()
+	defer f.writeLock.Unlock()
+
+	table, ok := f.tables[kind]
+	if !ok {
+		return errUnknownTable
+	}
+	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])
+	if err != nil {
+		return err
+	}
+	batch := newTable.newBatch()
+
+	numAncients, err := f.Ancients()
+	if err != nil {
+		return err
+	}
+	i := uint64(0)
+	// Number of the file in which the first up-to-date receipt appers
+	var filenum uint32
+	copyOver := true
+	// Iterate through entries and transform them
+	// until reaching first non-legacy one.
+	for ; i < numAncients; i++ {
+		if i%500000 == 0 {
+			log.Info("Processing legacy elements", "number", i)
+		}
+		blob, err := table.Retrieve(i)
+		if err != nil {
+			return err
+		}
+		out, stop, err := fn(blob)
+		if err != nil {
+			return err
+		}
+		if !stop {
+			// Entry is legacy, push transformed version
+			if err := batch.AppendRaw(i, out); err != nil {
+				return err
+			}
+		} else {
+			// Reached the first up-to-date entry.
+			// Remember in which file the switch happens.
+			indices, err := table.getIndices(i, 1)
+			if err != nil {
+				return err
+			}
+			// First non-legacy entry coincidentally is located in a new
+			// file and we have a clean switch-over boundary. No need to
+			// copy over further elements.
+			if indices[0].filenum != indices[1].filenum {
+				copyOver = false
+			}
+			filenum = indices[1].filenum
+			log.Info("Found first non-legacy element", "number", i, "filenum", filenum, "copyOver", copyOver)
+			break
+		}
+	}
+	if copyOver {
+		// Copy over left-over receipts in the file with last legacy receipt:
+		// 1. loop getBounds until filenum exceeds threshold filenum
+		// 2. copy verbatim to new table
+		for ; i < numAncients; i++ {
+			indices, err := table.getIndices(i, 1)
+			if err != nil {
+				return err
+			}
+			if indices[1].filenum > filenum {
+				log.Info("Copied over rest of switch-over file", "number", i, "nextFile", indices[1].filenum)
+				break
+			}
+			blob, err := table.Retrieve(i)
+			if err != nil {
+				return err
+			}
+			if err := batch.AppendRaw(i, blob); err != nil {
+				return err
+			}
+		}
+	}
+
+	if err := batch.commit(); err != nil {
+		return err
+	}
+
+	var fileCountDiff int32
+	toRename := make(map[uint32]struct{})
+
+	// 3. need to copy rest of old index and repair the filenum in the entries
+	if i < numAncients {
+		indices, err := table.getIndices(i, 1)
+		if err != nil {
+			return err
+		}
+
+		log.Info("Duplicating rest of index", "fromFile", indices[1].filenum, "toFile", newTable.headId+1)
+		fileCountDiff = int32(newTable.headId+1) - int32(indices[1].filenum)
+		for ; i < numAncients; i++ {
+			indices, err := table.getIndices(i, 1)
+			if err != nil {
+				return err
+			}
+			idx := indexEntry{
+				// (idx.filenum + fileCountDiff) is always > 0
+				filenum: uint32(int32(indices[1].filenum) + fileCountDiff),
+				offset:  indices[1].offset,
+			}
+			newTable.writeEntry(idx)
+			toRename[indices[1].filenum] = struct{}{}
+		}
+	}
+
+	log.Info("Replacing table files")
+	// Warning: file juggling to follow.
+	// First release open table files because we need to move some of them.
+	table.releaseFilesAfter(0, false)
+	// Move table files after the switchover point to migration dir.
+	for k := range toRename {
+		if err := os.Rename(table.tableFilePath(k), newTable.tableFilePath(uint32(int32(k)+fileCountDiff))); err != nil {
+			return err
+		}
+	}
+
+	// Now we can delete all the rest.
+	if err := table.deleteFiles(); err != nil {
+		return err
+	}
+
+	// Move migrated files to ancients dir.
+	if err := newTable.Close(); err != nil {
+		return err
+	}
+	files, err := ioutil.ReadDir(migrationPath)
+	if err != nil {
+		return err
+	}
+	for _, f := range files {
+		// This will replace the index + table files up to and including the switchover file.
+		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
+}

core/rawdb/freezer_table.go

@@ -431,13 +431,8 @@ func (t *freezerTable) Close() error {
 func (t *freezerTable) openFile(num uint32, opener func(string) (*os.File, error)) (f *os.File, err error) {
 	var exist bool
 	if f, exist = t.files[num]; !exist {
-		var name string
-		if t.noCompression {
-			name = fmt.Sprintf("%s.%04d.rdat", t.name, num)
-		} else {
-			name = fmt.Sprintf("%s.%04d.cdat", t.name, num)
-		}
-		f, err = opener(filepath.Join(t.path, name))
+		path := t.tableFilePath(num)
+		f, err = opener(path)
 		if err != nil {
 			return nil, err
 		}
@@ -446,6 +441,17 @@ func (t *freezerTable) openFile(num uint32, opener func(string) (*os.File, error
 	return f, err
 }
 
+// tableFileName returns path to a table file of a given index.
+func (t *freezerTable) tableFilePath(num uint32) string {
+	var name string
+	if t.noCompression {
+		name = fmt.Sprintf("%s.%04d.rdat", t.name, num)
+	} else {
+		name = fmt.Sprintf("%s.%04d.cdat", t.name, num)
+	}
+	return filepath.Join(t.path, name)
+}
+
 // releaseFile closes a file, and removes it from the open file cache.
 // Assumes that the caller holds the write lock
 func (t *freezerTable) releaseFile(num uint32) {
@@ -468,6 +474,20 @@ func (t *freezerTable) releaseFilesAfter(num uint32, remove bool) {
 	}
 }
 
+// deleteFiles deletes table files based on tail and head
+// indices, and assumes these files are not open.
+func (t *freezerTable) deleteFiles() error {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	for i := t.tailId; i < t.headId; i++ {
+		path := t.tableFilePath(i)
+		if err := os.Remove(path); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // getIndices returns the index entries for the given from-item, covering 'count' items.
 // N.B: The actual number of returned indices for N items will always be N+1 (unless an
 // error is returned).
@@ -740,3 +760,13 @@ func (t *freezerTable) dumpIndex(w io.Writer, start, stop int64) {
 	}
 	fmt.Fprintf(w, "|--------------------------|\n")
 }
+
+// low-level, doesnt increase counters, assumes lock
+func (t *freezerTable) writeEntry(idx indexEntry) error {
+	// Ensure the table is still accessible
+	if t.index == nil {
+		return errClosed
+	}
+	_, err := t.index.Write(idx.append([]byte{}))
+	return err
+}

core/rawdb/table.go

@@ -101,6 +101,10 @@ func (t *table) Sync() error {
 	return t.db.Sync()
 }
 
+func (t *table) MigrateTable(kind string, fn TransformerFn) error {
+	return t.db.MigrateTable(kind, fn)
+}
+
 // Put inserts the given value into the database at a prefixed version of the
 // provided key.
 func (t *table) Put(key []byte, value []byte) error {