Merge pull request #23 from node-real/op-node-request-coordinator

feat: coordinator_requestBuildingBlock

Author: keroro520

op-e2e/actions/l2_sequencer.go

@@ -46,9 +46,10 @@ func NewL2Sequencer(t Testing, log log.Logger, l1 derive.L1Fetcher, eng L2API, c
 	l1OriginSelector := &MockL1OriginSelector{
 		actual: driver.NewL1OriginSelector(log, cfg, seqConfDepthL1),
 	}
+	var coordinatorClient *rollup.CoordinatorClient
 	return &L2Sequencer{
 		L2Verifier:              *ver,
-		sequencer:               driver.NewSequencer(log, cfg, ver.derivation, attrBuilder, l1OriginSelector, metrics.NoopMetrics),
+		sequencer:               driver.NewSequencer(log, cfg, ver.derivation, coordinatorClient, attrBuilder, l1OriginSelector, metrics.NoopMetrics),
 		mockL1OriginSelector:    l1OriginSelector,
 		failL2GossipUnsafeBlock: nil,
 	}

op-node/flags/flags.go

@@ -207,6 +207,24 @@ var (
 		EnvVar:   prefixEnvVar("L2_BACKUP_UNSAFE_SYNC_RPC_TRUST_RPC"),
 		Required: false,
 	}
+	CoordinatorEnabledFlag = cli.BoolFlag{
+		Name:     "coordinator.enabled",
+		Usage:    "Enable the external coordinator mode",
+		EnvVar:   prefixEnvVar("COORDINATOR_ENABLED"),
+		Required: false,
+	}
+	CoordinatorAddrFlag = cli.StringFlag{
+		Name:     "coordinator.addr",
+		Usage:    "Coordinator listening address",
+		EnvVar:   prefixEnvVar("COORDINATOR_ADDR"),
+		Required: false,
+	}
+	CoordinatorSequencerIdFlag = cli.StringFlag{
+		Name:     "coordinator.sequencer-id",
+		Usage:    "the sequencer id configured in the coordinator",
+		EnvVar:   prefixEnvVar("COORDINATOR_SEQUENCER_ID"),
+		Required: false,
+	}
 )
 
 var requiredFlags = []cli.Flag{
@@ -244,6 +262,9 @@ var optionalFlags = []cli.Flag{
 	HeartbeatURLFlag,
 	BackupL2UnsafeSyncRPC,
 	BackupL2UnsafeSyncRPCTrustRPC,
+	CoordinatorEnabledFlag,
+	CoordinatorAddrFlag,
+	CoordinatorSequencerIdFlag,
 }
 
 // Flags contains the list of configuration options available to the binary.

op-node/node/client.go

@@ -36,6 +36,33 @@ type L1EndpointSetup interface {
 	Check() error
 }
 
+type CoordinatorConfig struct {
+	// Enabled is true when the driver should request permission from op-coordinator before building new blocks.
+	// Default is false.
+	Enabled bool
+
+	// Identifier of the sequencer node to request blocks from.
+	// It must be unique and same as the name of the sequencer node configured in the Coordinator service.
+	SequencerId string
+
+	// Address of the Coordinator JSON-RPC endpoint to use (opcoordinator namespace required).
+	CoordinatorAddr string
+}
+
+func (cfg *CoordinatorConfig) Check() error {
+	if !cfg.Enabled {
+		return nil
+	}
+	if cfg.SequencerId == "" {
+		return errors.New("empty Sequencer Id")
+	}
+	if cfg.CoordinatorAddr == "" {
+		return errors.New("empty Coordinator Address")
+	}
+
+	return nil
+}
+
 type L2EndpointConfig struct {
 	L2EngineAddr string // Address of L2 Engine JSON-RPC endpoint to use (engine and eth namespace required)
 

op-node/node/config.go

@@ -17,6 +17,8 @@ type Config struct {
 	L2     L2EndpointSetup
 	L2Sync L2SyncEndpointSetup
 
+	Coordinator CoordinatorConfig
+
 	Driver driver.Config
 
 	Rollup rollup.Config
@@ -86,6 +88,9 @@ func (cfg *Config) Check() error {
 	if err := cfg.Rollup.Check(); err != nil {
 		return fmt.Errorf("rollup config error: %w", err)
 	}
+	if err := cfg.Coordinator.Check(); err != nil {
+		return fmt.Errorf("coordinator config error: %w", err)
+	}
 	if err := cfg.Metrics.Check(); err != nil {
 		return fmt.Errorf("metrics config error: %w", err)
 	}

op-node/node/node.go

@@ -17,6 +17,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-node/metrics"
 	"github.com/ethereum-optimism/optimism/op-node/p2p"
+	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/driver"
 	"github.com/ethereum-optimism/optimism/op-node/sources"
 )
@@ -40,6 +41,8 @@ type OpNode struct {
 	tracer    Tracer                // tracer to get events for testing/debugging
 	runCfg    *RuntimeConfig        // runtime configurables
 
+	coordinatorClient *rollup.CoordinatorClient // Coordinator RPC
+
 	// some resources cannot be stopped directly, like the p2p gossipsub router (not our design),
 	// and depend on this ctx to be closed.
 	resourcesCtx   context.Context
@@ -81,6 +84,9 @@ func (n *OpNode) init(ctx context.Context, cfg *Config, snapshotLog log.Logger)
 	if err := n.initL1(ctx, cfg); err != nil {
 		return err
 	}
+	if err := n.initOpCoordinator(ctx, cfg); err != nil {
+		return err
+	}
 	if err := n.initRuntimeConfig(ctx, cfg); err != nil {
 		return err
 	}
@@ -115,6 +121,22 @@ func (n *OpNode) initTracer(ctx context.Context, cfg *Config) error {
 	return nil
 }
 
+func (n *OpNode) initOpCoordinator(ctx context.Context, cfg *Config) error {
+	if err := cfg.Coordinator.Check(); err != nil {
+		return fmt.Errorf("coordinator config is invalid: %w", err)
+	}
+
+	if cfg.Coordinator.Enabled {
+		var err error
+		n.coordinatorClient, err = rollup.NewCoordinatorClient(cfg.Coordinator.CoordinatorAddr, cfg.Coordinator.SequencerId)
+		if err != nil {
+			return fmt.Errorf("failed to get Coordinator RPC client: %w", err)
+		}
+	}
+
+	return nil
+}
+
 func (n *OpNode) initL1(ctx context.Context, cfg *Config) error {
 	l1Node, rpcCfg, err := cfg.L1.Setup(ctx, n.log, &cfg.Rollup)
 	if err != nil {
@@ -199,7 +221,7 @@ func (n *OpNode) initL2(ctx context.Context, cfg *Config, snapshotLog log.Logger
 		return err
 	}
 
-	n.l2Driver = driver.NewDriver(&cfg.Driver, &cfg.Rollup, n.l2Source, n.l1Source, n, n, n.log, snapshotLog, n.metrics)
+	n.l2Driver = driver.NewDriver(&cfg.Driver, &cfg.Rollup, n.l2Source, n.l1Source, n, n.coordinatorClient, n, n.log, snapshotLog, n.metrics)
 
 	return nil
 }

op-node/rollup/coordinator.go

@@ -0,0 +1,36 @@
+package rollup
+
+import (
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+type CoordinatorClient struct {
+	sequencerId string
+	rpc         *rpc.Client
+}
+
+func NewCoordinatorClient(url string, sequencerId string) (*CoordinatorClient, error) {
+	rpc, err := rpc.Dial(url)
+	if err != nil {
+		return nil, err
+	}
+	return &CoordinatorClient{
+		sequencerId: sequencerId,
+		rpc:         rpc,
+	}, nil
+}
+
+func (c *CoordinatorClient) RequestBuildingBlock() bool {
+	var respErr error
+	err := c.rpc.Call(respErr, "coordinator_requestBuildingBlock", c.sequencerId)
+	if err != nil {
+		log.Warn("Failed to call coordinator_requestBuildingBlock", "error", err)
+		return false
+	}
+	if respErr != nil {
+		log.Warn("coordinator_requestBuildingBlock refused request", "error", respErr)
+		return false
+	}
+	return true
+}

op-node/rollup/driver/driver.go

@@ -102,7 +102,7 @@ type AltSync interface {
 }
 
 // NewDriver composes an events handler that tracks L1 state, triggers L2 derivation, and optionally sequences new L2 blocks.
-func NewDriver(driverCfg *Config, cfg *rollup.Config, l2 L2Chain, l1 L1Chain, altSync AltSync, network Network, log log.Logger, snapshotLog log.Logger, metrics Metrics) *Driver {
+func NewDriver(driverCfg *Config, cfg *rollup.Config, l2 L2Chain, l1 L1Chain, altSync AltSync, coordinatorClient *rollup.CoordinatorClient, network Network, log log.Logger, snapshotLog log.Logger, metrics Metrics) *Driver {
 	l1State := NewL1State(log, metrics)
 	sequencerConfDepth := NewConfDepth(driverCfg.SequencerConfDepth, l1State.L1Head, l1)
 	findL1Origin := NewL1OriginSelector(log, cfg, sequencerConfDepth)
@@ -111,7 +111,7 @@ func NewDriver(driverCfg *Config, cfg *rollup.Config, l2 L2Chain, l1 L1Chain, al
 	attrBuilder := derive.NewFetchingAttributesBuilder(cfg, l1, l2)
 	engine := derivationPipeline
 	meteredEngine := NewMeteredEngine(cfg, engine, metrics, log)
-	sequencer := NewSequencer(log, cfg, meteredEngine, attrBuilder, findL1Origin, metrics)
+	sequencer := NewSequencer(log, cfg, meteredEngine, coordinatorClient, attrBuilder, findL1Origin, metrics)
 
 	return &Driver{
 		l1State:          l1State,

op-node/rollup/driver/sequencer.go

@@ -34,6 +34,8 @@ type Sequencer struct {
 	log    log.Logger
 	config *rollup.Config
 
+	coordinatorClient *rollup.CoordinatorClient
+
 	engine derive.ResettableEngineControl
 
 	attrBuilder      derive.AttributesBuilder
@@ -47,20 +49,27 @@ type Sequencer struct {
 	nextAction time.Time
 }
 
-func NewSequencer(log log.Logger, cfg *rollup.Config, engine derive.ResettableEngineControl, attributesBuilder derive.AttributesBuilder, l1OriginSelector L1OriginSelectorIface, metrics SequencerMetrics) *Sequencer {
+func NewSequencer(log log.Logger, cfg *rollup.Config, engine derive.ResettableEngineControl, coordinatorClient *rollup.CoordinatorClient, attributesBuilder derive.AttributesBuilder, l1OriginSelector L1OriginSelectorIface, metrics SequencerMetrics) *Sequencer {
 	return &Sequencer{
-		log:              log,
-		config:           cfg,
-		engine:           engine,
-		timeNow:          time.Now,
-		attrBuilder:      attributesBuilder,
-		l1OriginSelector: l1OriginSelector,
-		metrics:          metrics,
+		log:               log,
+		config:            cfg,
+		engine:            engine,
+		coordinatorClient: coordinatorClient,
+		timeNow:           time.Now,
+		attrBuilder:       attributesBuilder,
+		l1OriginSelector:  l1OriginSelector,
+		metrics:           metrics,
 	}
 }
 
 // StartBuildingBlock initiates a block building job on top of the given L2 head, safe and finalized blocks, and using the provided l1Origin.
 func (d *Sequencer) StartBuildingBlock(ctx context.Context) error {
+	// External coordinator mode (configured by --coordinator.enabled=true): Sequencer requests permission to build
+	// new blocks from the external coordinator by calling RequestBuildingBlock().
+	if d.coordinatorClient != nil && !d.coordinatorClient.RequestBuildingBlock() {
+		return errors.New("failed to request permission for building block from coordinator")
+	}
+
 	l2Head := d.engine.UnsafeL2Head()
 
 	// Figure out which L1 origin block we're going to be building on top of.

op-node/service.go

@@ -56,13 +56,15 @@ func NewConfig(ctx *cli.Context, log log.Logger) (*node.Config, error) {
 	}
 
 	l2SyncEndpoint := NewL2SyncEndpointConfig(ctx)
+	coordinator := NewCoordinatorConfig(ctx)
 
 	cfg := &node.Config{
-		L1:     l1Endpoint,
-		L2:     l2Endpoint,
-		L2Sync: l2SyncEndpoint,
-		Rollup: *rollupConfig,
-		Driver: *driverConfig,
+		L1:          l1Endpoint,
+		L2:          l2Endpoint,
+		L2Sync:      l2SyncEndpoint,
+		Coordinator: *coordinator,
+		Rollup:      *rollupConfig,
+		Driver:      *driverConfig,
 		RPC: node.RPCConfig{
 			ListenAddr:  ctx.GlobalString(flags.RPCListenAddr.Name),
 			ListenPort:  ctx.GlobalInt(flags.RPCListenPort.Name),
@@ -143,6 +145,14 @@ func NewL2SyncEndpointConfig(ctx *cli.Context) *node.L2SyncEndpointConfig {
 	}
 }
 
+func NewCoordinatorConfig(ctx *cli.Context) *node.CoordinatorConfig {
+	return &node.CoordinatorConfig{
+		Enabled:         ctx.GlobalBool(flags.CoordinatorEnabledFlag.Name),
+		CoordinatorAddr: ctx.GlobalString(flags.CoordinatorAddrFlag.Name),
+		SequencerId:     ctx.GlobalString(flags.CoordinatorSequencerIdFlag.Name),
+	}
+}
+
 func NewDriverConfig(ctx *cli.Context) *driver.Config {
 	return &driver.Config{
 		VerifierConfDepth:   ctx.GlobalUint64(flags.VerifierL1Confs.Name),