indexer

type CollectionExecutedMetricImpl ¶ added in v0.33.15

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

CollectionExecutedMetricImpl tracks metrics to measure how long it takes for tx to reach each step in their lifecycle

func NewCollectionExecutedMetricImpl(
	log zerolog.Logger,
	accessMetrics module.AccessMetrics,
	collectionsToMarkFinalized *stdmap.Times,
	collectionsToMarkExecuted *stdmap.Times,
	blocksToMarkExecuted *stdmap.Times,
	collections storage.Collections,
	blocks storage.Blocks,
	blockTransactions *stdmap.IdentifierMap,
) (*CollectionExecutedMetricImpl, error)

func (c *CollectionExecutedMetricImpl) BlockFinalized(block *flow.Block)

func (c *CollectionExecutedMetricImpl) CollectionExecuted(light *flow.LightCollection)

func (c *CollectionExecutedMetricImpl) CollectionFinalized(light *flow.LightCollection)

func (c *CollectionExecutedMetricImpl) ExecutionReceiptReceived(r *flow.ExecutionReceipt)

func (c *CollectionExecutedMetricImpl) UpdateLastFullBlockHeight(height uint64)

type InMemoryIndexer ¶ added in v0.43.0

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

<component_spec> InMemoryIndexer indexes block execution data for a single ExecutionResult into a mempool. It is designed to be used as part of the optimistic syncing processing pipeline, to index data for unsealed execution results which is eventually persisted when the execution result is sealed.

The data contained within the BlockExecutionData is verified by verifications nodes as part of the approval process. Once the execution result is sealed, the Access node can accept it as valid without further verification. However, with optimistic syncing, the Access node may index data for execution results that are not sealed. Since this data is not certified by the protocol, it must not be persisted to disk. It may be used by the Access node to serve Access API requests, with the understanding that it may be later determined to be invalid.

The provided BlockExecutionData is received over the network and its hash is compared to the value included in an ExecutionResult within a certified block. This guarantees it is the same data that was produced by an execution node whose stake is at risk if the data is incorrect. It is not practical for an Access node to verify all data, but the indexer may perform opportunistic checks to ensure the data is generally consistent.

Transaction error messages are received directly from execution nodes with no protocol guarantees. The node must validate that there is a one-to-one mapping between failed transactions and transaction error messages. Since the error messages are requested directly from the execution nodes, it's possible that they are delayed. To avoid blocking the indexing process if ENs are unresponsive, the processing pipeline may skip the call to `ValidateTxErrors()` if the error messages are not ready. In this case, the the error messages may be validated and backfilled later. </component_spec>

Safe for concurrent use.

func NewInMemoryIndexer(
	log zerolog.Logger,
	block *flow.Block,
	executionResult *flow.ExecutionResult,
) (*InMemoryIndexer, error)

func (i *InMemoryIndexer) IndexBlockData(data *execution_data.BlockExecutionData) (*IndexerData, error)

type Indexer ¶

type Indexer struct {
	component.Component
	execution_data.ProcessedHeightRecorder
	// contains filtered or unexported fields
}

Indexer handles ingestion of new execution data available and uses the execution data indexer module to index the data. The processing of new available data is done by creating a jobqueue that uses the execution data reader to obtain new jobs. The worker also implements the `highestConsecutiveHeight` method which is used by the execution data reader, so it doesn't surpass the highest sealed block height when fetching the data. The execution state worker has a callback that is used by the upstream queues which download new execution data to notify new data is available and kick off indexing.

func NewIndexer(
	log zerolog.Logger,
	initHeight uint64,
	registers storage.RegisterIndex,
	indexer *IndexerCore,
	executionCache *cache.ExecutionDataCache,
	executionDataLatestHeight func() (uint64, error),
	processedHeightInitializer storage.ConsumerProgressInitializer,
) (*Indexer, error)

func (i *Indexer) HighestIndexedHeight() (uint64, error)

func (i *Indexer) LowestIndexedHeight() (uint64, error)

func (i *Indexer) OnExecutionData(_ *execution_data.BlockExecutionDataEntity)

func (i *Indexer) Start(ctx irrecoverable.SignalerContext)

type IndexerCore ¶

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

IndexerCore indexes the execution state.

func New(
	log zerolog.Logger,
	metrics module.ExecutionStateIndexerMetrics,
	protocolDB storage.DB,
	registers storage.RegisterIndex,
	headers storage.Headers,
	events storage.Events,
	collections storage.Collections,
	transactions storage.Transactions,
	results storage.LightTransactionResults,
	scheduledTransactions storage.ScheduledTransactions,
	chainID flow.ChainID,
	derivedChainData *derived.DerivedChainData,
	collectionExecutedMetric module.CollectionExecutedMetric,
	lockManager lockctx.Manager,
) *IndexerCore

func (c *IndexerCore) IndexBlockData(data *execution_data.BlockExecutionDataEntity) error

func (c *IndexerCore) RegisterValue(ID flow.RegisterID, height uint64) (flow.RegisterValue, error)

type IndexerData ¶

type IndexerData struct {
	Events       []flow.Event
	Collections  []*flow.Collection
	Transactions []*flow.TransactionBody
	Results      []flow.LightTransactionResult
	Registers    []flow.RegisterEntry
}

IndexerData is the collection of data ingested by the indexer.