Merge pull request #75 from sei-protocol/ArchiveMigrationOnline

Archive migration online

go.mod

@@ -4,6 +4,7 @@ go 1.19
 
 require (
 	github.com/alitto/pond v1.8.3
+	github.com/armon/go-metrics v0.4.1
 	github.com/cockroachdb/pebble v0.0.0-20230819001538-1798fbf5956c
 	github.com/confio/ics23/go v0.9.0
 	github.com/cosmos/iavl v0.21.0-alpha.1.0.20230904092046-df3db2d96583
@@ -45,6 +46,8 @@ require (
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/flatbuffers v1.12.1 // indirect
 	github.com/google/uuid v1.3.0 // indirect
+	github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
+	github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d // indirect
 	github.com/inconshreveable/mousetrap v1.0.1 // indirect
 	github.com/jmhodges/levigo v1.0.0 // indirect
 	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 // indirect

go.sum

@@ -125,6 +125,8 @@ github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
 github.com/armon/go-metrics v0.3.9/go.mod h1:4O98XIr/9W0sxpJ8UaYkvjk10Iff7SnFrb4QAOwNTFc=
 github.com/armon/go-metrics v0.3.10/go.mod h1:4O98XIr/9W0sxpJ8UaYkvjk10Iff7SnFrb4QAOwNTFc=
+github.com/armon/go-metrics v0.4.1 h1:hR91U9KYmb6bLBYLQjyM+3j+rcd/UhE+G78SFnF8gJA=
+github.com/armon/go-metrics v0.4.1/go.mod h1:E6amYzXo6aW1tqzoZGT755KkbgrJsSdpwZ+3JqfkOG4=
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/armon/go-radix v1.0.0/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/ashanbrown/forbidigo v1.3.0/go.mod h1:vVW7PEdqEFqapJe95xHkTfB1+XvZXBFg8t0sG2FIxmI=
@@ -547,6 +549,7 @@ github.com/hashicorp/go-hclog v0.16.2/go.mod h1:whpDNt7SSdeAju8AWKIWsul05p54N/39
 github.com/hashicorp/go-hclog v1.0.0/go.mod h1:whpDNt7SSdeAju8AWKIWsul05p54N/39EeqMAyrmvFQ=
 github.com/hashicorp/go-hclog v1.2.0/go.mod h1:whpDNt7SSdeAju8AWKIWsul05p54N/39EeqMAyrmvFQ=
 github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
+github.com/hashicorp/go-immutable-radix v1.3.1 h1:DKHmCUm2hRBK510BaiZlwvpD40f8bJFeZnpfm2KLowc=
 github.com/hashicorp/go-immutable-radix v1.3.1/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
 github.com/hashicorp/go-msgpack v0.5.3/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM=
 github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
@@ -557,13 +560,16 @@ github.com/hashicorp/go-rootcerts v1.0.2/go.mod h1:pqUvnprVnM5bf7AOirdbb01K4ccR3
 github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU=
 github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=
 github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
+github.com/hashicorp/go-uuid v1.0.1 h1:fv1ep09latC32wFoVwnqcnKJGnMSdBanPczbHAYm1BE=
 github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
 github.com/hashicorp/go-version v1.2.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/hashicorp/go-version v1.2.1/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/hashicorp/go-version v1.4.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
+github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d h1:dg1dEPuWpEqDnvIw251EVy4zlP8gWbsGj4BsUKCRpYs=
+github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=
 github.com/hashicorp/mdns v1.0.1/go.mod h1:4gW7WsVCke5TE7EPeYliwHlRUyBtfCwuFwuMg2DmyNY=
@@ -843,6 +849,7 @@ github.com/otiai10/curr v1.0.0/go.mod h1:LskTG5wDwr8Rs+nNQ+1LlxRjAtTZZjtJW4rMXl6
 github.com/otiai10/mint v1.3.0/go.mod h1:F5AjcsTsWUqX+Na9fpHb52P8pcRX2CI6A3ctIT91xUo=
 github.com/otiai10/mint v1.3.1/go.mod h1:/yxELlJQ0ufhjUwhshSj+wFjZ78CnZ48/1wtmBH1OTc=
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
+github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0MwY=
 github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=

ss/pebbledb/db.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/armon/go-metrics"
 	"github.com/cockroachdb/pebble"
 	"github.com/cockroachdb/pebble/bloom"
 	errorutils "github.com/sei-protocol/sei-db/common/errors"
@@ -25,12 +26,14 @@ import (
 const (
 	VersionSize = 8
 
-	PrefixStore        = "s/k:"
-	LenPrefixStore     = 4
-	StorePrefixTpl     = "s/k:%s/"   // s/k:<storeKey>
-	latestVersionKey   = "s/_latest" // NB: latestVersionKey key must be lexically smaller than StorePrefixTpl
-	earliestVersionKey = "s/_earliest"
-	tombstoneVal       = "TOMBSTONE"
+	PrefixStore                  = "s/k:"
+	LenPrefixStore               = 4
+	StorePrefixTpl               = "s/k:%s/"   // s/k:<storeKey>
+	latestVersionKey             = "s/_latest" // NB: latestVersionKey key must be lexically smaller than StorePrefixTpl
+	earliestVersionKey           = "s/_earliest"
+	latestMigratedKeyMetadata    = "s/_latestMigratedKey"
+	latestMigratedModuleMetadata = "s/_latestMigratedModule"
+	tombstoneVal                 = "TOMBSTONE"
 
 	// TODO: Make configurable
 	ImportCommitBatchSize = 10000
@@ -155,7 +158,9 @@ func (db *Database) Close() error {
 func (db *Database) SetLatestVersion(version int64) error {
 	var ts [VersionSize]byte
 	binary.LittleEndian.PutUint64(ts[:], uint64(version))
-	return db.storage.Set([]byte(latestVersionKey), ts[:], defaultWriteOpts)
+	err := db.storage.Set([]byte(latestVersionKey), ts[:], defaultWriteOpts)
+	fmt.Printf("SetLatestVersion: version=%d, err=%v, latestVersionKey=%s\n", version, err, latestVersionKey)
+	return err
 }
 
 func (db *Database) GetLatestVersion() (int64, error) {
@@ -210,6 +215,42 @@ func retrieveEarliestVersion(db *pebble.DB) (int64, error) {
 	return int64(binary.LittleEndian.Uint64(bz)), closer.Close()
 }
 
+// SetLatestKey sets the latest key processed during migration.
+func (db *Database) SetLatestMigratedKey(key []byte) error {
+	return db.storage.Set([]byte(latestMigratedKeyMetadata), key, defaultWriteOpts)
+}
+
+// GetLatestKey retrieves the latest key processed during migration.
+func (db *Database) GetLatestMigratedKey() ([]byte, error) {
+	bz, closer, err := db.storage.Get([]byte(latestMigratedKeyMetadata))
+	if err != nil {
+		if errors.Is(err, pebble.ErrNotFound) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	defer closer.Close()
+	return bz, nil
+}
+
+// SetLatestModule sets the latest module processed during migration.
+func (db *Database) SetLatestMigratedModule(module string) error {
+	return db.storage.Set([]byte(latestMigratedModuleMetadata), []byte(module), defaultWriteOpts)
+}
+
+// GetLatestModule retrieves the latest module processed during migration.
+func (db *Database) GetLatestMigratedModule() (string, error) {
+	bz, closer, err := db.storage.Get([]byte(latestMigratedModuleMetadata))
+	if err != nil {
+		if errors.Is(err, pebble.ErrNotFound) {
+			return "", nil
+		}
+		return "", err
+	}
+	defer closer.Close()
+	return string(bz), nil
+}
+
 func (db *Database) Has(storeKey string, version int64, key []byte) (bool, error) {
 	if version < db.earliestVersion {
 		return false, nil
@@ -341,7 +382,7 @@ func (db *Database) writeAsyncInBackground() {
 // We add a heuristic to skip over a module's keys during pruning if it hasn't been updated
 // since the last time pruning occurred.
 // NOTE: There is a rare case when a module's keys are skipped during pruning even though
-// it has been updated. This occurs when that module is updated in between pruning runs, the node after is restarted.
+// it has been updated. This occurs when that module's keys are updated in between pruning runs, the node after is restarted.
 // This is not a large issue given the next time that module is updated, it will be properly pruned thereafter.
 func (db *Database) Prune(version int64) error {
 	earliestVersion := version + 1 // we increment by 1 to include the provided version
@@ -556,29 +597,63 @@ func (db *Database) RawImport(ch <-chan types.RawSnapshotNode) error {
 		}
 
 		var counter int
+		var latestKey []byte // store the latest key from the batch
+		var latestModule string
 		for entry := range ch {
 			err := batch.Set(entry.StoreKey, entry.Key, entry.Value, entry.Version)
 			if err != nil {
 				panic(err)
 			}
 
+			latestKey = entry.Key // track the latest key
+			latestModule = entry.StoreKey
 			counter++
+
 			if counter%ImportCommitBatchSize == 0 {
+				startTime := time.Now()
+
+				// Commit the batch and record the latest key as metadata
 				if err := batch.Write(); err != nil {
 					panic(err)
 				}
 
+				// Persist the latest key in the metadata
+				if err := db.SetLatestMigratedKey(latestKey); err != nil {
+					panic(err)
+				}
+
+				if err := db.SetLatestMigratedModule(latestModule); err != nil {
+					panic(err)
+				}
+
+				if counter%1000000 == 0 {
+					fmt.Printf("Time taken to write batch counter %d: %v\n", counter, time.Since(startTime))
+					metrics.IncrCounterWithLabels([]string{"sei", "migration", "nodes_imported"}, float32(1000000), []metrics.Label{
+						{Name: "module", Value: latestModule},
+					})
+				}
+
 				batch, err = NewRawBatch(db.storage)
 				if err != nil {
 					panic(err)
 				}
 			}
 		}
 
+		// Final batch write
 		if batch.Size() > 0 {
 			if err := batch.Write(); err != nil {
 				panic(err)
 			}
+
+			// Persist the final latest key
+			if err := db.SetLatestMigratedKey(latestKey); err != nil {
+				panic(err)
+			}
+
+			if err := db.SetLatestMigratedModule(latestModule); err != nil {
+				panic(err)
+			}
 		}
 	}
 

ss/types/store.go

@@ -18,6 +18,10 @@ type StateStore interface {
 	SetLatestVersion(version int64) error
 	GetEarliestVersion() (int64, error)
 	SetEarliestVersion(version int64) error
+	GetLatestMigratedKey() ([]byte, error)
+	SetLatestMigratedKey(key []byte) error
+	GetLatestMigratedModule() (string, error)
+	SetLatestMigratedModule(module string) error
 
 	// ApplyChangeset Persist the change set of a block,
 	// the `changeSet` should be ordered by (storeKey, key),