Finish conductor rpc e2e tests (#8929)

op-conductor/conductor/service.go

@@ -405,7 +405,7 @@ func (oc *OpConductor) Leader(_ context.Context) bool {
 }
 
 // LeaderWithID returns the current leader's server ID and address.
-func (oc *OpConductor) LeaderWithID(_ context.Context) (string, string) {
+func (oc *OpConductor) LeaderWithID(_ context.Context) *consensus.ServerInfo {
 	return oc.cons.LeaderWithID()
 }
 
@@ -444,6 +444,16 @@ func (oc *OpConductor) SequencerHealthy(_ context.Context) bool {
 	return oc.healthy.Load()
 }
 
+// ClusterMembership returns current cluster's membership information.
+func (oc *OpConductor) ClusterMembership(_ context.Context) ([]*consensus.ServerInfo, error) {
+	return oc.cons.ClusterMembership()
+}
+
+// LatestUnsafePayload returns the latest unsafe payload envelope from FSM.
+func (oc *OpConductor) LatestUnsafePayload(_ context.Context) *eth.ExecutionPayloadEnvelope {
+	return oc.cons.LatestUnsafePayload()
+}
+
 func (oc *OpConductor) loop() {
 	defer oc.wg.Done()
 

op-conductor/consensus/iface.go

@@ -4,6 +4,34 @@ import (
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
 
+// ServerSuffrage determines whether a Server in a Configuration gets a vote.
+type ServerSuffrage int
+
+const (
+	// Voter is a server whose vote is counted in elections.
+	Voter ServerSuffrage = iota
+	// Nonvoter is a server that receives log entries but is not considered for
+	// elections or commitment purposes.
+	Nonvoter
+)
+
+func (s ServerSuffrage) String() string {
+	switch s {
+	case Voter:
+		return "Voter"
+	case Nonvoter:
+		return "Nonvoter"
+	}
+	return "ServerSuffrage"
+}
+
+// ServerInfo defines the server information.
+type ServerInfo struct {
+	ID       string         `json:"id"`
+	Addr     string         `json:"addr"`
+	Suffrage ServerSuffrage `json:"suffrage"`
+}
+
 // Consensus defines the consensus interface for leadership election.
 //
 //go:generate mockery --name Consensus --output mocks/ --with-expecter=true
@@ -21,13 +49,15 @@ type Consensus interface {
 	// Leader returns if it is the leader of the cluster.
 	Leader() bool
 	// LeaderWithID returns the leader's server ID and address.
-	LeaderWithID() (string, string)
+	LeaderWithID() *ServerInfo
 	// ServerID returns the server ID of the consensus.
 	ServerID() string
 	// TransferLeader triggers leadership transfer to another member in the cluster.
 	TransferLeader() error
 	// TransferLeaderTo triggers leadership transfer to a specific member in the cluster.
 	TransferLeaderTo(id, addr string) error
+	// ClusterMembership returns the current cluster membership configuration.
+	ClusterMembership() ([]*ServerInfo, error)
 
 	// CommitPayload commits latest unsafe payload to the FSM.
 	CommitUnsafePayload(payload *eth.ExecutionPayloadEnvelope) error

op-conductor/consensus/mocks/Consensus.go

@@ -3,7 +3,9 @@
 package mocks
 
 import (
+	consensus "github.com/ethereum-optimism/optimism/op-conductor/consensus"
 	eth "github.com/ethereum-optimism/optimism/op-service/eth"
+
 	mock "github.com/stretchr/testify/mock"
 )
 
@@ -114,6 +116,63 @@ func (_c *Consensus_AddVoter_Call) RunAndReturn(run func(string, string) error)
 	return _c
 }
 
+// ClusterMembership provides a mock function with given fields:
+func (_m *Consensus) ClusterMembership() ([]*consensus.ServerInfo, error) {
+	ret := _m.Called()
+
+	if len(ret) == 0 {
+		panic("no return value specified for ClusterMembership")
+	}
+
+	var r0 []*consensus.ServerInfo
+	var r1 error
+	if rf, ok := ret.Get(0).(func() ([]*consensus.ServerInfo, error)); ok {
+		return rf()
+	}
+	if rf, ok := ret.Get(0).(func() []*consensus.ServerInfo); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*consensus.ServerInfo)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func() error); ok {
+		r1 = rf()
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Consensus_ClusterMembership_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ClusterMembership'
+type Consensus_ClusterMembership_Call struct {
+	*mock.Call
+}
+
+// ClusterMembership is a helper method to define mock.On call
+func (_e *Consensus_Expecter) ClusterMembership() *Consensus_ClusterMembership_Call {
+	return &Consensus_ClusterMembership_Call{Call: _e.mock.On("ClusterMembership")}
+}
+
+func (_c *Consensus_ClusterMembership_Call) Run(run func()) *Consensus_ClusterMembership_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run()
+	})
+	return _c
+}
+
+func (_c *Consensus_ClusterMembership_Call) Return(_a0 []*consensus.ServerInfo, _a1 error) *Consensus_ClusterMembership_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *Consensus_ClusterMembership_Call) RunAndReturn(run func() ([]*consensus.ServerInfo, error)) *Consensus_ClusterMembership_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // CommitUnsafePayload provides a mock function with given fields: payload
 func (_m *Consensus) CommitUnsafePayload(payload *eth.ExecutionPayloadEnvelope) error {
 	ret := _m.Called(payload)
@@ -346,31 +405,23 @@ func (_c *Consensus_LeaderCh_Call) RunAndReturn(run func() <-chan bool) *Consens
 }
 
 // LeaderWithID provides a mock function with given fields:
-func (_m *Consensus) LeaderWithID() (string, string) {
+func (_m *Consensus) LeaderWithID() *consensus.ServerInfo {
 	ret := _m.Called()
 
 	if len(ret) == 0 {
 		panic("no return value specified for LeaderWithID")
 	}
 
-	var r0 string
-	var r1 string
-	if rf, ok := ret.Get(0).(func() (string, string)); ok {
-		return rf()
-	}
-	if rf, ok := ret.Get(0).(func() string); ok {
+	var r0 *consensus.ServerInfo
+	if rf, ok := ret.Get(0).(func() *consensus.ServerInfo); ok {
 		r0 = rf()
 	} else {
-		r0 = ret.Get(0).(string)
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*consensus.ServerInfo)
 		}
-
-	if rf, ok := ret.Get(1).(func() string); ok {
-		r1 = rf()
-	} else {
-		r1 = ret.Get(1).(string)
 	}
 
-	return r0, r1
+	return r0
 }
 
 // Consensus_LeaderWithID_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'LeaderWithID'
@@ -390,12 +441,12 @@ func (_c *Consensus_LeaderWithID_Call) Run(run func()) *Consensus_LeaderWithID_C
 	return _c
 }
 
-func (_c *Consensus_LeaderWithID_Call) Return(_a0 string, _a1 string) *Consensus_LeaderWithID_Call {
-	_c.Call.Return(_a0, _a1)
+func (_c *Consensus_LeaderWithID_Call) Return(_a0 *consensus.ServerInfo) *Consensus_LeaderWithID_Call {
+	_c.Call.Return(_a0)
 	return _c
 }
 
-func (_c *Consensus_LeaderWithID_Call) RunAndReturn(run func() (string, string)) *Consensus_LeaderWithID_Call {
+func (_c *Consensus_LeaderWithID_Call) RunAndReturn(run func() *consensus.ServerInfo) *Consensus_LeaderWithID_Call {
 	_c.Call.Return(run)
 	return _c
 }

op-conductor/consensus/raft.go

@@ -145,9 +145,13 @@ func (rc *RaftConsensus) Leader() bool {
 }
 
 // LeaderWithID implements Consensus, it returns the leader's server ID and address.
-func (rc *RaftConsensus) LeaderWithID() (string, string) {
+func (rc *RaftConsensus) LeaderWithID() *ServerInfo {
 	addr, id := rc.r.LeaderWithID()
-	return string(id), string(addr)
+	return &ServerInfo{
+		ID:       string(id),
+		Addr:     string(addr),
+		Suffrage: Voter, // leader will always be Voter
+	}
 }
 
 // LeaderCh implements Consensus, it returns a channel that will be notified when leadership status changes (true = leader, false = follower).
@@ -221,3 +225,21 @@ func (rc *RaftConsensus) LatestUnsafePayload() *eth.ExecutionPayloadEnvelope {
 	payload := rc.unsafeTracker.UnsafeHead()
 	return payload
 }
+
+// ClusterMembership implements Consensus, it returns the current cluster membership configuration.
+func (rc *RaftConsensus) ClusterMembership() ([]*ServerInfo, error) {
+	var future raft.ConfigurationFuture
+	if future = rc.r.GetConfiguration(); future.Error() != nil {
+		return nil, future.Error()
+	}
+
+	var servers []*ServerInfo
+	for _, srv := range future.Configuration().Servers {
+		servers = append(servers, &ServerInfo{
+			ID:       string(srv.ID),
+			Addr:     string(srv.Address),
+			Suffrage: ServerSuffrage(srv.Suffrage),
+		})
+	}
+	return servers, nil
+}

op-conductor/rpc/api.go

@@ -6,17 +6,13 @@ import (
 
 	"github.com/ethereum/go-ethereum/rpc"
 
+	"github.com/ethereum-optimism/optimism/op-conductor/consensus"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
 
 var ErrNotLeader = errors.New("refusing to proxy request to non-leader sequencer")
 
-type ServerInfo struct {
-	ID   string `json:"id"`
-	Addr string `json:"addr"`
-}
-
 // API defines the interface for the op-conductor API.
 type API interface {
 	// Pause pauses op-conductor.
@@ -30,7 +26,7 @@ type API interface {
 	// Leader returns true if the server is the leader.
 	Leader(ctx context.Context) (bool, error)
 	// LeaderWithID returns the current leader's server info.
-	LeaderWithID(ctx context.Context) (*ServerInfo, error)
+	LeaderWithID(ctx context.Context) (*consensus.ServerInfo, error)
 	// AddServerAsVoter adds a server as a voter to the cluster.
 	AddServerAsVoter(ctx context.Context, id string, addr string) error
 	// AddServerAsNonvoter adds a server as a non-voter to the cluster. non-voter will not participate in leader election.
@@ -41,6 +37,8 @@ type API interface {
 	TransferLeader(ctx context.Context) error
 	// TransferLeaderToServer transfers leadership to a specific server.
 	TransferLeaderToServer(ctx context.Context, id string, addr string) error
+	// ClusterMembership returns the current cluster membership configuration.
+	ClusterMembership(ctx context.Context) ([]*consensus.ServerInfo, error)
 
 	// APIs called by op-node
 	// Active returns true if op-conductor is active.

op-conductor/rpc/backend.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/log"
 
+	"github.com/ethereum-optimism/optimism/op-conductor/consensus"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
 
@@ -16,13 +17,14 @@ type conductor interface {
 	SequencerHealthy(ctx context.Context) bool
 
 	Leader(ctx context.Context) bool
-	LeaderWithID(ctx context.Context) (string, string)
+	LeaderWithID(ctx context.Context) *consensus.ServerInfo
 	AddServerAsVoter(ctx context.Context, id string, addr string) error
 	AddServerAsNonvoter(ctx context.Context, id string, addr string) error
 	RemoveServer(ctx context.Context, id string) error
 	TransferLeader(ctx context.Context) error
 	TransferLeaderToServer(ctx context.Context, id string, addr string) error
 	CommitUnsafePayload(ctx context.Context, payload *eth.ExecutionPayloadEnvelope) error
+	ClusterMembership(ctx context.Context) ([]*consensus.ServerInfo, error)
 }
 
 // APIBackend is the backend implementation of the API.
@@ -69,12 +71,8 @@ func (api *APIBackend) Leader(ctx context.Context) (bool, error) {
 }
 
 // LeaderWithID implements API, returns the leader's server ID and address (not necessarily the current conductor).
-func (api *APIBackend) LeaderWithID(ctx context.Context) (*ServerInfo, error) {
-	id, addr := api.con.LeaderWithID(ctx)
-	return &ServerInfo{
-		ID:   id,
-		Addr: addr,
-	}, nil
+func (api *APIBackend) LeaderWithID(ctx context.Context) (*consensus.ServerInfo, error) {
+	return api.con.LeaderWithID(ctx), nil
 }
 
 // Pause implements API.
@@ -92,12 +90,14 @@ func (api *APIBackend) Resume(ctx context.Context) error {
 	return api.con.Resume(ctx)
 }
 
-// TransferLeader implements API.
+// TransferLeader implements API. With Raft implementation, a successful call does not mean that leadership transfer is complete
+// It just means that leadership transfer is in progress (current leader has initiated a new leader election round and stepped down as leader)
 func (api *APIBackend) TransferLeader(ctx context.Context) error {
 	return api.con.TransferLeader(ctx)
 }
 
-// TransferLeaderToServer implements API.
+// TransferLeaderToServer implements API. With Raft implementation, a successful call does not mean that leadership transfer is complete
+// It just means that leadership transfer is in progress (current leader has initiated a new leader election round and stepped down as leader)
 func (api *APIBackend) TransferLeaderToServer(ctx context.Context, id string, addr string) error {
 	return api.con.TransferLeaderToServer(ctx, id, addr)
 }
@@ -106,3 +106,8 @@ func (api *APIBackend) TransferLeaderToServer(ctx context.Context, id string, ad
 func (api *APIBackend) SequencerHealthy(ctx context.Context) (bool, error) {
 	return api.con.SequencerHealthy(ctx), nil
 }
+
+// ClusterMembership implements API.
+func (api *APIBackend) ClusterMembership(ctx context.Context) ([]*consensus.ServerInfo, error) {
+	return api.con.ClusterMembership(ctx)
+}

op-conductor/rpc/client.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/rpc"
 
+	"github.com/ethereum-optimism/optimism/op-conductor/consensus"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
 
@@ -61,8 +62,8 @@ func (c *APIClient) Leader(ctx context.Context) (bool, error) {
 }
 
 // LeaderWithID implements API.
-func (c *APIClient) LeaderWithID(ctx context.Context) (*ServerInfo, error) {
-	var info *ServerInfo
+func (c *APIClient) LeaderWithID(ctx context.Context) (*consensus.ServerInfo, error) {
+	var info *consensus.ServerInfo
 	err := c.c.CallContext(ctx, &info, prefixRPC("leaderWithID"))
 	return info, err
 }
@@ -98,3 +99,10 @@ func (c *APIClient) SequencerHealthy(ctx context.Context) (bool, error) {
 	err := c.c.CallContext(ctx, &healthy, prefixRPC("sequencerHealthy"))
 	return healthy, err
 }
+
+// ClusterMembership implements API.
+func (c *APIClient) ClusterMembership(ctx context.Context) ([]*consensus.ServerInfo, error) {
+	var info []*consensus.ServerInfo
+	err := c.c.CallContext(ctx, &info, prefixRPC("clusterMembership"))
+	return info, err
+}

op-e2e/sequencer_failover_setup.go

@@ -299,19 +299,15 @@ func sequencerCfg(rpcPort int) *rollupNode.Config {
 	}
 }
 
-func waitForLeadershipChange(t *testing.T, c *conductor, leader bool) error {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+func waitFor(t *testing.T, duration time.Duration, condition func() bool) error {
+	ctx, cancel := context.WithTimeout(context.Background(), duration)
 	defer cancel()
 	for {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
 		default:
-			isLeader, err := c.client.Leader(ctx)
-			if err != nil {
-				return err
-			}
-			if isLeader == leader {
+			if condition() {
 				return nil
 			}
 			time.Sleep(500 * time.Millisecond)
@@ -319,6 +315,30 @@ func waitForLeadershipChange(t *testing.T, c *conductor, leader bool) error {
 	}
 }
 
+func waitForLeadershipChange(t *testing.T, c *conductor, leader bool) error {
+	condiction := func() bool {
+		isLeader, err := c.client.Leader(context.Background())
+		if err != nil {
+			return false
+		}
+		return isLeader == leader
+	}
+
+	return waitFor(t, 10*time.Second, condiction)
+}
+
+func waitForSequencerStatusChange(t *testing.T, rollupClient *sources.RollupClient, active bool) error {
+	condiction := func() bool {
+		isActive, err := rollupClient.SequencerActive(context.Background())
+		if err != nil {
+			return false
+		}
+		return isActive == active
+	}
+
+	return waitFor(t, 5*time.Second, condiction)
+}
+
 func leader(t *testing.T, ctx context.Context, con *conductor) bool {
 	leader, err := con.client.Leader(ctx)
 	require.NoError(t, err)
@@ -371,3 +391,12 @@ func findLeader(t *testing.T, conductors map[string]*conductor) (string, *conduc
 	}
 	return "", nil
 }
+
+func findFollower(t *testing.T, conductors map[string]*conductor) (string, *conductor) {
+	for id, con := range conductors {
+		if !leader(t, context.Background(), con) {
+			return id, con
+		}
+	}
+	return "", nil
+}

op-e2e/sequencer_failover_test.go

 package op_e2e
 
 import (
+	"context"
+	"sort"
 	"testing"
 
 	"github.com/stretchr/testify/require"
+
+	"github.com/ethereum-optimism/optimism/op-conductor/consensus"
 )
 
 // [Category: Initial Setup]
@@ -17,3 +21,124 @@ func TestSequencerFailover_SetupCluster(t *testing.T) {
 		require.NotNil(t, con, "Expected conductor to be non-nil")
 	}
 }
+
+// [Category: conductor rpc]
+// In this test, we test all rpcs exposed by conductor.
+func TestSequencerFailover_ConductorRPC(t *testing.T) {
+	ctx := context.Background()
+	sys, conductors := setupSequencerFailoverTest(t)
+	defer sys.Close()
+
+	// SequencerHealthy, Leader, AddServerAsVoter are used in setup already.
+
+	// Test ClusterMembership
+	t.Log("Testing ClusterMembership")
+	c1 := conductors[Sequencer1Name]
+	c2 := conductors[Sequencer2Name]
+	c3 := conductors[Sequencer3Name]
+	membership, err := c1.client.ClusterMembership(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 3, len(membership), "Expected 3 members in cluster")
+	ids := make([]string, 0)
+	for _, member := range membership {
+		ids = append(ids, member.ID)
+		require.Equal(t, consensus.Voter, member.Suffrage, "Expected all members to be voters")
+	}
+	sort.Strings(ids)
+	require.Equal(t, []string{Sequencer1Name, Sequencer2Name, Sequencer3Name}, ids, "Expected all sequencers to be in cluster")
+
+	// Test Active & Pause & Resume
+	t.Log("Testing Active & Pause & Resume")
+	active, err := c1.client.Active(ctx)
+	require.NoError(t, err)
+	require.True(t, active, "Expected conductor to be active")
+
+	err = c1.client.Pause(ctx)
+	require.NoError(t, err)
+	active, err = c1.client.Active(ctx)
+	require.NoError(t, err)
+	require.False(t, active, "Expected conductor to be paused")
+
+	err = c1.client.Resume(ctx)
+	require.NoError(t, err)
+	active, err = c1.client.Active(ctx)
+	require.NoError(t, err)
+	require.True(t, active, "Expected conductor to be active")
+
+	// Test LeaderWithID
+	t.Log("Testing LeaderWithID")
+	leader1, err := c1.client.LeaderWithID(ctx)
+	require.NoError(t, err)
+	leader2, err := c2.client.LeaderWithID(ctx)
+	require.NoError(t, err)
+	leader3, err := c3.client.LeaderWithID(ctx)
+	require.NoError(t, err)
+	require.Equal(t, leader1.ID, leader2.ID, "Expected leader ID to be the same")
+	require.Equal(t, leader1.ID, leader3.ID, "Expected leader ID to be the same")
+
+	// Test TransferLeader & TransferLeaderToServer
+	t.Log("Testing TransferLeader")
+	lid, leader := findLeader(t, conductors)
+	err = leader.client.TransferLeader(ctx)
+	require.NoError(t, err, "Expected leader to transfer leadership")
+	require.NoError(t, waitForLeadershipChange(t, leader, false))
+	newLeader, err := leader.client.LeaderWithID(ctx)
+	require.NoError(t, err)
+	isLeader, err := leader.client.Leader(ctx)
+	require.NoError(t, err)
+	require.False(t, isLeader, "Expected leader to transfer leadership")
+	require.NotEqual(t, lid, newLeader.ID, "Expected leader to change")
+	require.NoError(t, waitForSequencerStatusChange(t, sys.RollupClient(newLeader.ID), true))
+
+	// old leader now became follower, we're trying to transfer leadership directly back to it.
+	t.Log("Testing TransferLeaderToServer")
+	fid, follower := lid, leader
+	_, leader = findLeader(t, conductors)
+	err = leader.client.TransferLeaderToServer(ctx, fid, follower.ConsensusEndpoint())
+	require.NoError(t, err, "Expected leader to transfer leadership to follower")
+	require.NoError(t, waitForLeadershipChange(t, follower, true))
+	require.NoError(t, waitForSequencerStatusChange(t, sys.RollupClient(fid), true))
+	leader = follower
+
+	// Test AddServerAsNonvoter, do not start a new sequencer just for this purpose, use Sequencer3's rpc to start conductor.
+	// This is fine as this mainly tests conductor's ability to add itself into the raft consensus cluster as a nonvoter.
+	t.Log("Testing AddServerAsNonvoter")
+	nonvoter := setupConductor(
+		t, VerifierName, t.TempDir(),
+		sys.RollupEndpoint(Sequencer3Name),
+		sys.NodeEndpoint(Sequencer3Name),
+		findAvailablePort(t),
+		false,
+		*sys.RollupConfig,
+	)
+
+	err = leader.client.AddServerAsNonvoter(ctx, VerifierName, nonvoter.ConsensusEndpoint())
+	require.NoError(t, err, "Expected leader to add non-voter")
+	membership, err = leader.client.ClusterMembership(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 4, len(membership), "Expected 4 members in cluster")
+	require.Equal(t, consensus.Nonvoter, membership[3].Suffrage, "Expected last member to be non-voter")
+
+	t.Log("Testing RemoveServer")
+	lid, leader = findLeader(t, conductors)
+	fid, follower = findFollower(t, conductors)
+	err = follower.client.RemoveServer(ctx, lid)
+	require.ErrorContains(t, err, "node is not the leader", "Expected follower to fail to remove leader")
+	membership, err = c1.client.ClusterMembership(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 4, len(membership), "Expected 4 members in cluster")
+
+	err = leader.client.RemoveServer(ctx, VerifierName)
+	require.NoError(t, err, "Expected leader to remove non-voter")
+	membership, err = c1.client.ClusterMembership(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 3, len(membership), "Expected 2 members in cluster after removal")
+	require.NotContains(t, membership, VerifierName, "Expected follower to be removed from cluster")
+
+	err = leader.client.RemoveServer(ctx, fid)
+	require.NoError(t, err, "Expected leader to remove follower")
+	membership, err = c1.client.ClusterMembership(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 2, len(membership), "Expected 2 members in cluster after removal")
+	require.NotContains(t, membership, fid, "Expected follower to be removed from cluster")
+}