add SyncHelper [NIT-1241]

arbnode/inbox_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/offchainlabs/nitro/arbos/arbostypes"
 	"github.com/offchainlabs/nitro/arbos/l2pricing"
 	"github.com/offchainlabs/nitro/arbutil"
+	"github.com/offchainlabs/nitro/execution"
 	"github.com/offchainlabs/nitro/execution/gethexec"
 	"github.com/offchainlabs/nitro/statetransfer"
 
@@ -37,6 +38,9 @@ type execClientWrapper struct {
 func (w *execClientWrapper) Pause()                     { w.t.Error("not supported") }
 func (w *execClientWrapper) Activate()                  { w.t.Error("not supported") }
 func (w *execClientWrapper) ForwardTo(url string) error { w.t.Error("not supported"); return nil }
+func (w *execClientWrapper) SetConfirmedNodeHelper(confirmedNodeHelper execution.ConfirmedNodeHelper) {
+	w.t.Error("not supported")
+}
 
 func NewTransactionStreamerForTest(t *testing.T, ownerAddress common.Address) (*gethexec.ExecutionEngine, *TransactionStreamer, ethdb.Database, *core.BlockChain) {
 	chainConfig := params.ArbitrumDevTestChainConfig()
@@ -54,7 +58,7 @@ func NewTransactionStreamerForTest(t *testing.T, ownerAddress common.Address) (*
 	arbDb := rawdb.NewMemoryDatabase()
 	initReader := statetransfer.NewMemoryInitDataReader(&initData)
 
-	bc, err := gethexec.WriteOrTestBlockChain(chainDb, nil, initReader, chainConfig, arbostypes.TestInitMessage, gethexec.ConfigDefaultTest().TxLookupLimit, 0)
+	bc, err := gethexec.WriteOrTestBlockChain(chainDb, nil, initReader, chainConfig, arbostypes.TestInitMessage, gethexec.ConfigDefaultTest().TxLookupLimit, 0, nil)
 
 	if err != nil {
 		Fail(t, err)

arbnode/message_pruner.go

@@ -65,7 +65,7 @@ func (m *MessagePruner) Start(ctxIn context.Context) {
 	m.StopWaiter.Start(ctxIn, m)
 }
 
-func (m *MessagePruner) UpdateLatestConfirmed(count arbutil.MessageIndex, globalState validator.GoGlobalState) {
+func (m *MessagePruner) UpdateLatestConfirmed(count arbutil.MessageIndex, globalState validator.GoGlobalState, _ uint64) {
 	locked := m.pruningLock.TryLock()
 	if !locked {
 		return

arbnode/node.go

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
+	"reflect"
 	"strings"
 	"time"
 
@@ -246,6 +247,7 @@ type Node struct {
 	DASLifecycleManager     *das.LifecycleManager
 	ClassicOutboxRetriever  *ClassicOutboxRetriever
 	SyncMonitor             *SyncMonitor
+	ConfirmedNodeHelper     *staker.ConfirmedNodeHelper
 	configFetcher           ConfigFetcher
 	ctx                     context.Context
 }
@@ -467,6 +469,7 @@ func createNodeImpl(
 			DASLifecycleManager:     nil,
 			ClassicOutboxRetriever:  classicOutbox,
 			SyncMonitor:             syncMonitor,
+			ConfirmedNodeHelper:     nil,
 			configFetcher:           configFetcher,
 			ctx:                     ctx,
 		}, nil
@@ -559,6 +562,12 @@ func createNodeImpl(
 		}
 	}
 
+	var confirmedNodeHelper *staker.ConfirmedNodeHelper
+	if l1client != nil && !reflect.ValueOf(l1client).IsNil() {
+		confirmedNodeHelper = staker.NewConfirmedNodeHelper(deployInfo.Rollup, l1client)
+		exec.SetConfirmedNodeHelper(confirmedNodeHelper)
+	}
+
 	var stakerObj *staker.Staker
 	var messagePruner *MessagePruner
 
@@ -610,6 +619,9 @@ func createNodeImpl(
 			messagePruner = NewMessagePruner(txStreamer, inboxTracker, func() *MessagePrunerConfig { return &configFetcher.Get().MessagePruner })
 			confirmedNotifiers = append(confirmedNotifiers, messagePruner)
 		}
+		if confirmedNodeHelper != nil {
+			confirmedNotifiers = append(confirmedNotifiers, confirmedNodeHelper)
+		}
 
 		stakerObj, err = staker.NewStaker(l1Reader, wallet, bind.CallOpts{}, config.Staker, blockValidator, statelessBlockValidator, nil, confirmedNotifiers, deployInfo.ValidatorUtils, fatalErrChan)
 		if err != nil {
@@ -682,6 +694,7 @@ func createNodeImpl(
 		DASLifecycleManager:     dasLifecycleManager,
 		ClassicOutboxRetriever:  classicOutbox,
 		SyncMonitor:             syncMonitor,
+		ConfirmedNodeHelper:     confirmedNodeHelper,
 		configFetcher:           configFetcher,
 		ctx:                     ctx,
 	}, nil
@@ -752,6 +765,9 @@ func (n *Node) Start(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("error starting geth stack: %w", err)
 	}
+	if n.ConfirmedNodeHelper != nil {
+		n.ConfirmedNodeHelper.Start(ctx)
+	}
 	err = n.Execution.Start(ctx)
 	if err != nil {
 		return fmt.Errorf("error starting exec client: %w", err)
@@ -923,6 +939,9 @@ func (n *Node) StopAndWait() {
 	if n.Execution != nil {
 		n.Execution.StopAndWait()
 	}
+	if n.ConfirmedNodeHelper != nil {
+		n.ConfirmedNodeHelper.StopAndWait()
+	}
 	if err := n.Stack.Close(); err != nil {
 		log.Error("error on stack close", "err", err)
 	}

cmd/nitro/init.go

@@ -159,7 +159,7 @@ func validateBlockChain(blockChain *core.BlockChain, chainConfig *params.ChainCo
 	return nil
 }
 
-func openInitializeChainDb(ctx context.Context, stack *node.Node, config *NodeConfig, chainId *big.Int, cacheConfig *core.CacheConfig, l1Client arbutil.L1Interface, rollupAddrs chaininfo.RollupAddresses) (ethdb.Database, *core.BlockChain, error) {
+func openInitializeChainDb(ctx context.Context, stack *node.Node, config *NodeConfig, chainId *big.Int, cacheConfig *core.CacheConfig, l1Client arbutil.L1Interface, rollupAddrs chaininfo.RollupAddresses, forceTriedbCommitHook core.ForceTriedbCommitHook) (ethdb.Database, *core.BlockChain, error) {
 	if !config.Init.Force {
 		if readOnlyDb, err := stack.OpenDatabaseWithFreezer("l2chaindata", 0, 0, "", "", true); err == nil {
 			if chainConfig := gethexec.TryReadStoredChainConfig(readOnlyDb); chainConfig != nil {
@@ -175,7 +175,7 @@ func openInitializeChainDb(ctx context.Context, stack *node.Node, config *NodeCo
 				if err != nil {
 					return chainDb, nil, fmt.Errorf("error pruning: %w", err)
 				}
-				l2BlockChain, err := gethexec.GetBlockChain(chainDb, cacheConfig, chainConfig, config.Execution.TxLookupLimit)
+				l2BlockChain, err := gethexec.GetBlockChain(chainDb, cacheConfig, chainConfig, config.Execution.TxLookupLimit, nil)
 				if err != nil {
 					return chainDb, nil, err
 				}
@@ -258,7 +258,7 @@ func openInitializeChainDb(ctx context.Context, stack *node.Node, config *NodeCo
 		if chainConfig == nil {
 			return chainDb, nil, errors.New("no --init.* mode supplied and chain data not in expected directory")
 		}
-		l2BlockChain, err = gethexec.GetBlockChain(chainDb, cacheConfig, chainConfig, config.Execution.TxLookupLimit)
+		l2BlockChain, err = gethexec.GetBlockChain(chainDb, cacheConfig, chainConfig, config.Execution.TxLookupLimit, forceTriedbCommitHook)
 		if err != nil {
 			return chainDb, nil, err
 		}
@@ -356,7 +356,7 @@ func openInitializeChainDb(ctx context.Context, stack *node.Node, config *NodeCo
 			log.Warn("Created fake init message as L1Reader is disabled and serialized chain config from init message is not available", "json", string(serializedChainConfig))
 		}
 
-		l2BlockChain, err = gethexec.WriteOrTestBlockChain(chainDb, cacheConfig, initDataReader, chainConfig, parsedInitMessage, config.Execution.TxLookupLimit, config.Init.AccountsPerSync)
+		l2BlockChain, err = gethexec.WriteOrTestBlockChain(chainDb, cacheConfig, initDataReader, chainConfig, parsedInitMessage, config.Execution.TxLookupLimit, config.Init.AccountsPerSync, forceTriedbCommitHook)
 		if err != nil {
 			return chainDb, nil, err
 		}

cmd/nitro/nitro.go

@@ -467,7 +467,9 @@ func mainImpl() int {
 		}
 	}
 
-	chainDb, l2BlockChain, err := openInitializeChainDb(ctx, stack, nodeConfig, new(big.Int).SetUint64(nodeConfig.Chain.ID), gethexec.DefaultCacheConfigFor(stack, &nodeConfig.Execution.Caching), l1Client, rollupAddrs)
+	syncHelperConfigFetcher := func() *gethexec.NitroSyncHelperConfig { return &liveNodeConfig.Get().Execution.SyncHelper }
+	forceTriedbCommitHook := gethexec.GetForceTriedbCommitHookForConfig(syncHelperConfigFetcher)
+	chainDb, l2BlockChain, err := openInitializeChainDb(ctx, stack, nodeConfig, new(big.Int).SetUint64(nodeConfig.Chain.ID), gethexec.DefaultCacheConfigFor(stack, &nodeConfig.Execution.Caching), l1Client, rollupAddrs, forceTriedbCommitHook)
 	if l2BlockChain != nil {
 		deferFuncs = append(deferFuncs, func() { l2BlockChain.Stop() })
 	}

execution/gethexec/blockchain.go

@@ -165,19 +165,18 @@ func WriteOrTestChainConfig(chainDb ethdb.Database, config *params.ChainConfig)
 	return nil
 }
 
-func GetBlockChain(chainDb ethdb.Database, cacheConfig *core.CacheConfig, chainConfig *params.ChainConfig, txLookupLimit uint64) (*core.BlockChain, error) {
+func GetBlockChain(chainDb ethdb.Database, cacheConfig *core.CacheConfig, chainConfig *params.ChainConfig, txLookupLimit uint64, forceTriedbCommitHook core.ForceTriedbCommitHook) (*core.BlockChain, error) {
 	engine := arbos.Engine{
 		IsSequencer: true,
 	}
 
 	vmConfig := vm.Config{
 		EnablePreimageRecording: false,
 	}
-
-	return core.NewBlockChain(chainDb, cacheConfig, chainConfig, nil, nil, engine, vmConfig, shouldPreserveFalse, &txLookupLimit)
+	return core.NewArbBlockChain(chainDb, cacheConfig, chainConfig, nil, nil, engine, vmConfig, shouldPreserveFalse, &txLookupLimit, forceTriedbCommitHook)
 }
 
-func WriteOrTestBlockChain(chainDb ethdb.Database, cacheConfig *core.CacheConfig, initData statetransfer.InitDataReader, chainConfig *params.ChainConfig, initMessage *arbostypes.ParsedInitMessage, txLookupLimit uint64, accountsPerSync uint) (*core.BlockChain, error) {
+func WriteOrTestBlockChain(chainDb ethdb.Database, cacheConfig *core.CacheConfig, initData statetransfer.InitDataReader, chainConfig *params.ChainConfig, initMessage *arbostypes.ParsedInitMessage, txLookupLimit uint64, accountsPerSync uint, forceTriedbCommitHook core.ForceTriedbCommitHook) (*core.BlockChain, error) {
 	err := WriteOrTestGenblock(chainDb, initData, chainConfig, initMessage, accountsPerSync)
 	if err != nil {
 		return nil, err
@@ -186,7 +185,7 @@ func WriteOrTestBlockChain(chainDb ethdb.Database, cacheConfig *core.CacheConfig
 	if err != nil {
 		return nil, err
 	}
-	return GetBlockChain(chainDb, cacheConfig, chainConfig, txLookupLimit)
+	return GetBlockChain(chainDb, cacheConfig, chainConfig, txLookupLimit, forceTriedbCommitHook)
 }
 
 // Don't preserve reorg'd out blocks

execution/gethexec/node.go

@@ -49,6 +49,7 @@ type Config struct {
 	Caching                   CachingConfig                    `koanf:"caching"`
 	RPC                       arbitrum.Config                  `koanf:"rpc"`
 	TxLookupLimit             uint64                           `koanf:"tx-lookup-limit"`
+	SyncHelper                NitroSyncHelperConfig            `koanf:"sync-helper"`
 	Dangerous                 DangerousConfig                  `koanf:"dangerous"`
 
 	forwardingTarget string
@@ -136,6 +137,7 @@ type ExecutionNode struct {
 	TxPublisher       TransactionPublisher
 	ConfigFetcher     ConfigFetcher
 	ParentChainReader *headerreader.HeaderReader
+	SyncHelper        *NitroSyncHelper
 	started           atomic.Bool
 }
 
@@ -199,6 +201,12 @@ func CreateExecutionNode(
 		return nil, err
 	}
 
+	var syncHelper *NitroSyncHelper
+	if config.SyncHelper.Enabled {
+		syncHelperConfigFetcher := func() *NitroSyncHelperConfig { return &configFetcher().SyncHelper }
+		syncHelper = NewNitroSyncHelper(syncHelperConfigFetcher, l2BlockChain)
+	}
+
 	apis := []rpc.API{{
 		Namespace: "arb",
 		Version:   "1.0",
@@ -243,6 +251,7 @@ func CreateExecutionNode(
 		TxPublisher:       txPublisher,
 		ConfigFetcher:     configFetcher,
 		ParentChainReader: parentChainReader,
+		SyncHelper:        syncHelper,
 	}, nil
 
 }
@@ -282,13 +291,22 @@ func (n *ExecutionNode) Start(ctx context.Context) error {
 	if n.ParentChainReader != nil {
 		n.ParentChainReader.Start(ctx)
 	}
+	if n.SyncHelper != nil {
+		err := n.SyncHelper.Start(ctx)
+		if err != nil {
+			return fmt.Errorf("Failed to start sync helper: %w", err)
+		}
+	}
 	return nil
 }
 
 func (n *ExecutionNode) StopAndWait() {
 	if !n.started.Load() {
 		return
 	}
+	if n.SyncHelper != nil {
+		n.SyncHelper.StopAndWait()
+	}
 	// TODO after separation
 	// n.Stack.StopRPC() // does nothing if not running
 	if n.TxPublisher.Started() {
@@ -332,6 +350,11 @@ func (n *ExecutionNode) SequenceDelayedMessage(message *arbostypes.L1IncomingMes
 func (n *ExecutionNode) ResultAtPos(pos arbutil.MessageIndex) (*execution.MessageResult, error) {
 	return n.ExecEngine.ResultAtPos(pos)
 }
+func (n *ExecutionNode) SetConfirmedNodeHelper(confirmedNodeHelper execution.ConfirmedNodeHelper) {
+	if n.SyncHelper != nil {
+		n.SyncHelper.SetConfirmedNodeHelper(confirmedNodeHelper)
+	}
+}
 
 func (n *ExecutionNode) RecordBlockCreation(
 	ctx context.Context,