blocksync

package
v1.7.0 Latest Latest
Warning

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

 Go to latest Published: Aug 17, 2026 License: Apache-2.0 Imports: 26 Imported by: 0

Documentation

Overview

Package blocksync implements two versions of a reactor Service that are responsible for block propagation and gossip between peers. This mechanism was formerly known as fast-sync.

In order for a full node to successfully participate in consensus, it must have the latest view of state. The blocksync protocol is a mechanism in which peers may exchange and gossip entire blocks with one another, in a request/response type model, until they've successfully synced to the latest head block. Once succussfully synced, the full node can switch to an active role in consensus and will no longer blocksync and thus no longer run the blocksync process.

Note, the blocksync reactor Service gossips entire block and relevant data such that each receiving peer may construct the entire view of the blocksync state.

There is currently only one version of the blocksync reactor Service that is battle-tested, but whose test coverage is lacking and is not formally verified.

The v0 blocksync reactor Service has one p2p channel, BlockchainChannel. This channel is responsible for handling messages that both request blocks and respond to block requests from peers. For every block request from a peer, the reactor will execute respondToPeer which will fetch the block from the node's state store and respond to the peer. For every block response, the node will add the block to its synchronizer.

Internally, v0 runs a poolRoutine that constantly checks for what blocks it needs and requests them. The poolRoutine is also responsible for taking blocks from the synchronizer, saving and executing each block.

Index

Constants

View Source 
const (
	MaxMsgSize = types.MaxBlockSizeBytes +
		p2pproto.BlockResponseMessagePrefixSize +
		p2pproto.BlockResponseMessageFieldKeySize
)

Variables

This section is empty.

Functions

func AddNumPending 

func AddNumPending(val int32) store.UpdateFunc[types.NodeID, PeerData]

AddNumPending adds a value to the numPending field

func ResetFailures added in v1.7.0 

func ResetFailures() store.UpdateFunc[types.NodeID, PeerData]

ResetFailures clears the count of consecutive failed requests, so that the threshold only ever trips on an unbroken run of failures

func ResetMonitor 

func ResetMonitor() store.UpdateFunc[types.NodeID, PeerData]

ResetMonitor replaces a peer monitor on a new one if numPending is zero

func UpdateMonitor 

func UpdateMonitor(recvSize int) store.UpdateFunc[types.NodeID, PeerData]

UpdateMonitor adds a block size value to the peer monitor if numPending is greater than zero

Types

type BlockResponse 

type BlockResponse struct {
	PeerID types.NodeID
	Block  *types.Block
	Commit *types.Commit
	// Size is the serialized size of Block in bytes, measured while decoding the
	// response. Deriving it from Block again means re-serializing the whole
	// block, which is too expensive to do on the block apply path.
	Size int
}

BlockResponse ...

func BlockResponseFromProto 

func BlockResponseFromProto(resp *bcproto.BlockResponse, peerID types.NodeID) (*BlockResponse, error)

func (*BlockResponse) Validate 

func (r *BlockResponse) Validate() error

type InMemPeerStore 

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

InMemPeerStore in-memory peer store

func NewInMemPeerStore 

func NewInMemPeerStore(peers ...PeerData) *InMemPeerStore

NewInMemPeerStore creates a new in-memory peer store

func (*InMemPeerStore) AddFailure added in v1.7.0 

func (p *InMemPeerStore) AddFailure(peerID types.NodeID, maxFailures int32) bool

AddFailure records a block request that this peer failed to answer: the request stops being pending, and the peer's consecutive failure count grows.

It reports true once the peer has failed maxFailures requests in a row and should be dropped. An unknown peer reports false, so the failures of a peer that was already removed do not report it again.

func (*InMemPeerStore) All 

func (p *InMemPeerStore) All() []PeerData

All returns all stored peers in the store

func (*InMemPeerStore) Delete 

func (p *InMemPeerStore) Delete(peerID types.NodeID)

Delete deletes the peer data from the store

func (*InMemPeerStore) FindPeer 

func (p *InMemPeerStore) FindPeer(height int64) (PeerData, bool)

FindPeer finds a peer for the request criteria by which the peer is looked up: 1. the number of pending requests must be less allowed (maxPendingRequestsPerPeer) 2. the height must be between two values base and height otherwise return the empty peer data and false

func (*InMemPeerStore) FindTimedoutPeers 

func (p *InMemPeerStore) FindTimedoutPeers() []PeerData

FindTimedoutPeers finds and returns the timed out peers

func (*InMemPeerStore) Get 

func (p *InMemPeerStore) Get(peerID types.NodeID) (PeerData, bool)

Get returns peer's data and true if the peer is found otherwise empty structure and false

func (*InMemPeerStore) GetAndDelete 

func (p *InMemPeerStore) GetAndDelete(peerID types.NodeID) (PeerData, bool)

GetAndDelete combines Get operation and Delete in one call

func (*InMemPeerStore) HasPeerForHeight added in v1.7.0 

func (p *InMemPeerStore) HasPeerForHeight(height int64) bool

HasPeerForHeight reports whether height is fetchable at all: some peer both advertises a range covering it and is one FindPeer would still select.

It takes two of FindPeer's three criteria, and the line between them is transient versus permanent unusability rather than holding versus serving.

The receive-rate check is in. A peer whose rate has fallen below minRecvRate is not merely slow, it is out of the game for good: FindPeer will not select it, so nextJob never resets its monitor, so the rate never recovers; and FindTimedoutPeers only removes peers with requests outstanding, so once its last request completes nothing evicts it either. Counting such a peer as fetchable is the same lock-out this predicate exists to prevent, reached by a merely slow peer instead of a lying one.

The pending-request limit is out. A peer at its cap is busy for as long as its in-flight requests take and then selectable again, so excluding it would make the verdict flap under load and end block sync exactly when the request pipeline is fullest.

func (*InMemPeerStore) IsZero 

func (p *InMemPeerStore) IsZero() bool

IsZero returns true if the store doesn't have a peer yet otherwise false

func (*InMemPeerStore) Len 

func (p *InMemPeerStore) Len() int

Len returns the count of all stored peers

func (*InMemPeerStore) MaxHeight 

func (p *InMemPeerStore) MaxHeight() int64

MaxHeight looks at all the peers in the store to get the maximum peer height.

It is derived on read rather than cached. A cached maximum has to be maintained by every path that adds, updates or removes a peer, each holding its own lock, so two of them interleaving can publish a height that belonged to a peer already gone or already lowered. This value decides whether the node considers itself caught up and whether a stalled sync gives up, so a height no peer can serve does not merely misreport - it keeps the node in block sync waiting for a block that will never arrive. The scan is O(peers), the same order as FindPeer and FindTimedoutPeers, which the producing loop already runs for every job.

func (*InMemPeerStore) Put 

func (p *InMemPeerStore) Put(peerID types.NodeID, newPeer PeerData)

Put adds the peer data to the store if the peer does not exist, otherwise update the current value

func (*InMemPeerStore) Query 

func (p *InMemPeerStore) Query(spec store.QueryFunc[types.NodeID, PeerData], limit int) []PeerData

Query finds and returns the copy of peers by specification conditions

func (*InMemPeerStore) Update 

func (p *InMemPeerStore) Update(peerID types.NodeID, updates ...store.UpdateFunc[types.NodeID, PeerData])

Update applies update functions to the peer if it exists

func (*InMemPeerStore) Upsert added in v1.7.0 

func (p *InMemPeerStore) Upsert(newPeer PeerData)

Upsert records the range of blocks a peer advertises, adding the peer if it is not known yet.

The advertised base and height are the only things a peer tells us about itself; everything else held in PeerData - the count of requests we have issued and not yet accounted for, the run of consecutive failures, the receive-rate monitor and the time we first saw the peer - describes our own outstanding requests and the peer's service quality. Peers re-advertise their range every few seconds, far more often than a block request times out, so replacing those counters here would reset them faster than any threshold built on them can be reached and would leave requests in flight that nothing accounts for.

Insert and merge are a single store operation. Peers are made known by the p2p consumer goroutine while the job producer is already issuing requests against them, so a lookup followed by a separate write would let a peer inserted in between be overwritten - losing precisely the state this exists to keep.

type OptionFunc 

type OptionFunc func(v *Synchronizer)

func WithClock 

func WithClock(clock clockwork.Clock) OptionFunc

func WithLogger 

func WithLogger(logger log.Logger) OptionFunc

func WithMonitorInterval added in v1.5.0 

func WithMonitorInterval(blocks int64) OptionFunc

func WithWorkerPool 

func WithWorkerPool(wp *workerpool.WorkerPool) OptionFunc

type PeerAdder 

type PeerAdder interface {
	AddPeer(peer PeerData)
}

type PeerData 

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

PeerData uses to keep peer related data like base height and the current height etc

type PeerRemover 

type PeerRemover interface {
	RemovePeer(peerID types.NodeID)
}

type Reactor 

type Reactor struct {
	service.BaseService
	// contains filtered or unexported fields
}

Reactor handles long-term catchup syncing.

func NewReactor 

func NewReactor(
	logger log.Logger,
	stateStore sm.Store,
	blockExec *sm.BlockExecutor,
	store *store.BlockStore,
	nodeProTxHash crypto.ProTxHash,
	consReactor consensusReactor,
	p2pClient *client.Client,
	peerEvents p2p.PeerEventSubscriber,
	blockSync bool,
	metrics *consensus.Metrics,
	eventBus *eventbus.EventBus,
	opts ...ReactorOption,
) *Reactor

NewReactor returns new reactor instance.

func (*Reactor) GetMaxPeerBlockHeight 

func (r *Reactor) GetMaxPeerBlockHeight() int64

func (*Reactor) GetRemainingSyncTime 

func (r *Reactor) GetRemainingSyncTime() time.Duration

func (*Reactor) GetTotalSyncedTime 

func (r *Reactor) GetTotalSyncedTime() time.Duration

func (*Reactor) OnStart 

func (r *Reactor) OnStart(ctx context.Context) error

OnStart starts separate go routines for each p2p Channel and listens for envelopes on each. In addition, it also listens for peer updates and handles messages on that p2p channel accordingly. The caller must be sure to execute OnStop to ensure the outbound p2p Channels are closed.

If blockSyncFlag is enabled, we also start the synchronizer If the synchronizer fails to start, an error is returned.

func (*Reactor) OnStop 

func (r *Reactor) OnStop()

OnStop stops the reactor by signaling to all spawned goroutines to exit and blocking until they all exit.

func (*Reactor) PublishStatus 

func (r *Reactor) PublishStatus(event types.EventDataBlockSyncStatus) error

func (*Reactor) SwitchToBlockSync 

func (r *Reactor) SwitchToBlockSync(ctx context.Context, state sm.State) error

SwitchToBlockSync is called by the state sync reactor when switching to fast sync.

type ReactorOption added in v1.5.2 

type ReactorOption func(*Reactor)

func WithStatusUpdateInterval added in v1.5.2 

func WithStatusUpdateInterval(interval time.Duration) ReactorOption

WithStatusUpdateInterval overrides the interval used to poll peers for their status updates.

func WithSynchronizerOptions added in v1.6.0 

func WithSynchronizerOptions(opts ...OptionFunc) ReactorOption

WithSynchronizerOptions forwards options to the synchronizer built in OnStart.

type Synchronizer 

type Synchronizer struct {
	service.BaseService
	// contains filtered or unexported fields
}

Synchronizer keeps track of the block sync peers, block requests and block responses.

func NewSynchronizer 

func NewSynchronizer(start int64, client client.BlockClient, blockExec *blockApplier, opts ...OptionFunc) *Synchronizer

NewSynchronizer returns a new Synchronizer with the height equal to start

func (*Synchronizer) AddPeer 

func (s *Synchronizer) AddPeer(peer PeerData)

AddPeer records the peer's alleged blockchain base and height. Peers report their range repeatedly, so for a peer we already track this only moves that range and leaves everything we know about its outstanding requests intact.

func (*Synchronizer) GetStatus 

func (s *Synchronizer) GetStatus() (int64, int32)

GetStatus returns synchronizer's height, count of in progress requests

func (*Synchronizer) IsCaughtUp 

func (s *Synchronizer) IsCaughtUp() bool

IsCaughtUp returns true if this node is caught up, false - otherwise.

func (*Synchronizer) LastAdvance 

func (s *Synchronizer) LastAdvance() time.Time

LastAdvance returns the time when the last block was processed (or start time if no blocks were processed).

func (*Synchronizer) MaxPeerHeight 

func (s *Synchronizer) MaxPeerHeight() int64

MaxPeerHeight returns the highest reported height.

func (*Synchronizer) OnStart 

func (s *Synchronizer) OnStart(ctx context.Context) error

OnStart implements service.Service by spawning requesters routine and recording synchronizer's start time.

func (*Synchronizer) OnStop 

func (s *Synchronizer) OnStop()

func (*Synchronizer) RemovePeer 

func (s *Synchronizer) RemovePeer(peerID types.NodeID)

RemovePeer removes the peer with peerID from the synchronizer. If there's no peer with peerID, function is a no-op.

func (*Synchronizer) WaitForSync 

func (s *Synchronizer) WaitForSync(ctx context.Context) (caughtUp bool)

WaitForSync blocks until block sync is finished, and reports whether the node actually caught up.

A stall on its own is not a reason to stop. Handing over to consensus is a one-way door - only the state sync path ever switches back to block sync - and consensus catch-up is far slower than block sync, so a node that gives up while it is still thousands of blocks behind stays behind. As long as some peer holds the block we are waiting for there is something to retry, so keep going and say so loudly. Give up on the stall only once no peer can serve that block, or once the stall has outlasted maxSyncStall, so that a wedged synchronizer can still hand over rather than blocking forever.

Source Files

View all Source files

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.