prysm-pulse/beacon-chain/archiver/service.go
terence tsao 81c53c26fb Update run time to v0.9 (#3935)
* Update seed domains (#3872)

* Remove Transfers (#3870)

* Remove active index roots and compact committee roots (#3869)

* Update inclusion reward (#3886)

* Alter proposer selection logic (#3884)

* Fix early committee bias (#3888)

* Remove shards and committees (#3896)

* Epoch spec tests v0.9 (#3907)

* Block spec test v0.9 (#3905)

* rm'ed in protobuf

* build proto

* build proto

* build proto

* fix core package

* Gazelle

* Fixed all the tests

* Fixed static test

* Comment out spec test for now

* One more skip

* fix-roundRobinSync (#3862)

* Starting but need new seed function

* Revert initial sync

* Updated Proposer Slashing

* Fixed all tests

* Lint

* Update inclusion reward

* Fill randao mixes with eth1 data hash

* Test

* Fixing test part1

* All tests passing

* One last test

* Updated config

* Build proto

* Proper skip message

* Conflict and fmt

* Removed crosslinks and shards. Built

* Format and gazelle

* Fixed all the block package tests

* Fixed all the helper tests

* All epoch package tests pass

* All core package tests pass

* Fixed operation tests

* Started fixing rpc test

* RPC tests passed!

* Fixed all init sync tests

* All tests pass

* Fixed blockchain tests

* Lint

* Lint

* Preston's feedback

* Starting

* Remove container

* Fixed block spec tests

* All passing except for block_processing test

* Failing block processing test

* Starting

* Add AggregateAndProof

* All mainnet test passes

* Update deposit contract (#3906)

* Proto spec tests v0.9 (#3908)

* Starting

* Add AggregateAndProof

* Unskip block util tests (#3910)

* rm'ed in protobuf

* build proto

* build proto

* build proto

* fix core package

* Gazelle

* Fixed all the tests

* Fixed static test

* Comment out spec test for now

* One more skip

* fix-roundRobinSync (#3862)

* Starting but need new seed function

* Revert initial sync

* Updated Proposer Slashing

* Fixed all tests

* Lint

* Update inclusion reward

* Fill randao mixes with eth1 data hash

* Test

* Fixing test part1

* All tests passing

* One last test

* Updated config

* Build proto

* Proper skip message

* Conflict and fmt

* Removed crosslinks and shards. Built

* Format and gazelle

* Fixed all the block package tests

* Fixed all the helper tests

* All epoch package tests pass

* All core package tests pass

* Fixed operation tests

* Started fixing rpc test

* RPC tests passed!

* Fixed all init sync tests

* All tests pass

* Fixed blockchain tests

* Lint

* Lint

* Preston's feedback

* Starting

* Remove container

* Fixed block spec tests

* All passing except for block_processing test

* Failing block processing test

* Starting

* Add AggregateAndProof

* All mainnet test passes

* Unskip block util tests

* Slot processing spec test V0.9 (#3912)

* Starting

* Add AggregateAndProof

* Unskip slot processing mainnet test

* Unskip minimal spec test for finalization (#3920)

* Remove outdated interop tests (#3922)

* Rm outdated interop tests

* Rm test runner

* Gazelle

* Update validator to use proposer slot (#3919)

* Fix committee assignment (#3931)

* Replace shard with committee index (#3930)

* Conflict

* Clean up (#3933)

* Remove shard filter in db (#3936)

* Remove lightouse compatibility test (#3939)

* Update Committee Cache for v0.9 (#3948)

* Updated committee cache

* Removed shuffled indices cache

* Started testing run time

* Lint

* Fixed test

* Safeguard against nil head state

* address edge case

* add test

* Fixed TestRoundRobinSync by doubling the epochs

* Unskip TestProtoCompatability (#3958)

* Unskip TestProtoCompatability

* Update WORKSPACE

* Fix minimal config (#3959)

* fix minimal configs

* fix hardcoded value in test

* Simplify verify att time (#3961)

* update readme for deposit contract, regen bindings for vyper 0.1.0b12 (#3963)

* update readme for deposit contract, regen bindings

* medium

* Check nil base state (#3964)

* Copy Block When Receiving it From Sync (#3966)

* copy block

* clone for other service methods too

* Change logging of Bitfield  (#3956)

* change logging of bits

* preston's review

* Unskip Beacon Server Test (#3962)

* run test till the end

* fix up proto message types

* fmt

* resolve broken tests

* better error handling

* fixing new logic to use archived proposer info

* fix up logic

* clip using the max effective balance

* broken build fix with num arg mismatch

* amend archive

* archival logic changed

* rename test

* archive both proposer and attester seeds

* page size 100

* further experiments

* further experimentation, archivedProposerIndex seems wrong

* test passes

* rem log

* fix broken test

* fix test

* gaz

* fix imports

* ethapis
2019-11-11 17:03:43 -05:00

175 lines
5.9 KiB
Go

package archiver
import (
"context"
"fmt"
"github.com/pkg/errors"
"github.com/prysmaticlabs/prysm/beacon-chain/blockchain"
"github.com/prysmaticlabs/prysm/beacon-chain/core/epoch"
"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
"github.com/prysmaticlabs/prysm/beacon-chain/core/validators"
"github.com/prysmaticlabs/prysm/beacon-chain/db"
pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
"github.com/prysmaticlabs/prysm/shared/params"
"github.com/sirupsen/logrus"
)
var log = logrus.WithField("prefix", "archiver")
// Service defining archiver functionality for persisting checkpointed
// beacon chain information to a database backend for historical purposes.
type Service struct {
ctx context.Context
cancel context.CancelFunc
beaconDB db.Database
headFetcher blockchain.HeadFetcher
newHeadNotifier blockchain.NewHeadNotifier
newHeadRootChan chan [32]byte
}
// Config options for the archiver service.
type Config struct {
BeaconDB db.Database
HeadFetcher blockchain.HeadFetcher
NewHeadNotifier blockchain.NewHeadNotifier
}
// NewArchiverService initializes the service from configuration options.
func NewArchiverService(ctx context.Context, cfg *Config) *Service {
ctx, cancel := context.WithCancel(ctx)
return &Service{
ctx: ctx,
cancel: cancel,
beaconDB: cfg.BeaconDB,
headFetcher: cfg.HeadFetcher,
newHeadNotifier: cfg.NewHeadNotifier,
newHeadRootChan: make(chan [32]byte, 1),
}
}
// Start the archiver service event loop.
func (s *Service) Start() {
go s.run(s.ctx)
}
// Stop the archiver service event loop.
func (s *Service) Stop() error {
defer s.cancel()
return nil
}
// Status reports the healthy status of the archiver. Returning nil means service
// is correctly running without error.
func (s *Service) Status() error {
return nil
}
// We archive committee information pertaining to the head state's epoch.
func (s *Service) archiveCommitteeInfo(ctx context.Context, headState *pb.BeaconState) error {
currentEpoch := helpers.SlotToEpoch(headState.Slot)
committeeCount, err := helpers.CommitteeCountAtSlot(headState, helpers.StartSlot(currentEpoch))
if err != nil {
return errors.Wrap(err, "could not get committee count")
}
proposerSeed, err := helpers.Seed(headState, currentEpoch, params.BeaconConfig().DomainBeaconProposer)
if err != nil {
return errors.Wrap(err, "could not generate seed")
}
attesterSeed, err := helpers.Seed(headState, currentEpoch, params.BeaconConfig().DomainBeaconAttester)
if err != nil {
return errors.Wrap(err, "could not generate seed")
}
info := &ethpb.ArchivedCommitteeInfo{
ProposerSeed: proposerSeed[:],
AttesterSeed: attesterSeed[:],
CommitteeCount: committeeCount * params.BeaconConfig().SlotsPerEpoch,
}
if err := s.beaconDB.SaveArchivedCommitteeInfo(ctx, currentEpoch, info); err != nil {
return errors.Wrap(err, "could not archive committee info")
}
return nil
}
// We archive active validator set changes that happened during the epoch.
func (s *Service) archiveActiveSetChanges(ctx context.Context, headState *pb.BeaconState) error {
activations := validators.ActivatedValidatorIndices(headState)
slashings := validators.SlashedValidatorIndices(headState)
exited, err := validators.ExitedValidatorIndices(headState)
if err != nil {
return errors.Wrap(err, "could not determine exited validator indices")
}
activeSetChanges := &ethpb.ArchivedActiveSetChanges{
Activated: activations,
Exited: exited,
Slashed: slashings,
}
if err := s.beaconDB.SaveArchivedActiveValidatorChanges(ctx, helpers.CurrentEpoch(headState), activeSetChanges); err != nil {
return errors.Wrap(err, "could not archive active validator set changes")
}
return nil
}
// We compute participation metrics by first retrieving the head state and
// matching validator attestations during the epoch.
func (s *Service) archiveParticipation(ctx context.Context, headState *pb.BeaconState) error {
participation, err := epoch.ComputeValidatorParticipation(headState)
if err != nil {
return errors.Wrap(err, "could not compute participation")
}
return s.beaconDB.SaveArchivedValidatorParticipation(ctx, helpers.SlotToEpoch(headState.Slot), participation)
}
// We archive validator balances and active indices.
func (s *Service) archiveBalances(ctx context.Context, headState *pb.BeaconState) error {
balances := headState.Balances
currentEpoch := helpers.CurrentEpoch(headState)
if err := s.beaconDB.SaveArchivedBalances(ctx, currentEpoch, balances); err != nil {
return errors.Wrap(err, "could not archive balances")
}
return nil
}
func (s *Service) run(ctx context.Context) {
sub := s.newHeadNotifier.HeadUpdatedFeed().Subscribe(s.newHeadRootChan)
defer sub.Unsubscribe()
for {
select {
case r := <-s.newHeadRootChan:
log.WithField("headRoot", fmt.Sprintf("%#x", r)).Debug("New chain head event")
headState := s.headFetcher.HeadState()
if !helpers.IsEpochEnd(headState.Slot) {
continue
}
if err := s.archiveCommitteeInfo(ctx, headState); err != nil {
log.WithError(err).Error("Could not archive committee info")
continue
}
if err := s.archiveActiveSetChanges(ctx, headState); err != nil {
log.WithError(err).Error("Could not archive active validator set changes")
continue
}
if err := s.archiveParticipation(ctx, headState); err != nil {
log.WithError(err).Error("Could not archive validator participation")
continue
}
if err := s.archiveBalances(ctx, headState); err != nil {
log.WithError(err).Error("Could not archive validator balances and active indices")
continue
}
log.WithField(
"epoch",
helpers.CurrentEpoch(headState),
).Debug("Successfully archived beacon chain data during epoch")
case <-s.ctx.Done():
log.Debug("Context closed, exiting goroutine")
return
case err := <-sub.Err():
log.WithError(err).Error("Subscription to new chain head notifier failed")
return
}
}
}