feat: add builder observer data indexer by rose2221 · Pull Request #802 · primev/mev-commit

rose2221 · 2025-09-24T15:42:56Z

feat(indexer): builder-observer data ingestor

Adds comprehensive blockchain indexing with MEV relay bid data and BeaconChain metadata for builder observer functionality.

Local Setup (StarRocks in Docker)

# 1) Start StarRocks all-in-one
docker run -d --name starrocks-local \
  -p 9030:9030 -p 8030:8030 -p 8040:8040 \
  starrocks/allin1-ubuntu:latest

# 2) Open MySQL shell into FE
docker exec -it starrocks-local mysql -h 127.0.0.1 -P 9030 -u root

Schema DDL (Updated for Builder Observer)

CREATE DATABASE IF NOT EXISTS indexer_db;
USE indexer_db;

-- Relays configuration
CREATE TABLE IF NOT EXISTS relays (
  relay_id BIGINT NOT NULL,
  name      VARCHAR(100),
  tag       VARCHAR(100),
  base_url  VARCHAR(500),
  is_active TINYINT
)
ENGINE=OLAP
PRIMARY KEY(relay_id)
DISTRIBUTED BY HASH(relay_id) BUCKETS 1
PROPERTIES (
  "replication_num" = "1",
  "enable_persistent_index" = "true"
);


-- Blocks with complete builder observer data
CREATE TABLE IF NOT EXISTS blocks (
  -- Core block data
  slot BIGINT,
  block_number BIGINT,
  timestamp DATETIME,
  proposer_index BIGINT,
  
  -- MEV/Relay data
  winning_relay VARCHAR(100),
  winning_builder_pubkey VARCHAR(200),
  fee_recipient VARCHAR(200),
  producer_reward_eth DECIMAL(20,8),
  
  -- Validator data
  validator_pubkey VARCHAR(200),
  validator_opted_in TINYINT,
  -- Tracking
  created_at DATETIME DEFAULT CURRENT_TIMESTAMP
) ENGINE=OLAP
PRIMARY KEY(slot)
DISTRIBUTED BY HASH(slot) BUCKETS 10
PROPERTIES("replication_num" = "1");

-- MEV relay bids
CREATE TABLE IF NOT EXISTS bids (
  slot BIGINT,
  relay_id BIGINT,
  builder_pubkey VARCHAR(200),
  proposer_pubkey VARCHAR(200),
  proposer_fee_recipient VARCHAR(200),
  value_wei VARCHAR(100),
  block_number BIGINT,
  timestamp_ms BIGINT
) ENGINE=OLAP
PRIMARY KEY(slot, relay_id, builder_pubkey)
DISTRIBUTED BY HASH(slot) BUCKETS 10
PROPERTIES("replication_num" = "1");

-- Application state and configuration storage
CREATE TABLE ingestor_state (
  id TINYINT NOT NULL,
  last_block_number BIGINT
)
ENGINE=OLAP
PRIMARY KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES (
  "replication_num" = "1",
  "enable_persistent_index" = "true"
);


## Environment Variables

Required:
```bash
DATABASE_URL=root@tcp(127.0.0.1:9030)/indexer_db
INFURA_RPC=https://mainnet.infura.io/v3/YOUR_KEY
BEACON_BASE=https://beaconcha.in/api/v1  # For BeaconChain.in API

Optional:

ETHERSCAN_API_KEY=your_key_here
HTTP_TIMEOUT=15s
BLOCK_TICK=12s
VALIDATOR_WAIT=1.5s
BACKFILL_EVERY=5m0s

Export and Run

export DATABASE_URL=root@tcp(127.0.0.1:9030)/indexer_db
export INFURA_RPC=https://mainnet.infura.io/v3/YOUR_KEY
export BEACON_BASE=https://beaconcha.in/api/v1

go run indexer/cmd/indexer/main.go

aloknerurkar · 2025-09-25T06:04:54Z

+	ctx2, cancel := context.WithTimeout(ctx, 3*time.Second)
+	defer cancel()
+
+	_, err := db.Conn.ExecContext(ctx2, `DELETE FROM ingestor_state WHERE id = 1`)


Both these operations should be done atomically in a transaction.

aloknerurkar · 2025-09-25T06:09:40Z

+	lastBN, found := db.LoadLastBlockNumber(ctx)
+	if !found || lastBN == 0 {
+		log.Printf(" [INIT] No previous state found, checking database for latest block...")
+		err := db.Conn.QueryRowContext(ctx, `SELECT COALESCE(MAX(block_number),0) FROM blocks`).Scan(&lastBN)


If the point of the DB is to mask the underlying storage layer, we shouldnt be accessing the Conn directly. Also if the DB layer is changed, for eg a new table instead of blocks is added, this will also have to be changed. The Conn should be unexported. Any functionality should be provided with the help of functions.

aloknerurkar · 2025-09-25T06:16:04Z

+
+	// Validate required configuration
+
+	log.Printf("[INIT] Starting blockchain indexer with StarRocks database")


In all other parts of the code we use slog. This outputs structured json logs which are easier to parse. It has option for text logging also and it can be configured. The other services also use this and we have some util pkg also. Better to start using this in the codebase.

All this manual [COMPONENT] can be added using log tags. This gives us the ability to filter using the tag in the logs instead of doing a text search.

aloknerurkar · 2025-09-25T06:22:20Z

+			// Fetch execution block data
+			ei, err := beacon.FetchCombinedBlockData(httpc, infuraRPC, beaconBase, nextBN)
+			if err != nil || ei == nil {
+				log.Printf("⏳ [BLOCK] Block %d not available yet: %v", nextBN, err)


Although I dont mind all these characters, not sure if anyone will be looking at this if things are running correctly. In which case you will only be looking at them when you have errors. Easier to parse the logs with lesser characters.

aloknerurkar · 2025-09-25T06:34:22Z

+
+		case <-backfillTicker.C:
+			log.Printf("[BACKFILL] Starting backfill operations...")
+			backfill.RunAll(ctx, db, httpc, createOptionsFromCLI(c), relays)


Why is ticker needed for backfill? This operation should only happen once. Also it can run in parallel with the main go routine so it should just be started before we enter this loop.

aloknerurkar · 2025-09-25T06:36:39Z

+	httputil "github.com/primev/mev-commit/indexer/pkg/http"
+)
+
+const optInABIJSON = `[


I gave this comment in the last PR as well. We can use the generated client in contracts-abi. All it needs is ethereum client and it will handle all the data types.

aloknerurkar · 2025-09-25T06:38:46Z

+		},
+	}
+}
+func FetchJSONWithRetry(ctx context.Context, httpc *http.Client, url string, out any, attempts int, baseDelay time.Duration) error {


If you want to provide these type of utility functions we cannot just assume this is doing the right thing. Unit tests should be added for this pkg. If not we can use some library that provides this functionality.

aloknerurkar · 2025-09-25T06:40:23Z

+
+}
+func (db *DB) Close() {
+	db.Conn.Close()


This doesnt return error?

Also if its the job of the DB to close this, we should not be exposing the conn directly outside.

kant777 · 2025-10-01T18:16:10Z

@rose2221 This PR seems to have lot of unrelated changes. Is it because of rebase? if so, please fix it

aloknerurkar · 2025-10-06T12:04:24Z

+	infuraRPC := c.String(optionInfuraRPC.Name)
+	beaconBase := c.String(optionBeaconBase.Name)
+	// Initialize random seed
+	rand.Seed(time.Now().UnixNano())


Why is this needed?

aloknerurkar · 2025-10-06T12:05:10Z

+		"validator_delay", c.Duration("validator-delay"))
+
+	// Setup graceful shutdown
+	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)


I think we spoke about this. This is already being done in main.go. This is not needed.

aloknerurkar · 2025-10-06T12:14:25Z

+
+		latestBlock, err := ethereum.GetLatestBlockNumber(httpc.HTTPClient, infuraRPC)
+		if err != nil {
+			initLogger.Error("failed to get latest block from RPC", "error", err)


What if there is an error here? lastBN will be initialized with -10

aloknerurkar · 2025-10-06T12:15:40Z

+			"lookback", c.Int("backfill-lookback"),
+			"batch", c.Int("backfill-batch"))
+		go backfill.RunAll(ctx, db, httpc, createOptionsFromCLI(c), relays)
+		initLogger.Info("[BACKFILL] completed startup backfill")


As we spoke this log message is wrong. This does not mean the backfill was completed.

aloknerurkar · 2025-10-06T12:16:25Z

+			// Log block details
+			initLogger.Info("[BLOCK] processing block", "block", nextBN, "slot", ei.Slot)
+
+			if ei.Timestamp != nil {


All these log messages are spammy. We should have just 1 log message which prints all the fields.

aloknerurkar · 2025-10-06T12:19:10Z

+	TsMS        *int64
+}
+
+func MustConnect(ctx context.Context, dsn string, maxConns, minConns int) (*DB, error) {


Usually the MustXXX terminology is used when function panics if there is error. If you are returning error this can just be called Connect or NewDB.

aloknerurkar · 2025-10-06T13:59:57Z

+				initLogger.Warn("[BLOCK] not available yet", "block", nextBN, "error", err)
+				continue
+			}
+			fields := []any{


Not sure why all this is required.

The ideal solution to the comment should just be one log message.

log.Info("processing block", "key1", value1, "key2", value2)

aloknerurkar · 2025-10-06T14:00:26Z

+				}
+
+				// final flush
+				if len(batch) > 0 {


If main context is cancelled, this insert will be called with cancelled context.

aloknerurkar · 2025-10-06T17:15:14Z

+	return db, nil
+}
+
+func closeDatabase(db *database.DB, logger *slog.Logger) {


This just feels like its blindly broken up into functions for the sake of refactor. For eg. this function is not needed.

aloknerurkar · 2025-10-06T17:16:06Z

+	}
+}
+
+func handleShutdown(ctx context.Context, db *database.DB, lastBN int64, logger *slog.Logger) error {


This function and saveBlockProgress are doing the exact same thing.

aloknerurkar · 2025-10-06T17:17:55Z

+	initLogger.Info("indexer configuration", "lookback", c.Int("backfill-lookback"), "batch", c.Int("backfill-batch"))
+
+	// Run backfill if configured
+	runBackfillIfConfigured(ctx, c, db, httpc, relays, initLogger)


This is happening in a single routine. The main loop will not start till the backfill is finished. This is not what we were planning.

aloknerurkar · 2025-10-06T19:57:52Z

+func initializeDatabase(ctx context.Context, dbURL string, logger *slog.Logger) (*database.DB, error) {
+	db, err := database.Connect(ctx, dbURL, 20, 5)
+	if err != nil {
+		logger.Error("[DB] connection failed", "error", err)


Not really sure why the log messages need [SOME KEY]. It can just say DB connection failed. What is achieved with this square brackets? Both cases you will have to search for DB.

aloknerurkar · 2025-10-06T19:58:36Z

+
+	logger.Info("[RELAY] loaded active relays", "count", len(relays))
+	for _, r := range relays {
+		logger.Info("[RELAY] relay found", "id", r.ID, "url", r.URL)


Printing 'Relay' twice in the log message makes no sense.

aloknerurkar · 2025-10-06T20:03:42Z

+	logger.Info("[DB] block saved successfully", "block", nextBN)
+
+	processBidsForBlock(ctx, db, httpc, relays, ei, logger)
+	launchAsyncValidatorTasks(ctx, c, db, httpc, ei, beaconBase, logger)


We launch async tasks and save block progress? What if there was an error in the async task?

aloknerurkar · 2025-10-06T20:06:14Z

+
+func launchAsyncValidatorTasks(ctx context.Context, c *cli.Context, db *database.DB, httpc *retryablehttp.Client, ei *beacon.ExecInfo, beaconBase string, logger *slog.Logger) { // Async validator pubkey fetch
+	if ei.ProposerIdx != nil {
+		go func(slot int64, proposerIdx int64) {


So the first go routine is supposed to populate the Validator public key which the second routine is using? What is the point of doing it in separate go routines then? The second one will fail till first one completes. Not sure if I understand this logic.

aloknerurkar · 2025-10-06T20:08:32Z

+
+}
+
+func saveBlockProgress(db *database.DB, blockNum int64, logger *slog.Logger) {


So out of the two functions, you removed 1. Now this function is still here. As I mentioned before, just changing code for the sake of it is not the right way. You need to understand first if you even need the function. If you create smaller functions for everything, the readability is hampered. There is a balance.

aloknerurkar · 2025-10-06T20:11:38Z

+	var slotsForBids []int64
+	var validatorsToCheck []SlotData
+
+	for data := range slotChan {


Better way is to have the slotChan be a channel of some type

type result struct { data SlotData err Error }

This way you need only 1 for loop and you can check

if data.err != nil { // handle the error }

aloknerurkar · 2025-10-06T20:18:10Z

+			logger.Error("opt-in status update failed", "slot", v.Slot, "error", uerr)
+		}
+	}
+	if err := recentBids(ctx, db, httpc, relays, slotsForBids); err != nil {


So again, I am not sure you understood what I meant when I asked you to use channels. The idea was that we do it one by one, so get missing slot, then get the validator info and opted in status for that slot and then get the bids for that slot, then move onto the next slot.

You have just changed this validator check to be sequential but then the bids again use batches. This does not make any sense. Either you do one by one or do batch like you were doing before.

aloknerurkar · 2025-10-07T14:00:38Z

+			}
+		}
+
+		for _, r := range relays {


This for loop should be inside the above for loop no?

aloknerurkar · 2025-10-07T14:04:16Z

+
+}
+
+func launchAsyncValidatorTasks(ctx context.Context, c *cli.Context, db *database.DB, httpc *retryablehttp.Client, ei *beacon.ExecInfo, beaconBase string, logger *slog.Logger) error { // Async validator pubkey fetch


This is no longer async. Naming should be changed.

aloknerurkar · 2025-10-07T14:07:47Z

+	}
+	return v.Elem().Interface()
+}
+func processNextBlock(ctx context.Context, c *cli.Context, db *database.DB, httpc *retryablehttp.Client, relays []relay.Row, infuraRPC, beaconBase string, lastBN int64, logger *slog.Logger) int64 {


So these 3 functions are duplicates.

processNextBlock
processBidsForBlock
launchValidatorTask

The backfill does these exact 3 things. Only difference is that in the forward operation, the block number is obtained by querying the current height and in cast of backfill it is obtained by querying DB for missing slots.

So ideally you could have 1 file with these 3 functions:

ProcessBlockData(BlockNumber)
ProcessRelayBidData(BlockNumber)
ProcessValidatorOptInData(BlockNumber, validators)

Both backfill and forward operations can just use these functions. Only the logic to get block number will be different.

aloknerurkar reviewed Sep 25, 2025

View reviewed changes

Rose Jethani added 4 commits October 3, 2025 23:31

feat(indexer): add builder observer data indexer

faa1319

Fix StarRocks compatibility and migrate to slog

a3cb8f9

fixed db close error

b3ec4d0

fixes

1b503b2

rose2221 force-pushed the feat/builder-observer-indexer branch from 99823e3 to 1b503b2 Compare October 3, 2025 18:04

fixed backfill operations and refactoring

98e511e

rose2221 force-pushed the feat/builder-observer-indexer branch from 7115450 to 98e511e Compare October 6, 2025 11:41

fixed lint issues

18b2f5c

aloknerurkar reviewed Oct 6, 2025

View reviewed changes

fixed formatting issues

51afda4

aloknerurkar reviewed Oct 6, 2025

View reviewed changes

Rose Jethani added 2 commits October 6, 2025 20:40

updated ctx for a func

b9140d2

improved backfill logic andrefactorign of start.go

30cfa45

aloknerurkar reviewed Oct 6, 2025

View reviewed changes

Rose Jethani added 5 commits October 7, 2025 00:38

fixed refactoring

d0c31a5

fixes

d957908

fixed backfill error logging

936ac83

lint error fixes

369f882

lint error fixes

aa9c997

aloknerurkar reviewed Oct 6, 2025

View reviewed changes

Rose Jethani added 3 commits October 7, 2025 14:29

fixed the start.go changes

650b111

fixed the start.go changes

afbe5df

fixed the start.go changes

634be45

aloknerurkar reviewed Oct 7, 2025

View reviewed changes

fixed the naming of the fucntions

83c1a4d

aloknerurkar approved these changes Oct 7, 2025

View reviewed changes

kant777 merged commit e67c6d7 into main Oct 8, 2025
7 of 9 checks passed

kant777 deleted the feat/builder-observer-indexer branch October 8, 2025 20:49

aloknerurkar mentioned this pull request Oct 14, 2025

feat: make relay bid collection optional via CLI flag #828

Merged


		// Validate required configuration

		log.Printf("[INIT] Starting blockchain indexer with StarRocks database")


		}

		func saveBlockProgress(db database.DB, blockNum int64, logger slog.Logger) {


		}

		func launchAsyncValidatorTasks(ctx context.Context, c cli.Context, db database.DB, httpc retryablehttp.Client, ei beacon.ExecInfo, beaconBase string, logger *slog.Logger) error { // Async validator pubkey fetch

Conversation

rose2221 commented Sep 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

feat(indexer): builder-observer data ingestor

Local Setup (StarRocks in Docker)

Schema DDL (Updated for Builder Observer)

Export and Run

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kant777 commented Oct 1, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

rose2221 commented Sep 24, 2025 •

edited

Loading