Refactor relay aggregator registrations and tests, add better logging

builder/local_relay.go

@@ -80,6 +80,7 @@ func NewLocalRelay(sk *bls.SecretKey, beaconClient IBeaconClient, builderSigning
 }
 
 func (r *LocalRelay) SubmitBlock(msg *boostTypes.BuilderSubmitBlockRequest, _ ValidatorData) error {
+	log.Info("submitting block to local relay", "block", msg.ExecutionPayload.BlockHash.String())
 	return r.submitBlock(msg)
 }
 

builder/relay.go

@@ -126,12 +126,13 @@ func (r *RemoteRelay) GetValidatorForSlot(nextSlot uint64) (ValidatorData, error
 }
 
 func (r *RemoteRelay) SubmitBlock(msg *boostTypes.BuilderSubmitBlockRequest, _ ValidatorData) error {
+	log.Info("submitting block to remote relay", "endpoint", r.endpoint)
 	code, err := server.SendHTTPRequest(context.TODO(), *http.DefaultClient, http.MethodPost, r.endpoint+"/relay/v1/builder/blocks", msg, nil)
 	if err != nil {
-		return err
+		return fmt.Errorf("error sending http request to relay %s. err: %w", r.endpoint, err)
 	}
 	if code > 299 {
-		return fmt.Errorf("non-ok response code %d from relay ", code)
+		return fmt.Errorf("non-ok response code %d from relay %s", code, r.endpoint)
 	}
 
 	if r.localRelay != nil {

builder/relay_aggregator.go

@@ -66,58 +66,59 @@ func (r *RemoteRelayAggregator) GetValidatorForSlot(nextSlot uint64) (ValidatorD
 	}
 
 	topRegistrationCh := make(chan ValidatorData, 1)
-	go func() {
-		registrations := make([]*RelayValidatorRegistration, 0, len(r.relays))
-		bestRelayIndex := len(r.relays)  // relay index of the topmost relay that gave us the registration
-		bestRelayRegistrationIndex := -1 // index into the registrations for the registration returned by topmost relay
-		for i := 0; i < len(r.relays); i++ {
-			relayRegistration := <-registrationsCh
-			if relayRegistration != nil {
-				registrations = append(registrations, relayRegistration)
-				// happy path for primary
-				if relayRegistration.relayI == 0 {
-					topRegistrationCh <- relayRegistration.vd
-				}
-				if relayRegistration.relayI < bestRelayIndex {
-					bestRelayIndex = relayRegistration.relayI
-					bestRelayRegistrationIndex = len(registrations) - 1
-				}
-			}
-		}
+	go r.updateRelayRegistrations(nextSlot, registrationsCh, topRegistrationCh)
 
-		if len(registrations) == 0 {
-			close(topRegistrationCh)
-			return
-		}
+	if vd, ok := <-topRegistrationCh; ok {
+		return vd, nil
+	}
+	return ValidatorData{}, ErrValidatorNotFound
+}
 
-		r.registrationsCacheLock.Lock()
-		defer r.registrationsCacheLock.Unlock()
+func (r *RemoteRelayAggregator) updateRelayRegistrations(nextSlot uint64, registrationsCh chan *RelayValidatorRegistration, topRegistrationCh chan ValidatorData) {
+	defer close(topRegistrationCh)
 
-		if nextSlot < r.registrationsCacheSlot {
-			// slot is outdated
-			close(topRegistrationCh)
-			return
-		}
+	r.registrationsCacheLock.Lock()
+	defer r.registrationsCacheLock.Unlock()
 
-		if nextSlot > r.registrationsCacheSlot {
-			// clear the cache
-			r.registrationsCache = make(map[ValidatorData][]IRelay)
-		}
+	if nextSlot < r.registrationsCacheSlot {
+		// slot is outdated
+		return
+	}
 
-		if bestRelayIndex != 0 {
-			// if bestRelayIndex == 0 it was already sent
-			topRegistrationCh <- registrations[bestRelayRegistrationIndex].vd
+	registrations := make([]*RelayValidatorRegistration, 0, len(r.relays))
+	bestRelayIndex := len(r.relays)  // relay index of the topmost relay that gave us the registration
+	bestRelayRegistrationIndex := -1 // index into the registrations for the registration returned by topmost relay
+	for i := 0; i < len(r.relays); i++ {
+		relayRegistration := <-registrationsCh
+		if relayRegistration != nil {
+			registrations = append(registrations, relayRegistration)
+			// happy path for primary
+			if relayRegistration.relayI == 0 {
+				topRegistrationCh <- relayRegistration.vd
+			}
+			if relayRegistration.relayI < bestRelayIndex {
+				bestRelayIndex = relayRegistration.relayI
+				bestRelayRegistrationIndex = len(registrations) - 1
+			}
 		}
+	}
 
-		close(topRegistrationCh)
+	if len(registrations) == 0 {
+		return
+	}
 
-		for _, relayRegistration := range registrations {
-			r.registrationsCache[relayRegistration.vd] = append(r.registrationsCache[relayRegistration.vd], r.relays[relayRegistration.relayI])
-		}
-	}()
+	if bestRelayIndex != 0 {
+		// if bestRelayIndex == 0 it was already sent
+		topRegistrationCh <- registrations[bestRelayRegistrationIndex].vd
+	}
 
-	if vd, ok := <-topRegistrationCh; ok {
-		return vd, nil
+	if nextSlot > r.registrationsCacheSlot {
+		// clear the cache
+		r.registrationsCache = make(map[ValidatorData][]IRelay)
+		r.registrationsCacheSlot = nextSlot
+	}
+
+	for _, relayRegistration := range registrations {
+		r.registrationsCache[relayRegistration.vd] = append(r.registrationsCache[relayRegistration.vd], r.relays[relayRegistration.relayI])
 	}
-	return ValidatorData{}, ErrValidatorNotFound
 }

builder/relay_aggregator_test.go

@@ -26,6 +26,24 @@ type testRelay struct {
 	submittedRegistration ValidatorData
 }
 
+type testRelayAggBackend struct {
+	relays []*testRelay
+	ragg   *RemoteRelayAggregator
+}
+
+func newTestRelayAggBackend(numRelay int) *testRelayAggBackend {
+	testRelays := make([]*testRelay, numRelay)
+	secondaryRelays := make([]IRelay, numRelay-1)
+	for i := 0; i < numRelay; i++ {
+		testRelays[i] = &testRelay{}
+		if i > 0 {
+			secondaryRelays[i-1] = testRelays[i]
+		}
+	}
+	ragg := NewRemoteRelayAggregator(testRelays[0], secondaryRelays)
+	return &testRelayAggBackend{testRelays, ragg}
+}
+
 func (r *testRelay) SubmitBlock(msg *boostTypes.BuilderSubmitBlockRequest, registration ValidatorData) error {
 	if r.submittedMsgCh != nil {
 		select {
@@ -42,100 +60,155 @@ func (r *testRelay) GetValidatorForSlot(nextSlot uint64) (ValidatorData, error)
 }
 
 func TestRemoteRelayAggregator(t *testing.T) {
-	testRelays := []*testRelay{&testRelay{}, &testRelay{}, &testRelay{}}
-	ragg := NewRemoteRelayAggregator(testRelays[0], []IRelay{testRelays[1], testRelays[2]})
+	t.Run("should return error if no relays return validator data", func(t *testing.T) {
+		backend := newTestRelayAggBackend(3)
+		// make all error out
+		for _, r := range backend.relays {
+			r.gvsErr = errors.New("error!")
+		}
 
-	// make all error out
-	for _, r := range testRelays {
-		r.gvsErr = errors.New("error!")
-	}
+		// Check getting validator slot - should error out if no relays return
+		_, err := backend.ragg.GetValidatorForSlot(10)
+		require.Error(t, ErrValidatorNotFound, err)
+	})
+
+	t.Run("should return validator if one relay returns validator data", func(t *testing.T) {
+		backend := newTestRelayAggBackend(3)
+
+		// If primary returns should not error out
+		backend.relays[1].gvsErr = errors.New("error!")
+		backend.relays[2].gvsErr = errors.New("error!")
+		_, err := backend.ragg.GetValidatorForSlot(10)
+		require.NoError(t, err)
+
+		// If any returns should not error out
+		backend.relays[0].gvsErr = errors.New("error!")
+		backend.relays[2].gvsErr = nil
+		_, err = backend.ragg.GetValidatorForSlot(10)
+		require.NoError(t, err)
+	})
+
+	t.Run("should return the more important (lower index) relay validator data", func(t *testing.T) {
+		backend := newTestRelayAggBackend(3)
+
+		// Should return the more important relay if primary fails
+		backend.relays[0].gvsErr = errors.New("error!")
+		backend.relays[1].gvsVd.GasLimit = 20
+		backend.relays[2].gvsVd.GasLimit = 30
+		vd, err := backend.ragg.GetValidatorForSlot(10)
+		require.NoError(t, err)
+		require.Equal(t, uint64(20), vd.GasLimit)
+
+		// Should return the primary if it returns
+		backend.relays[0].gvsErr = nil
+		backend.relays[0].gvsVd.GasLimit = 10
+		vd, err = backend.ragg.GetValidatorForSlot(11)
+		require.NoError(t, err)
+		require.Equal(t, uint64(10), vd.GasLimit)
+	})
+
+	t.Run("should error submitting to unseen validator data", func(t *testing.T) {
+		backend := newTestRelayAggBackend(3)
+
+		backend.relays[0].gvsVd.GasLimit = 10
+
+		vd, err := backend.ragg.GetValidatorForSlot(10)
+		require.NoError(t, err)
+		require.Equal(t, uint64(10), vd.GasLimit)
+
+		// let the validator registrations finish
+		// TODO: notify from the test relays
+		time.Sleep(10 * time.Millisecond)
+
+		// if submitting for unseen VD should error out
+		msg := &boostTypes.BuilderSubmitBlockRequest{}
+		err = backend.ragg.SubmitBlock(msg, ValidatorData{GasLimit: 40})
+		require.Error(t, err)
+	})
+
+	t.Run("should submit to relay with matching validator data", func(t *testing.T) {
+		backend := newTestRelayAggBackend(3)
+
+		backend.relays[0].gvsVd.GasLimit = 10
+		backend.relays[1].gvsVd.GasLimit = 20
+		backend.relays[2].gvsVd.GasLimit = 30
+
+		vd, err := backend.ragg.GetValidatorForSlot(11)
+		require.NoError(t, err)
+		require.Equal(t, uint64(10), vd.GasLimit)
+
+		// let the validator registrations finish
+		// TODO: notify from the test relays
+		time.Sleep(10 * time.Millisecond)
+
+		// if submitting for unseen VD should error out
+		msg := &boostTypes.BuilderSubmitBlockRequest{}
+		err = backend.ragg.SubmitBlock(msg, ValidatorData{GasLimit: 40})
+		require.Error(t, err)
+
+		// should submit to the single pirmary if its the only one matching
+		backend.relays[0].submittedMsgCh = make(chan *boostTypes.BuilderSubmitBlockRequest, 1)
+		err = backend.ragg.SubmitBlock(msg, ValidatorData{GasLimit: 10})
+		require.NoError(t, err)
+		select {
+		case rsMsg := <-backend.relays[0].submittedMsgCh:
+			require.Equal(t, msg, rsMsg)
+		case <-time.After(time.Second):
+			t.Fail()
+		}
 
-	// Check getting validator slot - should error out if no relays return
-	_, err := ragg.GetValidatorForSlot(10)
-	require.Error(t, ErrValidatorNotFound, err)
-
-	// If primary returns should not error out
-	testRelays[0].gvsErr = nil
-	_, err = ragg.GetValidatorForSlot(10)
-	require.NoError(t, err)
-
-	// If any returns should not error out
-	testRelays[0].gvsErr = errors.New("error!")
-	testRelays[2].gvsErr = nil
-	_, err = ragg.GetValidatorForSlot(10)
-	require.NoError(t, err)
-
-	// Should return the more important relay if primary fails
-	testRelays[1].gvsErr = nil
-	testRelays[1].gvsVd.GasLimit = 20
-	testRelays[2].gvsVd.GasLimit = 30
-	vd, err := ragg.GetValidatorForSlot(10)
-	require.NoError(t, err)
-	require.Equal(t, uint64(20), vd.GasLimit)
-
-	// Should return the primary if it returns
-	testRelays[0].gvsErr = nil
-	testRelays[0].gvsVd.GasLimit = 10
-	vd, err = ragg.GetValidatorForSlot(11)
-	require.NoError(t, err)
-	require.Equal(t, uint64(10), vd.GasLimit)
-
-	// let the validator registrations finish
-	// TODO: notify from the test relays
-	time.Sleep(10 * time.Millisecond)
-
-	// if submitting for unseen VD should error out
-	msg := &boostTypes.BuilderSubmitBlockRequest{}
-	err = ragg.SubmitBlock(msg, ValidatorData{GasLimit: 40})
-	require.Error(t, err)
-
-	// should submit to the single pirmary if its the only one matching
-	testRelays[0].submittedMsgCh = make(chan *boostTypes.BuilderSubmitBlockRequest, 1)
-	err = ragg.SubmitBlock(msg, ValidatorData{GasLimit: 10})
-	require.NoError(t, err)
-	select {
-	case rsMsg := <-testRelays[0].submittedMsgCh:
-		require.Equal(t, msg, rsMsg)
-	case <-time.After(time.Second):
-		t.Fail()
-	}
+		// no other relay should have been asked
+		require.Nil(t, backend.relays[1].submittedMsg)
+		require.Nil(t, backend.relays[2].submittedMsg)
+	})
 
-	// no other relay should have been asked
-	require.Nil(t, testRelays[1].submittedMsg)
-	require.Nil(t, testRelays[2].submittedMsg)
-	testRelays[0].submittedMsgCh = nil
+	t.Run("should submit to relays with matching validator data and drop registrations on next slot", func(t *testing.T) {
+		backend := newTestRelayAggBackend(3)
 
-	testRelays[0].gvsVd.GasLimit = 30
-	testRelays[1].gvsVd.GasLimit = 20
-	testRelays[2].gvsVd.GasLimit = 30
+		backend.relays[0].gvsVd.GasLimit = 10
+		backend.relays[1].gvsVd.GasLimit = 20
+		backend.relays[2].gvsVd.GasLimit = 30
 
-	// should drop registrations if asked for the next slot
-	vd, err = ragg.GetValidatorForSlot(12)
-	require.NoError(t, err)
-	require.Equal(t, uint64(30), vd.GasLimit)
+		vd, err := backend.ragg.GetValidatorForSlot(11)
+		require.NoError(t, err)
+		require.Equal(t, uint64(10), vd.GasLimit)
 
-	time.Sleep(10 * time.Millisecond)
+		// let the validator registrations finish
+		time.Sleep(10 * time.Millisecond)
 
-	// should submit to multiple matching relays
-	testRelays[0].submittedMsgCh = make(chan *boostTypes.BuilderSubmitBlockRequest, 1)
-	testRelays[2].submittedMsgCh = make(chan *boostTypes.BuilderSubmitBlockRequest, 1)
-	err = ragg.SubmitBlock(msg, ValidatorData{GasLimit: 10})
-	require.Error(t, err)
+		backend.relays[0].gvsVd.GasLimit = 30
+		backend.relays[1].gvsVd.GasLimit = 20
+		backend.relays[2].gvsVd.GasLimit = 30
 
-	err = ragg.SubmitBlock(msg, ValidatorData{GasLimit: 30})
-	require.NoError(t, err)
+		// should drop registrations if asked for the next slot
+		vd, err = backend.ragg.GetValidatorForSlot(12)
+		require.NoError(t, err)
+		require.Equal(t, uint64(30), vd.GasLimit)
 
-	select {
-	case rsMsg := <-testRelays[0].submittedMsgCh:
-		require.Equal(t, msg, rsMsg)
-	case <-time.After(time.Second):
-		t.Fail()
-	}
+		time.Sleep(10 * time.Millisecond)
 
-	select {
-	case rsMsg := <-testRelays[2].submittedMsgCh:
-		require.Equal(t, msg, rsMsg)
-	case <-time.After(time.Second):
-		t.Fail()
-	}
+		// should submit to multiple matching relays
+		backend.relays[0].submittedMsgCh = make(chan *boostTypes.BuilderSubmitBlockRequest, 1)
+		backend.relays[2].submittedMsgCh = make(chan *boostTypes.BuilderSubmitBlockRequest, 1)
+		msg := &boostTypes.BuilderSubmitBlockRequest{}
+		err = backend.ragg.SubmitBlock(msg, ValidatorData{GasLimit: 10})
+		require.Error(t, err)
+
+		err = backend.ragg.SubmitBlock(msg, ValidatorData{GasLimit: 30})
+		require.NoError(t, err)
+
+		select {
+		case rsMsg := <-backend.relays[0].submittedMsgCh:
+			require.Equal(t, msg, rsMsg)
+		case <-time.After(time.Second):
+			t.Fail()
+		}
+
+		select {
+		case rsMsg := <-backend.relays[2].submittedMsgCh:
+			require.Equal(t, msg, rsMsg)
+		case <-time.After(time.Second):
+			t.Fail()
+		}
+	})
 }