Add support for stopping / starting the batcher

op-batcher/batcher/batch_submitter.go

@@ -3,6 +3,7 @@ package batcher
 import (
 	"context"
 	"fmt"
+	"github.com/ethereum-optimism/optimism/op-batcher/rpc"
 	_ "net/http/pprof"
 	"os"
 	"os/signal"
@@ -15,6 +16,7 @@ import (
 	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"
+	gethrpc "github.com/ethereum/go-ethereum/rpc"
 )
 
 const (
@@ -44,10 +46,12 @@ func Main(version string, cliCtx *cli.Context) error {
 
 	l.Info("Starting Batch Submitter")
 
+	if !cfg.Stopped {
 		if err := batchSubmitter.Start(); err != nil {
 			l.Error("Unable to start Batch Submitter", "error", err)
 			return err
 		}
+	}
 	defer batchSubmitter.Stop()
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -81,6 +85,12 @@ func Main(version string, cliCtx *cli.Context) error {
 		rpcCfg.ListenPort,
 		version,
 	)
+	if rpcCfg.EnableAdmin {
+		server.AddAPI(gethrpc.API{
+			Namespace: "admin",
+			Service:   rpc.NewAdminAPI(batchSubmitter),
+		})
+	}
 	if err := server.Start(); err != nil {
 		cancel()
 		return fmt.Errorf("error starting RPC server: %w", err)
@@ -97,5 +107,4 @@ func Main(version string, cliCtx *cli.Context) error {
 	cancel()
 	_ = server.Stop()
 	return nil
-
 }

op-batcher/batcher/config.go

 package batcher
 
 import (
+	"github.com/ethereum-optimism/optimism/op-batcher/rpc"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -14,7 +15,6 @@ import (
 	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,45 @@ 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),
 	}, nil
 
 }
 
 func (l *BatchSubmitter) Start() error {
+	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()
 	return nil
 }
 
-func (l *BatchSubmitter) Stop() {
+func (l *BatchSubmitter) Stop() error {
+	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()
+	return nil
 }
 
 // loadBlocksIntoState loads all blocks since the previous stored block
@@ -199,7 +218,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 +282,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

 package flags
 
 import (
+	"github.com/ethereum-optimism/optimism/op-batcher/rpc"
 	"github.com/urfave/cli"
 
 	opservice "github.com/ethereum-optimism/optimism/op-service"
@@ -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 (
+	opservice "github.com/ethereum-optimism/optimism/op-service"
+	oprpc "github.com/ethereum-optimism/optimism/op-service/rpc"
+	"github.com/urfave/cli"
+)
+
+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),
+	}
+}

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: