Publish block before returning to proposer (flashbots#330)

* publish block first

* Add concurrent release of the execution payload

* Fix capture

* Update logging

* Log the whole URI

* add 1s delay on returning getPaylaod

* fix naming

---------

Co-authored-by: Mateusz Morusiewicz <[email protected]>

beaconclient/multi_beacon_client.go

@@ -17,6 +17,7 @@ var (
 	ErrBeaconNodeSyncing        = errors.New("beacon node is syncing or unavailable")
 	ErrBeaconNodesUnavailable   = errors.New("all beacon nodes responded with error")
 	ErrWithdrawalsBeforeCapella = errors.New("withdrawals are not supported before capella")
+	ErrBeaconBlock202           = errors.New("beacon block failed validation but was still broadcast (202)")
 )
 
 // IMultiBeaconClient is the interface for the MultiBeaconClient, which can manage several beacon client instances under the hood
@@ -201,6 +202,12 @@ func (c *MultiBeaconClient) beaconInstancesByLastResponse() []IBeaconInstance {
 	return instances
 }
 
+type publishResp struct {
+	index int
+	code  int
+	err   error
+}
+
 // PublishBlock publishes the signed beacon block via https://ethereum.github.io/beacon-APIs/#/ValidatorRequiredApi/publishBlock
 func (c *MultiBeaconClient) PublishBlock(block *common.SignedBeaconBlock) (code int, err error) {
 	log := c.log.WithFields(logrus.Fields{
@@ -209,21 +216,46 @@ func (c *MultiBeaconClient) PublishBlock(block *common.SignedBeaconBlock) (code
 	})
 
 	clients := c.beaconInstancesByLastResponse()
-	for _, client := range clients {
+
+	// The chan will be cleaner up automatically once the function exists even if it was still being written to
+	resChans := make(chan publishResp, len(clients))
+
+	for i, client := range clients {
 		log := log.WithField("uri", client.GetURI())
 		log.Debug("publishing block")
+		go func(index int, client IBeaconInstance) {
+			code, err := client.PublishBlock(block)
+			resChans <- publishResp{
+				index: index,
+				code:  code,
+				err:   err,
+			}
+		}(i, client)
+	}
 
-		if code, err = client.PublishBlock(block); err != nil {
-			log.WithField("statusCode", code).WithError(err).Warn("failed to publish block")
+	var lastErrPublishResp publishResp
+	for i := 0; i < len(clients); i++ {
+		res := <-resChans
+		if res.err != nil {
+			log.WithField("beacon", clients[res.index].GetURI()).WithField("statusCode", res.code).WithError(res.err).Error("failed to publish block")
+			lastErrPublishResp = res
+			continue
+		} else if res.code == 202 {
+			// Should the block fail full validation, a separate success response code (202) is used to indicate that the block was successfully broadcast but failed integration.
+			// https://ethereum.github.io/beacon-APIs/?urls.primaryName=dev#/Beacon/publishBlock
+			log.WithField("beacon", clients[res.index].GetURI()).WithField("statusCode", res.code).WithError(res.err).Error("block failed validation but was still broadcast")
+			lastErrPublishResp = res
 			continue
 		}
 
-		log.WithField("statusCode", code).Info("published block")
-		return code, nil
+		c.bestBeaconIndex.Store(int64(res.index))
+
+		log.WithField("beacon", clients[res.index].GetURI()).WithField("statusCode", res.code).Info("published block")
+		return res.code, nil
 	}
 
-	log.WithField("statusCode", code).WithError(err).Error("failed to publish block on any CL node")
-	return code, err
+	log.Error("failed to publish block on any CL node")
+	return lastErrPublishResp.code, lastErrPublishResp.err
 }
 
 // GetGenesis returns the genesis info - https://ethereum.github.io/beacon-APIs/#/Beacon/getGenesis

services/api/service.go

@@ -79,6 +79,7 @@ var (
 	numActiveValidatorProcessors = cli.GetEnvInt("NUM_ACTIVE_VALIDATOR_PROCESSORS", 10)
 	numValidatorRegProcessors    = cli.GetEnvInt("NUM_VALIDATOR_REG_PROCESSORS", 10)
 	timeoutGetPayloadRetryMs     = cli.GetEnvInt("GETPAYLOAD_RETRY_TIMEOUT_MS", 100)
+	getPayloadResponseDelayMs    = cli.GetEnvInt("GETPAYLOAD_DELAY_MS", 1000)
 
 	apiReadTimeoutMs       = cli.GetEnvInt("API_TIMEOUT_READ_MS", 1500)
 	apiReadHeaderTimeoutMs = cli.GetEnvInt("API_TIMEOUT_READHEADER_MS", 600)
@@ -255,11 +256,6 @@ func NewRelayAPI(opts RelayAPIOpts) (api *RelayAPI, err error) {
 		api.ffForceGetHeader204 = true
 	}
 
-	if os.Getenv("DISABLE_BLOCK_PUBLISHING") == "1" {
-		api.log.Warn("env: DISABLE_BLOCK_PUBLISHING - disabling publishing blocks on getPayload")
-		api.ffDisableBlockPublishing = true
-	}
-
 	if os.Getenv("DISABLE_LOWPRIO_BUILDERS") == "1" {
 		api.log.Warn("env: DISABLE_LOWPRIO_BUILDERS - allowing only high-level builders")
 		api.ffDisableLowPrioBuilders = true
@@ -1031,6 +1027,18 @@ func (api *RelayAPI) handleGetPayload(w http.ResponseWriter, req *http.Request)
 		}
 	}
 
+	// Publish the signed beacon block via beacon-node
+	signedBeaconBlock := SignedBlindedBeaconBlockToBeaconBlock(payload, getPayloadResp)
+	code, err := api.beaconClient.PublishBlock(signedBeaconBlock) // errors are logged inside
+	if err != nil {
+		log.WithError(err).WithField("code", code).Error("failed to publish block")
+		api.RespondError(w, http.StatusBadRequest, "failed to publish block")
+		return
+	}
+
+	// give the beacon network some time to propagate the block
+	time.Sleep(time.Duration(getPayloadResponseDelayMs) * time.Millisecond)
+
 	api.RespondOK(w, getPayloadResp)
 	log = log.WithFields(logrus.Fields{
 		"numTx":       getPayloadResp.NumTx(),
@@ -1117,16 +1125,6 @@ func (api *RelayAPI) handleGetPayload(w http.ResponseWriter, req *http.Request)
 			}).WithError(err).Error("unable to update builder demotion with refund justification")
 		}
 	}()
-
-	// Publish the signed beacon block via beacon-node
-	go func() {
-		if api.ffDisableBlockPublishing {
-			log.Info("publishing the block is disabled")
-			return
-		}
-		signedBeaconBlock := SignedBlindedBeaconBlockToBeaconBlock(payload, getPayloadResp)
-		_, _ = api.beaconClient.PublishBlock(signedBeaconBlock) // errors are logged inside
-	}()
 }
 
 // --------------------