2021-01-11 23:59:17 +00:00
|
|
|
package kv
|
|
|
|
|
|
|
|
import (
|
|
|
|
"bytes"
|
|
|
|
"context"
|
|
|
|
|
2021-02-22 23:20:57 +00:00
|
|
|
types "github.com/prysmaticlabs/eth2-types"
|
2021-01-11 23:59:17 +00:00
|
|
|
"github.com/prysmaticlabs/prysm/shared/bytesutil"
|
2021-01-15 21:35:21 +00:00
|
|
|
"github.com/prysmaticlabs/prysm/shared/params"
|
2021-01-11 23:59:17 +00:00
|
|
|
"github.com/prysmaticlabs/prysm/shared/progressutil"
|
|
|
|
bolt "go.etcd.io/bbolt"
|
|
|
|
)
|
|
|
|
|
|
|
|
var migrationOptimalAttesterProtectionKey = []byte("optimal_attester_protection_0")
|
|
|
|
|
|
|
|
// Migrate attester protection to a more optimal format in the DB. Given we
|
|
|
|
// stored attesting history as large, 2Mb arrays per validator, we need to perform
|
|
|
|
// this migration differently than the rest, ensuring we perform each expensive bolt
|
|
|
|
// update in its own transaction to prevent having everything on the heap.
|
2021-01-20 14:39:07 +00:00
|
|
|
func (s *Store) migrateOptimalAttesterProtectionUp(ctx context.Context) error {
|
2021-01-11 23:59:17 +00:00
|
|
|
publicKeyBytes := make([][]byte, 0)
|
|
|
|
attestingHistoryBytes := make([][]byte, 0)
|
|
|
|
numKeys := 0
|
2021-01-20 14:39:07 +00:00
|
|
|
err := s.db.Update(func(tx *bolt.Tx) error {
|
2021-01-11 23:59:17 +00:00
|
|
|
mb := tx.Bucket(migrationsBucket)
|
|
|
|
if b := mb.Get(migrationOptimalAttesterProtectionKey); bytes.Equal(b, migrationCompleted) {
|
|
|
|
return nil // Migration already completed.
|
|
|
|
}
|
|
|
|
|
2021-01-15 21:35:21 +00:00
|
|
|
bkt := tx.Bucket(deprecatedAttestationHistoryBucket)
|
2021-01-11 23:59:17 +00:00
|
|
|
numKeys = bkt.Stats().KeyN
|
2021-01-22 23:12:22 +00:00
|
|
|
return bkt.ForEach(func(k, v []byte) error {
|
2021-01-11 23:59:17 +00:00
|
|
|
if v == nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
bucket := tx.Bucket(pubKeysBucket)
|
|
|
|
pkBucket, err := bucket.CreateBucketIfNotExists(k)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
_, err = pkBucket.CreateBucketIfNotExists(attestationSourceEpochsBucket)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
_, err = pkBucket.CreateBucketIfNotExists(attestationSigningRootsBucket)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
nk := make([]byte, len(k))
|
|
|
|
copy(nk, k)
|
|
|
|
nv := make([]byte, len(v))
|
|
|
|
copy(nv, v)
|
|
|
|
publicKeyBytes = append(publicKeyBytes, nk)
|
|
|
|
attestingHistoryBytes = append(attestingHistoryBytes, nv)
|
|
|
|
return nil
|
2021-01-22 23:12:22 +00:00
|
|
|
})
|
2021-01-11 23:59:17 +00:00
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
bar := progressutil.InitializeProgressBar(numKeys, "Migrating attesting history to more efficient format")
|
|
|
|
for i, publicKey := range publicKeyBytes {
|
2021-01-22 23:12:22 +00:00
|
|
|
attestingHistory := deprecatedEncodedAttestingHistory(attestingHistoryBytes[i])
|
2021-01-20 14:39:07 +00:00
|
|
|
err = s.db.Update(func(tx *bolt.Tx) error {
|
2021-01-12 01:10:11 +00:00
|
|
|
if attestingHistory == nil {
|
2021-01-11 23:59:17 +00:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
bucket := tx.Bucket(pubKeysBucket)
|
|
|
|
pkBucket := bucket.Bucket(publicKey)
|
|
|
|
sourceEpochsBucket := pkBucket.Bucket(attestationSourceEpochsBucket)
|
|
|
|
|
|
|
|
signingRootsBucket := pkBucket.Bucket(attestationSigningRootsBucket)
|
|
|
|
|
|
|
|
// Extract every single source, target, signing root
|
|
|
|
// from the attesting history then insert them into the
|
|
|
|
// respective buckets under the new db schema.
|
2021-01-12 15:32:13 +00:00
|
|
|
latestEpochWritten, err := attestingHistory.getLatestEpochWritten(ctx)
|
2021-01-11 23:59:17 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
// For every epoch since genesis up to the highest epoch written, we then
|
|
|
|
// extract historical data and insert it into the new schema.
|
2021-02-09 10:05:22 +00:00
|
|
|
for targetEpoch := types.Epoch(0); targetEpoch <= latestEpochWritten; targetEpoch++ {
|
2021-01-12 15:32:13 +00:00
|
|
|
historicalAtt, err := attestingHistory.getTargetData(ctx, targetEpoch)
|
2021-01-11 23:59:17 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2021-01-12 15:32:13 +00:00
|
|
|
if historicalAtt.isEmpty() {
|
2021-01-11 23:59:17 +00:00
|
|
|
continue
|
|
|
|
}
|
2021-02-09 10:05:22 +00:00
|
|
|
targetEpochBytes := bytesutil.EpochToBytesBigEndian(targetEpoch)
|
|
|
|
sourceEpochBytes := bytesutil.EpochToBytesBigEndian(historicalAtt.Source)
|
2021-01-11 23:59:17 +00:00
|
|
|
if err := sourceEpochsBucket.Put(sourceEpochBytes, targetEpochBytes); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := signingRootsBucket.Put(targetEpochBytes, historicalAtt.SigningRoot); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return bar.Add(1)
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-01-20 14:39:07 +00:00
|
|
|
return s.db.Update(func(tx *bolt.Tx) error {
|
2021-01-11 23:59:17 +00:00
|
|
|
mb := tx.Bucket(migrationsBucket)
|
2021-01-22 23:12:22 +00:00
|
|
|
return mb.Put(migrationOptimalAttesterProtectionKey, migrationCompleted)
|
2021-01-11 23:59:17 +00:00
|
|
|
})
|
|
|
|
}
|
2021-01-15 21:35:21 +00:00
|
|
|
|
|
|
|
// Migrate attester protection from the more optimal format to the old format in the DB.
|
2021-01-20 14:39:07 +00:00
|
|
|
func (s *Store) migrateOptimalAttesterProtectionDown(ctx context.Context) error {
|
2021-01-15 21:35:21 +00:00
|
|
|
// First we extract the public keys we are migrating down for.
|
|
|
|
pubKeys := make([][48]byte, 0)
|
2021-01-20 14:39:07 +00:00
|
|
|
err := s.view(func(tx *bolt.Tx) error {
|
2021-01-15 21:35:21 +00:00
|
|
|
mb := tx.Bucket(migrationsBucket)
|
|
|
|
if b := mb.Get(migrationOptimalAttesterProtectionKey); b == nil {
|
|
|
|
// Migration has not occurred, meaning data is already in old format
|
|
|
|
// so no need to perform a down migration.
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
bkt := tx.Bucket(pubKeysBucket)
|
|
|
|
if bkt == nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
return bkt.ForEach(func(pubKey, v []byte) error {
|
|
|
|
if pubKey == nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
pkBucket := bkt.Bucket(pubKey)
|
|
|
|
if pkBucket == nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
pubKeys = append(pubKeys, bytesutil.ToBytes48(pubKey))
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Next up, we extract the data for attested epochs and signing roots
|
|
|
|
// from the optimized db schema into maps we can use later.
|
2021-02-09 10:05:22 +00:00
|
|
|
signingRootsByTarget := make(map[types.Epoch][]byte)
|
|
|
|
targetEpochsBySource := make(map[types.Epoch][]types.Epoch)
|
2021-01-20 14:39:07 +00:00
|
|
|
err = s.view(func(tx *bolt.Tx) error {
|
2021-01-15 21:35:21 +00:00
|
|
|
bkt := tx.Bucket(pubKeysBucket)
|
|
|
|
if bkt == nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
for _, pubKey := range pubKeys {
|
|
|
|
pubKeyBkt := bkt.Bucket(pubKey[:])
|
|
|
|
if pubKeyBkt == nil {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
sourceEpochsBucket := pubKeyBkt.Bucket(attestationSourceEpochsBucket)
|
|
|
|
signingRootsBucket := pubKeyBkt.Bucket(attestationSigningRootsBucket)
|
|
|
|
// Extract signing roots.
|
|
|
|
if err := signingRootsBucket.ForEach(func(targetBytes, signingRoot []byte) error {
|
|
|
|
var sr [32]byte
|
|
|
|
copy(sr[:], signingRoot)
|
2021-02-09 10:05:22 +00:00
|
|
|
signingRootsByTarget[bytesutil.BytesToEpochBigEndian(targetBytes)] = sr[:]
|
2021-01-15 21:35:21 +00:00
|
|
|
return nil
|
|
|
|
}); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
// Next up, extract the target epochs by source.
|
|
|
|
if err := sourceEpochsBucket.ForEach(func(sourceBytes, targetEpochsBytes []byte) error {
|
2021-02-09 10:05:22 +00:00
|
|
|
targetEpochs := make([]types.Epoch, 0)
|
2021-01-15 21:35:21 +00:00
|
|
|
for i := 0; i < len(targetEpochsBytes); i += 8 {
|
2021-02-09 10:05:22 +00:00
|
|
|
targetEpochs = append(targetEpochs, bytesutil.BytesToEpochBigEndian(targetEpochsBytes[i:i+8]))
|
2021-01-15 21:35:21 +00:00
|
|
|
}
|
2021-02-09 10:05:22 +00:00
|
|
|
targetEpochsBySource[bytesutil.BytesToEpochBigEndian(sourceBytes)] = targetEpochs
|
2021-01-15 21:35:21 +00:00
|
|
|
return nil
|
|
|
|
}); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Then, we use the data we extracted to recreate the old
|
|
|
|
// attesting history format and for each public key, we save it
|
|
|
|
// to the appropriate bucket.
|
2021-01-20 14:39:07 +00:00
|
|
|
err = s.update(func(tx *bolt.Tx) error {
|
2021-01-15 21:35:21 +00:00
|
|
|
bkt := tx.Bucket(pubKeysBucket)
|
|
|
|
if bkt == nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
bar := progressutil.InitializeProgressBar(len(pubKeys), "Migrating attesting history to old format")
|
|
|
|
for _, pubKey := range pubKeys {
|
|
|
|
// Now we write the attesting history using the data we extracted
|
|
|
|
// from the buckets accordingly.
|
|
|
|
history := newDeprecatedAttestingHistory(0)
|
2021-02-09 10:05:22 +00:00
|
|
|
var maxTargetWritten types.Epoch
|
2021-01-15 21:35:21 +00:00
|
|
|
for source, targetEpochs := range targetEpochsBySource {
|
|
|
|
for _, target := range targetEpochs {
|
|
|
|
signingRoot := params.BeaconConfig().ZeroHash[:]
|
|
|
|
if sr, ok := signingRootsByTarget[target]; ok {
|
|
|
|
signingRoot = sr
|
|
|
|
}
|
|
|
|
newHist, err := history.setTargetData(ctx, target, &deprecatedHistoryData{
|
|
|
|
Source: source,
|
|
|
|
SigningRoot: signingRoot,
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
history = newHist
|
|
|
|
if target > maxTargetWritten {
|
|
|
|
maxTargetWritten = target
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
newHist, err := history.setLatestEpochWritten(ctx, maxTargetWritten)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
history = newHist
|
|
|
|
deprecatedBkt, err := tx.CreateBucketIfNotExists(deprecatedAttestationHistoryBucket)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := deprecatedBkt.Put(pubKey[:], history); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := bar.Add(1); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Finally, we clear the migration key.
|
2021-01-20 14:39:07 +00:00
|
|
|
return s.update(func(tx *bolt.Tx) error {
|
2021-01-15 21:35:21 +00:00
|
|
|
migrationsBkt := tx.Bucket(migrationsBucket)
|
|
|
|
return migrationsBkt.Delete(migrationOptimalAttesterProtectionKey)
|
|
|
|
})
|
|
|
|
}
|