refactor: revise and add tests for exporter/modules/relays.go (#1372)

* refactor: updated relays.go handlers

* refactor: move and update relays db functions

* test: add ConsensusDBI mocks wth mockery

* refactor: add ctx cancellation in exporter and fetch payloads loop

* refactor: updated base.go

* refactor: add RelayClient to fetch delivered payloads in relays.go

* test: add relays tests

* fix: lint

* refactor: updated relayClient interface, rm SaveBlocksTags & SaveBlocksRelays db func and added SaveBlockTagsAndRelays db func, moved BidTrace type to /commons/types, updated tests

* refactor:rename ConsensusDBI to ConsensusRepository and ConsensusDB to DBConsensusRepository, moved repository implementation to a separate file

* test: rm ConsensusDBI mocks, create new mocks for ConsensusRepository woth mockery and update tests

* refactor: update consensus repository

* fix: update import in base.go

* refactor: move ConsensusRepository to db2/consensus.go and rename consensusRepository to ConsensusDB

* test: update ConsensusRepository mocks with mockery

Author: Monika-Bitfly

backend/pkg/commons/db2/consensus.go

@@ -1,8 +1,10 @@
 package db2
 
 import (
+	"database/sql"
 	"fmt"
 
+	"github.com/gobitfly/beaconchain/pkg/commons/log"
 	"github.com/gobitfly/beaconchain/pkg/commons/types"
 	"github.com/gobitfly/beaconchain/pkg/commons/utils"
 	"github.com/jmoiron/sqlx"
@@ -16,6 +18,13 @@ type ConsensusRepository interface {
 	UpdateBlockDepositCount(count int) error
 	SaveNetworkLivenessData(head *types.ChainHead) error
 	GetNetworkLivenessPreviousHeadEpoch() (uint64, error)
+	GetRelays() ([]types.Relay, error)
+	UpdateRelay(tagID, endpoint string) error
+	UpdateRelayLastExportTry(tagID, endpoint string) error
+	UpdateRelayExportFailureCount(exportFailureCount uint64, tagID, endpoint string) error
+	GetFirstRelayBlock(tagID string) (types.RelayBlock, error)
+	GetLastRelayBlock(tagID string) (types.RelayBlock, error)
+	SaveBlockTagsAndRelays(tagID string, payload types.BidTrace) error
 }
 
 type ConsensusDB struct {
@@ -125,3 +134,124 @@ func (c *ConsensusDB) GetNetworkLivenessPreviousHeadEpoch() (uint64, error) {
 	err := c.WriterDb.Get(&headEpoch, "SELECT COALESCE(MAX(headepoch), 0) FROM network_liveness")
 	return headEpoch, err
 }
+
+func (c *ConsensusDB) GetRelays() ([]types.Relay, error) {
+	var relays []types.Relay
+	err := c.ReaderDb.Select(&relays, `
+		SELECT tag_id, endpoint, public_link, is_censoring, is_ethical, export_failure_count, last_export_try_ts, last_export_success_ts
+		FROM relays`)
+
+	if err != nil && err != sql.ErrNoRows {
+		return nil, err
+	}
+	return relays, nil
+}
+
+func (c *ConsensusDB) UpdateRelay(tagID, endpoint string) error {
+	_, err := c.WriterDb.Exec(`
+		UPDATE relays SET
+			export_failure_count = 0,
+			last_export_try_ts = NOW() AT TIME ZONE 'utc',
+			last_export_success_ts = NOW() AT TIME ZONE 'utc'
+		WHERE tag_id = $1 AND endpoint = $2`, tagID, endpoint)
+
+	return err
+}
+
+func (c *ConsensusDB) UpdateRelayLastExportTry(tagID, endpoint string) error {
+	_, err := c.WriterDb.Exec(`
+			UPDATE relays SET
+				last_export_try_ts = (NOW() AT TIME ZONE 'utc')
+			WHERE tag_id = $1 AND endpoint = $2`, tagID, endpoint)
+
+	return err
+}
+
+func (c *ConsensusDB) UpdateRelayExportFailureCount(exportFailureCount uint64, tagID, endpoint string) error {
+	_, err := c.WriterDb.Exec(`
+	UPDATE relays SET
+		export_failure_count = $1,
+		last_export_try_ts = (NOW() AT TIME ZONE 'utc')
+	WHERE tag_id = $2 AND endpoint = $3`, exportFailureCount+1, tagID, endpoint)
+
+	return err
+}
+
+func (c *ConsensusDB) GetFirstRelayBlock(tagID string) (types.RelayBlock, error) {
+	var block types.RelayBlock
+	err := c.ReaderDb.Get(&block, `
+		SELECT tag_id, block_slot, block_root, exec_block_hash, value, builder_pubkey, proposer_pubkey, proposer_fee_recipient
+		FROM relays_blocks
+		WHERE tag_id=$1
+		ORDER BY block_slot ASC
+		LIMIT 1`, tagID)
+
+	return block, err
+}
+
+func (c *ConsensusDB) GetLastRelayBlock(tagID string) (types.RelayBlock, error) {
+	var block types.RelayBlock
+	err := c.ReaderDb.Get(&block, `
+		SELECT tag_id, block_slot, block_root, exec_block_hash, value, builder_pubkey, proposer_pubkey, proposer_fee_recipient
+		FROM relays_blocks
+		WHERE tag_id=$1
+		ORDER BY block_slot DESC
+		LIMIT 1`, tagID)
+
+	return block, err
+}
+
+func (c *ConsensusDB) SaveBlockTagsAndRelays(tagID string, payload types.BidTrace) error {
+	tx, err := c.WriterDb.Beginx()
+	if err != nil {
+		return err
+	}
+	defer utils.Rollback(tx)
+
+	// first insert the tag into the blocks_tags table
+	_, err = tx.Exec(`
+	INSERT INTO blocks_tags
+	SELECT blocks.slot, blocks.blockroot, $1
+	FROM blocks
+	WHERE blocks.slot = $2 AND blocks.exec_block_hash = $3
+	ON CONFLICT DO NOTHING`, tagID, payload.Slot,
+		utils.MustParseHex(payload.BlockHash))
+
+	if err != nil {
+		log.Error(fmt.Errorf("failed to insert payload into blocks_tags table"), "", 0, map[string]interface{}{"relay": tagID})
+		return err
+	}
+
+	// save relays
+	_, err = tx.Exec(`
+		INSERT INTO relays_blocks
+		(
+			tag_id,
+			block_slot,
+			block_root,
+			exec_block_hash,
+			value,
+			builder_pubkey,
+			proposer_pubkey,
+			proposer_fee_recipient
+		)
+		SELECT
+			$1,	blocks.slot, blocks.blockroot, blocks.exec_block_hash, $4, $5, $6, $7
+		FROM blocks
+		WHERE
+			blocks.slot = $2 and
+			blocks.exec_block_hash = $3
+		ON CONFLICT (block_slot, block_root, tag_id) DO NOTHING`,
+		tagID, payload.Slot, payload.Value,
+		utils.MustParseHex(payload.BlockHash),
+		utils.MustParseHex(payload.BuilderPubkey),
+		utils.MustParseHex(payload.ProposerPubkey),
+		utils.MustParseHex(payload.ProposerFeeRecipient))
+
+	if err != nil {
+		log.Error(fmt.Errorf("failed to insert payload into relays_blocks table"), "", 0, map[string]interface{}{"relay": tagID})
+		return err
+	}
+
+	return tx.Commit()
+}

backend/pkg/commons/db2/mocks/ConsensusRepository.go

@@ -575,6 +575,18 @@ type RelayBlock struct {
 	ProposerFeeRecipient string `db:"proposer_fee_recipient" json:"proposer_fee_recipient"`
 }
 
+type BidTrace struct {
+	Slot                 uint64    `json:"slot,string"`
+	ParentHash           string    `json:"parent_hash"`
+	BlockHash            string    `json:"block_hash"`
+	BuilderPubkey        string    `json:"builder_pubkey"`
+	ProposerPubkey       string    `json:"proposer_pubkey"`
+	ProposerFeeRecipient string    `json:"proposer_fee_recipient"`
+	GasLimit             uint64    `json:"gas_limit,string"`
+	GasUsed              uint64    `json:"gas_used,string"`
+	Value                WeiString `json:"value"`
+}
+
 type BlockTag struct {
 	ID        string `db:"tag_id"`
 	BlockSlot uint64 `db:"slot"`

backend/pkg/commons/types/exporter.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/gobitfly/beaconchain/pkg/commons/config"
+
 	"github.com/gobitfly/beaconchain/pkg/commons/db"
 	db2 "github.com/gobitfly/beaconchain/pkg/commons/db2"
 	"github.com/gobitfly/beaconchain/pkg/commons/log"
@@ -64,7 +65,8 @@ func StartAll(moduleCtx ModuleContext, modules []ModuleInterface, justV2 bool) {
 		}
 
 		if utils.Config.MevBoostRelayExporter.Enabled {
-			go mevBoostRelaysExporter()
+			relaysExporter := newRelaysExporter(ctx, consDB)
+			go relaysExporter.MEVBoostRelaysExporter()
 		}
 	}
 	// wait until the beacon-node is available