Skip to content

Commit e887a8b

Browse files
author
kant
committed
feat: indexer indexes blocks and txs of mev-commit chain to any pluggable storage
1 parent 11e33f9 commit e887a8b

12 files changed

Lines changed: 1367 additions & 0 deletions

File tree

‎indexer/README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
# indexer

‎indexer/cmd/indexer.go‎

Lines changed: 391 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,391 @@
1+
package main
2+
3+
import (
4+
"context"
5+
"encoding/hex"
6+
"fmt"
7+
"log/slog"
8+
"math/big"
9+
"time"
10+
11+
"github.com/ethereum/go-ethereum/common"
12+
"github.com/ethereum/go-ethereum/core/types"
13+
"github.com/primev/mev-commit/indexer/pkg/ethclient"
14+
"github.com/primev/mev-commit/indexer/pkg/store"
15+
)
16+
17+
const TimeLayOut = "2006-01-02T15:04:05.000Z"
18+
19+
type Config struct {
20+
EthClient ethclient.EthereumClient
21+
Storage store.Storage
22+
IndexInterval time.Duration
23+
AccountAddresses []string
24+
MinBlocksToFetchAccountAddresses uint
25+
TimeoutToFetchAccountAddresses time.Duration
26+
}
27+
28+
type BlockchainIndexer struct {
29+
ethClient ethclient.EthereumClient
30+
storage store.Storage
31+
forwardBlockChan chan *types.Block
32+
backwardBlockChan chan *types.Block
33+
txChan chan *types.Transaction
34+
indexInterval time.Duration
35+
lastForwardIndexedBlock *big.Int
36+
lastBackwardIndexedBlock *big.Int
37+
logger *slog.Logger
38+
accountAddresses []string
39+
minBlocksToFetchAccountAddresses uint
40+
timeoutToFetchAccountAddresses time.Duration
41+
}
42+
43+
func NewBlockchainIndexer(config Config) *BlockchainIndexer {
44+
return &BlockchainIndexer{
45+
ethClient: config.EthClient,
46+
storage: config.Storage,
47+
forwardBlockChan: make(chan *types.Block, 100),
48+
backwardBlockChan: make(chan *types.Block, 100),
49+
txChan: make(chan *types.Transaction, 100),
50+
indexInterval: config.IndexInterval,
51+
logger: slog.Default(),
52+
accountAddresses: config.AccountAddresses,
53+
minBlocksToFetchAccountAddresses: config.MinBlocksToFetchAccountAddresses,
54+
timeoutToFetchAccountAddresses: config.TimeoutToFetchAccountAddresses,
55+
}
56+
}
57+
58+
func (bi *BlockchainIndexer) Start(ctx context.Context) error {
59+
if err := bi.storage.CreateIndices(ctx); err != nil {
60+
return fmt.Errorf("failed to create indices: %w", err)
61+
}
62+
63+
latestBlockNumber, err := bi.ethClient.BlockNumber(ctx)
64+
bi.logger.Info("latest block number", "block number", latestBlockNumber)
65+
if err != nil {
66+
return fmt.Errorf("failed to get latest block number: %w", err)
67+
}
68+
69+
if err = bi.initializeForwardIndex(ctx, latestBlockNumber.Uint64()); err != nil {
70+
return err
71+
}
72+
73+
if err = bi.initializeBackwardIndex(ctx, latestBlockNumber.Uint64()); err != nil {
74+
return err
75+
}
76+
77+
go bi.fetchForwardBlocks(ctx)
78+
go bi.processForwardBlocks(ctx)
79+
go bi.fetchBackwardBlocks(ctx)
80+
go bi.processBackwardBlocks(ctx)
81+
go bi.IndexAccountBalances(ctx)
82+
83+
// Block the main function indefinitely
84+
select {
85+
case <-ctx.Done():
86+
return ctx.Err()
87+
}
88+
}
89+
90+
func (bi *BlockchainIndexer) initializeForwardIndex(ctx context.Context, latestBlockNumber uint64) error {
91+
lastForwardIndexedBlock, err := bi.storage.GetLastIndexedBlock(ctx, "forward")
92+
if err != nil {
93+
return fmt.Errorf("failed to get last forward indexed block: %w", err)
94+
}
95+
96+
bi.logger.Info("last indexed block", "blockNumber", lastForwardIndexedBlock, "direction", "forward")
97+
98+
if lastForwardIndexedBlock == nil || lastForwardIndexedBlock.Sign() == 0 {
99+
bi.lastForwardIndexedBlock = new(big.Int).SetUint64(latestBlockNumber - 1)
100+
} else {
101+
bi.lastForwardIndexedBlock = lastForwardIndexedBlock
102+
}
103+
104+
return nil
105+
}
106+
107+
func (bi *BlockchainIndexer) initializeBackwardIndex(ctx context.Context, latestBlockNumber uint64) error {
108+
lastBackwardIndexedBlock, err := bi.storage.GetLastIndexedBlock(ctx, "backward")
109+
if err != nil {
110+
return fmt.Errorf("failed to get last backward indexed block: %w", err)
111+
}
112+
113+
bi.logger.Info("last indexed block", "blockNumber", lastBackwardIndexedBlock, "direction", "backward")
114+
115+
if lastBackwardIndexedBlock == nil || lastBackwardIndexedBlock.Sign() == 0 {
116+
bi.lastBackwardIndexedBlock = new(big.Int).SetUint64(latestBlockNumber)
117+
} else {
118+
bi.lastBackwardIndexedBlock = lastBackwardIndexedBlock
119+
}
120+
121+
return nil
122+
}
123+
124+
func (bi *BlockchainIndexer) fetchForwardBlocks(ctx context.Context) {
125+
ticker := time.NewTicker(bi.indexInterval)
126+
defer ticker.Stop()
127+
128+
for {
129+
select {
130+
case <-ctx.Done():
131+
return
132+
case <-ticker.C:
133+
latestBlockNumber, err := bi.ethClient.BlockNumber(ctx)
134+
if err != nil {
135+
bi.logger.Error("failed to get latest block number", "error", err)
136+
continue
137+
}
138+
139+
for blockNum := new(big.Int).Add(bi.lastForwardIndexedBlock, big.NewInt(1)); blockNum.Cmp(latestBlockNumber) <= 0; blockNum.Add(blockNum, big.NewInt(5)) {
140+
endBlockNum := new(big.Int).Add(blockNum, big.NewInt(4))
141+
if endBlockNum.Cmp(latestBlockNumber) > 0 {
142+
endBlockNum.Set(latestBlockNumber)
143+
}
144+
145+
var blockNums []*big.Int
146+
for bn := new(big.Int).Set(blockNum); bn.Cmp(endBlockNum) <= 0; bn.Add(bn, big.NewInt(1)) {
147+
blockNums = append(blockNums, new(big.Int).Set(bn))
148+
}
149+
150+
blocks, err := bi.fetchBlocks(ctx, blockNums)
151+
if err != nil {
152+
bi.logger.Error("failed to fetch blocks", "start", blockNum, "end", endBlockNum, "error", err)
153+
continue
154+
}
155+
156+
for _, block := range blocks {
157+
bi.forwardBlockChan <- block
158+
bi.lastForwardIndexedBlock.Set(block.Number())
159+
}
160+
}
161+
}
162+
}
163+
}
164+
165+
func (bi *BlockchainIndexer) fetchBackwardBlocks(ctx context.Context) {
166+
ticker := time.NewTicker(bi.indexInterval)
167+
defer ticker.Stop()
168+
169+
for {
170+
select {
171+
case <-ctx.Done():
172+
return
173+
case <-ticker.C:
174+
if bi.lastBackwardIndexedBlock.Sign() <= 0 {
175+
return
176+
}
177+
zeroBigNum := big.NewInt(0)
178+
blockNum := new(big.Int).Sub(bi.lastBackwardIndexedBlock, big.NewInt(1))
179+
180+
for i := 0; blockNum.Cmp(zeroBigNum) >= 0; i++ {
181+
endBlockNum := new(big.Int).Sub(blockNum, big.NewInt(4))
182+
if endBlockNum.Cmp(zeroBigNum) < 0 {
183+
endBlockNum.Set(zeroBigNum)
184+
}
185+
186+
var blockNums []*big.Int
187+
for bn := new(big.Int).Set(blockNum); bn.Cmp(endBlockNum) >= 0; bn.Sub(bn, big.NewInt(1)) {
188+
blockNums = append(blockNums, new(big.Int).Set(bn))
189+
}
190+
191+
blocks, err := bi.fetchBlocks(ctx, blockNums)
192+
if err != nil {
193+
bi.logger.Error("failed to fetch blocks", "start", blockNum, "end", endBlockNum, "error", err)
194+
break
195+
}
196+
197+
for _, block := range blocks {
198+
bi.backwardBlockChan <- block
199+
bi.lastBackwardIndexedBlock.Set(block.Number())
200+
if block.Number().Cmp(zeroBigNum) == 0 {
201+
bi.logger.Info("done fetching backward blocks...")
202+
return
203+
}
204+
}
205+
blockNum.Sub(endBlockNum, big.NewInt(1))
206+
}
207+
}
208+
}
209+
}
210+
211+
func (bi *BlockchainIndexer) processForwardBlocks(ctx context.Context) {
212+
for {
213+
select {
214+
case <-ctx.Done():
215+
return
216+
case block := <-bi.forwardBlockChan:
217+
if err := bi.indexBlock(ctx, block); err != nil {
218+
bi.logger.Error("failed to index block", "error", err)
219+
}
220+
if err := bi.indexTransactions(ctx, block); err != nil {
221+
bi.logger.Error("failed to index transactions", "error", err)
222+
}
223+
}
224+
}
225+
}
226+
227+
func (bi *BlockchainIndexer) processBackwardBlocks(ctx context.Context) {
228+
for {
229+
select {
230+
case <-ctx.Done():
231+
return
232+
case block := <-bi.backwardBlockChan:
233+
if err := bi.indexBlock(ctx, block); err != nil {
234+
bi.logger.Error("failed to index block", "error", err)
235+
}
236+
if err := bi.indexTransactions(ctx, block); err != nil {
237+
bi.logger.Error("failed to index transactions", "error", err)
238+
}
239+
if block.Number().Cmp(big.NewInt(0)) == 0 {
240+
bi.logger.Info("done processing backward blocks...")
241+
return
242+
}
243+
}
244+
}
245+
}
246+
247+
func (bi *BlockchainIndexer) IndexAccountBalances(ctx context.Context) {
248+
var blockCounter uint
249+
timer := time.NewTimer(bi.timeoutToFetchAccountAddresses)
250+
defer timer.Stop()
251+
252+
for {
253+
select {
254+
case <-ctx.Done():
255+
return
256+
case block := <-bi.forwardBlockChan:
257+
blockCounter++
258+
if blockCounter >= bi.minBlocksToFetchAccountAddresses {
259+
if err := bi.indexBalances(ctx, block.NumberU64()); err != nil {
260+
return
261+
}
262+
blockCounter = 0
263+
timer.Reset(bi.timeoutToFetchAccountAddresses)
264+
}
265+
case <-timer.C:
266+
if err := bi.indexBalances(ctx, 0); err != nil {
267+
return
268+
}
269+
blockCounter = 0
270+
timer.Reset(bi.timeoutToFetchAccountAddresses)
271+
}
272+
}
273+
}
274+
275+
func (bi *BlockchainIndexer) indexBalances(ctx context.Context, blockNumber uint64) error {
276+
addresses, err := bi.storage.GetAddresses(ctx)
277+
if err != nil {
278+
return err
279+
}
280+
281+
for _, initialAddress := range bi.accountAddresses {
282+
addresses = append(addresses, initialAddress)
283+
}
284+
285+
addrs := make([]common.Address, len(addresses))
286+
for i, address := range addresses {
287+
addrs[i] = common.HexToAddress(address)
288+
}
289+
290+
accBalances, err := bi.ethClient.AccountBalances(ctx, addrs, blockNumber)
291+
if err != nil {
292+
return err
293+
}
294+
295+
return bi.storage.IndexAccountBalances(ctx, accBalances)
296+
}
297+
298+
func (bi *BlockchainIndexer) indexBlock(ctx context.Context, block *types.Block) error {
299+
timestamp := time.UnixMilli(int64(block.Time())).UTC().Format(TimeLayOut)
300+
indexBlock := &store.IndexBlock{
301+
Number: block.NumberU64(),
302+
Hash: block.Hash().Hex(),
303+
ParentHash: block.ParentHash().Hex(),
304+
Root: block.Root().Hex(),
305+
Nonce: block.Nonce(),
306+
Timestamp: timestamp,
307+
Transactions: len(block.Transactions()),
308+
BaseFee: block.BaseFee().Uint64(),
309+
GasLimit: block.GasLimit(),
310+
GasUsed: block.GasUsed(),
311+
Difficulty: block.Difficulty().Uint64(),
312+
ExtraData: hex.EncodeToString(block.Extra()),
313+
}
314+
315+
return bi.storage.IndexBlock(ctx, indexBlock)
316+
}
317+
318+
func (bi *BlockchainIndexer) indexTransactions(ctx context.Context, block *types.Block) error {
319+
var transactions []*store.IndexTransaction
320+
var txHashes []string
321+
322+
for _, tx := range block.Transactions() {
323+
from, err := types.Sender(types.NewCancunSigner(tx.ChainId()), tx)
324+
if err != nil {
325+
return fmt.Errorf("failed to derive sender: %w", err)
326+
}
327+
328+
v, r, s := tx.RawSignatureValues()
329+
timestamp := tx.Time().UTC().Format(TimeLayOut)
330+
transaction := &store.IndexTransaction{
331+
Hash: tx.Hash().Hex(),
332+
From: from.Hex(),
333+
Gas: tx.Gas(),
334+
Nonce: tx.Nonce(),
335+
BlockHash: block.Hash().Hex(),
336+
BlockNumber: block.NumberU64(),
337+
ChainId: tx.ChainId().String(),
338+
V: v.String(),
339+
R: r.String(),
340+
S: s.String(),
341+
Input: hex.EncodeToString(tx.Data()),
342+
Timestamp: timestamp,
343+
}
344+
345+
if tx.To() != nil {
346+
transaction.To = tx.To().Hex()
347+
}
348+
if tx.GasPrice() != nil {
349+
transaction.GasPrice = tx.GasPrice().Uint64()
350+
}
351+
if tx.GasTipCap() != nil {
352+
transaction.GasTipCap = tx.GasTipCap().Uint64()
353+
}
354+
if tx.GasFeeCap() != nil {
355+
transaction.GasFeeCap = tx.GasFeeCap().Uint64()
356+
}
357+
if tx.Value() != nil {
358+
transaction.Value = tx.Value().String()
359+
}
360+
361+
transactions = append(transactions, transaction)
362+
txHashes = append(txHashes, tx.Hash().Hex())
363+
}
364+
365+
receipts, err := bi.fetchReceipts(ctx, txHashes)
366+
if err != nil {
367+
return fmt.Errorf("failed to fetch transaction receipts: %w", err)
368+
}
369+
370+
for _, tx := range transactions {
371+
if receipt, ok := receipts[tx.Hash]; ok {
372+
tx.Status = receipt.Status
373+
tx.GasUsed = receipt.GasUsed
374+
tx.CumulativeGasUsed = receipt.CumulativeGasUsed
375+
tx.ContractAddress = receipt.ContractAddress.Hex()
376+
tx.TransactionIndex = receipt.TransactionIndex
377+
tx.ReceiptBlockHash = receipt.BlockHash.Hex()
378+
tx.ReceiptBlockNumber = receipt.BlockNumber.Uint64()
379+
}
380+
}
381+
382+
return bi.storage.IndexTransactions(ctx, transactions)
383+
}
384+
385+
func (bi *BlockchainIndexer) fetchReceipts(ctx context.Context, txHashes []string) (map[string]*types.Receipt, error) {
386+
return bi.ethClient.TxReceipts(ctx, txHashes)
387+
}
388+
389+
func (bi *BlockchainIndexer) fetchBlocks(ctx context.Context, blockNums []*big.Int) ([]*types.Block, error) {
390+
return bi.ethClient.GetBlocks(ctx, blockNums)
391+
}

0 commit comments

Comments
 (0)