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,9 @@ 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 := store.NewHeaders(&metrics.NoopCollector{}, c.db, flow.ChainID(chainID))
 	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,29 @@ func (fnb *FlowNodeBuilder) initStorageLockManager() error {
 	return nil
 }
 
+func (fnb *FlowNodeBuilder) determineChainID() error {
+	if ok, _ := badgerState.IsBootstrapped(fnb.ProtocolDB); ok {
+		chainID, err := badgerState.GetChainIDFromLatestFinalizedHeader(fnb.ProtocolDB)
+		if err == nil {
+			fnb.RootChainID = chainID
+			return nil
+		}
+	}
+	// could not read from DB; try reading root snapshot from disk
+	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)
+	}
+	// set root snapshot fields (including RootChainID)
+	if err := fnb.setRootSnapshot(rootSnapshot); err != nil {
+		return err
+	}
+	return nil
+}
+
 func (fnb *FlowNodeBuilder) initStorage() error {
-	headers := store.NewHeaders(fnb.Metrics.Cache, fnb.ProtocolDB)
+	headers := store.NewHeaders(fnb.Metrics.Cache, fnb.ProtocolDB, fnb.RootChainID)
 	guarantees := store.NewGuarantees(fnb.Metrics.Cache, fnb.ProtocolDB, fnb.BaseConfig.guaranteesCacheSize,
 		store.DefaultCacheSize)
 	seals := store.NewSeals(fnb.Metrics.Cache, fnb.ProtocolDB)
@@ -2081,6 +2102,10 @@ func (fnb *FlowNodeBuilder) onStart() error {
 		return err
 	}
 
+	if err := fnb.determineChainID(); err != nil {
+		return err
+	}
+
 	if err := fnb.initStorage(); 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,9 @@ func IsPebbleFolder(dataDir string) (bool, error) {
 	return pebblestorage.IsPebbleFolder(dataDir)
 }
 
-func InitStorages(db storage.DB) *store.All {
+func InitStorages(db storage.DB, chainID flow.ChainID) *store.All {
 	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,13 @@ 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.GetChainIDFromLatestFinalizedHeader(db)
+		if err != nil {
+			return err
+		}
 
 		cacheMetrics := &metrics.NoopCollector{}
-		headers := store.NewHeaders(cacheMetrics, db)
+		headers := store.NewHeaders(cacheMetrics, db, chainID)
 		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,13 @@ 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.GetChainIDFromLatestFinalizedHeader(db)
+		if err != nil {
+			return err
+		}
 
 		cacheMetrics := &metrics.NoopCollector{}
-		headers := store.NewHeaders(cacheMetrics, db)
+		headers := store.NewHeaders(cacheMetrics, db, chainID)
 
 		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,13 @@ 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.GetChainIDFromLatestFinalizedHeader(db)
+		if err != nil {
+			return err
+		}
 
 		cacheMetrics := &metrics.NoopCollector{}
-		headers := store.NewHeaders(cacheMetrics, db)
+		headers := store.NewHeaders(cacheMetrics, db, chainID)
 		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,13 @@ 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.GetChainIDFromLatestFinalizedHeader(db)
+		if err != nil {
+			return err
+		}
 
 		cacheMetrics := &metrics.NoopCollector{}
-		headers := store.NewHeaders(cacheMetrics, db)
+		headers := store.NewHeaders(cacheMetrics, db, chainID)
 		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,19 @@ func ExportExecutedTransactions(blockID flow.Identifier, dbPath string, outputPa
 	}
 	defer db.Close()
 
+	chainID, err := badgerstate.GetChainIDFromLatestFinalizedHeader(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 := store.NewHeaders(cacheMetrics, db, chainID)
 	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,11 @@ 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.GetChainIDFromLatestFinalizedHeader(db)
+		if err != nil {
+			return err
+		}
+		storages := common.InitStorages(db, chainID)
 
 		state, err := common.OpenProtocolState(lockManager, db, storages)
 		if err != nil {

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

@@ -60,7 +60,7 @@ func TestFindBlockTransactions(t *testing.T) {
 		)
 
 		// prepare dependencies
-		storages := common.InitStorages(db)
+		storages := common.InitStorages(db, b1.ChainID)
 		payloads, collections := storages.Payloads, storages.Collections
 		snap4 := &mock.Snapshot{}
 		snap4.On("Head").Return(b1.ToHeader(), 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,11 @@ 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.GetChainIDFromLatestFinalizedHeader(db)
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+	storages := common.InitStorages(db, chainID)
 	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,12 @@ 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 {
+			chainID, err := badgerstate.GetChainIDFromLatestFinalizedHeader(db)
+			if err != nil {
+				return err
+			}
 			cacheMetrics := &metrics.NoopCollector{}
-			headers := store.NewHeaders(cacheMetrics, db)
+			headers := store.NewHeaders(cacheMetrics, db, chainID)
 			index := store.NewIndex(cacheMetrics, db)
 			guarantees := store.NewGuarantees(cacheMetrics, db, store.DefaultCacheSize, store.DefaultCacheSize)
 			seals := store.NewSeals(cacheMetrics, db)
@@ -39,7 +44,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)

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

@@ -34,12 +34,12 @@ var clusterBlocksCmd = &cobra.Command{
 		return common.WithStorage(flagDatadir, func(db storage.DB) error {
 			metrics := metrics.NewNoopCollector()
 
-			headers := store.NewHeaders(metrics, db)
-			clusterPayloads := store.NewClusterPayloads(metrics, db)
-
 			// get chain id
 			log.Info().Msgf("got flag chain name: %s", flagChainName)
 			chainID := flow.ChainID(flagChainName)
+
+			headers := store.NewHeaders(metrics, db, chainID)
+			clusterPayloads := store.NewClusterPayloads(metrics, db)
 			clusterBlocks := store.NewClusterBlocks(db, chainID, headers, clusterPayloads)
 
 			if flagClusterBlockID != "" && flagHeight != 0 {

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

@@ -8,6 +8,7 @@ import (
 
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	"github.com/onflow/flow-go/model/flow"
+	badgerstate "github.com/onflow/flow-go/state/protocol/badger"
 	"github.com/onflow/flow-go/storage"
 )
 
@@ -28,7 +29,11 @@ var collectionsCmd = &cobra.Command{
 	Short: "get collection by collection or transaction ID",
 	RunE: func(cmd *cobra.Command, args []string) error {
 		return common.WithStorage(flagDatadir, func(db storage.DB) error {
-			storages := common.InitStorages(db)
+			chainID, err := badgerstate.GetChainIDFromLatestFinalizedHeader(db)
+			if err != nil {
+				return err
+			}
+			storages := common.InitStorages(db, chainID) // TODO(4204) - header storage not used
 
 			if flagCollectionID != "" {
 				log.Info().Msgf("got flag collection id: %s", flagCollectionID)

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

@@ -8,6 +8,7 @@ import (
 
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	"github.com/onflow/flow-go/model/flow"
+	badgerstate "github.com/onflow/flow-go/state/protocol/badger"
 	"github.com/onflow/flow-go/storage"
 )
 
@@ -25,7 +26,11 @@ var epochCommitCmd = &cobra.Command{
 	Short: "get epoch commit by ID",
 	RunE: func(cmd *cobra.Command, args []string) error {
 		return common.WithStorage(flagDatadir, func(db storage.DB) error {
-			storages := common.InitStorages(db)
+			chainID, err := badgerstate.GetChainIDFromLatestFinalizedHeader(db)
+			if err != nil {
+				return err
+			}
+			storages := common.InitStorages(db, chainID) // TODO(4204) - header storage not used
 
 			log.Info().Msgf("got flag commit id: %s", flagEpochCommitID)
 			commitID, err := flow.HexStringToIdentifier(flagEpochCommitID)

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

@@ -8,6 +8,7 @@ import (
 
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	"github.com/onflow/flow-go/model/flow"
+	badgerstate "github.com/onflow/flow-go/state/protocol/badger"
 	"github.com/onflow/flow-go/storage"
 )
 
@@ -23,7 +24,11 @@ var epochProtocolStateCmd = &cobra.Command{
 	Short: "get epoch protocol state by block ID",
 	RunE: func(cmd *cobra.Command, args []string) error {
 		return common.WithStorage(flagDatadir, func(db storage.DB) error {
-			storages := common.InitStorages(db)
+			chainID, err := badgerstate.GetChainIDFromLatestFinalizedHeader(db)
+			if err != nil {
+				return err
+			}
+			storages := common.InitStorages(db, chainID) // TODO(4204) - header storage not used
 
 			log.Info().Msgf("got flag block id: %s", flagBlockID)
 			blockID, err := flow.HexStringToIdentifier(flagBlockID)