started implementation of mev block indexer

ethpandaops · Jun 16, 2024 · f72f53b · f72f53b
1 parent 71b686d
commit f72f53b
Show file tree

Hide file tree

Showing 6 changed files with 370 additions and 3 deletions.
diff --git a/db/mev_blocks.go b/db/mev_blocks.go
@@ -51,7 +51,7 @@ func InsertMevBlocks(mevBlocks []*dbtypes.MevBlock, tx *sqlx.Tx) error {
 		argIdx += fieldCount
 	}
 	fmt.Fprint(&sql, EngineQuery(map[dbtypes.DBEngineType]string{
-		dbtypes.DBEnginePgsql:  " ON CONFLICT (block_hash) DO UPDATE SET proposed = excluded.proposed",
+		dbtypes.DBEnginePgsql:  " ON CONFLICT (block_hash) DO UPDATE SET proposed = excluded.proposed, seenby_relays = excluded.seenby_relays",
 		dbtypes.DBEngineSqlite: "",
 	}))
 
@@ -62,7 +62,21 @@ func InsertMevBlocks(mevBlocks []*dbtypes.MevBlock, tx *sqlx.Tx) error {
 	return nil
 }
 
-func GetMevBlocksFiltered(offset uint64, limit uint32, finalizedBlock uint64, filter *dbtypes.MevBlockFilter) ([]*dbtypes.MevBlock, uint64, error) {
+func GetHighestMevBlockSlotByRelay(relayId uint8) (uint64, error) {
+	highestSlot := uint64(0)
+	err := ReaderDb.Get(&highestSlot, `
+	SELECT
+		MAX(slot_number)
+	FROM mev_blocks
+	WHERE (seenby_relays & $1) != 0
+	`, uint64(1)<<relayId)
+	if err != nil {
+		return 0, err
+	}
+	return highestSlot, nil
+}
+
+func GetMevBlocksFiltered(offset uint64, limit uint32, filter *dbtypes.MevBlockFilter) ([]*dbtypes.MevBlock, uint64, error) {
 	var sql strings.Builder
 	args := []any{}
 	fmt.Fprint(&sql, `

diff --git a/db/slots.go b/db/slots.go
@@ -183,6 +183,25 @@ func GetSlotByRoot(root []byte) *dbtypes.Slot {
 	return &block
 }
 
+func GetSlotsByBlockHash(blockHash []byte) []*dbtypes.Slot {
+	slots := []*dbtypes.Slot{}
+	err := ReaderDb.Select(&slots, `
+	SELECT
+		slot, proposer, status, root, parent_root, state_root, graffiti, graffiti_text,
+		attestation_count, deposit_count, exit_count, withdraw_count, withdraw_amount, attester_slashing_count, 
+		proposer_slashing_count, bls_change_count, eth_transaction_count, eth_block_number, eth_block_hash, 
+		eth_block_extra, eth_block_extra_text, sync_participation
+	FROM slots
+	WHERE eth_block_hash = $1
+	ORDER BY slot DESC
+	`, blockHash)
+	if err != nil {
+		logger.Errorf("Error while fetching slots by block hash: %v", err)
+		return nil
+	}
+	return slots
+}
+
 func parseAssignedSlots(rows *sql.Rows, fields []string, fieldsOffset int) []*dbtypes.AssignedSlot {
 	blockAssignments := []*dbtypes.AssignedSlot{}
 

diff --git a/services/chainservice.go b/services/chainservice.go
@@ -40,13 +40,14 @@ func StartChainService() error {
 		return nil
 	}
 
+	// init validator names & load inventory
 	validatorNames := NewValidatorNames()
 	loadingChan := validatorNames.LoadValidatorNames()
 	<-loadingChan
 
+	// reset sync state if configured
 	if utils.Config.Indexer.ResyncFromEpoch != nil {
 		err := db.RunDBTransaction(func(tx *sqlx.Tx) error {
-			// reset sync state
 			syncState := &dbtypes.IndexerSyncState{
 				Epoch: *utils.Config.Indexer.ResyncFromEpoch,
 			}
@@ -58,6 +59,7 @@ func StartChainService() error {
 		logrus.Warnf("Reset explorer synchronization status to epoch %v as configured! Please remove this setting again.", *utils.Config.Indexer.ResyncFromEpoch)
 	}
 
+	// configure and start beaconchain indexer
 	indexer, err := indexer.NewIndexer()
 	if err != nil {
 		return err
@@ -71,8 +73,13 @@ func StartChainService() error {
 		indexer.AddExecutionClient(uint16(idx), &endpoint)
 	}
 
+	// start validator names updater
 	validatorNames.StartUpdater(indexer)
 
+	// start mev index updater
+	mevIndexer := NewMevIndexer()
+	mevIndexer.StartUpdater(indexer)
+
 	GlobalBeaconService = &ChainService{
 		indexer:          indexer,
 		validatorNames:   validatorNames,