spanner.go

package kv

import (
	"context"

	"github.com/pkg/errors"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"github.com/prysmaticlabs/prysm/shared/bytesutil"
	"github.com/prysmaticlabs/prysm/shared/params"
	"github.com/prysmaticlabs/prysm/slasher/detection/attestations/types"
	log "github.com/sirupsen/logrus"
	bolt "go.etcd.io/bbolt"
	"go.opencensus.io/trace"
)

// Tracks the highest and lowest observed epochs from the validator span maps
// used for attester slashing detection. This value is purely used
// as a cache key and only needs to be maintained in memory.
var highestObservedEpoch uint64
var lowestObservedEpoch = params.BeaconConfig().FarFutureEpoch

var (
	slasherLowestObservedEpoch = promauto.NewGauge(prometheus.GaugeOpts{
		Name: "slasher_lowest_observed_epoch",
		Help: "The lowest epoch number seen by slasher",
	})
	slasherHighestObservedEpoch = promauto.NewGauge(prometheus.GaugeOpts{
		Name: "slasher_highest_observed_epoch",
		Help: "The highest epoch number seen by slasher",
	})
	epochSpansCacheEvictions = promauto.NewCounter(prometheus.CounterOpts{
		Name: "epoch_spans_cache_evictions_total",
		Help: "The number of cache evictions seen by slasher",
	})
)

// This function defines a function which triggers upon a span map being
// evicted from the cache. It allows us to persist the span map by the epoch value
// to the database itself in the validatorsMinMaxSpanBucket.
func persistSpanMapsOnEviction(db *Store) func(key interface{}, value interface{}) {
	// We use a closure here so we can access the database itself
	// on the eviction of a span map from the cache. The function has the signature
	// required by the ristretto cache OnEvict method.
	// See https://godoc.org/github.com/dgraph-io/ristretto#Config.
	return func(key interface{}, value interface{}) {
		log.Tracef("Evicting span map for epoch: %d", key)
		err := db.update(func(tx *bolt.Tx) error {
			epoch, keyOK := key.(uint64)
			spanMap, valueOK := value.(map[uint64]types.Span)
			if !keyOK || !valueOK {
				return errors.New("could not cast key and value into needed types")
			}

			bucket := tx.Bucket(validatorsMinMaxSpanBucket)
			epochBucket, err := bucket.CreateBucketIfNotExists(bytesutil.Bytes8(epoch))
			if err != nil {
				return err
			}
			for k, v := range spanMap {
				if err = epochBucket.Put(bytesutil.Bytes8(k), marshalSpan(v)); err != nil {
					return err
				}
			}
			epochSpansCacheEvictions.Inc()
			return nil
		})
		if err != nil {
			log.Errorf("Failed to save span map to db on cache eviction: %v", err)
		}
	}
}

// Unmarshal a span map from an encoded, flattened array.
func unmarshalSpan(ctx context.Context, enc []byte) (types.Span, error) {
	ctx, span := trace.StartSpan(ctx, "slasherDB.unmarshalSpan")
	defer span.End()
	r := types.Span{}
	if len(enc) != spannerEncodedLength {
		return r, errors.New("wrong data length for min max span")
	}
	r.MinSpan = bytesutil.FromBytes2(enc[:2])
	r.MaxSpan = bytesutil.FromBytes2(enc[2:4])
	sigB := [2]byte{}
	copy(sigB[:], enc[4:6])
	r.SigBytes = sigB
	r.HasAttested = bytesutil.ToBool(enc[6])
	return r, nil
}

// Convert the span struct into a flattened array.
func marshalSpan(span types.Span) []byte {
	return append(append(append(
		bytesutil.Bytes2(uint64(span.MinSpan)),
		bytesutil.Bytes2(uint64(span.MaxSpan))...),
		span.SigBytes[:]...),
		bytesutil.FromBool(span.HasAttested),
	)
}

// EpochSpansMap accepts epoch and returns the corresponding spans map epoch=>spans
// for slashing detection. This function reads spans from cache if caching is
// enabled and the epoch key exists.
// Returns span maps, retrieved from cache bool,
// and error in case of db error. returns empty map if the span map
// for this validator index does not exist.
func (db *Store) EpochSpansMap(ctx context.Context, epoch uint64) (map[uint64]types.Span, bool, error) {
	ctx, span := trace.StartSpan(ctx, "slasherDB.EpochSpansMap")
	defer span.End()
	if db.spanCacheEnabled {
		spanMap, ok := db.spanCache.Get(epoch)
		if ok {
			return spanMap, true, nil
		}
	}

	var err error
	var spanMap map[uint64]types.Span
	err = db.view(func(tx *bolt.Tx) error {
		b := tx.Bucket(validatorsMinMaxSpanBucket)
		epochBucket := b.Bucket(bytesutil.Bytes8(epoch))
		if epochBucket == nil {
			return nil
		}
		keysLength := epochBucket.Stats().KeyN
		spanMap = make(map[uint64]types.Span, keysLength)
		return epochBucket.ForEach(func(k, v []byte) error {
			key := bytesutil.FromBytes8(k)
			value, err := unmarshalSpan(ctx, v)
			if err != nil {
				return err
			}
			spanMap[key] = value
			return nil
		})
	})
	if spanMap == nil {
		spanMap = make(map[uint64]types.Span)
	}
	return spanMap, false, err
}

// EpochSpanByValidatorIndex accepts validator index and epoch returns the corresponding spans
// for slashing detection.
// it reads the epoch spans from cache and gets the requested value from there if it exists
// when caching is enabled.
// Returns error if the spans for this validator index and epoch does not exist.
func (db *Store) EpochSpanByValidatorIndex(ctx context.Context, validatorIdx uint64, epoch uint64) (types.Span, error) {
	ctx, span := trace.StartSpan(ctx, "slasherDB.EpochSpanByValidatorIndex")
	defer span.End()
	if db.spanCacheEnabled {
		setObservedEpochs(epoch)
		spanMap, err := db.findOrLoadEpochInCache(ctx, epoch)
		if err != nil {
			return types.Span{}, err
		}
		spans, ok := spanMap[validatorIdx]
		if ok {
			return spans, nil
		}
		return types.Span{}, nil
	}

	var spans types.Span
	err := db.view(func(tx *bolt.Tx) error {
		b := tx.Bucket(validatorsMinMaxSpanBucket)
		epochBucket := b.Bucket(bytesutil.Bytes8(epoch))
		if epochBucket == nil {
			return nil
		}
		key := bytesutil.Bytes8(validatorIdx)
		v := epochBucket.Get(key)
		if v == nil {
			return nil
		}
		value, err := unmarshalSpan(ctx, v)
		if err != nil {
			return err
		}
		spans = value
		return nil
	})
	return spans, err
}

// EpochsSpanByValidatorsIndices accepts validator indices and epoch and
// returns all their previous corresponding spans for slashing detection epoch=> validator index => spammap.
// Returns empty map if no values exists and error on db error.
func (db *Store) EpochsSpanByValidatorsIndices(ctx context.Context, validatorIndices []uint64, maxEpoch uint64) (map[uint64]map[uint64]types.Span, error) {
	ctx, span := trace.StartSpan(ctx, "slasherDB.EpochsSpanByValidatorsIndices")
	defer span.End()

	var err error
	epochsSpanMap := make(map[uint64]map[uint64]types.Span)
	err = db.view(func(tx *bolt.Tx) error {
		b := tx.Bucket(validatorsMinMaxSpanBucket)
		epoch := maxEpoch
		epochBucket := b.Bucket(bytesutil.Bytes8(epoch))

		for epochBucket != nil {
			valSpans := make(map[uint64]types.Span, len(validatorIndices))
			for _, v := range validatorIndices {
				enc := epochBucket.Get(bytesutil.Bytes8(v))
				if enc == nil {
					continue
				}
				value, err := unmarshalSpan(ctx, enc)
				if err != nil {
					return err
				}
				valSpans[v] = value
			}
			epochsSpanMap[epoch] = valSpans
			if epoch == 0 {
				break
			}
			epoch--
			epochBucket = b.Bucket(bytesutil.Bytes8(epoch))
		}
		return nil
	})
	return epochsSpanMap, err
}

// SaveEpochsSpanByValidatorsIndices accepts epochs span maps by validator indices and
// writes them to db.
// Returns error on db write error.
func (db *Store) SaveEpochsSpanByValidatorsIndices(ctx context.Context, epochsSpans map[uint64]map[uint64]types.Span) error {
	ctx, span := trace.StartSpan(ctx, "slasherDB.SaveEpochsSpanByValidatorsIndices")
	defer span.End()

	err := db.update(func(tx *bolt.Tx) error {
		b := tx.Bucket(validatorsMinMaxSpanBucket)
		for epoch, indicesSpanMaps := range epochsSpans {
			epochBucket, err := b.CreateBucketIfNotExists(bytesutil.Bytes8(epoch))
			if err != nil {
				return err
			}
			for idx, v := range indicesSpanMaps {
				enc := marshalSpan(v)
				if err := epochBucket.Put(bytesutil.Bytes8(idx), enc); err != nil {
					return err
				}
			}
		}
		return nil
	})
	return err
}

// SaveValidatorEpochSpan accepts validator index epoch and spans returns.
// it reads the epoch spans from cache, updates it and save it back to cache
// if caching is enabled.
// Returns error if the spans for this validator index and epoch does not exist.
func (db *Store) SaveValidatorEpochSpan(
	ctx context.Context,
	validatorIdx uint64,
	epoch uint64,
	span types.Span,
) error {
	ctx, traceSpan := trace.StartSpan(ctx, "slasherDB.SaveValidatorEpochSpan")
	defer traceSpan.End()
	if db.spanCacheEnabled {
		setObservedEpochs(epoch)
		spanMap, err := db.findOrLoadEpochInCache(ctx, epoch)
		if err != nil {
			return err
		}
		spanMap[validatorIdx] = span
		db.spanCache.Set(epoch, spanMap)
		return nil
	}

	return db.update(func(tx *bolt.Tx) error {
		b := tx.Bucket(validatorsMinMaxSpanBucket)
		epochBucket, err := b.CreateBucketIfNotExists(bytesutil.Bytes8(epoch))
		if err != nil {
			return err
		}
		key := bytesutil.Bytes8(validatorIdx)
		value := marshalSpan(span)
		return epochBucket.Put(key, value)
	})
}

// SaveEpochSpansMap accepts a epoch and span map epoch=>spans and writes it to disk.
// saves the spans to cache if caching is enabled. The key in the cache is the
// epoch and the value is the span map itself.
func (db *Store) SaveEpochSpansMap(ctx context.Context, epoch uint64, spanMap map[uint64]types.Span) error {
	ctx, span := trace.StartSpan(ctx, "slasherDB.SaveEpochSpansMap")
	defer span.End()
	if db.spanCacheEnabled {
		setObservedEpochs(epoch)
		db.spanCache.Set(epoch, spanMap)
		return nil
	}

	return db.update(func(tx *bolt.Tx) error {
		bucket := tx.Bucket(validatorsMinMaxSpanBucket)
		valBucket, err := bucket.CreateBucketIfNotExists(bytesutil.Bytes8(epoch))
		if err != nil {
			return err
		}
		for k, v := range spanMap {
			err = valBucket.Put(bytesutil.Bytes8(k), marshalSpan(v))
			if err != nil {
				return err
			}
		}
		return nil
	})
}

// EnableSpanCache used to enable or disable span map cache in tests.
func (db *Store) EnableSpanCache(enable bool) {
	db.spanCacheEnabled = enable
}

// SaveCachedSpansMaps saves all span maps that are currently
// in memory into the DB. if no span maps are in db or cache is disabled it returns nil.
func (db *Store) SaveCachedSpansMaps(ctx context.Context) error {
	ctx, span := trace.StartSpan(ctx, "slasherDB.SaveCachedSpansMaps")
	defer span.End()
	if db.spanCacheEnabled {
		db.EnableSpanCache(false)
		defer db.EnableSpanCache(true)
		for epoch := lowestObservedEpoch; epoch <= highestObservedEpoch; epoch++ {
			spanMap, ok := db.spanCache.Get(epoch)
			if ok {
				if err := db.SaveEpochSpansMap(ctx, epoch, spanMap); err != nil {
					return errors.Wrap(err, "failed to save span maps from cache")
				}
			}
		}
		// Reset the observed epochs after saving to the DB.
		lowestObservedEpoch = params.BeaconConfig().FarFutureEpoch
		highestObservedEpoch = 0
		log.Debugf("Epochs %d to %d have been saved", lowestObservedEpoch, highestObservedEpoch)
	}
	return nil
}

// DeleteEpochSpans deletes a epochs validators span map using a epoch index as bucket key.
func (db *Store) DeleteEpochSpans(ctx context.Context, epoch uint64) error {
	ctx, span := trace.StartSpan(ctx, "slasherDB.DeleteEpochSpans")
	defer span.End()
	if db.spanCacheEnabled {
		_ = db.spanCache.Delete(epoch)
		return nil
	}
	return db.update(func(tx *bolt.Tx) error {
		bucket := tx.Bucket(validatorsMinMaxSpanBucket)
		key := bytesutil.Bytes8(epoch)
		return bucket.DeleteBucket(key)
	})
}

// DeleteValidatorSpanByEpoch deletes a validator span for a certain epoch
// deletes spans from cache if caching is enabled.
// using a validator index as bucket key.
func (db *Store) DeleteValidatorSpanByEpoch(ctx context.Context, validatorIdx uint64, epoch uint64) error {
	ctx, span := trace.StartSpan(ctx, "slasherDB.DeleteValidatorSpanByEpoch")
	defer span.End()
	if db.spanCacheEnabled {
		spanMap, ok := db.spanCache.Get(epoch)
		if ok {
			delete(spanMap, validatorIdx)
			db.spanCache.Set(epoch, spanMap)
			return nil
		}
	}

	return db.update(func(tx *bolt.Tx) error {
		bucket := tx.Bucket(validatorsMinMaxSpanBucket)
		e := bytesutil.Bytes8(epoch)
		epochBucket := bucket.Bucket(e)
		v := bytesutil.Bytes8(validatorIdx)
		return epochBucket.Delete(v)
	})
}

// findOrLoadEpochInCache checks if the requested epoch is in the cache, and if not, we load it from the DB.
func (db *Store) findOrLoadEpochInCache(ctx context.Context, epoch uint64) (map[uint64]types.Span, error) {
	ctx, span := trace.StartSpan(ctx, "slasherDB.findOrLoadEpochInCache")
	defer span.End()
	spanMap, epochFound := db.spanCache.Get(epoch)
	if epochFound {
		return spanMap, nil
	}

	db.EnableSpanCache(false)
	defer db.EnableSpanCache(true)
	// If the epoch we want isn't in the cache, load it in.
	spanForEpoch, _, err := db.EpochSpansMap(ctx, epoch)
	if err != nil {
		return make(map[uint64]types.Span), errors.Wrap(err, "failed to get span map for epoch")
	}
	db.spanCache.Set(epoch, spanForEpoch)
	return spanForEpoch, nil
}

func setObservedEpochs(epoch uint64) {
	if epoch > highestObservedEpoch {
		slasherHighestObservedEpoch.Set(float64(epoch))
		highestObservedEpoch = epoch
	}
	if epoch < lowestObservedEpoch {
		slasherLowestObservedEpoch.Set(float64(epoch))
		lowestObservedEpoch = epoch
	}
}