Add support for stopping / starting sequencer from admin API

op-e2e/actions/l2_verifier.go

@@ -113,6 +113,14 @@ func (s *l2VerifierBackend) ResetDerivationPipeline(ctx context.Context) error {
 	return nil
 }
 
+func (s *l2VerifierBackend) StartSequencer(ctx context.Context) error {
+	return nil
+}
+
+func (s *l2VerifierBackend) StopSequencer(ctx context.Context) error {
+	return errors.New("stopping the L2Verifier sequencer is not supported")
+}
+
 func (s *L2Verifier) L2Finalized() eth.L2BlockRef {
 	return s.derivation.Finalized()
 }

op-e2e/system_test.go

@@ -1129,6 +1129,57 @@ func TestFees(t *testing.T) {
 	require.Equal(t, balanceDiff, totalFee, "balances should add up")
 }
 
+func TestStopStartSequencer(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()
+
+	l2Seq := sys.Clients["sequencer"]
+	rollupNode := sys.RollupNodes["sequencer"]
+
+	nodeRPC, err := rpc.DialContext(context.Background(), rollupNode.HTTPEndpoint())
+	require.Nil(t, err, "Error dialing node")
+
+	blockBefore := latestBlock(t, l2Seq)
+	time.Sleep(time.Duration(cfg.DeployConfig.L2BlockTime+1) * time.Second)
+	blockAfter := latestBlock(t, l2Seq)
+	require.Greaterf(t, blockAfter, blockBefore, "Chain did not advance")
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	err = nodeRPC.CallContext(ctx, nil, "admin_stopSequencer")
+	require.Nil(t, err, "Error stopping sequencer")
+
+	blockBefore = latestBlock(t, l2Seq)
+	time.Sleep(time.Duration(cfg.DeployConfig.L2BlockTime+1) * time.Second)
+	blockAfter = latestBlock(t, l2Seq)
+	require.Equal(t, blockAfter, blockBefore, "Chain advanced after stopping sequencer")
+
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	err = nodeRPC.CallContext(ctx, nil, "admin_startSequencer")
+	require.Nil(t, err, "Error starting sequencer")
+
+	blockBefore = latestBlock(t, l2Seq)
+	time.Sleep(time.Duration(cfg.DeployConfig.L2BlockTime+1) * time.Second)
+	blockAfter = latestBlock(t, l2Seq)
+	require.Greater(t, blockAfter, blockBefore, "Chain did not advance after starting sequencer")
+}
+
 func safeAddBig(a *big.Int, b *big.Int) *big.Int {
 	return new(big.Int).Add(a, b)
 }
+
+func latestBlock(t *testing.T, client *ethclient.Client) uint64 {
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancel()
+	blockAfter, err := client.BlockNumber(ctx)
+	require.Nil(t, err, "Error getting latest block")
+	return blockAfter
+}

op-node/node/api.go

@@ -26,6 +26,8 @@ type driverClient interface {
 	SyncStatus(ctx context.Context) (*eth.SyncStatus, error)
 	BlockRefWithStatus(ctx context.Context, num uint64) (eth.L2BlockRef, *eth.SyncStatus, error)
 	ResetDerivationPipeline(context.Context) error
+	StartSequencer(context.Context) error
+	StopSequencer(context.Context) error
 }
 
 type rpcMetrics interface {
@@ -51,6 +53,18 @@ func (n *adminAPI) ResetDerivationPipeline(ctx context.Context) error {
 	return n.dr.ResetDerivationPipeline(ctx)
 }
 
+func (n *adminAPI) StartSequencer(ctx context.Context) error {
+	recordDur := n.m.RecordRPCServerRequest("admin_startSequencer")
+	defer recordDur()
+	return n.dr.StartSequencer(ctx)
+}
+
+func (n *adminAPI) StopSequencer(ctx context.Context) error {
+	recordDur := n.m.RecordRPCServerRequest("admin_stopSequencer")
+	defer recordDur()
+	return n.dr.StopSequencer(ctx)
+}
+
 type nodeAPI struct {
 	config *rollup.Config
 	client l2EthClient

op-node/node/server_test.go

@@ -218,3 +218,11 @@ func (c *mockDriverClient) SyncStatus(ctx context.Context) (*eth.SyncStatus, err
 func (c *mockDriverClient) ResetDerivationPipeline(ctx context.Context) error {
 	return c.Mock.MethodCalled("ResetDerivationPipeline").Get(0).(error)
 }
+
+func (c *mockDriverClient) StartSequencer(ctx context.Context) error {
+	return c.Mock.MethodCalled("StartSequencer").Get(0).(error)
+}
+
+func (c *mockDriverClient) StopSequencer(ctx context.Context) error {
+	return c.Mock.MethodCalled("StopSequencer").Get(0).(error)
+}

op-node/rollup/driver/driver.go

@@ -97,6 +97,8 @@ func NewDriver(driverCfg *Config, cfg *rollup.Config, l2 L2Chain, l1 L1Chain, ne
 		idleDerivation:   false,
 		stateReq:         make(chan chan struct{}),
 		forceReset:       make(chan chan struct{}, 10),
+		startSequencer:   make(chan chan struct{}, 10),
+		stopSequencer:    make(chan chan struct{}, 10),
 		config:           cfg,
 		driverConfig:     driverCfg,
 		done:             make(chan struct{}),

op-node/rollup/driver/state.go

@@ -42,6 +42,14 @@ type Driver struct {
 	// It tells the caller that the reset occurred by closing the passed in channel.
 	forceReset chan chan struct{}
 
+	// Upon receiving a channel in this channel, the sequencer is started.
+	// It tells the caller that the sequencer started by closing the passed in channel.
+	startSequencer chan chan struct{}
+
+	// Upon receiving a channel in this channel, the sequencer is stopped.
+	// It tells the caller that the sequencer stopped by closing the passed in channel.
+	stopSequencer chan chan struct{}
+
 	// Rollup config: rollup chain configuration
 	config *rollup.Config
 
@@ -367,6 +375,15 @@ func (s *Driver) eventLoop() {
 			s.derivation.Reset()
 			s.metrics.RecordPipelineReset()
 			close(respCh)
+		case respCh := <-s.startSequencer:
+			s.log.Info("Sequencer has been started")
+			s.driverConfig.SequencerEnabled = true
+			sequencingPlannedOnto = eth.BlockID{}
+			close(respCh)
+		case respCh := <-s.stopSequencer:
+			s.log.Warn("Sequencer has been stopped")
+			s.driverConfig.SequencerEnabled = false
+			close(respCh)
 		case <-s.done:
 			return
 		}
@@ -391,6 +408,36 @@ func (s *Driver) ResetDerivationPipeline(ctx context.Context) error {
 	}
 }
 
+func (s *Driver) StartSequencer(ctx context.Context) error {
+	respCh := make(chan struct{}, 1)
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case s.startSequencer <- respCh:
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-respCh:
+			return nil
+		}
+	}
+}
+
+func (s *Driver) StopSequencer(ctx context.Context) error {
+	respCh := make(chan struct{}, 1)
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case s.stopSequencer <- respCh:
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-respCh:
+			return nil
+		}
+	}
+}
+
 // syncStatus returns the current sync status, and should only be called synchronously with
 // the driver event loop to avoid retrieval of an inconsistent status.
 func (s *Driver) syncStatus() *eth.SyncStatus {

ops-bedrock/docker-compose.yml

@@ -65,6 +65,7 @@ services:
       --metrics.addr=0.0.0.0
       --metrics.port=7300
       --pprof.enabled
+      --rpc.enable-admin
     ports:
       - "7545:8545"
       - "9003:9003"