[Access] POC Collection Syncing

cmd/observer/node_builder/observer_builder.go

@@ -38,7 +38,6 @@ import (
 	"github.com/onflow/flow-go/engine"
 	"github.com/onflow/flow-go/engine/access/apiproxy"
 	"github.com/onflow/flow-go/engine/access/index"
-	"github.com/onflow/flow-go/engine/access/ingestion/collections"
 	"github.com/onflow/flow-go/engine/access/rest"
 	restapiproxy "github.com/onflow/flow-go/engine/access/rest/apiproxy"
 	"github.com/onflow/flow-go/engine/access/rest/router"
@@ -70,7 +69,6 @@ import (
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/blobs"
 	"github.com/onflow/flow-go/module/chainsync"
-	"github.com/onflow/flow-go/module/counters"
 	"github.com/onflow/flow-go/module/execution"
 	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
 	execdatacache "github.com/onflow/flow-go/module/executiondatasync/execution_data/cache"
@@ -1438,31 +1436,7 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				return nil, fmt.Errorf("could not create derived chain data: %w", err)
 			}
 
-			rootBlockHeight := node.State.Params().FinalizedRoot().Height
-			progress, err := store.NewConsumerProgress(builder.ProtocolDB, module.ConsumeProgressLastFullBlockHeight).Initialize(rootBlockHeight)
-			if err != nil {
-				return nil, fmt.Errorf("could not create last full block height consumer progress: %w", err)
-			}
-
-			lastFullBlockHeight, err := counters.NewPersistentStrictMonotonicCounter(progress)
-			if err != nil {
-				return nil, fmt.Errorf("could not create last full block height counter: %w", err)
-			}
-
 			var collectionExecutedMetric module.CollectionExecutedMetric = metrics.NewNoopCollector()
-			collectionIndexer, err := collections.NewIndexer(
-				builder.Logger,
-				builder.ProtocolDB,
-				collectionExecutedMetric,
-				builder.State,
-				builder.Storage.Blocks,
-				builder.Storage.Collections,
-				lastFullBlockHeight,
-				builder.StorageLockMgr,
-			)
-			if err != nil {
-				return nil, fmt.Errorf("could not create collection indexer: %w", err)
-			}
 
 			builder.ExecutionIndexerCore = indexer.New(
 				builder.Logger,
@@ -1477,7 +1451,6 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				builder.scheduledTransactions,
 				builder.RootChainID,
 				indexerDerivedChainData,
-				collectionIndexer,
 				collectionExecutedMetric,
 				node.StorageLockMgr,
 			)
@@ -1556,10 +1529,6 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 			}
 			builder.stateStreamConf.RpcMetricsEnabled = builder.rpcMetricsEnabled
 
-			highestAvailableHeight, err := builder.ExecutionDataRequester.HighestConsecutiveHeight()
-			if err != nil {
-				return nil, fmt.Errorf("could not get highest consecutive height: %w", err)
-			}
 			broadcaster := engine.NewBroadcaster()
 
 			eventQueryMode, err := query_mode.ParseIndexQueryMode(builder.rpcConf.BackendConfig.EventQueryMode)
@@ -1578,7 +1547,7 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				builder.executionDataConfig.InitialBlockHeight,
 				node.Storage.Headers,
 				broadcaster,
-				highestAvailableHeight,
+				builder.ExecutionDataRequester,
 				builder.EventsIndex,
 				useIndex,
 			)

consensus/hotstuff/distributor.go

@@ -0,0 +1,8 @@
+package hotstuff
+
+import "github.com/onflow/flow-go/consensus/hotstuff/model"
+
+type Distributor interface {
+	AddOnBlockFinalizedConsumer(consumer func(block *model.Block))
+	AddOnBlockIncorporatedConsumer(consumer func(block *model.Block))
+}

engine/access/collection_sync/collection_sync.go

@@ -0,0 +1,76 @@
+package collection_sync
+
+import (
+	"context"
+
+	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module/component"
+	"github.com/onflow/flow-go/module/irrecoverable"
+)
+
+// Tracks missing collections per height and invokes job callbacks when complete.
+type MissingCollectionQueue interface {
+	EnqueueMissingCollections(blockHeight uint64, ids []flow.Identifier, callback func()) error
+	OnIndexedForBlock(blockHeight uint64) (func(), bool)
+
+	// On receipt of a collection, MCQ updates internal state and, if a block
+	// just became complete, returns: (collections, height, true).
+	// Otherwise, returns (nil, 0, false).
+	OnReceivedCollection(collection *flow.Collection) ([]*flow.Collection, uint64, bool)
+
+	// IsHeightQueued returns true if the given height is still being tracked (has not been indexed yet).
+	IsHeightQueued(height uint64) bool
+
+	// Size returns the number of missing collections currently in the queue.
+	Size() uint
+}
+
+// Requests collections by their IDs.
+type CollectionRequester interface {
+	RequestCollections(ids []flow.Identifier) error
+}
+
+// BlockCollectionIndexer stores and indexes collections for a given block height.
+type BlockCollectionIndexer interface {
+	// IndexCollectionsForBlock stores and indexes collections for a given block height.
+	// No error is exepcted during normal operation.
+	IndexCollectionsForBlock(blockHeight uint64, cols []*flow.Collection) error
+
+	// GetMissingCollections retrieves the block and returns collection guarantees that whose collections
+	// are missing in storage.
+	// Only garantees whose collections that are not already in storage are returned.
+	GetMissingCollections(block *flow.Block) ([]*flow.CollectionGuarantee, error)
+}
+
+// Implements the job lifecycle for a single block height.
+type BlockProcessor interface {
+	FetchCollections(ctx irrecoverable.SignalerContext, block *flow.Block, done func()) error
+	// MissingCollectionQueueSize returns the number of missing collections currently in the queue.
+	MissingCollectionQueueSize() uint
+}
+
+// Fetcher is a component that consumes finalized block jobs and processes them
+// to index collections. It uses a job consumer with windowed throttling to prevent node overload.
+type Fetcher interface {
+	component.Component
+	ProgressReader
+	OnFinalizedBlock()
+	Size() uint
+}
+
+// ExecutionDataProvider provides the latest height for which execution data indexer has collections.
+// This can be nil if execution data indexing is disabled.
+type ExecutionDataProvider interface {
+	HighestIndexedHeight() uint64
+	GetExecutionDataByHeight(ctx context.Context, height uint64) ([]*flow.Collection, error)
+}
+
+// ExecutionDataProcessor processes execution data when new execution data is available.
+type ExecutionDataProcessor interface {
+	OnNewExectuionData()
+}
+
+// ProgressReader provides the current progress of collection fetching/indexing.
+type ProgressReader interface {
+	ProcessedHeight() uint64
+}

engine/access/collection_sync/execution_data_index/processor.go

@@ -0,0 +1,130 @@
+package execution_data_index
+
+import (
+	"fmt"
+
+	"github.com/rs/zerolog"
+
+	"github.com/onflow/flow-go/engine/access/collection_sync"
+	"github.com/onflow/flow-go/module"
+	"github.com/onflow/flow-go/module/component"
+	"github.com/onflow/flow-go/module/counters"
+	"github.com/onflow/flow-go/module/irrecoverable"
+)
+
+type ExecutionDataProcessor struct {
+	component.Component
+	log                     zerolog.Logger
+	newExecutionDataIndexed chan struct{}
+	provider                collection_sync.ExecutionDataProvider
+	indexer                 collection_sync.BlockCollectionIndexer
+	metrics                 module.CollectionSyncMetrics
+	// state
+	processedHeight *counters.PersistentStrictMonotonicCounter
+}
+
+var _ collection_sync.ExecutionDataProcessor = (*ExecutionDataProcessor)(nil)
+var _ collection_sync.ProgressReader = (*ExecutionDataProcessor)(nil)
+var _ component.Component = (*ExecutionDataProcessor)(nil)
+
+func NewExecutionDataProcessor(
+	log zerolog.Logger,
+	provider collection_sync.ExecutionDataProvider,
+	indexer collection_sync.BlockCollectionIndexer,
+	processedHeight *counters.PersistentStrictMonotonicCounter,
+	metrics module.CollectionSyncMetrics,
+) *ExecutionDataProcessor {
+	edp := &ExecutionDataProcessor{
+		log:                     log.With().Str("component", "coll_sync_ed_processor").Logger(),
+		newExecutionDataIndexed: make(chan struct{}, 1),
+		provider:                provider,
+		indexer:                 indexer,
+		metrics:                 metrics,
+		processedHeight:         processedHeight,
+	}
+
+	// Initialize the channel so that even if no new execution data comes in,
+	// the worker loop can still be triggered to process any existing data.
+	edp.newExecutionDataIndexed <- struct{}{}
+
+	// Build component manager with worker loop
+	cm := component.NewComponentManagerBuilder().
+		AddWorker(edp.workerLoop).
+		Build()
+
+	edp.Component = cm
+
+	return edp
+}
+
+func (edp *ExecutionDataProcessor) OnNewExectuionData() {
+	select {
+	case edp.newExecutionDataIndexed <- struct{}{}:
+	default:
+		// if the channel is full, no need to block, just return.
+		// once the worker loop processes the buffered signal, it will
+		// process the next height all the way to the highest available height.
+	}
+}
+
+func (edp *ExecutionDataProcessor) workerLoop(ctx irrecoverable.SignalerContext, ready component.ReadyFunc) {
+	ready()
+
+	// using a single threaded loop to index each execution for height
+	// since indexing collections is blocking anyway, and reading the execution data
+	// is quick, because we cache for 100 heights.
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-edp.newExecutionDataIndexed:
+			highestAvailableHeight := edp.provider.HighestIndexedHeight()
+			lowestMissing := edp.processedHeight.Value() + 1
+
+			for height := lowestMissing; height <= highestAvailableHeight; height++ {
+				collections, err := edp.provider.GetExecutionDataByHeight(ctx, height)
+				if err != nil {
+					ctx.Throw(fmt.Errorf("failed to get execution data for height %d: %w", height, err))
+					return
+				}
+
+				// TODO: since both collections and execution data processor are the data source of
+				// collections, before indexing the collections, double check if it was indexed
+				// by the collections already by simply comparing the missing height with the
+				// collections's lowest height.
+				// if collections's lowest height is higher than the missing height, it means the collections
+				// has been indexed by the collections already, no need to index again.
+				// And make sure reading the collections's lowest height is cheap operation (only hitting RW lock)
+
+				err = edp.indexer.IndexCollectionsForBlock(height, collections)
+				if err != nil {
+					ctx.Throw(fmt.Errorf("failed to index collections for block height %d: %w", height, err))
+					return
+				}
+
+				// Update processed height after successful indexing
+				err = edp.processedHeight.Set(height)
+				if err != nil {
+					ctx.Throw(fmt.Errorf("failed to update processed height to %d: %w", height, err))
+					return
+				}
+
+				// Log progress for each height with all relevant information
+				edp.log.Info().
+					Uint64("indexed", height).
+					Uint64("lowest_missing", lowestMissing).
+					Uint64("highest_available", highestAvailableHeight).
+					Uint64("processed_count", height-lowestMissing+1).
+					Uint64("remaining_count", highestAvailableHeight-height).
+					Uint64("total_to_process", highestAvailableHeight-lowestMissing+1).
+					Msg("indexed execution data progress")
+
+				edp.metrics.CollectionSyncedHeight(height)
+			}
+		}
+	}
+}
+
+func (edp *ExecutionDataProcessor) ProcessedHeight() uint64 {
+	return edp.processedHeight.Value()
+}

engine/access/collection_sync/execution_data_index/provider.go

@@ -0,0 +1,45 @@
+package execution_data_index
+
+import (
+	"context"
+
+	"github.com/onflow/flow-go/engine/access/collection_sync"
+	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
+	"github.com/onflow/flow-go/module/state_synchronization"
+)
+
+var _ collection_sync.ExecutionDataProvider = (*executionDataProvider)(nil)
+
+// executionDataProvider implements ExecutionDataProvider by querying ExecutionDataCache.
+type executionDataProvider struct {
+	cache                      execution_data.ExecutionDataCache
+	highestExectuionDataHeight state_synchronization.ExecutionDataIndexedHeight
+}
+
+// NewExecutionDataProvider creates a new ExecutionDataProvider that reads from the given ExecutionDataCache.
+// The headers storage is used to determine the search range for finding available heights.
+func NewExecutionDataProvider(
+	cache execution_data.ExecutionDataCache,
+	highestExectuionDataHeight state_synchronization.ExecutionDataIndexedHeight,
+) *executionDataProvider {
+	return &executionDataProvider{
+		cache:                      cache,
+		highestExectuionDataHeight: highestExectuionDataHeight,
+	}
+}
+
+// HighestIndexedHeight returns the highest block height for which execution data is available.
+func (p *executionDataProvider) HighestIndexedHeight() uint64 {
+	return p.highestExectuionDataHeight.HighestConsecutiveHeight()
+}
+
+// GetExecutionDataByHeight returns the execution data for the given block height.
+func (p *executionDataProvider) GetExecutionDataByHeight(ctx context.Context, height uint64) ([]*flow.Collection, error) {
+	blockExecutionData, err := p.cache.ByHeight(ctx, height)
+	if err != nil {
+		return nil, err
+	}
+
+	return blockExecutionData.StandardCollections(), nil
+}