op-service, op-batcher, op-proposer: Active sequencer follow mode

op-batcher/batcher/config.go

@@ -20,10 +20,10 @@ type CLIConfig struct {
 	// L1EthRpc is the HTTP provider URL for L1.
 	L1EthRpc string
 
-	// L2EthRpc is the HTTP provider URL for the L2 execution engine.
+	// L2EthRpc is the HTTP provider URL for the L2 execution engine. A comma-separated list enables the active L2 provider. Such a list needs to match the number of RollupRpcs provided.
 	L2EthRpc string
 
-	// RollupRpc is the HTTP provider URL for the L2 rollup node.
+	// RollupRpc is the HTTP provider URL for the L2 rollup node. A comma-separated list enables the active L2 provider. Such a list needs to match the number of L2EthRpcs provided.
 	RollupRpc string
 
 	// MaxChannelDuration is the maximum duration (in #L1-blocks) to keep a

op-batcher/batcher/service.go

@@ -8,6 +8,7 @@ import (
 	"net"
 	_ "net/http/pprof"
 	"strconv"
+	"strings"
 	"sync/atomic"
 	"time"
 
@@ -125,9 +126,16 @@ func (bs *BatcherService) initRPCClients(ctx context.Context, cfg *CLIConfig) er
 	}
 	bs.L1Client = l1Client
 
-	endpointProvider, err := dial.NewStaticL2EndpointProvider(ctx, bs.Log, cfg.L2EthRpc, cfg.RollupRpc)
+	var endpointProvider dial.L2EndpointProvider
+	if strings.Contains(cfg.RollupRpc, ",") || strings.Contains(cfg.L2EthRpc, ",") {
+		rollupUrls := strings.Split(cfg.RollupRpc, ",")
+		ethUrls := strings.Split(cfg.L2EthRpc, ",")
+		endpointProvider, err = dial.NewActiveL2EndpointProvider(ctx, ethUrls, rollupUrls, dial.DefaultActiveSequencerFollowerCheckDuration, dial.DefaultDialTimeout, bs.Log, dial.DialEthClientInterfaceWithTimeout, dial.DialRollupClientInterfaceWithTimeout)
+	} else {
+		endpointProvider, err = dial.NewStaticL2EndpointProvider(ctx, bs.Log, cfg.L2EthRpc, cfg.RollupRpc)
+	}
 	if err != nil {
-		return fmt.Errorf("failed to create L2 endpoint provider: %w", err)
+		return fmt.Errorf("failed to build L2 endpoint provider: %w", err)
 	}
 	bs.EndpointProvider = endpointProvider
 

op-batcher/flags/flags.go

@@ -30,12 +30,12 @@ var (
 	}
 	L2EthRpcFlag = &cli.StringFlag{
 		Name:    "l2-eth-rpc",
-		Usage:   "HTTP provider URL for L2 execution engine",
+		Usage:   "HTTP provider URL for L2 execution engine. A comma-separated list enables the active L2 endpoint provider. Such a list needs to match the number of rollup-rpcs provided.",
 		EnvVars: prefixEnvVars("L2_ETH_RPC"),
 	}
 	RollupRpcFlag = &cli.StringFlag{
 		Name:    "rollup-rpc",
-		Usage:   "HTTP provider URL for Rollup node",
+		Usage:   "HTTP provider URL for Rollup node. A comma-separated list enables the active L2 endpoint provider. Such a list needs to match the number of l2-eth-rpcs provided.",
 		EnvVars: prefixEnvVars("ROLLUP_RPC"),
 	}
 	// Optional flags

op-proposer/flags/flags.go

@@ -29,7 +29,7 @@ var (
 	}
 	RollupRpcFlag = &cli.StringFlag{
 		Name:    "rollup-rpc",
-		Usage:   "HTTP provider URL for the rollup node",
+		Usage:   "HTTP provider URL for the rollup node. A comma-separated list enables the active rollup provider.",
 		EnvVars: prefixEnvVars("ROLLUP_RPC"),
 	}
 	L2OOAddressFlag = &cli.StringFlag{

op-proposer/proposer/config.go

@@ -22,7 +22,7 @@ type CLIConfig struct {
 	// L1EthRpc is the HTTP provider URL for L1.
 	L1EthRpc string
 
-	// RollupRpc is the HTTP provider URL for the rollup node.
+	// RollupRpc is the HTTP provider URL for the rollup node. A comma-separated list enables the active rollup provider.
 	RollupRpc string
 
 	// L2OOAddress is the L2OutputOracle contract address.

op-proposer/proposer/driver.go

@@ -206,7 +206,7 @@ func (l *L2OutputSubmitter) fetchOutput(ctx context.Context, block *big.Int) (*e
 	}
 	output, err := rollupClient.OutputAtBlock(ctx, block.Uint64())
 	if err != nil {
-		l.Log.Error("failed to fetch output at block %d: %w", block, err)
+		l.Log.Error("failed to fetch output at block", "block", block, "err", err)
 		return nil, false, err
 	}
 	if output.Version != supportedL2OutputVersion {

op-proposer/proposer/service.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net"
 	"strconv"
+	"strings"
 	"sync/atomic"
 	"time"
 
@@ -121,7 +122,13 @@ func (ps *ProposerService) initRPCClients(ctx context.Context, cfg *CLIConfig) e
 	}
 	ps.L1Client = l1Client
 
-	rollupProvider, err := dial.NewStaticL2RollupProvider(ctx, ps.Log, cfg.RollupRpc)
+	var rollupProvider dial.RollupProvider
+	if strings.Contains(cfg.RollupRpc, ",") {
+		rollupUrls := strings.Split(cfg.RollupRpc, ",")
+		rollupProvider, err = dial.NewActiveL2RollupProvider(ctx, rollupUrls, dial.DefaultActiveSequencerFollowerCheckDuration, dial.DefaultDialTimeout, ps.Log, dial.DialRollupClientInterfaceWithTimeout)
+	} else {
+		rollupProvider, err = dial.NewStaticL2RollupProvider(ctx, ps.Log, cfg.RollupRpc)
+	}
 	if err != nil {
 		return fmt.Errorf("failed to build L2 endpoint provider: %w", err)
 	}

op-service/dial/active_l2_provider.go

@@ -0,0 +1,131 @@
+package dial
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+)
+
+const DefaultActiveSequencerFollowerCheckDuration = 2 * DefaultDialTimeout
+
+type ethDialer func(ctx context.Context, timeout time.Duration, log log.Logger, url string) (EthClientInterface, error)
+
+type ActiveL2EndpointProvider struct {
+	ActiveL2RollupProvider
+	currentEthClient EthClientInterface
+	ethDialer        ethDialer
+}
+
+func NewActiveL2EndpointProvider(
+	ctx context.Context,
+	ethUrls, rollupUrls []string,
+	checkDuration time.Duration,
+	networkTimeout time.Duration,
+	logger log.Logger,
+	ethDialer ethDialer,
+	rollupDialer rollupDialer,
+) (*ActiveL2EndpointProvider, error) {
+	if len(rollupUrls) == 0 {
+		return nil, errors.New("empty rollup urls list")
+	}
+	if len(ethUrls) != len(rollupUrls) {
+		return nil, errors.New("number of eth urls and rollup urls mismatch")
+	}
+
+	rollupProvider, err := NewActiveL2RollupProvider(ctx, rollupUrls, checkDuration, networkTimeout, logger, rollupDialer)
+	if err != nil {
+		return nil, err
+	}
+	cctx, cancel := context.WithTimeout(ctx, networkTimeout)
+	defer cancel()
+	ethClient, err := ethDialer(cctx, networkTimeout, logger, ethUrls[0])
+	if err != nil {
+		return nil, fmt.Errorf("dialing eth client: %w", err)
+	}
+	return &ActiveL2EndpointProvider{
+		ActiveL2RollupProvider: *rollupProvider,
+		currentEthClient:       ethClient,
+		ethDialer:              ethDialer,
+	}, nil
+}
+
+func (p *ActiveL2EndpointProvider) EthClient(ctx context.Context) (EthClientInterface, error) {
+	p.clientLock.Lock()
+	defer p.clientLock.Unlock()
+	err := p.ensureActiveEndpoint(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return p.currentEthClient, nil
+}
+
+func (p *ActiveL2EndpointProvider) ensureActiveEndpoint(ctx context.Context) error {
+	if !p.shouldCheck() {
+		return nil
+	}
+
+	if err := p.findActiveEndpoints(ctx); err != nil {
+		return err
+	}
+	p.activeTimeout = time.Now().Add(p.checkDuration)
+	return nil
+}
+
+func (p *ActiveL2EndpointProvider) findActiveEndpoints(ctx context.Context) error {
+	const maxRetries = 20
+	totalAttempts := 0
+
+	for totalAttempts < maxRetries {
+		active, err := p.checkCurrentSequencer(ctx)
+		if err != nil {
+			p.log.Warn("Error querying active sequencer, closing connection and trying next.", "err", err, "try", totalAttempts)
+			p.currentRollupClient.Close()
+			p.currentEthClient.Close()
+		} else if active {
+			p.log.Debug("Current sequencer active.", "try", totalAttempts)
+			return nil
+		} else {
+			p.log.Info("Current sequencer inactive, closing connection and trying next.", "try", totalAttempts)
+			p.currentRollupClient.Close()
+			p.currentEthClient.Close()
+		}
+		if err := p.dialNextSequencer(ctx); err != nil {
+			return fmt.Errorf("dialing next sequencer: %w", err)
+		}
+
+		totalAttempts++
+		if p.currentIndex >= p.numEndpoints() {
+			p.currentIndex = 0
+		}
+	}
+	return fmt.Errorf("failed to find an active sequencer after %d retries", maxRetries)
+}
+
+func (p *ActiveL2EndpointProvider) dialNextSequencer(ctx context.Context) error {
+	cctx, cancel := context.WithTimeout(ctx, p.networkTimeout)
+	defer cancel()
+	p.currentIndex++
+	ep := p.rollupUrls[p.currentIndex]
+	p.log.Debug("Dialing next sequencer.", "url", ep)
+	rollupClient, err := p.rollupDialer(cctx, p.networkTimeout, p.log, ep)
+	if err != nil {
+		return fmt.Errorf("dialing rollup client: %w", err)
+	}
+	ethClient, err := p.ethDialer(cctx, p.networkTimeout, p.log, ep)
+	if err != nil {
+		return fmt.Errorf("dialing eth client: %w", err)
+	}
+
+	p.currentRollupClient = rollupClient
+	p.currentEthClient = ethClient
+	return nil
+}
+
+func (p *ActiveL2EndpointProvider) Close() {
+	p.currentEthClient.Close()
+	p.ActiveL2RollupProvider.Close()
+}

op-service/dial/active_l2_provider_test.go

@@ -0,0 +1,104 @@
+package dial
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/ethereum-optimism/optimism/op-service/testlog"
+	"github.com/ethereum-optimism/optimism/op-service/testutils"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/stretchr/testify/require"
+)
+
+// TestActiveSequencerFailoverBehavior_RollupProvider tests that the ActiveL2RollupProvider
+// will failover to the next provider if the current one is not active.
+func TestActiveSequencerFailoverBehavior_RollupProviders(t *testing.T) {
+	// Create two mock rollup clients, one of which will declare itself inactive after first check.
+	primarySequencer := testutils.MockRollupClient{}
+	primarySequencer.ExpectSequencerActive(true, nil)
+	primarySequencer.ExpectSequencerActive(false, nil)
+	primarySequencer.ExpectClose()
+	secondarySequencer := testutils.MockRollupClient{}
+	secondarySequencer.ExpectSequencerActive(true, nil)
+
+	mockRollupDialer := func(ctx context.Context, timeout time.Duration, log log.Logger, url string) (RollupClientInterface, error) {
+		if url == "primary" {
+			return &primarySequencer, nil
+		} else if url == "secondary" {
+			return &secondarySequencer, nil
+		} else {
+			return nil, fmt.Errorf("unknown test url: %s", url)
+		}
+	}
+
+	endpointProvider, err := NewActiveL2RollupProvider(
+		context.Background(),
+		[]string{"primary", "secondary"},
+		1*time.Microsecond,
+		1*time.Minute,
+		testlog.Logger(t, log.LvlDebug),
+		mockRollupDialer,
+	)
+	require.NoError(t, err)
+	// Check that the first client is used, then the second once the first declares itself inactive.
+	firstSequencerUsed, err := endpointProvider.RollupClient(context.Background())
+	require.NoError(t, err)
+	require.True(t, &primarySequencer == firstSequencerUsed) // avoids copying the struct (and its mutex, etc.)
+	secondSequencerUsed, err := endpointProvider.RollupClient(context.Background())
+	require.NoError(t, err)
+	require.True(t, &secondarySequencer == secondSequencerUsed)
+}
+
+// TestActiveSequencerFailoverBehavior_L2Providers tests that the ActiveL2EndpointProvider
+// will failover to the next provider if the current one is not active.
+func TestActiveSequencerFailoverBehavior_L2Providers(t *testing.T) {
+	// as TestActiveSequencerFailoverBehavior_RollupProviders,
+	// but ensure the added `EthClient()` method also triggers the failover.
+	primarySequencer := testutils.MockRollupClient{}
+	primarySequencer.ExpectSequencerActive(true, nil)
+	primarySequencer.ExpectSequencerActive(false, nil)
+	primarySequencer.ExpectClose()
+	secondarySequencer := testutils.MockRollupClient{}
+	secondarySequencer.ExpectSequencerActive(true, nil)
+
+	mockRollupDialer := func(ctx context.Context, timeout time.Duration, log log.Logger, url string) (RollupClientInterface, error) {
+		if url == "primary" {
+			return &primarySequencer, nil
+		} else if url == "secondary" {
+			return &secondarySequencer, nil
+		} else {
+			return nil, fmt.Errorf("unknown test url: %s", url)
+		}
+	}
+	primaryEthClient := testutils.MockEthClient{}
+	primaryEthClient.ExpectClose()
+	secondaryEthClient := testutils.MockEthClient{}
+	mockEthDialer := func(ctx context.Context, timeout time.Duration, log log.Logger, url string) (EthClientInterface, error) {
+		if url == "primary" {
+			return &primaryEthClient, nil
+		} else if url == "secondary" {
+			return &secondaryEthClient, nil
+		} else {
+			return nil, fmt.Errorf("unknown test url: %s", url)
+		}
+	}
+	endpointProvider, err := NewActiveL2EndpointProvider(
+		context.Background(),
+		[]string{"primary", "secondary"},
+		[]string{"primary", "secondary"},
+		1*time.Microsecond,
+		1*time.Minute,
+		testlog.Logger(t, log.LvlDebug),
+		mockEthDialer,
+		mockRollupDialer,
+	)
+	require.NoError(t, err)
+	firstClientUsed, err := endpointProvider.EthClient(context.Background())
+	require.NoError(t, err)
+	require.True(t, &primaryEthClient == firstClientUsed) // avoids copying the struct (and its mutex, etc.)
+	secondClientUsed, err := endpointProvider.EthClient(context.Background())
+	require.NoError(t, err)
+	require.True(t, &secondaryEthClient == secondClientUsed)
+}