mirror of
https://gitlab.com/pulsechaincom/prysm-pulse.git
synced 2024-12-26 05:17:22 +00:00
f77049ae74
* Fmt * Starting * Cont * Store aggregate attestation is better * Conflict * Done * Merge branch 'master' of git+ssh://github.com/prysmaticlabs/prysm into process-pending-atts * Comment * Better logs * Better logs * Fix existing tests * Update metric names * Preston's feedback * Broadcast atts once it's valid * Gazelle * Test for validating atts and pruning * Tests * Removed debug log * Conflict * Feedback * Merge refs/heads/master into process-pending-atts * Merge refs/heads/master into process-pending-atts
123 lines
3.7 KiB
Go
123 lines
3.7 KiB
Go
package sync
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
|
|
"github.com/kevinms/leakybucket-go"
|
|
"github.com/pkg/errors"
|
|
ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/blockchain"
|
|
statefeed "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/state"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/db"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/operations/attestations"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/operations/voluntaryexits"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
|
|
"github.com/prysmaticlabs/prysm/shared"
|
|
)
|
|
|
|
var _ = shared.Service(&Service{})
|
|
|
|
const allowedBlocksPerSecond = 32.0
|
|
const allowedBlocksBurst = 10 * allowedBlocksPerSecond
|
|
|
|
// Config to set up the regular sync service.
|
|
type Config struct {
|
|
P2P p2p.P2P
|
|
DB db.NoHeadAccessDatabase
|
|
AttPool attestations.Pool
|
|
ExitPool *voluntaryexits.Pool
|
|
Chain blockchainService
|
|
InitialSync Checker
|
|
StateNotifier statefeed.Notifier
|
|
}
|
|
|
|
// This defines the interface for interacting with block chain service
|
|
type blockchainService interface {
|
|
blockchain.BlockReceiver
|
|
blockchain.HeadFetcher
|
|
blockchain.FinalizationFetcher
|
|
blockchain.ForkFetcher
|
|
blockchain.AttestationReceiver
|
|
blockchain.TimeFetcher
|
|
}
|
|
|
|
// NewRegularSync service.
|
|
func NewRegularSync(cfg *Config) *Service {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
r := &Service{
|
|
ctx: ctx,
|
|
cancel: cancel,
|
|
db: cfg.DB,
|
|
p2p: cfg.P2P,
|
|
attPool: cfg.AttPool,
|
|
exitPool: cfg.ExitPool,
|
|
chain: cfg.Chain,
|
|
initialSync: cfg.InitialSync,
|
|
slotToPendingBlocks: make(map[uint64]*ethpb.SignedBeaconBlock),
|
|
seenPendingBlocks: make(map[[32]byte]bool),
|
|
blkRootToPendingAtts: make(map[[32]byte][]*ethpb.AggregateAttestationAndProof),
|
|
stateNotifier: cfg.StateNotifier,
|
|
blocksRateLimiter: leakybucket.NewCollector(allowedBlocksPerSecond, allowedBlocksBurst, false /* deleteEmptyBuckets */),
|
|
}
|
|
|
|
r.registerRPCHandlers()
|
|
r.registerSubscribers()
|
|
|
|
return r
|
|
}
|
|
|
|
// Service is responsible for handling all run time p2p related operations as the
|
|
// main entry point for network messages.
|
|
type Service struct {
|
|
ctx context.Context
|
|
cancel context.CancelFunc
|
|
p2p p2p.P2P
|
|
db db.NoHeadAccessDatabase
|
|
attPool attestations.Pool
|
|
exitPool *voluntaryexits.Pool
|
|
chain blockchainService
|
|
slotToPendingBlocks map[uint64]*ethpb.SignedBeaconBlock
|
|
seenPendingBlocks map[[32]byte]bool
|
|
blkRootToPendingAtts map[[32]byte][]*ethpb.AggregateAttestationAndProof
|
|
pendingAttsLock sync.RWMutex
|
|
pendingQueueLock sync.RWMutex
|
|
chainStarted bool
|
|
initialSync Checker
|
|
validateBlockLock sync.RWMutex
|
|
stateNotifier statefeed.Notifier
|
|
blocksRateLimiter *leakybucket.Collector
|
|
}
|
|
|
|
// Start the regular sync service.
|
|
func (r *Service) Start() {
|
|
r.p2p.AddConnectionHandler(r.sendRPCStatusRequest)
|
|
r.p2p.AddDisconnectionHandler(r.removeDisconnectedPeerStatus)
|
|
r.processPendingBlocksQueue()
|
|
r.processPendingAttsQueue()
|
|
r.maintainPeerStatuses()
|
|
r.resyncIfBehind()
|
|
}
|
|
|
|
// Stop the regular sync service.
|
|
func (r *Service) Stop() error {
|
|
defer r.cancel()
|
|
return nil
|
|
}
|
|
|
|
// Status of the currently running regular sync service.
|
|
func (r *Service) Status() error {
|
|
if r.chainStarted && r.initialSync.Syncing() {
|
|
return errors.New("waiting for initial sync")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Checker defines a struct which can verify whether a node is currently
|
|
// synchronizing a chain with the rest of peers in the network.
|
|
type Checker interface {
|
|
Syncing() bool
|
|
Status() error
|
|
Resync() error
|
|
}
|