[op-conductor] e2e tests part 1 - test setup (#8886)

* Seq HA e2e test setup

* Finish test setup

* Finish test setup

* Update

* Fix test flakiness

* Fix test flakiness

* Fix typo

* Fix flakiness of main test

* Minor update

* Fix merge issue

op-conductor/conductor/config.go

@@ -2,6 +2,7 @@ package conductor
 
 import (
 	"fmt"
+	"math"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/pkg/errors"
@@ -58,8 +59,8 @@ func (c *Config) Check() error {
 	if c.ConsensusAddr == "" {
 		return fmt.Errorf("missing consensus address")
 	}
-	if c.ConsensusPort == 0 {
-		return fmt.Errorf("missing consensus port")
+	if c.ConsensusPort < 0 || c.ConsensusPort > math.MaxUint16 {
+		return fmt.Errorf("invalid RPC port")
 	}
 	if c.RaftServerID == "" {
 		return fmt.Errorf("missing raft server ID")

op-conductor/conductor/service.go

@@ -339,6 +339,13 @@ func (oc *OpConductor) Paused() bool {
 	return oc.paused.Load()
 }
 
+func (oc *OpConductor) HTTPEndpoint() string {
+	if oc.rpcServer == nil {
+		return ""
+	}
+	return fmt.Sprintf("http://%s", oc.rpcServer.Endpoint())
+}
+
 // Leader returns true if OpConductor is the leader.
 func (oc *OpConductor) Leader(_ context.Context) bool {
 	return oc.cons.Leader()
@@ -379,6 +386,11 @@ func (oc *OpConductor) CommitUnsafePayload(_ context.Context, payload *eth.Execu
 	return oc.cons.CommitUnsafePayload(payload)
 }
 
+// SequencerHealthy returns true if sequencer is healthy.
+func (oc *OpConductor) SequencerHealthy(_ context.Context) bool {
+	return oc.healthy.Load()
+}
+
 func (oc *OpConductor) loop() {
 	defer oc.wg.Done()
 

op-conductor/consensus/raft_test.go

@@ -18,7 +18,7 @@ import (
 func TestCommitAndRead(t *testing.T) {
 	log := testlog.Logger(t, log.LvlInfo)
 	serverID := "SequencerA"
-	serverAddr := "127.0.0.1:50050"
+	serverAddr := "127.0.0.1:0"
 	bootstrap := true
 	now := uint64(time.Now().Unix())
 	rollupCfg := &rollup.Config{

op-conductor/rpc/api.go

@@ -17,6 +17,8 @@ type API interface {
 	Pause(ctx context.Context) error
 	// Resume resumes op-conductor.
 	Resume(ctx context.Context) error
+	// SequencerHealthy returns true if the sequencer is healthy.
+	SequencerHealthy(ctx context.Context) (bool, error)
 
 	// Consensus related APIs
 	// Leader returns true if the server is the leader.

op-conductor/rpc/backend.go

@@ -13,6 +13,7 @@ type conductor interface {
 	Resume(ctx context.Context) error
 	Paused() bool
 	Stopped() bool
+	SequencerHealthy(ctx context.Context) bool
 
 	Leader(ctx context.Context) bool
 	LeaderWithID(ctx context.Context) (string, string)
@@ -100,3 +101,8 @@ func (api *APIBackend) TransferLeader(ctx context.Context) error {
 func (api *APIBackend) TransferLeaderToServer(ctx context.Context, id string, addr string) error {
 	return api.con.TransferLeaderToServer(ctx, id, addr)
 }
+
+// SequencerHealthy implements API.
+func (api *APIBackend) SequencerHealthy(ctx context.Context) (bool, error) {
+	return api.con.SequencerHealthy(ctx), nil
+}

op-conductor/rpc/client.go

@@ -86,3 +86,10 @@ func (c *APIClient) TransferLeader(ctx context.Context) error {
 func (c *APIClient) TransferLeaderToServer(ctx context.Context, id string, addr string) error {
 	return c.c.CallContext(ctx, nil, prefixRPC("transferLeaderToServer"), id, addr)
 }
+
+// SequencerHealthy implements API.
+func (c *APIClient) SequencerHealthy(ctx context.Context) (bool, error) {
+	var healthy bool
+	err := c.c.CallContext(ctx, &healthy, prefixRPC("sequencerHealthy"))
+	return healthy, err
+}

op-e2e/e2eutils/wait/blocks.go

@@ -6,9 +6,11 @@ import (
 	"math/big"
 	"time"
 
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/rpc"
+
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 	"github.com/ethereum-optimism/optimism/op-service/sources"
-	"github.com/ethereum/go-ethereum/core/types"
 )
 
 // BlockCaller is a subset of the [ethclient.Client] interface
@@ -20,20 +22,21 @@ type BlockCaller interface {
 
 func ForBlock(ctx context.Context, client BlockCaller, n uint64) error {
 	for {
-		if ctx.Done() != nil {
+		select {
+		case <-ctx.Done():
 			return ctx.Err()
+		default:
+			height, err := client.BlockNumber(ctx)
+			if err != nil {
+				return err
+			}
+			if height < n {
+				time.Sleep(500 * time.Millisecond)
+				continue
+			}
+			return nil
 		}
-		height, err := client.BlockNumber(ctx)
-		if err != nil {
-			return err
-		}
-		if height < n {
-			time.Sleep(500 * time.Millisecond)
-			continue
-		}
-		break
 	}
-	return nil
 }
 
 func ForBlockWithTimestamp(ctx context.Context, client BlockCaller, target uint64) error {
@@ -68,3 +71,42 @@ func ForProcessingFullBatch(ctx context.Context, rollupCl *sources.RollupClient)
 	})
 	return err
 }
+
+func ForUnsafeBlock(ctx context.Context, rollupCl *sources.RollupClient, n uint64) error {
+	ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
+	defer cancel()
+
+	_, err := AndGet(ctx, time.Second, func() (*eth.SyncStatus, error) {
+		return rollupCl.SyncStatus(ctx)
+	}, func(syncStatus *eth.SyncStatus) bool {
+		return syncStatus.UnsafeL2.Number >= n
+	})
+	return err
+}
+
+func ForNextSafeBlock(ctx context.Context, client BlockCaller) error {
+	safeBlockNumber := big.NewInt(rpc.SafeBlockNumber.Int64())
+	current, err := client.BlockByNumber(ctx, safeBlockNumber)
+	if err != nil {
+		return err
+	}
+
+	// Long timeout so we don't have to care what the block time is. If the test passes this will complete early anyway.
+	ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
+	defer cancel()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			next, err := client.BlockByNumber(ctx, safeBlockNumber)
+			if err != nil {
+				return err
+			}
+			if next.NumberU64() > current.NumberU64() {
+				return nil
+			}
+			time.Sleep(500 * time.Millisecond)
+		}
+	}
+}

op-e2e/sequencer_failover_test.go

+package op_e2e
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// [Category: Initial Setup]
+// In this test, we test that we can successfully setup a working cluster.
+func TestSequencerFailover_SetupCluster(t *testing.T) {
+	sys, conductors := setupSequencerFailoverTest(t)
+	defer sys.Close()
+
+	require.Equal(t, 3, len(conductors), "Expected 3 conductors")
+	for _, con := range conductors {
+		require.NotNil(t, con, "Expected conductor to be non-nil")
+	}
+}

op-e2e/system_adminrpc_test.go

@@ -202,10 +202,7 @@ func TestPostUnsafePayload(t *testing.T) {
 	require.NoError(t, err)
 	err = rollupClient.PostUnsafePayload(ctx, payload)
 	require.NoError(t, err)
-
-	ss, err := rollupClient.SyncStatus(ctx)
-	require.NoError(t, err)
-	require.Equal(t, uint64(1), ss.UnsafeL2.Number)
+	require.NoError(t, wait.ForUnsafeBlock(ctx, rollupClient, 1), "Chain did not advance after posting payload")
 
 	// Test validation
 	blockNumberTwo, err := l2Seq.BlockByNumber(ctx, big.NewInt(2))

op-service/rpc/server.go

@@ -10,13 +10,13 @@ import (
 	"strconv"
 	"time"
 
-	oplog "github.com/ethereum-optimism/optimism/op-service/log"
-	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
-	optls "github.com/ethereum-optimism/optimism/op-service/tls"
-
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/rpc"
+
+	oplog "github.com/ethereum-optimism/optimism/op-service/log"
+	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
+	optls "github.com/ethereum-optimism/optimism/op-service/tls"
 )
 
 var wildcardHosts = []string{"*"}
@@ -33,6 +33,7 @@ type Server struct {
 	healthzPath    string
 	httpRecorder   opmetrics.HTTPRecorder
 	httpServer     *http.Server
+	listener       net.Listener
 	log            log.Logger
 	tls            *ServerTLSConfig
 	middlewares    []Middleware
@@ -149,7 +150,7 @@ func NewServer(host string, port int, appVersion string, opts ...ServerOption) *
 }
 
 func (b *Server) Endpoint() string {
-	return b.endpoint
+	return b.listener.Addr().String()
 }
 
 func (b *Server) AddAPI(api rpc.API) {
@@ -180,14 +181,22 @@ func (b *Server) Start() error {
 	handler = oplog.NewLoggingMiddleware(b.log, handler)
 	b.httpServer.Handler = handler
 
+	listener, err := net.Listen("tcp", b.endpoint)
+	if err != nil {
+		return fmt.Errorf("failed to listen: %w", err)
+	}
+	b.listener = listener
+	// override endpoint with the actual listener address, in case the port was 0 during test.
+	b.httpServer.Addr = listener.Addr().String()
+	b.endpoint = listener.Addr().String()
 	errCh := make(chan error, 1)
 	go func() {
 		if b.tls != nil {
-			if err := b.httpServer.ListenAndServeTLS("", ""); err != nil {
+			if err := b.httpServer.ServeTLS(b.listener, "", ""); err != nil {
 				errCh <- err
 			}
 		} else {
-			if err := b.httpServer.ListenAndServe(); err != nil {
+			if err := b.httpServer.Serve(b.listener); err != nil {
 				errCh <- err
 			}
 		}

op-service/rpc/server_test.go

@@ -3,8 +3,9 @@ package rpc
 import (
 	"fmt"
 	"io"
-	"math/rand"
+	"net"
 	"net/http"
+	"strconv"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/rpc"
@@ -21,7 +22,7 @@ func TestBaseServer(t *testing.T) {
 	appVersion := "test"
 	server := NewServer(
 		"127.0.0.1",
-		10000+rand.Intn(22768),
+		0,
 		appVersion,
 		WithAPIs([]rpc.API{
 			{
@@ -58,4 +59,13 @@ func TestBaseServer(t *testing.T) {
 		require.NoError(t, rpcClient.Call(&res, "test_frobnicate", 2))
 		require.Equal(t, 4, res)
 	})
+
+	t.Run("supports 0 port", func(t *testing.T) {
+		endpoint := server.Endpoint()
+		_, portStr, err := net.SplitHostPort(endpoint)
+		require.NoError(t, err)
+		port, err := strconv.Atoi(portStr)
+		require.NoError(t, err)
+		require.Greater(t, port, 0)
+	})
 }