Index Withdrawal objects (#25)

Indexes the new validator withdrawal objects (from Shanghai/Capella fork: https://eips.ethereum.org/EIPS/eip-4895)
- new table `eth.withdrawal_cids`
- new column `withdrawals_root` in `eth.header_cids`
- add unit tests
- use new external stack repo in CI/CT job (cerc-io/fixturenet-eth-stacks#14)

Reviewed-on: #25

indexer/database/sql/indexer.go

@@ -105,7 +105,7 @@ func (sdi *StateDiffIndexer) PushBlock(block *types.Block, receipts types.Receip
 	}
 
 	// Generate the block iplds
-	txNodes, rctNodes, logNodes, err := ipld.FromBlockAndReceipts(block, receipts)
+	txNodes, rctNodes, logNodes, wdNodes, err := ipld.FromBlockAndReceipts(block, receipts)
 	if err != nil {
 		return nil, fmt.Errorf("error creating IPLD nodes from block and receipts: %w", err)
 	}
@@ -148,16 +148,18 @@ func (sdi *StateDiffIndexer) PushBlock(block *types.Block, receipts types.Receip
 	}
 	metrics2.IndexerMetrics.UncleProcessingTimer.Update(time.Since(t))
 	t = time.Now()
-	// Publish and index receipts and txs
-	err = sdi.processReceiptsAndTxs(batch, processArgs{
+
+	err = sdi.processObjects(batch, processArgs{
 		headerID:    headerID,
 		blockNumber: block.Number(),
 		blockTime:   block.Time(),
 		receipts:    receipts,
+		withdrawals: block.Withdrawals(),
 		txs:         transactions,
 		rctNodes:    rctNodes,
 		txNodes:     txNodes,
 		logNodes:    logNodes,
+		wdNodes:     wdNodes,
 	})
 	if err != nil {
 		return nil, err
@@ -185,7 +187,7 @@ func (sdi *StateDiffIndexer) PushHeader(batch interfaces.Batch, header *types.He
 		return "", fmt.Errorf("sql: batch is expected to be of type %T, got %T", &BatchTx{}, batch)
 	}
 	// Process the header
-	headerNode, err := ipld.NewEthHeader(header)
+	headerNode, err := ipld.EncodeHeader(header)
 	if err != nil {
 		return "", err
 	}
@@ -208,6 +210,7 @@ func (sdi *StateDiffIndexer) PushHeader(batch interfaces.Batch, header *types.He
 		Timestamp:       header.Time,
 		Coinbase:        header.Coinbase.String(),
 		Canonical:       true,
+		WithdrawalsRoot: shared.MaybeStringHash(header.WithdrawalsHash),
 	})
 }
 
@@ -258,13 +261,15 @@ type processArgs struct {
 	blockTime   uint64
 	receipts    types.Receipts
 	txs         types.Transactions
-	rctNodes    []*ipld.EthReceipt
-	txNodes     []*ipld.EthTx
-	logNodes    [][]*ipld.EthLog
+	withdrawals types.Withdrawals
+	rctNodes    []ipld.IPLD
+	txNodes     []ipld.IPLD
+	logNodes    [][]ipld.IPLD
+	wdNodes     []ipld.IPLD
 }
 
-// processReceiptsAndTxs publishes and indexes receipt and transaction IPLDs in Postgres
-func (sdi *StateDiffIndexer) processReceiptsAndTxs(tx *BatchTx, args processArgs) error {
+// processObjects publishes and indexes receipt and transaction IPLDs in Postgres
+func (sdi *StateDiffIndexer) processObjects(tx *BatchTx, args processArgs) error {
 	// Process receipts and txs
 	signer := types.MakeSigner(sdi.chainConfig, args.blockNumber, args.blockTime)
 	for i, receipt := range args.receipts {
@@ -348,7 +353,23 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(tx *BatchTx, args processArgs
 			return err
 		}
 	}
-
+	// Process withdrawals
+	for i, withdrawal := range args.withdrawals {
+		wdNode := args.wdNodes[i]
+		tx.cacheIPLD(wdNode)
+		wdModel := models.WithdrawalModel{
+			BlockNumber: args.blockNumber.String(),
+			HeaderID:    args.headerID,
+			CID:         wdNode.Cid().String(),
+			Index:       withdrawal.Index,
+			Validator:   withdrawal.Validator,
+			Address:     withdrawal.Address.String(),
+			Amount:      withdrawal.Amount,
+		}
+		if err := sdi.dbWriter.upsertWithdrawalCID(tx.dbtx, wdModel); err != nil {
+			return err
+		}
+	}
 	return nil
 }
 

indexer/database/sql/interfaces.go

@@ -54,6 +54,7 @@ type Statements interface {
 	InsertTxStm() string
 	InsertRctStm() string
 	InsertLogStm() string
+	InsertWithdrawalStm() string
 	InsertStateStm() string
 	InsertStorageStm() string
 	InsertIPLDStm() string

indexer/database/sql/pgx_indexer_test.go

@@ -96,6 +96,14 @@ func TestPGXIndexer(t *testing.T) {
 		test.DoTestPublishAndIndexReceiptIPLDs(t, db)
 	})
 
+	t.Run("Publish and index withdrawal IPLDs in a single tx", func(t *testing.T) {
+		setupPGX(t)
+		defer tearDown(t)
+		defer checkTxClosure(t, 1, 0, 1)
+
+		test.DoTestPublishAndIndexWithdrawalIPLDs(t, db)
+	})
+
 	t.Run("Publish and index state IPLDs in a single tx", func(t *testing.T) {
 		setupPGX(t)
 		defer tearDown(t)

indexer/database/sql/postgres/database.go

@@ -18,6 +18,7 @@ package postgres
 
 import (
 	"fmt"
+	"strings"
 
 	"github.com/cerc-io/plugeth-statediff/indexer/database/sql"
 	"github.com/cerc-io/plugeth-statediff/indexer/shared/schema"
@@ -43,7 +44,9 @@ type DB struct {
 
 // MaxHeaderStm satisfies the sql.Statements interface
 func (db *DB) MaxHeaderStm() string {
-	return fmt.Sprintf("SELECT block_number, block_hash, parent_hash, cid, td, node_ids, reward, state_root, tx_root, receipt_root, uncles_hash, bloom, timestamp, coinbase FROM %s ORDER BY block_number DESC LIMIT 1", schema.TableHeader.Name)
+	return fmt.Sprintf("SELECT %s FROM %s ORDER BY block_number DESC LIMIT 1",
+		strings.Join(schema.TableHeader.ColumnNames(), ","),
+		schema.TableHeader.Name)
 }
 
 // ExistsHeaderStm satisfies the sql.Statements interface
@@ -59,7 +62,7 @@ func (db *DB) DetectGapsStm() string {
 // InsertHeaderStm satisfies the sql.Statements interface
 // Stm == Statement
 func (db *DB) InsertHeaderStm() string {
-	return schema.TableHeader.ToInsertStatement(db.upsert)
+	return schema.TableHeader.PreparedInsert(db.upsert)
 }
 
 // SetCanonicalHeaderStm satisfies the sql.Statements interface
@@ -70,37 +73,42 @@ func (db *DB) SetCanonicalHeaderStm() string {
 
 // InsertUncleStm satisfies the sql.Statements interface
 func (db *DB) InsertUncleStm() string {
-	return schema.TableUncle.ToInsertStatement(db.upsert)
+	return schema.TableUncle.PreparedInsert(db.upsert)
 }
 
 // InsertTxStm satisfies the sql.Statements interface
 func (db *DB) InsertTxStm() string {
-	return schema.TableTransaction.ToInsertStatement(db.upsert)
+	return schema.TableTransaction.PreparedInsert(db.upsert)
 }
 
 // InsertRctStm satisfies the sql.Statements interface
 func (db *DB) InsertRctStm() string {
-	return schema.TableReceipt.ToInsertStatement(db.upsert)
+	return schema.TableReceipt.PreparedInsert(db.upsert)
 }
 
 // InsertLogStm satisfies the sql.Statements interface
 func (db *DB) InsertLogStm() string {
-	return schema.TableLog.ToInsertStatement(db.upsert)
+	return schema.TableLog.PreparedInsert(db.upsert)
+}
+
+// InsertLogStm satisfies the sql.Statements interface
+func (db *DB) InsertWithdrawalStm() string {
+	return schema.TableWithdrawal.PreparedInsert(db.upsert)
 }
 
 // InsertStateStm satisfies the sql.Statements interface
 func (db *DB) InsertStateStm() string {
-	return schema.TableStateNode.ToInsertStatement(db.upsert)
+	return schema.TableStateNode.PreparedInsert(db.upsert)
 }
 
 // InsertStorageStm satisfies the sql.Statements interface
 func (db *DB) InsertStorageStm() string {
-	return schema.TableStorageNode.ToInsertStatement(db.upsert)
+	return schema.TableStorageNode.PreparedInsert(db.upsert)
 }
 
 // InsertIPLDStm satisfies the sql.Statements interface
 func (db *DB) InsertIPLDStm() string {
-	return schema.TableIPLDBlock.ToInsertStatement(db.upsert)
+	return schema.TableIPLDBlock.PreparedInsert(db.upsert)
 }
 
 // InsertIPLDsStm satisfies the sql.Statements interface

indexer/database/sql/sqlx_indexer_test.go

@@ -82,6 +82,14 @@ func TestSQLXIndexer(t *testing.T) {
 		test.DoTestPublishAndIndexReceiptIPLDs(t, db)
 	})
 
+	t.Run("Publish and index withdrawal IPLDs in a single tx", func(t *testing.T) {
+		setupSQLX(t)
+		defer tearDown(t)
+		defer checkTxClosure(t, 0, 0, 0)
+
+		test.DoTestPublishAndIndexWithdrawalIPLDs(t, db)
+	})
+
 	t.Run("Publish and index state IPLDs in a single tx", func(t *testing.T) {
 		setupSQLX(t)
 		defer tearDown(t)

indexer/database/sql/writer.go

@@ -95,6 +95,7 @@ func (w *Writer) maxHeader() (*models.HeaderModel, error) {
 		&model.Timestamp,
 		&model.Coinbase,
 		&model.Canonical,
+		&model.WithdrawalsRoot,
 	)
 	model.BlockNumber = strconv.FormatUint(number, 10)
 	model.TotalDifficulty = strconv.FormatUint(td, 10)
@@ -125,6 +126,7 @@ func (w *Writer) upsertHeaderCID(tx Tx, header models.HeaderModel) error {
 		header.Timestamp,
 		header.Coinbase,
 		header.Canonical,
+		header.WithdrawalsRoot,
 	)
 	if err != nil {
 		return insertError{"eth.header_cids", err, w.db.InsertHeaderStm(), header}
@@ -286,6 +288,41 @@ func (w *Writer) upsertLogCID(tx Tx, logs []*models.LogsModel) error {
 	return nil
 }
 
+func (w *Writer) upsertWithdrawalCID(tx Tx, withdrawal models.WithdrawalModel) error {
+	if w.useCopyForTx(tx) {
+		blockNum, err := strconv.ParseUint(withdrawal.BlockNumber, 10, 64)
+		if err != nil {
+			return insertError{"eth.withdrawal_cids", err, "COPY", withdrawal}
+		}
+
+		_, err = tx.CopyFrom(w.db.Context(), schema.TableWithdrawal.TableName(), schema.TableWithdrawal.ColumnNames(),
+			toRows(toRow(blockNum,
+				withdrawal.HeaderID,
+				withdrawal.CID,
+				withdrawal.Index,
+				withdrawal.Validator,
+				withdrawal.Address,
+				withdrawal.Amount)))
+		if err != nil {
+			return insertError{"eth.withdrawal_cids", err, "COPY", withdrawal}
+		}
+	} else {
+		_, err := tx.Exec(w.db.Context(), w.db.InsertWithdrawalStm(),
+			withdrawal.BlockNumber,
+			withdrawal.HeaderID,
+			withdrawal.CID,
+			withdrawal.Index,
+			withdrawal.Validator,
+			withdrawal.Address,
+			withdrawal.Amount)
+		if err != nil {
+			return insertError{"eth.withdrawal_cids", err, w.db.InsertWithdrawalStm(), withdrawal}
+		}
+	}
+	metrics.IndexerMetrics.WithdrawalsCounter.Inc(1)
+	return nil
+}
+
 /*
 INSERT INTO eth.state_cids (block_number, header_id, state_leaf_key, cid, removed, diff, balance, nonce, code_hash, storage_root) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
 ON CONFLICT (header_id, state_leaf_key, block_number) DO NOTHING