Add storage and resend for validators preferences #149
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -8,6 +8,7 @@ import ( | |
| "net/http" | ||
| "strconv" | ||
| "sync" | ||
| "sync/atomic" | ||
| "time" | ||
|
|
||
| "github.com/flashbots/go-boost-utils/types" | ||
|
|
@@ -56,10 +57,17 @@ type BoostService struct { | |
| serverTimeouts HTTPServerTimeouts | ||
|
|
||
| httpClient http.Client | ||
|
|
||
| // Used to stop registerValidatorAtInterval | ||
| done chan bool | ||
| // Used by registerValidator to share new incoming registration request with the goroutine holding the ticker | ||
|
There was a problem hiding this comment. It'd be great to avoid fields shared across requests as they usually introduce a race |
||
| newRegistrationsRequests chan []types.SignedValidatorRegistration | ||
| // Used by registerValidatorAtInterval to share the number of successful requests with the registerValidator handler | ||
| numSuccessRequestsToRelay chan uint64 | ||
| } | ||
|
|
||
| // NewBoostService created a new BoostService | ||
| func NewBoostService(listenAddr string, relays []RelayEntry, log *logrus.Entry, genesisForkVersionHex string, relayRequestTimeout, validatorPreferencesResendInterval time.Duration) (*BoostService, error) { | ||
| if len(relays) == 0 { | ||
| return nil, errors.New("no relays") | ||
| } | ||
|
|
@@ -77,6 +85,10 @@ func NewBoostService(listenAddr string, relays []RelayEntry, log *logrus.Entry, | |
| builderSigningDomain: builderSigningDomain, | ||
| serverTimeouts: NewDefaultHTTPServerTimeouts(), | ||
| httpClient: http.Client{Timeout: relayRequestTimeout}, | ||
|
|
||
| done: make(chan bool), | ||
| newRegistrationsRequests: make(chan []types.SignedValidatorRegistration), | ||
| numSuccessRequestsToRelay: make(chan uint64), | ||
| }, nil | ||
| } | ||
|
|
||
|
|
@@ -94,8 +106,8 @@ func (m *BoostService) getRouter() http.Handler { | |
| return loggedRouter | ||
| } | ||
|
|
||
| // StartServer starts the HTTP server for this boost service instance | ||
| func (m *BoostService) StartServer() error { | ||
| if m.srv != nil { | ||
| return errServerAlreadyRunning | ||
| } | ||
|
|
@@ -110,10 +122,15 @@ func (m *BoostService) StartHTTPServer() error { | |
| IdleTimeout: m.serverTimeouts.Idle, | ||
| } | ||
|
|
||
| // Start separate process to send validator preferences at regular interval. | ||
| go m.registerValidatorAtInterval(time.Second*384, m.done) | ||
|
There was a problem hiding this comment. I'm guessing this should be registerValidatorIntervalSec There was a problem hiding this comment. Missed that, thanks ! Fixed in e1d6572. |
||
| defer m.shutdown() | ||
|
|
||
| err := m.srv.ListenAndServe() | ||
| if err == http.ErrServerClosed { | ||
| return nil | ||
| } | ||
|
|
||
| return err | ||
| } | ||
|
|
||
|
|
@@ -129,6 +146,62 @@ func (m *BoostService) handleStatus(w http.ResponseWriter, req *http.Request) { | |
| fmt.Fprintf(w, `{}`) | ||
| } | ||
|
|
||
| // sendValidatorPreferences is used to send the validators preferences to the registered relays | ||
| func (m *BoostService) sendValidatorPreferences(log *logrus.Entry, payload []types.SignedValidatorRegistration) uint64 { | ||
| // We need a wait group to manage each routine used to perform the requests. | ||
| var wg sync.WaitGroup | ||
|
|
||
| // Use an atomic counter to count successful requests. | ||
| numSuccessRequestsToRelay := uint64(0) | ||
|
|
||
| // Send the validators preferences to each registered relay. | ||
| for _, relay := range m.relays { | ||
| wg.Add(1) | ||
|
|
||
| go func(relayAddr string) { | ||
| defer wg.Done() | ||
|
|
||
| url := relayAddr + pathRegisterValidator | ||
| log := log.WithField("url", url) | ||
|
|
||
| err := SendHTTPRequest(context.Background(), m.httpClient, http.MethodPost, url, payload, nil) | ||
| if err != nil { | ||
| log.WithError(err).Warn("error in registerValidator to relay") | ||
| return | ||
| } | ||
|
|
||
| atomic.AddUint64(&numSuccessRequestsToRelay, 1) | ||
| }(relay.Address) | ||
| } | ||
|
|
||
| wg.Wait() | ||
|
|
||
| return numSuccessRequestsToRelay | ||
| } | ||
|
|
||
| func (m *BoostService) registerValidatorAtInterval(interval time.Duration, done chan bool) { | ||
| var payload []types.SignedValidatorRegistration | ||
| log := m.log.WithField("method", "registerValidatorAtInterval") | ||
|
|
||
| ticker := time.NewTicker(interval) | ||
| defer ticker.Stop() | ||
|
|
||
| for { | ||
| select { | ||
| case <-done: | ||
| // mev-boost has probably stopped | ||
| return | ||
| case payload = <-m.newRegistrationsRequests: | ||
|
There was a problem hiding this comment. this will overwrite the previous payload, but it's conceivable that there's multiple validators using one BN that would then overwrite each others registrations. perhaps better to make this additive? There was a problem hiding this comment. I'm actually not sure about it, the expectation is to have a single BN connecting to mev-boost, and forcing this to be additive would mean you cannot ever remove a registration. There was a problem hiding this comment. I've implemented the merging process between the old (local payload list) and the new one (received from the channel) but not pushed it, so you'll just have to tell me ! There was a problem hiding this comment. @Ruteri expectation is a single BN, but this single BN could have many validator clients connected! We can not ignore this case. One approach could be to add a time to each registration and remove old ones. |
||
| // registerValidator has received new registrations and forwards them to here | ||
| m.numSuccessRequestsToRelay <- m.sendValidatorPreferences(log, payload) | ||
| // Reset the timer to avoid overload | ||
| ticker.Reset(interval) | ||
| case <-ticker.C: | ||
|
There was a problem hiding this comment. One issue that arises is that we can send twice in a short interval (if a registration request comes near the timer tick)
This has some nice benefits:
The only downside is that you'll need to return the registration status back over a channel (or a callback), but it should be okay. We can submit a PR to this PR if you prefer not implementing it There was a problem hiding this comment. Totally agree with this comment ! Thanks a lot for raising the issue and proposing an alternative 👍 |
||
| m.sendValidatorPreferences(log, payload) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // RegisterValidatorV1 - returns 200 if at least one relay returns 200 | ||
| func (m *BoostService) handleRegisterValidator(w http.ResponseWriter, req *http.Request) { | ||
| log := m.log.WithField("method", "registerValidator") | ||
|
|
@@ -164,32 +237,10 @@ func (m *BoostService) handleRegisterValidator(w http.ResponseWriter, req *http. | |
| } | ||
| } | ||
|
|
||
| // Send the payload to the goroutine responsible for handling the resend at interval | ||
| m.newRegistrationsRequests <- payload | ||
| // Block until we get the number of successful requests back from this goroutine | ||
| numSuccessRequestsToRelay := <-m.numSuccessRequestsToRelay | ||
|
There was a problem hiding this comment. Once context (timeout) handling is implemented this will become a race, the result channel (or callback) should be specific to the request. There was a problem hiding this comment. I really like the idea of pushing the response channel, I've added this in 2d26f24 |
||
|
|
||
| if numSuccessRequestsToRelay > 0 { | ||
| w.Header().Set("Content-Type", "application/json") | ||
|
|
@@ -384,3 +435,7 @@ func (m *BoostService) handleGetPayload(w http.ResponseWriter, req *http.Request | |
| return | ||
| } | ||
| } | ||
|
|
||
| func (m *BoostService) shutdown() { | ||
| m.done <- true | ||
|
There was a problem hiding this comment. using a context would be better because if there's multiple things to shut down, this channel would only work for one of them. could be okay for now, and we can also do this in another PR. There was a problem hiding this comment. Same as above, fixed in a2bb054 |
||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Another way to implement shutdowns will be to have a context with cancel, and storing the cancel function.
Since we are not using contexts yet this is good too.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I agree, that's what they've been made after all.
Fixed in a2bb054