index recent deposits & add signature verification

db/deposits.go

@@ -61,17 +61,6 @@ func InsertDepositTxs(depositTxs []*dbtypes.DepositTx, tx *sqlx.Tx) error {
 	return nil
 }
 
-type Deposit struct {
-	Index                 uint64 `db:"deposit_index"`
-	SlotNumber            uint64 `db:"slot_number"`
-	SlotIndex             uint64 `db:"slot_index"`
-	SlotRoot              []byte `db:"slot_root"`
-	Orphaned              bool   `db:"orphaned"`
-	PublicKey             []byte `db:"publickey"`
-	WithdrawalCredentials []byte `db:"withdrawalcredentials"`
-	Amount                uint64 `db:"amount"`
-}
-
 func InsertDeposits(deposits []*dbtypes.Deposit, tx *sqlx.Tx) error {
 	var sql strings.Builder
 	fmt.Fprint(&sql,

db/schema/pgsql/20240429121225_deposit-index.sql

@@ -21,6 +21,10 @@ CREATE INDEX IF NOT EXISTS "deposit_txs_deposit_index_idx"
     ON public."deposit_txs"
     ("deposit_index" ASC NULLS FIRST);
 
+CREATE INDEX IF NOT EXISTS "deposit_txs_block_number_idx"
+    ON public."deposit_txs"
+    ("block_number" ASC NULLS FIRST);
+
 CREATE INDEX IF NOT EXISTS "deposit_txs_publickey_idx"
     ON public."deposit_txs"
     ("publickey" ASC NULLS FIRST);

db/schema/sqlite/20240429121225_deposit-index.sql

@@ -21,6 +21,10 @@ CREATE INDEX IF NOT EXISTS "deposit_txs_deposit_index_idx"
     ON "deposit_txs"
     ("deposit_index" ASC);
 
+CREATE INDEX IF NOT EXISTS "deposit_txs_block_number_idx"
+    ON "deposit_txs"
+    ("block_number" ASC);
+
 CREATE INDEX IF NOT EXISTS "deposit_txs_publickey_idx"
     ON "deposit_txs"
     ("publickey" ASC);

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/mattn/go-sqlite3 v1.14.22
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/pressly/goose/v3 v3.20.0
+	github.com/protolambda/zrnt v0.32.3
 	github.com/prysmaticlabs/go-bitfield v0.0.0-20240328144219-a1caa50c3a1e
 	github.com/rs/zerolog v1.32.0
 	github.com/shopspring/decimal v1.4.0
@@ -82,6 +83,7 @@ require (
 	github.com/jackc/pgtype v1.14.3 // indirect
 	github.com/jackc/puddle v1.3.0 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
+	github.com/kilic/bls12-381 v0.1.0 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
@@ -93,6 +95,8 @@ require (
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.53.0 // indirect
 	github.com/prometheus/procfs v0.14.0 // indirect
+	github.com/protolambda/bls12-381-util v0.1.0 // indirect
+	github.com/protolambda/ztyp v0.2.2 // indirect
 	github.com/r3labs/sse/v2 v2.10.0 // indirect
 	github.com/sethvargo/go-retry v0.2.4 // indirect
 	github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible // indirect

go.sum

@@ -163,6 +163,8 @@ github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb h1:PBC98N2aIaM3XXi
 github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/gofuzz v1.2.0 h1:xRy4A+RhZaiKjJ1bPfwQ8sedCA+YS2YcCHW6ec7JMi0=
+github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/subcommands v1.2.0/go.mod h1:ZjhPrFU+Olkh9WazFPsl27BQ4UPiG37m3yTrtFlrHVk=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
@@ -179,6 +181,7 @@ github.com/holiman/billy v0.0.0-20240216141850-2abb0c79d3c4 h1:X4egAf/gcS1zATw6w
 github.com/holiman/billy v0.0.0-20240216141850-2abb0c79d3c4/go.mod h1:5GuXa7vkL8u9FkFuWdVvfR5ix8hRB7DbOAaYULamFpc=
 github.com/holiman/bloomfilter/v2 v2.0.3 h1:73e0e/V0tCydx14a0SCYS/EWCxgwLZ18CZcZKVu0fao=
 github.com/holiman/bloomfilter/v2 v2.0.3/go.mod h1:zpoh+gs7qcpqrHr3dB55AMiJwo0iURXE7ZOP9L9hSkA=
+github.com/holiman/uint256 v1.2.0/go.mod h1:y4ga/t+u+Xwd7CpDgZESaRcWy0I7XMlTMA25ApIH5Jw=
 github.com/holiman/uint256 v1.2.4 h1:jUc4Nk8fm9jZabQuqr2JzednajVmBpC+oiTiXZJEApU=
 github.com/holiman/uint256 v1.2.4/go.mod h1:EOMSn4q6Nyt9P6efbI3bueV4e1b3dGlUCXeiRV4ng7E=
 github.com/huandu/go-assert v1.1.5 h1:fjemmA7sSfYHJD7CUqs9qTwwfdNAx7/j2/ZlHXzNB3c=
@@ -253,6 +256,8 @@ github.com/juliangruber/go-intersect v1.1.0 h1:sc+y5dCjMMx0pAdYk/N6KBm00tD/f3tq+
 github.com/juliangruber/go-intersect v1.1.0/go.mod h1:WMau+1kAmnlQnKiikekNJbtGtfmILU/mMU6H7AgKbWQ=
 github.com/kelseyhightower/envconfig v1.4.0 h1:Im6hONhd3pLkfDFsbRgu68RDNkGF1r3dvMUtDTo2cv8=
 github.com/kelseyhightower/envconfig v1.4.0/go.mod h1:cccZRl6mQpaq41TPp5QxidR+Sa3axMbJDNb//FQX6Gg=
+github.com/kilic/bls12-381 v0.1.0 h1:encrdjqKMEvabVQ7qYOKu1OvhqpK4s47wDYtNiPtlp4=
+github.com/kilic/bls12-381 v0.1.0/go.mod h1:vDTTHJONJ6G+P2R74EhnyotQDTliQDnFEwhdmfzw1ig=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
 github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
@@ -332,6 +337,12 @@ github.com/prometheus/common v0.53.0 h1:U2pL9w9nmJwJDa4qqLQ3ZaePJ6ZTwt7cMD3AG3+a
 github.com/prometheus/common v0.53.0/go.mod h1:BrxBKv3FWBIGXw89Mg1AeBq7FSyRzXWI3l3e7W3RN5U=
 github.com/prometheus/procfs v0.14.0 h1:Lw4VdGGoKEZilJsayHf0B+9YgLGREba2C6xr+Fdfq6s=
 github.com/prometheus/procfs v0.14.0/go.mod h1:XL+Iwz8k8ZabyZfMFHPiilCniixqQarAy5Mu67pHlNQ=
+github.com/protolambda/bls12-381-util v0.1.0 h1:05DU2wJN7DTU7z28+Q+zejXkIsA/MF8JZQGhtBZZiWk=
+github.com/protolambda/bls12-381-util v0.1.0/go.mod h1:cdkysJTRpeFeuUVx/TXGDQNMTiRAalk1vQw3TYTHcE4=
+github.com/protolambda/zrnt v0.32.3 h1:b3mkBEjcmxtft115cBIQk+2qz1HEb2ExDdduVQqN4v0=
+github.com/protolambda/zrnt v0.32.3/go.mod h1:A0fezkp9Tt3GBLATSPIbuY4ywYESyAuc/FFmPKg8Lqs=
+github.com/protolambda/ztyp v0.2.2 h1:rVcL3vBu9W/aV646zF6caLS/dyn9BN8NYiuJzicLNyY=
+github.com/protolambda/ztyp v0.2.2/go.mod h1:9bYgKGqg3wJqT9ac1gI2hnVb0STQq7p/1lapqrqY1dU=
 github.com/prysmaticlabs/go-bitfield v0.0.0-20240328144219-a1caa50c3a1e h1:ATgOe+abbzfx9kCPeXIW4fiWyDdxlwHw07j8UGhdTd4=
 github.com/prysmaticlabs/go-bitfield v0.0.0-20240328144219-a1caa50c3a1e/go.mod h1:wmuf/mdK4VMD+jA9ThwcUKjg3a2XWM9cVfFYjDyY4j4=
 github.com/r3labs/sse/v2 v2.10.0 h1:hFEkLLFY4LDifoHdiCN/LlGBAdVJYsANaLqNYa1l/v0=
@@ -479,6 +490,7 @@ golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201101102859-da207088b7d1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

indexer/deposit_indexer.go

@@ -15,6 +15,9 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
+	blsu "github.com/protolambda/bls12-381-util"
+	zrnt_common "github.com/protolambda/zrnt/eth2/beacon/common"
+	"github.com/protolambda/ztyp/tree"
 
 	"github.com/ethpandaops/dora/db"
 	"github.com/ethpandaops/dora/dbtypes"
@@ -28,6 +31,7 @@ type DepositIndexer struct {
 	depositContract    common.Address
 	depositContractAbi *abi.ABI
 	depositEventTopic  []byte
+	depositSigDomain   zrnt_common.BLSDomain
 }
 
 const depositContractAbi = `[{"inputs":[],"stateMutability":"nonpayable","type":"constructor"},{"anonymous":false,"inputs":[{"indexed":false,"internalType":"bytes","name":"pubkey","type":"bytes"},{"indexed":false,"internalType":"bytes","name":"withdrawal_credentials","type":"bytes"},{"indexed":false,"internalType":"bytes","name":"amount","type":"bytes"},{"indexed":false,"internalType":"bytes","name":"signature","type":"bytes"},{"indexed":false,"internalType":"bytes","name":"index","type":"bytes"}],"name":"DepositEvent","type":"event"},{"inputs":[{"internalType":"bytes","name":"pubkey","type":"bytes"},{"internalType":"bytes","name":"withdrawal_credentials","type":"bytes"},{"internalType":"bytes","name":"signature","type":"bytes"},{"internalType":"bytes32","name":"deposit_data_root","type":"bytes32"}],"name":"deposit","outputs":[],"stateMutability":"payable","type":"function"},{"inputs":[],"name":"get_deposit_count","outputs":[{"internalType":"bytes","name":"","type":"bytes"}],"stateMutability":"view","type":"function"},{"inputs":[],"name":"get_deposit_root","outputs":[{"internalType":"bytes32","name":"","type":"bytes32"}],"stateMutability":"view","type":"function"},{"inputs":[{"internalType":"bytes4","name":"interfaceId","type":"bytes4"}],"name":"supportsInterface","outputs":[{"internalType":"bool","name":"","type":"bool"}],"stateMutability":"pure","type":"function"}]`
@@ -45,12 +49,16 @@ func newDepositIndexer(indexer *Indexer) *DepositIndexer {
 
 	depositEventTopic := crypto.Keccak256Hash([]byte(contractAbi.Events["DepositEvent"].Sig))
 
+	genesisForkVersion := common.FromHex(utils.Config.Chain.Config.GenesisForkVersion)
+	depositSigDomain := zrnt_common.ComputeDomain(zrnt_common.DOMAIN_DEPOSIT, zrnt_common.Version(genesisForkVersion), zrnt_common.Root{})
+
 	ds := &DepositIndexer{
 		indexer:            indexer,
 		batchSize:          batchSize,
 		depositContract:    common.HexToAddress(utils.Config.Chain.Config.DepositContractAddress),
 		depositContractAbi: &contractAbi,
 		depositEventTopic:  depositEventTopic[:],
+		depositSigDomain:   depositSigDomain,
 	}
 
 	go ds.runDepositIndexerLoop()
@@ -62,7 +70,7 @@ func (ds *DepositIndexer) runDepositIndexerLoop() {
 	defer utils.HandleSubroutinePanic("runCacheLoop")
 
 	for {
-		time.Sleep(30 * time.Second)
+		time.Sleep(60 * time.Second)
 		logger.Debugf("run deposit indexer logic")
 
 		err := ds.runDepositIndexer()
@@ -109,6 +117,8 @@ func (ds *DepositIndexer) runDepositIndexer() error {
 		}
 	}
 
+	ds.processRecentBlocks()
+
 	return nil
 }
 
@@ -188,15 +198,125 @@ func (ds *DepositIndexer) processFinalizedBlocks(finalizedBlockNumber uint64) er
 				TxSender:              txFrom[:],
 				TxTarget:              txTo[:],
 			}
+			ds.checkDepositValidity(depositTx)
 			depositTxs = append(depositTxs, depositTx)
 		}
 
 		if len(depositTxs) > 0 {
 			logger.Infof("crawled deposits for block %v - %v: %v deposits", ds.state.FinalBlock, toBlock, len(depositTxs))
 
-			err = ds.persistFinalizedDepositTxs(toBlock, depositTxs)
+			depositCount := len(depositTxs)
+			for depositIdx := 0; depositIdx < depositCount; depositIdx += 500 {
+				endIdx := depositIdx + 500
+				if endIdx > depositCount {
+					endIdx = depositCount
+				}
+
+				err = ds.persistFinalizedDepositTxs(toBlock, depositTxs[depositIdx:endIdx])
+				if err != nil {
+					return fmt.Errorf("could not persist deposit txs: %v", err)
+				}
+			}
+
+			time.Sleep(1 * time.Second)
+		}
+	}
+	return nil
+}
+
+func (ds *DepositIndexer) processRecentBlocks() error {
+	insertMap := map[string]bool{}
+
+	headForks := ds.indexer.GetHeadForks(true)
+	for _, headFork := range headForks {
+		client := ds.indexer.GetReadyElClient(false, headFork.Root, nil)
+		if client == nil {
+			return fmt.Errorf("no ready execution client found")
+		}
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		query := ethereum.FilterQuery{
+			FromBlock: big.NewInt(0).SetUint64(ds.state.FinalBlock),
+			Addresses: []common.Address{
+				ds.depositContract,
+			},
+		}
+
+		logs, err := client.GetRpcClient().GetEthClient().FilterLogs(ctx, query)
+		if err != nil {
+			return fmt.Errorf("error fetching deposit contract logs: %v", err)
+		}
+
+		var txHash []byte
+		var txDetails *types.Transaction
+
+		depositTxs := []*dbtypes.DepositTx{}
+
+		for _, log := range logs {
+			if !bytes.Equal(log.Topics[0][:], ds.depositEventTopic) {
+				continue
+			}
+
+			event, err := ds.depositContractAbi.Unpack("DepositEvent", log.Data)
 			if err != nil {
-				return fmt.Errorf("could not persist deposit txs: %v", err)
+				return fmt.Errorf("error decoding deposit event (%v): %v", log.TxHash, err)
+
+			}
+
+			if txHash == nil || !bytes.Equal(txHash, log.TxHash[:]) {
+				txDetails, _, err = client.GetRpcClient().GetEthClient().TransactionByHash(ctx, log.TxHash)
+				if err != nil {
+					return fmt.Errorf("could not load tx details (%v): %v", log.TxHash, err)
+				}
+
+			}
+
+			txFrom, err := types.Sender(types.LatestSignerForChainID(txDetails.ChainId()), txDetails)
+			if err != nil {
+				return fmt.Errorf("could not decode tx sender (%v): %v", log.TxHash, err)
+			}
+			txTo := *txDetails.To()
+
+			depositTx := &dbtypes.DepositTx{
+				Index:                 binary.LittleEndian.Uint64(event[4].([]byte)),
+				BlockNumber:           log.BlockNumber,
+				BlockRoot:             log.BlockHash[:],
+				PublicKey:             event[0].([]byte),
+				WithdrawalCredentials: event[1].([]byte),
+				Amount:                binary.LittleEndian.Uint64(event[2].([]byte)),
+				Signature:             event[3].([]byte),
+				Orphaned:              true,
+				TxHash:                log.TxHash[:],
+				TxSender:              txFrom[:],
+				TxTarget:              txTo[:],
+			}
+
+			depositKey := fmt.Sprintf("%v-%x", depositTx.Index, depositTx.BlockRoot)
+			if insertMap[depositKey] {
+				break
+			}
+			insertMap[depositKey] = true
+
+			ds.checkDepositValidity(depositTx)
+			depositTxs = append(depositTxs, depositTx)
+		}
+
+		if len(depositTxs) > 0 {
+			logger.Infof("crawled recent deposits since block %v: %v deposits", ds.state.FinalBlock, len(depositTxs))
+
+			depositCount := len(depositTxs)
+			for depositIdx := 0; depositIdx < depositCount; depositIdx += 500 {
+				endIdx := depositIdx + 500
+				if endIdx > depositCount {
+					endIdx = depositCount
+				}
+
+				err = ds.persistRecentDepositTxs(depositTxs[depositIdx:endIdx])
+				if err != nil {
+					return fmt.Errorf("could not persist deposit txs: %v", err)
+				}
 			}
 
 			time.Sleep(1 * time.Second)
@@ -205,6 +325,26 @@ func (ds *DepositIndexer) processFinalizedBlocks(finalizedBlockNumber uint64) er
 	return nil
 }
 
+func (ds *DepositIndexer) checkDepositValidity(depositTx *dbtypes.DepositTx) {
+	depositMsg := &zrnt_common.DepositMessage{
+		Pubkey:                zrnt_common.BLSPubkey(depositTx.PublicKey),
+		WithdrawalCredentials: tree.Root(depositTx.WithdrawalCredentials),
+		Amount:                zrnt_common.Gwei(depositTx.Amount),
+	}
+	depositRoot := depositMsg.HashTreeRoot(tree.GetHashFn())
+	signingRoot := zrnt_common.ComputeSigningRoot(
+		depositRoot,
+		ds.depositSigDomain,
+	)
+
+	pubkey, err := depositMsg.Pubkey.Pubkey()
+	sigData := zrnt_common.BLSSignature(depositTx.Signature)
+	sig, err2 := sigData.Signature()
+	if err == nil && err2 == nil && blsu.Verify(pubkey, signingRoot[:], sig) {
+		depositTx.ValidSignature = true
+	}
+}
+
 func (ds *DepositIndexer) persistFinalizedDepositTxs(toBlockNumber uint64, deposits []*dbtypes.DepositTx) error {
 	tx, err := db.WriterDb.Beginx()
 	if err != nil {
@@ -218,6 +358,9 @@ func (ds *DepositIndexer) persistFinalizedDepositTxs(toBlockNumber uint64, depos
 	}
 
 	ds.state.FinalBlock = toBlockNumber
+	if toBlockNumber > ds.state.HeadBlock {
+		ds.state.HeadBlock = toBlockNumber
+	}
 	err = db.SetExplorerState("indexer.depositstate", ds.state, tx)
 	if err != nil {
 		return fmt.Errorf("error while updating deposit state: %v", err)
@@ -228,3 +371,21 @@ func (ds *DepositIndexer) persistFinalizedDepositTxs(toBlockNumber uint64, depos
 	}
 	return nil
 }
+
+func (ds *DepositIndexer) persistRecentDepositTxs(deposits []*dbtypes.DepositTx) error {
+	tx, err := db.WriterDb.Beginx()
+	if err != nil {
+		return fmt.Errorf("error starting db transactions: %v", err)
+	}
+	defer tx.Rollback()
+
+	err = db.InsertDepositTxs(deposits, tx)
+	if err != nil {
+		return fmt.Errorf("error while inserting deposit txs: %v", err)
+	}
+
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("error committing db transaction: %v", err)
+	}
+	return nil
+}