go-ethereum/statediff/service.go
Elizabeth 1162162c0a Statediffing geth
* Write state diff to CSV (#2)

* port statediff from 9b7fd9af80/statediff/statediff.go; minor fixes

* integrating state diff extracting, building, and persisting into geth processes

* work towards persisting created statediffs in ipfs; based off github.com/vulcanize/eth-block-extractor

* Add a state diff service

* Remove diff extractor from blockchain

* Update imports

* Move statediff on/off check to geth cmd config

* Update starting state diff service

* Add debugging logs for creating diff

* Add statediff extractor and builder tests and small refactoring

* Start to write statediff to a CSV

* Restructure statediff directory

* Pull CSV publishing methods into their own file

* Reformatting due to go fmt

* Add gomega to vendor dir

* Remove testing focuses

* Update statediff tests to use golang test pkg

instead of ginkgo

- builder_test
- extractor_test
- publisher_test

* Use hexutil.Encode instead of deprecated common.ToHex

* Remove OldValue from DiffBigInt and DiffUint64 fields

* Update builder test

* Remove old storage value from updated accounts

* Remove old values from created/deleted accounts

* Update publisher to account for only storing current account values

* Update service loop and fetching previous block

* Update testing

- remove statediff ginkgo test suite file
- move mocks to their own dir

* Updates per go fmt

* Updates to tests

* Pass statediff mode and path in through cli

* Return filename from publisher

* Remove some duplication in builder

* Remove code field from state diff output

this is the contract byte code, and it can still be obtained by querying
the db by the codeHash

* Consolidate acct diff structs for updated & updated/deleted accts

* Include block number in csv filename

* Clean up error logging

* Cleanup formatting, spelling, etc

* Address PR comments

* Add contract address and storage value to csv

* Refactor accumulating account row in csv publisher

* Add DiffStorage struct

* Add storage key to csv

* Address PR comments

* Fix publisher to include rows for accounts that don't have store updates

* Update builder test after merging in release/1.8

* Update test contract to include storage on contract intialization

- so that we're able to test that storage diffing works for created and
deleted accounts (not just updated accounts).

* Factor out a common trie iterator method in builder

* Apply goimports to statediff

* Apply gosimple changes to statediff

* Gracefully exit geth command(#4)

* Statediff for full node (#6)

* Open a trie from the in-memory database

* Use a node's LeafKey as an identifier instead of the address

It was proving difficult to find look the address up from a given path
with a full node (sometimes the value wouldn't exist in the disk db).
So, instead, for now we are using the node's LeafKey with is a Keccak256
hash of the address, so if we know the address we can figure out which
LeafKey it matches up to.

* Make sure that statediff has been processed before pruning

* Use blockchain stateCache.OpenTrie for storage diffs

* Clean up log lines and remove unnecessary fields from builder

* Apply go fmt changes

* Add a sleep to the blockchain test

* Address PR comments

* Address PR comments

* refactoring/reorganizing packages

* refactoring statediff builder and types and adjusted to relay proofs and paths (still need to make this optional)

* refactoring state diff service and adding api which allows for streaming state diff payloads over an rpc websocket subscription

* make proofs and paths optional + compress service loop into single for loop (may be missing something here)

* option to process intermediate nodes

* make state diff rlp serializable

* cli parameter to limit statediffing to select account addresses + test

* review fixes and fixes for issues ran into in integration

* review fixes; proper method signature for api; adjust service so that statediff processing is halted/paused until there is at least one subscriber listening for the results

* adjust buffering to improve stability; doc.go; fix notifier
err handling

* relay receipts with the rest of the data + review fixes/changes

* rpc method to get statediff at specific block; requires archival node or the block be within the pruning range

* review fixes

* fixes after rebase

* statediff verison meta

* fix linter issues

* include total difficulty to the payload

* fix state diff builder: emit actual leaf nodes instead of value nodes; diff on the leaf not on the value; emit correct path for intermediate nodes

* adjust statediff builder tests to changes and extend to test intermediate nodes; golint

* add genesis block to test; handle block 0 in StateDiffAt

* rlp files for mainnet blocks 0-3, for tests

* builder test on mainnet blocks

* common.BytesToHash(path) => crypto.Keaccak256(hash) in builder; BytesToHash produces same hash for e.g. []byte{} and []byte{\x00} - prefix \x00 steps are inconsequential to the hash result

* complete tests for early mainnet blocks

* diff type for representing deleted accounts

* fix builder so that we handle account deletions properly and properly diff storage when an account is moved to a new path; update params

* remove cli params; moving them to subscriber defined

* remove unneeded bc methods

* update service and api; statediffing params are now defined by user through api rather than by service provider by cli

* update top level tests

* add ability to watch specific storage slots (leaf keys) only

* comments; explain logic

* update mainnet blocks test

* update api_test.go

* storage leafkey filter test

* cleanup chain maker

* adjust chain maker for tests to add an empty account in block1 and switch to EIP-158 afterwards (now we just need to generate enough accounts until one causes the empty account to be touched and removed post-EIP-158 so we can simulate and test that process...); also added 2 new blocks where more contract storage is set and old slots are set to zero so they are removed so we can test that

* found an account whose creation causes the empty account to be moved to a new path; this should count as 'touching; the empty account and cause it to be removed according to eip-158... but it doesn't

* use new contract in unit tests that has self-destruct ability, so we can test eip-158 since simply moving an account to new path doesn't count as 'touchin' it

* handle storage deletions

* tests for eip-158 account removal and storage value deletions; there is one edge case left to test where we remove 1 account when only two exist such that the remaining account is moved up and replaces the root branch node

* finish testing known edge cases

* add endpoint to fetch all state and storage nodes at a given blockheight; useful for generating a recent atate cache/snapshot that we can diff forward from rather than needing to collect all diffs from genesis

* test for state trie builder

* minor changes/fixes

* update version meta

* if statediffing is on, lock tries in triedb until the statediffing service signals they are done using them

* update version meta

* fix mock blockchain; golint; bump patch

* increase maxRequestContentLength; bump patch

* log the sizes of the state objects we are sending

* CI build (#20)

* CI: run build on PR and on push to master

* CI: debug building geth

* CI: fix coping file

* CI: fix coping file v2

* CI: temporary upload file to release asset

* CI: get release upload_url by tag, upload asset to current relase

* CI: fix tag name

* fix ci build on statediff_at_anyblock-1.9.11 branch

* fix publishing assets in release

* bump version meta

* use context deadline for timeout in eth_call

* collect and emit codehash=>code mappings for state objects

* subscription endpoint for retrieving all the codehash=>code mappings that exist at provided height

* bump version meta

* Implement WriteStateDiffAt

* Writes state diffs directly to postgres

* Adds CLI flags to configure PG

* Refactors builder output with callbacks

* Copies refactored postgres handling code from ipld-eth-indexer

* rename PostgresCIDWriter.{index->upsert}*

* less ambiguous

* go.mod update

* rm unused

* cleanup

* output code & codehash iteratively

* had to rf some types for this

* prometheus metrics output

* duplicate recent eth-indexer changes

* migrations and metrics...

* [wip] prom.Init() here? another CLI flag?

* cleanup

* tidy & DRY

* statediff WriteLoop service + CLI flag

* [wip] update test mocks

* todo - do something meaningful to test write loop

* logging

* use geth log

* port tests to go testing

* drop ginkgo/gomega

* fix and cleanup tests

* fail before defer statement

* delete vendor/ dir

* unused

* bump version meta

* fixes after rebase onto 1.9.23

* bump version meta

* fix API registration

* bump version meta

* use golang 1.15.5 version (#34)

* bump version meta; add 0.0.11 branch to actions

* bump version meta; update github actions workflows

* statediff: refactor metrics

* Remove redundant statediff/indexer/prom tooling and use existing
prometheus integration.

* cleanup

* "indexer" namespace for metrics

* add reporting loop for db metrics

* doc

* metrics for statediff stats

* metrics namespace/subsystem = statediff/{indexer,service}

* statediff: use a worker pool (for direct writes)

* fix test

* fix chain event subscription

* log tweaks

* func name

* unused import

* intermediate chain event channel for metrics

* cleanup

* bump version meta

* update github actions; linting

* add poststate and status to receipt ipld indexes

* bump statediff version

* stateDiffFor endpoints for fetching or writing statediff object by blockhash; bump statediff version

* fixes after rebase on to v1.10.1

* update github actions and version meta; go fmt

* add leaf key to removed 'nodes'

* include Postgres migrations and schema

* service documentation

* touching up
2021-04-08 22:03:04 -05:00

659 lines
23 KiB
Go

// Copyright 2019 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library 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 Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package statediff
import (
"bytes"
"math/big"
"strconv"
"sync"
"sync/atomic"
"github.com/ethereum/go-ethereum/eth/ethconfig"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/trie"
ind "github.com/ethereum/go-ethereum/statediff/indexer"
nodeinfo "github.com/ethereum/go-ethereum/statediff/indexer/node"
"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
. "github.com/ethereum/go-ethereum/statediff/types"
)
const chainEventChanSize = 20000
var writeLoopParams = Params{
IntermediateStateNodes: true,
IntermediateStorageNodes: true,
IncludeBlock: true,
IncludeReceipts: true,
IncludeTD: true,
IncludeCode: true,
}
var statediffMetrics = RegisterStatediffMetrics(metrics.DefaultRegistry)
type blockChain interface {
SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
GetBlockByHash(hash common.Hash) *types.Block
GetBlockByNumber(number uint64) *types.Block
GetReceiptsByHash(hash common.Hash) types.Receipts
GetTdByHash(hash common.Hash) *big.Int
UnlockTrie(root common.Hash)
StateCache() state.Database
}
// IService is the state-diffing service interface
type IService interface {
// Start() and Stop()
node.Lifecycle
// Method to getting API(s) for this service
APIs() []rpc.API
// Main event loop for processing state diffs
Loop(chainEventCh chan core.ChainEvent)
// Method to subscribe to receive state diff processing output
Subscribe(id rpc.ID, sub chan<- Payload, quitChan chan<- bool, params Params)
// Method to unsubscribe from state diff processing
Unsubscribe(id rpc.ID) error
// Method to get state diff object at specific block
StateDiffAt(blockNumber uint64, params Params) (*Payload, error)
// Method to get state diff object at specific block
StateDiffFor(blockHash common.Hash, params Params) (*Payload, error)
// Method to get state trie object at specific block
StateTrieAt(blockNumber uint64, params Params) (*Payload, error)
// Method to stream out all code and codehash pairs
StreamCodeAndCodeHash(blockNumber uint64, outChan chan<- CodeAndCodeHash, quitChan chan<- bool)
// Method to write state diff object directly to DB
WriteStateDiffAt(blockNumber uint64, params Params) error
// Method to write state diff object directly to DB
WriteStateDiffFor(blockHash common.Hash, params Params) error
// Event loop for progressively processing and writing diffs directly to DB
WriteLoop(chainEventCh chan core.ChainEvent)
}
// Wraps consructor parameters
type ServiceParams struct {
DBParams *DBParams
// Whether to enable writing state diffs directly to track blochain head
EnableWriteLoop bool
// Size of the worker pool
NumWorkers uint
}
// Service is the underlying struct for the state diffing service
type Service struct {
// Used to sync access to the Subscriptions
sync.Mutex
// Used to build the state diff objects
Builder Builder
// Used to subscribe to chain events (blocks)
BlockChain blockChain
// Used to signal shutdown of the service
QuitChan chan bool
// A mapping of rpc.IDs to their subscription channels, mapped to their subscription type (hash of the Params rlp)
Subscriptions map[common.Hash]map[rpc.ID]Subscription
// A mapping of subscription params rlp hash to the corresponding subscription params
SubscriptionTypes map[common.Hash]Params
// Cache the last block so that we can avoid having to lookup the next block's parent
BlockCache blockCache
// Whether or not we have any subscribers; only if we do, do we processes state diffs
subscribers int32
// Interface for publishing statediffs as PG-IPLD objects
indexer ind.Indexer
// Whether to enable writing state diffs directly to track blochain head
enableWriteLoop bool
// Size of the worker pool
numWorkers uint
}
// Wrap the cached last block for safe access from different service loops
type blockCache struct {
sync.Mutex
blocks map[common.Hash]*types.Block
maxSize uint
}
func NewBlockCache(max uint) blockCache {
return blockCache{
blocks: make(map[common.Hash]*types.Block),
maxSize: max,
}
}
// New creates a new statediff.Service
// func New(stack *node.Node, ethServ *eth.Ethereum, dbParams *DBParams, enableWriteLoop bool) error {
func New(stack *node.Node, ethServ *eth.Ethereum, cfg *ethconfig.Config, params ServiceParams) error {
blockChain := ethServ.BlockChain()
var indexer ind.Indexer
if params.DBParams != nil {
info := nodeinfo.Info{
GenesisBlock: blockChain.Genesis().Hash().Hex(),
NetworkID: strconv.FormatUint(cfg.NetworkId, 10),
ChainID: blockChain.Config().ChainID.Uint64(),
ID: params.DBParams.ID,
ClientName: params.DBParams.ClientName,
}
// TODO: pass max idle, open, lifetime?
db, err := postgres.NewDB(params.DBParams.ConnectionURL, postgres.ConnectionConfig{}, info)
if err != nil {
return err
}
indexer = ind.NewStateDiffIndexer(blockChain.Config(), db)
}
workers := params.NumWorkers
if workers == 0 {
workers = 1
}
sds := &Service{
Mutex: sync.Mutex{},
BlockChain: blockChain,
Builder: NewBuilder(blockChain.StateCache()),
QuitChan: make(chan bool),
Subscriptions: make(map[common.Hash]map[rpc.ID]Subscription),
SubscriptionTypes: make(map[common.Hash]Params),
BlockCache: NewBlockCache(workers),
indexer: indexer,
enableWriteLoop: params.EnableWriteLoop,
numWorkers: workers,
}
stack.RegisterLifecycle(sds)
stack.RegisterAPIs(sds.APIs())
return nil
}
// 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 statediff.Service offers
func (sds *Service) APIs() []rpc.API {
return []rpc.API{
{
Namespace: APIName,
Version: APIVersion,
Service: NewPublicStateDiffAPI(sds),
Public: true,
},
}
}
// Return the parent block of currentBlock, using the cached block if available;
// and cache the passed block
func (lbc *blockCache) getParentBlock(currentBlock *types.Block, bc blockChain) *types.Block {
lbc.Lock()
parentHash := currentBlock.ParentHash()
var parentBlock *types.Block
if block, ok := lbc.blocks[parentHash]; ok {
parentBlock = block
if len(lbc.blocks) > int(lbc.maxSize) {
delete(lbc.blocks, parentHash)
}
} else {
parentBlock = bc.GetBlockByHash(parentHash)
}
lbc.blocks[currentBlock.Hash()] = currentBlock
lbc.Unlock()
return parentBlock
}
type workerParams struct {
chainEventCh <-chan core.ChainEvent
errCh <-chan error
wg *sync.WaitGroup
id uint
}
func (sds *Service) WriteLoop(chainEventCh chan core.ChainEvent) {
chainEventSub := sds.BlockChain.SubscribeChainEvent(chainEventCh)
defer chainEventSub.Unsubscribe()
errCh := chainEventSub.Err()
var wg sync.WaitGroup
// Process metrics for chain events, then forward to workers
chainEventFwd := make(chan core.ChainEvent, chainEventChanSize)
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case chainEvent := <-chainEventCh:
statediffMetrics.lastEventHeight.Update(int64(chainEvent.Block.Number().Uint64()))
statediffMetrics.writeLoopChannelLen.Update(int64(len(chainEventCh)))
chainEventFwd <- chainEvent
case <-sds.QuitChan:
return
}
}
}()
wg.Add(int(sds.numWorkers))
for worker := uint(0); worker < sds.numWorkers; worker++ {
params := workerParams{chainEventCh: chainEventFwd, errCh: errCh, wg: &wg, id: worker}
go sds.writeLoopWorker(params)
}
wg.Wait()
}
func (sds *Service) writeLoopWorker(params workerParams) {
defer params.wg.Done()
for {
select {
//Notify chain event channel of events
case chainEvent := <-params.chainEventCh:
log.Debug("WriteLoop(): chain event received", "event", chainEvent)
currentBlock := chainEvent.Block
parentBlock := sds.BlockCache.getParentBlock(currentBlock, sds.BlockChain)
if parentBlock == nil {
log.Error("Parent block is nil, skipping this block", "block height", currentBlock.Number())
continue
}
log.Info("Writing state diff", "block height", currentBlock.Number().Uint64(), "worker", params.id)
err := sds.writeStateDiff(currentBlock, parentBlock.Root(), writeLoopParams)
if err != nil {
log.Error("statediff.Service.WriteLoop: processing error", "block height", currentBlock.Number().Uint64(), "error", err.Error(), "worker", params.id)
continue
}
// TODO: how to handle with concurrent workers
statediffMetrics.lastStatediffHeight.Update(int64(currentBlock.Number().Uint64()))
case err := <-params.errCh:
log.Warn("Error from chain event subscription", "error", err, "worker", params.id)
sds.close()
return
case <-sds.QuitChan:
log.Info("Quitting the statediff writing process", "worker", params.id)
sds.close()
return
}
}
}
// Loop is the main processing method
func (sds *Service) Loop(chainEventCh chan core.ChainEvent) {
chainEventSub := sds.BlockChain.SubscribeChainEvent(chainEventCh)
defer chainEventSub.Unsubscribe()
errCh := chainEventSub.Err()
for {
select {
//Notify chain event channel of events
case chainEvent := <-chainEventCh:
statediffMetrics.serviceLoopChannelLen.Update(int64(len(chainEventCh)))
log.Debug("Loop(): chain event received", "event", chainEvent)
// if we don't have any subscribers, do not process a statediff
if atomic.LoadInt32(&sds.subscribers) == 0 {
log.Debug("Currently no subscribers to the statediffing service; processing is halted")
continue
}
currentBlock := chainEvent.Block
parentBlock := sds.BlockCache.getParentBlock(currentBlock, sds.BlockChain)
if parentBlock == nil {
log.Error("Parent block is nil, skipping this block", "block height", currentBlock.Number())
continue
}
sds.streamStateDiff(currentBlock, parentBlock.Root())
case err := <-errCh:
log.Warn("Error from chain event subscription", "error", err)
sds.close()
return
case <-sds.QuitChan:
log.Info("Quitting the statediffing process")
sds.close()
return
}
}
}
// streamStateDiff method builds the state diff payload for each subscription according to their subscription type and sends them the result
func (sds *Service) streamStateDiff(currentBlock *types.Block, parentRoot common.Hash) {
sds.Lock()
for ty, subs := range sds.Subscriptions {
params, ok := sds.SubscriptionTypes[ty]
if !ok {
log.Error("no parameter set associated with this subscription", "subscription type", ty.Hex())
sds.closeType(ty)
continue
}
// create payload for this subscription type
payload, err := sds.processStateDiff(currentBlock, parentRoot, params)
if err != nil {
log.Error("statediff processing error", "block height", currentBlock.Number().Uint64(), "parameters", params, "error", err.Error())
continue
}
for id, sub := range subs {
select {
case sub.PayloadChan <- *payload:
log.Debug("sending statediff payload at head", "height", currentBlock.Number(), "subscription id", id)
default:
log.Info("unable to send statediff payload; channel has no receiver", "subscription id", id)
}
}
}
sds.Unlock()
}
// 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 Params) (*Payload, error) {
currentBlock := sds.BlockChain.GetBlockByNumber(blockNumber)
log.Info("sending state diff", "block height", blockNumber)
if blockNumber == 0 {
return sds.processStateDiff(currentBlock, common.Hash{}, params)
}
parentBlock := sds.BlockChain.GetBlockByHash(currentBlock.ParentHash())
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 Params) (*Payload, error) {
currentBlock := sds.BlockChain.GetBlockByHash(blockHash)
log.Info("sending state diff", "block hash", blockHash)
if currentBlock.NumberU64() == 0 {
return sds.processStateDiff(currentBlock, common.Hash{}, params)
}
parentBlock := sds.BlockChain.GetBlockByHash(currentBlock.ParentHash())
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 Params) (*Payload, error) {
stateDiff, err := sds.Builder.BuildStateDiffObject(Args{
NewStateRoot: currentBlock.Root(),
OldStateRoot: parentRoot,
BlockHash: currentBlock.Hash(),
BlockNumber: currentBlock.Number(),
}, params)
// allow dereferencing of parent, keep current locked as it should be the next parent
sds.BlockChain.UnlockTrie(parentRoot)
if err != nil {
return nil, err
}
stateDiffRlp, err := rlp.EncodeToBytes(stateDiff)
if err != nil {
return nil, err
}
log.Info("state diff size", "at block height", currentBlock.Number().Uint64(), "rlp byte size", len(stateDiffRlp))
return sds.newPayload(stateDiffRlp, currentBlock, params)
}
func (sds *Service) newPayload(stateObject []byte, block *types.Block, params Params) (*Payload, error) {
payload := &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 {
payload.TotalDifficulty = sds.BlockChain.GetTdByHash(block.Hash())
}
if params.IncludeReceipts {
receiptBuff := new(bytes.Buffer)
receipts := sds.BlockChain.GetReceiptsByHash(block.Hash())
if err := rlp.Encode(receiptBuff, receipts); err != nil {
return nil, err
}
payload.ReceiptsRlp = receiptBuff.Bytes()
}
return payload, nil
}
// StateTrieAt returns a state trie object payload at the specified blockheight
// This operation cannot be performed back past the point of db pruning; it requires an archival node for historical data
func (sds *Service) StateTrieAt(blockNumber uint64, params Params) (*Payload, error) {
currentBlock := sds.BlockChain.GetBlockByNumber(blockNumber)
log.Info("sending state trie", "block height", blockNumber)
return sds.processStateTrie(currentBlock, params)
}
func (sds *Service) processStateTrie(block *types.Block, params Params) (*Payload, error) {
stateNodes, err := sds.Builder.BuildStateTrieObject(block)
if err != nil {
return nil, err
}
stateTrieRlp, err := rlp.EncodeToBytes(stateNodes)
if err != nil {
return nil, err
}
log.Info("state trie size", "at block height", block.Number().Uint64(), "rlp byte size", len(stateTrieRlp))
return sds.newPayload(stateTrieRlp, block, params)
}
// Subscribe is used by the API to subscribe to the service loop
func (sds *Service) Subscribe(id rpc.ID, sub chan<- Payload, quitChan chan<- bool, params Params) {
log.Info("Subscribing to the statediff service")
if atomic.CompareAndSwapInt32(&sds.subscribers, 0, 1) {
log.Info("State diffing subscription received; beginning statediff processing")
}
// Subscription type is defined as the hash of the rlp-serialized subscription params
by, err := rlp.EncodeToBytes(params)
if err != nil {
log.Error("State diffing params need to be rlp-serializable")
return
}
subscriptionType := crypto.Keccak256Hash(by)
// Add subscriber
sds.Lock()
if sds.Subscriptions[subscriptionType] == nil {
sds.Subscriptions[subscriptionType] = make(map[rpc.ID]Subscription)
}
sds.Subscriptions[subscriptionType][id] = Subscription{
PayloadChan: sub,
QuitChan: quitChan,
}
sds.SubscriptionTypes[subscriptionType] = params
sds.Unlock()
}
// Unsubscribe is used to unsubscribe from the service loop
func (sds *Service) Unsubscribe(id rpc.ID) error {
log.Info("Unsubscribing from the statediff service", "subscription id", id)
sds.Lock()
for ty := range sds.Subscriptions {
delete(sds.Subscriptions[ty], id)
if len(sds.Subscriptions[ty]) == 0 {
// If we removed the last subscription of this type, remove the subscription type outright
delete(sds.Subscriptions, ty)
delete(sds.SubscriptionTypes, ty)
}
}
if len(sds.Subscriptions) == 0 {
if atomic.CompareAndSwapInt32(&sds.subscribers, 1, 0) {
log.Info("No more subscriptions; halting statediff processing")
}
}
sds.Unlock()
return nil
}
// Start is used to begin the service
func (sds *Service) Start() error {
log.Info("Starting statediff service")
chainEventCh := make(chan core.ChainEvent, chainEventChanSize)
go sds.Loop(chainEventCh)
if sds.enableWriteLoop {
log.Info("Starting statediff DB write loop", "params", writeLoopParams)
chainEventCh := make(chan core.ChainEvent, chainEventChanSize)
go sds.WriteLoop(chainEventCh)
}
return nil
}
// Stop is used to close down the service
func (sds *Service) Stop() error {
log.Info("Stopping statediff service")
close(sds.QuitChan)
return nil
}
// close is used to close all listening subscriptions
func (sds *Service) close() {
sds.Lock()
for ty, subs := range sds.Subscriptions {
for id, sub := range subs {
select {
case sub.QuitChan <- true:
log.Info("closing subscription", "id", id)
default:
log.Info("unable to close subscription; channel has no receiver", "subscription id", id)
}
delete(sds.Subscriptions[ty], id)
}
delete(sds.Subscriptions, ty)
delete(sds.SubscriptionTypes, ty)
}
sds.Unlock()
}
// closeType is used to close all subscriptions of given type
// closeType needs to be called with subscription access locked
func (sds *Service) closeType(subType common.Hash) {
subs := sds.Subscriptions[subType]
for id, sub := range subs {
sendNonBlockingQuit(id, sub)
}
delete(sds.Subscriptions, subType)
delete(sds.SubscriptionTypes, subType)
}
func sendNonBlockingQuit(id rpc.ID, sub Subscription) {
select {
case sub.QuitChan <- true:
log.Info("closing subscription", "id", id)
default:
log.Info("unable to close subscription; channel has no receiver", "subscription id", id)
}
}
// StreamCodeAndCodeHash subscription method for extracting all the codehash=>code mappings that exist in the trie at the provided height
func (sds *Service) StreamCodeAndCodeHash(blockNumber uint64, outChan chan<- CodeAndCodeHash, quitChan chan<- bool) {
current := sds.BlockChain.GetBlockByNumber(blockNumber)
log.Info("sending code and codehash", "block height", blockNumber)
currentTrie, err := sds.BlockChain.StateCache().OpenTrie(current.Root())
if err != nil {
log.Error("error creating trie for block", "block height", current.Number(), "err", err)
close(quitChan)
return
}
it := currentTrie.NodeIterator([]byte{})
leafIt := trie.NewIterator(it)
go func() {
defer close(quitChan)
for leafIt.Next() {
select {
case <-sds.QuitChan:
return
default:
}
account := new(state.Account)
if err := rlp.DecodeBytes(leafIt.Value, account); err != nil {
log.Error("error decoding state account", "err", err)
return
}
codeHash := common.BytesToHash(account.CodeHash)
code, err := sds.BlockChain.StateCache().ContractCode(common.Hash{}, codeHash)
if err != nil {
log.Error("error collecting contract code", "err", err)
return
}
outChan <- CodeAndCodeHash{
Hash: codeHash,
Code: code,
}
}
}()
}
// 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 Params) error {
currentBlock := sds.BlockChain.GetBlockByNumber(blockNumber)
parentRoot := common.Hash{}
if blockNumber != 0 {
parentBlock := sds.BlockChain.GetBlockByHash(currentBlock.ParentHash())
parentRoot = parentBlock.Root()
}
return sds.writeStateDiff(currentBlock, parentRoot, params)
}
// 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 Params) error {
currentBlock := sds.BlockChain.GetBlockByHash(blockHash)
parentRoot := common.Hash{}
if currentBlock.NumberU64() != 0 {
parentBlock := sds.BlockChain.GetBlockByHash(currentBlock.ParentHash())
parentRoot = parentBlock.Root()
}
return sds.writeStateDiff(currentBlock, parentRoot, params)
}
// 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 Params) error {
// log.Info("Writing state diff", "block height", block.Number().Uint64())
var totalDifficulty *big.Int
var receipts types.Receipts
if params.IncludeTD {
totalDifficulty = sds.BlockChain.GetTdByHash(block.Hash())
}
if params.IncludeReceipts {
receipts = sds.BlockChain.GetReceiptsByHash(block.Hash())
}
tx, err := sds.indexer.PushBlock(block, receipts, totalDifficulty)
if err != nil {
return err
}
// defer handling of commit/rollback for any return case
defer tx.Close()
output := func(node StateNode) error {
return sds.indexer.PushStateNode(tx, node)
}
codeOutput := func(c CodeAndCodeHash) error {
return sds.indexer.PushCodeAndCodeHash(tx, c)
}
err = sds.Builder.WriteStateDiffObject(StateRoots{
NewStateRoot: block.Root(),
OldStateRoot: parentRoot,
}, params, output, codeOutput)
// allow dereferencing of parent, keep current locked as it should be the next parent
sds.BlockChain.UnlockTrie(parentRoot)
if err != nil {
return err
}
return nil
}