Merge pull request #6360 from onflow/leo/v0.33-pebble-storage-block-i…

…ndexer

[Pebble Storage] Refactor indexing new blocks

cmd/consensus/main.go

@@ -67,6 +67,7 @@ import (
 	pebbleState "github.com/onflow/flow-go/state/protocol/pebble"
 	"github.com/onflow/flow-go/storage"
 	bstorage "github.com/onflow/flow-go/storage/pebble"
+	"github.com/onflow/flow-go/storage/pebble/procedure"
 	"github.com/onflow/flow-go/utils/io"
 )
 
@@ -132,6 +133,7 @@ func main() {
 		getSealingConfigs   module.SealingConfigsGetter
 	)
 	var deprecatedFlagBlockRateDelay time.Duration
+	blockIndexer := procedure.NewBlockIndexer()
 
 	nodeBuilder := cmd.FlowNode(flow.RoleConsensus.String())
 	nodeBuilder.ExtraFlags(func(flags *pflag.FlagSet) {
@@ -285,6 +287,7 @@ func main() {
 				state,
 				node.Storage.Index,
 				node.Storage.Payloads,
+				blockIndexer,
 				blockTimer,
 				receiptValidator,
 				sealValidator,
@@ -730,6 +733,7 @@ func main() {
 				node.Storage.Seals,
 				node.Storage.Index,
 				node.Storage.Blocks,
+				blockIndexer,
 				node.Storage.Results,
 				node.Storage.Receipts,
 				guarantees,

cmd/util/cmd/read-light-block/read_light_block_test.go

@@ -26,9 +26,10 @@ func TestReadClusterRange(t *testing.T) {
 		err = operation.InsertClusterFinalizedHeight(parent.Header.ChainID, parent.Header.Height)(db)
 		require.NoError(t, err)
 
+		blockIndexer := procedure.NewClusterBlockIndexer()
 		// add blocks
 		for _, block := range blocks {
-			err := operation.WithReaderBatchWriter(db, procedure.InsertClusterBlock(&block))
+			err := operation.WithReaderBatchWriter(db, blockIndexer.InsertClusterBlock(&block))
 			require.NoError(t, err)
 
 			err = operation.WithReaderBatchWriter(db, procedure.FinalizeClusterBlock(block.Header.ID()))

consensus/integration/nodes_test.go

@@ -58,6 +58,7 @@ import (
 	"github.com/onflow/flow-go/state/protocol/util"
 	storagemock "github.com/onflow/flow-go/storage/mock"
 	storage "github.com/onflow/flow-go/storage/pebble"
+	"github.com/onflow/flow-go/storage/pebble/procedure"
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
@@ -403,6 +404,7 @@ func createNode(
 
 	blockTimer, err := blocktimer.NewBlockTimer(1*time.Millisecond, 90*time.Second)
 	require.NoError(t, err)
+	blockIndexer := procedure.NewBlockIndexer()
 
 	fullState, err := bprotocol.NewFullConsensusState(
 		log,
@@ -411,6 +413,7 @@ func createNode(
 		state,
 		indexDB,
 		payloadsDB,
+		blockIndexer,
 		blockTimer,
 		util.MockReceiptValidator(),
 		util.MockSealValidator(sealsDB),
@@ -458,7 +461,7 @@ func createNode(
 	seals := stdmap.NewIncorporatedResultSeals(sealLimit)
 
 	// initialize the block builder
-	build, err := builder.NewBuilderPebble(metricsCollector, db, fullState, headersDB, sealsDB, indexDB, blocksDB, resultsDB, receiptsDB,
+	build, err := builder.NewBuilderPebble(metricsCollector, db, fullState, headersDB, sealsDB, indexDB, blocksDB, blockIndexer, resultsDB, receiptsDB,
 		guarantees, consensusMempools.NewIncorporatedResultSeals(seals, receiptsDB), receipts, tracer)
 	require.NoError(t, err)
 

engine/collection/epochmgr/factories/builder.go

@@ -55,6 +55,7 @@ func (f *BuilderFactory) Create(
 	clusterState clusterstate.State,
 	clusterHeaders storage.Headers,
 	clusterPayloads storage.ClusterPayloads,
+	blockIndexer storage.ClusterBlockIndexer,
 	pool mempool.Transactions,
 	epoch uint64,
 ) (module.Builder, *finalizer.FinalizerPebble, error) {
@@ -67,6 +68,7 @@ func (f *BuilderFactory) Create(
 		f.mainChainHeaders,
 		clusterHeaders,
 		clusterPayloads,
+		blockIndexer,
 		pool,
 		f.log,
 		epoch,

engine/collection/epochmgr/factories/cluster_state.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/onflow/flow-go/module"
 	clusterkv "github.com/onflow/flow-go/state/cluster/pebble"
+	"github.com/onflow/flow-go/storage"
 	bstorage "github.com/onflow/flow-go/storage/pebble"
 )
 
@@ -29,7 +30,7 @@ func NewClusterStateFactory(
 	return factory, nil
 }
 
-func (f *ClusterStateFactory) Create(stateRoot *clusterkv.StateRoot) (
+func (f *ClusterStateFactory) Create(stateRoot *clusterkv.StateRoot, blockIndexer storage.ClusterBlockIndexer) (
 	*clusterkv.MutableState,
 	*bstorage.Headers,
 	*bstorage.ClusterPayloads,
@@ -58,7 +59,7 @@ func (f *ClusterStateFactory) Create(stateRoot *clusterkv.StateRoot) (
 		}
 	}
 
-	mutableState, err := clusterkv.NewMutableState(clusterState, f.tracer, headers, payloads)
+	mutableState, err := clusterkv.NewMutableState(clusterState, f.tracer, headers, payloads, blockIndexer)
 	if err != nil {
 		return nil, nil, nil, nil, fmt.Errorf("could create mutable cluster state: %w", err)
 	}

engine/collection/epochmgr/factories/epoch.go

@@ -12,6 +12,7 @@ import (
 	"github.com/onflow/flow-go/state/cluster/pebble"
 	"github.com/onflow/flow-go/state/protocol"
 	"github.com/onflow/flow-go/storage"
+	"github.com/onflow/flow-go/storage/pebble/procedure"
 )
 
 type EpochComponentsFactory struct {
@@ -114,8 +115,11 @@ func (factory *EpochComponentsFactory) Create(
 		err = fmt.Errorf("could not create valid state root: %w", err)
 		return
 	}
+
+	blockIndexer := procedure.NewClusterBlockIndexer()
+
 	var mutableState *pebble.MutableState
-	mutableState, headers, payloads, blocks, err = factory.state.Create(stateRoot)
+	mutableState, headers, payloads, blocks, err = factory.state.Create(stateRoot, blockIndexer)
 	state = mutableState
 	if err != nil {
 		err = fmt.Errorf("could not create cluster state: %w", err)
@@ -125,7 +129,7 @@ func (factory *EpochComponentsFactory) Create(
 	// get the transaction pool for the epoch
 	pool := factory.pools.ForEpoch(epochCounter)
 
-	builder, finalizer, err := factory.builder.Create(state, headers, payloads, pool, epochCounter)
+	builder, finalizer, err := factory.builder.Create(state, headers, payloads, blockIndexer, pool, epochCounter)
 	if err != nil {
 		err = fmt.Errorf("could not create builder/finalizer: %w", err)
 		return

engine/testutil/nodes.go

@@ -107,6 +107,7 @@ import (
 	"github.com/onflow/flow-go/state/protocol/util"
 	storage "github.com/onflow/flow-go/storage/pebble"
 	storagepebble "github.com/onflow/flow-go/storage/pebble"
+	"github.com/onflow/flow-go/storage/pebble/procedure"
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
@@ -271,6 +272,7 @@ func CompleteStateFixture(
 		state,
 		s.Index,
 		s.Payloads,
+		procedure.NewBlockIndexer(),
 		util.MockBlockTimer(),
 		util.MockReceiptValidator(),
 		util.MockSealValidator(s.Seals),

module/builder/collection/builder_pebble.go

@@ -20,7 +20,6 @@ import (
 	"github.com/onflow/flow-go/state/protocol"
 	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/storage/pebble/operation"
-	"github.com/onflow/flow-go/storage/pebble/procedure"
 	"github.com/onflow/flow-go/utils/logging"
 )
 
@@ -37,6 +36,7 @@ type BuilderPebble struct {
 	protoState     protocol.State
 	clusterState   clusterstate.State
 	payloads       storage.ClusterPayloads
+	blockIndexer   storage.ClusterBlockIndexer
 	transactions   mempool.Transactions
 	tracer         module.Tracer
 	config         Config
@@ -56,6 +56,7 @@ func NewBuilderPebble(
 	mainHeaders storage.Headers,
 	clusterHeaders storage.Headers,
 	payloads storage.ClusterPayloads,
+	blockIndexer storage.ClusterBlockIndexer,
 	transactions mempool.Transactions,
 	log zerolog.Logger,
 	epochCounter uint64,
@@ -69,6 +70,7 @@ func NewBuilderPebble(
 		mainHeaders:    mainHeaders,
 		clusterHeaders: clusterHeaders,
 		payloads:       payloads,
+		blockIndexer:   blockIndexer,
 		transactions:   transactions,
 		config:         DefaultConfig(),
 		log:            log.With().Str("component", "cluster_builder").Logger(),
@@ -193,7 +195,7 @@ func (b *BuilderPebble) BuildOn(parentID flow.Identifier, setter func(*flow.Head
 
 	// STEP 4: insert the cluster block to the database.
 	span, _ = b.tracer.StartSpanFromContext(ctx, trace.COLBuildOnDBInsert)
-	err = operation.WithReaderBatchWriter(b.db, procedure.InsertClusterBlock(&proposal))
+	err = operation.WithReaderBatchWriter(b.db, b.blockIndexer.InsertClusterBlock(&proposal))
 	span.End()
 	if err != nil {
 		return nil, fmt.Errorf("could not insert built block: %w", err)

module/builder/collection/builder_pebble_test.go

@@ -43,9 +43,10 @@ type BuilderPebbleSuite struct {
 	chainID      flow.ChainID
 	epochCounter uint64
 
-	headers  storage.Headers
-	payloads storage.ClusterPayloads
-	blocks   storage.Blocks
+	headers      storage.Headers
+	payloads     storage.ClusterPayloads
+	blocks       storage.Blocks
+	blockIndexer storage.ClusterBlockIndexer
 
 	state cluster.MutableState
 
@@ -77,6 +78,7 @@ func (suite *BuilderPebbleSuite) SetupTest() {
 	suite.headers = all.Headers
 	suite.blocks = all.Blocks
 	suite.payloads = bstorage.NewClusterPayloads(metrics, suite.db)
+	suite.blockIndexer = procedure.NewClusterBlockIndexer()
 
 	// just bootstrap with a genesis block, we'll use this as reference
 	root, result, seal := unittest.BootstrapFixture(unittest.IdentityListFixture(5, unittest.WithAllRoles()))
@@ -93,7 +95,7 @@ func (suite *BuilderPebbleSuite) SetupTest() {
 	clusterState, err := clusterkv.Bootstrap(suite.db, clusterStateRoot)
 	suite.Require().NoError(err)
 
-	suite.state, err = clusterkv.NewMutableState(clusterState, tracer, suite.headers, suite.payloads)
+	suite.state, err = clusterkv.NewMutableState(clusterState, tracer, suite.headers, suite.payloads, suite.blockIndexer)
 	suite.Require().NoError(err)
 
 	state, err := ppebble.Bootstrap(
@@ -134,7 +136,7 @@ func (suite *BuilderPebbleSuite) SetupTest() {
 		suite.Assert().True(added)
 	}
 
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, tracer, suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter)
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, tracer, suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter)
 }
 
 // runs after each test finishes
@@ -146,7 +148,7 @@ func (suite *BuilderPebbleSuite) TearDownTest() {
 }
 
 func (suite *BuilderPebbleSuite) InsertBlock(block model.Block) {
-	err := operation.WithReaderBatchWriter(suite.db, procedure.InsertClusterBlock(&block))
+	err := operation.WithReaderBatchWriter(suite.db, suite.blockIndexer.InsertClusterBlock(&block))
 	suite.Assert().NoError(err)
 }
 
@@ -489,7 +491,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_LargeHistory() {
 
 	// use a mempool with 2000 transactions, one per block
 	suite.pool = herocache.NewTransactions(2000, unittest.Logger(), metrics.NewNoopCollector())
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter, builder.WithMaxCollectionSize(10000))
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter, builder.WithMaxCollectionSize(10000))
 
 	// get a valid reference block ID
 	final, err := suite.protoState.Final().Head()
@@ -569,7 +571,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_LargeHistory() {
 
 func (suite *BuilderPebbleSuite) TestBuildOn_MaxCollectionSize() {
 	// set the max collection size to 1
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter, builder.WithMaxCollectionSize(1))
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter, builder.WithMaxCollectionSize(1))
 
 	// build a block
 	header, err := suite.builder.BuildOn(suite.genesis.ID(), noopSetter)
@@ -587,7 +589,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_MaxCollectionSize() {
 
 func (suite *BuilderPebbleSuite) TestBuildOn_MaxCollectionByteSize() {
 	// set the max collection byte size to 400 (each tx is about 150 bytes)
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter, builder.WithMaxCollectionByteSize(400))
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter, builder.WithMaxCollectionByteSize(400))
 
 	// build a block
 	header, err := suite.builder.BuildOn(suite.genesis.ID(), noopSetter)
@@ -605,7 +607,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_MaxCollectionByteSize() {
 
 func (suite *BuilderPebbleSuite) TestBuildOn_MaxCollectionTotalGas() {
 	// set the max gas to 20,000
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter, builder.WithMaxCollectionTotalGas(20000))
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter, builder.WithMaxCollectionTotalGas(20000))
 
 	// build a block
 	header, err := suite.builder.BuildOn(suite.genesis.ID(), noopSetter)
@@ -642,7 +644,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_ExpiredTransaction() {
 
 	// reset the pool and builder
 	suite.pool = herocache.NewTransactions(10, unittest.Logger(), metrics.NewNoopCollector())
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter)
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter)
 
 	// insert a transaction referring genesis (now expired)
 	tx1 := unittest.TransactionBodyFixture(func(tx *flow.TransactionBody) {
@@ -684,7 +686,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_EmptyMempool() {
 
 	// start with an empty mempool
 	suite.pool = herocache.NewTransactions(1000, unittest.Logger(), metrics.NewNoopCollector())
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter)
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter)
 
 	header, err := suite.builder.BuildOn(suite.genesis.ID(), noopSetter)
 	suite.Require().NoError(err)
@@ -711,7 +713,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_NoRateLimiting() {
 	suite.ClearPool()
 
 	// create builder with no rate limit and max 10 tx/collection
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter,
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter,
 		builder.WithMaxCollectionSize(10),
 		builder.WithMaxPayerTransactionRate(0),
 	)
@@ -752,7 +754,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_RateLimitNonPayer() {
 	suite.ClearPool()
 
 	// create builder with 5 tx/payer and max 10 tx/collection
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter,
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter,
 		builder.WithMaxCollectionSize(10),
 		builder.WithMaxPayerTransactionRate(5),
 	)
@@ -796,7 +798,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_HighRateLimit() {
 	suite.ClearPool()
 
 	// create builder with 5 tx/payer and max 10 tx/collection
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter,
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter,
 		builder.WithMaxCollectionSize(10),
 		builder.WithMaxPayerTransactionRate(5),
 	)
@@ -834,7 +836,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_LowRateLimit() {
 	suite.ClearPool()
 
 	// create builder with .5 tx/payer and max 10 tx/collection
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter,
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter,
 		builder.WithMaxCollectionSize(10),
 		builder.WithMaxPayerTransactionRate(.5),
 	)
@@ -876,7 +878,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_UnlimitedPayer() {
 	// create builder with 5 tx/payer and max 10 tx/collection
 	// configure an unlimited payer
 	payer := unittest.RandomAddressFixture()
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter,
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter,
 		builder.WithMaxCollectionSize(10),
 		builder.WithMaxPayerTransactionRate(5),
 		builder.WithUnlimitedPayers(payer),
@@ -917,7 +919,7 @@ func (suite *BuilderPebbleSuite) TestBuildOn_RateLimitDryRun() {
 	// create builder with 5 tx/payer and max 10 tx/collection
 	// configure an unlimited payer
 	payer := unittest.RandomAddressFixture()
-	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter,
+	suite.builder, _ = builder.NewBuilderPebble(suite.db, trace.NewNoopTracer(), suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter,
 		builder.WithMaxCollectionSize(10),
 		builder.WithMaxPayerTransactionRate(5),
 		builder.WithRateLimitDryRun(true),
@@ -993,7 +995,7 @@ func benchmarkBuildOnPebble(b *testing.B, size int) {
 		state, err := clusterkv.Bootstrap(suite.db, stateRoot)
 		assert.NoError(b, err)
 
-		suite.state, err = clusterkv.NewMutableState(state, tracer, suite.headers, suite.payloads)
+		suite.state, err = clusterkv.NewMutableState(state, tracer, suite.headers, suite.payloads, suite.blockIndexer)
 		assert.NoError(b, err)
 
 		// add some transactions to transaction pool
@@ -1004,14 +1006,14 @@ func benchmarkBuildOnPebble(b *testing.B, size int) {
 		}
 
 		// create the builder
-		suite.builder, _ = builder.NewBuilderPebble(suite.db, tracer, suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.pool, unittest.Logger(), suite.epochCounter)
+		suite.builder, _ = builder.NewBuilderPebble(suite.db, tracer, suite.protoState, suite.state, suite.headers, suite.headers, suite.payloads, suite.blockIndexer, suite.pool, unittest.Logger(), suite.epochCounter)
 	}
 
 	// create a block history to test performance against
 	final := suite.genesis
 	for i := 0; i < size; i++ {
 		block := unittest.ClusterBlockWithParent(final)
-		err := operation.WithReaderBatchWriter(suite.db, procedure.InsertClusterBlock(&block))
+		err := operation.WithReaderBatchWriter(suite.db, suite.blockIndexer.InsertClusterBlock(&block))
 		require.NoError(b, err)
 
 		// finalize the block 80% of the time, resulting in a fork-rate of 20%