453 lines
16 KiB
Go
453 lines
16 KiB
Go
// VulcanizeDB
|
|
// Copyright © 2022 Vulcanize
|
|
|
|
// This program is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU Affero General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
|
|
// This program is distributed in the hope that it will be useful,
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
// GNU Affero General Public License for more details.
|
|
|
|
// You should have received a copy of the GNU Affero General Public License
|
|
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
// This file will keep track of all the code needed to process a slot.
|
|
// To process a slot, it should have all the necessary data needed to write it to the DB.
|
|
// But not actually write it.
|
|
|
|
package beaconclient
|
|
|
|
import (
|
|
"context"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v4"
|
|
|
|
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"
|
|
)
|
|
|
|
type SlotProcessingDetails struct {
|
|
Context context.Context // A context generic context with multiple uses.
|
|
ServerEndpoint string // What is the endpoint of the beacon server.
|
|
Db sql.Database // Database object used for reads and writes.
|
|
Metrics *BeaconClientMetrics // An object used to keep track of certain BeaconClient Metrics.
|
|
KnownGapTableIncrement int // The max number of slots within a single known_gaps table entry.
|
|
CheckDb bool // Should we check the DB to see if the slot exists before processing it?
|
|
PerformBeaconStateProcessing bool // Should we process BeaconStates?
|
|
PerformBeaconBlockProcessing bool // Should we process BeaconBlocks?
|
|
|
|
StartingSlot int // If we're performing head tracking. What is the first slot we processed.
|
|
PreviousSlot int // Whats the previous slot we processed
|
|
PreviousBlockRoot string // Whats the previous block root, used to check the next blocks parent.
|
|
}
|
|
|
|
func (bc *BeaconClient) SlotProcessingDetails() SlotProcessingDetails {
|
|
return SlotProcessingDetails{
|
|
Context: bc.Context,
|
|
ServerEndpoint: bc.ServerEndpoint,
|
|
Db: bc.Db,
|
|
Metrics: bc.Metrics,
|
|
|
|
CheckDb: bc.CheckDb,
|
|
PerformBeaconBlockProcessing: bc.PerformBeaconBlockProcessing,
|
|
PerformBeaconStateProcessing: bc.PerformBeaconStateProcessing,
|
|
|
|
KnownGapTableIncrement: bc.KnownGapTableIncrement,
|
|
StartingSlot: bc.StartingSlot,
|
|
PreviousSlot: bc.PreviousSlot,
|
|
PreviousBlockRoot: bc.PreviousBlockRoot,
|
|
}
|
|
}
|
|
|
|
type ProcessSlot struct {
|
|
// Generic
|
|
|
|
Slot int // The slot number.
|
|
Epoch int // The epoch number.
|
|
BlockRoot string // The hex encoded string of the BlockRoot.
|
|
StateRoot string // The hex encoded string of the StateRoot.
|
|
ParentBlockRoot string // The hex encoded string of the parent block.
|
|
Status string // The status of the block
|
|
HeadOrHistoric string // Is this the head or a historic slot. This is critical when trying to analyze errors and skipped slots.
|
|
Db sql.Database // The DB object used to write to the DB.
|
|
Metrics *BeaconClientMetrics // An object to keep track of the beaconclient metrics
|
|
PerformanceMetrics PerformanceMetrics // An object to keep track of performance metrics.
|
|
// BeaconBlock
|
|
|
|
SszSignedBeaconBlock []byte // The entire SSZ encoded SignedBeaconBlock
|
|
FullSignedBeaconBlock *SignedBeaconBlock // The unmarshaled BeaconState object, the unmarshalling could have errors.
|
|
|
|
// BeaconState
|
|
FullBeaconState *BeaconState // The unmarshaled BeaconState object, the unmarshalling could have errors.
|
|
SszBeaconState []byte // The entire SSZ encoded BeaconState
|
|
|
|
// DB Write objects
|
|
DbSlotsModel *DbSlots // The model being written to the slots table.
|
|
DbSignedBeaconBlockModel *DbSignedBeaconBlock // The model being written to the signed_block table.
|
|
DbBeaconState *DbBeaconState // The model being written to the state table.
|
|
}
|
|
|
|
type PerformanceMetrics struct {
|
|
BeaconNodeBlockRetrievalTime time.Duration // How long it took to get the BeaconBlock from the Beacon Node.
|
|
BeaconNodeStateRetrievalTime time.Duration // How long it took to get the BeaconState from the Beacon Node.
|
|
ParseBeaconObjectForHash time.Duration // How long it took to get some information from the beacon objects.
|
|
CheckDbPreProcessing time.Duration // How long it takes to check the DB before processing a block.
|
|
CreateDbWriteObject time.Duration // How long it takes to create a DB write object.
|
|
TransactSlotOnly time.Duration // How long it takes to transact the slot information only.
|
|
CheckReorg time.Duration // How long it takes to check for Reorgs
|
|
CommitTransaction time.Duration // How long it takes to commit the final transaction.
|
|
TotalDbTransaction time.Duration // How long it takes from start to committing the entire DB transaction.
|
|
TotalProcessing time.Duration // How long it took to process the entire slot.
|
|
}
|
|
|
|
// This function will do all the work to process the slot and write it to the DB.
|
|
// It will return the error and error process. The error process is used for providing reach detail to the
|
|
// known_gaps table.
|
|
func processFullSlot(
|
|
ctx context.Context,
|
|
slot int,
|
|
blockRoot string,
|
|
stateRoot string,
|
|
previousSlot int,
|
|
previousBlockRoot string,
|
|
knownGapsTableIncrement int,
|
|
headOrHistoric string,
|
|
spd *SlotProcessingDetails) (error, string) {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, ""
|
|
default:
|
|
totalStart := time.Now()
|
|
ps := &ProcessSlot{
|
|
Slot: slot,
|
|
BlockRoot: blockRoot,
|
|
StateRoot: stateRoot,
|
|
HeadOrHistoric: headOrHistoric,
|
|
Db: spd.Db,
|
|
Metrics: spd.Metrics,
|
|
PerformanceMetrics: PerformanceMetrics{
|
|
BeaconNodeBlockRetrievalTime: 0,
|
|
BeaconNodeStateRetrievalTime: 0,
|
|
ParseBeaconObjectForHash: 0,
|
|
CheckDbPreProcessing: 0,
|
|
CreateDbWriteObject: 0,
|
|
TransactSlotOnly: 0,
|
|
CheckReorg: 0,
|
|
CommitTransaction: 0,
|
|
TotalDbTransaction: 0,
|
|
TotalProcessing: 0,
|
|
},
|
|
}
|
|
|
|
g, _ := errgroup.WithContext(context.Background())
|
|
|
|
if spd.PerformBeaconStateProcessing {
|
|
// Get the BeaconState.
|
|
g.Go(func() error {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
default:
|
|
start := time.Now()
|
|
err := ps.getBeaconState(spd.ServerEndpoint)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ps.PerformanceMetrics.BeaconNodeStateRetrievalTime = time.Since(start)
|
|
return nil
|
|
}
|
|
})
|
|
}
|
|
|
|
if spd.PerformBeaconBlockProcessing {
|
|
// Get the SignedBeaconBlock.
|
|
g.Go(func() error {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
default:
|
|
start := time.Now()
|
|
err := ps.getSignedBeaconBlock(spd.ServerEndpoint)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ps.PerformanceMetrics.BeaconNodeBlockRetrievalTime = time.Since(start)
|
|
return nil
|
|
}
|
|
})
|
|
}
|
|
|
|
if err := g.Wait(); err != nil {
|
|
return err, "processSlot"
|
|
}
|
|
|
|
parseBeaconTime := time.Now()
|
|
finalBlockRoot, finalStateRoot, finalEth1DataBlockHash, err := ps.provideFinalHash()
|
|
if err != nil {
|
|
return err, "CalculateBlockRoot"
|
|
}
|
|
ps.PerformanceMetrics.ParseBeaconObjectForHash = time.Since(parseBeaconTime)
|
|
|
|
if spd.CheckDb {
|
|
checkDbTime := time.Now()
|
|
inDb, err := IsSlotInDb(ctx, ps.Db, strconv.Itoa(ps.Slot), finalBlockRoot, finalStateRoot)
|
|
if err != nil {
|
|
return err, "checkDb"
|
|
}
|
|
if inDb {
|
|
log.WithField("slot", slot).Info("Slot already in the DB.")
|
|
return nil, ""
|
|
}
|
|
ps.PerformanceMetrics.CheckDbPreProcessing = time.Since(checkDbTime)
|
|
}
|
|
|
|
// Get this object ready to write
|
|
createDbWriteTime := time.Now()
|
|
dw, err := ps.createWriteObjects(finalBlockRoot, finalStateRoot, finalEth1DataBlockHash)
|
|
if err != nil {
|
|
return err, "blockRoot"
|
|
}
|
|
ps.PerformanceMetrics.CreateDbWriteObject = time.Since(createDbWriteTime)
|
|
|
|
// Write the object to the DB.
|
|
dbFullTransactionTime := time.Now()
|
|
defer func() {
|
|
err := dw.Tx.Rollback(dw.Ctx)
|
|
if err != nil && err != pgx.ErrTxClosed {
|
|
loghelper.LogError(err).Error("We were unable to Rollback a transaction")
|
|
}
|
|
}()
|
|
|
|
transactionTime := time.Now()
|
|
err = dw.transactFullSlot()
|
|
if err != nil {
|
|
return err, "processSlot"
|
|
}
|
|
ps.PerformanceMetrics.TransactSlotOnly = time.Since(transactionTime)
|
|
|
|
// Handle any reorgs or skipped slots.
|
|
reorgTime := time.Now()
|
|
headOrHistoric = strings.ToLower(headOrHistoric)
|
|
if headOrHistoric != "head" && headOrHistoric != "historic" {
|
|
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)
|
|
}
|
|
ps.PerformanceMetrics.CheckReorg = time.Since(reorgTime)
|
|
|
|
// Commit the transaction
|
|
commitTime := time.Now()
|
|
if err = dw.Tx.Commit(dw.Ctx); err != nil {
|
|
return err, "transactionCommit"
|
|
}
|
|
ps.PerformanceMetrics.CommitTransaction = time.Since(commitTime)
|
|
|
|
// Total metric capture time.
|
|
ps.PerformanceMetrics.TotalDbTransaction = time.Since(dbFullTransactionTime)
|
|
ps.PerformanceMetrics.TotalProcessing = time.Since(totalStart)
|
|
|
|
log.WithFields(log.Fields{
|
|
"slot": slot,
|
|
"performanceMetrics": fmt.Sprintf("%+v\n", ps.PerformanceMetrics),
|
|
}).Debug("Performance Metric output!")
|
|
|
|
return nil, ""
|
|
}
|
|
}
|
|
|
|
// Handle a slot that is at head. A wrapper function for calling `handleFullSlot`.
|
|
func processHeadSlot(slot int, blockRoot string, stateRoot string, spd SlotProcessingDetails) {
|
|
// Get the knownGaps at startUp
|
|
if spd.PreviousSlot == 0 && spd.PreviousBlockRoot == "" {
|
|
writeStartUpGaps(spd.Db, spd.KnownGapTableIncrement, slot, spd.Metrics)
|
|
}
|
|
// TODO(telackey): Why context.Background()?
|
|
err, errReason := processFullSlot(context.Background(), slot, blockRoot, stateRoot,
|
|
spd.PreviousSlot, spd.PreviousBlockRoot, spd.KnownGapTableIncrement, "head", &spd)
|
|
if err != nil {
|
|
writeKnownGaps(spd.Db, spd.KnownGapTableIncrement, slot, slot, err, errReason, spd.Metrics)
|
|
}
|
|
}
|
|
|
|
// Handle a historic slot. A wrapper function for calling `handleFullSlot`.
|
|
func handleHistoricSlot(ctx context.Context, slot int, spd SlotProcessingDetails) (error, string) {
|
|
return processFullSlot(ctx, slot, "", "", 0, "",
|
|
1, "historic", &spd)
|
|
}
|
|
|
|
// Update the SszSignedBeaconBlock and FullSignedBeaconBlock object with their respective values.
|
|
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
|
|
sszSignedBeaconBlock, rc, err := querySsz(blockEndpoint, strconv.Itoa(ps.Slot))
|
|
|
|
if err != nil || rc != 200 {
|
|
loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error("Unable to properly query the slot.")
|
|
ps.FullSignedBeaconBlock = nil
|
|
ps.SszSignedBeaconBlock = []byte{}
|
|
ps.ParentBlockRoot = ""
|
|
ps.Status = "skipped"
|
|
return err
|
|
}
|
|
|
|
var signedBeaconBlock SignedBeaconBlock
|
|
err = signedBeaconBlock.UnmarshalSSZ(sszSignedBeaconBlock)
|
|
if err != nil {
|
|
loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error("Unable to unmarshal SignedBeaconBlock for slot.")
|
|
ps.FullSignedBeaconBlock = nil
|
|
ps.SszSignedBeaconBlock = []byte{}
|
|
ps.ParentBlockRoot = ""
|
|
ps.Status = "skipped"
|
|
return err
|
|
}
|
|
|
|
ps.FullSignedBeaconBlock = &signedBeaconBlock
|
|
ps.SszSignedBeaconBlock = sszSignedBeaconBlock
|
|
|
|
ps.ParentBlockRoot = toHex(ps.FullSignedBeaconBlock.Block().ParentRoot())
|
|
return nil
|
|
}
|
|
|
|
// Update the SszBeaconState and FullBeaconState object with their respective values.
|
|
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
|
|
sszBeaconState, _, err := querySsz(stateEndpoint, strconv.Itoa(ps.Slot))
|
|
if err != nil {
|
|
loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error("Unable to properly query the BeaconState.")
|
|
return err
|
|
}
|
|
|
|
var beaconState BeaconState
|
|
err = beaconState.UnmarshalSSZ(sszBeaconState)
|
|
if err != nil {
|
|
loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error("Unable to unmarshal the 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) {
|
|
if nil == ps.FullSignedBeaconBlock {
|
|
log.Debug("Can't check block root, no current block.")
|
|
return
|
|
}
|
|
parentRoot := toHex(ps.FullSignedBeaconBlock.Block().ParentRoot())
|
|
slot := ps.Slot
|
|
if previousSlot == slot {
|
|
log.WithFields(log.Fields{
|
|
"slot": slot,
|
|
"fork": true,
|
|
}).Warn("A fork occurred! The previous slot and current slot match.")
|
|
transactReorgs(tx, ctx, strconv.Itoa(ps.Slot), ps.BlockRoot, ps.Metrics)
|
|
} else if previousSlot > slot {
|
|
log.WithFields(log.Fields{
|
|
"previousSlot": previousSlot,
|
|
"curSlot": slot,
|
|
}).Warn("We noticed the previous slot is greater than the current slot.")
|
|
} else if previousSlot+1 != slot {
|
|
log.WithFields(log.Fields{
|
|
"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)
|
|
} else if previousBlockRoot != parentRoot {
|
|
log.WithFields(log.Fields{
|
|
"previousBlockRoot": previousBlockRoot,
|
|
"currentBlockParent": parentRoot,
|
|
}).Error("The previousBlockRoot does not match the current blocks parent, an unprocessed fork might have occurred.")
|
|
transactReorgs(tx, ctx, strconv.Itoa(previousSlot), parentRoot, ps.Metrics)
|
|
} else {
|
|
log.Debug("Previous Slot and Current Slot are one distance from each other.")
|
|
}
|
|
}
|
|
|
|
// Transforms all the raw data into DB models that can be written to the DB.
|
|
func (ps *ProcessSlot) createWriteObjects(blockRoot, stateRoot, eth1DataBlockHash string) (*DatabaseWriter, error) {
|
|
var status string
|
|
if ps.Status != "" {
|
|
status = ps.Status
|
|
} else {
|
|
status = "proposed"
|
|
}
|
|
|
|
dw, err := CreateDatabaseWrite(ps.Db, ps.Slot, stateRoot, blockRoot, ps.ParentBlockRoot, eth1DataBlockHash, status, &ps.SszSignedBeaconBlock, &ps.SszBeaconState, ps.Metrics)
|
|
if err != nil {
|
|
return dw, err
|
|
}
|
|
|
|
return dw, nil
|
|
}
|
|
|
|
// This function will return the final blockRoot, stateRoot, and eth1DataBlockHash that will be
|
|
// used to write to a DB
|
|
func (ps *ProcessSlot) provideFinalHash() (string, string, string, error) {
|
|
var (
|
|
stateRoot string
|
|
blockRoot string
|
|
eth1DataBlockHash string
|
|
)
|
|
if ps.Status == "skipped" {
|
|
stateRoot = ""
|
|
blockRoot = ""
|
|
eth1DataBlockHash = ""
|
|
} else {
|
|
if ps.StateRoot != "" {
|
|
stateRoot = ps.StateRoot
|
|
} else {
|
|
if nil != ps.FullSignedBeaconBlock {
|
|
stateRoot = toHex(ps.FullSignedBeaconBlock.Block().StateRoot())
|
|
log.Debug("BeaconBlock StateRoot: ", stateRoot)
|
|
} else {
|
|
log.Debug("BeaconBlock StateRoot: <nil beacon block>")
|
|
}
|
|
}
|
|
|
|
if ps.BlockRoot != "" {
|
|
blockRoot = ps.BlockRoot
|
|
} else {
|
|
if nil != ps.FullSignedBeaconBlock {
|
|
rawBlockRoot := ps.FullSignedBeaconBlock.Block().HashTreeRoot()
|
|
blockRoot = toHex(rawBlockRoot)
|
|
log.WithFields(log.Fields{"blockRoot": blockRoot}).Debug("Block Root from ssz")
|
|
} else {
|
|
log.Debug("BeaconBlock HashTreeRoot: <nil beacon block>")
|
|
}
|
|
}
|
|
if nil != ps.FullSignedBeaconBlock {
|
|
eth1DataBlockHash = toHex(ps.FullSignedBeaconBlock.Block().Body().Eth1Data().BlockHash)
|
|
}
|
|
}
|
|
return blockRoot, stateRoot, eth1DataBlockHash, nil
|
|
}
|
|
|
|
func toHex(r [32]byte) string {
|
|
return "0x" + hex.EncodeToString(r[:])
|
|
}
|