refactor index block by view

state/protocol/badger/mutator.go

@@ -365,6 +365,12 @@ func (m *FollowerState) headerExtend(ctx context.Context, candidate *flow.Block,
 				return fmt.Errorf("could not store incorporated qc: %w", err)
 			}
 		} else {
+			// parent is a block that has been received and certified by a QC.
+			err := transaction.WithTx(operation.IndexBlockView(parent.View, qc.BlockID))(tx)
+			if err != nil {
+				return fmt.Errorf("could not index certified block: %w", err)
+			}
+
 			// trigger BlockProcessable for parent block above root height
 			if parent.Height > m.finalizedRootHeight {
 				tx.OnSucceed(func() {
@@ -389,6 +395,13 @@ func (m *FollowerState) headerExtend(ctx context.Context, candidate *flow.Block,
 			if err != nil {
 				return fmt.Errorf("could not store certifying qc: %w", err)
 			}
+
+			// candidate is a block that has been received and certified by a QC
+			err := transaction.WithTx(operation.IndexBlockView(candidate.Header.View, blockID))(tx)
+			if err != nil {
+				return fmt.Errorf("could not index certified block: %w", err)
+			}
+
 			tx.OnSucceed(func() { // queue a BlockProcessable event for candidate block, since it is certified
 				m.consumer.BlockProcessable(candidate.Header, certifyingQC)
 			})

storage/badger/blocks.go

@@ -83,6 +83,17 @@ func (b *Blocks) ByHeight(height uint64) (*flow.Block, error) {
 	return b.retrieveTx(blockID)(tx)
 }
 
+func (b *Blocks) ByView(view uint64) (*flow.Block, error) {
+	header, err := b.headers.ByView(view)
+	if err != nil {
+		return nil, err
+	}
+
+	tx := b.db.NewTransaction(false)
+	defer tx.Discard()
+	return b.retrieveTx(header.ID())(tx)
+}
+
 // ByCollectionID ...
 func (b *Blocks) ByCollectionID(collID flow.Identifier) (*flow.Block, error) {
 	var blockID flow.Identifier

storage/badger/qcs.go

@@ -1,8 +1,6 @@
 package badger
 
 import (
-	"fmt"
-
 	"github.com/dgraph-io/badger/v2"
 
 	"github.com/onflow/flow-go/model/flow"
@@ -31,12 +29,6 @@ func NewQuorumCertificates(collector module.CacheMetrics, db *badger.DB, cacheSi
 				return err
 			}
 
-			// qc are for certified block, which is unique by view
-			// while storing qc, we also index the block by view in the same transaction.
-			err = operation.SkipDuplicatesTx(transaction.WithTx(operation.IndexBlockView(qc.View, qc.BlockID)))(tx)
-			if err != nil {
-				return fmt.Errorf("could not index block by view (blockID: %v, view: %v): %w", qc.BlockID, qc.View, err)
-			}
 			return nil
 		}
 	}
@@ -51,7 +43,7 @@ func NewQuorumCertificates(collector module.CacheMetrics, db *badger.DB, cacheSi
 
 	return &QuorumCertificates{
 		db: db,
-		cache: newCache[flow.Identifier, *flow.QuorumCertificate](collector, metrics.ResourceQC,
+		cache: newCache(collector, metrics.ResourceQC,
 			withLimit[flow.Identifier, *flow.QuorumCertificate](cacheSize),
 			withStore(store),
 			withRetrieve(retrieve)),

storage/blocks.go

@@ -23,6 +23,12 @@ type Blocks interface {
 	// for finalized blocks.
 	ByHeight(height uint64) (*flow.Block, error)
 
+	// ByView returns the block with the given view. It is only available for certified blocks.
+	// certified blocks are the blocks that have received QC.
+	//
+	// TODO: this method is not available until next spork (mainnet27) or a migration that builds the index.
+	// ByView(view uint64) (*flow.Header, error)
+
 	// ByCollectionID returns the block for the given collection ID.
 	ByCollectionID(collID flow.Identifier) (*flow.Block, error)