Add tool for database conversion

Dockerfile

@@ -249,6 +249,7 @@ COPY --from=node-builder /workspace/target/bin/nitro /usr/local/bin/
 COPY --from=node-builder /workspace/target/bin/relay /usr/local/bin/
 COPY --from=node-builder /workspace/target/bin/nitro-val /usr/local/bin/
 COPY --from=node-builder /workspace/target/bin/seq-coordinator-manager /usr/local/bin/
+COPY --from=node-builder /workspace/target/bin/dbconv /usr/local/bin/
 COPY --from=machine-versions /workspace/machines /home/user/target/machines
 USER root
 RUN export DEBIAN_FRONTEND=noninteractive && \

Makefile

@@ -153,7 +153,7 @@ push: lint test-go .make/fmt
 all: build build-replay-env test-gen-proofs
 	@touch .make/all
 
-build: $(patsubst %,$(output_root)/bin/%, nitro deploy relay daserver datool seq-coordinator-invalidate nitro-val seq-coordinator-manager)
+build: $(patsubst %,$(output_root)/bin/%, nitro deploy relay daserver datool seq-coordinator-invalidate nitro-val seq-coordinator-manager dbconv)
 	@printf $(done)
 
 build-node-deps: $(go_source) build-prover-header build-prover-lib build-jit .make/solgen .make/cbrotli-lib
@@ -266,6 +266,9 @@ $(output_root)/bin/nitro-val: $(DEP_PREDICATE) build-node-deps
 $(output_root)/bin/seq-coordinator-manager: $(DEP_PREDICATE) build-node-deps
 	go build $(GOLANG_PARAMS) -o $@ "$(CURDIR)/cmd/seq-coordinator-manager"
 
+$(output_root)/bin/dbconv: $(DEP_PREDICATE) build-node-deps
+	go build $(GOLANG_PARAMS) -o $@ "$(CURDIR)/cmd/dbconv"
+
 # recompile wasm, but don't change timestamp unless files differ
 $(replay_wasm): $(DEP_PREDICATE) $(go_source) .make/solgen
 	mkdir -p `dirname $(replay_wasm)`

cmd/dbconv/dbconv/config.go

@@ -0,0 +1,82 @@
+package dbconv
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/offchainlabs/nitro/cmd/conf"
+	"github.com/offchainlabs/nitro/cmd/genericconf"
+	"github.com/offchainlabs/nitro/execution/gethexec"
+	flag "github.com/spf13/pflag"
+)
+
+type DBConfig struct {
+	Data      string            `koanf:"data"`
+	DBEngine  string            `koanf:"db-engine"`
+	Handles   int               `koanf:"handles"`
+	Cache     int               `koanf:"cache"`
+	Namespace string            `koanf:"namespace"`
+	Pebble    conf.PebbleConfig `koanf:"pebble"`
+}
+
+var DBConfigDefault = DBConfig{
+	Handles: conf.PersistentConfigDefault.Handles,
+	Cache:   gethexec.DefaultCachingConfig.DatabaseCache,
+	Pebble:  conf.PebbleConfigDefault,
+}
+
+func DBConfigAddOptions(prefix string, f *flag.FlagSet, defaultNamespace string) {
+	f.String(prefix+".data", DBConfigDefault.Data, "directory of stored chain state")
+	f.String(prefix+".db-engine", DBConfigDefault.DBEngine, "backing database implementation to use ('leveldb' or 'pebble')")
+	f.Int(prefix+".handles", DBConfigDefault.Handles, "number of file descriptor handles to use for the database")
+	f.Int(prefix+".cache", DBConfigDefault.Cache, "the capacity(in megabytes) of the data caching")
+	f.String(prefix+".namespace", defaultNamespace, "metrics namespace")
+	conf.PebbleConfigAddOptions(prefix+".pebble", f)
+}
+
+type DBConvConfig struct {
+	Src            DBConfig                        `koanf:"src"`
+	Dst            DBConfig                        `koanf:"dst"`
+	IdealBatchSize int                             `koanf:"ideal-batch-size"`
+	Convert        bool                            `koanf:"convert"`
+	Compact        bool                            `koanf:"compact"`
+	Verify         int                             `koanf:"verify"`
+	LogLevel       string                          `koanf:"log-level"`
+	LogType        string                          `koanf:"log-type"`
+	Metrics        bool                            `koanf:"metrics"`
+	MetricsServer  genericconf.MetricsServerConfig `koanf:"metrics-server"`
+}
+
+var DefaultDBConvConfig = DBConvConfig{
+	IdealBatchSize: 100 * 1024 * 1024, // 100 MB
+	Convert:        false,
+	Compact:        false,
+	Verify:         0,
+	LogLevel:       "INFO",
+	LogType:        "plaintext",
+	Metrics:        false,
+	MetricsServer:  genericconf.MetricsServerConfigDefault,
+}
+
+func DBConvConfigAddOptions(f *flag.FlagSet) {
+	DBConfigAddOptions("src", f, "srcdb/")
+	DBConfigAddOptions("dst", f, "destdb/")
+	f.Int("ideal-batch-size", DefaultDBConvConfig.IdealBatchSize, "ideal write batch size")
+	f.Bool("convert", DefaultDBConvConfig.Convert, "enables conversion step")
+	f.Bool("compact", DefaultDBConvConfig.Compact, "enables compaction step")
+	f.Int("verify", DefaultDBConvConfig.Verify, "enables verification step (0 = disabled, 1 = only keys, 2 = keys and values)")
+	f.String("log-level", DefaultDBConvConfig.LogLevel, "log level, valid values are CRIT, ERROR, WARN, INFO, DEBUG, TRACE")
+	f.String("log-type", DefaultDBConvConfig.LogType, "log type (plaintext or json)")
+	f.Bool("metrics", DefaultDBConvConfig.Metrics, "enable metrics")
+	genericconf.MetricsServerAddOptions("metrics-server", f)
+}
+
+func (c *DBConvConfig) Validate() error {
+	if c.Verify < 0 || c.Verify > 2 {
+		return fmt.Errorf("Invalid verify config value: %v", c.Verify)
+	}
+	if !c.Convert && c.Verify == 0 && !c.Compact {
+		return errors.New("nothing to be done, conversion, verification and compaction disabled")
+	}
+	return nil
+}

cmd/dbconv/dbconv/dbconv.go

@@ -0,0 +1,152 @@
+package dbconv
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+type DBConverter struct {
+	config *DBConvConfig
+	stats  Stats
+
+	src ethdb.Database
+	dst ethdb.Database
+}
+
+func NewDBConverter(config *DBConvConfig) *DBConverter {
+	return &DBConverter{
+		config: config,
+	}
+}
+
+func openDB(config *DBConfig, name string, readonly bool) (ethdb.Database, error) {
+	return rawdb.Open(rawdb.OpenOptions{
+		Type:               config.DBEngine,
+		Directory:          config.Data,
+		AncientsDirectory:  "", // don't open freezer
+		Namespace:          config.Namespace,
+		Cache:              config.Cache,
+		Handles:            config.Handles,
+		ReadOnly:           readonly,
+		PebbleExtraOptions: config.Pebble.ExtraOptions(name),
+	})
+}
+
+func (c *DBConverter) Convert(ctx context.Context) error {
+	var err error
+	defer c.Close()
+	c.src, err = openDB(&c.config.Src, "src", true)
+	if err != nil {
+		return err
+	}
+	c.dst, err = openDB(&c.config.Dst, "dst", false)
+	if err != nil {
+		return err
+	}
+	c.stats.Reset()
+	it := c.src.NewIterator(nil, nil)
+	defer it.Release()
+	batch := c.dst.NewBatch()
+	entriesInBatch := 0
+	for it.Next() && ctx.Err() == nil {
+		if err = batch.Put(it.Key(), it.Value()); err != nil {
+			return err
+		}
+		entriesInBatch++
+		if batchSize := batch.ValueSize(); batchSize >= c.config.IdealBatchSize {
+			if err = batch.Write(); err != nil {
+				return err
+			}
+			c.stats.AddEntries(int64(entriesInBatch))
+			c.stats.AddBytes(int64(batchSize))
+			batch.Reset()
+			entriesInBatch = 0
+		}
+	}
+	if err = ctx.Err(); err == nil {
+		batchSize := batch.ValueSize()
+		err = batch.Write()
+		c.stats.AddEntries(int64(entriesInBatch))
+		c.stats.AddBytes(int64(batchSize))
+	}
+	return err
+}
+
+func (c *DBConverter) CompactDestination() error {
+	var err error
+	c.dst, err = openDB(&c.config.Dst, "dst", false)
+	if err != nil {
+		return err
+	}
+	defer c.dst.Close()
+	start := time.Now()
+	log.Info("Compacting destination database", "data", c.config.Dst.Data)
+	if err := c.dst.Compact(nil, nil); err != nil {
+		return err
+	}
+	log.Info("Compaction done", "elapsed", time.Since(start))
+	return nil
+}
+
+func (c *DBConverter) Verify(ctx context.Context) error {
+	if c.config.Verify == 1 {
+		log.Info("Starting quick verification - verifying only keys existence")
+	} else {
+		log.Info("Starting full verification - verifying keys and values")
+	}
+	var err error
+	defer c.Close()
+	c.src, err = openDB(&c.config.Src, "src", true)
+	if err != nil {
+		return err
+	}
+	c.dst, err = openDB(&c.config.Dst, "dst", true)
+	if err != nil {
+		return err
+	}
+
+	c.stats.Reset()
+	it := c.src.NewIterator(nil, nil)
+	defer it.Release()
+	for it.Next() && ctx.Err() == nil {
+		switch c.config.Verify {
+		case 1:
+			if has, err := c.dst.Has(it.Key()); !has {
+				return fmt.Errorf("Missing key in destination db, key: %v, err: %w", it.Key(), err)
+			}
+			c.stats.AddBytes(int64(len(it.Key())))
+		case 2:
+			dstValue, err := c.dst.Get(it.Key())
+			if err != nil {
+				return err
+			}
+			if !bytes.Equal(dstValue, it.Value()) {
+				return fmt.Errorf("Value mismatch for key: %v, src value: %v, dst value: %s", it.Key(), it.Value(), dstValue)
+			}
+			c.stats.AddBytes(int64(len(it.Key()) + len(dstValue)))
+		default:
+			return fmt.Errorf("Invalid verify config value: %v", c.config.Verify)
+		}
+		c.stats.AddEntries(1)
+	}
+	return ctx.Err()
+}
+
+func (c *DBConverter) Stats() *Stats {
+	return &c.stats
+}
+
+func (c *DBConverter) Close() {
+	if c.src != nil {
+		c.src.Close()
+	}
+	if c.dst != nil {
+		c.dst.Close()
+	}
+}

cmd/dbconv/dbconv/dbconv_test.go

@@ -0,0 +1,85 @@
+package dbconv
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/offchainlabs/nitro/util/testhelpers"
+)
+
+func TestConversion(t *testing.T) {
+	_ = testhelpers.InitTestLog(t, log.LvlTrace)
+	oldDBConfig := DBConfigDefault
+	oldDBConfig.Data = t.TempDir()
+	oldDBConfig.DBEngine = "leveldb"
+
+	newDBConfig := DBConfigDefault
+	newDBConfig.Data = t.TempDir()
+	newDBConfig.DBEngine = "pebble"
+
+	func() {
+		oldDb, err := openDB(&oldDBConfig, "", false)
+		defer oldDb.Close()
+		Require(t, err)
+		err = oldDb.Put([]byte{}, []byte{0xde, 0xed, 0xbe, 0xef})
+		Require(t, err)
+		for i := 0; i < 20; i++ {
+			err = oldDb.Put([]byte{byte(i)}, []byte{byte(i + 1)})
+			Require(t, err)
+		}
+	}()
+
+	config := DefaultDBConvConfig
+	config.Src = oldDBConfig
+	config.Dst = newDBConfig
+	config.IdealBatchSize = 5
+	conv := NewDBConverter(&config)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	err := conv.Convert(ctx)
+	Require(t, err)
+	conv.Close()
+
+	oldDb, err := openDB(&oldDBConfig, "", true)
+	Require(t, err)
+	defer oldDb.Close()
+	newDb, err := openDB(&newDBConfig, "", true)
+	Require(t, err)
+	defer newDb.Close()
+
+	func() {
+		it := oldDb.NewIterator(nil, nil)
+		defer it.Release()
+		for it.Next() {
+			if has, _ := newDb.Has(it.Key()); !has {
+				t.Log("Missing key in the converted db, key:", it.Key())
+			}
+			newValue, err := newDb.Get(it.Key())
+			Require(t, err)
+			if !bytes.Equal(newValue, it.Value()) {
+				Fail(t, "Value mismatch, old:", it.Value(), "new:", newValue)
+			}
+		}
+	}()
+	func() {
+		it := newDb.NewIterator(nil, nil)
+		defer it.Release()
+		for it.Next() {
+			if has, _ := oldDb.Has(it.Key()); !has {
+				Fail(t, "Unexpected key in the converted db, key:", it.Key())
+			}
+		}
+	}()
+}
+
+func Require(t *testing.T, err error, printables ...interface{}) {
+	t.Helper()
+	testhelpers.RequireImpl(t, err, printables...)
+}
+
+func Fail(t *testing.T, printables ...interface{}) {
+	t.Helper()
+	testhelpers.FailImpl(t, printables...)
+}