index blob hashes

indexer/database/dump/indexer.go

@@ -273,6 +273,20 @@ func (sdi *StateDiffIndexer) processObjects(tx *BatchTx, args processArgs) error
 			return err
 		}
 
+		if trx.Type() == types.BlobTxType {
+			blobHashes := trx.BlobHashes()
+			for i, hash := range blobHashes {
+				bhModel := models.BlobHashModel{
+					TxHash:   trxID,
+					Index:    uint64(i),
+					BlobHash: hash,
+				}
+				if _, err := fmt.Fprintf(sdi.dump, "%+v\r\n", bhModel); err != nil {
+					return err
+				}
+			}
+		}
+
 		// this is the contract address if this receipt is for a contract creation tx
 		contract := shared.HandleZeroAddr(receipt.ContractAddress)
 

indexer/database/file/csv_writer.go

@@ -254,6 +254,13 @@ func (csw *CSVWriter) upsertTransactionCID(transaction models.TxModel) {
 	metrics.IndexerMetrics.TransactionsCounter.Inc(1)
 }
 
+func (csw *CSVWriter) upsertBlobHash(blobHash models.BlobHashModel) {
+	var values []interface{}
+	values = append(values, blobHash.TxHash, blobHash.Index, blobHash.BlobHash)
+	csw.rows <- tableRow{&schema.TableBlobHash, values}
+	metrics.IndexerMetrics.BlobHashCounter.Inc(1)
+}
+
 func (csw *CSVWriter) upsertReceiptCID(rct *models.ReceiptModel) {
 	var values []interface{}
 	values = append(values, rct.BlockNumber, rct.HeaderID, rct.TxID, rct.CID, rct.Contract,

indexer/database/file/indexer.go

@@ -338,6 +338,17 @@ func (sdi *StateDiffIndexer) processObjects(args processArgs) error {
 		}
 		sdi.fileWriter.upsertTransactionCID(txModel)
 
+		if trx.Type() == types.BlobTxType {
+			blobHashes := trx.BlobHashes()
+			for i, hash := range blobHashes {
+				sdi.fileWriter.upsertBlobHash(models.BlobHashModel{
+					TxHash:   txID,
+					Index:    uint64(i),
+					BlobHash: hash,
+				})
+			}
+		}
+
 		// this is the contract address if this receipt is for a contract creation tx
 		contract := shared.HandleZeroAddr(receipt.ContractAddress)
 

indexer/database/file/interfaces.go

@@ -39,6 +39,7 @@ type FileWriter interface {
 	upsertHeaderCID(header models.HeaderModel)
 	upsertUncleCID(uncle models.UncleModel)
 	upsertTransactionCID(transaction models.TxModel)
+	upsertBlobHash(models.BlobHashModel)
 	upsertReceiptCID(rct *models.ReceiptModel)
 	upsertLogCID(logs []*models.LogsModel)
 	upsertWithdrawalCID(models.WithdrawalModel)

indexer/database/file/sql_writer.go

@@ -170,6 +170,7 @@ const (
 
 var (
 	withdrawalsInsert = schema.TableWithdrawal.FmtStringInsert() + ";\n"
+	blobHashesInsert  = schema.TableBlobHash.FmtStringInsert() + ";\n"
 )
 
 func (sqw *SQLWriter) upsertNode(node nodeinfo.Info) {
@@ -230,6 +231,11 @@ func (sqw *SQLWriter) upsertTransactionCID(transaction models.TxModel) {
 	metrics.IndexerMetrics.TransactionsCounter.Inc(1)
 }
 
+func (sqw *SQLWriter) upsertBlobHash(bh models.BlobHashModel) {
+	sqw.stmts <- []byte(fmt.Sprintf(blobHashesInsert, bh.TxHash, bh.Index, bh.BlobHash))
+	metrics.IndexerMetrics.BlobHashCounter.Inc(1)
+}
+
 func (sqw *SQLWriter) upsertReceiptCID(rct *models.ReceiptModel) {
 	sqw.stmts <- []byte(fmt.Sprintf(rctInsert, rct.BlockNumber, rct.HeaderID, rct.TxID, rct.CID, rct.Contract,
 		rct.PostState, rct.PostStatus))

indexer/database/metrics/metrics.go

@@ -52,6 +52,8 @@ type IndexerMetricsHandles struct {
 	BlocksCounter metrics.Counter
 	// The total number of processed transactions
 	TransactionsCounter metrics.Counter
+	// The total number of indexed blob hashes
+	BlobHashCounter metrics.Counter
 	// The total number of processed receipts
 	ReceiptsCounter metrics.Counter
 	// The total number of processed logs
@@ -90,6 +92,7 @@ func RegisterIndexerMetrics(reg metrics.Registry) IndexerMetricsHandles {
 	ctx := IndexerMetricsHandles{
 		BlocksCounter:                     metrics.NewCounter(),
 		TransactionsCounter:               metrics.NewCounter(),
+		BlobHashCounter:                   metrics.NewCounter(),
 		ReceiptsCounter:                   metrics.NewCounter(),
 		LogsCounter:                       metrics.NewCounter(),
 		WithdrawalsCounter:                metrics.NewCounter(),
@@ -114,6 +117,7 @@ func RegisterIndexerMetrics(reg metrics.Registry) IndexerMetricsHandles {
 	subsys := "indexer"
 	reg.Register(metricName(subsys, "blocks"), ctx.BlocksCounter)
 	reg.Register(metricName(subsys, "transactions"), ctx.TransactionsCounter)
+	reg.Register(metricName(subsys, "blob_hashes"), ctx.BlobHashCounter)
 	reg.Register(metricName(subsys, "receipts"), ctx.ReceiptsCounter)
 	reg.Register(metricName(subsys, "logs"), ctx.LogsCounter)
 	reg.Register(metricName(subsys, "withdrawals"), ctx.WithdrawalsCounter)

indexer/database/sql/indexer.go

@@ -306,6 +306,19 @@ func (sdi *StateDiffIndexer) processObjects(tx *BatchTx, args processArgs) error
 			return err
 		}
 
+		if trx.Type() == types.BlobTxType {
+			blobHashes := trx.BlobHashes()
+			for i, hash := range blobHashes {
+				if err := sdi.dbWriter.upsertBlobHash(tx.dbtx, models.BlobHashModel{
+					TxHash:   txID,
+					Index:    uint64(i),
+					BlobHash: hash,
+				}); err != nil {
+					return err
+				}
+			}
+		}
+
 		// this is the contract address if this receipt is for a contract creation tx
 		contract := shared.HandleZeroAddr(receipt.ContractAddress)
 

indexer/database/sql/interfaces.go

@@ -52,6 +52,7 @@ type Statements interface {
 	SetCanonicalHeaderStm() string
 	InsertUncleStm() string
 	InsertTxStm() string
+	InsertBlobHashStm() string
 	InsertRctStm() string
 	InsertLogStm() string
 	InsertWithdrawalStm() string

indexer/database/sql/postgres/database.go

@@ -81,6 +81,11 @@ func (db *DB) InsertTxStm() string {
 	return schema.TableTransaction.PreparedInsert(db.upsert)
 }
 
+// InsertBlobHashStm satisfies the sql.Statements interface
+func (db *DB) InsertBlobHashStm() string {
+	return schema.TableBlobHash.PreparedInsert(db.upsert)
+}
+
 // InsertRctStm satisfies the sql.Statements interface
 func (db *DB) InsertRctStm() string {
 	return schema.TableReceipt.PreparedInsert(db.upsert)

indexer/database/sql/writer.go

@@ -209,6 +209,30 @@ func (w *Writer) upsertTransactionCID(tx Tx, transaction models.TxModel) error {
 	return nil
 }
 
+/*
+INSERT INTO eth.blob_hashes (tx_hash, index, blob_hash) VALUES ($1, $2, $3)
+ON CONFLICT (tx_hash, index) DO NOTHING
+*/
+func (w *Writer) upsertBlobHash(tx Tx, blobHash models.BlobHashModel) error {
+	if w.useCopyForTx(tx) {
+		rows := toRows(toRow(blobHash.TxHash, blobHash.Index, blobHash.BlobHash))
+		_, err := tx.CopyFrom(w.db.Context(), schema.TableBlobHash.TableName(), schema.TableBlobHash.ColumnNames(), rows)
+		if err != nil {
+			return insertError{"eth.blob_hashes", err, "COPY", blobHash}
+		}
+	} else {
+		_, err := tx.Exec(w.db.Context(), w.db.InsertBlobHashStm(),
+			blobHash.TxHash,
+			blobHash.Index,
+			blobHash.BlobHash,
+		)
+		if err != nil {
+			return insertError{"eth.blob_hashes", err, w.db.InsertBlobHashStm(), blobHash}
+		}
+	}
+	return nil
+}
+
 /*
 INSERT INTO eth.receipt_cids (block_number, header_id, tx_id, cid, contract, post_state, post_status) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
 ON CONFLICT (tx_id, header_id, block_number) DO NOTHING

indexer/mocks/test_data.go

@@ -50,7 +50,7 @@ var (
 	// canonical block at London height
 	// includes 5 transactions: 3 Legacy + 1 EIP-2930 + 1 EIP-1559
 	MockHeader = types.Header{
-		Time:        0,
+		Time:        BlockTime,
 		Number:      new(big.Int).Set(BlockNumber),
 		Root:        common.HexToHash("0x0"),
 		TxHash:      common.HexToHash("0x0"),
@@ -434,13 +434,15 @@ func createLegacyTransactionsAndReceipts(config *params.ChainConfig, blockNumber
 	return types.Transactions{signedTrx1, signedTrx2, signedTrx3}, types.Receipts{mockReceipt1, mockReceipt2, mockReceipt3}, senderAddr
 }
 
-// createTransactionsAndReceipts is a helper function to generate signed mock transactions and mock receipts with mock logs
+// createTransactionsAndReceipts generates signed mock transactions and mock receipts with mock logs, and returns the address of the sender with them.
 func createTransactionsAndReceipts(config *params.ChainConfig, blockNumber *big.Int, blockTime uint64) (types.Transactions, types.Receipts, common.Address) {
+	const txCount = 6
 	// make transactions
-	trx1 := types.NewTransaction(0, Address, big.NewInt(1000), 50, big.NewInt(100), []byte{})
+	txs := make(types.Transactions, txCount)
-	trx2 := types.NewTransaction(1, AnotherAddress, big.NewInt(2000), 100, big.NewInt(200), []byte{})
+	txs[0] = types.NewTransaction(0, Address, big.NewInt(1000), 50, big.NewInt(100), []byte{})
-	trx3 := types.NewContractCreation(2, big.NewInt(1500), 75, big.NewInt(150), MockContractByteCode)
+	txs[1] = types.NewTransaction(1, AnotherAddress, big.NewInt(2000), 100, big.NewInt(200), []byte{})
-	trx4 := types.NewTx(&types.AccessListTx{
+	txs[2] = types.NewContractCreation(2, big.NewInt(1500), 75, big.NewInt(150), MockContractByteCode)
+	txs[3] = types.NewTx(&types.AccessListTx{
 		ChainID:  config.ChainID,
 		Nonce:    0,
 		GasPrice: big.NewInt(100),
@@ -453,7 +455,7 @@ func createTransactionsAndReceipts(config *params.ChainConfig, blockNumber *big.
 			AccessListEntry2,
 		},
 	})
-	trx5 := types.NewTx(&types.DynamicFeeTx{
+	txs[4] = types.NewTx(&types.DynamicFeeTx{
 		ChainID:   config.ChainID,
 		Nonce:     0,
 		GasTipCap: big.NewInt(100),
@@ -467,6 +469,20 @@ func createTransactionsAndReceipts(config *params.ChainConfig, blockNumber *big.
 			AccessListEntry2,
 		},
 	})
+	txs[5] = types.NewTx(&types.BlobTx{
+		ChainID:    uint256.MustFromBig(config.ChainID),
+		Nonce:      0,
+		GasTipCap:  uint256.NewInt(100),
+		GasFeeCap:  uint256.NewInt(100),
+		Gas:        50,
+		To:         AnotherAddress,
+		Value:      uint256.NewInt(0),
+		BlobFeeCap: uint256.NewInt(1e6),
+		BlobHashes: []common.Hash{
+			common.HexToHash("0x0100000000000000000000000000000000000000000000000000000000000001"),
+			common.HexToHash("0x0100000000000000000000000000000000000000000000000000000000000002"),
+		},
+	})
 
 	transactionSigner := types.MakeSigner(config, blockNumber, blockTime)
 	mockCurve := elliptic.P256()
@@ -474,60 +490,56 @@ func createTransactionsAndReceipts(config *params.ChainConfig, blockNumber *big.
 	if err != nil {
 		log.Crit(err.Error())
 	}
-	signedTrx1, err := types.SignTx(trx1, transactionSigner, mockPrvKey)
+	var signedTxs types.Transactions
-	if err != nil {
+	for _, tx := range txs {
-		log.Crit(err.Error())
+		signed, err := types.SignTx(tx, transactionSigner, mockPrvKey)
+		if err != nil {
+			log.Crit(err.Error())
+		}
+		signedTxs = append(signedTxs, signed)
 	}
-	signedTrx2, err := types.SignTx(trx2, transactionSigner, mockPrvKey)
+	senderAddr, err := types.Sender(transactionSigner, signedTxs[0]) // same for both trx
-	if err != nil {
-		log.Crit(err.Error())
-	}
-	signedTrx3, err := types.SignTx(trx3, transactionSigner, mockPrvKey)
-	if err != nil {
-		log.Crit(err.Error())
-	}
-	signedTrx4, err := types.SignTx(trx4, transactionSigner, mockPrvKey)
-	if err != nil {
-		log.Crit(err.Error())
-	}
-	signedTrx5, err := types.SignTx(trx5, transactionSigner, mockPrvKey)
-	if err != nil {
-		log.Crit(err.Error())
-	}
-
-	senderAddr, err := types.Sender(transactionSigner, signedTrx1) // same for both trx
 	if err != nil {
 		log.Crit(err.Error())
 	}
 
 	// make receipts
-	mockReceipt1 := types.NewReceipt(nil, false, 50)
+	receipts := make(types.Receipts, txCount)
-	mockReceipt1.Logs = []*types.Log{MockLog1}
+	receipts[0] = types.NewReceipt(nil, false, 50)
-	mockReceipt1.TxHash = signedTrx1.Hash()
+	receipts[0].Logs = []*types.Log{MockLog1}
-	mockReceipt2 := types.NewReceipt(common.HexToHash("0x1").Bytes(), false, 100)
+	receipts[0].TxHash = signedTxs[0].Hash()
-	mockReceipt2.Logs = []*types.Log{MockLog2, ShortLog1}
+	receipts[1] = types.NewReceipt(common.HexToHash("0x1").Bytes(), false, 100)
-	mockReceipt2.TxHash = signedTrx2.Hash()
+	receipts[1].Logs = []*types.Log{MockLog2, ShortLog1}
-	mockReceipt3 := types.NewReceipt(common.HexToHash("0x2").Bytes(), false, 75)
+	receipts[1].TxHash = signedTxs[1].Hash()
-	mockReceipt3.Logs = []*types.Log{}
+	receipts[2] = types.NewReceipt(common.HexToHash("0x2").Bytes(), false, 75)
-	mockReceipt3.TxHash = signedTrx3.Hash()
+	receipts[2].Logs = []*types.Log{}
-	mockReceipt4 := &types.Receipt{
+	receipts[2].TxHash = signedTxs[2].Hash()
+	receipts[3] = &types.Receipt{
 		Type:              types.AccessListTxType,
 		PostState:         common.HexToHash("0x3").Bytes(),
 		Status:            types.ReceiptStatusSuccessful,
 		CumulativeGasUsed: 175,
 		Logs:              []*types.Log{MockLog3, MockLog4, ShortLog2},
-		TxHash:            signedTrx4.Hash(),
+		TxHash:            signedTxs[3].Hash(),
 	}
-	mockReceipt5 := &types.Receipt{
+	receipts[4] = &types.Receipt{
 		Type:              types.DynamicFeeTxType,
 		PostState:         common.HexToHash("0x3").Bytes(),
 		Status:            types.ReceiptStatusSuccessful,
 		CumulativeGasUsed: 175,
 		Logs:              []*types.Log{},
-		TxHash:            signedTrx5.Hash(),
+		TxHash:            signedTxs[4].Hash(),
+	}
+	receipts[5] = &types.Receipt{
+		Type:              types.BlobTxType,
+		PostState:         common.HexToHash("0x3").Bytes(),
+		Status:            types.ReceiptStatusSuccessful,
+		CumulativeGasUsed: 175,
+		Logs:              []*types.Log{},
+		TxHash:            signedTxs[5].Hash(),
 	}
 
-	return types.Transactions{signedTrx1, signedTrx2, signedTrx3, signedTrx4, signedTrx5}, types.Receipts{mockReceipt1, mockReceipt2, mockReceipt3, mockReceipt4, mockReceipt5}, senderAddr
+	return signedTxs, receipts, senderAddr
 }
 
 // createNonCanonicalBlockReceipts is a helper function to generate mock receipts with mock logs for non-canonical blocks

indexer/models/models.go

@@ -16,7 +16,10 @@
 
 package models
 
-import "github.com/lib/pq"
+import (
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/lib/pq"
+)
 
 // IPLDModel is the db model for ipld.blocks
 type IPLDModel struct {
@@ -130,3 +133,10 @@ type WithdrawalModel struct {
 	Address     string `db:"address"`
 	Amount      uint64 `db:"amount"`
 }
+
+// BlobHashModel is the DB model for eth.blob_hashes
+type BlobHashModel struct {
+	TxHash   string      `db:"tx_hash"`
+	Index    uint64      `db:"index"`
+	BlobHash common.Hash `db:"blob_hash"`
+}

indexer/shared/schema/schema.go

@@ -149,6 +149,15 @@ var TableTransaction = Table{
 	UpsertClause: OnConflict("block_number", "header_id", "tx_hash"),
 }
 
+var TableBlobHash = Table{
+	Name: "eth.blob_hashes",
+	Columns: []Column{
+		{Name: "tx_hash", Type: Dvarchar},
+		{Name: "index", Type: Dinteger},
+		{Name: "blob_hash", Type: Dbytea},
+	},
+}
+
 var TableReceipt = Table{
 	Name: "eth.receipt_cids",
 	Columns: []Column{

indexer/test/test.go

@@ -109,16 +109,14 @@ func DoTestPublishAndIndexTransactionIPLDs(t *testing.T, db sql.Database) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	require.Equal(t, 5, len(trxs))
+	require.Equal(t, len(txCIDs), len(trxs))
-	require.Contains(t, trxs, trx1CID.String())
+	for _, c := range txCIDs {
-	require.Contains(t, trxs, trx2CID.String())
+		require.Contains(t, trxs, c.String())
-	require.Contains(t, trxs, trx3CID.String())
+	}
-	require.Contains(t, trxs, trx4CID.String())
-	require.Contains(t, trxs, trx5CID.String())
 
 	transactions := mocks.MockBlock.Transactions()
 	type txResult struct {
-		TxType uint8 `db:"tx_type"`
+		TxType int `db:"tx_type"`
 		Value  string
 	}
 	for _, c := range trxs {
@@ -132,9 +130,10 @@ func DoTestPublishAndIndexTransactionIPLDs(t *testing.T, db sql.Database) {
 			t.Fatal(err)
 		}
 		txTypeAndValueStr := `SELECT tx_type, CAST(value as TEXT) FROM eth.transaction_cids WHERE cid = $1`
+		txBlobHashQuery := `SELECT blob_hash FROM eth.blob_hashes WHERE tx_hash = $1`
 		switch c {
-		case trx1CID.String():
+		case txCIDs[0].String():
-			require.Equal(t, tx1, data)
+			require.Equal(t, encodedTxs[0], data)
 			txRes := new(txResult)
 			err = db.QueryRow(context.Background(), txTypeAndValueStr, c).Scan(&txRes.TxType, &txRes.Value)
 			if err != nil {
@@ -146,8 +145,8 @@ func DoTestPublishAndIndexTransactionIPLDs(t *testing.T, db sql.Database) {
 			if txRes.Value != transactions[0].Value().String() {
 				t.Fatalf("expected tx value %s got %s", transactions[0].Value().String(), txRes.Value)
 			}
-		case trx2CID.String():
+		case txCIDs[1].String():
-			require.Equal(t, tx2, data)
+			require.Equal(t, encodedTxs[1], data)
 			txRes := new(txResult)
 			err = db.QueryRow(context.Background(), txTypeAndValueStr, c).Scan(&txRes.TxType, &txRes.Value)
 			if err != nil {
@@ -159,8 +158,8 @@ func DoTestPublishAndIndexTransactionIPLDs(t *testing.T, db sql.Database) {
 			if txRes.Value != transactions[1].Value().String() {
 				t.Fatalf("expected tx value %s got %s", transactions[1].Value().String(), txRes.Value)
 			}
-		case trx3CID.String():
+		case txCIDs[2].String():
-			require.Equal(t, tx3, data)
+			require.Equal(t, encodedTxs[2], data)
 			txRes := new(txResult)
 			err = db.QueryRow(context.Background(), txTypeAndValueStr, c).Scan(&txRes.TxType, &txRes.Value)
 			if err != nil {
@@ -172,8 +171,8 @@ func DoTestPublishAndIndexTransactionIPLDs(t *testing.T, db sql.Database) {
 			if txRes.Value != transactions[2].Value().String() {
 				t.Fatalf("expected tx value %s got %s", transactions[2].Value().String(), txRes.Value)
 			}
-		case trx4CID.String():
+		case txCIDs[3].String():
-			require.Equal(t, tx4, data)
+			require.Equal(t, encodedTxs[3], data)
 			txRes := new(txResult)
 			err = db.QueryRow(context.Background(), txTypeAndValueStr, c).Scan(&txRes.TxType, &txRes.Value)
 			if err != nil {
@@ -185,8 +184,8 @@ func DoTestPublishAndIndexTransactionIPLDs(t *testing.T, db sql.Database) {
 			if txRes.Value != transactions[3].Value().String() {
 				t.Fatalf("expected tx value %s got %s", transactions[3].Value().String(), txRes.Value)
 			}
-		case trx5CID.String():
+		case txCIDs[4].String():
-			require.Equal(t, tx5, data)
+			require.Equal(t, encodedTxs[4], data)
 			txRes := new(txResult)
 			err = db.QueryRow(context.Background(), txTypeAndValueStr, c).Scan(&txRes.TxType, &txRes.Value)
 			if err != nil {
@@ -198,6 +197,21 @@ func DoTestPublishAndIndexTransactionIPLDs(t *testing.T, db sql.Database) {
 			if txRes.Value != transactions[4].Value().String() {
 				t.Fatalf("expected tx value %s got %s", transactions[4].Value().String(), txRes.Value)
 			}
+		case txCIDs[5].String():
+			require.Equal(t, encodedTxs[5], data)
+			var txRes txResult
+			err = db.QueryRow(context.Background(), txTypeAndValueStr, c).Scan(&txRes.TxType, &txRes.Value)
+			if err != nil {
+				t.Fatal(err)
+			}
+			require.Equal(t, types.BlobTxType, txRes.TxType)
+			require.Equal(t, transactions[5].Value().String(), txRes.Value)
+			var txBlobHashes []common.Hash
+			err = db.Select(context.Background(), &txBlobHashes, txBlobHashQuery, transactions[5].Hash().String())
+			if err != nil {
+				t.Fatal(err)
+			}
+			require.Equal(t, transactions[5].BlobHashes(), txBlobHashes)
 		}
 	}
 }
@@ -255,12 +269,10 @@ func DoTestPublishAndIndexReceiptIPLDs(t *testing.T, db sql.Database) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	require.Equal(t, 5, len(rcts))
+	require.Equal(t, len(rctCIDs), len(rcts))
-	require.Contains(t, rcts, rct1CID.String())
+	for _, c := range rctCIDs {
-	require.Contains(t, rcts, rct2CID.String())
+		require.Contains(t, rcts, c.String())
-	require.Contains(t, rcts, rct3CID.String())
+	}
-	require.Contains(t, rcts, rct4CID.String())
-	require.Contains(t, rcts, rct5CID.String())
 
 	for idx, c := range rcts {
 		result := make([]models.IPLDModel, 0)
@@ -289,8 +301,8 @@ func DoTestPublishAndIndexReceiptIPLDs(t *testing.T, db sql.Database) {
 
 		postStatePgStr := `SELECT post_state FROM eth.receipt_cids WHERE cid = $1`
 		switch c {
-		case rct1CID.String():
+		case rctCIDs[0].String():
-			require.Equal(t, rct1, data)
+			require.Equal(t, encodedRcts[0], data)
 			var postStatus uint64
 			pgStr = `SELECT post_status FROM eth.receipt_cids WHERE cid = $1`
 			err = db.Get(context.Background(), &postStatus, pgStr, c)
@@ -298,32 +310,40 @@ func DoTestPublishAndIndexReceiptIPLDs(t *testing.T, db sql.Database) {
 				t.Fatal(err)
 			}
 			require.Equal(t, mocks.ExpectedPostStatus, postStatus)
-		case rct2CID.String():
+		case rctCIDs[1].String():
-			require.Equal(t, rct2, data)
+			require.Equal(t, encodedRcts[1], data)
 			var postState string
 			err = db.Get(context.Background(), &postState, postStatePgStr, c)
 			if err != nil {
 				t.Fatal(err)
 			}
 			require.Equal(t, mocks.ExpectedPostState1, postState)
-		case rct3CID.String():
+		case rctCIDs[2].String():
-			require.Equal(t, rct3, data)
+			require.Equal(t, encodedRcts[2], data)
 			var postState string
 			err = db.Get(context.Background(), &postState, postStatePgStr, c)
 			if err != nil {
 				t.Fatal(err)
 			}
 			require.Equal(t, mocks.ExpectedPostState2, postState)
-		case rct4CID.String():
+		case rctCIDs[3].String():
-			require.Equal(t, rct4, data)
+			require.Equal(t, encodedRcts[3], data)
 			var postState string
 			err = db.Get(context.Background(), &postState, postStatePgStr, c)
 			if err != nil {
 				t.Fatal(err)
 			}
 			require.Equal(t, mocks.ExpectedPostState3, postState)
-		case rct5CID.String():
+		case rctCIDs[4].String():
-			require.Equal(t, rct5, data)
+			require.Equal(t, encodedRcts[4], data)
+			var postState string
+			err = db.Get(context.Background(), &postState, postStatePgStr, c)
+			if err != nil {
+				t.Fatal(err)
+			}
+			require.Equal(t, mocks.ExpectedPostState3, postState)
+		case rctCIDs[5].String():
+			require.Equal(t, encodedRcts[5], data)
 			var postState string
 			err = db.Get(context.Background(), &postState, postStatePgStr, c)
 			if err != nil {
@@ -723,62 +743,19 @@ func DoTestPublishAndIndexTransactionsNonCanonical(t *testing.T, db sql.Database
 
 	// expected transactions in the canonical block
 	mockBlockTxs := mocks.MockBlock.Transactions()
-	expectedBlockTxs := []models.TxModel{
+	expectedBlockTxs := make([]models.TxModel, len(mockBlockTxs))
-		{
+	for i, tx := range mockBlockTxs {
+		expectedBlockTxs[i] = models.TxModel{
 			BlockNumber: mockBlock.Number().String(),
 			HeaderID:    mockBlock.Hash().String(),
-			TxHash:      mockBlockTxs[0].Hash().String(),
+			TxHash:      tx.Hash().String(),
-			CID:         trx1CID.String(),
+			CID:         txCIDs[i].String(),
-			Dst:         shared.HandleZeroAddrPointer(mockBlockTxs[0].To()),
+			Dst:         shared.HandleZeroAddrPointer(tx.To()),
 			Src:         mocks.SenderAddr.String(),
-			Index:       0,
+			Index:       int64(i),
-			Type:        mockBlockTxs[0].Type(),
+			Type:        tx.Type(),
-			Value:       mockBlockTxs[0].Value().String(),
+			Value:       tx.Value().String(),
-		},
+		}
-		{
-			BlockNumber: mockBlock.Number().String(),
-			HeaderID:    mockBlock.Hash().String(),
-			TxHash:      mockBlockTxs[1].Hash().String(),
-			CID:         trx2CID.String(),
-			Dst:         shared.HandleZeroAddrPointer(mockBlockTxs[1].To()),
-			Src:         mocks.SenderAddr.String(),
-			Index:       1,
-			Type:        mockBlockTxs[1].Type(),
-			Value:       mockBlockTxs[1].Value().String(),
-		},
-		{
-			BlockNumber: mockBlock.Number().String(),
-			HeaderID:    mockBlock.Hash().String(),
-			TxHash:      mockBlockTxs[2].Hash().String(),
-			CID:         trx3CID.String(),
-			Dst:         shared.HandleZeroAddrPointer(mockBlockTxs[2].To()),
-			Src:         mocks.SenderAddr.String(),
-			Index:       2,
-			Type:        mockBlockTxs[2].Type(),
-			Value:       mockBlockTxs[2].Value().String(),
-		},
-		{
-			BlockNumber: mockBlock.Number().String(),
-			HeaderID:    mockBlock.Hash().String(),
-			TxHash:      mockBlockTxs[3].Hash().String(),
-			CID:         trx4CID.String(),
-			Dst:         shared.HandleZeroAddrPointer(mockBlockTxs[3].To()),
-			Src:         mocks.SenderAddr.String(),
-			Index:       3,
-			Type:        mockBlockTxs[3].Type(),
-			Value:       mockBlockTxs[3].Value().String(),
-		},
-		{
-			BlockNumber: mockBlock.Number().String(),
-			HeaderID:    mockBlock.Hash().String(),
-			TxHash:      mockBlockTxs[4].Hash().String(),
-			CID:         trx5CID.String(),
-			Dst:         shared.HandleZeroAddrPointer(mockBlockTxs[4].To()),
-			Src:         mocks.SenderAddr.String(),
-			Index:       4,
-			Type:        mockBlockTxs[4].Type(),
-			Value:       mockBlockTxs[4].Value().String(),
-		},
 	}
 
 	// expected transactions in the non-canonical block at London height
@@ -788,7 +765,7 @@ func DoTestPublishAndIndexTransactionsNonCanonical(t *testing.T, db sql.Database
 			BlockNumber: mockNonCanonicalBlock.Number().String(),
 			HeaderID:    mockNonCanonicalBlock.Hash().String(),
 			TxHash:      mockNonCanonicalBlockTxs[0].Hash().String(),
-			CID:         trx2CID.String(),
+			CID:         txCIDs[1].String(),
 			Dst:         mockNonCanonicalBlockTxs[0].To().String(),
 			Src:         mocks.SenderAddr.String(),
 			Index:       0,
@@ -799,7 +776,7 @@ func DoTestPublishAndIndexTransactionsNonCanonical(t *testing.T, db sql.Database
 			BlockNumber: mockNonCanonicalBlock.Number().String(),
 			HeaderID:    mockNonCanonicalBlock.Hash().String(),
 			TxHash:      mockNonCanonicalBlockTxs[1].Hash().String(),
-			CID:         trx5CID.String(),
+			CID:         txCIDs[4].String(),
 			Dst:         mockNonCanonicalBlockTxs[1].To().String(),
 			Src:         mocks.SenderAddr.String(),
 			Index:       1,
@@ -815,7 +792,7 @@ func DoTestPublishAndIndexTransactionsNonCanonical(t *testing.T, db sql.Database
 			BlockNumber: mockNonCanonicalBlock2.Number().String(),
 			HeaderID:    mockNonCanonicalBlock2.Hash().String(),
 			TxHash:      mockNonCanonicalBlock2Txs[0].Hash().String(),
-			CID:         trx3CID.String(),
+			CID:         txCIDs[2].String(),
 			Dst:         "",
 			Src:         mocks.SenderAddr.String(),
 			Index:       0,
@@ -826,7 +803,7 @@ func DoTestPublishAndIndexTransactionsNonCanonical(t *testing.T, db sql.Database
 			BlockNumber: mockNonCanonicalBlock2.Number().String(),
 			HeaderID:    mockNonCanonicalBlock2.Hash().String(),
 			TxHash:      mockNonCanonicalBlock2Txs[1].Hash().String(),
-			CID:         trx5CID.String(),
+			CID:         txCIDs[4].String(),
 			Dst:         mockNonCanonicalBlock2Txs[1].To().String(),
 			Src:         mocks.SenderAddr.String(),
 			Index:       1,
@@ -863,14 +840,12 @@ func DoTestPublishAndIndexTransactionsNonCanonical(t *testing.T, db sql.Database
 	// check indexed IPLD blocks
 	var data []byte
 
-	txCIDs := []cid.Cid{trx1CID, trx2CID, trx3CID, trx4CID, trx5CID}
-	txRLPs := [][]byte{tx1, tx2, tx3, tx4, tx5}
 	for i, txCID := range txCIDs {
 		err = db.Get(context.Background(), &data, ipfsPgGet, txCID.String(), mocks.BlockNumber.Uint64())
 		if err != nil {
 			t.Fatal(err)
 		}
-		require.Equal(t, txRLPs[i], data)
+		require.Equal(t, encodedTxs[i], data)
 	}
 }
 
@@ -886,11 +861,10 @@ func DoTestPublishAndIndexReceiptsNonCanonical(t *testing.T, db sql.Database) {
 	}
 
 	// expected receipts in the canonical block
-	rctCids := []cid.Cid{rct1CID, rct2CID, rct3CID, rct4CID, rct5CID}
 	expectedBlockRctsMap := make(map[string]models.ReceiptModel, len(mocks.MockReceipts))
 	for i, mockBlockRct := range mocks.MockReceipts {
-		rctModel := createRctModel(mockBlockRct, rctCids[i], mockBlock.Number().String())
+		rctModel := createRctModel(mockBlockRct, rctCIDs[i], mockBlock.Number().String())
-		expectedBlockRctsMap[rctCids[i].String()] = rctModel
+		expectedBlockRctsMap[rctCIDs[i].String()] = rctModel
 	}
 
 	// expected receipts in the non-canonical block at London height
@@ -945,10 +919,8 @@ func DoTestPublishAndIndexReceiptsNonCanonical(t *testing.T, db sql.Database) {
 	// check indexed rct IPLD blocks
 	var data []byte
 
-	rctRLPs := [][]byte{
+	rctRLPs := append(encodedRcts, nonCanonicalBlockRct1, nonCanonicalBlockRct2)
-		rct1, rct2, rct3, rct4, rct5, nonCanonicalBlockRct1, nonCanonicalBlockRct2,
+	for i, rctCid := range append(rctCIDs, nonCanonicalBlockRctCids...) {
-	}
-	for i, rctCid := range append(rctCids, nonCanonicalBlockRctCids...) {
 		err = db.Get(context.Background(), &data, ipfsPgGet, rctCid.String(), mocks.BlockNumber.Uint64())
 		if err != nil {
 			t.Fatal(err)

indexer/test/test_init.go

@@ -37,14 +37,13 @@ var (
 					WHERE key = $1 AND block_number = $2`
 	watchedAddressesPgGet = `SELECT *
 					FROM eth_meta.watched_addresses`
-	tx1, tx2, tx3, tx4, tx5, rct1, rct2, rct3, rct4, rct5            []byte
+	encodedTxs, encodedRcts                                          [][]byte
 	wd1, wd2                                                         []byte
 	nonCanonicalBlockRct1, nonCanonicalBlockRct2                     []byte
 	nonCanonicalBlock2Rct1, nonCanonicalBlock2Rct2                   []byte
 	mockBlock, mockNonCanonicalBlock, mockNonCanonicalBlock2         *types.Block
 	headerCID, mockNonCanonicalHeaderCID, mockNonCanonicalHeader2CID cid.Cid
-	trx1CID, trx2CID, trx3CID, trx4CID, trx5CID                      cid.Cid
+	txCIDs, rctCIDs                                                  []cid.Cid
-	rct1CID, rct2CID, rct3CID, rct4CID, rct5CID                      cid.Cid
 	wd1CID, wd2CID                                                   cid.Cid
 	nonCanonicalBlockRct1CID, nonCanonicalBlockRct2CID               cid.Cid
 	nonCanonicalBlock2Rct1CID, nonCanonicalBlock2Rct2CID             cid.Cid
@@ -64,60 +63,28 @@ func init() {
 	mockNonCanonicalBlock2 = mocks.MockNonCanonicalBlock2
 	nonCanonicalBlock2Rcts := mocks.MockNonCanonicalBlock2Receipts
 
-	// encode mock receipts
+	// encode mock txs and receipts
 	buf := new(bytes.Buffer)
-	txs.EncodeIndex(0, buf)
+	encodedTxs = make([][]byte, len(txs))
-	tx1 = make([]byte, buf.Len())
+	encodedRcts = make([][]byte, len(rcts))
-	copy(tx1, buf.Bytes())
-	buf.Reset()
 
-	txs.EncodeIndex(1, buf)
+	for i := 0; i < len(txs); i++ {
-	tx2 = make([]byte, buf.Len())
+		txs.EncodeIndex(i, buf)
-	copy(tx2, buf.Bytes())
+		tx := make([]byte, buf.Len())
-	buf.Reset()
+		copy(tx, buf.Bytes())
+		buf.Reset()
+		encodedTxs[i] = tx
+	}
 
-	txs.EncodeIndex(2, buf)
+	for i := 0; i < len(rcts); i++ {
-	tx3 = make([]byte, buf.Len())
+		rcts.EncodeIndex(i, buf)
-	copy(tx3, buf.Bytes())
+		rct := make([]byte, buf.Len())
-	buf.Reset()
+		copy(rct, buf.Bytes())
-
+		buf.Reset()
-	txs.EncodeIndex(3, buf)
+		encodedRcts[i] = rct
-	tx4 = make([]byte, buf.Len())
+	}
-	copy(tx4, buf.Bytes())
-	buf.Reset()
-
-	txs.EncodeIndex(4, buf)
-	tx5 = make([]byte, buf.Len())
-	copy(tx5, buf.Bytes())
-	buf.Reset()
-
-	rcts.EncodeIndex(0, buf)
-	rct1 = make([]byte, buf.Len())
-	copy(rct1, buf.Bytes())
-	buf.Reset()
-
-	rcts.EncodeIndex(1, buf)
-	rct2 = make([]byte, buf.Len())
-	copy(rct2, buf.Bytes())
-	buf.Reset()
-
-	rcts.EncodeIndex(2, buf)
-	rct3 = make([]byte, buf.Len())
-	copy(rct3, buf.Bytes())
-	buf.Reset()
-
-	rcts.EncodeIndex(3, buf)
-	rct4 = make([]byte, buf.Len())
-	copy(rct4, buf.Bytes())
-	buf.Reset()
-
-	rcts.EncodeIndex(4, buf)
-	rct5 = make([]byte, buf.Len())
-	copy(rct5, buf.Bytes())
-	buf.Reset()
 
 	// encode mock withdrawals
-	// wds
 	mocks.MockWithdrawals.EncodeIndex(0, buf)
 	wd1 = make([]byte, buf.Len())
 	copy(wd1, buf.Bytes())
@@ -152,19 +119,20 @@ func init() {
 	headerCID, _ = ipld.RawdataToCid(ipld.MEthHeader, mocks.MockHeaderRlp, multihash.KECCAK_256)
 	mockNonCanonicalHeaderCID, _ = ipld.RawdataToCid(ipld.MEthHeader, mocks.MockNonCanonicalHeaderRlp, multihash.KECCAK_256)
 	mockNonCanonicalHeader2CID, _ = ipld.RawdataToCid(ipld.MEthHeader, mocks.MockNonCanonicalHeader2Rlp, multihash.KECCAK_256)
-	trx1CID, _ = ipld.RawdataToCid(ipld.MEthTx, tx1, multihash.KECCAK_256)
+
-	trx2CID, _ = ipld.RawdataToCid(ipld.MEthTx, tx2, multihash.KECCAK_256)
+	for i := 0; i < len(txs); i++ {
-	trx3CID, _ = ipld.RawdataToCid(ipld.MEthTx, tx3, multihash.KECCAK_256)
+		tx, _ := ipld.RawdataToCid(ipld.MEthTx, encodedTxs[i], multihash.KECCAK_256)
-	trx4CID, _ = ipld.RawdataToCid(ipld.MEthTx, tx4, multihash.KECCAK_256)
+		txCIDs = append(txCIDs, tx)
-	trx5CID, _ = ipld.RawdataToCid(ipld.MEthTx, tx5, multihash.KECCAK_256)
+	}
+
 	state1CID, _ = ipld.RawdataToCid(ipld.MEthStateTrie, mocks.ContractLeafNode, multihash.KECCAK_256)
 	state2CID, _ = ipld.RawdataToCid(ipld.MEthStateTrie, mocks.AccountLeafNode, multihash.KECCAK_256)
 	storageCID, _ = ipld.RawdataToCid(ipld.MEthStorageTrie, mocks.StorageLeafNode, multihash.KECCAK_256)
-	rct1CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, rct1, multihash.KECCAK_256)
+
-	rct2CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, rct2, multihash.KECCAK_256)
+	for i := 0; i < len(rcts); i++ {
-	rct3CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, rct3, multihash.KECCAK_256)
+		rct, _ := ipld.RawdataToCid(ipld.MEthTxReceipt, encodedRcts[i], multihash.KECCAK_256)
-	rct4CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, rct4, multihash.KECCAK_256)
+		rctCIDs = append(rctCIDs, rct)
-	rct5CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, rct5, multihash.KECCAK_256)
+	}
 
 	wd1CID, _ = ipld.RawdataToCid(ipld.MEthWithdrawal, wd1, multihash.KECCAK_256)
 	wd2CID, _ = ipld.RawdataToCid(ipld.MEthWithdrawal, wd2, multihash.KECCAK_256)

indexer/test_helpers/test_helpers.go

@@ -25,6 +25,21 @@ import (
 	"github.com/cerc-io/plugeth-statediff/indexer/database/sql"
 )
 
+var TruncateStatements = []string{
+	`TRUNCATE nodes`,
+	`TRUNCATE ipld.blocks`,
+	`TRUNCATE eth.header_cids`,
+	`TRUNCATE eth.uncle_cids`,
+	`TRUNCATE eth.transaction_cids`,
+	`TRUNCATE eth.blob_hashes`,
+	`TRUNCATE eth.receipt_cids`,
+	`TRUNCATE eth.state_cids`,
+	`TRUNCATE eth.storage_cids`,
+	`TRUNCATE eth.log_cids`,
+	`TRUNCATE eth.withdrawal_cids`,
+	`TRUNCATE eth_meta.watched_addresses`,
+}
+
 // DedupFile removes duplicates from the given file
 func DedupFile(filePath string) error {
 	f, err := os.OpenFile(filePath, os.O_CREATE|os.O_RDONLY, os.ModePerm)
@@ -39,10 +54,6 @@ func DedupFile(filePath string) error {
 		s := sc.Text()
 		stmts[s] = struct{}{}
 	}
-	if err != nil {
-		return err
-	}
-
 	f.Close()
 
 	f, err = os.Create(filePath)
@@ -60,31 +71,23 @@ func DedupFile(filePath string) error {
 
 // TearDownDB is used to tear down the watcher dbs after tests
 func TearDownDB(t *testing.T, db sql.Database) {
-	ctx := context.Background()
+	err := ClearDB(db)
-	tx, err := db.Begin(ctx)
 	if err != nil {
 		t.Fatal(err)
 	}
+}
 
-	statements := []string{
+func ClearDB(db sql.Database) error {
-		`TRUNCATE nodes`,
+	ctx := context.Background()
-		`TRUNCATE ipld.blocks`,
+	tx, err := db.Begin(ctx)
-		`TRUNCATE eth.header_cids`,
+	if err != nil {
-		`TRUNCATE eth.uncle_cids`,
+		return err
-		`TRUNCATE eth.transaction_cids`,
-		`TRUNCATE eth.receipt_cids`,
-		`TRUNCATE eth.state_cids`,
-		`TRUNCATE eth.storage_cids`,
-		`TRUNCATE eth.log_cids`,
-		`TRUNCATE eth.withdrawal_cids`,
-		`TRUNCATE eth_meta.watched_addresses`,
 	}
-	for _, stm := range statements {
+
+	for _, stm := range TruncateStatements {
 		if _, err = tx.Exec(ctx, stm); err != nil {
-			t.Fatal(err)
+			return err
 		}
 	}
-	if err = tx.Commit(ctx); err != nil {
+	return tx.Commit(ctx)
-		t.Fatal(err)
-	}
 }

test_helpers/chaingen/gen.go

@@ -11,8 +11,10 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/crypto/kzg4844"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/params"
+	"github.com/holiman/uint256"
 )
 
 const secondsPerBlock = 12
@@ -148,3 +150,58 @@ func (gen *GenContext) createTx(from common.Address, to *common.Address, amount
 	}
 	return types.SignTx(tx, signer, priv)
 }
+
+func (gen *GenContext) createBlobTx(
+	from common.Address,
+	to common.Address,
+	amount *uint256.Int,
+	gasLimit uint64,
+	blobData []byte,
+) (*types.Transaction, error) {
+	signer := types.MakeSigner(gen.ChainConfig, gen.block.Number(), gen.time)
+	nonce := gen.block.TxNonce(from)
+	priv, ok := gen.Keys[from]
+	if !ok {
+		return nil, errors.New("No private key for sender address" + from.String())
+	}
+
+	if !gen.ChainConfig.IsCancun(gen.block.Number(), gen.time) {
+		return nil, errors.New("blob tx is only supported from Cancun fork")
+	}
+
+	sidecar := MakeSidecar([][]byte{blobData})
+	tx := types.NewTx(&types.BlobTx{
+		ChainID:    uint256.MustFromBig(gen.ChainConfig.ChainID),
+		Nonce:      nonce,
+		To:         to,
+		Gas:        gasLimit,
+		GasTipCap:  uint256.NewInt(50),
+		GasFeeCap:  uint256.NewInt(1000000000),
+		Value:      amount,
+		BlobFeeCap: uint256.NewInt(1000000),
+		BlobHashes: sidecar.BlobHashes(),
+		Sidecar:    sidecar,
+	})
+	return types.SignTx(tx, signer, priv)
+}
+
+// From go-ethereum/cmd/devp2p/internal/ethtest/chain.go
+func MakeSidecar(data [][]byte) *types.BlobTxSidecar {
+	var (
+		blobs       = make([]kzg4844.Blob, len(data))
+		commitments []kzg4844.Commitment
+		proofs      []kzg4844.Proof
+	)
+	for i := range blobs {
+		copy(blobs[i][:], data[i])
+		c, _ := kzg4844.BlobToCommitment(blobs[i])
+		p, _ := kzg4844.ComputeBlobProof(blobs[i], c)
+		commitments = append(commitments, c)
+		proofs = append(proofs, p)
+	}
+	return &types.BlobTxSidecar{
+		Blobs:       blobs,
+		Commitments: commitments,
+		Proofs:      proofs,
+	}
+}

test_helpers/db.go

@@ -3,6 +3,7 @@ package test_helpers
 import (
 	"fmt"
 
+	"github.com/cerc-io/plugeth-statediff/indexer/test_helpers"
 	"github.com/jmoiron/sqlx"
 )
 
@@ -12,20 +13,7 @@ func ClearDB(db *sqlx.DB) error {
 	if err != nil {
 		return err
 	}
-	statements := []string{
+	for _, stm := range test_helpers.TruncateStatements {
-		`TRUNCATE nodes`,
-		`TRUNCATE ipld.blocks`,
-		`TRUNCATE eth.header_cids`,
-		`TRUNCATE eth.uncle_cids`,
-		`TRUNCATE eth.transaction_cids`,
-		`TRUNCATE eth.receipt_cids`,
-		`TRUNCATE eth.state_cids`,
-		`TRUNCATE eth.storage_cids`,
-		`TRUNCATE eth.log_cids`,
-		`TRUNCATE eth.withdrawal_cids`,
-		`TRUNCATE eth_meta.watched_addresses`,
-	}
-	for _, stm := range statements {
 		if _, err = tx.Exec(stm); err != nil {
 			return fmt.Errorf("error executing `%s`: %w", stm, err)
 		}