[Access] POC Collection Syncing

cmd/execution_builder.go

@@ -413,6 +413,10 @@ func (exeNode *ExecutionNode) LoadBlobService(
 		opts = append(opts, blob.WithReprovideInterval(-1))
 	}
 
+	if !node.BitswapBloomCacheEnabled {
+		opts = append(opts, blob.WithSkipBloomCache(true))
+	}
+
 	if exeNode.exeConf.blobstoreRateLimit > 0 && exeNode.exeConf.blobstoreBurstLimit > 0 {
 		opts = append(opts, blob.WithRateLimit(float64(exeNode.exeConf.blobstoreRateLimit), exeNode.exeConf.blobstoreBurstLimit))
 	}

cmd/node_builder.go

@@ -186,6 +186,13 @@ type BaseConfig struct {
 	// This is only meaningful to Access and Execution nodes.
 	BitswapReprovideEnabled bool
 
+	// BitswapBloomCacheEnabled configures whether the Bitswap bloom cache is enabled.
+	// When disabled, uses a plain blockstore instead of cached blockstore, avoiding
+	// the CPU cost of building the bloom filter on startup. Pebble's built-in bloom
+	// filters (persisted in SSTables) are still used for efficient lookups.
+	// This is only meaningful to Access and Execution nodes.
+	BitswapBloomCacheEnabled bool
+
 	TransactionFeesDisabled bool
 }
 
@@ -297,12 +304,13 @@ func DefaultBaseConfig() *BaseConfig {
 			Duration: 10 * time.Second,
 		},
 
-		HeroCacheMetricsEnable:  false,
-		SyncCoreConfig:          chainsync.DefaultConfig(),
-		CodecFactory:            codecFactory,
-		ComplianceConfig:        compliance.DefaultConfig(),
-		DhtSystemEnabled:        true,
-		BitswapReprovideEnabled: true,
+		HeroCacheMetricsEnable:   false,
+		SyncCoreConfig:           chainsync.DefaultConfig(),
+		CodecFactory:             codecFactory,
+		ComplianceConfig:         compliance.DefaultConfig(),
+		DhtSystemEnabled:         true,
+		BitswapReprovideEnabled:  true,
+		BitswapBloomCacheEnabled: true, // default: use cached blockstore TODO leo: change default to false
 	}
 }
 

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"
@@ -1104,7 +1102,6 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 	var ds datastore.Batching
 	var bs network.BlobService
 	var processedBlockHeight storage.ConsumerProgressInitializer
-	var processedNotifications storage.ConsumerProgressInitializer
 	var publicBsDependable *module.ProxiedReadyDoneAware
 	var execDataDistributor *edrequester.ExecutionDataDistributor
 	var execDataCacheBackend *herocache.BlockExecutionData
@@ -1152,14 +1149,6 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 			processedBlockHeight = store.NewConsumerProgress(db, module.ConsumeProgressExecutionDataRequesterBlockHeight)
 			return nil
 		}).
-		Module("processed notifications consumer progress", func(node *cmd.NodeConfig) error {
-			// Note: progress is stored in the datastore's DB since that is where the jobqueue
-			// writes execution data to.
-			db := builder.ExecutionDatastoreManager.DB()
-
-			processedNotifications = store.NewConsumerProgress(db, module.ConsumeProgressExecutionDataRequesterNotification)
-			return nil
-		}).
 		Module("blobservice peer manager dependencies", func(node *cmd.NodeConfig) error {
 			publicBsDependable = module.NewProxiedReadyDoneAware()
 			builder.PeerManagerDependencies.Add(publicBsDependable)
@@ -1200,6 +1189,10 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				),
 			}
 
+			if !builder.BitswapBloomCacheEnabled {
+				opts = append(opts, blob.WithSkipBloomCache(true))
+			}
+
 			var err error
 			bs, err = node.EngineRegistry.RegisterBlobService(channels.PublicExecutionDataService, ds, opts...)
 			if err != nil {
@@ -1292,7 +1285,6 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				builder.ExecutionDataDownloader,
 				executionDataCache,
 				processedBlockHeight,
-				processedNotifications,
 				builder.State,
 				builder.Storage.Headers,
 				builder.executionDataConfig,
@@ -1438,31 +1430,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 +1445,6 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				builder.scheduledTransactions,
 				builder.RootChainID,
 				indexerDerivedChainData,
-				collectionIndexer,
 				collectionExecutedMetric,
 				node.StorageLockMgr,
 			)
@@ -1556,10 +1523,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 +1541,7 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				builder.executionDataConfig.InitialBlockHeight,
 				node.Storage.Headers,
 				broadcaster,
-				highestAvailableHeight,
+				builder.ExecutionDataRequester,
 				builder.EventsIndex,
 				useIndex,
 			)

cmd/scaffold.go

@@ -213,6 +213,10 @@ func (fnb *FlowNodeBuilder) BaseFlags() {
 		"bitswap-reprovide-enabled",
 		defaultConfig.BitswapReprovideEnabled,
 		"[experimental] whether to enable bitswap reproviding. This is an experimental feature. Use with caution.")
+	fnb.flags.BoolVar(&fnb.BaseConfig.BitswapBloomCacheEnabled,
+		"bitswap-bloom-cache-enabled",
+		defaultConfig.BitswapBloomCacheEnabled,
+		"[experimental] whether to enable bitswap bloom cache. When disabled, uses a plain blockstore instead of cached blockstore, avoiding the CPU cost of building the bloom filter on startup. Pebble's built-in bloom filters (persisted in SSTables) are still used. This is an experimental feature. Use with caution.")
 
 	// dynamic node startup flags
 	fnb.flags.StringVar(&fnb.BaseConfig.DynamicStartupANPubkey,

cmd/util/cmd/read-protocol-state/cmd/blocks.go

@@ -222,6 +222,8 @@ func runE(*cobra.Command, []string) error {
 				return fmt.Errorf("could not get root block: %w", err)
 			}
 
+			log.Info().Msgf("searching executed block between heights %v and %v", root.Height, sealed.Height)
+
 			// find the last executed and sealed block
 			for h := sealed.Height; h >= root.Height; h-- {
 				block, err := reader.GetBlockByHeight(h)
@@ -238,6 +240,8 @@ func runE(*cobra.Command, []string) error {
 					common.PrettyPrintEntity(block)
 					return nil
 				}
+
+				log.Info().Msgf("block at height %v is not executed yet", h)
 			}
 
 			return fmt.Errorf("could not find executed block")

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,98 @@
+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 tracks the given missing collection IDs for the given block height.
+	EnqueueMissingCollections(blockHeight uint64, ids []flow.Identifier, callback func()) error
+
+	// OnIndexedForBlock returns the callback function for the given block height
+	OnIndexedForBlock(blockHeight uint64) (func(), bool)
+
+	// On receipt of a collection, MCQ updates internal state and, if a block
+	// just became complete, returns: (collections, height, missingCollectionID, true).
+	// Otherwise, returns (nil, height, missingCollectionID, false).
+	// missingCollectionID is an arbitrary ID from the remaining missing collections, or ZeroID if none.
+	OnReceivedCollection(collection *flow.Collection) ([]*flow.Collection, uint64, flow.Identifier, 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 heights currently in the queue.
+	Size() uint
+
+	// GetMissingCollections returns all collection IDs that are currently missing across all block heights.
+	GetMissingCollections() []flow.Identifier
+
+	// GetMissingCollectionsByHeight returns a map of block height to collection IDs that are missing for that height.
+	GetMissingCollectionsByHeight() map[uint64][]flow.Identifier
+}
+
+// Requests collections by their guarantees.
+type CollectionRequester interface {
+	RequestCollectionsByGuarantees(guarantees []*flow.CollectionGuarantee) 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)
+}
+
+// BlockProcessor processes blocks to fetch and index their collections.
+type BlockProcessor interface {
+	// RequestCollectionsForBlock requests all missing collections for the given block.
+	FetchCollections(ctx irrecoverable.SignalerContext, block *flow.Block, done func()) error
+	// MissingCollectionQueueSize returns the number of missing collections currently in the queue.
+	MissingCollectionQueueSize() uint
+	// RetryFetchingMissingCollections retries fetching all missing collections currently in the queue.
+	RetryFetchingMissingCollections() error
+}
+
+// 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 notifies the fetcher that a new block has been finalized.
+	OnFinalizedBlock()
+	// MissingCollectionQueueSize returns the number of missing height currently in the queue.
+	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 returns the highest block height for which execution data is available.
+	// It garautnees that all heights below it also have execution data available to be called
+	// with GetExecutionDataByHeight.
+	HighestIndexedHeight() uint64
+
+	// It might return [execution_data.BlobNotFoundError] error, if some CID in the blob tree could not be found from the blobstore.
+	GetExecutionDataByHeight(ctx context.Context, height uint64) ([]*flow.Collection, error)
+}
+
+// ExecutionDataProcessor processes execution data when new execution data is available.
+type ExecutionDataProcessor interface {
+	// OnNewExectuionData notifies the processor that new execution data is available for processing.
+	OnNewExectuionData()
+}
+
+// ProgressReader provides the current progress of collection fetching/indexing.
+type ProgressReader interface {
+	// ProcessedHeight returns the highest block height for which collections have been processed.
+	ProcessedHeight() uint64
+}