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

use configurable timeout for geth batch http requests; additional error log info in payload fetchers

Browse Source
This commit is contained in:
Ian Norden
2020-04-20 11:05:12 -05:00
parent b3be8aaa05
commit eceaa0aecb
16 changed files with 128 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/super_node/backfiller.go
+1 -1
View File
@@ -85,7 +85,7 @@ func NewBackFillService(settings *Config, screenAndServeChan chan shared.Convert
if err != nil {
return nil, err
}
fetcher, err := NewPaylaodFetcher(settings.Chain, settings.HTTPClient)
fetcher, err := NewPaylaodFetcher(settings.Chain, settings.HTTPClient, settings.Timeout)
if err != nil {
return nil, err
}
pkg/super_node/btc/payload_fetcher.go
+4 -2
View File
@@ -17,6 +17,8 @@
package btc
import (
"fmt"
"github.com/btcsuite/btcd/rpcclient"
"github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btcutil"
@@ -48,11 +50,11 @@ func (fetcher *PayloadFetcher) FetchAt(blockHeights []uint64) ([]shared.RawChain
for i, height := range blockHeights {
hash, err := fetcher.client.GetBlockHash(int64(height))
if err != nil {
return nil, err
return nil, fmt.Errorf("bitcoin PayloadFetcher GetBlockHash err at blockheight %d: %s", height, err.Error())
}
block, err := fetcher.client.GetBlock(hash)
if err != nil {
return nil, err
return nil, fmt.Errorf("bitcoin PayloadFetcher GetBlock err at blockheight %d: %s", height, err.Error())
}
blockPayloads[i] = BlockPayload{
BlockHeight: int64(height),
pkg/super_node/config.go
+8
View File
@@ -72,6 +72,7 @@ type Config struct {
BatchSize uint64
BatchNumber uint64
ValidationLevel int
Timeout time.Duration // HTTP connection timeout in seconds
}
// NewSuperNodeConfig is used to initialize a SuperNode config from a .toml file
@@ -90,6 +91,13 @@ func NewSuperNodeConfig() (*Config, error) {
viper.BindEnv("superNode.ipcPath", SUPERNODE_IPC_PATH)
viper.BindEnv("superNode.httpPath", SUPERNODE_HTTP_PATH)
viper.BindEnv("superNode.backFill", SUPERNODE_BACKFILL)
viper.BindEnv("superNode.timeout", shared.HTTP_TIMEOUT)
timeout := viper.GetInt("superNode.timeout")
if timeout < 15 {
timeout = 15
}
c.Timeout = time.Second * time.Duration(timeout)
chain := viper.GetString("superNode.chain")
c.Chain, err = shared.NewChainType(chain)
pkg/super_node/constructors.go
+3 -2
View File
@@ -18,6 +18,7 @@ package super_node
import (
"fmt"
"time"
"github.com/btcsuite/btcd/chaincfg"
@@ -92,7 +93,7 @@ func NewPayloadStreamer(chain shared.ChainType, clientOrConfig interface{}) (sha
}
// NewPaylaodFetcher constructs a PayloadFetcher for the provided chain type
func NewPaylaodFetcher(chain shared.ChainType, client interface{}) (shared.PayloadFetcher, error) {
func NewPaylaodFetcher(chain shared.ChainType, client interface{}, timeout time.Duration) (shared.PayloadFetcher, error) {
switch chain {
case shared.Ethereum:
batchClient, ok := client.(eth.BatchClient)
@@ -100,7 +101,7 @@ func NewPaylaodFetcher(chain shared.ChainType, client interface{}) (shared.Paylo
var expectedClient eth.BatchClient
return nil, fmt.Errorf("ethereum payload fetcher constructor expected client type %T got %T", expectedClient, client)
}
return eth.NewPayloadFetcher(batchClient), nil
return eth.NewPayloadFetcher(batchClient, timeout), nil
case shared.Bitcoin:
connConfig, ok := client.(*rpcclient.ConnConfig)
if !ok {
pkg/super_node/eth/payload_fetcher.go
+14 -9
View File
@@ -17,33 +17,36 @@
package eth
import (
"context"
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
"time"
"github.com/ethereum/go-ethereum/statediff"
"github.com/vulcanize/vulcanizedb/pkg/eth/client"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
)
// BatchClient is an interface to a batch-fetching geth rpc client; created to allow mock insertion
type BatchClient interface {
BatchCall(batch []client.BatchElem) error
BatchCallContext(ctx context.Context, batch []client.BatchElem) error
}
// PayloadFetcher satisfies the PayloadFetcher interface for ethereum
type PayloadFetcher struct {
// PayloadFetcher is thread-safe as long as the underlying client is thread-safe, since it has/modifies no other state
// http.Client is thread-safe
client BatchClient
client BatchClient
timeout time.Duration
}
const method = "statediff_stateDiffAt"
// NewStateDiffFetcher returns a PayloadFetcher
func NewPayloadFetcher(bc BatchClient) *PayloadFetcher {
func NewPayloadFetcher(bc BatchClient, timeout time.Duration) *PayloadFetcher {
return &PayloadFetcher{
client: bc,
client: bc,
timeout: timeout,
}
}
@@ -58,14 +61,16 @@ func (fetcher *PayloadFetcher) FetchAt(blockHeights []uint64) ([]shared.RawChain
Result: new(statediff.Payload),
})
}
batchErr := fetcher.client.BatchCall(batch)
ctx, cancel := context.WithTimeout(context.Background(), fetcher.timeout)
defer cancel()
batchErr := fetcher.client.BatchCallContext(ctx, batch)
if batchErr != nil {
return nil, fmt.Errorf("PayloadFetcher err: %s", batchErr.Error())
return nil, fmt.Errorf("ethereum PayloadFetcher batch err for block range %d-%d: %s", blockHeights[0], blockHeights[len(blockHeights)-1], batchErr.Error())
}
results := make([]shared.RawChainData, 0, len(blockHeights))
for _, batchElem := range batch {
if batchElem.Error != nil {
return nil, fmt.Errorf("PayloadFetcher err: %s", batchElem.Error.Error())
return nil, fmt.Errorf("ethereum PayloadFetcher err at blockheight %d: %s", batchElem.Args[0].(uint64), batchElem.Error.Error())
}
payload, ok := batchElem.Result.(*statediff.Payload)
if ok {
pkg/super_node/eth/payload_fetcher_test.go
+3 -1
View File
@@ -17,6 +17,8 @@
package eth_test
import (
"time"
"github.com/ethereum/go-ethereum/statediff"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
@@ -38,7 +40,7 @@ var _ = Describe("StateDiffFetcher", func() {
Expect(setDiffAtErr1).ToNot(HaveOccurred())
setDiffAtErr2 := mc.SetReturnDiffAt(test_data.BlockNumber2.Uint64(), test_data.MockStatediffPayload2)
Expect(setDiffAtErr2).ToNot(HaveOccurred())
stateDiffFetcher = eth.NewPayloadFetcher(mc)
stateDiffFetcher = eth.NewPayloadFetcher(mc, time.Second*60)
})
It("Batch calls statediff_stateDiffAt", func() {
blockHeights := []uint64{
pkg/super_node/resync/config.go
+13 -4
View File
@@ -18,6 +18,7 @@ package resync
import (
"fmt"
"time"
"github.com/spf13/viper"
@@ -52,10 +53,11 @@ type Config struct {
DBConfig config.Database
IPFSPath string
HTTPClient interface{} // Note this client is expected to support the retrieval of the specified data type(s)
NodeInfo core.Node // Info for the associated node
Ranges [][2]uint64 // The block height ranges to resync
BatchSize uint64 // BatchSize for the resync http calls (client has to support batch sizing)
HTTPClient interface{} // Note this client is expected to support the retrieval of the specified data type(s)
NodeInfo core.Node // Info for the associated node
Ranges [][2]uint64 // The block height ranges to resync
BatchSize uint64 // BatchSize for the resync http calls (client has to support batch sizing)
Timeout time.Duration // HTTP connection timeout in seconds
BatchNumber uint64
Quit chan bool // Channel for shutting down
@@ -76,6 +78,13 @@ func NewReSyncConfig() (*Config, error) {
viper.BindEnv("resync.batchSize", RESYNC_BATCH_SIZE)
viper.BindEnv("resync.batchNumber", RESYNC_BATCH_NUMBER)
viper.BindEnv("resync.resetValidation", RESYNC_RESET_VALIDATION)
viper.BindEnv("resync.timeout", shared.HTTP_TIMEOUT)
timeout := viper.GetInt("resync.timeout")
if timeout < 15 {
timeout = 15
}
c.Timeout = time.Second * time.Duration(timeout)
start := uint64(viper.GetInt64("resync.start"))
stop := uint64(viper.GetInt64("resync.stop"))
pkg/super_node/resync/service.go
+1 -1
View File
@@ -80,7 +80,7 @@ func NewResyncService(settings *Config) (Resync, error) {
if err != nil {
return nil, err
}
fetcher, err := super_node.NewPaylaodFetcher(settings.Chain, settings.HTTPClient)
fetcher, err := super_node.NewPaylaodFetcher(settings.Chain, settings.HTTPClient, settings.Timeout)
if err != nil {
return nil, err
}
pkg/super_node/shared/env.go
+21 -10
View File
@@ -24,17 +24,20 @@ import (
"github.com/btcsuite/btcd/rpcclient"
"github.com/spf13/viper"
"github.com/vulcanize/vulcanizedb/pkg/eth/client"
"github.com/vulcanize/vulcanizedb/pkg/eth/core"
"github.com/vulcanize/vulcanizedb/pkg/eth/node"
)
// Env variables
const (
IPFS_PATH = "IPFS_PATH"
IPFS_PATH = "IPFS_PATH"
HTTP_TIMEOUT = "HTTP_TIMEOUT"
ETH_WS_PATH = "ETH_WS_PATH"
ETH_HTTP_PATH = "ETH_HTTP_PATH"
ETH_WS_PATH = "ETH_WS_PATH"
ETH_HTTP_PATH = "ETH_HTTP_PATH"
ETH_NODE_ID = "ETH_NODE_ID"
ETH_CLIENT_NAME = "ETH_CLIENT_NAME"
ETH_GENESIS_BLOCK = "ETH_GENESIS_BLOCK"
ETH_NETWORK_ID = "ETH_NETWORK_ID"
BTC_WS_PATH = "BTC_WS_PATH"
BTC_HTTP_PATH = "BTC_HTTP_PATH"
@@ -47,14 +50,22 @@ const (
)
// GetEthNodeAndClient returns eth node info and client from path url
func GetEthNodeAndClient(path string) (core.Node, core.RPCClient, error) {
rawRPCClient, err := rpc.Dial(path)
func GetEthNodeAndClient(path string) (core.Node, *rpc.Client, error) {
viper.BindEnv("ethereum.nodeID", ETH_NODE_ID)
viper.BindEnv("ethereum.clientName", ETH_CLIENT_NAME)
viper.BindEnv("ethereum.genesisBlock", ETH_GENESIS_BLOCK)
viper.BindEnv("ethereum.networkID", ETH_NETWORK_ID)
rpcClient, err := rpc.Dial(path)
if err != nil {
return core.Node{}, nil, err
}
rpcClient := client.NewRPCClient(rawRPCClient, path)
vdbNode := node.MakeNode(rpcClient)
return vdbNode, rpcClient, nil
return core.Node{
ID: viper.GetString("ethereum.nodeID"),
ClientName: viper.GetString("ethereum.clientName"),
GenesisBlock: viper.GetString("ethereum.genesisBlock"),
NetworkID: viper.GetString("ethereum.networkID"),
}, rpcClient, nil
}
// GetIPFSPath returns the ipfs path from the config or env variable