package kv import ( "time" "github.com/patrickmn/go-cache" "github.com/pkg/errors" ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1" "github.com/prysmaticlabs/go-ssz" "github.com/prysmaticlabs/prysm/beacon-chain/core/helpers" ) // SaveAggregatedAttestation saves an aggregated attestation in cache. func (p *AttCaches) SaveAggregatedAttestation(att *ethpb.Attestation) error { if !helpers.IsAggregated(att) { return errors.New("attestation is not aggregated") } r, err := ssz.HashTreeRoot(att.Data) if err != nil { return errors.Wrap(err, "could not tree hash attestation") } d, expTime, ok := p.aggregatedAtt.GetWithExpiration(string(r[:])) // If we have not seen the attestation data before, store in in cache with // the default expiration time out. if !ok { atts := []*ethpb.Attestation{att} p.aggregatedAtt.Set(string(r[:]), atts, cache.DefaultExpiration) return nil } atts, ok := d.([]*ethpb.Attestation) if !ok { return errors.New("cached value is not of type []*ethpb.Attestation") } atts, err = helpers.AggregateAttestations(append(atts, att)) if err != nil { return err } // Delete attestation if the current time has passed the expiration time. if time.Now().Unix() >= expTime.Unix() { p.aggregatedAtt.Delete(string(r[:])) return nil } // Reset expiration time given how much time has passed. expDuration := time.Duration(expTime.Unix() - time.Now().Unix()) p.aggregatedAtt.Set(string(r[:]), atts, expDuration*time.Second) return nil } // SaveAggregatedAttestations saves a list of aggregated attestations in cache. func (p *AttCaches) SaveAggregatedAttestations(atts []*ethpb.Attestation) error { for _, att := range atts { if err := p.SaveAggregatedAttestation(att); err != nil { return err } } return nil } // AggregatedAttestations returns the aggregated attestations in cache. func (p *AttCaches) AggregatedAttestations() []*ethpb.Attestation { // Delete all expired aggregated attestations before returning them. p.aggregatedAtt.DeleteExpired() atts := make([]*ethpb.Attestation, 0, p.aggregatedAtt.ItemCount()) for s, i := range p.aggregatedAtt.Items() { // Type assertion for the worst case. This shouldn't happen. a, ok := i.Object.([]*ethpb.Attestation) if !ok { p.aggregatedAtt.Delete(s) continue } atts = append(atts, a...) } return atts } // AggregatedAttestationsBySlotIndex returns the aggregated attestations in cache, // filtered by committee index and slot. func (p *AttCaches) AggregatedAttestationsBySlotIndex(slot uint64, committeeIndex uint64) []*ethpb.Attestation { // Delete all expired aggregated attestations before returning them. p.aggregatedAtt.DeleteExpired() atts := make([]*ethpb.Attestation, 0, p.aggregatedAtt.ItemCount()) for s, i := range p.aggregatedAtt.Items() { // Type assertion for the worst case. This shouldn't happen. a, ok := i.Object.([]*ethpb.Attestation) if !ok { p.aggregatedAtt.Delete(s) continue } if slot == a[0].Data.Slot && committeeIndex == a[0].Data.CommitteeIndex { atts = append(atts, a...) } } return atts } // DeleteAggregatedAttestation deletes the aggregated attestations in cache. func (p *AttCaches) DeleteAggregatedAttestation(att *ethpb.Attestation) error { if !helpers.IsAggregated(att) { return errors.New("attestation is not aggregated") } r, err := ssz.HashTreeRoot(att.Data) if err != nil { return errors.Wrap(err, "could not tree hash attestation data") } a, expTime, ok := p.aggregatedAtt.GetWithExpiration(string(r[:])) if !ok { return nil } atts, ok := a.([]*ethpb.Attestation) if !ok { return errors.New("cached value is not of type []*ethpb.Attestation") } filtered := make([]*ethpb.Attestation, 0) for _, a := range atts { if !att.AggregationBits.Contains(a.AggregationBits) { filtered = append(filtered, a) } } if len(filtered) == 0 { p.aggregatedAtt.Delete(string(r[:])) } else { // Delete attestation if the current time has passed the expiration time. if time.Now().Unix() >= expTime.Unix() { p.aggregatedAtt.Delete(string(r[:])) return nil } // Reset expiration time given how much time has passed. expDuration := time.Duration(expTime.Unix() - time.Now().Unix()) p.aggregatedAtt.Set(string(r[:]), filtered, expDuration*time.Second) } return nil } // HasAggregatedAttestation checks if the input attestations has already existed in cache. func (p *AttCaches) HasAggregatedAttestation(att *ethpb.Attestation) (bool, error) { r, err := ssz.HashTreeRoot(att.Data) if err != nil { return false, errors.Wrap(err, "could not tree hash attestation") } if atts, ok := p.aggregatedAtt.Get(string(r[:])); ok { for _, a := range atts.([]*ethpb.Attestation) { if a.AggregationBits.Contains(att.AggregationBits) { return true, nil } } } if atts, ok := p.blockAtt.Get(string(r[:])); ok { for _, a := range atts.([]*ethpb.Attestation) { if a.AggregationBits.Contains(att.AggregationBits) { return true, nil } } } return false, nil } // AggregatedAttestationCount returns the number of aggregated attestations key in the pool. func (p *AttCaches) AggregatedAttestationCount() int { return p.aggregatedAtt.ItemCount() }