Merge pull request #5012 from mdehoog/michael/stop-start-batcher

[op-batcher] Add support for stopping / starting the batcher

.gitignore

@@ -39,6 +39,9 @@ packages/contracts-bedrock/deployments/anvil
 # vim
 *.sw*
 
+# jetbrains
+.idea/
+
 .env
 .env*
 !.env.example

op-batcher/batcher/batch_submitter.go

@@ -9,8 +9,10 @@ import (
 	"syscall"
 	"time"
 
+	gethrpc "github.com/ethereum/go-ethereum/rpc"
 	"github.com/urfave/cli"
 
+	"github.com/ethereum-optimism/optimism/op-batcher/rpc"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
 	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
 	oppprof "github.com/ethereum-optimism/optimism/op-service/pprof"
@@ -42,17 +44,16 @@ func Main(version string, cliCtx *cli.Context) error {
 		return err
 	}
 
-	l.Info("Starting Batch Submitter")
-
-	if err := batchSubmitter.Start(); err != nil {
-		l.Error("Unable to start Batch Submitter", "error", err)
-		return err
+	if !cfg.Stopped {
+		if err := batchSubmitter.Start(); err != nil {
+			l.Error("Unable to start Batch Submitter", "error", err)
+			return err
+		}
 	}
-	defer batchSubmitter.Stop()
+	defer batchSubmitter.StopIfRunning()
 
 	ctx, cancel := context.WithCancel(context.Background())
 
-	l.Info("Batch Submitter started")
 	pprofConfig := cfg.PprofConfig
 	if pprofConfig.Enabled {
 		l.Info("starting pprof", "addr", pprofConfig.ListenAddr, "port", pprofConfig.ListenPort)
@@ -81,6 +82,13 @@ func Main(version string, cliCtx *cli.Context) error {
 		rpcCfg.ListenPort,
 		version,
 	)
+	if rpcCfg.EnableAdmin {
+		server.AddAPI(gethrpc.API{
+			Namespace: "admin",
+			Service:   rpc.NewAdminAPI(batchSubmitter),
+		})
+		l.Info("Admin RPC enabled")
+	}
 	if err := server.Start(); err != nil {
 		cancel()
 		return fmt.Errorf("error starting RPC server: %w", err)
@@ -97,5 +105,4 @@ func Main(version string, cliCtx *cli.Context) error {
 	cancel()
 	_ = server.Stop()
 	return nil
-
 }

op-batcher/batcher/config.go

@@ -9,12 +9,12 @@ import (
 	"github.com/urfave/cli"
 
 	"github.com/ethereum-optimism/optimism/op-batcher/flags"
+	"github.com/ethereum-optimism/optimism/op-batcher/rpc"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/sources"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
 	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
 	oppprof "github.com/ethereum-optimism/optimism/op-service/pprof"
-	oprpc "github.com/ethereum-optimism/optimism/op-service/rpc"
 	"github.com/ethereum-optimism/optimism/op-service/txmgr"
 	opsigner "github.com/ethereum-optimism/optimism/op-signer/client"
 )
@@ -81,7 +81,7 @@ type CLIConfig struct {
 	// PrivateKey is the private key used to submit sequencer transactions.
 	PrivateKey string
 
-	RPCConfig oprpc.CLIConfig
+	RPCConfig rpc.CLIConfig
 
 	/* Optional Params */
 
@@ -98,6 +98,8 @@ type CLIConfig struct {
 	// compression algorithm.
 	ApproxComprRatio float64
 
+	Stopped bool
+
 	LogConfig oplog.CLIConfig
 
 	MetricsConfig opmetrics.CLIConfig
@@ -145,10 +147,11 @@ func NewConfig(ctx *cli.Context) CLIConfig {
 		TargetL1TxSize:   ctx.GlobalUint64(flags.TargetL1TxSizeBytesFlag.Name),
 		TargetNumFrames:  ctx.GlobalInt(flags.TargetNumFramesFlag.Name),
 		ApproxComprRatio: ctx.GlobalFloat64(flags.ApproxComprRatioFlag.Name),
+		Stopped:          ctx.GlobalBool(flags.StoppedFlag.Name),
 		Mnemonic:         ctx.GlobalString(flags.MnemonicFlag.Name),
 		SequencerHDPath:  ctx.GlobalString(flags.SequencerHDPathFlag.Name),
 		PrivateKey:       ctx.GlobalString(flags.PrivateKeyFlag.Name),
-		RPCConfig:        oprpc.ReadCLIConfig(ctx),
+		RPCConfig:        rpc.ReadCLIConfig(ctx),
 		LogConfig:        oplog.ReadCLIConfig(ctx),
 		MetricsConfig:    opmetrics.ReadCLIConfig(ctx),
 		PprofConfig:      oppprof.ReadCLIConfig(ctx),

op-batcher/batcher/driver.go

@@ -29,6 +29,9 @@ type BatchSubmitter struct {
 	ctx    context.Context
 	cancel context.CancelFunc
 
+	mutex   sync.Mutex
+	running bool
+
 	// lastStoredBlock is the last block loaded into `state`. If it is empty it should be set to the l2 safe head.
 	lastStoredBlock eth.BlockID
 
@@ -95,17 +98,14 @@ func NewBatchSubmitterFromCLIConfig(cfg CLIConfig, l log.Logger) (*BatchSubmitte
 		},
 	}
 
-	return NewBatchSubmitter(batcherCfg, l)
+	return NewBatchSubmitter(ctx, batcherCfg, l)
 }
 
 // NewBatchSubmitter initializes the BatchSubmitter, gathering any resources
 // that will be needed during operation.
-func NewBatchSubmitter(cfg Config, l log.Logger) (*BatchSubmitter, error) {
-	ctx, cancel := context.WithCancel(context.Background())
-
+func NewBatchSubmitter(ctx context.Context, cfg Config, l log.Logger) (*BatchSubmitter, error) {
 	balance, err := cfg.L1Client.BalanceAt(ctx, cfg.From, nil)
 	if err != nil {
-		cancel()
 		return nil, err
 	}
 
@@ -117,26 +117,59 @@ func NewBatchSubmitter(cfg Config, l log.Logger) (*BatchSubmitter, error) {
 		txMgr: NewTransactionManager(l,
 			cfg.TxManagerConfig, cfg.Rollup.BatchInboxAddress, cfg.Rollup.L1ChainID,
 			cfg.From, cfg.L1Client),
-		done: make(chan struct{}),
-		// TODO: this context only exists because the event loop doesn't reach done
-		// if the tx manager is blocking forever due to e.g. insufficient balance.
-		ctx:    ctx,
-		cancel: cancel,
-		state:  NewChannelManager(l, cfg.Channel),
+		state: NewChannelManager(l, cfg.Channel),
 	}, nil
 
 }
 
 func (l *BatchSubmitter) Start() error {
+	l.log.Info("Starting Batch Submitter")
+
+	l.mutex.Lock()
+	defer l.mutex.Unlock()
+
+	if l.running {
+		return errors.New("batcher is already running")
+	}
+	l.running = true
+
+	l.done = make(chan struct{})
+	// TODO: this context only exists because the event loop doesn't reach done
+	// if the tx manager is blocking forever due to e.g. insufficient balance.
+	l.ctx, l.cancel = context.WithCancel(context.Background())
+	l.state.Clear()
+	l.lastStoredBlock = eth.BlockID{}
+
 	l.wg.Add(1)
 	go l.loop()
+
+	l.log.Info("Batch Submitter started")
+
 	return nil
 }
 
-func (l *BatchSubmitter) Stop() {
+func (l *BatchSubmitter) StopIfRunning() {
+	_ = l.Stop()
+}
+
+func (l *BatchSubmitter) Stop() error {
+	l.log.Info("Stopping Batch Submitter")
+
+	l.mutex.Lock()
+	defer l.mutex.Unlock()
+
+	if !l.running {
+		return errors.New("batcher is not running")
+	}
+	l.running = false
+
 	l.cancel()
 	close(l.done)
 	l.wg.Wait()
+
+	l.log.Info("Batch Submitter stopped")
+
+	return nil
 }
 
 // loadBlocksIntoState loads all blocks since the previous stored block
@@ -199,7 +232,7 @@ func (l *BatchSubmitter) calculateL2BlockRangeToStore(ctx context.Context) (eth.
 	}
 
 	// Check last stored to see if it needs to be set on startup OR set if is lagged behind.
-	// It lagging implies that the op-node processed some batches that where submitted prior to the current instance of the batcher being alive.
+	// It lagging implies that the op-node processed some batches that were submitted prior to the current instance of the batcher being alive.
 	if l.lastStoredBlock == (eth.BlockID{}) {
 		l.log.Info("Starting batch-submitter work at safe-head", "safe", syncStatus.SafeL2)
 		l.lastStoredBlock = syncStatus.SafeL2.ID()
@@ -263,7 +296,7 @@ func (l *BatchSubmitter) loop() {
 
 				// hack to exit this loop. Proper fix is to do request another send tx or parallel tx sending
 				// from the channel manager rather than sending the channel in a loop. This stalls b/c if the
-				// context is cancelled while sending, it will never fuilly clearing the pending txns.
+				// context is cancelled while sending, it will never fully clear the pending txns.
 				select {
 				case <-l.ctx.Done():
 					break blockLoop

op-batcher/flags/flags.go

@@ -3,6 +3,7 @@ package flags
 import (
 	"github.com/urfave/cli"
 
+	"github.com/ethereum-optimism/optimism/op-batcher/rpc"
 	opservice "github.com/ethereum-optimism/optimism/op-service"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
 	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
@@ -98,6 +99,11 @@ var (
 		Value:  1.0,
 		EnvVar: opservice.PrefixEnvVar(envVarPrefix, "APPROX_COMPR_RATIO"),
 	}
+	StoppedFlag = cli.BoolFlag{
+		Name:   "stopped",
+		Usage:  "Initialize the batcher in a stopped state. The batcher can be started using the admin_startBatcher RPC",
+		EnvVar: opservice.PrefixEnvVar(envVarPrefix, "STOPPED"),
+	}
 	MnemonicFlag = cli.StringFlag{
 		Name: "mnemonic",
 		Usage: "The mnemonic used to derive the wallets for either the " +
@@ -133,6 +139,7 @@ var optionalFlags = []cli.Flag{
 	TargetL1TxSizeBytesFlag,
 	TargetNumFramesFlag,
 	ApproxComprRatioFlag,
+	StoppedFlag,
 	MnemonicFlag,
 	SequencerHDPathFlag,
 	PrivateKeyFlag,
@@ -145,6 +152,7 @@ func init() {
 	optionalFlags = append(optionalFlags, opmetrics.CLIFlags(envVarPrefix)...)
 	optionalFlags = append(optionalFlags, oppprof.CLIFlags(envVarPrefix)...)
 	optionalFlags = append(optionalFlags, opsigner.CLIFlags(envVarPrefix)...)
+	optionalFlags = append(optionalFlags, rpc.CLIFlags(envVarPrefix)...)
 
 	Flags = append(requiredFlags, optionalFlags...)
 }

op-batcher/rpc/api.go

+package rpc
+
+import (
+	"context"
+)
+
+type batcherClient interface {
+	Start() error
+	Stop() error
+}
+
+type adminAPI struct {
+	b batcherClient
+}
+
+func NewAdminAPI(dr batcherClient) *adminAPI {
+	return &adminAPI{
+		b: dr,
+	}
+}
+
+func (a *adminAPI) StartBatcher(_ context.Context) error {
+	return a.b.Start()
+}
+
+func (a *adminAPI) StopBatcher(_ context.Context) error {
+	return a.b.Stop()
+}

op-batcher/rpc/config.go

+package rpc
+
+import (
+	"github.com/urfave/cli"
+
+	opservice "github.com/ethereum-optimism/optimism/op-service"
+	oprpc "github.com/ethereum-optimism/optimism/op-service/rpc"
+)
+
+const (
+	EnableAdminFlagName = "rpc.enable-admin"
+)
+
+func CLIFlags(envPrefix string) []cli.Flag {
+	return []cli.Flag{
+		cli.BoolFlag{
+			Name:   EnableAdminFlagName,
+			Usage:  "Enable the admin API (experimental)",
+			EnvVar: opservice.PrefixEnvVar(envPrefix, "RPC_ENABLE_ADMIN"),
+		},
+	}
+}
+
+type CLIConfig struct {
+	oprpc.CLIConfig
+	EnableAdmin bool
+}
+
+func ReadCLIConfig(ctx *cli.Context) CLIConfig {
+	return CLIConfig{
+		CLIConfig:   oprpc.ReadCLIConfig(ctx),
+		EnableAdmin: ctx.GlobalBool(EnableAdminFlagName),
+	}
+}

op-e2e/migration_test.go

@@ -340,7 +340,7 @@ func TestMigration(t *testing.T) {
 	}, lgr.New("module", "batcher"))
 	require.NoError(t, err)
 	t.Cleanup(func() {
-		batcher.Stop()
+		batcher.StopIfRunning()
 	})
 
 	proposer, err := l2os.NewL2OutputSubmitterFromCLIConfig(l2os.CLIConfig{

op-e2e/setup.go

@@ -212,7 +212,7 @@ func (sys *System) Close() {
 		sys.L2OutputSubmitter.Stop()
 	}
 	if sys.BatchSubmitter != nil {
-		sys.BatchSubmitter.Stop()
+		sys.BatchSubmitter.StopIfRunning()
 	}
 
 	for _, node := range sys.RollupNodes {

op-e2e/system_test.go

@@ -1173,6 +1173,101 @@ func TestStopStartSequencer(t *testing.T) {
 	require.Greater(t, blockAfter, blockBefore, "Chain did not advance after starting sequencer")
 }
 
+func TestStopStartBatcher(t *testing.T) {
+	parallel(t)
+	if !verboseGethNodes {
+		log.Root().SetHandler(log.DiscardHandler())
+	}
+
+	cfg := DefaultSystemConfig(t)
+	sys, err := cfg.Start()
+	require.Nil(t, err, "Error starting up system")
+	defer sys.Close()
+
+	rollupRPCClient, err := rpc.DialContext(context.Background(), sys.RollupNodes["verifier"].HTTPEndpoint())
+	require.Nil(t, err)
+	rollupClient := sources.NewRollupClient(client.NewBaseRPCClient(rollupRPCClient))
+
+	l2Seq := sys.Clients["sequencer"]
+	l2Verif := sys.Clients["verifier"]
+
+	// retrieve the initial sync status
+	seqStatus, err := rollupClient.SyncStatus(context.Background())
+	require.Nil(t, err)
+
+	nonce := uint64(0)
+	sendTx := func() *types.Receipt {
+		// Submit TX to L2 sequencer node
+		tx := types.MustSignNewTx(cfg.Secrets.Alice, types.LatestSignerForChainID(cfg.L2ChainIDBig()), &types.DynamicFeeTx{
+			ChainID:   cfg.L2ChainIDBig(),
+			Nonce:     nonce,
+			To:        &common.Address{0xff, 0xff},
+			Value:     big.NewInt(1_000_000_000),
+			GasTipCap: big.NewInt(10),
+			GasFeeCap: big.NewInt(200),
+			Gas:       21000,
+		})
+		nonce++
+		err = l2Seq.SendTransaction(context.Background(), tx)
+		require.Nil(t, err, "Sending L2 tx to sequencer")
+
+		// Let it show up on the unsafe chain
+		receipt, err := waitForTransaction(tx.Hash(), l2Seq, 3*time.Duration(cfg.DeployConfig.L1BlockTime)*time.Second)
+		require.Nil(t, err, "Waiting for L2 tx on sequencer")
+
+		return receipt
+	}
+	// send a transaction
+	receipt := sendTx()
+
+	// wait until the block the tx was first included in shows up in the safe chain on the verifier
+	safeBlockInclusionDuration := time.Duration(3*cfg.DeployConfig.L1BlockTime) * time.Second
+	_, err = waitForBlock(receipt.BlockNumber, l2Verif, safeBlockInclusionDuration)
+	require.Nil(t, err, "Waiting for block on verifier")
+
+	// ensure the safe chain advances
+	newSeqStatus, err := rollupClient.SyncStatus(context.Background())
+	require.Nil(t, err)
+	require.Greater(t, newSeqStatus.SafeL2.Number, seqStatus.SafeL2.Number, "Safe chain did not advance")
+
+	// stop the batch submission
+	err = sys.BatchSubmitter.Stop()
+	require.Nil(t, err)
+
+	// wait for any old safe blocks being submitted / derived
+	time.Sleep(safeBlockInclusionDuration)
+
+	// get the initial sync status
+	seqStatus, err = rollupClient.SyncStatus(context.Background())
+	require.Nil(t, err)
+
+	// send another tx
+	sendTx()
+	time.Sleep(safeBlockInclusionDuration)
+
+	// ensure that the safe chain does not advance while the batcher is stopped
+	newSeqStatus, err = rollupClient.SyncStatus(context.Background())
+	require.Nil(t, err)
+	require.Equal(t, newSeqStatus.SafeL2.Number, seqStatus.SafeL2.Number, "Safe chain advanced while batcher was stopped")
+
+	// start the batch submission
+	err = sys.BatchSubmitter.Start()
+	require.Nil(t, err)
+	time.Sleep(safeBlockInclusionDuration)
+
+	// send a third tx
+	receipt = sendTx()
+
+	// wait until the block the tx was first included in shows up in the safe chain on the verifier
+	_, err = waitForBlock(receipt.BlockNumber, l2Verif, safeBlockInclusionDuration)
+	require.Nil(t, err, "Waiting for block on verifier")
+
+	// ensure that the safe chain advances after restarting the batcher
+	newSeqStatus, err = rollupClient.SyncStatus(context.Background())
+	require.Nil(t, err)
+	require.Greater(t, newSeqStatus.SafeL2.Number, seqStatus.SafeL2.Number, "Safe chain did not advance after batcher was restarted")
+}
+
 func safeAddBig(a *big.Int, b *big.Int) *big.Int {
 	return new(big.Int).Add(a, b)
 }

op-node/node/node.go

@@ -215,6 +215,7 @@ func (n *OpNode) initRPCServer(ctx context.Context, cfg *Config) error {
 	}
 	if cfg.RPC.EnableAdmin {
 		server.EnableAdminAPI(NewAdminAPI(n.l2Driver, n.metrics))
+		n.log.Info("Admin RPC enabled")
 	}
 	n.log.Info("Starting JSON-RPC server")
 	if err := server.Start(); err != nil {

ops-bedrock/docker-compose.yml

@@ -115,6 +115,7 @@ services:
     ports:
       - "6061:6060"
       - "7301:7300"
+      - "6545:8545"
     environment:
       OP_BATCHER_L1_ETH_RPC: http://l1:8545
       OP_BATCHER_L2_ETH_RPC: http://l2:8545
@@ -130,10 +131,10 @@ services:
       OP_BATCHER_RESUBMISSION_TIMEOUT: 30s
       OP_BATCHER_MNEMONIC: test test test test test test test test test test test junk
       OP_BATCHER_SEQUENCER_HD_PATH: "m/44'/60'/0'/0/2"
-      OP_BATCHER_SEQUENCER_BATCH_INBOX_ADDRESS: "${SEQUENCER_BATCH_INBOX_ADDRESS}"
       OP_BATCHER_LOG_TERMINAL: "true"
       OP_BATCHER_PPROF_ENABLED: "true"
       OP_BATCHER_METRICS_ENABLED: "true"
+      OP_BATCHER_RPC_ENABLE_ADMIN: "true"
 
   stateviz:
     build: