cerc-io/ipld-eth-beacon-indexer
19
0
Fork 0
Code Issues 4 Pull Requests 2 Actions Packages Projects Releases 9 Wiki Activity

76: Add indexing of ExecutionPayloads (and other Merge-related updates). #73

Merged
telackey merged 30 commits from telackey/the_merge into main 2022-09-29 01:39:56 +00:00
Conversation 0 Commits 30 Files Changed 38 +151 -482
7 changed files with 151 additions and 482 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit b49854884b - Show all commits

10
config/example.ipld-eth-beacon-indexer-config.json
View File

@ -1,15 +1,15 @@
{
"db": {
"address": "localhost",
"password": "password",
"port": 8076,
"username": "vdbm",
"name": "vulcanize_testing",
"password": "secret12",
"port": 45432,
"username": "postgres",
"name": "postgres",
"driver": "PGX"
},
"bc": {
"address": "localhost",
"port": 5052,
"port": 8001,
"type": "lighthouse",
"bootRetryInterval": 30,
"bootMaxRetry": 5,

46
go.mod
View File

@ -6,19 +6,22 @@ require (
github.com/ipfs/go-ipfs-blockstore v1.2.0
github.com/ipfs/go-ipfs-ds-help v1.1.0
github.com/jackc/pgconn v1.12.0
github.com/multiformats/go-multihash v0.1.0
github.com/multiformats/go-multihash v0.2.0
github.com/onsi/ginkgo/v2 v2.1.4
github.com/onsi/gomega v1.19.0
github.com/prometheus/client_golang v1.12.1
github.com/prometheus/client_golang v1.12.2
github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc
github.com/sirupsen/logrus v1.8.1
)
require (
github.com/BurntSushi/toml v1.1.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/btcsuite/btcd/btcec/v2 v2.2.0 // indirect
github.com/cespare/xxhash/v2 v2.1.2 // indirect
github.com/cpuguy83/go-md2man/v2 v2.0.1 // indirect
github.com/ethereum/go-ethereum v1.10.17 // indirect
github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
github.com/ethereum/go-ethereum v1.10.20 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
github.com/golang/protobuf v1.5.2 // indirect
@ -28,12 +31,12 @@ require (
github.com/herumi/bls-eth-go-binary v0.0.0-20210917013441-d37c07cfda4e // indirect
github.com/ipfs/bbloom v0.0.4 // indirect
github.com/ipfs/go-block-format v0.0.3 // indirect
github.com/ipfs/go-cid v0.1.0 // indirect
github.com/ipfs/go-cid v0.2.0 // indirect
github.com/ipfs/go-datastore v0.5.0 // indirect
github.com/ipfs/go-ipfs-util v0.0.2 // indirect
github.com/ipfs/go-ipld-format v0.3.0 // indirect
github.com/ipfs/go-log v1.0.5 // indirect
github.com/ipfs/go-log/v2 v2.5.0 // indirect
github.com/ipfs/go-log/v2 v2.5.1 // indirect
github.com/ipfs/go-metrics-interface v0.0.1 // indirect
github.com/jackc/chunkreader/v2 v2.0.1 // indirect
github.com/jackc/pgio v1.0.0 // indirect
@ -43,37 +46,42 @@ require (
github.com/jackc/pgtype v1.11.0 // indirect
github.com/jackc/puddle v1.2.1 // indirect
github.com/jbenet/goprocess v0.1.4 // indirect
github.com/klauspost/cpuid/v2 v2.0.12 // indirect
github.com/klauspost/cpuid/v2 v2.0.14 // indirect
github.com/lib/pq v1.10.5 // indirect
github.com/mattn/go-isatty v0.0.14 // indirect
github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1 // indirect
github.com/minio/highwayhash v1.0.1 // indirect
github.com/minio/sha256-simd v1.0.0 // indirect
github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 // indirect
github.com/mr-tron/base58 v1.2.0 // indirect
github.com/multiformats/go-base32 v0.0.4 // indirect
github.com/multiformats/go-base36 v0.1.0 // indirect
github.com/multiformats/go-multibase v0.0.3 // indirect
github.com/multiformats/go-multibase v0.1.1 // indirect
github.com/multiformats/go-varint v0.0.6 // indirect
github.com/opentracing/opentracing-go v1.2.0 // indirect
github.com/pkg/errors v0.9.1 // indirect
github.com/prometheus/client_model v0.2.0 // indirect
github.com/prometheus/common v0.32.1 // indirect
github.com/prometheus/common v0.35.0 // indirect
github.com/prometheus/procfs v0.7.3 // indirect
github.com/prysmaticlabs/fastssz v0.0.0-20220628121656-93dfe28febab // indirect
github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7 // indirect
github.com/prysmaticlabs/gohashtree v0.0.1-alpha.0.20220303211031-f753e083138c // indirect
github.com/prysmaticlabs/gohashtree v0.0.2-alpha // indirect
github.com/prysmaticlabs/prysm/v3 v3.0.0 // indirect
github.com/russross/blackfriday/v2 v2.1.0 // indirect
github.com/spaolacci/murmur3 v1.1.0 // indirect
github.com/supranational/blst v0.3.5 // indirect
github.com/supranational/blst v0.3.10 // indirect
github.com/thomaso-mirodin/intmath v0.0.0-20160323211736-5dc6d854e46e // indirect
github.com/urfave/cli/v2 v2.3.0 // indirect
github.com/umbracle/go-eth-consensus v0.1.1-0.20220824100056-6a6a6f5dbe76 // indirect
github.com/urfave/cli/v2 v2.10.2 // indirect
github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
go.opencensus.io v0.23.0 // indirect
go.uber.org/atomic v1.9.0 // indirect
go.uber.org/multierr v1.8.0 // indirect
go.uber.org/zap v1.21.0 // indirect
golang.org/x/crypto v0.0.0-20220427172511-eb4f295cb31f // indirect
golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4 // indirect
golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d // indirect
golang.org/x/net v0.0.0-20220624214902-1bab6f366d9e // indirect
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
golang.org/x/tools v0.1.11 // indirect
google.golang.org/genproto v0.0.0-20220505152158-f39f71e6c8f3 // indirect
google.golang.org/grpc v1.46.0 // indirect
google.golang.org/protobuf v1.28.0 // indirect
@ -82,7 +90,7 @@ require (
)
require (
github.com/ferranbt/fastssz v0.0.0-20220303160658-88bb965b6747 // indirect
github.com/ferranbt/fastssz v0.1.1 // indirect
github.com/fsnotify/fsnotify v1.5.4 // indirect
github.com/georgysavva/scany v0.3.0
github.com/hashicorp/hcl v1.0.0 // indirect
@ -101,10 +109,10 @@ require (
github.com/spf13/pflag v1.0.5 // indirect
github.com/spf13/viper v1.11.0
github.com/subosito/gotenv v1.2.0 // indirect
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6 // indirect
golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f
golang.org/x/sys v0.0.0-20220702020025-31831981b65f // indirect
golang.org/x/text v0.3.7 // indirect
gopkg.in/ini.v1 v1.66.4 // indirect
gopkg.in/yaml.v2 v2.4.0 // indirect
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)

436
go.sum
View File

File diff suppressed because it is too large Load Diff

2
pkg/beaconclient/checkbeaconserverstatus.go
View File

@ -46,7 +46,7 @@ type SyncData struct {
}
// This function will check to see if we are synced up with the head of chain.
//{"data":{"is_syncing":true,"head_slot":"62528","sync_distance":"3734299"}}
// {"data":{"is_syncing":true,"head_slot":"62528","sync_distance":"3734299"}}
func (bc BeaconClient) CheckHeadSync() (bool, error) {
syncStatus, err := bc.QueryHeadSync()
if err != nil {

4
pkg/beaconclient/metrics.go
View File

@ -23,7 +23,7 @@ import (
"github.com/vulcanize/ipld-eth-beacon-indexer/pkg/loghelper"
)
//Create a metric struct and register each channel with prometheus
// Create a metric struct and register each channel with prometheus
func CreateBeaconClientMetrics() (*BeaconClientMetrics, error) {
metrics := &BeaconClientMetrics{
SlotInserts: 0,
@ -138,7 +138,7 @@ func (m *BeaconClientMetrics) IncrementReorgError(inc uint64) {
}
// Wrapper function to increment the number of knownGaps that were updated with reprocessing errors.
//If we want to use mutexes later we can easily update all occurrences here.
// If we want to use mutexes later we can easily update all occurrences here.
func (m *BeaconClientMetrics) IncrementKnownGapsReprocessError(inc uint64) {
atomic.AddUint64(&m.KnownGapsReprocessError, inc)
}

106
pkg/beaconclient/processslot.go
View File

@ -23,31 +23,21 @@ import (
"context"
"encoding/hex"
"fmt"
consensus "github.com/umbracle/go-eth-consensus"
"github.com/umbracle/go-eth-consensus/http"
"strconv"
"strings"
"time"
"github.com/jackc/pgx/v4"
si "github.com/prysmaticlabs/prysm/consensus-types/interfaces"
"github.com/prysmaticlabs/prysm/consensus-types/wrapper"
dt "github.com/prysmaticlabs/prysm/encoding/ssz/detect"
// The below is temporary, once https://github.com/prysmaticlabs/prysm/issues/10006 has been resolved we wont need it.
// pb "github.com/prysmaticlabs/prysm/proto/prysm/v2"
state "github.com/prysmaticlabs/prysm/beacon-chain/state"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/ipld-eth-beacon-indexer/pkg/database/sql"
"github.com/vulcanize/ipld-eth-beacon-indexer/pkg/loghelper"
"golang.org/x/sync/errgroup"
)
var (
ParentRootUnmarshalError = "Unable to properly unmarshal the ParentRoot field in the SignedBeaconBlock."
MissingEth1Data = "Can't get the Eth1 block_hash"
VersionedUnmarshalerError = "Unable to create a versioned unmarshaler"
)
type ProcessSlot struct {
// Generic
@ -64,10 +54,10 @@ type ProcessSlot struct {
// BeaconBlock
SszSignedBeaconBlock []byte // The entire SSZ encoded SignedBeaconBlock
FullSignedBeaconBlock si.SignedBeaconBlock // The unmarshaled BeaconState object, the unmarshalling could have errors.
FullSignedBeaconBlock consensus.SignedBeaconBlockBellatrix // The unmarshaled BeaconState object, the unmarshalling could have errors.
// BeaconState
FullBeaconState state.BeaconState // The unmarshaled BeaconState object, the unmarshalling could have errors.
FullBeaconState consensus.BeaconStateBellatrix // The unmarshaled BeaconState object, the unmarshalling could have errors.
SszBeaconState []byte // The entire SSZ encoded BeaconState
// DB Write objects
@ -120,7 +110,6 @@ func processFullSlot(ctx context.Context, db sql.Database, serverAddress string,
}
g, _ := errgroup.WithContext(context.Background())
vUnmarshalerCh := make(chan *dt.VersionedUnmarshaler, 1)
// Get the BeaconState.
g.Go(func() error {
@ -129,7 +118,7 @@ func processFullSlot(ctx context.Context, db sql.Database, serverAddress string,
return nil
default:
start := time.Now()
err := ps.getBeaconState(serverAddress, vUnmarshalerCh)
err := ps.getBeaconState(serverAddress)
if err != nil {
return err
}
@ -145,7 +134,7 @@ func processFullSlot(ctx context.Context, db sql.Database, serverAddress string,
return nil
default:
start := time.Now()
err := ps.getSignedBeaconBlock(serverAddress, vUnmarshalerCh)
err := ps.getSignedBeaconBlock(serverAddress)
if err != nil {
return err
}
@ -206,7 +195,7 @@ func processFullSlot(ctx context.Context, db sql.Database, serverAddress string,
reorgTime := time.Now()
headOrHistoric = strings.ToLower(headOrHistoric)
if headOrHistoric != "head" && headOrHistoric != "historic" {
return fmt.Errorf("headOrHistoric must be either historic or head!"), ""
return fmt.Errorf("headOrHistoric must be either historic or head"), ""
}
if ps.HeadOrHistoric == "head" && previousSlot != 0 && previousBlockRoot != "" && ps.Status != "skipped" {
ps.checkPreviousSlot(dw.Tx, dw.Ctx, previousSlot, previousBlockRoot, knownGapsTableIncrement)
@ -251,74 +240,71 @@ func handleHistoricSlot(ctx context.Context, db sql.Database, serverAddress stri
}
// Update the SszSignedBeaconBlock and FullSignedBeaconBlock object with their respective values.
func (ps *ProcessSlot) getSignedBeaconBlock(serverAddress string, vmCh <-chan *dt.VersionedUnmarshaler) error {
func (ps *ProcessSlot) getSignedBeaconBlock(serverAddress string) error {
var blockIdentifier string // Used to query the block
if ps.BlockRoot != "" {
blockIdentifier = ps.BlockRoot
} else {
blockIdentifier = strconv.Itoa(ps.Slot)
}
blockEndpoint := serverAddress + BcBlockQueryEndpoint + blockIdentifier
var err error
var rc int
ps.SszSignedBeaconBlock, rc, err = querySsz(blockEndpoint, strconv.Itoa(ps.Slot))
client := http.New(serverAddress)
var signedBeaconBlock consensus.SignedBeaconBlockBellatrix
err := client.Get(BcBlockQueryEndpoint+blockIdentifier, &signedBeaconBlock)
if err != nil {
loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error("Unable to properly query the slot.")
return err
}
vm := <-vmCh
if rc != 200 {
ps.FullSignedBeaconBlock = &wrapper.Phase0SignedBeaconBlock{}
sszSignedBeaconBlock, err := signedBeaconBlock.MarshalSSZ()
if err != nil {
loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error("Unable to marshal SignedBeaconBlock to SSZ.")
ps.FullSignedBeaconBlock = consensus.SignedBeaconBlockBellatrix{}
ps.SszSignedBeaconBlock = []byte{}
ps.ParentBlockRoot = ""
ps.Status = "skipped"
return nil
}
if vm == nil {
return fmt.Errorf(VersionedUnmarshalerError)
}
ps.FullSignedBeaconBlock = signedBeaconBlock
ps.SszSignedBeaconBlock = sszSignedBeaconBlock
ps.FullSignedBeaconBlock, err = vm.UnmarshalBeaconBlock(ps.SszSignedBeaconBlock)
if err != nil {
loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Warn("Unable to process the slots SignedBeaconBlock")
return nil
}
ps.ParentBlockRoot = "0x" + hex.EncodeToString(ps.FullSignedBeaconBlock.Block().ParentRoot())
ps.ParentBlockRoot = rootToHex(&ps.FullSignedBeaconBlock.Block.ParentRoot)
return nil
}
// Update the SszBeaconState and FullBeaconState object with their respective values.
func (ps *ProcessSlot) getBeaconState(serverEndpoint string, vmCh chan<- *dt.VersionedUnmarshaler) error {
func (ps *ProcessSlot) getBeaconState(serverEndpoint string) error {
var stateIdentifier string // Used to query the state
if ps.StateRoot != "" {
stateIdentifier = ps.StateRoot
} else {
stateIdentifier = strconv.Itoa(ps.Slot)
}
stateEndpoint := serverEndpoint + BcStateQueryEndpoint + stateIdentifier
ps.SszBeaconState, _, _ = querySsz(stateEndpoint, strconv.Itoa(ps.Slot))
client := http.New(serverEndpoint)
versionedUnmarshaler, err := dt.FromState(ps.SszBeaconState)
var beaconState consensus.BeaconStateBellatrix
err := client.Get(BcStateQueryEndpoint+stateIdentifier, &beaconState)
if err != nil {
loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error(VersionedUnmarshalerError)
vmCh <- nil
return fmt.Errorf(VersionedUnmarshalerError)
}
vmCh <- versionedUnmarshaler
ps.FullBeaconState, err = versionedUnmarshaler.UnmarshalBeaconState(ps.SszBeaconState)
if err != nil {
loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error("Unable to process the slots BeaconState")
loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error("Unable to properly query the BeaconState.")
return err
}
sszBeaconState, err := beaconState.MarshalSSZ()
if err != nil {
loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error("Unable to marshal BeaconState.")
return err
}
ps.FullBeaconState = beaconState
ps.SszBeaconState = sszBeaconState
return nil
}
// Check to make sure that the previous block we processed is the parent of the current block.
func (ps *ProcessSlot) checkPreviousSlot(tx sql.Tx, ctx context.Context, previousSlot int, previousBlockRoot string, knownGapsTableIncrement int) {
parentRoot := "0x" + hex.EncodeToString(ps.FullSignedBeaconBlock.Block().ParentRoot())
slot := int(ps.FullBeaconState.Slot())
parentRoot := rootToHex(&ps.FullSignedBeaconBlock.Block.ParentRoot)
slot := int(ps.FullBeaconState.Slot)
if previousSlot == slot {
log.WithFields(log.Fields{
"slot": slot,
@ -335,7 +321,7 @@ func (ps *ProcessSlot) checkPreviousSlot(tx sql.Tx, ctx context.Context, previou
"previousSlot": previousSlot,
"currentSlot": slot,
}).Error("We skipped a few slots.")
transactKnownGaps(tx, ctx, knownGapsTableIncrement, previousSlot+1, slot-1, fmt.Errorf("Gaps during head processing"), "headGaps", ps.Metrics)
transactKnownGaps(tx, ctx, knownGapsTableIncrement, previousSlot+1, slot-1, fmt.Errorf("gaps during head processing"), "headGaps", ps.Metrics)
} else if previousBlockRoot != parentRoot {
log.WithFields(log.Fields{
"previousBlockRoot": previousBlockRoot,
@ -380,23 +366,29 @@ func (ps *ProcessSlot) provideFinalHash() (string, string, string, error) {
if ps.StateRoot != "" {
stateRoot = ps.StateRoot
} else {
stateRoot = "0x" + hex.EncodeToString(ps.FullSignedBeaconBlock.Block().StateRoot())
stateRoot = rootToHex(&ps.FullSignedBeaconBlock.Block.StateRoot)
log.Debug("StateRoot: ", stateRoot)
}
if ps.BlockRoot != "" {
blockRoot = ps.BlockRoot
} else {
var err error
rawBlockRoot, err := ps.FullSignedBeaconBlock.Block().HashTreeRoot()
//blockRoot, err = queryBlockRoot(blockRootEndpoint, strconv.Itoa(ps.Slot))
rawBlockRoot, err := ps.FullSignedBeaconBlock.Block.HashTreeRoot()
if err != nil {
return "", "", "", err
}
blockRoot = "0x" + hex.EncodeToString(rawBlockRoot[:])
blockRoot = byteArrayToHex(&rawBlockRoot)
log.WithFields(log.Fields{"blockRoot": blockRoot}).Debug("Block Root from ssz")
}
eth1BlockHash = "0x" + hex.EncodeToString(ps.FullSignedBeaconBlock.Block().Body().Eth1Data().BlockHash)
eth1BlockHash = byteArrayToHex(&ps.FullSignedBeaconBlock.Block.Body.Eth1Data.BlockHash)
}
return blockRoot, stateRoot, eth1BlockHash, nil
}
func rootToHex(r *consensus.Root) string {
return "0x" + hex.EncodeToString(r[:])
}
func byteArrayToHex(r *[32]byte) string {
return "0x" + hex.EncodeToString(r[:])
}

25
pkg/beaconclient/queryserver.go
View File

@ -27,31 +27,6 @@ import (
"github.com/vulcanize/ipld-eth-beacon-indexer/pkg/loghelper"
)
// A helper function to query endpoints that utilize slots.
func querySsz(endpoint string, slot string) ([]byte, int, error) {
log.WithFields(log.Fields{"endpoint": endpoint}).Debug("Querying endpoint")
client := &http.Client{}
req, err := http.NewRequest("GET", endpoint, nil)
if err != nil {
loghelper.LogSlotError(slot, err).Error("Unable to create a request!")
return nil, 0, fmt.Errorf("Unable to create a request!: %s", err.Error())
}
req.Header.Set("Accept", "application/octet-stream")
response, err := client.Do(req)
if err != nil {
loghelper.LogSlotError(slot, err).Error("Unable to query Beacon Node!")
return nil, 0, fmt.Errorf("Unable to query Beacon Node: %s", err.Error())
}
defer response.Body.Close()
rc := response.StatusCode
body, err := ioutil.ReadAll(response.Body)
if err != nil {
loghelper.LogSlotError(slot, err).Error("Unable to turn response into a []bytes array!")
return nil, rc, fmt.Errorf("Unable to turn response into a []bytes array!: %s", err.Error())
}
return body, rc, nil
}
// Object to unmarshal the BlockRootResponse
type BlockRootResponse struct {
Data BlockRootMessage `json:"data"`