Differentiate between Consensus and Cluster Headers storage

admin/commands/storage/read_range_cluster_blocks.go

@@ -10,6 +10,7 @@ import (
 	"github.com/onflow/flow-go/admin/commands"
 	"github.com/onflow/flow-go/cmd/util/cmd/read-light-block"
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module/metrics"
 	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/storage/store"
 )
@@ -22,14 +23,12 @@ const Max_Range_Cluster_Block_Limit = uint64(10001)
 
 type ReadRangeClusterBlocksCommand struct {
 	db       storage.DB
-	headers  *store.Headers
 	payloads *store.ClusterPayloads
 }
 
-func NewReadRangeClusterBlocksCommand(db storage.DB, headers *store.Headers, payloads *store.ClusterPayloads) commands.AdminCommand {
+func NewReadRangeClusterBlocksCommand(db storage.DB, payloads *store.ClusterPayloads) commands.AdminCommand {
 	return &ReadRangeClusterBlocksCommand{
 		db:       db,
-		headers:  headers,
 		payloads: payloads,
 	}
 }
@@ -51,8 +50,12 @@ func (c *ReadRangeClusterBlocksCommand) Handler(ctx context.Context, req *admin.
 		return nil, admin.NewInvalidAdminReqErrorf("getting for more than %v blocks at a time might have an impact to node's performance and is not allowed", Max_Range_Cluster_Block_Limit)
 	}
 
+	clusterHeaders, err := store.NewClusterHeaders(metrics.NewNoopCollector(), c.db, flow.ChainID(chainID))
+	if err != nil {
+		return nil, err
+	}
 	clusterBlocks := store.NewClusterBlocks(
-		c.db, flow.ChainID(chainID), c.headers, c.payloads,
+		c.db, flow.ChainID(chainID), clusterHeaders, c.payloads,
 	)
 
 	lights, err := read.ReadClusterLightBlockByHeightRange(clusterBlocks, reqData.startHeight, reqData.endHeight)

cmd/collection/main.go

@@ -235,8 +235,8 @@ func main() {
 		}).
 		AdminCommand("read-range-cluster-blocks", func(conf *cmd.NodeConfig) commands.AdminCommand {
 			clusterPayloads := store.NewClusterPayloads(&metrics.NoopCollector{}, conf.ProtocolDB)
-			headers := store.NewHeaders(&metrics.NoopCollector{}, conf.ProtocolDB)
-			return storageCommands.NewReadRangeClusterBlocksCommand(conf.ProtocolDB, headers, clusterPayloads)
+			// defer construction of Headers since the cluster's ChainID is provided by the command
+			return storageCommands.NewReadRangeClusterBlocksCommand(conf.ProtocolDB, clusterPayloads)
 		}).
 		Module("follower distributor", func(node *cmd.NodeConfig) error {
 			followerDistributor = pubsub.NewFollowerDistributor()

cmd/scaffold.go

@@ -1186,8 +1186,45 @@ func (fnb *FlowNodeBuilder) initStorageLockManager() error {
 	return nil
 }
 
+// determineChainID attempts to determine the chain this node is running on
+// directly from the database or root snapshot, before storage interfaces have been initialized.
+// No errors expected during normal operation.
+func (fnb *FlowNodeBuilder) determineChainID() error {
+	bootstrapped, err := badgerState.IsBootstrapped(fnb.ProtocolDB)
+	if err != nil {
+		return err
+	}
+	if bootstrapped {
+		chainID, err := badgerState.GetChainID(fnb.ProtocolDB)
+		if err != nil {
+			return err
+		}
+		fnb.RootChainID = chainID
+	} else {
+		// if no root snapshot is configured, attempt to load the file from disk
+		var rootSnapshot = fnb.RootSnapshot
+		if rootSnapshot == nil {
+			fnb.Logger.Info().Msgf("loading root protocol state snapshot from disk")
+			rootSnapshot, err = loadRootProtocolSnapshot(fnb.BaseConfig.BootstrapDir)
+			if err != nil {
+				return fmt.Errorf("failed to read protocol snapshot from disk: %w", err)
+			}
+		}
+		// retrieve ChainID from the snapshot
+		sealingSegment, err := rootSnapshot.SealingSegment()
+		if err != nil {
+			return fmt.Errorf("failed to read ChainID from root snapshot: %w", err)
+		}
+		fnb.RootChainID = sealingSegment.Highest().ChainID
+	}
+	return nil
+}
+
 func (fnb *FlowNodeBuilder) initStorage() error {
-	headers := store.NewHeaders(fnb.Metrics.Cache, fnb.ProtocolDB)
+	headers, err := store.NewHeaders(fnb.Metrics.Cache, fnb.ProtocolDB, fnb.RootChainID)
+	if err != nil {
+		return err
+	}
 	guarantees := store.NewGuarantees(fnb.Metrics.Cache, fnb.ProtocolDB, fnb.BaseConfig.guaranteesCacheSize,
 		store.DefaultCacheSize)
 	seals := store.NewSeals(fnb.Metrics.Cache, fnb.ProtocolDB)
@@ -1457,7 +1494,6 @@ func (fnb *FlowNodeBuilder) setRootSnapshot(rootSnapshot protocol.Snapshot) erro
 		return fmt.Errorf("failed to read root QC: %w", err)
 	}
 
-	fnb.RootChainID = fnb.FinalizedRootBlock.ChainID
 	fnb.SporkID = fnb.RootSnapshot.Params().SporkID()
 
 	return nil
@@ -2081,16 +2117,20 @@ func (fnb *FlowNodeBuilder) onStart() error {
 		return err
 	}
 
-	if err := fnb.initStorage(); err != nil {
-		return err
-	}
-
 	for _, f := range fnb.preInitFns {
 		if err := fnb.handlePreInit(f); err != nil {
 			return err
 		}
 	}
 
+	if err := fnb.determineChainID(); err != nil {
+		return err
+	}
+
+	if err := fnb.initStorage(); err != nil {
+		return err
+	}
+
 	if err := fnb.initState(); err != nil {
 		return err
 	}

cmd/util/cmd/common/storage.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/rs/zerolog/log"
 
+	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/metrics"
 	"github.com/onflow/flow-go/storage"
 	storagebadger "github.com/onflow/flow-go/storage/badger"
@@ -53,9 +54,14 @@ func IsPebbleFolder(dataDir string) (bool, error) {
 	return pebblestorage.IsPebbleFolder(dataDir)
 }
 
-func InitStorages(db storage.DB) *store.All {
+// InitStorages initializes the common storage abstractions used by all node roles (with default cache sizes
+// suitable for mainnet). However, no metrics are collected (if you need metrics, use [store.InitAll] directly).
+// The chain ID indicates which Flow network the node is operating, referencing the ID of the blockchain
+// build by the main consensus, i.e. security nodes (not the chains built by collector clusters).
+// No errors are expected during normal operations.
+func InitStorages(db storage.DB, chainID flow.ChainID) (*store.All, error) {
 	metrics := &metrics.NoopCollector{}
-	return store.InitAll(metrics, db)
+	return store.InitAll(metrics, db, chainID)
 }
 
 // WithStorage runs the given function with the storage depending on the flags.

cmd/util/cmd/exec-data-json-export/block_exporter.go

@@ -12,6 +12,7 @@ import (
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/metrics"
+	badgerstate "github.com/onflow/flow-go/state/protocol/badger"
 	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/storage/store"
 )
@@ -37,9 +38,16 @@ func ExportBlocks(blockID flow.Identifier, dbPath string, outputPath string) (fl
 
 	// traverse backward from the given block (parent block) and fetch by blockHash
 	err := common.WithStorage(dbPath, func(db storage.DB) error {
+		chainID, err := badgerstate.GetChainID(db)
+		if err != nil {
+			return err
+		}
 
 		cacheMetrics := &metrics.NoopCollector{}
-		headers := store.NewHeaders(cacheMetrics, db)
+		headers, err := store.NewHeaders(cacheMetrics, db, chainID)
+		if err != nil {
+			return err
+		}
 		index := store.NewIndex(cacheMetrics, db)
 		guarantees := store.NewGuarantees(cacheMetrics, db, store.DefaultCacheSize, store.DefaultCacheSize)
 		seals := store.NewSeals(cacheMetrics, db)

cmd/util/cmd/exec-data-json-export/delta_snapshot_exporter.go

@@ -11,6 +11,7 @@ import (
 	"github.com/onflow/flow-go/fvm/storage/snapshot"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/metrics"
+	badgerstate "github.com/onflow/flow-go/state/protocol/badger"
 	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/storage/operation"
 	"github.com/onflow/flow-go/storage/store"
@@ -26,9 +27,16 @@ func ExportDeltaSnapshots(blockID flow.Identifier, dbPath string, outputPath str
 
 	// traverse backward from the given block (parent block) and fetch by blockHash
 	return common.WithStorage(dbPath, func(db storage.DB) error {
+		chainID, err := badgerstate.GetChainID(db)
+		if err != nil {
+			return err
+		}
 
 		cacheMetrics := &metrics.NoopCollector{}
-		headers := store.NewHeaders(cacheMetrics, db)
+		headers, err := store.NewHeaders(cacheMetrics, db, chainID)
+		if err != nil {
+			return err
+		}
 
 		activeBlockID := blockID
 		outputFile := filepath.Join(outputPath, "delta.jsonl")

cmd/util/cmd/exec-data-json-export/event_exporter.go

@@ -11,6 +11,7 @@ import (
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/metrics"
+	badgerstate "github.com/onflow/flow-go/state/protocol/badger"
 	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/storage/store"
 )
@@ -30,9 +31,16 @@ func ExportEvents(blockID flow.Identifier, dbPath string, outputPath string) err
 
 	// traverse backward from the given block (parent block) and fetch by blockHash
 	return common.WithStorage(dbPath, func(db storage.DB) error {
+		chainID, err := badgerstate.GetChainID(db)
+		if err != nil {
+			return err
+		}
 
 		cacheMetrics := &metrics.NoopCollector{}
-		headers := store.NewHeaders(cacheMetrics, db)
+		headers, err := store.NewHeaders(cacheMetrics, db, chainID)
+		if err != nil {
+			return err
+		}
 		events := store.NewEvents(cacheMetrics, db)
 		activeBlockID := blockID
 

cmd/util/cmd/exec-data-json-export/result_exporter.go

@@ -11,6 +11,7 @@ import (
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/metrics"
+	badgerstate "github.com/onflow/flow-go/state/protocol/badger"
 	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/storage/store"
 )
@@ -28,9 +29,16 @@ func ExportResults(blockID flow.Identifier, dbPath string, outputPath string) er
 
 	// traverse backward from the given block (parent block) and fetch by blockHash
 	return common.WithStorage(dbPath, func(db storage.DB) error {
+		chainID, err := badgerstate.GetChainID(db)
+		if err != nil {
+			return err
+		}
 
 		cacheMetrics := &metrics.NoopCollector{}
-		headers := store.NewHeaders(cacheMetrics, db)
+		headers, err := store.NewHeaders(cacheMetrics, db, chainID)
+		if err != nil {
+			return err
+		}
 		results := store.NewExecutionResults(cacheMetrics, db)
 		activeBlockID := blockID
 

cmd/util/cmd/exec-data-json-export/transaction_exporter.go

@@ -11,6 +11,7 @@ import (
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/metrics"
+	badgerstate "github.com/onflow/flow-go/state/protocol/badger"
 	"github.com/onflow/flow-go/storage/store"
 )
 
@@ -48,14 +49,22 @@ func ExportExecutedTransactions(blockID flow.Identifier, dbPath string, outputPa
 	}
 	defer db.Close()
 
+	chainID, err := badgerstate.GetChainID(db)
+	if err != nil {
+		return err
+	}
+
 	cacheMetrics := &metrics.NoopCollector{}
 	index := store.NewIndex(cacheMetrics, db)
 	guarantees := store.NewGuarantees(cacheMetrics, db, store.DefaultCacheSize, store.DefaultCacheSize)
 	seals := store.NewSeals(cacheMetrics, db)
 	results := store.NewExecutionResults(cacheMetrics, db)
 	receipts := store.NewExecutionReceipts(cacheMetrics, db, results, store.DefaultCacheSize)
 	transactions := store.NewTransactions(cacheMetrics, db)
-	headers := store.NewHeaders(cacheMetrics, db)
+	headers, err := store.NewHeaders(cacheMetrics, db, chainID)
+	if err != nil {
+		return err
+	}
 	payloads := store.NewPayloads(db, index, guarantees, seals, receipts, results)
 	blocks := store.NewBlocks(db, headers, payloads)
 	collections := store.NewCollections(db, transactions)

cmd/util/cmd/export-json-transactions/cmd.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	"github.com/onflow/flow-go/cmd/util/cmd/export-json-transactions/transactions"
+	badgerstate "github.com/onflow/flow-go/state/protocol/badger"
 	"github.com/onflow/flow-go/storage"
 )
 
@@ -67,7 +68,14 @@ func ExportTransactions(lockManager lockctx.Manager, dataDir string, outputDir s
 
 	// init dependencies
 	return common.WithStorage(flagDatadir, func(db storage.DB) error {
-		storages := common.InitStorages(db)
+		chainID, err := badgerstate.GetChainID(db)
+		if err != nil {
+			return err
+		}
+		storages, err := common.InitStorages(db, chainID)
+		if err != nil {
+			return err
+		}
 
 		state, err := common.OpenProtocolState(lockManager, db, storages)
 		if err != nil {

cmd/util/cmd/export-json-transactions/transactions/range_test.go

@@ -60,7 +60,8 @@ func TestFindBlockTransactions(t *testing.T) {
 		)
 
 		// prepare dependencies
-		storages := common.InitStorages(db)
+		storages, err := common.InitStorages(db, flow.Emulator)
+		require.NoError(t, err)
 		payloads, collections := storages.Payloads, storages.Collections
 		snap4 := &mock.Snapshot{}
 		snap4.On("Head").Return(b1.ToHeader(), nil)
@@ -73,7 +74,7 @@ func TestFindBlockTransactions(t *testing.T) {
 		// store into database
 		p1 := unittest.ProposalFromBlock(b1)
 		p2 := unittest.ProposalFromBlock(b2)
-		err := unittest.WithLock(t, lockManager, storage.LockInsertBlock, func(lctx lockctx.Context) error {
+		err = unittest.WithLock(t, lockManager, storage.LockInsertBlock, func(lctx lockctx.Context) error {
 			return db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
 				err := storages.Blocks.BatchStore(lctx, rw, p1)
 				if err != nil {

cmd/util/cmd/find-inconsistent-result/cmd.go

@@ -11,6 +11,7 @@ import (
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/block_iterator/latest"
 	"github.com/onflow/flow-go/state/protocol"
+	badgerstate "github.com/onflow/flow-go/state/protocol/badger"
 	"github.com/onflow/flow-go/storage"
 )
 
@@ -95,7 +96,14 @@ func findFirstMismatch(datadir string, startHeight, endHeight uint64, lockManage
 
 func createStorages(db storage.DB, lockManager lockctx.Manager) (
 	storage.Headers, storage.ExecutionResults, storage.Seals, protocol.State, error) {
-	storages := common.InitStorages(db)
+	chainID, err := badgerstate.GetChainID(db)
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+	storages, err := common.InitStorages(db, chainID)
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
 	state, err := common.OpenProtocolState(lockManager, db, storages)
 	if err != nil {
 		return nil, nil, nil, nil, fmt.Errorf("could not open protocol state: %v", err)

cmd/util/cmd/read-badger/cmd/blocks.go

@@ -9,6 +9,7 @@ import (
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/metrics"
+	badgerstate "github.com/onflow/flow-go/state/protocol/badger"
 	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/storage/store"
 )
@@ -28,8 +29,15 @@ var blocksCmd = &cobra.Command{
 	Short: "get a block by block ID or height",
 	RunE: func(cmd *cobra.Command, args []string) error {
 		return common.WithStorage(flagDatadir, func(db storage.DB) error {
-			cacheMetrics := &metrics.NoopCollector{}
-			headers := store.NewHeaders(cacheMetrics, db)
+			chainID, err := badgerstate.GetChainID(db)
+			if err != nil {
+				return err
+			}
+			cacheMetrics := metrics.NewNoopCollector()
+			headers, err := store.NewHeaders(cacheMetrics, db, chainID)
+			if err != nil {
+				return err
+			}
 			index := store.NewIndex(cacheMetrics, db)
 			guarantees := store.NewGuarantees(cacheMetrics, db, store.DefaultCacheSize, store.DefaultCacheSize)
 			seals := store.NewSeals(cacheMetrics, db)
@@ -39,7 +47,6 @@ var blocksCmd = &cobra.Command{
 			blocks := store.NewBlocks(db, headers, payloads)
 
 			var block *flow.Block
-			var err error
 
 			if flagBlockID != "" {
 				log.Info().Msgf("got flag block id: %s", flagBlockID)