erigon22: working PlainState unwind (#5051)

cmd/integration/commands/stages.go

@@ -642,17 +642,7 @@ func stageExec(db kv.RwDB, ctx context.Context) error {
 	must(sync.SetCurrentStage(stages.Execution))
 	dirs := datadir.New(datadirCli)
 
-	var exec22 bool
 	var err error
-	if err := db.View(context.Background(), func(tx kv.Tx) error {
-		exec22, err = rawdb.HistoryV2.Enabled(tx)
-		if err != nil {
-			return err
-		}
-		return nil
-	}); err != nil {
-		panic(err)
-	}
 	if reset {
 		if historyV2 {
 			dir.Recreate(path.Join(dirs.DataDir, "agg22"))
@@ -686,7 +676,7 @@ func stageExec(db kv.RwDB, ctx context.Context) error {
 	genesis := core.DefaultGenesisBlockByChainName(chain)
 	cfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, nil, chainConfig, engine, vmConfig, nil,
 		/*stateStream=*/ false,
-		/*badBlockHalt=*/ false, exec22, dirs, getBlockReader(db), nil, genesis, 1)
+		/*badBlockHalt=*/ false, historyV2, dirs, getBlockReader(db), nil, genesis, 1)
 	if unwind > 0 {
 		u := sync.NewUnwindState(stages.Execution, s.BlockNumber-unwind, s.BlockNumber)
 		err := stagedsync.UnwindExecutionStage(u, s, nil, ctx, cfg, false)

common/changeset/storage_changeset.go

@@ -101,7 +101,7 @@ func RewindData(db kv.Tx, timestampSrc, timestampDst uint64, changes *etl.Collec
 }
 
 func walkAndCollect(collectorFunc func([]byte, []byte) error, db kv.Tx, bucket string, timestampDst, timestampSrc uint64, quit <-chan struct{}) error {
-	return ForRange(db, bucket, timestampDst, timestampSrc+1, func(_ uint64, k, v []byte) error {
+	return ForRange(db, bucket, timestampDst, timestampSrc+1, func(bl uint64, k, v []byte) error {
 		if err := libcommon.Stopped(quit); err != nil {
 			return err
 		}

core/state/history_reader_22.go

@@ -67,6 +67,17 @@ func (hr *HistoryReader22) ReadAccountData(address common.Address) (*accounts.Ac
 		return nil, nil
 	}
 	var a accounts.Account
+	if err := deserialise2(&a, enc); err != nil {
+		return nil, fmt.Errorf("ReadAccountData(%x): %w", address, err)
+	}
+
+	if hr.trace {
+		fmt.Printf("ReadAccountData [%x] => [nonce: %d, balance: %d, codeHash: %x]\n", address, a.Nonce, &a.Balance, a.CodeHash)
+	}
+	return &a, nil
+}
+
+func deserialise2(a *accounts.Account, enc []byte) error {
 	a.Reset()
 	pos := 0
 	nonceBytes := int(enc[pos])
@@ -88,17 +99,14 @@ func (hr *HistoryReader22) ReadAccountData(address common.Address) (*accounts.Ac
 		pos += codeHashBytes
 	}
 	if pos >= len(enc) {
-		fmt.Printf("panic ReadAccountData(%x)=>[%x]\n", address, enc)
+		return fmt.Errorf("deserialse2: %d >= %d ", pos, len(enc))
 	}
 	incBytes := int(enc[pos])
 	pos++
 	if incBytes > 0 {
 		a.Incarnation = bytesToUint64(enc[pos : pos+incBytes])
 	}
-	if hr.trace {
-		fmt.Printf("ReadAccountData [%x] => [nonce: %d, balance: %d, codeHash: %x]\n", address, a.Nonce, &a.Balance, a.CodeHash)
-	}
-	return &a, nil
+	return nil
 }
 
 func (hr *HistoryReader22) ReadAccountStorage(address common.Address, incarnation uint64, key *common.Hash) ([]byte, error) {

core/state/rw22.go

@@ -3,6 +3,7 @@ package state
 import (
 	"bytes"
 	"container/heap"
+	"context"
 	"encoding/binary"
 	"fmt"
 	"math/bits"
@@ -12,7 +13,8 @@ import (
 	"github.com/google/btree"
 	"github.com/holiman/uint256"
 	libcommon "github.com/ledgerwatch/erigon-lib/common"
-	"github.com/ledgerwatch/erigon-lib/common/dbg"
+	"github.com/ledgerwatch/erigon-lib/common/length"
+	"github.com/ledgerwatch/erigon-lib/etl"
 	"github.com/ledgerwatch/erigon-lib/kv"
 	libstate "github.com/ledgerwatch/erigon-lib/state"
 	"github.com/ledgerwatch/erigon/common"
@@ -20,6 +22,7 @@ import (
 	"github.com/ledgerwatch/erigon/core/types"
 	"github.com/ledgerwatch/erigon/core/types/accounts"
 	"github.com/ledgerwatch/erigon/params"
+	"github.com/ledgerwatch/erigon/turbo/shards"
 )
 
 // ReadWriteSet contains ReadSet, WriteSet and BalanceIncrease of a transaction,
@@ -110,10 +113,6 @@ func NewState22() *State22 {
 }
 
 func (rs *State22) put(table string, key, val []byte) {
-	if table == kv.PlainState {
-		fmt.Printf("table: %s, %s\n", table, dbg.Stack())
-	}
-
 	t, ok := rs.changes[table]
 	if !ok {
 		t = btree.NewG[StateItem](32, stateItemLess)
@@ -479,6 +478,92 @@ func (rs *State22) Apply(emptyRemoval bool, roTx kv.Tx, txTask *TxTask, agg *lib
 	return nil
 }
 
+func recoverCodeHashPlain(acc *accounts.Account, db kv.Tx, key []byte) {
+	var address common.Address
+	copy(address[:], key)
+	if acc.Incarnation > 0 && acc.IsEmptyCodeHash() {
+		if codeHash, err2 := db.GetOne(kv.PlainContractCode, dbutils.PlainGenerateStoragePrefix(address[:], acc.Incarnation)); err2 == nil {
+			copy(acc.CodeHash[:], codeHash)
+		}
+	}
+}
+
+func (rs *State22) Unwind(ctx context.Context, tx kv.RwTx, txUnwindTo uint64, agg *libstate.Aggregator22, accumulator *shards.Accumulator) error {
+	var currentInc uint64
+	if err := agg.Unwind(ctx, txUnwindTo, func(k, v []byte, table etl.CurrentTableReader, next etl.LoadNextFunc) error {
+		if len(k) == 20 {
+			if len(v) > 0 {
+				var acc accounts.Account
+				if err := deserialise2(&acc, v); err != nil {
+					return fmt.Errorf("%w, %x", err, v)
+				}
+				currentInc = acc.Incarnation
+				// Fetch the code hash
+				recoverCodeHashPlain(&acc, tx, k)
+				var address common.Address
+				copy(address[:], k)
+
+				// cleanup contract code bucket
+				original, err := NewPlainStateReader(tx).ReadAccountData(address)
+				if err != nil {
+					return fmt.Errorf("read account for %x: %w", address, err)
+				}
+				if original != nil {
+					// clean up all the code incarnations original incarnation and the new one
+					for incarnation := original.Incarnation; incarnation > acc.Incarnation && incarnation > 0; incarnation-- {
+						err = tx.Delete(kv.PlainContractCode, dbutils.PlainGenerateStoragePrefix(address[:], incarnation))
+						if err != nil {
+							return fmt.Errorf("writeAccountPlain for %x: %w", address, err)
+						}
+					}
+				}
+
+				newV := make([]byte, acc.EncodingLengthForStorage())
+				acc.EncodeForStorage(newV)
+				if accumulator != nil {
+					accumulator.ChangeAccount(address, acc.Incarnation, newV)
+				}
+				if err := next(k, k, newV); err != nil {
+					return err
+				}
+			} else {
+				if accumulator != nil {
+					var address common.Address
+					copy(address[:], k)
+					accumulator.DeleteAccount(address)
+				}
+				if err := next(k, k, nil); err != nil {
+					return err
+				}
+			}
+			return nil
+		}
+		if accumulator != nil {
+			var address common.Address
+			var incarnation uint64
+			var location common.Hash
+			copy(address[:], k[:length.Addr])
+			incarnation = binary.BigEndian.Uint64(k[length.Addr:])
+			copy(location[:], k[length.Addr+length.Incarnation:])
+			accumulator.ChangeStorage(address, incarnation, location, common.CopyBytes(v))
+		}
+		newKeys := dbutils.PlainGenerateCompositeStorageKey(k[20:], currentInc, k[20:])
+		if len(v) > 0 {
+			if err := next(k, newKeys, v); err != nil {
+				return err
+			}
+		} else {
+			if err := next(k, newKeys, nil); err != nil {
+				return err
+			}
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	return nil
+}
+
 func (rs *State22) DoneCount() uint64 {
 	rs.lock.RLock()
 	defer rs.lock.RUnlock()

eth/stagedsync/stage_execute.go

@@ -395,9 +395,42 @@ func UnwindExec22(u *UnwindState, s *StageState, tx kv.RwTx, ctx context.Context
 
 	agg.SetTx(tx)
 	agg.SetTxNum(txNums[prevStageProgress])
-	if err := agg.Unwind(txNums[u.UnwindPoint]); err != nil {
+	rs := state.NewState22()
+	if err := rs.Unwind(ctx, tx, txNums[u.UnwindPoint], agg, cfg.accumulator); err != nil {
 		return err
 	}
+	if err := rs.Flush(tx); err != nil {
+		return err
+	}
+
+	/*
+		if err := rawdb.TruncateReceipts(tx, u.UnwindPoint+1); err != nil {
+			return fmt.Errorf("truncate receipts: %w", err)
+		}
+		if err := rawdb.TruncateBorReceipts(tx, u.UnwindPoint+1); err != nil {
+			return fmt.Errorf("truncate bor receipts: %w", err)
+		}
+		if err := rawdb.DeleteNewerEpochs(tx, u.UnwindPoint+1); err != nil {
+			return fmt.Errorf("delete newer epochs: %w", err)
+		}
+
+		// Truncate CallTraceSet
+		keyStart := dbutils.EncodeBlockNumber(u.UnwindPoint + 1)
+		c, err := tx.RwCursorDupSort(kv.CallTraceSet)
+		if err != nil {
+			return err
+		}
+		defer c.Close()
+		for k, _, err := c.Seek(keyStart); k != nil; k, _, err = c.NextNoDup() {
+			if err != nil {
+				return err
+			}
+			err = c.DeleteCurrentDuplicates()
+			if err != nil {
+				return err
+			}
+		}
+	*/
 	if err = u.Done(tx); err != nil {
 		return err
 	}

go.mod

@@ -3,7 +3,7 @@ module github.com/ledgerwatch/erigon
 go 1.18
 
 require (
-	github.com/ledgerwatch/erigon-lib v0.0.0-20220813104409-4999fb010739
+	github.com/ledgerwatch/erigon-lib v0.0.0-20220814105353-404276494a0c
 	github.com/ledgerwatch/erigon-snapshot v1.0.1-0.20220809023834-6309df4da4b1
 	github.com/ledgerwatch/log/v3 v3.4.1
 	github.com/ledgerwatch/secp256k1 v1.0.0

go.sum

@@ -390,8 +390,8 @@ github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v0.0.0-20170224010052-a616ab194758 h1:0D5M2HQSGD3PYPwICLl+/9oulQauOuETfgFvhBDffs0=
 github.com/leanovate/gopter v0.2.9 h1:fQjYxZaynp97ozCzfOyOuAGOU4aU/z37zf/tOujFk7c=
 github.com/leanovate/gopter v0.2.9/go.mod h1:U2L/78B+KVFIx2VmW6onHJQzXtFb+p5y3y2Sh+Jxxv8=
-github.com/ledgerwatch/erigon-lib v0.0.0-20220813104409-4999fb010739 h1:J0X0kujYuZKkzbHw/IamcjC51nNPqLrrc+92TBrWuWk=
-github.com/ledgerwatch/erigon-lib v0.0.0-20220813104409-4999fb010739/go.mod h1:Sb+16YoG+izjT/6ZPxUZixa2gI2jLbi1rrxBCtc2QKg=
+github.com/ledgerwatch/erigon-lib v0.0.0-20220814105353-404276494a0c h1:ctv3n6W5p0oDnxqpBRSY0HlwnHjly9bdL+x4JqgGWio=
+github.com/ledgerwatch/erigon-lib v0.0.0-20220814105353-404276494a0c/go.mod h1:Sb+16YoG+izjT/6ZPxUZixa2gI2jLbi1rrxBCtc2QKg=
 github.com/ledgerwatch/erigon-snapshot v1.0.1-0.20220809023834-6309df4da4b1 h1:qRIJu6cs6fbI8L52DSdPF27j3sOrEriXz1zQSuQvYpA=
 github.com/ledgerwatch/erigon-snapshot v1.0.1-0.20220809023834-6309df4da4b1/go.mod h1:3AuPxZc85jkehh/HA9h8gabv5MSi3kb/ddtzBsTVJFo=
 github.com/ledgerwatch/log/v3 v3.4.1 h1:/xGwlVulXnsO9Uq+tzaExc8OWmXXHU0dnLalpbnY5Bc=

turbo/trie/trie_root.go

@@ -636,10 +636,6 @@ func (r *RootHashAggregator) genStructAccount() error {
 		if r.hc == nil {
 			return nil
 		}
-
-		//if bytes.HasPrefix(keyHex, common.FromHex("060e")) {
-		//	fmt.Printf("collect: %x,%b,%b, del:%t\n", keyHex, hasHash, hasTree, hashes == nil)
-		//}
 		return r.hc(keyHex, hasState, hasTree, hasHash, hashes, rootHash)
 	}, data, r.groups, r.hasTree, r.hasHash,
 		false,