collections

type CollectionIndexer ¶

type CollectionIndexer interface {
	// OnCollectionReceived notifies the collection indexer that a new collection is available to be indexed.
	// Calling this method multiple times with the same collection is a no-op.
	// This method is non-blocking.
	OnCollectionReceived(collection *flow.Collection)

	// IndexCollections indexes a set of collections, skipping any collections which already exist in storage.
	// Calling this method multiple times with the same collections is a no-op.
	//
	// No error returns are expected during normal operation.
	IndexCollections(collections []*flow.Collection) error

	// MissingCollectionsAtHeight returns all collections that are not present in storage for a specific
	// finalized block height.
	//
	// Expected error returns during normal operation:
	//   - [storage.ErrNotFound]: if provided block height is not finalized or below this node's root block
	MissingCollectionsAtHeight(height uint64) ([]*flow.CollectionGuarantee, error)
}

type ExecutionDataSyncer ¶

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

ExecutionDataSyncer submits collections from execution data to the collections indexer. It is designed to be used within the collection syncer to optimize indexing when collection data is already available on the node.

func NewExecutionDataSyncer(
	log zerolog.Logger,
	executionDataCache execution_data.ExecutionDataCache,
	indexer CollectionIndexer,
) *ExecutionDataSyncer

func (s *ExecutionDataSyncer) IndexForHeight(ctx context.Context, height uint64) (bool, error)

type Indexer ¶

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

Indexer stores and indexes collections received from the network. It is designed to be the central point for accumulating collections from various subsystems that my receive them from the network. For example, collections may be received from execution data sync, the collection syncer, or the execution state indexer. Depending on the node's configuration, one or more of these subsystems will feed the indexer with collections.

The indexer also maintains the last full block height state, which is the highest block height for which all collections are stored and indexed.

func NewIndexer(
	log zerolog.Logger,
	db storage.DB,
	metrics module.CollectionExecutedMetric,
	state protocol.State,
	blocks storage.Blocks,
	collections storage.Collections,
	lastFullBlockHeight *counters.PersistentStrictMonotonicCounter,
	lockManager lockctx.Manager,
) (*Indexer, error)

func (ci *Indexer) IndexCollections(collections []*flow.Collection) error

func (ci *Indexer) MissingCollectionsAtHeight(height uint64) ([]*flow.CollectionGuarantee, error)

func (ci *Indexer) OnCollectionReceived(collection *flow.Collection)

func (ci *Indexer) WorkerLoop(ctx irrecoverable.SignalerContext, ready component.ReadyFunc)

type Syncer ¶

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

The Syncer is responsible for syncing collections for finalized blocks from the network. It has three main responsibilities:

1. Handle requests for collections for finalized blocks from the ingestion engine by submitting the requests to Collection nodes.
2. Track blocks with missing collections, and periodically re-request the missing collections from the network.
3. Submit collections received to the Indexer for storage and indexing.

The Syncer guarantees that all collections for finalized blocks will eventually be received as long as there are honest Collection nodes in each cluster, and the node is able to successfully communicate with them over the networking layer.

It is responsible for ensuring the local node has all collections contained within finalized blocks starting from the last fully synced height. It works by periodically scanning the finalized block range from the last full block height up to the latest finalized block height, identifying missing collections, and triggering requests to fetch them from the network. Once collections are retrieved, it submits them to the Indexer for storage and indexing.

It is meant to operate in a background goroutine as part of the node's ingestion pipeline.

func NewSyncer(
	log zerolog.Logger,
	requester module.Requester,
	state protocol.State,
	collections storage.Collections,
	lastFullBlockHeight counters.Reader,
	collectionIndexer CollectionIndexer,
	execDataSyncer *ExecutionDataSyncer,
) *Syncer

func (s *Syncer) OnCollectionDownloaded(_ flow.Identifier, entity flow.Entity)

func (s *Syncer) RequestCollectionsForBlock(height uint64, missingCollections []*flow.CollectionGuarantee) error

func (s *Syncer) WorkerLoop(ctx irrecoverable.SignalerContext, ready component.ReadyFunc)