Integration check history (#1040)

* integration check history

* reset tx pool

* reset tx pool

* reset tx pool

cmd/integration/commands/reset_state.go

@@ -102,6 +102,9 @@ func resetState(_ context.Context) error {
 	if err := resetTxLookup(db); err != nil {
 		return err
 	}
+	if err := resetTxPool(db); err != nil {
+		return err
+	}
 
 	// set genesis after reset all buckets
 	if _, _, err := core.DefaultGenesisBlock().CommitGenesisState(db, false); err != nil {
@@ -193,6 +196,18 @@ func resetTxLookup(db *ethdb.ObjectDatabase) error {
 
 	return nil
 }
+
+func resetTxPool(db ethdb.Putter) error {
+	if err := stages.SaveStageProgress(db, stages.TxPool, 0, nil); err != nil {
+		return err
+	}
+	if err := stages.SaveStageUnwind(db, stages.TxPool, 0, nil); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func printStages(db *ethdb.ObjectDatabase) error {
 	var err error
 	var progress uint64

cmd/integration/commands/state_stages.go

@@ -3,9 +3,10 @@ package commands
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
-
 	"github.com/ledgerwatch/turbo-geth/cmd/utils"
+	"github.com/ledgerwatch/turbo-geth/common/dbutils"
 
 	"github.com/ledgerwatch/turbo-geth/common"
 	"github.com/ledgerwatch/turbo-geth/common/changeset"
@@ -120,7 +121,8 @@ func syncBySmallSteps(ctx context.Context, chaindata string) error {
 		}
 
 		// All stages forward to `execStage + unwindEvery` block
-		execToBlock := progress(stages.Execution).BlockNumber + unwindEvery
+		execAtBlock := progress(stages.Execution).BlockNumber
+		execToBlock := execAtBlock + unwindEvery
 		if execToBlock > stopAt {
 			execToBlock = stopAt + 1
 			unwind = 0
@@ -146,6 +148,13 @@ func syncBySmallSteps(ctx context.Context, chaindata string) error {
 			delete(expectedStorageChanges, blockN)
 		}
 
+		if err := checkHistory(tx, dbutils.AccountChangeSetBucket, execAtBlock); err != nil {
+			return err
+		}
+		if err := checkHistory(tx, dbutils.StorageChangeSetBucket, execAtBlock); err != nil {
+			return err
+		}
+
 		// Unwind all stages to `execStage - unwind` block
 		if unwind == 0 {
 			continue
@@ -249,3 +258,47 @@ func checkChangeSet(db ethdb.Getter, blockNum uint64, expectedAccountChanges []b
 	}
 	return nil
 }
+
+func checkHistory(db ethdb.Getter, changeSetBucket string, blockNum uint64) error {
+	currentKey := dbutils.EncodeTimestamp(blockNum)
+
+	var walker func([]byte) changeset.Walker
+	if dbutils.AccountChangeSetBucket == changeSetBucket {
+		walker = func(cs []byte) changeset.Walker {
+			return changeset.AccountChangeSetBytes(cs)
+		}
+	}
+
+	if dbutils.StorageChangeSetBucket == changeSetBucket {
+		walker = func(cs []byte) changeset.Walker {
+			return changeset.StorageChangeSetBytes(cs)
+		}
+	}
+
+	vv, ok := changeset.Mapper[changeSetBucket]
+	if !ok {
+		return errors.New("unknown bucket type")
+	}
+
+	if err := db.Walk(changeSetBucket, currentKey, 0, func(k, v []byte) (b bool, e error) {
+		blockNum, _ := dbutils.DecodeTimestamp(k)
+		if err := walker(v).Walk(func(key, val []byte) error {
+			indexBytes, innerErr := db.GetIndexChunk(vv.IndexBucket, key, blockNum)
+			if innerErr != nil {
+				return innerErr
+			}
+
+			index := dbutils.WrapHistoryIndex(indexBytes)
+			if findVal, _, ok := index.Search(blockNum); !ok {
+				return fmt.Errorf("%v,%v,%v", blockNum, findVal, common.Bytes2Hex(key))
+			}
+			return nil
+		}); err != nil {
+			return false, err
+		}
+		return true, nil
+	}); err != nil {
+		return err
+	}
+	return nil
+}