eth-statediff-service/pkg/service.go
2023-09-02 01:58:47 +08:00

445 lines
14 KiB
Go

// Copyright © 2020 Vulcanize, Inc
//
// 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/>.
package statediff
import (
"bytes"
"fmt"
"math/big"
"sync"
"time"
"github.com/cerc-io/plugeth-statediff"
"github.com/cerc-io/plugeth-statediff/adapt"
"github.com/cerc-io/plugeth-statediff/indexer/interfaces"
sdtypes "github.com/cerc-io/plugeth-statediff/types"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
"github.com/sirupsen/logrus"
"github.com/cerc-io/eth-statediff-service/pkg/prom"
)
const defaultQueueSize = 1024
// Service is the underlying struct for the state diffing service
type Service struct {
// Used to build the state diff objects
Builder statediff.Builder
// Used to read data from LevelDB
lvlDBReader Reader
// Used to signal shutdown of the service
quitChan chan struct{}
// Interface for publishing statediffs as PG-IPLD objects
indexer interfaces.StateDiffIndexer
// range queue
queue chan RangeRequest
// number of ranges we can work over concurrently
workers uint
// ranges configured locally
preruns []RangeRequest
}
// NewStateDiffService creates a new Service
func NewStateDiffService(lvlDBReader Reader, indexer interfaces.StateDiffIndexer, conf ServiceConfig) *Service {
builder := statediff.NewBuilder(adapt.GethStateView(lvlDBReader.StateDB()))
builder.SetSubtrieWorkers(conf.TrieWorkers)
if conf.WorkerQueueSize == 0 {
conf.WorkerQueueSize = defaultQueueSize
}
return &Service{
lvlDBReader: lvlDBReader,
Builder: builder,
indexer: indexer,
workers: conf.ServiceWorkers,
queue: make(chan RangeRequest, conf.WorkerQueueSize),
preruns: conf.PreRuns,
}
}
// Protocols exports the services p2p protocols, this service has none
func (sds *Service) Protocols() []p2p.Protocol {
return []p2p.Protocol{}
}
// APIs returns the RPC descriptors the Service offers
func (sds *Service) APIs() []rpc.API {
return []rpc.API{
{
Namespace: APIName,
Version: APIVersion,
Service: NewPublicStateDiffAPI(sds),
Public: true,
},
}
}
func segmentRange(workers, start, stop uint64, params statediff.Params) []RangeRequest {
segmentSize := ((stop - start) + 1) / workers
remainder := ((stop - start) + 1) % workers
numOfSegments := workers
if remainder > 0 {
numOfSegments++
}
segments := make([]RangeRequest, numOfSegments)
for i := range segments {
end := start + segmentSize - 1
if end > stop {
end = stop
}
segments[i] = RangeRequest{start, end, params}
start = end + 1
}
return segments
}
// Run does a one-off processing run on the provided RangeRequests + any pre-runs, exiting afterwards
func (sds *Service) Run(rngs []RangeRequest, parallel bool) error {
for _, preRun := range sds.preruns {
// if the rangeSize is smaller than the number of workers
// make sure we do synchronous processing to avoid quantization issues
rangeSize := (preRun.Stop - preRun.Start) + 1
numWorkers := uint64(sds.workers)
if rangeSize < numWorkers {
parallel = false
}
if parallel {
logrus.Infof("parallel processing prerun range (%d, %d) (%d blocks) divided into %d sized chunks with %d workers", preRun.Start, preRun.Stop,
rangeSize, rangeSize/numWorkers, numWorkers)
workChan := make(chan RangeRequest)
quitChan := make(chan struct{})
// spin up numWorkers number of worker goroutines
wg := new(sync.WaitGroup)
for i := 0; i < int(numWorkers); i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
for {
select {
case workerSegment := <-workChan:
for j := workerSegment.Start; j <= workerSegment.Stop; j++ {
if err := sds.WriteStateDiffAt(j, workerSegment.Params); err != nil {
logrus.Errorf("error writing statediff at height %d in range (%d, %d) : %v", id, workerSegment.Start, workerSegment.Stop, err)
}
}
logrus.Infof("prerun worker %d finished processing range (%d, %d)", id, workerSegment.Start, workerSegment.Stop)
case <-quitChan:
return
}
}
}(i)
}
// break range up into segments
segments := segmentRange(numWorkers, preRun.Start, preRun.Stop, preRun.Params)
// send the segments to the work channel
for _, segment := range segments {
workChan <- segment
}
close(quitChan)
wg.Wait()
} else {
logrus.Infof("sequential processing prerun range (%d, %d)", preRun.Start, preRun.Stop)
for i := preRun.Start; i <= preRun.Stop; i++ {
if err := sds.WriteStateDiffAt(i, preRun.Params); err != nil {
return fmt.Errorf("error writing statediff at height %d in range (%d, %d) : %v", i, preRun.Start, preRun.Stop, err)
}
}
}
}
sds.preruns = nil
// At present this code is never called so we have not written the parallel version:
for _, rng := range rngs {
logrus.Infof("processing requested range (%d, %d)", rng.Start, rng.Stop)
for i := rng.Start; i <= rng.Stop; i++ {
if err := sds.WriteStateDiffAt(i, rng.Params); err != nil {
return fmt.Errorf("error writing statediff at height %d in range (%d, %d) : %v", i, rng.Start, rng.Stop, err)
}
}
}
return nil
}
// Loop is an empty service loop for awaiting rpc requests
func (sds *Service) Loop(wg *sync.WaitGroup) error {
if sds.quitChan != nil {
return fmt.Errorf("service loop is already running")
}
sds.quitChan = make(chan struct{})
for i := 0; i < int(sds.workers); i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
for {
select {
case blockRange := <-sds.queue:
log := logrus.WithField("range", blockRange).WithField("worker", id)
log.Debug("processing range")
prom.DecQueuedRanges()
for j := blockRange.Start; j <= blockRange.Stop; j++ {
if err := sds.WriteStateDiffAt(j, blockRange.Params); err != nil {
log.Errorf("error writing statediff at block %d: %v", j, err)
}
select {
case <-sds.quitChan:
log.Infof("closing service worker (last processed block: %d)", j)
return
default:
log.Infof("Finished processing block %d", j)
}
}
log.Debugf("Finished processing range")
case <-sds.quitChan:
logrus.Debugf("closing the statediff service loop worker %d", id)
return
}
}
}(i)
}
for _, preRun := range sds.preruns {
if err := sds.WriteStateDiffsInRange(preRun.Start, preRun.Stop, preRun.Params); err != nil {
close(sds.quitChan)
return err
}
}
return nil
}
// StateDiffAt returns a state diff object payload at the specific blockheight
// This operation cannot be performed back past the point of db pruning; it requires an archival node for historical data
func (sds *Service) StateDiffAt(blockNumber uint64, params statediff.Params) (*statediff.Payload, error) {
currentBlock, err := sds.lvlDBReader.GetBlockByNumber(blockNumber)
if err != nil {
return nil, err
}
logrus.Infof("sending state diff at block %d", blockNumber)
// compute leaf paths of watched addresses in the params
params.ComputeWatchedAddressesLeafPaths()
if blockNumber == 0 {
return sds.processStateDiff(currentBlock, common.Hash{}, params)
}
parentBlock, err := sds.lvlDBReader.GetBlockByHash(currentBlock.ParentHash())
if err != nil {
return nil, err
}
return sds.processStateDiff(currentBlock, parentBlock.Root(), params)
}
// StateDiffFor returns a state diff object payload for the specific blockhash
// This operation cannot be performed back past the point of db pruning; it requires an archival node for historical data
func (sds *Service) StateDiffFor(blockHash common.Hash, params statediff.Params) (*statediff.Payload, error) {
currentBlock, err := sds.lvlDBReader.GetBlockByHash(blockHash)
if err != nil {
return nil, err
}
logrus.Infof("sending state diff at block %s", blockHash)
// compute leaf paths of watched addresses in the params
params.ComputeWatchedAddressesLeafPaths()
if currentBlock.NumberU64() == 0 {
return sds.processStateDiff(currentBlock, common.Hash{}, params)
}
parentBlock, err := sds.lvlDBReader.GetBlockByHash(currentBlock.ParentHash())
if err != nil {
return nil, err
}
return sds.processStateDiff(currentBlock, parentBlock.Root(), params)
}
// processStateDiff method builds the state diff payload from the current block, parent state root, and provided params
func (sds *Service) processStateDiff(currentBlock *types.Block, parentRoot common.Hash, params statediff.Params) (*statediff.Payload, error) {
stateDiff, err := sds.Builder.BuildStateDiffObject(statediff.Args{
BlockHash: currentBlock.Hash(),
BlockNumber: currentBlock.Number(),
OldStateRoot: parentRoot,
NewStateRoot: currentBlock.Root(),
}, params)
if err != nil {
return nil, err
}
stateDiffRlp, err := rlp.EncodeToBytes(&stateDiff)
if err != nil {
return nil, err
}
logrus.Infof("state diff object at block %d is %d bytes in length", currentBlock.Number().Uint64(), len(stateDiffRlp))
return sds.newPayload(stateDiffRlp, currentBlock, params)
}
func (sds *Service) newPayload(stateObject []byte, block *types.Block, params statediff.Params) (*statediff.Payload, error) {
payload := &statediff.Payload{
StateObjectRlp: stateObject,
}
if params.IncludeBlock {
blockBuff := new(bytes.Buffer)
if err := block.EncodeRLP(blockBuff); err != nil {
return nil, err
}
payload.BlockRlp = blockBuff.Bytes()
}
if params.IncludeTD {
var err error
payload.TotalDifficulty, err = sds.lvlDBReader.GetTdByHash(block.Hash())
if err != nil {
return nil, err
}
}
if params.IncludeReceipts {
receiptBuff := new(bytes.Buffer)
receipts, err := sds.lvlDBReader.GetReceiptsByHash(block.Hash())
if err != nil {
return nil, err
}
if err := rlp.Encode(receiptBuff, receipts); err != nil {
return nil, err
}
payload.ReceiptsRlp = receiptBuff.Bytes()
}
return payload, nil
}
// Start is used to begin the service
func (sds *Service) Start() error {
logrus.Info("starting statediff service")
return sds.Loop(new(sync.WaitGroup))
}
// Stop is used to close down the service
func (sds *Service) Stop() error {
logrus.Info("stopping statediff service")
close(sds.quitChan)
return nil
}
// WriteStateDiffAt writes a state diff at the specific blockheight directly to the database
// This operation cannot be performed back past the point of db pruning; it requires an archival node
// for historical data
func (sds *Service) WriteStateDiffAt(blockNumber uint64, params statediff.Params) error {
logrus.Infof("Writing state diff at block %d", blockNumber)
t := time.Now()
currentBlock, err := sds.lvlDBReader.GetBlockByNumber(blockNumber)
if err != nil {
return err
}
// compute leaf paths of watched addresses in the params
params.ComputeWatchedAddressesLeafPaths()
parentRoot := common.Hash{}
if blockNumber != 0 {
parentBlock, err := sds.lvlDBReader.GetBlockByHash(currentBlock.ParentHash())
if err != nil {
return err
}
parentRoot = parentBlock.Root()
}
return sds.writeStateDiff(currentBlock, parentRoot, params, t)
}
// WriteStateDiffFor writes a state diff for the specific blockHash directly to the database
// This operation cannot be performed back past the point of db pruning; it requires an archival node
// for historical data
func (sds *Service) WriteStateDiffFor(blockHash common.Hash, params statediff.Params) error {
logrus.Infof("Writing state diff for block %s", blockHash)
t := time.Now()
currentBlock, err := sds.lvlDBReader.GetBlockByHash(blockHash)
if err != nil {
return err
}
// compute leaf paths of watched addresses in the params
params.ComputeWatchedAddressesLeafPaths()
parentRoot := common.Hash{}
if currentBlock.NumberU64() != 0 {
parentBlock, err := sds.lvlDBReader.GetBlockByHash(currentBlock.ParentHash())
if err != nil {
return err
}
parentRoot = parentBlock.Root()
}
return sds.writeStateDiff(currentBlock, parentRoot, params, t)
}
// Writes a state diff from the current block, parent state root, and provided params
func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, params statediff.Params, t time.Time) error {
var totalDifficulty *big.Int
var receipts types.Receipts
var err error
if params.IncludeTD {
totalDifficulty, err = sds.lvlDBReader.GetTdByHash(block.Hash())
}
if err != nil {
return err
}
if params.IncludeReceipts {
receipts, err = sds.lvlDBReader.GetReceiptsByHash(block.Hash())
}
if err != nil {
return err
}
height := block.Number().Int64()
prom.SetLastLoadedHeight(height)
prom.SetTimeMetric(prom.T_BLOCK_LOAD, time.Now().Sub(t))
t = time.Now()
tx, err := sds.indexer.PushBlock(block, receipts, totalDifficulty)
if err != nil {
return err
}
// defer handling of commit/rollback for any return case
output := func(node sdtypes.StateLeafNode) error {
return sds.indexer.PushStateNode(tx, node, block.Hash().String())
}
codeOutput := func(c sdtypes.IPLD) error {
return sds.indexer.PushIPLD(tx, c)
}
prom.SetTimeMetric(prom.T_BLOCK_PROCESSING, time.Now().Sub(t))
t = time.Now()
err = sds.Builder.WriteStateDiff(statediff.Args{
NewStateRoot: block.Root(),
OldStateRoot: parentRoot,
BlockNumber: block.Number(),
BlockHash: block.Hash(),
}, params, output, codeOutput)
prom.SetTimeMetric(prom.T_STATE_PROCESSING, time.Now().Sub(t))
t = time.Now()
err = tx.Submit()
prom.SetLastProcessedHeight(height)
prom.SetTimeMetric(prom.T_POSTGRES_TX_COMMIT, time.Now().Sub(t))
return err
}
// WriteStateDiffsInRange adds a RangeRequest to the work queue
func (sds *Service) WriteStateDiffsInRange(start, stop uint64, params statediff.Params) error {
if stop < start {
return fmt.Errorf("invalid block range (%d, %d): stop height must be greater or equal to start height", start, stop)
}
blocked := time.NewTimer(30 * time.Second)
select {
case sds.queue <- RangeRequest{Start: start, Stop: stop, Params: params}:
prom.IncQueuedRanges()
logrus.Infof("Added range (%d, %d) to the worker queue", start, stop)
return nil
case <-blocked.C:
return fmt.Errorf("unable to add range (%d, %d) to the worker queue", start, stop)
}
}