Merge pull request #650 from onflow/janez/use-batch-to-init-blocks

Use batch to init blocks

bootstrap/bootstrap.go

@@ -519,7 +519,7 @@ func setupStorage(
 			return nil, fmt.Errorf("could not store register updates: %w", err)
 		}
 
-		if err := blocks.InitHeights(cadenceHeight, cadenceBlock.ID); err != nil {
+		if err := blocks.InitHeights(cadenceHeight, cadenceBlock.ID, batch); err != nil {
 			return nil, fmt.Errorf(
 				"failed to init the database for block height: %d and ID: %s, with : %w",
 				cadenceHeight,

services/ingestion/engine.go

@@ -168,8 +168,14 @@ func (e *Engine) processEvents(events *models.CadenceEvents) error {
 		return nil // nothing else to do this was heartbeat event with not event payloads
 	}
 
+	// TODO(JanezP): accounts need an indexed batch. Investigate why and try to switch to non-indexed batch
 	batch := e.store.NewIndexedBatch()
-	defer batch.Close()
+	defer func(batch *pebbleDB.Batch) {
+		err := batch.Close()
+		if err != nil {
+			e.log.Fatal().Err(err).Msg("failed to close batch")
+		}
+	}(batch)
 
 	// Step 1: Re-execute all transactions on the latest EVM block
 
@@ -205,6 +211,8 @@ func (e *Engine) processEvents(events *models.CadenceEvents) error {
 	}
 
 	// Step 2.2: Write the latest EVM block to `Blocks` storage
+	// This verifies the EVM height is sequential, and if not it will return an error
+	// TODO(janezp): can we do this before re-execution of the block?
 	err = e.indexBlock(
 		events.CadenceHeight(),
 		events.CadenceBlockID(),

services/replayer/blocks_provider_test.go

@@ -3,6 +3,8 @@ package replayer
 import (
 	"testing"
 
+	pebble2 "github.com/cockroachdb/pebble"
+
 	"github.com/goccy/go-json"
 	"github.com/onflow/flow-evm-gateway/config"
 	"github.com/onflow/flow-evm-gateway/models"
@@ -269,11 +271,15 @@ func setupBlocksDB(t *testing.T) storage.BlockIndexer {
 	dir := t.TempDir()
 	db, err := pebble.New(dir, zerolog.Nop())
 	require.NoError(t, err)
+	batch := db.NewBatch()
 
 	chainID := flowGo.Emulator
 	blocks := pebble.NewBlocks(db, chainID)
 
-	err = blocks.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1})
+	err = blocks.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1}, batch)
+	require.NoError(t, err)
+
+	err = batch.Commit(pebble2.Sync)
 	require.NoError(t, err)
 
 	return blocks

storage/pebble/blocks.go

@@ -222,24 +222,24 @@ func (b *Blocks) SetLatestCadenceHeight(height uint64, batch *pebble.Batch) erro
 }
 
 // InitHeights sets the Cadence height to zero as well as EVM heights. Used for empty database init.
-func (b *Blocks) InitHeights(cadenceHeight uint64, cadenceID flow.Identifier) error {
+func (b *Blocks) InitHeights(cadenceHeight uint64, cadenceID flow.Identifier, batch *pebble.Batch) error {
 	// sanity check, make sure we don't have any heights stored, disable overwriting the database
 	_, err := b.LatestEVMHeight()
 	if !errors.Is(err, errs.ErrStorageNotInitialized) {
 		return fmt.Errorf("can't init the database that already has data stored")
 	}
 
-	if err := b.store.set(latestCadenceHeightKey, nil, uint64Bytes(cadenceHeight), nil); err != nil {
+	if err := b.store.set(latestCadenceHeightKey, nil, uint64Bytes(cadenceHeight), batch); err != nil {
 		return fmt.Errorf("failed to init latest Cadence height at: %d, with: %w", cadenceHeight, err)
 	}
 
-	if err := b.store.set(latestEVMHeightKey, nil, uint64Bytes(0), nil); err != nil {
+	if err := b.store.set(latestEVMHeightKey, nil, uint64Bytes(0), batch); err != nil {
 		return fmt.Errorf("failed to init latest EVM height at: %d, with: %w", 0, err)
 	}
 
 	// we store genesis block because it isn't emitted over the network
 	genesisBlock := models.GenesisBlock(b.chainID)
-	if err := b.Store(cadenceHeight, cadenceID, genesisBlock, nil); err != nil {
+	if err := b.Store(cadenceHeight, cadenceID, genesisBlock, batch); err != nil {
 		return fmt.Errorf("failed to store genesis block at Cadence height: %d, with: %w", cadenceHeight, err)
 	}
 

storage/pebble/storage_test.go

@@ -23,8 +23,14 @@ import (
 func TestBlocks(t *testing.T) {
 	runDB("blocks", t, func(t *testing.T, db *Storage) {
 		bl := NewBlocks(db, flowGo.Emulator)
-		err := bl.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1})
+		batch := db.NewBatch()
+
+		err := bl.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1}, batch)
+		require.NoError(t, err)
+
+		err = batch.Commit(pebble.Sync)
 		require.NoError(t, err)
+
 		suite.Run(t, &storage.BlockTestSuite{Blocks: bl})
 	})
 }
@@ -33,11 +39,16 @@ func TestReceipts(t *testing.T) {
 	runDB("receipts", t, func(t *testing.T, db *Storage) {
 		// prepare the blocks database since they track heights which are used in receipts as well
 		bl := NewBlocks(db, flowGo.Emulator)
-		err := bl.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1})
+		batch := db.NewBatch()
+
+		err := bl.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1}, batch)
+		require.NoError(t, err)
+		err = bl.Store(30, flow.Identifier{0x1}, mocks.NewBlock(10), batch) // update first and latest height
 		require.NoError(t, err)
-		err = bl.Store(30, flow.Identifier{0x1}, mocks.NewBlock(10), nil) // update first and latest height
+		err = bl.Store(30, flow.Identifier{0x1}, mocks.NewBlock(300), batch) // update latest
 		require.NoError(t, err)
-		err = bl.Store(30, flow.Identifier{0x1}, mocks.NewBlock(300), nil) // update latest
+
+		err = batch.Commit(pebble.Sync)
 		require.NoError(t, err)
 
 		suite.Run(t, &storage.ReceiptTestSuite{ReceiptIndexer: NewReceipts(db)})
@@ -67,10 +78,15 @@ func TestBlock(t *testing.T) {
 	runDB("store block", t, func(t *testing.T, db *Storage) {
 		bl := mocks.NewBlock(10)
 		blocks := NewBlocks(db, flowGo.Emulator)
-		err := blocks.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1})
+		batch := db.NewBatch()
+
+		err := blocks.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1}, batch)
 		require.NoError(t, err)
 
-		err = blocks.Store(20, flow.Identifier{0x1}, bl, nil)
+		err = blocks.Store(20, flow.Identifier{0x1}, bl, batch)
+		require.NoError(t, err)
+
+		err = batch.Commit(pebble.Sync)
 		require.NoError(t, err)
 	})
 
@@ -81,10 +97,14 @@ func TestBlock(t *testing.T) {
 		bl := mocks.NewBlock(height)
 
 		blocks := NewBlocks(db, flowGo.Emulator)
-		err := blocks.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1})
+		batch := db.NewBatch()
+		err := blocks.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1}, batch)
+		require.NoError(t, err)
+
+		err = blocks.Store(cadenceHeight, cadenceID, bl, batch)
 		require.NoError(t, err)
 
-		err = blocks.Store(cadenceHeight, cadenceID, bl, nil)
+		err = batch.Commit(pebble.Sync)
 		require.NoError(t, err)
 
 		block, err := blocks.GetByHeight(height)
@@ -109,9 +129,15 @@ func TestBlock(t *testing.T) {
 
 	runDB("get not found block error", t, func(t *testing.T, db *Storage) {
 		blocks := NewBlocks(db, flowGo.Emulator)
-		err := blocks.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1})
+
+		batch := db.NewBatch()
+		err := blocks.InitHeights(config.EmulatorInitCadenceHeight, flow.Identifier{0x1}, batch)
+		require.NoError(t, err)
+		err = blocks.Store(2, flow.Identifier{0x1}, mocks.NewBlock(1), batch) // init
+		require.NoError(t, err)
+
+		err = batch.Commit(pebble.Sync)
 		require.NoError(t, err)
-		_ = blocks.Store(2, flow.Identifier{0x1}, mocks.NewBlock(1), nil) // init
 
 		bl, err := blocks.GetByHeight(11)
 		require.ErrorIs(t, err, errors.ErrEntityNotFound)