Add new option to allow BlocksReExecutor to commit state to disk

Add new boolean option to BlocksReExecutor called
CommitStateToDisk that will allow BlocksReExecutor.Blocks
range to not only re-executes blocks but it also commit
their state to triedb.

Signed-off-by: Igor Braga <[email protected]>

Author: bragaigor

blocks_reexecutor/blocks_reexecutor.go

@@ -34,6 +34,7 @@ type Config struct {
 	Enable             bool   `koanf:"enable"`
 	Mode               string `koanf:"mode"`
 	Blocks             string `koanf:"blocks"` // Range of blocks to be executed in json format
+	CommitStateToDisk  bool   `koanf:"commit-state-to-disk"`
 	Room               int    `koanf:"room"`
 	MinBlocksPerThread uint64 `koanf:"min-blocks-per-thread"`
 	TrieCleanLimit     int    `koanf:"trie-clean-limit"`
@@ -71,6 +72,7 @@ var DefaultConfig = Config{
 	Mode:               "random",
 	Room:               util.GoMaxProcs(),
 	Blocks:             `[[0,0]]`, // execute from chain start to chain end
+	CommitStateToDisk:  false,
 	MinBlocksPerThread: 0,
 	TrieCleanLimit:     0,
 	ValidateMultiGas:   false,
@@ -81,6 +83,7 @@ var TestConfig = Config{
 	Enable:             true,
 	Mode:               "full",
 	Blocks:             `[[0,0]]`, // execute from chain start to chain end
+	CommitStateToDisk:  false,
 	Room:               util.GoMaxProcs(),
 	TrieCleanLimit:     600,
 	MinBlocksPerThread: 0,
@@ -93,6 +96,7 @@ func ConfigAddOptions(prefix string, f *pflag.FlagSet) {
 	f.Bool(prefix+".enable", DefaultConfig.Enable, "enables re-execution of a range of blocks against historic state")
 	f.String(prefix+".mode", DefaultConfig.Mode, "mode to run the blocks-reexecutor on. Valid modes full and random. full - execute all the blocks in the given range. random - execute a random sample range of blocks with in a given range")
 	f.String(prefix+".blocks", DefaultConfig.Blocks, "json encoded list of block ranges in the form of start and end block numbers in a list of size 2")
+	f.Bool(prefix+".commit-state-to-disk", DefaultConfig.CommitStateToDisk, "if set, blocks-reexecutor not only re-executes blocks but it also commits their state to triedb")
 	f.Int(prefix+".room", DefaultConfig.Room, "number of threads to parallelize blocks re-execution")
 	f.Uint64(prefix+".min-blocks-per-thread", DefaultConfig.MinBlocksPerThread, "minimum number of blocks to execute per thread. When mode is random this acts as the size of random block range sample")
 	f.Int(prefix+".trie-clean-limit", DefaultConfig.TrieCleanLimit, "memory allowance (MB) to use for caching trie nodes in memory")
@@ -110,9 +114,10 @@ type BlocksReExecutor struct {
 	fatalErrChan chan error
 	blocks       [][3]uint64 // start, end and minBlocksPerThread of block ranges
 	mutex        sync.Mutex
+	success      chan struct{}
 }
 
-func New(c *Config, blockchain *core.BlockChain, ethDb ethdb.Database, fatalErrChan chan error) (*BlocksReExecutor, error) {
+func New(c *Config, blockchain *core.BlockChain, ethDb ethdb.Database) (*BlocksReExecutor, error) {
 	if blockchain.TrieDB().Scheme() == rawdb.PathScheme {
 		return nil, errors.New("blocksReExecutor not supported on pathdb")
 	}
@@ -197,7 +202,8 @@ func New(c *Config, blockchain *core.BlockChain, ethDb ethdb.Database, fatalErrC
 		stateFor:     stateForFunc,
 		blocks:       blocks,
 		done:         make(chan struct{}, c.Room),
-		fatalErrChan: fatalErrChan,
+		fatalErrChan: make(chan error, 10),
+		success:      make(chan struct{}),
 		mutex:        sync.Mutex{},
 	}
 	return blocksReExecutor, nil
@@ -260,7 +266,7 @@ func (s *BlocksReExecutor) Impl(ctx context.Context, startBlock, currentBlock, m
 	return currentBlock
 }
 
-func (s *BlocksReExecutor) Start(ctx context.Context, done chan struct{}) {
+func (s *BlocksReExecutor) Start(ctx context.Context) {
 	s.StopWaiter.Start(ctx, s)
 	s.LaunchThread(func(ctx context.Context) {
 		// Using returned value from Impl we can avoid duplicate reexecution of blocks
@@ -273,12 +279,23 @@ func (s *BlocksReExecutor) Start(ctx context.Context, done chan struct{}) {
 				log.Info("BlocksReExecutor successfully completed re-execution of blocks against historic state", "stateAt", blocks[0], "startBlock", blocks[0]+1, "endBlock", blocks[1])
 			}
 		}
-		if done != nil {
-			close(done)
+		if s.success != nil {
+			close(s.success)
 		}
 	})
 }
 
+func (s *BlocksReExecutor) WaitForReExecution(ctx context.Context) error {
+	select {
+	case err := <-s.fatalErrChan:
+		log.Error("shutting BlocksReExecutor down due to fatal error", "err", err)
+		return fmt.Errorf("shutting BlocksReExecutor down due to fatal error %v", err)
+	case <-s.success:
+	}
+
+	return nil
+}
+
 func (s *BlocksReExecutor) StopAndWait() {
 	s.StopWaiter.StopAndWait()
 }
@@ -299,6 +316,14 @@ func (s *BlocksReExecutor) commitStateAndVerify(statedb *state.StateDB, expected
 	if result != expected {
 		return nil, arbitrum.NoopStateRelease, fmt.Errorf("bad root hash expected: %v got: %v", expected, result)
 	}
+
+	if s.config.CommitStateToDisk {
+		err = s.db.TrieDB().Commit(expected, false)
+		if err != nil {
+			return nil, arbitrum.NoopStateRelease, fmt.Errorf("trieDB commit failed in commitStateAndVerify, number %d root %v: %w", blockNumber, expected, err)
+		}
+	}
+
 	sdb, err := state.New(result, s.db)
 	if err == nil {
 		_ = s.db.TrieDB().Reference(result, common.Hash{})

cmd/nitro/nitro.go

@@ -466,27 +466,23 @@ func mainImpl() int {
 		return 1
 	}
 
-	fatalErrChan := make(chan error, 10)
-
 	if nodeConfig.BlocksReExecutor.Enable && l2BlockChain != nil {
 		if !nodeConfig.Init.ThenQuit {
 			log.Error("blocks-reexecutor cannot be enabled without --init.then-quit")
 			return 1
 		}
-		blocksReExecutor, err := blocksreexecutor.New(&nodeConfig.BlocksReExecutor, l2BlockChain, chainDb, fatalErrChan)
+		blocksReExecutor, err := blocksreexecutor.New(&nodeConfig.BlocksReExecutor, l2BlockChain, chainDb)
 		if err != nil {
 			log.Error("error initializing blocksReExecutor", "err", err)
 			return 1
 		}
-		success := make(chan struct{})
-		blocksReExecutor.Start(ctx, success)
+
+		blocksReExecutor.Start(ctx)
 		deferFuncs = append(deferFuncs, func() { blocksReExecutor.StopAndWait() })
-		select {
-		case err := <-fatalErrChan:
-			log.Error("shutting down due to fatal error", "err", err)
+		err = blocksReExecutor.WaitForReExecution(ctx)
+		if err != nil {
 			defer log.Error("shut down due to fatal error", "err", err)
 			return 1
-		case <-success:
 		}
 	}
 
@@ -510,6 +506,8 @@ func mainImpl() int {
 		return 1
 	}
 
+	fatalErrChan := make(chan error, 10)
+
 	var valNode *valnode.ValidationNode
 	if sameProcessValidationNodeEnabled {
 		valNode, err = valnode.CreateValidationNode(

system_tests/blocks_reexecutor_test.go

@@ -2,10 +2,14 @@ package arbtest
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/trie"
 
 	blocksreexecutor "github.com/offchainlabs/nitro/blocks_reexecutor"
 )
@@ -36,7 +40,6 @@ func testBlocksReExecutorModes(t *testing.T, onMultipleRanges bool) {
 	l2info := builder.L2Info
 	client := builder.L2.Client
 	blockchain := builder.L2.ExecNode.Backend.ArbInterface().BlockChain()
-	feedErrChan := make(chan error, 10)
 
 	l2info.GenerateAccount("User2")
 	genesis, err := client.BlockNumber(ctx)
@@ -62,27 +65,144 @@ func testBlocksReExecutorModes(t *testing.T, onMultipleRanges bool) {
 	// Reexecute blocks at mode full
 	c.MinBlocksPerThread = 10
 	Require(t, c.Validate())
-	executorFull, err := blocksreexecutor.New(&c, blockchain, builder.L2.ExecNode.ChainDB, feedErrChan)
+	executorFull, err := blocksreexecutor.New(&c, blockchain, builder.L2.ExecNode.ChainDB)
+	Require(t, err)
+	executorFull.Start(ctx)
+	err = executorFull.WaitForReExecution(ctx)
 	Require(t, err)
-	success := make(chan struct{})
-	executorFull.Start(ctx, success)
-	select {
-	case err := <-feedErrChan:
-		t.Fatalf("error occurred: %v", err)
-	case <-success:
-	}
 
 	// Reexecute blocks at mode random
 	c.Mode = "random"
 	c.MinBlocksPerThread = 20
 	Require(t, c.Validate())
-	executorRandom, err := blocksreexecutor.New(&c, blockchain, builder.L2.ExecNode.ChainDB, feedErrChan)
+	executorRandom, err := blocksreexecutor.New(&c, blockchain, builder.L2.ExecNode.ChainDB)
+	Require(t, err)
+	executorRandom.Start(ctx)
+	err = executorFull.WaitForReExecution(ctx)
+	Require(t, err)
+}
+
+func TestBlocksReExecutorCommitState(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	builder := NewNodeBuilder(ctx).DefaultConfig(t, false).WithDatabase(rawdb.DBPebble)
+	// For now PathDB is not supported
+	builder.RequireScheme(t, rawdb.HashScheme)
+
+	// 1. Setup builder to be run in sparse archive mode
+	builder.execConfig.Caching.Archive = true
+	builder.execConfig.Caching.SnapshotCache = 0 // disable snapshots
+	builder.execConfig.Caching.BlockAge = 0
+	builder.execConfig.Caching.MaxNumberOfBlocksToSkipStateSaving = 100
+	builder.execConfig.Caching.MaxAmountOfGasToSkipStateSaving = 0
+
+	maxRecreateStateDepth := int64(100 * 1000 * 1000)
+
+	builder.execConfig.RPC.MaxRecreateStateDepth = maxRecreateStateDepth
+	builder.execConfig.Sequencer.MaxBlockSpeed = 0
+	builder.execConfig.Sequencer.MaxTxDataSize = 150
+
+	cleanup := builder.Build(t)
+	defer cleanup()
+
+	l2info := builder.L2Info
+	client := builder.L2.Client
+	blockchain := builder.L2.ExecNode.Backend.ArbInterface().BlockChain()
+
+	// 2. Create about 500 blocks
+	l2info.GenerateAccount("User2")
+	genesis, err := client.BlockNumber(ctx)
+	Require(t, err)
+	for i := genesis; i < genesis+500; i++ {
+		tx := l2info.PrepareTx("Owner", "User2", l2info.TransferGas, common.Big1, nil)
+		err := client.SendTransaction(ctx, tx)
+		Require(t, err)
+		receipt, err := EnsureTxSucceeded(ctx, client, tx)
+		Require(t, err)
+		if have, want := receipt.BlockNumber.Uint64(), uint64(i)+1; have != want {
+			Fatal(t, "internal test error - tx got included in unexpected block number, have:", have, "want:", want)
+		}
+	}
+
+	// 3. Set 2 blocks as target block tests
+	bc := builder.L2.ExecNode.Backend.ArbInterface().BlockChain()
+	blockNum35 := 35
+	block35 := bc.GetBlockByNumber(uint64(blockNum35))
+	blockRoot35 := block35.Root()
+
+	blockNum150 := 150
+	block150 := bc.GetBlockByNumber(uint64(blockNum150))
+	blockRoot150 := block150.Root()
+
+	// 4. Since we started L1 as a sparse archive we should only expect states to be
+	// present for blocks 100, 200, 300, etc.
+	_, err = bc.StateAt(blockRoot35)
+	expectedErr := &trie.MissingNodeError{}
+	if !errors.As(err, &expectedErr) {
+		Fatal(t, "getting state failed with unexpected error, root:", block35.Root(), "blockNumber:", blockNum35, "blockHash:", block35.Hash(), "err:", err)
+	}
+
+	_, err = bc.StateAt(blockRoot150)
+	expectedErr = &trie.MissingNodeError{}
+	if !errors.As(err, &expectedErr) {
+		Fatal(t, "getting state failed with unexpected error, root:", block150.Root(), "blockNumber:", blockNum150, "blockHash:", block150.Hash(), "err:", err)
+	}
+
+	blockTraceConfig := map[string]interface{}{"tracer": "callTracer"}
+	blockTraceConfig["tracerConfig"] = map[string]interface{}{"onlyTopCall": false}
+
+	l2rpc := builder.L2.Stack.Attach()
+	var blockTrace json.RawMessage
+	err = l2rpc.CallContext(ctx, &blockTrace, "debug_traceBlockByNumber", rpc.BlockNumber(blockNum150), blockTraceConfig)
+	Require(t, err)
+
+	// 5. We first run BlocksReExecutor with ValidateMultiGas set to false to make sure
+	// BlocksReExecutor does not commit state to disk
+	c := blocksreexecutor.TestConfig
+	c.ValidateMultiGas = true
+	c.Blocks = `[[0, 42], [90, 160], [180, 200]]`
+	// We don't need to explicit set it to false since default is false, but we want to be explicit
+	c.CommitStateToDisk = false
+
+	// Reexecute blocks at mode full
+	c.MinBlocksPerThread = 10
+	Require(t, c.Validate())
+	executorFull, err := blocksreexecutor.New(&c, blockchain, builder.L2.ExecNode.ChainDB)
+	Require(t, err)
+	executorFull.Start(ctx)
+	err = executorFull.WaitForReExecution(ctx)
 	Require(t, err)
-	success = make(chan struct{})
-	executorRandom.Start(ctx, success)
-	select {
-	case err := <-feedErrChan:
-		t.Fatalf("error occurred: %v", err)
-	case <-success:
+
+	// 6. Now that we have run Block Re-executor CommitStateToDisk set to false
+	// we should expect state to NOT be present for both of the above blocks
+	_, err = bc.StateAt(blockRoot35)
+	if !errors.As(err, &expectedErr) {
+		Fatal(t, "getting state failed with unexpected error, root:", block35.Root(), "blockNumber:", blockNum35, "blockHash:", block35.Hash(), "err:", err)
+	}
+
+	_, err = bc.StateAt(blockRoot150)
+	expectedErr = &trie.MissingNodeError{}
+	if !errors.As(err, &expectedErr) {
+		Fatal(t, "getting state failed with unexpected error, root:", block150.Root(), "blockNumber:", blockNum150, "blockHash:", block150.Hash(), "err:", err)
 	}
+
+	// 7. Now we run BlocksReExecutor with ValidateMultiGas set to true to make sure
+	// BlocksReExecutor does indeed commit state of c.Blocks to disk.
+	// We don't set c.Blocks since we want to use the same blocks range
+	c.CommitStateToDisk = true
+	Require(t, c.Validate())
+	executorFullCommit, err := blocksreexecutor.New(&c, blockchain, builder.L2.ExecNode.ChainDB)
+	Require(t, err)
+	executorFullCommit.Start(ctx)
+	err = executorFullCommit.WaitForReExecution(ctx)
+	Require(t, err)
+
+	// 8. Now that we have run Block Re-executor with CommitStateToDisk set to true
+	// we should expect state to be present for both of the above blocks
+	_, err = bc.StateAt(blockRoot35)
+	Require(t, err)
+
+	_, err = bc.StateAt(blockRoot150)
+	Require(t, err)
 }