fix: circular deps finalized head

.github/workflows/publish.yml

@@ -5,6 +5,8 @@ on:
     branches:
       - 'chore/ci-publish-docker'
       - 'main'
+      # TODO: remove temp branch
+      - 'fix/circular-deps-finalized-head'
     tags:
       - '*'
 

Makefile

@@ -13,6 +13,7 @@ GIT_ROOT := $(shell git rev-parse --show-toplevel)
 mock-gen:
 	@which mockgen > /dev/null || CGO_ENABLED=0 go install go.uber.org/mock/mockgen@latest
 	mockgen -source=db/interface.go -package mocks -destination $(MOCKS_DIR)/db_mock.go
+	mockgen -source=finalitygadget/interface.go -package mocks -destination $(MOCKS_DIR)/finalitygadget_mock.go
 	mockgen -source=finalitygadget/expected_clients.go -package mocks -destination $(MOCKS_DIR)/expected_clients_mock.go
 
 test:

cmd/opfgd/start.go

@@ -55,7 +55,12 @@ func runStartCmd(ctx client.Context, cmd *cobra.Command, args []string) error {
 	if err != nil {
 		return fmt.Errorf("failed to create DB handler: %w", err)
 	}
-	defer db.Close()
+	defer func() {
+		logger.Info("Closing DB...")
+		if dbErr := db.Close(); dbErr != nil {
+			logger.Error("Error closing DB", zap.Error(dbErr))
+		}
+	}()
 	err = db.CreateInitialSchema()
 	if err != nil {
 		return fmt.Errorf("create initial buckets error: %w", err)
@@ -88,6 +93,11 @@ func runStartCmd(ctx client.Context, cmd *cobra.Command, args []string) error {
 		}
 	}()
 
+	// Start finality gadget
+	if err := fg.Startup(fgCtx); err != nil {
+		logger.Fatal("Error starting finality gadget", zap.Error(err))
+	}
+
 	// Run finality gadget in a separate goroutine
 	go func() {
 		if err := fg.ProcessBlocks(fgCtx); err != nil {

config.toml.example

@@ -9,4 +9,5 @@ BBNChainID = "euphrates-0.4.0"
 BBNRPCAddress = "https://rpc-euphrates.devnet.babylonchain.io"
 GRPCListener = "0.0.0.0:50051"
 HTTPListener = "0.0.0.0:8080"
-PollInterval = "10s"
+PollInterval = "10s"
+BatchSize = 10

config/config.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/spf13/viper"
@@ -19,6 +20,47 @@ type Config struct {
 	HTTPListener      string        `long:"http-listener" description:"host:port to listen for HTTP connections"`
 	BitcoinDisableTLS bool          `long:"bitcoin-disable-tls" description:"disable TLS for RPC connections"`
 	PollInterval      time.Duration `long:"retry-interval" description:"interval in seconds to recheck Babylon finality of block"`
+	BatchSize         uint64        `long:"batch-size" description:"number of blocks to process in a batch"`
+}
+
+func (c *Config) Validate() error {
+	// Required fields
+	if c.L2RPCHost == "" {
+		return fmt.Errorf("l2-rpc-host is required")
+	}
+	if c.BitcoinRPCHost == "" {
+		return fmt.Errorf("bitcoin-rpc-host is required")
+	}
+	if c.FGContractAddress == "" {
+		return fmt.Errorf("fg-contract-address is required")
+	}
+	if c.BBNChainID == "" {
+		return fmt.Errorf("bbn-chain-id is required")
+	}
+	if c.BBNRPCAddress == "" {
+		return fmt.Errorf("bbn-rpc-address is required")
+	}
+	// TODO: add some default values if missing
+	if c.DBFilePath == "" {
+		return fmt.Errorf("db-file-path is required")
+	}
+	if c.GRPCListener == "" {
+		return fmt.Errorf("grpc-listener is required")
+	}
+	if c.HTTPListener == "" {
+		return fmt.Errorf("http-listener is required")
+	}
+
+	// Numeric validations
+	// TODO: add more validations (max batch size, min poll interval)
+	if c.PollInterval <= 0 {
+		return fmt.Errorf("poll-interval must be positive")
+	}
+	if c.BatchSize == 0 {
+		return fmt.Errorf("batch-size must be greater than 0")
+	}
+
+	return nil
 }
 
 func Load(configPath string) (*Config, error) {
@@ -34,5 +76,9 @@ func Load(configPath string) (*Config, error) {
 		return nil, err
 	}
 
+	if err := config.Validate(); err != nil {
+		return nil, fmt.Errorf("invalid configuration: %w", err)
+	}
+
 	return &config, nil
 }

db/bbolt.go

@@ -64,80 +64,72 @@ func (bb *BBoltHandler) CreateInitialSchema() error {
 	})
 }
 
-func (bb *BBoltHandler) InsertBlock(block *types.Block) error {
-	bb.logger.Info("Inserting block to DB", zap.Uint64("block_height", block.BlockHeight))
-
-	// Store mapping number -> block
-	err := bb.db.Update(func(tx *bolt.Tx) error {
-		b := tx.Bucket([]byte(blocksBucket))
-		key := bb.itob(block.BlockHeight)
-		blockBytes, err := json.Marshal(block)
-		if err != nil {
-			return err
-		}
-		return b.Put(key, blockBytes)
-	})
-	if err != nil {
-		bb.logger.Error("Error inserting block", zap.Error(err))
-		return err
+func (bb *BBoltHandler) InsertBlocks(blocks []*types.Block) error {
+	if len(blocks) == 0 {
+		return nil
 	}
 
-	// Store mapping hash -> number
-	err = bb.db.Update(func(tx *bolt.Tx) error {
-		b := tx.Bucket([]byte(blockHeightsBucket))
-		return b.Put([]byte(block.BlockHash), bb.itob(block.BlockHeight))
-	})
-	if err != nil {
-		bb.logger.Error("Error inserting block", zap.Error(err))
-		return err
-	}
+	bb.logger.Info("Batch inserting blocks to DB", zap.Int("count", len(blocks)))
+
+	// Single transaction for all operations
+	return bb.db.Update(func(tx *bolt.Tx) error {
+		blocksBucket := tx.Bucket([]byte(blocksBucket))
+		heightsBucket := tx.Bucket([]byte(blockHeightsBucket))
+		indexBucket := tx.Bucket([]byte(indexerBucket))
+
+		var minHeight, maxHeight uint64 = math.MaxUint64, 0
+
+		// Insert all blocks
+		for _, block := range blocks {
+			// Update min/max heights
+			if block.BlockHeight < minHeight {
+				minHeight = block.BlockHeight
+			}
+			if block.BlockHeight > maxHeight {
+				maxHeight = block.BlockHeight
+			}
+
+			// Store block data
+			blockBytes, err := json.Marshal(block)
+			if err != nil {
+				bb.logger.Error("Error inserting block", zap.Error(err))
+				return err
+			}
+			if err := blocksBucket.Put(bb.itob(block.BlockHeight), blockBytes); err != nil {
+				return err
+			}
 
-	// Get current earliest block
-	// If it is unset, update earliest block
-	earliestBlock, err := bb.QueryEarliestFinalizedBlock()
-	if earliestBlock == nil || errors.Is(err, types.ErrBlockNotFound) {
-		err = bb.db.Update(func(tx *bolt.Tx) error {
-			b := tx.Bucket([]byte(indexerBucket))
-			return b.Put([]byte(earliestBlockKey), bb.itob(block.BlockHeight))
-		})
-		if err != nil {
-			bb.logger.Error("Error updating earliest block", zap.Error(err))
-			return err
+			// Store height mapping
+			if err := heightsBucket.Put([]byte(block.BlockHash), bb.itob(block.BlockHeight)); err != nil {
+				bb.logger.Error("Error inserting height mapping", zap.Error(err))
+				return err
+			}
 		}
-	}
-	if err != nil {
-		bb.logger.Error("Error getting earliest block", zap.Error(err))
-		return err
-	}
 
-	// Get current latest block
-	latestBlock, err := bb.QueryLatestFinalizedBlock()
-	if latestBlock == nil {
-		latestBlock = &types.Block{BlockHeight: 0}
-	}
-	if err != nil {
-		bb.logger.Error("Error getting latest block", zap.Error(err))
-		return err
-	}
+		// Update earliest block if needed
+		earliestBytes := indexBucket.Get([]byte(earliestBlockKey))
+		if earliestBytes == nil {
+			if err := indexBucket.Put([]byte(earliestBlockKey), bb.itob(minHeight)); err != nil {
+				bb.logger.Error("Error inserting earliest block", zap.Error(err))
+				return err
+			}
+		}
 
-	// Update latest block if it's the latest
-	err = bb.db.Update(func(tx *bolt.Tx) error {
-		b := tx.Bucket([]byte(indexerBucket))
-		if err != nil {
-			bb.logger.Error("Error getting latest block", zap.Error(err))
-			return err
+		// Update latest block if needed
+		latestBytes := indexBucket.Get([]byte(latestBlockKey))
+		var currentLatest uint64
+		if latestBytes != nil {
+			currentLatest = bb.btoi(latestBytes)
 		}
-		if latestBlock.BlockHeight < block.BlockHeight {
-			return b.Put([]byte(latestBlockKey), bb.itob(block.BlockHeight))
+		if maxHeight > currentLatest {
+			if err := indexBucket.Put([]byte(latestBlockKey), bb.itob(maxHeight)); err != nil {
+				bb.logger.Error("Error inserting latest block", zap.Error(err))
+				return err
+			}
 		}
+
 		return nil
 	})
-	if err != nil {
-		bb.logger.Error("Error updating latest block", zap.Error(err))
-		return err
-	}
-
-	return nil
 }
 
 func (bb *BBoltHandler) GetBlockByHeight(height uint64) (*types.Block, error) {
@@ -174,6 +166,39 @@ func (bb *BBoltHandler) GetBlockByHash(hash string) (*types.Block, error) {
 	return bb.GetBlockByHeight(blockHeight)
 }
 
+func (bb *BBoltHandler) QueryIsBlockRangeFinalizedByHeight(startHeight, endHeight uint64) ([]bool, error) {
+	if startHeight > endHeight {
+		return nil, types.ErrInvalidBlockRange
+	}
+
+	// Create result slice with size of the range
+	len := endHeight - startHeight + 1
+	results := make([]bool, len)
+
+	err := bb.db.View(func(tx *bolt.Tx) error {
+		bucket := tx.Bucket([]byte(blocksBucket))
+
+		// Check each height in the range
+		for i := uint64(0); i < len; i++ {
+			height := startHeight + i
+			blockExists := bucket.Get(bb.itob(height)) != nil
+			// break early if block not found, as we only store consecutive blocks
+			if !blockExists {
+				break
+			}
+			results[i] = blockExists
+		}
+
+		return nil
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	return results, nil
+}
+
 func (bb *BBoltHandler) QueryIsBlockFinalizedByHeight(height uint64) (bool, error) {
 	_, err := bb.GetBlockByHeight(height)
 	if err != nil {
@@ -275,6 +300,7 @@ func (bb *BBoltHandler) SaveActivatedTimestamp(timestamp uint64) error {
 }
 
 func (bb *BBoltHandler) Close() error {
+	bb.logger.Info("Closing DB...")
 	return bb.db.Close()
 }