mirror/erigon-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/erigon-pulse.git synced 2024-12-22 03:30:37 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

db migration to reset blocks (#4389)

Browse Source 
* save

* save

* save

* save

* save

* save

* Update reset_blocks.go

* Not to remove too many tx lookup files

* Fix truncate blocks and add reset txlookup

* Fix bodies

* Fix nil pointer

Co-authored-by: ledgerwatch <akhounov@gmail.com>
Co-authored-by: Alexey Sharp <alexeysharp@Alexeys-iMac.local>
Co-authored-by: Alex Sharp <alexsharp@Alexs-MacBook-Pro.local>
This commit is contained in:
Alex Sharov 2022-06-08 01:59:14 +07:00 committed by GitHub
parent 07df1eb598
commit d7d698f565
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 150 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
core/rawdb/accessors_chain.go
View File

@ -1332,10 +1332,13 @@ func TruncateBlocks(ctx context.Context, tx kv.RwTx, blockFrom uint64) error {
return err
}
}
if err := tx.Delete(kv.Headers, k, nil); err != nil {
// Copying k because otherwise the same memory will be reused
// for the next key and Delete below will end up deleting 1 more record than required
kCopy := common.CopyBytes(k)
if err := tx.Delete(kv.Headers, kCopy, nil); err != nil {
return err
}
if err := tx.Delete(kv.BlockBody, k, nil); err != nil {
if err := tx.Delete(kv.BlockBody, kCopy, nil); err != nil {
return err
}

9
eth/stagedsync/stage_bodies.go
View File

@ -58,8 +58,10 @@ func BodiesForward(
test bool, // Set to true in tests, allows the stage to fail rather than wait indefinitely
firstCycle bool,
) error {
var doUpdate bool
if cfg.snapshots != nil && s.BlockNumber < cfg.snapshots.BlocksAvailable() {
s.BlockNumber = cfg.snapshots.BlocksAvailable()
doUpdate = true
}
var d1, d2, d3, d4, d5, d6 time.Duration
@ -74,6 +76,13 @@ func BodiesForward(
}
timeout := cfg.timeout
// this update is required, because cfg.bd.UpdateFromDb(tx) below reads it and initialises requestedLow accordingly
// if not done, it will cause downloading from block 1
if doUpdate {
if err := s.Update(tx, s.BlockNumber); err != nil {
return err
}
}
// This will update bd.maxProgress
if _, _, _, err = cfg.bd.UpdateFromDb(tx); err != nil {
return err

2
eth/stagedsync/stage_call_traces.go
View File

@ -185,7 +185,7 @@ func promoteCallTraces(logPrefix string, tx kv.RwTx, startBlock, endBlock uint64
case <-logEvery.C:
var m runtime.MemStats
libcommon.ReadMemStats(&m)
log.Info(fmt.Sprintf("[%s] Pruning call trace intermediate table", logPrefix), "number", blockNum,
log.Info(fmt.Sprintf("[%s] Pruning call trace table", logPrefix), "number", blockNum,
"alloc", libcommon.ByteCount(m.Alloc), "sys", libcommon.ByteCount(m.Sys))
}
if err = traceCursor.DeleteCurrentDuplicates(); err != nil {

3
migrations/db_schema_version.go
View File

@ -4,11 +4,12 @@ import (
"context"
"github.com/ledgerwatch/erigon-lib/kv"
"github.com/ledgerwatch/erigon/node/nodecfg/datadir"
)
var dbSchemaVersion5 = Migration{
Name: "db_schema_version5",
Up: func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) (err error) {
Up: func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
tx, err := db.BeginRw(context.Background())
if err != nil {
return err

10
migrations/migrations.go
View File

@ -10,6 +10,7 @@ import (
"github.com/ledgerwatch/erigon-lib/kv"
"github.com/ledgerwatch/erigon/common"
"github.com/ledgerwatch/erigon/eth/stagedsync/stages"
"github.com/ledgerwatch/erigon/node/nodecfg/datadir"
"github.com/ledgerwatch/log/v3"
"github.com/ugorji/go/codec"
)
@ -33,6 +34,7 @@ var migrations = map[kv.Label][]Migration{
kv.ChainDB: {
dbSchemaVersion5,
txsBeginEnd,
resetBlocks,
},
kv.TxPoolDB: {},
kv.SentryDB: {},
@ -41,7 +43,7 @@ var migrations = map[kv.Label][]Migration{
type Callback func(tx kv.RwTx, progress []byte, isDone bool) error
type Migration struct {
Name string
Up func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) error
Up func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) error
}
var (
@ -151,10 +153,11 @@ func (m *Migrator) VerifyVersion(db kv.RwDB) error {
return nil
}
func (m *Migrator) Apply(db kv.RwDB, datadir string) error {
func (m *Migrator) Apply(db kv.RwDB, dataDir string) error {
if len(m.Migrations) == 0 {
return nil
}
dirs := datadir.New(dataDir)
var applied map[string][]byte
if err := db.View(context.Background(), func(tx kv.Tx) error {
@ -198,7 +201,8 @@ func (m *Migrator) Apply(db kv.RwDB, datadir string) error {
return fmt.Errorf("migrator.Apply: %w", err)
}
if err := v.Up(db, filepath.Join(datadir, "migrations", v.Name), progress, func(tx kv.RwTx, key []byte, isDone bool) error {
dirs.Tmp = filepath.Join(dirs.DataDir, "migrations", v.Name)
if err := v.Up(db, dirs, progress, func(tx kv.RwTx, key []byte, isDone bool) error {
if !isDone {
if key != nil {
if err := tx.Put(kv.Migrations, []byte("_progress_"+v.Name), key); err != nil {

19
migrations/migrations_test.go
View File

@ -8,6 +8,7 @@ import (
"github.com/ledgerwatch/erigon-lib/kv"
"github.com/ledgerwatch/erigon-lib/kv/memdb"
"github.com/ledgerwatch/erigon/eth/stagedsync/stages"
"github.com/ledgerwatch/erigon/node/nodecfg/datadir"
"github.com/stretchr/testify/require"
)
@ -17,7 +18,7 @@ func TestApplyWithInit(t *testing.T) {
m := []Migration{
{
"one",
func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) (err error) {
func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
tx, err := db.BeginRw(context.Background())
if err != nil {
return err
@ -32,7 +33,7 @@ func TestApplyWithInit(t *testing.T) {
},
{
"two",
func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) (err error) {
func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
tx, err := db.BeginRw(context.Background())
if err != nil {
return err
@ -81,14 +82,14 @@ func TestApplyWithoutInit(t *testing.T) {
m := []Migration{
{
"one",
func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) (err error) {
func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
t.Fatal("shouldn't been executed")
return nil
},
},
{
"two",
func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) (err error) {
func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
tx, err := db.BeginRw(context.Background())
if err != nil {
return err
@ -145,7 +146,7 @@ func TestWhenNonFirstMigrationAlreadyApplied(t *testing.T) {
m := []Migration{
{
"one",
func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) (err error) {
func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
tx, err := db.BeginRw(context.Background())
if err != nil {
return err
@ -160,7 +161,7 @@ func TestWhenNonFirstMigrationAlreadyApplied(t *testing.T) {
},
{
"two",
func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) (err error) {
func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
t.Fatal("shouldn't been executed")
return nil
},
@ -226,7 +227,7 @@ func TestValidation(t *testing.T) {
m := []Migration{
{
Name: "repeated_name",
Up: func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) (err error) {
Up: func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
tx, err := db.BeginRw(context.Background())
if err != nil {
return err
@ -241,7 +242,7 @@ func TestValidation(t *testing.T) {
},
{
Name: "repeated_name",
Up: func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) (err error) {
Up: func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
tx, err := db.BeginRw(context.Background())
if err != nil {
return err
@ -275,7 +276,7 @@ func TestCommitCallRequired(t *testing.T) {
m := []Migration{
{
Name: "one",
Up: func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) (err error) {
Up: func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
//don't call BeforeCommit
return nil
},

69
migrations/reset_blocks.go Normal file
View File

@ -0,0 +1,69 @@
package migrations
import (
"context"
"github.com/ledgerwatch/erigon-lib/kv"
"github.com/ledgerwatch/erigon/core/rawdb"
"github.com/ledgerwatch/erigon/core/rawdb/rawdbreset"
"github.com/ledgerwatch/erigon/node/nodecfg/datadir"
"github.com/ledgerwatch/erigon/turbo/snapshotsync/snap"
"github.com/ledgerwatch/log/v3"
)
var resetBlocks = Migration{
Name: "reset_blocks_3",
Up: func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
tx, err := db.BeginRw(context.Background())
if err != nil {
return err
}
defer tx.Rollback()
enabled, err := snap.Enabled(tx)
if err != nil {
return err
}
if !enabled {
if err := BeforeCommit(tx, nil, true); err != nil {
return err
}
return
}
genesisBlock := rawdb.ReadHeaderByNumber(tx, 0)
if genesisBlock == nil {
if err := BeforeCommit(tx, nil, true); err != nil {
return err
}
return nil
}
chainConfig, err := rawdb.ReadChainConfig(tx, genesisBlock.Hash())
if err != nil {
return err
}
log.Warn("NOTE: this migration will remove recent blocks (and senders) to fix several recent bugs. Your node will re-download last ~400K blocks, should not take very long")
if err := snap.RemoveNonPreverifiedFiles(chainConfig.ChainName, dirs.Snap); err != nil {
return err
}
if err := rawdbreset.ResetBlocks(tx); err != nil {
return err
}
if err := rawdbreset.ResetSenders(tx); err != nil {
return err
}
if err := rawdbreset.ResetTxLookup(tx); err != nil {
return err
}
// This migration is no-op, but it forces the migration mechanism to apply it and thus write the DB schema version info
if err := BeforeCommit(tx, nil, true); err != nil {
return err
}
return tx.Commit()
},
}

3
migrations/txs_begin_end.go
View File

@ -15,6 +15,7 @@ import (
"github.com/ledgerwatch/erigon/core/rawdb"
"github.com/ledgerwatch/erigon/core/types"
"github.com/ledgerwatch/erigon/eth/stagedsync/stages"
"github.com/ledgerwatch/erigon/node/nodecfg/datadir"
"github.com/ledgerwatch/erigon/rlp"
"github.com/ledgerwatch/log/v3"
)
@ -25,7 +26,7 @@ var ErrTxsBeginEndNoMigration = fmt.Errorf("in this Erigon version DB format was
var txsBeginEnd = Migration{
Name: "txs_begin_end",
Up: func(db kv.RwDB, tmpdir string, progress []byte, BeforeCommit Callback) (err error) {
Up: func(db kv.RwDB, dirs datadir.Dirs, progress []byte, BeforeCommit Callback) (err error) {
logEvery := time.NewTicker(10 * time.Second)
defer logEvery.Stop()

6
turbo/snapshotsync/block_snapshots.go
View File

@ -388,7 +388,7 @@ func (s *RoSnapshots) Reopen() error {
s.Txs.lock.Lock()
defer s.Txs.lock.Unlock()
s.closeSegmentsLocked()
files, err := segments2(s.dir)
files, err := segments(s.dir)
if err != nil {
return err
}
@ -483,7 +483,7 @@ func (s *RoSnapshots) ReopenSegments() error {
s.Txs.lock.Lock()
defer s.Txs.lock.Unlock()
s.closeSegmentsLocked()
files, err := segments2(s.dir)
files, err := segments(s.dir)
if err != nil {
return err
}
@ -830,7 +830,7 @@ func noOverlaps(in []snap.FileInfo) (res []snap.FileInfo) {
return res
}
func segments2(dir string) (res []snap.FileInfo, err error) {
func segments(dir string) (res []snap.FileInfo, err error) {
list, err := snap.Segments(dir)
if err != nil {
return nil, err

42
turbo/snapshotsync/snap/files.go
View File

@ -9,6 +9,7 @@ import (
"strconv"
"strings"
"github.com/ledgerwatch/erigon/turbo/snapshotsync/snapshothashes"
"golang.org/x/exp/slices"
)
@ -208,3 +209,44 @@ func ParseDir(dir string) (res []FileInfo, err error) {
return res, nil
}
func RemoveNonPreverifiedFiles(chainName, snapDir string) error {
preverified := snapshothashes.KnownConfig(chainName).Preverified
keep := map[string]struct{}{}
for _, p := range preverified {
ext := filepath.Ext(p.Name)
withoutExt := p.Name[0 : len(p.Name)-len(ext)]
keep[withoutExt] = struct{}{}
}
list, err := Segments(snapDir)
if err != nil {
return err
}
for _, f := range list {
_, fname := filepath.Split(f.Path)
ext := filepath.Ext(fname)
withoutExt := fname[0 : len(fname)-len(ext)]
if _, ok := keep[withoutExt]; !ok {
_ = os.Remove(f.Path)
} else {
if f.T == Transactions {
idxPath := IdxFileName(f.From, f.To, Transactions2Block.String())
idxExt := filepath.Ext(idxPath)
keep[idxPath[0:len(idxPath)-len(idxExt)]] = struct{}{}
}
}
}
list, err = IdxFiles(snapDir)
if err != nil {
return err
}
for _, f := range list {
_, fname := filepath.Split(f.Path)
ext := filepath.Ext(fname)
withoutExt := fname[0 : len(fname)-len(ext)]
if _, ok := keep[withoutExt]; !ok {
_ = os.Remove(f.Path)
}
}
return nil
}