sync

package
v7.1.8 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 23, 2026 License: GPL-3.0 Imports: 109 Imported by: 1

Documentation

Overview

Package sync TODO(3147): Add details on how sync works.

Code generated by hack/gen-logs.sh; DO NOT EDIT. This file is created and regenerated automatically. Anything added here might get removed.

Package sync includes all chain-synchronization logic for the beacon node, including gossip-sub validators for blocks, attestations, and other p2p messages, as well as ability to process and respond to block requests by peers.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrNoPeersCoverNeeded is returned when no peers are able to cover the needed columns.
	ErrNoPeersCoverNeeded = errors.New("no peers able to cover needed columns")
	// ErrNoPeersAvailable is returned when no peers are available for block requests.
	ErrNoPeersAvailable = errors.New("no peers available")
)
View Source 
var (
	ErrNoValidDigest       = errors.New("no valid digest matched")
	ErrUnrecognizedVersion = errors.New("cannot determine context bytes for unrecognized object")
)
View Source 
var (
	ErrOptimisticParent = errors.New("parent of the block is optimistic")

	ErrSlashingSignatureFailure = errors.New("proposer slashing signature verification failed")
)
View Source 
var (
	// ErrInvalidFetchedData is used to signal that an error occurred which should result in peer downscoring.
	ErrInvalidFetchedData = errors.New("invalid data returned from peer")
)

Any error from the following declaration block should result in peer downscoring.

View Source 
var ErrSidecarHeaderMismatch = errors.New("data column sidecar signed block header does not match local block")

Functions

func BlobRPCMinValidSlot 

func BlobRPCMinValidSlot(current primitives.Slot) (primitives.Slot, error)

BlobRPCMinValidSlot returns the lowest slot that we should expect peers to respect as the start slot in a BlobSidecarsByRange request. This can be used to validate incoming requests and to avoid pestering peers with requests for blobs that are outside the retention window.

func DataColumnSidecarsByRangeRequest added in v7.1.0 

func DataColumnSidecarsByRangeRequest(columns []uint64, start, end primitives.Slot) (*ethpb.DataColumnSidecarsByRangeRequest, error)

func FetchDataColumnSidecars 

func FetchDataColumnSidecars(
	params DataColumnSidecarsParams,
	roBlocks []blocks.ROBlock,
	requestedIndices map[uint64]bool,
) (map[[fieldparams.RootLength]byte][]blocks.VerifiedRODataColumn, map[[fieldparams.RootLength]byte]map[uint64]bool, error)

FetchDataColumnSidecars retrieves data column sidecars for the given blocks and indices using a series of fallback strategies.

For each block in `roBlocks` that has commitments, the function attempts to obtain all sidecars corresponding to the indices listed in `requestedIndices`.

The function returns:

  • A map from block root to the sidecars successfully retrieved.
  • A set of block roots for which not all requested sidecars could be retrieved.

Retrieval strategy (proceeds to the next step only if not all requested sidecars were successfully obtained at the current step):

  1. Attempt to load the requested sidecars from storage, reconstructing them from other available sidecars in storage if necessary.
  2. Request any missing sidecars from peers. If some are still missing, attempt to reconstruct them using both stored sidecars and those retrieved from peers.
  3. Request all remaining possible sidecars from peers that are not already in storage or retrieved in step 2. Stop once either all requested sidecars are retrieved, or enough sidecars are available (from storage, step 2, and step 3) to reconstruct the requested ones.

func ReadChunkedBlock 

func ReadChunkedBlock(stream libp2pcore.Stream, tor blockchain.TemporalOracle, p2p p2p.EncodingProvider, isFirstChunk bool) (interfaces.ReadOnlySignedBeaconBlock, error)

ReadChunkedBlock handles each response chunk that is sent by the peer and converts it into a beacon block.

func ReadStatusCode 

func ReadStatusCode(stream network.Stream, encoding encoder.NetworkEncoding) (uint8, string, error)

ReadStatusCode response from a RPC stream.

func SendBeaconBlocksByRangeRequest 

func SendBeaconBlocksByRangeRequest(
	ctx context.Context, tor blockchain.TemporalOracle, p2pProvider p2p.SenderEncoder, pid peer.ID,
	req *ethpb.BeaconBlocksByRangeRequest, blockProcessor BeaconBlockProcessor,
) ([]interfaces.ReadOnlySignedBeaconBlock, error)

SendBeaconBlocksByRangeRequest sends BeaconBlocksByRange and returns fetched blocks, if any.

func SendBeaconBlocksByRootRequest 

func SendBeaconBlocksByRootRequest(
	ctx context.Context, clock blockchain.TemporalOracle, p2pProvider p2p.P2P, pid peer.ID,
	req *p2ptypes.BeaconBlockByRootsReq, blockProcessor BeaconBlockProcessor,
) ([]interfaces.ReadOnlySignedBeaconBlock, error)

SendBeaconBlocksByRootRequest sends BeaconBlocksByRoot and returns fetched blocks, if any.

func SendBlobSidecarByRoot 

func SendBlobSidecarByRoot(
	ctx context.Context, tor blockchain.TemporalOracle, p2pApi p2p.P2P, pid peer.ID,
	ctxMap ContextByteVersions, req *p2ptypes.BlobSidecarsByRootReq, slot primitives.Slot,
) ([]blocks.ROBlob, error)

func SendBlobsByRangeRequest 

func SendBlobsByRangeRequest(ctx context.Context, tor blockchain.TemporalOracle, p2pApi p2p.SenderEncoder, pid peer.ID, ctxMap ContextByteVersions, req *ethpb.BlobSidecarsByRangeRequest, bvs ...BlobResponseValidation) ([]blocks.ROBlob, error)

func SendDataColumnSidecarsByRangeRequest 

func SendDataColumnSidecarsByRangeRequest(
	p DataColumnSidecarsParams,
	pid peer.ID,
	request *ethpb.DataColumnSidecarsByRangeRequest,
	vfs ...DataColumnResponseValidation,
) ([]blocks.RODataColumn, error)

SendDataColumnSidecarsByRangeRequest sends a request for data column sidecars by range and returns the fetched data column sidecars.

func SendDataColumnSidecarsByRootRequest 

func SendDataColumnSidecarsByRootRequest(p DataColumnSidecarsParams, peer goPeer.ID, identifiers p2ptypes.DataColumnsByRootIdentifiers) ([]blocks.RODataColumn, error)

SendDataColumnSidecarsByRootRequest sends a request for data column sidecars by root and returns the fetched data column sidecars.

func SendExecutionPayloadEnvelopesByRangeRequest added in v7.1.4 

func SendExecutionPayloadEnvelopesByRangeRequest(
	ctx context.Context,
	tor blockchain.TemporalOracle,
	p2pProvider p2p.SenderEncoder,
	pid peer.ID,
	ctxMap ContextByteVersions,
	req *ethpb.ExecutionPayloadEnvelopesByRangeRequest,
) ([]*ethpb.SignedExecutionPayloadEnvelope, error)

SendExecutionPayloadEnvelopesByRangeRequest sends ExecutionPayloadEnvelopesByRange and returns fetched envelopes, if any.

func SendExecutionPayloadEnvelopesByRootRequest added in v7.1.4 

func SendExecutionPayloadEnvelopesByRootRequest(
	ctx context.Context, tor blockchain.TemporalOracle, p2pApi p2p.P2P, pid peer.ID,
	ctxMap ContextByteVersions, req *p2ptypes.ExecutionPayloadEnvelopesByRootReq,
) ([]*ethpb.SignedExecutionPayloadEnvelope, error)

SendExecutionPayloadEnvelopesByRootRequest sends ExecutionPayloadEnvelopesByRoot and returns fetched envelopes, if any.

func SetRPCStreamDeadlines 

func SetRPCStreamDeadlines(stream network.Stream)

SetRPCStreamDeadlines sets read and write deadlines for libp2p-based connection streams.

func SetStreamReadDeadline 

func SetStreamReadDeadline(stream network.Stream, duration time.Duration)

SetStreamReadDeadline for reading from libp2p connection streams, deciding when to close a connection based on a particular duration.

NOTE: libp2p uses the system clock time for determining the deadline so we use time.Now() instead of the synchronized roughtime.Now(). If the system time is corrupted (i.e. time does not advance), the node will experience issues being able to properly close streams, leading to unexpected failures and possible memory leaks.

func SetStreamWriteDeadline 

func SetStreamWriteDeadline(stream network.Stream, duration time.Duration)

SetStreamWriteDeadline for writing to libp2p connection streams, deciding when to close a connection based on a particular duration.

NOTE: libp2p uses the system clock time for determining the deadline so we use time.Now() instead of the synchronized roughtime.Now(). If the system time is corrupted (i.e. time does not advance), the node will experience issues being able to properly close streams, leading to unexpected failures and possible memory leaks.

func WriteBlobSidecarChunk 

func WriteBlobSidecarChunk(stream libp2pcore.Stream, tor blockchain.TemporalOracle, encoding encoder.NetworkEncoding, sidecar blocks.VerifiedROBlob) error

WriteBlobSidecarChunk writes blob chunk object to stream. response_chunk ::= <result> | <context-bytes> | <encoding-dependent-header> | <encoded-payload>

func WriteBlockChunk 

func WriteBlockChunk(stream libp2pcore.Stream, tor blockchain.TemporalOracle, encoding encoder.NetworkEncoding, blk interfaces.ReadOnlySignedBeaconBlock) error

WriteBlockChunk writes block chunk object to stream. response_chunk ::= <result> | <context-bytes> | <encoding-dependent-header> | <encoded-payload>

func WriteDataColumnSidecarChunk 

func WriteDataColumnSidecarChunk(stream libp2pcore.Stream, tor blockchain.TemporalOracle, encoding encoder.NetworkEncoding, sidecar blocks.RODataColumn) error

WriteDataColumnSidecarChunk writes data column chunk object to stream. response_chunk ::= <result> | <context-bytes> | <encoding-dependent-header> | <encoded-payload>

func WriteExecutionPayloadEnvelopeChunk added in v7.1.4 

func WriteExecutionPayloadEnvelopeChunk(stream libp2pcore.Stream, encoding encoder.NetworkEncoding, envelope *ethpb.SignedExecutionPayloadEnvelope) error

WriteExecutionPayloadEnvelopeChunk writes execution payload envelope chunk object to stream. response_chunk ::= <result> | <context-bytes> | <encoding-dependent-header> | <encoded-payload>

func WriteLightClientBootstrapChunk 

func WriteLightClientBootstrapChunk(stream libp2pcore.Stream, tor blockchain.TemporalOracle, encoding encoder.NetworkEncoding, bootstrap interfaces.LightClientBootstrap) error

func WriteLightClientFinalityUpdateChunk 

func WriteLightClientFinalityUpdateChunk(stream libp2pcore.Stream, tor blockchain.TemporalOracle, encoding encoder.NetworkEncoding, update interfaces.LightClientFinalityUpdate) error

func WriteLightClientOptimisticUpdateChunk 

func WriteLightClientOptimisticUpdateChunk(stream libp2pcore.Stream, tor blockchain.TemporalOracle, encoding encoder.NetworkEncoding, update interfaces.LightClientOptimisticUpdate) error

func WriteLightClientUpdateChunk 

func WriteLightClientUpdateChunk(stream libp2pcore.Stream, tor blockchain.TemporalOracle, encoding encoder.NetworkEncoding, update interfaces.LightClientUpdate) error

Types

type BeaconBlockProcessor 

type BeaconBlockProcessor func(block interfaces.ReadOnlySignedBeaconBlock) error

BeaconBlockProcessor defines a block processing function, which allows to start utilizing blocks even before all blocks are ready.

type BlobResponseValidation 

type BlobResponseValidation func(blocks.ROBlob) error

BlobResponseValidation represents a function that can validate aspects of a single unmarshaled blob sidecar that was received from a peer in response to an rpc request.

type Checker 

type Checker interface {
	Initialized() bool
	Syncing() bool
	Synced() bool
	Status() error
	Resync() error
}

Checker defines a struct which can verify whether a node is currently synchronizing a chain with the rest of peers in the network.

type ContextByteVersions 

type ContextByteVersions map[[4]byte]int

ContextByteVersions is a mapping between expected values for context bytes and the runtime/version identifier they correspond to. This can be used to look up the type needed to unmarshal a wire-encoded value.

func ContextByteVersionsForValRoot 

func ContextByteVersionsForValRoot(valRoot [32]byte) (ContextByteVersions, error)

ContextByteVersionsForValRoot computes a mapping between all possible context bytes values and the runtime/version identifier for the corresponding fork.

type DASPeerCache added in v7.1.0 

type DASPeerCache struct {
	// contains filtered or unexported fields
}

DASPeerCache caches information about a set of peers DAS peering decisions.

func NewDASPeerCache added in v7.1.0 

func NewDASPeerCache(p2pSvc p2p.P2P) *DASPeerCache

NewDASPeerCache initializes a DASPeerCache. This type is not currently thread safe.

func (*DASPeerCache) NewPicker added in v7.1.0 

func (c *DASPeerCache) NewPicker(pids []peer.ID, toCustody peerdas.ColumnIndices, reqInterval time.Duration) (*PeerPicker, error)

NewColumnScarcityRanking computes the ColumnScarcityRanking based on the current view of columns custodied by the given set of peers. New PeerPickers should be created somewhat frequently, as the status of peers can change, including the set of columns each peer custodies. reqInterval sets the frequency that a peer can be picked in terms of time. A peer can be picked once per reqInterval, eg a value of time.Second would allow 1 request per second to the peer, or a value of 500 * time.Millisecond would allow 2 req/sec.

type DataColumnResponseValidation 

type DataColumnResponseValidation func(column blocks.RODataColumn) error

DataColumnResponseValidation represents a function that can validate aspects of a single unmarshaled data column sidecar that was received from a peer in response to an rpc request.

type DataColumnSidecarsParams 

type DataColumnSidecarsParams struct {
	Ctx                     context.Context                     // Context
	Tor                     blockchain.TemporalOracle           // Temporal oracle, useful to get the current slot
	P2P                     prysmP2P.P2P                        // P2P network interface
	RateLimiter             *leakybucket.Collector              // Rate limiter for outgoing requests
	CtxMap                  ContextByteVersions                 // Context map, useful to know if a message is mapped to the correct fork
	Storage                 filesystem.DataColumnStorageReader  // Data columns storage
	NewVerifier             verification.NewDataColumnsVerifier // Data columns verifier to check to conformity of incoming data column sidecars
	DownscorePeerOnRPCFault bool                                // Downscore a peer if it commits an RPC fault. Not responding sidecars at all is considered as a fault.
}

DataColumnSidecarsParams stores the common parameters needed to fetch data column sidecars from peers.

type Option 

type Option func(s *Service) error

func WithAttestationCache 

func WithAttestationCache(c *cache.AttestationCache) Option

func WithAttestationNotifier 

func WithAttestationNotifier(notifier operation.Notifier) Option

func WithAttestationPool 

func WithAttestationPool(attPool attestations.Pool) Option

func WithAvailableBlocker 

func WithAvailableBlocker(avb coverage.AvailableBlocker) Option

WithAvailableBlocker allows the sync package to access the current status of backfill.

func WithBatchVerifierLimit 

func WithBatchVerifierLimit(limit int) Option

WithBatchVerifierLimit sets the maximum number of signatures to batch verify at once.

func WithBlobStorage 

func WithBlobStorage(b *filesystem.BlobStorage) Option

WithBlobStorage gives the sync package direct access to BlobStorage.

func WithBlockNotifier 

func WithBlockNotifier(blockNotifier blockfeed.Notifier) Option

func WithBlsToExecPool 

func WithBlsToExecPool(blsToExecPool blstoexec.PoolManager) Option

func WithChainService 

func WithChainService(chain blockchainService) Option

func WithClockWaiter 

func WithClockWaiter(cw startup.ClockWaiter) Option

func WithDataColumnStorage 

func WithDataColumnStorage(b *filesystem.DataColumnStorage) Option

WithDataColumnStorage gives the sync package direct access to DataColumnStorage.

func WithDatabase 

func WithDatabase(db db.NoHeadAccessDatabase) Option

func WithExitPool 

func WithExitPool(exitPool voluntaryexits.PoolManager) Option

func WithInitialSync 

func WithInitialSync(initialSync Checker) Option

func WithInitialSyncComplete 

func WithInitialSyncComplete(c chan struct{}) Option

func WithLightClientStore 

func WithLightClientStore(lcs *lightClient.Store) Option

WithLightClientStore allows the sync package to access light client data.

func WithOperationNotifier 

func WithOperationNotifier(operationNotifier operation.Notifier) Option

func WithP2P 

func WithP2P(p2p p2p.P2P) Option

func WithPayloadAttestationCache added in v7.1.4 

func WithPayloadAttestationCache(c *cache.PayloadAttestationCache) Option

func WithPayloadAttestationPool added in v7.1.4 

func WithPayloadAttestationPool(pool payloadattestation.PoolManager) Option

func WithProposerPreferencesCache added in v7.1.4 

func WithProposerPreferencesCache(c *cache.ProposerPreferencesCache) Option

func WithReconstructionRandGen 

func WithReconstructionRandGen(rg *rand.Rand) Option

WithReconstructionRandGen sets the random generator for reconstruction delays.

func WithReconstructor 

func WithReconstructor(r execution.Reconstructor) Option

func WithSlasherAttestationsFeed 

func WithSlasherAttestationsFeed(slasherAttestationsFeed *event.Feed) Option

func WithSlasherBlockHeadersFeed 

func WithSlasherBlockHeadersFeed(slasherBlockHeadersFeed *event.Feed) Option

func WithSlasherEnabled 

func WithSlasherEnabled(enabled bool) Option

WithSlasherEnabled configures the sync package to support slashing detection.

func WithSlashingPool 

func WithSlashingPool(slashingPool slashings.PoolManager) Option

func WithStateGen 

func WithStateGen(stateGen *stategen.State) Option

func WithStateNotifier 

func WithStateNotifier(n statefeed.Notifier) Option

WithStateNotifier to notify an event feed of state processing.

func WithSubscribedValidatorsCache added in v7.1.6 

func WithSubscribedValidatorsCache(c *cache.SubscribedValidatorsCache) Option

func WithSyncCommsPool 

func WithSyncCommsPool(syncCommsPool synccommittee.Pool) Option

func WithVerifierWaiter 

func WithVerifierWaiter(v *verification.InitializerWaiter) Option

WithVerifierWaiter gives the sync package direct access to the verifier waiter.

type PeerPicker added in v7.1.0 

type PeerPicker struct {
	// contains filtered or unexported fields
}

PeerPicker is a structure that maps out the intersection of peer custody and column indices to weight each peer based on the scarcity of the columns they custody. This allows us to prioritize requests for more scarce columns to peers that custody them, so that we don't waste our bandwidth allocation making requests for more common columns from peers that can provide the more scarce columns.

func (*PeerPicker) ForBlocks added in v7.1.0 

func (m *PeerPicker) ForBlocks(busy map[peer.ID]bool) (peer.ID, error)

ForBlocks returns the lowest scoring peer in the set. This can be used to pick a peer for block requests, preserving the peers that have the highest coverage scores for column requests.

func (*PeerPicker) ForColumns added in v7.1.0 

func (m *PeerPicker) ForColumns(needed peerdas.ColumnIndices, busy map[peer.ID]bool) (peer.ID, []uint64, error)

ForColumns returns the best peer to request columns from, based on the scarcity of the columns needed.

type Service 

type Service struct {
	// contains filtered or unexported fields
}

Service is responsible for handling all run time p2p related operations as the main entry point for network messages.

func NewService 

func NewService(ctx context.Context, opts ...Option) *Service

NewService initializes new regular sync service.

func (*Service) HighestExecutionPayloadBidCache added in v7.1.5 

func (s *Service) HighestExecutionPayloadBidCache() *cache.HighestExecutionPayloadBidCache

HighestExecutionPayloadBidCache exposes sync's cache to the proposer RPC. Sync is the sole writer (gossip); the proposer is a reader.

func (*Service) Start 

func (s *Service) Start()

Start the regular sync service.

func (*Service) Status 

func (s *Service) Status() error

Status of the currently running regular sync service.

func (*Service) Stop 

func (s *Service) Stop() error

Stop the regular sync service.

Source Files

View all Source files

Directories

Path Synopsis
Code generated by hack/gen-logs.sh; DO NOT EDIT.
 Code generated by hack/gen-logs.sh; DO NOT EDIT.
coverage
Code generated by hack/gen-logs.sh; DO NOT EDIT.
 Code generated by hack/gen-logs.sh; DO NOT EDIT.
Code generated by hack/gen-logs.sh; DO NOT EDIT.
 Code generated by hack/gen-logs.sh; DO NOT EDIT.
testing
Package testing includes useful mocks for testing initial sync status in unit tests.
 Package testing includes useful mocks for testing initial sync status in unit tests.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.