Sync progress of optimistic block validation across builder instances via redis

Author: austonst

datastore/redis.go

@@ -24,6 +24,7 @@ var (
 	redisPrefix = "boost-relay"
 
 	expiryBidCache = 45 * time.Second
+	expiryLock     = 24 * time.Second
 
 	RedisConfigFieldPubkey         = "pubkey"
 	RedisStatsFieldLatestSlot      = "latest-slot"
@@ -91,6 +92,7 @@ type RedisCache struct {
 	prefixTopBidValue                 string
 	prefixFloorBid                    string
 	prefixFloorBidValue               string
+	prefixProcessingSlot              string
 
 	// keys
 	keyValidatorRegistrationTimestamp string
@@ -101,6 +103,8 @@ type RedisCache struct {
 	keyBlockBuilderStatus string
 	keyLastSlotDelivered  string
 	keyLastHashDelivered  string
+
+	currentSlot uint64
 }
 
 func NewRedisCache(prefix, redisURI, readonlyURI string) (*RedisCache, error) {
@@ -132,6 +136,7 @@ func NewRedisCache(prefix, redisURI, readonlyURI string) (*RedisCache, error) {
 		prefixTopBidValue:                 fmt.Sprintf("%s/%s:top-bid-value", redisPrefix, prefix),                  // prefix:slot_parentHash_proposerPubkey
 		prefixFloorBid:                    fmt.Sprintf("%s/%s:bid-floor", redisPrefix, prefix),                      // prefix:slot_parentHash_proposerPubkey
 		prefixFloorBidValue:               fmt.Sprintf("%s/%s:bid-floor-value", redisPrefix, prefix),                // prefix:slot_parentHash_proposerPubkey
+		prefixProcessingSlot:              fmt.Sprintf("%s/%s:processing-slot", redisPrefix, prefix),                // prefix:slot
 
 		keyValidatorRegistrationTimestamp: fmt.Sprintf("%s/%s:validator-registration-timestamp", redisPrefix, prefix),
 		keyRelayConfig:                    fmt.Sprintf("%s/%s:relay-config", redisPrefix, prefix),
@@ -190,6 +195,11 @@ func (r *RedisCache) keyFloorBidValue(slot uint64, parentHash, proposerPubkey st
 	return fmt.Sprintf("%s:%d_%s_%s", r.prefixFloorBidValue, slot, parentHash, proposerPubkey)
 }
 
+// keyProcessingSlot returns the key for the counter of builder processes working on a given slot
+func (r *RedisCache) keyProcessingSlot(slot uint64) string {
+	return fmt.Sprintf("%s:%d", r.prefixProcessingSlot, slot)
+}
+
 func (r *RedisCache) GetObj(key string, obj any) (err error) {
 	value, err := r.client.Get(context.Background(), key).Result()
 	if err != nil {
@@ -800,6 +810,51 @@ func (r *RedisCache) SetFloorBidValue(slot uint64, parentHash, proposerPubkey, v
 	return err
 }
 
+// BeginProcessingSlot signals that a builder process is handling blocks for a given slot
+func (r *RedisCache) BeginProcessingSlot(ctx context.Context, slot uint64) (err error) {
+	// Should never process more than one slot at a time
+	if r.currentSlot != 0 {
+		return fmt.Errorf("already processing slot %d", r.currentSlot)
+	}
+
+	keyProcessingSlot := r.keyProcessingSlot(slot)
+	err = r.client.Incr(ctx, keyProcessingSlot).Err()
+	if err != nil {
+		return err
+	}
+	r.currentSlot = slot
+	err = r.client.Expire(ctx, keyProcessingSlot, expiryLock).Err()
+	return err
+}
+
+// EndProcessingSlot signals that a builder process is done handling blocks for the current slot
+func (r *RedisCache) EndProcessingSlot(ctx context.Context) (err error) {
+	// Do not decrement if called multiple times
+	if r.currentSlot == 0 {
+		return nil
+	}
+
+	keyProcessingSlot := r.keyProcessingSlot(r.currentSlot)
+	err = r.client.Decr(ctx, keyProcessingSlot).Err()
+	r.currentSlot = 0
+	return err
+}
+
+// WaitForSlotComplete waits for a slot to be completed by all builder processes
+func (r *RedisCache) WaitForSlotComplete(ctx context.Context, slot uint64) (err error) {
+	keyProcessingSlot := r.keyProcessingSlot(slot)
+	for {
+		processing, err := r.client.Get(ctx, keyProcessingSlot).Uint64()
+		if err != nil {
+			return err
+		}
+		if processing == 0 {
+			return nil
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+}
+
 func (r *RedisCache) NewPipeline() redis.Pipeliner { //nolint:ireturn,nolintlint
 	return r.client.Pipeline()
 }

services/api/service.go

@@ -516,6 +516,7 @@ func (api *RelayAPI) IsReady() bool {
 // - Stop returning bids
 // - Set ready /readyz to negative status
 // - Wait a bit to allow removal of service from load balancer and draining of requests
+// - If in the middle of proccessing optimistic blocks, wait for those to finish and release redis lock
 func (api *RelayAPI) StopServer() (err error) {
 	// avoid running this twice. setting srvShutdown to true makes /readyz switch to negative status
 	if wasStopping := api.srvShutdown.Swap(true); wasStopping {
@@ -538,6 +539,10 @@ func (api *RelayAPI) StopServer() (err error) {
 	// wait for any active getPayload call to finish
 	api.getPayloadCallsInFlight.Wait()
 
+	// wait for optimistic blocks
+	api.optimisticBlocksWG.Wait()
+	api.redis.EndProcessingSlot(context.Background())
+
 	// shutdown
 	return api.srv.Shutdown(context.Background())
 }
@@ -826,10 +831,21 @@ func (api *RelayAPI) updateProposerDuties(headSlot uint64) {
 }
 
 func (api *RelayAPI) prepareBuildersForSlot(headSlot uint64) {
-	// Wait until there are no optimistic blocks being processed. Then we can
-	// safely update the slot.
+	// First wait for this process to finish processing optimistic blocks
 	api.optimisticBlocksWG.Wait()
+
+	// Now we release our lock and wait for all other builder processes to wrap up
+	api.redis.EndProcessingSlot(context.Background())
+	api.redis.WaitForSlotComplete(context.Background(), headSlot)
+
+	// Prevent race with StopServer, make sure we don't lock up redis if the server is shutting down
+	if api.srvShutdown.Load() {
+		return
+	}
+
+	// Update the optimistic slot and signal processing of the next slot
 	api.optimisticSlot.Store(headSlot + 1)
+	api.redis.BeginProcessingSlot(context.Background(), headSlot + 1)
 
 	builders, err := api.db.GetBlockBuilders()
 	if err != nil {