cerc-io/ipld-eth-server
17
0
Fork 1
Code Issues 10 Pull Requests 1 Actions Packages Projects Releases 37 Wiki Activity

cache misses trigger call out to statediffing geth to fill in the gap in Postgres

Browse Source
This commit is contained in:
Ian Norden 2021-02-24 10:50:26 -06:00
parent 7df5bbc99a
commit 211ec12009
16 changed files with 151 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
environments/example.toml
View File

@ -20,6 +20,7 @@
defaultSender = "" # $ETH_DEFAULT_SENDER_ADDR
rpcGasCap = "1000000000000" # $ETH_RPC_GAS_CAP
httpPath = "127.0.0.1:8545" # $ETH_HTTP_PATH
supportsStateDiff = true # $ETH_SUPPORTS_STATEDIFF
nodeID = "arch1" # $ETH_NODE_ID
clientName = "Geth" # $ETH_CLIENT_NAME
genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3" # $ETH_GENESIS_BLOCK

1
go.mod
View File

@ -25,4 +25,5 @@ require (
)
replace github.com/ethereum/go-ethereum v1.9.25 => github.com/vulcanize/go-ethereum v1.9.25-statediff-0.0.14
replace github.com/vulcanize/ipfs-ethdb v0.0.2-alpha => github.com/vulcanize/pg-ipfs-ethdb v0.0.2-alpha

116
pkg/eth/api.go
View File

@ -18,12 +18,15 @@ package eth
import (
"context"
"encoding/json"
"errors"
"fmt"
"math"
"math/big"
"time"
"github.com/ethereum/go-ethereum/statediff"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
@ -51,20 +54,22 @@ type PublicEthAPI struct {
B *Backend
// Remote node for forwarding cache misses
rpc *rpc.Client
ethClient *ethclient.Client
supportsStateDiff bool // Whether or not the remote node supports the statediff_writeStateDiffAt endpoint, if it does we can fill the local cache when we hit a miss
rpc *rpc.Client
ethClient *ethclient.Client
}
// NewPublicEthAPI creates a new PublicEthAPI with the provided underlying Backend
func NewPublicEthAPI(b *Backend, client *rpc.Client) *PublicEthAPI {
func NewPublicEthAPI(b *Backend, client *rpc.Client, supportsStateDiff bool) *PublicEthAPI {
var ethClient *ethclient.Client
if client != nil {
ethClient = ethclient.NewClient(client)
}
return &PublicEthAPI{
B: b,
rpc: client,
ethClient: ethClient,
B: b,
supportsStateDiff: supportsStateDiff,
rpc: client,
ethClient: ethClient,
}
}
@ -84,6 +89,7 @@ func (pea *PublicEthAPI) GetHeaderByNumber(ctx context.Context, number rpc.Block
}
if pea.ethClient != nil {
if header, err := pea.ethClient.HeaderByNumber(ctx, big.NewInt(number.Int64())); header != nil && err == nil {
go pea.writeStateDiffAt(number.Int64())
return pea.rpcMarshalHeader(header)
}
}
@ -100,6 +106,7 @@ func (pea *PublicEthAPI) GetHeaderByHash(ctx context.Context, hash common.Hash)
}
if pea.ethClient != nil {
if header, err := pea.ethClient.HeaderByHash(ctx, hash); header != nil && err == nil {
go pea.writeStateDiffFor(hash)
if res, err := pea.rpcMarshalHeader(header); err != nil {
return res
}
@ -137,6 +144,7 @@ func (pea *PublicEthAPI) GetBlockByNumber(ctx context.Context, number rpc.BlockN
}
if pea.ethClient != nil {
if block, err := pea.ethClient.BlockByNumber(ctx, big.NewInt(number.Int64())); block != nil && err == nil {
go pea.writeStateDiffAt(number.Int64())
return pea.rpcMarshalBlock(block, true, fullTx)
}
}
@ -152,6 +160,7 @@ func (pea *PublicEthAPI) GetBlockByHash(ctx context.Context, hash common.Hash, f
}
if pea.ethClient != nil {
if block, err := pea.ethClient.BlockByHash(ctx, hash); block != nil && err == nil {
go pea.writeStateDiffFor(hash)
return pea.rpcMarshalBlock(block, true, fullTx)
}
}
@ -179,6 +188,7 @@ func (pea *PublicEthAPI) GetUncleByBlockNumberAndIndex(ctx context.Context, bloc
}
if pea.rpc != nil {
if uncle, uncleHashes, err := getBlockAndUncleHashes(pea.rpc, ctx, "eth_getUncleByBlockNumberAndIndex", blockNr, index); uncle != nil && err == nil {
go pea.writeStateDiffAt(blockNr.Int64())
return pea.rpcMarshalBlockWithUncleHashes(uncle, uncleHashes, false, false)
}
}
@ -200,6 +210,7 @@ func (pea *PublicEthAPI) GetUncleByBlockHashAndIndex(ctx context.Context, blockH
}
if pea.rpc != nil {
if uncle, uncleHashes, err := getBlockAndUncleHashes(pea.rpc, ctx, "eth_getUncleByBlockHashAndIndex", blockHash, index); uncle != nil && err == nil {
go pea.writeStateDiffFor(blockHash)
return pea.rpcMarshalBlockWithUncleHashes(uncle, uncleHashes, false, false)
}
}
@ -215,6 +226,7 @@ func (pea *PublicEthAPI) GetUncleCountByBlockNumber(ctx context.Context, blockNr
if pea.rpc != nil {
var num *hexutil.Uint
if err := pea.rpc.CallContext(ctx, &num, "eth_getUncleCountByBlockNumber", blockNr); num != nil && err == nil {
go pea.writeStateDiffAt(blockNr.Int64())
return num
}
}
@ -230,6 +242,7 @@ func (pea *PublicEthAPI) GetUncleCountByBlockHash(ctx context.Context, blockHash
if pea.rpc != nil {
var num *hexutil.Uint
if err := pea.rpc.CallContext(ctx, &num, "eth_getUncleCountByBlockHash", blockHash); num != nil && err == nil {
go pea.writeStateDiffFor(blockHash)
return num
}
}
@ -251,6 +264,7 @@ func (pea *PublicEthAPI) GetTransactionCount(ctx context.Context, address common
if pea.rpc != nil {
var num *hexutil.Uint64
if err := pea.rpc.CallContext(ctx, &num, "eth_getTransactionCount", address, blockNrOrHash); num != nil && err == nil {
go pea.writeStateDiffAtOrFor(blockNrOrHash)
return num, nil
}
}
@ -275,6 +289,7 @@ func (pea *PublicEthAPI) GetBlockTransactionCountByNumber(ctx context.Context, b
if pea.rpc != nil {
var num *hexutil.Uint
if err := pea.rpc.CallContext(ctx, &num, "eth_getBlockTransactionCountByNumber", blockNr); num != nil && err == nil {
go pea.writeStateDiffAt(blockNr.Int64())
return num
}
}
@ -290,6 +305,7 @@ func (pea *PublicEthAPI) GetBlockTransactionCountByHash(ctx context.Context, blo
if pea.rpc != nil {
var num *hexutil.Uint
if err := pea.rpc.CallContext(ctx, &num, "eth_getBlockTransactionCountByHash", blockHash); num != nil && err == nil {
go pea.writeStateDiffFor(blockHash)
return num
}
}
@ -304,6 +320,7 @@ func (pea *PublicEthAPI) GetTransactionByBlockNumberAndIndex(ctx context.Context
if pea.rpc != nil {
var tx *RPCTransaction
if err := pea.rpc.CallContext(ctx, &tx, "eth_getTransactionByBlockNumberAndIndex", blockNr, index); tx != nil && err == nil {
go pea.writeStateDiffAt(blockNr.Int64())
return tx
}
}
@ -318,6 +335,7 @@ func (pea *PublicEthAPI) GetTransactionByBlockHashAndIndex(ctx context.Context,
if pea.rpc != nil {
var tx *RPCTransaction
if err := pea.rpc.CallContext(ctx, &tx, "eth_getTransactionByBlockHashAndIndex", blockHash, index); tx != nil && err == nil {
go pea.writeStateDiffFor(blockHash)
return tx
}
}
@ -332,6 +350,7 @@ func (pea *PublicEthAPI) GetRawTransactionByBlockNumberAndIndex(ctx context.Cont
if pea.rpc != nil {
var tx hexutil.Bytes
if err := pea.rpc.CallContext(ctx, &tx, "eth_getRawTransactionByBlockNumberAndIndex", blockNr, index); tx != nil && err == nil {
go pea.writeStateDiffAt(blockNr.Int64())
return tx
}
}
@ -346,6 +365,7 @@ func (pea *PublicEthAPI) GetRawTransactionByBlockHashAndIndex(ctx context.Contex
if pea.rpc != nil {
var tx hexutil.Bytes
if err := pea.rpc.CallContext(ctx, &tx, "eth_getRawTransactionByBlockHashAndIndex", blockHash, index); tx != nil && err == nil {
go pea.writeStateDiffFor(blockHash)
return tx
}
}
@ -362,6 +382,7 @@ func (pea *PublicEthAPI) GetTransactionByHash(ctx context.Context, hash common.H
if pea.rpc != nil {
var tx *RPCTransaction
if err := pea.rpc.CallContext(ctx, &tx, "eth_getTransactionByHash", hash); tx != nil && err == nil {
go pea.writeStateDiffFor(hash)
return tx, nil
}
}
@ -378,6 +399,7 @@ func (pea *PublicEthAPI) GetRawTransactionByHash(ctx context.Context, hash commo
if pea.rpc != nil {
var tx hexutil.Bytes
if err := pea.rpc.CallContext(ctx, &tx, "eth_getRawTransactionByHash", hash); tx != nil && err == nil {
go pea.writeStateDiffFor(hash)
return tx, nil
}
}
@ -398,6 +420,7 @@ func (pea *PublicEthAPI) GetTransactionReceipt(ctx context.Context, hash common.
}
if pea.rpc != nil {
if receipt := pea.remoteGetTransactionReceipt(ctx, hash); receipt != nil {
go pea.writeStateDiffFor(hash)
return receipt, nil
}
}
@ -489,6 +512,7 @@ func (pea *PublicEthAPI) GetLogs(ctx context.Context, crit ethereum.FilterQuery)
if arg, err := toFilterArg(crit); err == nil {
var res []*types.Log
if err := pea.rpc.CallContext(ctx, &res, "eth_getLogs", arg); err == nil {
go pea.writeStateDiffWithCriteria(crit)
return res, nil
}
}
@ -601,6 +625,7 @@ func (pea *PublicEthAPI) GetBalance(ctx context.Context, address common.Address,
if pea.rpc != nil {
var res *hexutil.Big
if err := pea.rpc.CallContext(ctx, &res, "eth_getBalance", address, blockNrOrHash); res != nil && err == nil {
go pea.writeStateDiffAtOrFor(blockNrOrHash)
return res, nil
}
}
@ -626,6 +651,7 @@ func (pea *PublicEthAPI) GetStorageAt(ctx context.Context, address common.Addres
if pea.rpc != nil {
var res hexutil.Bytes
if err := pea.rpc.CallContext(ctx, &res, "eth_getStorageAt", address, key, blockNrOrHash); res != nil && err == nil {
go pea.writeStateDiffAtOrFor(blockNrOrHash)
return res, nil
}
}
@ -641,6 +667,7 @@ func (pea *PublicEthAPI) GetCode(ctx context.Context, address common.Address, bl
if pea.rpc != nil {
var res hexutil.Bytes
if err := pea.rpc.CallContext(ctx, &res, "eth_getCode", address, blockNrOrHash); res != nil && err == nil {
go pea.writeStateDiffAtOrFor(blockNrOrHash)
return res, nil
}
}
@ -656,6 +683,7 @@ func (pea *PublicEthAPI) GetProof(ctx context.Context, address common.Address, s
if pea.rpc != nil {
var res *AccountResult
if err := pea.rpc.CallContext(ctx, &res, "eth_getProof", address, storageKeys, blockNrOrHash); res != nil && err == nil {
go pea.writeStateDiffAtOrFor(blockNrOrHash)
return res, nil
}
}
@ -726,6 +754,7 @@ func (pea *PublicEthAPI) Call(ctx context.Context, args CallArgs, blockNrOrHash
if (failed || err != nil) && pea.rpc != nil {
var hex hexutil.Bytes
if err := pea.rpc.CallContext(ctx, &hex, "eth_call", args, blockNrOrHash, overrides); hex != nil && err == nil {
go pea.writeStateDiffAtOrFor(blockNrOrHash)
return hex, nil
}
}
@ -845,6 +874,81 @@ func DoCall(ctx context.Context, b *Backend, args CallArgs, blockNrOrHash rpc.Bl
return result.Return(), result.UsedGas, result.Failed(), err
}
// writeStateDiffAtOrFor calls out to the statediffing geth client to fill in a gap in the index
func (pea *PublicEthAPI) writeStateDiffAtOrFor(blockNrOrHash rpc.BlockNumberOrHash) {
if blockNr, ok := blockNrOrHash.Number(); ok {
pea.writeStateDiffAt(blockNr.Int64())
return
}
if hash, ok := blockNrOrHash.Hash(); ok {
pea.writeStateDiffFor(hash)
}
}
// writeStateDiffWithCriteria calls out to the statediffing geth client to fill in a gap in the index
func (pea *PublicEthAPI) writeStateDiffWithCriteria(crit ethereum.FilterQuery) {
if crit.BlockHash != nil {
pea.writeStateDiffFor(*crit.BlockHash)
return
}
var start, end int64
if crit.FromBlock != nil {
start = crit.FromBlock.Int64()
}
if crit.ToBlock != nil {
end = crit.ToBlock.Int64()
} else {
end = start
}
for i := start; i <= end; i++ {
pea.writeStateDiffAt(i)
}
}
// writeStateDiffAt calls out to the statediffing geth client to fill in a gap in the index
func (pea *PublicEthAPI) writeStateDiffAt(height int64) {
if !pea.supportsStateDiff {
return
}
// we use a separate context than the one provided by the client
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
var data json.RawMessage
params := statediff.Params{
IntermediateStateNodes: true,
IntermediateStorageNodes: true,
IncludeBlock: true,
IncludeReceipts: true,
IncludeTD: true,
IncludeCode: true,
}
if err := pea.rpc.CallContext(ctx, &data, "statediff_writeStateDiffAt", uint64(height), params); err != nil {
logrus.Errorf("writeStateDiffAt %d faild with err %s", height, err.Error())
}
}
// writeStateDiffFor calls out to the statediffing geth client to fill in a gap in the index
func (pea *PublicEthAPI) writeStateDiffFor(blockHash common.Hash) {
if !pea.supportsStateDiff {
return
}
// we use a separate context than the one provided by the client
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
var data json.RawMessage
params := statediff.Params{
IntermediateStateNodes: true,
IntermediateStorageNodes: true,
IncludeBlock: true,
IncludeReceipts: true,
IncludeTD: true,
IncludeCode: true,
}
if err := pea.rpc.CallContext(ctx, &data, "statediff_writeStateDiffFor", blockHash, params); err != nil {
logrus.Errorf("writeStateDiffFor %s faild with err %s", blockHash.Hex(), err.Error())
}
}
// rpcMarshalBlock uses the generalized output filler, then adds the total difficulty field
func (pea *PublicEthAPI) rpcMarshalBlock(b *types.Block, inclTx bool, fullTx bool) (map[string]interface{}, error) {
fields, err := RPCMarshalBlock(b, inclTx, fullTx)

2
pkg/eth/api_test.go
View File

@ -189,7 +189,7 @@ var _ = Describe("API", func() {
indexAndPublisher := eth2.NewIPLDPublisher(db)
backend, err := eth.NewEthBackend(db, &eth.Config{})
Expect(err).ToNot(HaveOccurred())
api = eth.NewPublicEthAPI(backend, nil)
api = eth.NewPublicEthAPI(backend, nil, false)
err = indexAndPublisher.Publish(test_helpers.MockConvertedPayload)
Expect(err).ToNot(HaveOccurred())
err = publishCode(db, test_helpers.ContractCodeHash, test_helpers.ContractCode)

3
pkg/eth/backend.go
View File

@ -20,9 +20,10 @@ import (
"context"
"errors"
"fmt"
"github.com/ethereum/go-ethereum/trie"
"math/big"
"github.com/ethereum/go-ethereum/trie"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/common"

3
pkg/eth/cid_retriever_test.go
View File

@ -17,9 +17,10 @@
package eth_test
import (
"math/big"
"github.com/ethereum/go-ethereum/trie"
"github.com/vulcanize/ipld-eth-indexer/pkg/shared"
"math/big"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"

7
pkg/eth/eth_state_test.go
View File

@ -19,10 +19,11 @@ package eth_test
import (
"bytes"
"context"
"github.com/vulcanize/ipld-eth-indexer/pkg/shared"
"io/ioutil"
"math/big"
"github.com/vulcanize/ipld-eth-indexer/pkg/shared"
"github.com/ethereum/go-ethereum/accounts/abi"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
@ -83,7 +84,7 @@ var _ = Describe("eth state reading tests", func() {
RPCGasCap: big.NewInt(10000000000),
})
Expect(err).ToNot(HaveOccurred())
api = eth.NewPublicEthAPI(backend, nil)
api = eth.NewPublicEthAPI(backend, nil, false)
// make the test blockchain (and state)
blocks, receipts, chain = test_helpers.MakeChain(5, test_helpers.Genesis, test_helpers.TestChainGen)
@ -153,7 +154,7 @@ var _ = Describe("eth state reading tests", func() {
// Insert some non-canonical data into the database so that we test our ability to discern canonicity
indexAndPublisher := eth2.NewIPLDPublisher(db)
api = eth.NewPublicEthAPI(backend, nil)
api = eth.NewPublicEthAPI(backend, nil, false)
err = indexAndPublisher.Publish(test_helpers.MockConvertedPayload)
Expect(err).ToNot(HaveOccurred())
// The non-canonical header has a child

1
pkg/eth/filterer.go
View File

@ -18,6 +18,7 @@ package eth
import (
"bytes"
sdtypes "github.com/ethereum/go-ethereum/statediff/types"
"github.com/ethereum/go-ethereum/common"

1
pkg/eth/filterer_test.go
View File

@ -18,6 +18,7 @@ package eth_test
import (
"bytes"
sdtypes "github.com/ethereum/go-ethereum/statediff/types"
. "github.com/onsi/ginkgo"

1
pkg/eth/helpers.go
View File

@ -18,6 +18,7 @@ package eth
import (
"fmt"
sdtypes "github.com/ethereum/go-ethereum/statediff/types"
"github.com/ethereum/go-ethereum/params"

3
pkg/eth/test_helpers/test_data.go
View File

@ -20,9 +20,10 @@ import (
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"math/big"
sdtypes "github.com/ethereum/go-ethereum/statediff/types"
"github.com/ethereum/go-ethereum/trie"
"math/big"
"github.com/vulcanize/ipld-eth-indexer/pkg/shared"

5
pkg/graphql/service.go
View File

@ -18,11 +18,12 @@ package graphql
import (
"fmt"
"github.com/ethereum/go-ethereum/cmd/utils"
"github.com/ethereum/go-ethereum/node"
"net"
"net/http"
"github.com/ethereum/go-ethereum/cmd/utils"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rpc"
"github.com/graph-gophers/graphql-go"

1
pkg/rpc/http.go
View File

@ -18,6 +18,7 @@ package rpc
import (
"fmt"
"github.com/ethereum/go-ethereum/cmd/utils"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/rpc"

5
pkg/rpc/ws.go
View File

@ -17,11 +17,12 @@
package rpc
import (
"github.com/ethereum/go-ethereum/cmd/utils"
"github.com/ethereum/go-ethereum/node"
"net"
"net/http"
"github.com/ethereum/go-ethereum/cmd/utils"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/rpc"
"github.com/vulcanize/ipld-eth-server/pkg/prom"
)

22
pkg/serve/config.go
View File

@ -48,19 +48,21 @@ const (
ETH_DEFAULT_SENDER_ADDR = "ETH_DEFAULT_SENDER_ADDR"
ETH_RPC_GAS_CAP = "ETH_RPC_GAS_CAP"
ETH_SUPPORTS_STATEDIFF = "ETH_SUPPORTS_STATEDIFF"
)
// Config struct
type Config struct {
DB *postgres.DB
DBConfig postgres.Config
WSEndpoint string
HTTPEndpoint string
IPCEndpoint string
ChainConfig *params.ChainConfig
DefaultSender *common.Address
RPCGasCap *big.Int
Client *rpc.Client
DB *postgres.DB
DBConfig postgres.Config
WSEndpoint string
HTTPEndpoint string
IPCEndpoint string
ChainConfig *params.ChainConfig
DefaultSender *common.Address
RPCGasCap *big.Int
Client *rpc.Client
SupportStateDiff bool
}
// NewConfig is used to initialize a watcher config from a .toml file
@ -74,6 +76,7 @@ func NewConfig() (*Config, error) {
viper.BindEnv("ethereum.httpPath", shared.ETH_HTTP_PATH)
viper.BindEnv("ethereum.defaultSender", ETH_DEFAULT_SENDER_ADDR)
viper.BindEnv("ethereum.rpcGasCap", ETH_RPC_GAS_CAP)
viper.BindEnv("ethereum.supportsStateDiff", ETH_SUPPORTS_STATEDIFF)
c.DBConfig.Init()
@ -83,6 +86,7 @@ func NewConfig() (*Config, error) {
return nil, err
}
c.Client = cli
c.SupportStateDiff = viper.GetBool("ethereum.supportsStateDiff")
wsPath := viper.GetString("server.wsPath")
if wsPath == "" {

6
pkg/serve/service.go
View File

@ -80,6 +80,8 @@ type Service struct {
serveWg *sync.WaitGroup
// rpc client for forwarding cache misses
client *rpc.Client
// whether the proxied client supports state diffing
supportsStateDiffing bool
// backend for the server
backend *eth.Backend
}
@ -94,6 +96,8 @@ func NewServer(settings *Config) (Server, error) {
sap.QuitChan = make(chan bool)
sap.Subscriptions = make(map[common.Hash]map[rpc.ID]Subscription)
sap.SubscriptionTypes = make(map[common.Hash]eth.SubscriptionSettings)
sap.client = settings.Client
sap.supportsStateDiffing = settings.SupportStateDiff
var err error
sap.backend, err = eth.NewEthBackend(sap.db, &eth.Config{
ChainConfig: settings.ChainConfig,
@ -141,7 +145,7 @@ func (sap *Service) APIs() []rpc.API {
return append(apis, rpc.API{
Namespace: eth.APIName,
Version: eth.APIVersion,
Service: eth.NewPublicEthAPI(sap.backend, sap.client),
Service: eth.NewPublicEthAPI(sap.backend, sap.client, sap.supportsStateDiffing),
Public: true,
})
}