small refactor

storage/badger/commits.go

@@ -70,10 +70,6 @@ func (c *Commits) BatchStore(blockID flow.Identifier, commit flow.StateCommitmen
 	return operation.BatchIndexStateCommitment(blockID, commit)(writeBatch)
 }
 
-func (c *Commits) BatchStore2(blockID flow.Identifier, commit flow.StateCommitment, tx storage.BatchWriter) error {
-	return operation.BatchIndexStateCommitment(blockID, commit)(tx)
-}
-
 func (c *Commits) ByBlockID(blockID flow.Identifier) (flow.StateCommitment, error) {
 	tx := c.db.NewTransaction(false)
 	defer tx.Discard()

storage/pebble/headers.go

@@ -21,6 +21,10 @@ type Headers struct {
 
 func NewHeaders(collector module.CacheMetrics, db *pebble.DB) *Headers {
 
+	store := func(blockID flow.Identifier, header *flow.Header) func(storage.PebbleReaderBatchWriter) error {
+		return storage.OnlyWriter(operation.InsertHeader(blockID, header))
+	}
+
 	// CAUTION: should only be used to index FINALIZED blocks by their
 	// respective height
 	storeHeight := func(height uint64, id flow.Identifier) func(storage.PebbleReaderBatchWriter) error {
@@ -47,6 +51,7 @@ func NewHeaders(collector module.CacheMetrics, db *pebble.DB) *Headers {
 		db: db,
 		cache: newCache(collector, metrics.ResourceHeader,
 			withLimit[flow.Identifier, *flow.Header](4*flow.DefaultTransactionExpiry),
+			withStore(store),
 			withRetrieve(retrieve)),
 
 		heightCache: newCache(collector, metrics.ResourceFinalizedHeight,
@@ -59,19 +64,7 @@ func NewHeaders(collector module.CacheMetrics, db *pebble.DB) *Headers {
 }
 
 func (h *Headers) storePebble(blockID flow.Identifier, header *flow.Header) func(storage.PebbleReaderBatchWriter) error {
-	return func(rw storage.PebbleReaderBatchWriter) error {
-		rw.AddCallback(func() {
-			h.cache.Insert(blockID, header)
-		})
-
-		_, tx := rw.ReaderWriter()
-		err := operation.InsertHeader(blockID, header)(tx)
-		if err != nil {
-			return fmt.Errorf("could not store header %v: %w", blockID, err)
-		}
-
-		return nil
-	}
+	return h.cache.PutPebble(blockID, header)
 }
 
 func (h *Headers) retrieveTx(blockID flow.Identifier) func(pebble.Reader) (*flow.Header, error) {

storage/pebble/operation/batch.go

@@ -9,6 +9,7 @@ import (
 	"github.com/onflow/flow-go/storage"
 )
 
+// batchWriter wraps the storage.BatchWriter to make it compatible with pebble.Writer
 type batchWriter struct {
 	batch storage.BatchWriter
 }

storage/pebble/operation/common.go

@@ -379,7 +379,7 @@ func findHighestAtOrBelow(
 
 		key := append(prefix, b(height)...)
 		it, err := r.NewIter(&pebble.IterOptions{
-			UpperBound: append(key, 0xFF),
+			UpperBound: append(key, ffBytes...),
 		})
 		if err != nil {
 			return fmt.Errorf("can not create iterator: %w", err)

storage/pebble/qcs.go

@@ -15,7 +15,7 @@ import (
 
 // QuorumCertificates implements persistent storage for quorum certificates.
 type QuorumCertificates struct {
-	storing *sync.Mutex
+	storing sync.Mutex
 	db      *pebble.DB
 	cache   *Cache[flow.Identifier, *flow.QuorumCertificate]
 }