batch upload known validators to Redis

datastore/redis.go

@@ -216,8 +216,17 @@ func (r *RedisCache) GetKnownValidators() (map[uint64]boostTypes.PubkeyHex, erro
 	return validators, nil
 }
 
+func (r *RedisCache) SetMultiKnownValidator(indexPkMap map[uint64]boostTypes.PubkeyHex) error {
+	values := []string{}
+	for proposerIndex, publickeyHex := range indexPkMap {
+		values = append(values, strconv.FormatUint(proposerIndex, 10), PubkeyHexToLowerStr(publickeyHex))
+	}
+
+	return r.client.HMSet(context.Background(), r.keyKnownValidators, values).Err()
+}
+
 func (r *RedisCache) SetKnownValidator(pubkeyHex boostTypes.PubkeyHex, proposerIndex uint64) error {
-	return r.client.HSet(context.Background(), r.keyKnownValidators, proposerIndex, PubkeyHexToLowerStr(pubkeyHex)).Err()
+	return r.SetMultiKnownValidator(map[uint64]boostTypes.PubkeyHex{proposerIndex: pubkeyHex})
 }
 
 func (r *RedisCache) GetValidatorRegistrationTimestamp(proposerPubkey boostTypes.PubkeyHex) (uint64, error) {

datastore/redis_test.go

@@ -100,6 +100,24 @@ func TestRedisKnownValidators(t *testing.T) {
 		require.Contains(t, knownVals, index2)
 		require.Equal(t, key2, knownVals[index2])
 	})
+
+	t.Run("Can save multi and get known validators", func(t *testing.T) {
+		key1 := types.NewPubkeyHex("0x1a1d7b8dd64e0aafe7ea7b6c95065c9364cf99d38470c12ee807d55f7de1529ad29ce2c422e0b65e3d5a05c02caca249")
+		index1 := uint64(1)
+		key2 := types.NewPubkeyHex("0x2a1d7b8dd64e0aafe7ea7b6c95065c9364cf99d38470c12ee807d55f7de1529ad29ce2c422e0b65e3d5a05c02caca249")
+		index2 := uint64(2)
+
+		indexPkMap := map[uint64]types.PubkeyHex{index1: key1, index2: key2}
+		require.NoError(t, cache.SetMultiKnownValidator(indexPkMap))
+
+		knownVals, err := cache.GetKnownValidators()
+		require.NoError(t, err)
+		require.Equal(t, 2, len(knownVals))
+		require.Contains(t, knownVals, index1)
+		require.Equal(t, key1, knownVals[index1])
+		require.Contains(t, knownVals, index2)
+		require.Equal(t, key2, knownVals[index2])
+	})
 }
 
 func TestRedisValidatorRegistrations(t *testing.T) {

services/housekeeper/housekeeper.go

@@ -133,6 +133,17 @@ func (hk *Housekeeper) processNewSlot(headSlot uint64) {
 	}).Infof("updated headSlot to %d", headSlot)
 }
 
+func (hk *Housekeeper) saveKnownValidators(indexPkMap map[uint64]types.PubkeyHex) {
+	err := hk.redis.SetMultiKnownValidator(indexPkMap)
+	if err != nil {
+		hk.log.WithError(err).Error("failed to set known validators in Redis")
+	} else {
+		for proposerIndex, publickeyHex := range indexPkMap {
+			hk.proposersAlreadySaved[proposerIndex] = publickeyHex.String()
+		}
+	}
+}
+
 // updateKnownValidators queries the full list of known validators from the beacon node
 // and stores it in redis. For the CL client this is an expensive operation and takes a bunch
 // of resources. This is why we schedule the requests for slot 4 and 20 of every epoch,
@@ -212,27 +223,35 @@ func (hk *Housekeeper) updateKnownValidators() {
 
 	i := 0
 	newValidators := 0
+	bufferSize := 10000
+	indexPkMap := make(map[uint64]types.PubkeyHex)
 	for _, validator := range validators {
 		i++
-		if printCounter && i%10000 == 0 {
-			log.Debugf("writing to redis: %d / %d", i, numValidators)
-		}
 
 		// avoid resaving if index->pubkey mapping is the same
 		prevPubkeyForIndex := hk.proposersAlreadySaved[validator.Index]
 		if prevPubkeyForIndex == validator.Validator.Pubkey {
 			continue
 		}
 
-		err := hk.redis.SetKnownValidator(types.PubkeyHex(validator.Validator.Pubkey), validator.Index)
-		if err != nil {
-			log.WithError(err).WithField("pubkey", validator.Validator.Pubkey).Error("failed to set known validator in Redis")
-		} else {
-			hk.proposersAlreadySaved[validator.Index] = validator.Validator.Pubkey
-			newValidators++
+		indexPkMap[validator.Index] = types.PubkeyHex(validator.Validator.Pubkey)
+
+		if i%bufferSize == 0 {
+			hk.saveKnownValidators(indexPkMap)
+			indexPkMap = make(map[uint64]types.PubkeyHex)
+			newValidators += bufferSize
+			if printCounter {
+				hk.log.Debugf("writing to redis: %d / %d", i, numValidators)
+			}
 		}
 	}
 
+	hk.saveKnownValidators(indexPkMap)
+	newValidators += len(indexPkMap)
+	if printCounter {
+		hk.log.Debugf("writing to redis: %d / %d", i, numValidators)
+	}
+
 	log.WithFields(logrus.Fields{
 		"durationRedisWrite": time.Since(timeStartWriting).Seconds(),
 		"newValidators":      newValidators,