deep-stack/ipld-eth-server
4
0
Fork 0
forked from cerc-io/ipld-eth-server
Code Pull Requests Activity

major refactor part 2: remove cold import, full sync, generalize node table

Browse Source
This commit is contained in:
Ian Norden
2020-02-20 16:14:17 -06:00
parent 0f765df12c
commit ca273a026d
171 changed files with 196 additions and 6330 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/super_node/btc/indexer.go
+7 -7
View File
@@ -24,7 +24,7 @@ import (
"github.com/jmoiron/sqlx"
"github.com/lib/pq"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
)
@@ -47,7 +47,7 @@ func (in *CIDIndexer) Index(cids shared.CIDsForIndexing) error {
if err != nil {
return err
}
headerID, err := in.indexHeaderCID(tx, cidWrapper.HeaderCID)
headerID, err := in.indexHeaderCID(tx, cidWrapper.HeaderCID, in.db.NodeID)
if err != nil {
logrus.Error("btc indexer error when indexing header")
return err
@@ -59,13 +59,13 @@ func (in *CIDIndexer) Index(cids shared.CIDsForIndexing) error {
return tx.Commit()
}
func (in *CIDIndexer) indexHeaderCID(tx *sqlx.Tx, header HeaderModel) (int64, error) {
func (in *CIDIndexer) indexHeaderCID(tx *sqlx.Tx, header HeaderModel, nodeID int64) (int64, error) {
var headerID int64
err := tx.QueryRowx(`INSERT INTO btc.header_cids (block_number, block_hash, parent_hash, cid, timestamp, bits)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (block_number, block_hash) DO UPDATE SET (parent_hash, cid, timestamp, bits) = ($3, $4, $5, $6)
err := tx.QueryRowx(`INSERT INTO btc.header_cids (block_number, block_hash, parent_hash, cid, timestamp, bits, node_id)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (block_number, block_hash) DO UPDATE SET (parent_hash, cid, timestamp, bits, node_id) = ($3, $4, $5, $6, $7)
RETURNING id`,
header.BlockNumber, header.BlockHash, header.ParentHash, header.CID, header.Timestamp, header.Bits).Scan(&headerID)
header.BlockNumber, header.BlockHash, header.ParentHash, header.CID, header.Timestamp, header.Bits, nodeID).Scan(&headerID)
return headerID, err
}
pkg/super_node/btc/indexer_test.go
+1 -1
View File
@@ -20,7 +20,7 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
"github.com/vulcanize/vulcanizedb/pkg/super_node/btc"
"github.com/vulcanize/vulcanizedb/pkg/super_node/btc/mocks"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
pkg/super_node/btc/models.go
+1
View File
@@ -27,6 +27,7 @@ type HeaderModel struct {
CID string `db:"cid"`
Timestamp int64 `db:"timestamp"`
Bits uint32 `db:"bits"`
NodeID int64 `db:"node_id"`
}
// TxModel is the db model for btc.transaction_cids table
pkg/super_node/btc/payload_fetcher.go
+7 -3
View File
@@ -32,10 +32,14 @@ type PayloadFetcher struct {
}
// NewStateDiffFetcher returns a PayloadFetcher
func NewPayloadFetcher(c *rpcclient.Client) *PayloadFetcher {
return &PayloadFetcher{
client: c,
func NewPayloadFetcher(c *rpcclient.ConnConfig) (*PayloadFetcher, error) {
client, err := rpcclient.New(c, nil)
if err != nil {
return nil, err
}
return &PayloadFetcher{
client: client,
}, nil
}
// FetchAt fetches the block payloads at the given block heights
pkg/super_node/btc/retriever.go
+1 -1
View File
@@ -26,7 +26,7 @@ import (
"github.com/jmoiron/sqlx"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
)
pkg/super_node/btc/test_helpers.go
+1 -1
View File
@@ -19,7 +19,7 @@ package btc
import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
)
// TearDownDB is used to tear down the super node dbs after tests
pkg/super_node/constructors.go
+3 -3
View File
@@ -26,7 +26,7 @@ import (
"github.com/ethereum/go-ethereum/rpc"
"github.com/vulcanize/vulcanizedb/pkg/eth/core"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
"github.com/vulcanize/vulcanizedb/pkg/super_node/btc"
"github.com/vulcanize/vulcanizedb/pkg/super_node/eth"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
@@ -102,11 +102,11 @@ func NewPaylaodFetcher(chain shared.ChainType, client interface{}) (shared.Paylo
}
return eth.NewPayloadFetcher(batchClient), nil
case shared.Bitcoin:
rpcClient, ok := client.(*rpcclient.Client)
connConfig, ok := client.(*rpcclient.ConnConfig)
if !ok {
return nil, fmt.Errorf("bitcoin payload fetcher constructor expected client type %T got %T", &rpcclient.Client{}, client)
}
return btc.NewPayloadFetcher(rpcClient), nil
return btc.NewPayloadFetcher(connConfig)
default:
return nil, fmt.Errorf("invalid chain %s for payload fetcher constructor", chain.String())
}
pkg/super_node/eth/backend.go
+1 -1
View File
@@ -28,7 +28,7 @@ import (
"github.com/ethereum/go-ethereum/rpc"
"github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
)
var (
pkg/super_node/eth/indexer.go
+6 -6
View File
@@ -25,7 +25,7 @@ import (
"github.com/jmoiron/sqlx"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
)
// Indexer satisfies the Indexer interface for ethereum
@@ -50,7 +50,7 @@ func (in *CIDIndexer) Index(cids shared.CIDsForIndexing) error {
if err != nil {
return err
}
headerID, err := in.indexHeaderCID(tx, cidPayload.HeaderCID)
headerID, err := in.indexHeaderCID(tx, cidPayload.HeaderCID, in.db.NodeID)
if err != nil {
if err := tx.Rollback(); err != nil {
log.Error(err)
@@ -80,12 +80,12 @@ func (in *CIDIndexer) Index(cids shared.CIDsForIndexing) error {
return tx.Commit()
}
func (in *CIDIndexer) indexHeaderCID(tx *sqlx.Tx, header HeaderModel) (int64, error) {
func (in *CIDIndexer) indexHeaderCID(tx *sqlx.Tx, header HeaderModel, nodeID int64) (int64, error) {
var headerID int64
err := tx.QueryRowx(`INSERT INTO eth.header_cids (block_number, block_hash, parent_hash, cid, td) VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (block_number, block_hash) DO UPDATE SET (parent_hash, cid, td) = ($3, $4, $5)
err := tx.QueryRowx(`INSERT INTO eth.header_cids (block_number, block_hash, parent_hash, cid, td, node_id) VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (block_number, block_hash) DO UPDATE SET (parent_hash, cid, td, node_id) = ($3, $4, $5, $6)
RETURNING id`,
header.BlockNumber, header.BlockHash, header.ParentHash, header.CID, header.TotalDifficulty).Scan(&headerID)
header.BlockNumber, header.BlockHash, header.ParentHash, header.CID, header.TotalDifficulty, nodeID).Scan(&headerID)
return headerID, err
}
pkg/super_node/eth/indexer_test.go
+1 -1
View File
@@ -20,7 +20,7 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
"github.com/vulcanize/vulcanizedb/pkg/super_node/eth"
"github.com/vulcanize/vulcanizedb/pkg/super_node/eth/mocks"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
pkg/super_node/eth/models.go
+1
View File
@@ -26,6 +26,7 @@ type HeaderModel struct {
ParentHash string `db:"parent_hash"`
CID string `db:"cid"`
TotalDifficulty string `db:"td"`
NodeID int64 `db:"node_id"`
}
// UncleModel is the db model for eth.uncle_cids
pkg/super_node/eth/retriever.go
+1 -1
View File
@@ -26,7 +26,7 @@ import (
"github.com/lib/pq"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
)
pkg/super_node/eth/retriever_test.go
+2 -1
View File
@@ -22,7 +22,7 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
"github.com/vulcanize/vulcanizedb/pkg/super_node/eth"
eth2 "github.com/vulcanize/vulcanizedb/pkg/super_node/eth"
"github.com/vulcanize/vulcanizedb/pkg/super_node/eth/mocks"
@@ -236,6 +236,7 @@ var _ = Describe("Retriever", func() {
Expect(len(cidWrapper.Headers)).To(Equal(1))
expectedHeaderCIDs := mocks.MockCIDWrapper.Headers
expectedHeaderCIDs[0].ID = cidWrapper.Headers[0].ID
expectedHeaderCIDs[0].NodeID = cidWrapper.Headers[0].NodeID
Expect(cidWrapper.Headers).To(Equal(expectedHeaderCIDs))
Expect(len(cidWrapper.Transactions)).To(Equal(2))
Expect(eth.TxModelsContainsCID(cidWrapper.Transactions, mocks.MockCIDWrapper.Transactions[0].CID)).To(BeTrue())
pkg/super_node/eth/test_helpers.go
+1 -1
View File
@@ -19,7 +19,7 @@ package eth
import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
)
// TearDownDB is used to tear down the super node dbs after tests
pkg/super_node/service.go
+1 -1
View File
@@ -29,8 +29,8 @@ import (
log "github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/pkg/eth/core"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
)
pkg/super_node/shared/config.go
+43 -26
View File
@@ -17,22 +17,21 @@
package shared
import (
"fmt"
"os"
"path/filepath"
"time"
"github.com/btcsuite/btcd/rpcclient"
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/rpc"
"github.com/spf13/viper"
"os"
"path/filepath"
"time"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/eth"
"github.com/vulcanize/vulcanizedb/pkg/eth/client"
vRpc "github.com/vulcanize/vulcanizedb/pkg/eth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/eth/core"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/eth/node"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
"github.com/vulcanize/vulcanizedb/utils"
)
@@ -93,7 +92,21 @@ func NewSuperNodeConfig() (*SuperNodeConfig, error) {
workers = 1
}
sn.Workers = workers
sn.NodeInfo, sn.WSClient, err = getNodeAndClient(sn.Chain, viper.GetString("superNode.sync.wsPath"))
if sn.Chain == Ethereum {
sn.NodeInfo, sn.WSClient, err = getEthNodeAndClient(sn.Chain, viper.GetString("superNode.sync.wsPath"))
}
if sn.Chain == Bitcoin {
sn.NodeInfo = core.Node{
ID: "temporaryID",
ClientName: "omnicored",
GenesisBlock: "000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f",
NetworkID: "0xD9B4BEF9",
}
sn.WSClient = &rpcclient.ConnConfig{
Host: viper.GetString("superNode.sync.wsPath"),
Endpoint: "ws",
}
}
}
if sn.Serve {
wsPath := viper.GetString("superNode.server.wsPath")
@@ -130,9 +143,18 @@ func NewSuperNodeConfig() (*SuperNodeConfig, error) {
// BackFillFields is used to fill in the BackFill fields of the config
func (sn *SuperNodeConfig) BackFillFields() error {
sn.BackFill = true
_, httpClient, err := getNodeAndClient(sn.Chain, viper.GetString("superNode.backFill.httpPath"))
if err != nil {
return err
var httpClient interface{}
var err error
if sn.Chain == Ethereum {
_, httpClient, err = getEthNodeAndClient(sn.Chain, viper.GetString("superNode.backFill.httpPath"))
if err != nil {
return err
}
}
if sn.Chain == Bitcoin {
httpClient = &rpcclient.ConnConfig{
Host: viper.GetString("superNode.backFill.httpPath"),
}
}
sn.HTTPClient = httpClient
freq := viper.GetInt("superNode.backFill.frequency")
@@ -147,21 +169,16 @@ func (sn *SuperNodeConfig) BackFillFields() error {
return nil
}
func getNodeAndClient(chain ChainType, path string) (core.Node, interface{}, error) {
switch chain {
case Ethereum:
rawRPCClient, err := rpc.Dial(path)
if err != nil {
return core.Node{}, nil, err
}
rpcClient := client.NewRPCClient(rawRPCClient, path)
ethClient := ethclient.NewClient(rawRPCClient)
vdbEthClient := client.NewEthClient(ethClient)
vdbNode := node.MakeNode(rpcClient)
transactionConverter := vRpc.NewRPCTransactionConverter(ethClient)
blockChain := eth.NewBlockChain(vdbEthClient, rpcClient, vdbNode, transactionConverter)
return blockChain.Node(), rpcClient, nil
default:
return core.Node{}, nil, fmt.Errorf("unrecognized chain type %s", chain.String())
func getEthNodeAndClient(chain ChainType, path string) (core.Node, interface{}, error) {
rawRPCClient, err := rpc.Dial(path)
if err != nil {
return core.Node{}, nil, err
}
rpcClient := client.NewRPCClient(rawRPCClient, path)
ethClient := ethclient.NewClient(rawRPCClient)
vdbEthClient := client.NewEthClient(ethClient)
vdbNode := node.MakeNode(rpcClient)
transactionConverter := vRpc.NewRPCTransactionConverter(ethClient)
blockChain := eth.NewBlockChain(vdbEthClient, rpcClient, vdbNode, transactionConverter)
return blockChain.Node(), rpcClient, nil
}
pkg/super_node/shared/mocks/retriever.go
+1 -1
View File
@@ -17,7 +17,7 @@
package mocks
import (
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
)
pkg/super_node/shared/test_helpers.go
+1 -1
View File
@@ -19,7 +19,7 @@ package shared
import (
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/eth/core"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/postgres"
)
// SetupDB is use to setup a db for super node tests