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

major refactoring of super_node to make it easier to support other chains

Browse Source
This commit is contained in:
Ian Norden
2020-01-24 15:37:52 -06:00
parent 5ff05225c2
commit 7843312815
69 changed files with 2811 additions and 1509 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/root.go
+2 -1
View File
@@ -34,13 +34,14 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/eth/client"
vRpc "github.com/vulcanize/vulcanizedb/pkg/eth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/eth/node"
config2 "github.com/vulcanize/vulcanizedb/pkg/super_node/config"
)
var (
cfgFile string
databaseConfig config.Database
genConfig config.Plugin
subscriptionConfig config.Subscription
subscriptionConfig *config2.EthSubscription
ipc string
levelDbPath string
queueRecheckInterval time.Duration
cmd/screenAndServe.go
+36 -30
View File
@@ -26,10 +26,10 @@ import (
"github.com/spf13/cobra"
"github.com/spf13/viper"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
"github.com/vulcanize/vulcanizedb/pkg/super_node"
"github.com/vulcanize/vulcanizedb/pkg/super_node/config"
"github.com/vulcanize/vulcanizedb/utils"
)
@@ -52,18 +52,17 @@ func init() {
}
func screenAndServe() {
superNode, newNodeErr := newSuperNodeWithoutPairedGethNode()
if newNodeErr != nil {
logWithCommand.Fatal(newNodeErr)
superNode, err := newSuperNodeWithoutPairedGethNode()
if err != nil {
logWithCommand.Fatal(err)
}
wg := &syn.WaitGroup{}
quitChan := make(chan bool, 1)
emptyPayloadChan := make(chan ipfs.IPLDPayload)
emptyPayloadChan := make(chan interface{})
superNode.ScreenAndServe(wg, emptyPayloadChan, quitChan)
serverErr := startServers(superNode)
if serverErr != nil {
logWithCommand.Fatal(serverErr)
if err := startServers(superNode); err != nil {
logWithCommand.Fatal(err)
}
wg.Wait()
}
@@ -72,15 +71,15 @@ func startServers(superNode super_node.NodeInterface) error {
var ipcPath string
ipcPath = viper.GetString("server.ipcPath")
if ipcPath == "" {
home, homeDirErr := os.UserHomeDir()
if homeDirErr != nil {
return homeDirErr
home, err := os.UserHomeDir()
if err != nil {
return err
}
ipcPath = filepath.Join(home, ".vulcanize/vulcanize.ipc")
}
_, _, ipcErr := rpc.StartIPCEndpoint(ipcPath, superNode.APIs())
if ipcErr != nil {
return ipcErr
_, _, err := rpc.StartIPCEndpoint(ipcPath, superNode.APIs())
if err != nil {
return err
}
var wsEndpoint string
@@ -90,9 +89,9 @@ func startServers(superNode super_node.NodeInterface) error {
}
var exposeAll = true
var wsOrigins []string
_, _, wsErr := rpc.StartWSEndpoint(wsEndpoint, superNode.APIs(), []string{"vdb"}, wsOrigins, exposeAll)
if wsErr != nil {
return wsErr
_, _, err = rpc.StartWSEndpoint(wsEndpoint, superNode.APIs(), []string{"vdb"}, wsOrigins, exposeAll)
if err != nil {
return err
}
return nil
}
@@ -100,27 +99,34 @@ func startServers(superNode super_node.NodeInterface) error {
func newSuperNodeWithoutPairedGethNode() (super_node.NodeInterface, error) {
ipfsPath = viper.GetString("client.ipfsPath")
if ipfsPath == "" {
home, homeDirErr := os.UserHomeDir()
if homeDirErr != nil {
return nil, homeDirErr
home, err := os.UserHomeDir()
if err != nil {
return nil, err
}
ipfsPath = filepath.Join(home, ".ipfs")
}
ipfsInitErr := ipfs.InitIPFSPlugins()
if ipfsInitErr != nil {
return nil, ipfsInitErr
if err := ipfs.InitIPFSPlugins(); err != nil {
return nil, err
}
ipldFetcher, newFetcherErr := ipfs.NewIPLDFetcher(ipfsPath)
if newFetcherErr != nil {
return nil, newFetcherErr
ipldFetcher, err := super_node.NewIPLDFetcher(config.Ethereum, ipfsPath)
if err != nil {
return nil, err
}
db := utils.LoadPostgres(databaseConfig, core.Node{})
retriever, err := super_node.NewCIDRetriever(config.Ethereum, &db)
if err != nil {
return nil, err
}
resolver, err := super_node.NewIPLDResolver(config.Ethereum)
if err != nil {
return nil, err
}
return &super_node.Service{
IPLDFetcher: ipldFetcher,
Retriever: super_node.NewCIDRetriever(&db),
Resolver: ipfs.NewIPLDResolver(),
Retriever: retriever,
Resolver: resolver,
Subscriptions: make(map[common.Hash]map[rpc.ID]super_node.Subscription),
SubscriptionTypes: make(map[common.Hash]config.Subscription),
GethNode: core.Node{},
SubscriptionTypes: make(map[common.Hash]super_node.SubscriptionSettings),
NodeInfo: core.Node{},
}, nil
}
cmd/streamSubscribe.go
+20 -13
View File
@@ -30,9 +30,11 @@ import (
"github.com/spf13/viper"
"github.com/vulcanize/vulcanizedb/libraries/shared/streamer"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/eth/client"
"github.com/vulcanize/vulcanizedb/pkg/super_node"
"github.com/vulcanize/vulcanizedb/pkg/super_node/config"
"github.com/vulcanize/vulcanizedb/pkg/super_node/eth"
)
// streamSubscribeCmd represents the streamSubscribe command
@@ -61,10 +63,10 @@ func streamSubscribe() {
str := streamer.NewSuperNodeStreamer(rpcClient)
// Buffered channel for reading subscription payloads
payloadChan := make(chan streamer.SuperNodePayload, 20000)
payloadChan := make(chan super_node.Payload, 20000)
// Subscribe to the super node service with the given config/filter parameters
sub, err := str.Stream(payloadChan, subscriptionConfig)
sub, err := str.StreamETH(payloadChan, subscriptionConfig)
if err != nil {
logWithCommand.Fatal(err)
}
@@ -73,11 +75,16 @@ func streamSubscribe() {
for {
select {
case payload := <-payloadChan:
if payload.ErrMsg != "" {
logWithCommand.Error(payload.ErrMsg)
if payload.Err != "" {
logWithCommand.Error(payload.Err)
continue
}
for _, headerRlp := range payload.HeadersRlp {
data, ok := payload.Data.(eth.StreamPayload)
if !ok {
logWithCommand.Warnf("payload data expected type %T got %T", eth.StreamPayload{}, payload.Data)
continue
}
for _, headerRlp := range data.HeadersRlp {
var header types.Header
err = rlp.Decode(bytes.NewBuffer(headerRlp), &header)
if err != nil {
@@ -87,7 +94,7 @@ func streamSubscribe() {
fmt.Printf("Header number %d, hash %s\n", header.Number.Int64(), header.Hash().Hex())
fmt.Printf("header: %v\n", header)
}
for _, trxRlp := range payload.TransactionsRlp {
for _, trxRlp := range data.TransactionsRlp {
var trx types.Transaction
buff := bytes.NewBuffer(trxRlp)
stream := rlp.NewStream(buff, 0)
@@ -99,7 +106,7 @@ func streamSubscribe() {
fmt.Printf("Transaction with hash %s\n", trx.Hash().Hex())
fmt.Printf("trx: %v\n", trx)
}
for _, rctRlp := range payload.ReceiptsRlp {
for _, rctRlp := range data.ReceiptsRlp {
var rct types.ReceiptForStorage
buff := bytes.NewBuffer(rctRlp)
stream := rlp.NewStream(buff, 0)
@@ -121,7 +128,7 @@ func streamSubscribe() {
}
}
// This assumes leafs only
for key, stateRlp := range payload.StateNodesRlp {
for key, stateRlp := range data.StateNodesRlp {
var acct state.Account
err = rlp.Decode(bytes.NewBuffer(stateRlp), &acct)
if err != nil {
@@ -132,7 +139,7 @@ func streamSubscribe() {
key.Hex(), acct.Root.Hex(), acct.Balance.Int64())
fmt.Printf("state account: %v\n", acct)
}
for stateKey, mappedRlp := range payload.StorageNodesRlp {
for stateKey, mappedRlp := range data.StorageNodesRlp {
fmt.Printf("Storage for state key %s ", stateKey.Hex())
for storageKey, storageRlp := range mappedRlp {
fmt.Printf("with storage key %s\n", storageKey.Hex())
@@ -165,15 +172,15 @@ func streamSubscribe() {
func configureSubscription() {
logWithCommand.Info("loading subscription config")
subscriptionConfig = config.Subscription{
subscriptionConfig = &config.EthSubscription{
// Below default to false, which means we do not backfill by default
BackFill: viper.GetBool("subscription.backfill"),
BackFillOnly: viper.GetBool("subscription.backfillOnly"),
// Below default to 0
// 0 start means we start at the beginning and 0 end means we continue indefinitely
StartingBlock: big.NewInt(viper.GetInt64("subscription.startingBlock")),
EndingBlock: big.NewInt(viper.GetInt64("subscription.endingBlock")),
Start: big.NewInt(viper.GetInt64("subscription.startingBlock")),
End: big.NewInt(viper.GetInt64("subscription.endingBlock")),
// Below default to false, which means we get all headers by default
HeaderFilter: config.HeaderFilter{
cmd/syncAndPublish.go
+3 -2
View File
@@ -33,6 +33,7 @@ import (
vRpc "github.com/vulcanize/vulcanizedb/pkg/eth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/eth/node"
"github.com/vulcanize/vulcanizedb/pkg/super_node"
"github.com/vulcanize/vulcanizedb/pkg/super_node/config"
"github.com/vulcanize/vulcanizedb/utils"
)
@@ -107,7 +108,7 @@ func newSuperNode() (super_node.NodeInterface, error) {
if workers < 1 {
workers = 1
}
return super_node.NewSuperNode(ipfsPath, &db, rpcClient, quitChan, workers, blockChain.Node())
return super_node.NewSuperNode(config.Ethereum, ipfsPath, &db, rpcClient, quitChan, workers, blockChain.Node())
}
func newBackFiller() (super_node.BackFillInterface, error) {
@@ -120,5 +121,5 @@ func newBackFiller() (super_node.BackFillInterface, error) {
} else {
frequency = time.Duration(freq)
}
return super_node.NewBackFillService(ipfsPath, &db, archivalRPCClient, time.Minute*frequency, super_node.DefaultMaxBatchSize)
return super_node.NewBackFillService(config.Ethereum, ipfsPath, &db, archivalRPCClient, time.Minute*frequency, super_node.DefaultMaxBatchSize)
}
cmd/syncPublishScreenAndServe.go
+1 -3
View File
@@ -21,8 +21,6 @@ import (
log "github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"github.com/spf13/viper"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
)
// syncPublishScreenAndServeCmd represents the syncPublishScreenAndServe command
@@ -52,7 +50,7 @@ func syncPublishScreenAndServe() {
}
wg := &syn.WaitGroup{}
forwardPayloadChan := make(chan ipfs.IPLDPayload, 20000)
forwardPayloadChan := make(chan interface{}, 20000)
forwardQuitChan := make(chan bool, 1)
syncAndPubErr := superNode.SyncAndPublish(wg, forwardPayloadChan, forwardQuitChan)
if syncAndPubErr != nil {