Compare commits

..
14 Commits
Author SHA1 Message Date
Rob Mulholand 25f9c6c9e3 Merge pull request #150 from vulcanize/unique-headers-constraint-v2
Add constraint to prevent duplicate headers
2019-10-28 15:16:45 -05:00
Rob Mulholand e252229b8a Add constraint to prevent duplicate headers
- Disallow inserts of headers with the same number, hash, and node
  fingerprint, since it will enable duplicate log fetching for the
  same header
2019-10-28 14:57:13 -05:00
Rob Mulholand 62e1378e0c Merge pull request #163 from vulcanize/vdb-929-storage-key-lookup-cleanup
(VDB-929) Minimize storage key lookup bespoke code
2019-10-28 14:56:26 -05:00
Rob Mulholand b8fec5e4e3 (VDB-929) Minimize storage key lookup bespoke code
- Extract shared namespace for looking up and hashing keys
- Require storage transformers only to implement a loader that
  associates known keys with metadata
- Move key loader/lookup utils to utils directory to avoid
  multiple "storage" packages in imports
2019-10-28 14:29:09 -05:00
Ian Norden b7675316b4 Merge pull request #158 from vulcanize/missed_marked_checked_headers
Fix for issue #146
2019-10-28 12:34:47 -05:00
Ian Norden a2d249ca9d review fixes 2019-10-28 11:40:32 -05:00
Ian Norden 4fbde836d4 log sql.ErrNoRows which I suspect is what is leading to the flaky test 2019-10-28 09:37:21 -05:00
Ian Norden 65808998b3 goimports -w; golinting, remove some unused code 2019-10-28 09:37:21 -05:00
Ian Norden 11b5efbfe3 fix for issue #146; mark header checked for contract if it doesnt have
any logs at that header but other contracts do; test
2019-10-28 09:34:42 -05:00
Edvard Hübinette 3fff2896aa Rename geth to eth, signifying client independence (#161) 2019-10-28 12:30:24 +01:00
Edvard Hübinette f7c4a6736d VDB-919 Generalise converter (#152)
* Generalise transformer stack to use InsertionModel

* Add tests for event repository

* Restrict accepted values in InsertionModel

* Add call to repository.SetDB

* Improve error propagation/clarity on GetABI()

* Remove maker references in example

* Please golint

* refactor rollback error handling in repository

* Cleaner errors in repository, refactor tests
2019-10-28 11:48:31 +01:00
Rob Mulholand 6c055a9e12 Pin to urfave/cli version in go.mod (#154)
* Pin to urfave/cli version in go.mod

- Attempting to fix error: github.com/vulcanize/vulcanizedb@v0.0.8
  requires gopkg.in/urfave/cli.v1@v1.0.0-00010101000000-000000000000:
  invalid version: unknown revision 000000000000
2019-10-22 05:58:26 +09:00
Rob Mulholand 7be070fcea Merge pull request #156 from vulcanize/contract-watcher-init
Enable contractWatcher without prior headerSync
2019-10-10 08:37:48 +09:00
Rob Mulholand 2800e6df36 Enable contractWatcher without prior headerSync
- Previous setup would fail if there were no headers in the db. This
  makes sense because we need headers that haven't been checked for
  logs to exist so that we can fetch logs for those headers. But it
  also prevents us from kicking off the headerSync and contractWatcher
  processes concurrently. These changes enable kicking off both
  processes at the same time with the idea that we will have unchecked
  headers upon transformer execution.
2019-10-04 16:00:13 -05:00
115 changed files with 1237 additions and 569 deletions
+3 -3
View File
@@ -23,10 +23,10 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/crypto"
"github.com/vulcanize/vulcanizedb/pkg/datastore/ethereum"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
"github.com/vulcanize/vulcanizedb/pkg/eth/cold_import"
"github.com/vulcanize/vulcanizedb/pkg/eth/converters/cold_db"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/eth/converters/common"
"github.com/vulcanize/vulcanizedb/pkg/fs"
"github.com/vulcanize/vulcanizedb/pkg/geth/cold_import"
"github.com/vulcanize/vulcanizedb/pkg/geth/converters/cold_db"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/geth/converters/common"
"github.com/vulcanize/vulcanizedb/utils"
)
+2 -2
View File
@@ -25,7 +25,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/eth"
"github.com/vulcanize/vulcanizedb/pkg/history"
"github.com/vulcanize/vulcanizedb/utils"
)
@@ -100,7 +100,7 @@ func headerSync() {
}
}
func validateArgs(blockChain *geth.BlockChain) {
func validateArgs(blockChain *eth.BlockChain) {
lastBlock, err := blockChain.LastBlock()
if err != nil {
LogWithCommand.Error("validateArgs: Error getting last block: ", err)
+6 -6
View File
@@ -28,10 +28,10 @@ import (
"github.com/spf13/viper"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/geth/client"
vRpc "github.com/vulcanize/vulcanizedb/pkg/geth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/geth/node"
"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/node"
)
var (
@@ -155,12 +155,12 @@ func initConfig() {
}
}
func getBlockChain() *geth.BlockChain {
func getBlockChain() *eth.BlockChain {
rpcClient, ethClient := getClients()
vdbEthClient := client.NewEthClient(ethClient)
vdbNode := node.MakeNode(rpcClient)
transactionConverter := vRpc.NewRpcTransactionConverter(ethClient)
return geth.NewBlockChain(vdbEthClient, rpcClient, vdbNode, transactionConverter)
return eth.NewBlockChain(vdbEthClient, rpcClient, vdbNode, transactionConverter)
}
func getClients() (client.RpcClient, *ethclient.Client) {
+2 -1
View File
@@ -8,7 +8,8 @@ CREATE TABLE public.headers
block_timestamp NUMERIC,
check_count INTEGER NOT NULL DEFAULT 0,
eth_node_id INTEGER NOT NULL REFERENCES eth_nodes (id) ON DELETE CASCADE,
eth_node_fingerprint VARCHAR(128)
eth_node_fingerprint VARCHAR(128),
UNIQUE (block_number, hash, eth_node_fingerprint)
);
-- Index is removed when table is
+8
View File
@@ -921,6 +921,14 @@ ALTER TABLE ONLY public.header_sync_transactions
ADD CONSTRAINT header_sync_transactions_pkey PRIMARY KEY (id);
--
-- Name: headers headers_block_number_hash_eth_node_fingerprint_key; Type: CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.headers
ADD CONSTRAINT headers_block_number_hash_eth_node_fingerprint_key UNIQUE (block_number, hash, eth_node_fingerprint);
--
-- Name: headers headers_pkey; Type: CONSTRAINT; Schema: public; Owner: -
--
+1 -1
View File
@@ -49,7 +49,7 @@ require (
golang.org/x/sync v0.0.0-20190423024810-112230192c58
gopkg.in/olebedev/go-duktape.v3 v3.0.0-20190709231704-1e4459ed25ff // indirect
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7
gopkg.in/urfave/cli.v1 v1.0.0-00010101000000-000000000000 // indirect
gopkg.in/urfave/cli.v1 v1.20.0 // indirect
)
replace github.com/ethereum/go-ethereum => github.com/vulcanize/go-ethereum v0.0.0-20190731183759-8e20673bd101
+6 -6
View File
@@ -22,10 +22,10 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/geth/client"
vRpc "github.com/vulcanize/vulcanizedb/pkg/geth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/geth/node"
"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/node"
"github.com/vulcanize/vulcanizedb/test_config"
)
@@ -39,7 +39,7 @@ var _ = Describe("Rewards calculations", func() {
blockChainClient := client.NewEthClient(ethClient)
node := node.MakeNode(rpcClient)
transactionConverter := vRpc.NewRpcTransactionConverter(ethClient)
blockChain := geth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
blockChain := eth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
block, err := blockChain.GetBlockByNumber(1071819)
Expect(err).ToNot(HaveOccurred())
Expect(block.Reward).To(Equal("5313550000000000000"))
@@ -53,7 +53,7 @@ var _ = Describe("Rewards calculations", func() {
blockChainClient := client.NewEthClient(ethClient)
node := node.MakeNode(rpcClient)
transactionConverter := vRpc.NewRpcTransactionConverter(ethClient)
blockChain := geth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
blockChain := eth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
block, err := blockChain.GetBlockByNumber(1071819)
Expect(err).ToNot(HaveOccurred())
Expect(block.UnclesReward).To(Equal("6875000000000000000"))
+8 -8
View File
@@ -26,11 +26,11 @@ import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/geth/client"
rpc2 "github.com/vulcanize/vulcanizedb/pkg/geth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/geth/node"
"github.com/vulcanize/vulcanizedb/pkg/geth/testing"
"github.com/vulcanize/vulcanizedb/pkg/eth"
"github.com/vulcanize/vulcanizedb/pkg/eth/client"
rpc2 "github.com/vulcanize/vulcanizedb/pkg/eth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/eth/node"
"github.com/vulcanize/vulcanizedb/pkg/eth/testing"
"github.com/vulcanize/vulcanizedb/test_config"
)
@@ -56,7 +56,7 @@ var _ = Describe("Reading contracts", func() {
blockChainClient := client.NewEthClient(ethClient)
node := node.MakeNode(rpcClient)
transactionConverter := rpc2.NewRpcTransactionConverter(ethClient)
blockChain := geth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
blockChain := eth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
contract := testing.SampleContract()
logs, err := blockChain.GetFullSyncLogs(contract, big.NewInt(4703824), nil)
@@ -74,7 +74,7 @@ var _ = Describe("Reading contracts", func() {
blockChainClient := client.NewEthClient(ethClient)
node := node.MakeNode(rpcClient)
transactionConverter := rpc2.NewRpcTransactionConverter(ethClient)
blockChain := geth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
blockChain := eth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
logs, err := blockChain.GetFullSyncLogs(core.Contract{Hash: "0x123"}, big.NewInt(4703824), nil)
@@ -92,7 +92,7 @@ var _ = Describe("Reading contracts", func() {
blockChainClient := client.NewEthClient(ethClient)
node := node.MakeNode(rpcClient)
transactionConverter := rpc2.NewRpcTransactionConverter(ethClient)
blockChain := geth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
blockChain := eth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
contract := testing.SampleContract()
var balance = new(big.Int)
@@ -18,7 +18,6 @@ package integration
import (
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/config"
"math/rand"
"strings"
"time"
@@ -27,6 +26,7 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/full/transformer"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/constants"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers"
@@ -40,8 +40,8 @@ var _ = Describe("contractWatcher headerSync transformer", func() {
var blockChain core.BlockChain
var headerRepository repositories.HeaderRepository
var headerID int64
var ensAddr = strings.ToLower(constants.EnsContractAddress)
var tusdAddr = strings.ToLower(constants.TusdContractAddress)
var ensAddr = strings.ToLower(constants.EnsContractAddress) // 0x314159265dd8dbb310642f98f50c066173c1259b
var tusdAddr = strings.ToLower(constants.TusdContractAddress) // 0x8dd5fbce2f6a956c3022ba3663759011dd51e73e
BeforeEach(func() {
db, blockChain = test_helpers.SetupDBandBC()
@@ -78,11 +78,10 @@ var _ = Describe("contractWatcher headerSync transformer", func() {
Expect(c.Address).To(Equal(tusdAddr))
})
It("Fails to initialize if first and block cannot be fetched from vDB headers table", func() {
It("initializes when no headers available in db", func() {
t := transformer.NewTransformer(test_helpers.TusdConfig, blockChain, db)
err = t.Init()
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("no rows in result set"))
Expect(err).ToNot(HaveOccurred())
})
It("Does nothing if nothing if no addresses are configured", func() {
@@ -378,6 +377,42 @@ var _ = Describe("contractWatcher headerSync transformer", func() {
Expect(transferLog.Value).To(Equal("2800000000000000000000"))
})
It("Marks header checked for a contract that has no logs at that header", func() {
t := transformer.NewTransformer(test_helpers.ENSandTusdConfig, blockChain, db)
err = t.Init()
Expect(err).ToNot(HaveOccurred())
err = t.Execute()
Expect(err).ToNot(HaveOccurred())
Expect(t.Start).To(Equal(int64(6885702)))
newOwnerLog := test_helpers.HeaderSyncNewOwnerLog{}
err = db.QueryRowx(fmt.Sprintf("SELECT * FROM header_%s.newowner_event", ensAddr)).StructScan(&newOwnerLog)
Expect(err).ToNot(HaveOccurred())
transferLog := test_helpers.HeaderSyncTransferLog{}
err = db.QueryRowx(fmt.Sprintf("SELECT * FROM header_%s.transfer_event", tusdAddr)).StructScan(&transferLog)
Expect(err).ToNot(HaveOccurred())
Expect(transferLog.HeaderID).ToNot(Equal(newOwnerLog.HeaderID))
type checkedHeader struct {
ID int64 `db:"id"`
HeaderID int64 `db:"header_id"`
NewOwner int64 `db:"newowner_0x314159265dd8dbb310642f98f50c066173c1259b"`
Transfer int64 `db:"transfer_0x8dd5fbce2f6a956c3022ba3663759011dd51e73e"`
}
transferCheckedHeader := new(checkedHeader)
err = db.QueryRowx("SELECT * FROM public.checked_headers WHERE header_id = $1", transferLog.HeaderID).StructScan(transferCheckedHeader)
Expect(err).ToNot(HaveOccurred())
Expect(transferCheckedHeader.Transfer).To(Equal(int64(1)))
Expect(transferCheckedHeader.NewOwner).To(Equal(int64(1)))
newOwnerCheckedHeader := new(checkedHeader)
err = db.QueryRowx("SELECT * FROM public.checked_headers WHERE header_id = $1", newOwnerLog.HeaderID).StructScan(newOwnerCheckedHeader)
Expect(err).ToNot(HaveOccurred())
Expect(newOwnerCheckedHeader.NewOwner).To(Equal(int64(1)))
Expect(newOwnerCheckedHeader.Transfer).To(Equal(int64(1)))
})
It("Keeps track of contract-related hashes and addresses while transforming event data if they need to be used for later method polling", func() {
var testConf config.ContractConfig
testConf = test_helpers.ENSandTusdConfig
+6 -6
View File
@@ -24,17 +24,17 @@ import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/eth"
"github.com/vulcanize/vulcanizedb/pkg/eth/client"
rpc2 "github.com/vulcanize/vulcanizedb/pkg/eth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/eth/node"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/geth/client"
rpc2 "github.com/vulcanize/vulcanizedb/pkg/geth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/geth/node"
"github.com/vulcanize/vulcanizedb/pkg/history"
"github.com/vulcanize/vulcanizedb/test_config"
)
var _ = Describe("Reading from the Geth blockchain", func() {
var blockChain *geth.BlockChain
var blockChain *eth.BlockChain
BeforeEach(func() {
rawRpcClient, err := rpc.Dial(test_config.InfuraClient.IPCPath)
@@ -44,7 +44,7 @@ var _ = Describe("Reading from the Geth blockchain", func() {
blockChainClient := client.NewEthClient(ethClient)
node := node.MakeNode(rpcClient)
transactionConverter := rpc2.NewRpcTransactionConverter(ethClient)
blockChain = geth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
blockChain = eth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
})
It("reads two blocks", func(done Done) {
@@ -17,6 +17,8 @@
package integration_test
import (
"github.com/sirupsen/logrus"
"io/ioutil"
"testing"
. "github.com/onsi/ginkgo"
@@ -27,3 +29,7 @@ func TestIntegrationTest(t *testing.T) {
RegisterFailHandler(Fail)
RunSpecs(t, "IntegrationTest Suite")
}
var _ = BeforeSuite(func() {
logrus.SetOutput(ioutil.Discard)
})
@@ -16,9 +16,13 @@
package event
import "github.com/vulcanize/vulcanizedb/pkg/core"
import (
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
// Converter transforms log data into general InsertionModels the Repository can persist__
type Converter interface {
ToEntities(contractAbi string, ethLog []core.HeaderSyncLog) ([]interface{}, error)
ToModels([]interface{}) ([]interface{}, error)
ToModels(contractAbi string, ethLog []core.HeaderSyncLog) ([]InsertionModel, error)
SetDB(db *postgres.DB)
}
+156 -2
View File
@@ -16,9 +16,163 @@
package event
import "github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
import (
"database/sql/driver"
"fmt"
"github.com/vulcanize/vulcanizedb/utils"
"strings"
"github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
const SetLogTransformedQuery = `UPDATE public.header_sync_logs SET transformed = true WHERE id = $1`
// Repository persists transformed values to the DB
type Repository interface {
Create(models []interface{}) error
Create(models []InsertionModel) error
SetDB(db *postgres.DB)
}
// LogFK is the name of log foreign key columns
const LogFK ColumnName = "log_id"
// AddressFK is the name of address foreign key columns
const AddressFK ColumnName = "address_id"
// HeaderFK is the name of header foreign key columns
const HeaderFK ColumnName = "header_id"
// SchemaName is the schema to work with
type SchemaName string
// TableName identifies the table for inserting the data
type TableName string
// ColumnName identifies columns on the given table
type ColumnName string
// ColumnValues maps a column to the value for insertion. This is restricted to []byte, bool, float64, int64, string, time.Time
type ColumnValues map[ColumnName]interface{}
// ErrUnsupportedValue is thrown when a model supplies a type of value the postgres driver cannot handle.
var ErrUnsupportedValue = func(value interface{}) error {
return fmt.Errorf("unsupported type of value supplied in model: %v (%T)", value, value)
}
// InsertionModel is the generalised data structure a converter returns, and contains everything the repository needs to
// persist the converted data.
type InsertionModel struct {
SchemaName SchemaName
TableName TableName
OrderedColumns []ColumnName // Defines the fields to insert, and in which order the table expects them
ColumnValues ColumnValues // Associated values for columns, restricted to []byte, bool, float64, int64, string, time.Time
}
// ModelToQuery stores memoised insertion queries to minimise computation
var ModelToQuery = map[string]string{}
// GetMemoizedQuery gets/creates a DB insertion query for the model
func GetMemoizedQuery(model InsertionModel) string {
// The schema and table name uniquely determines the insertion query, use that for memoization
queryKey := string(model.SchemaName) + string(model.TableName)
query, queryMemoized := ModelToQuery[queryKey]
if !queryMemoized {
query = GenerateInsertionQuery(model)
ModelToQuery[queryKey] = query
}
return query
}
// GenerateInsertionQuery creates an SQL insertion query from an insertion model.
// Should be called through GetMemoizedQuery, so the query is not generated on each call to Create.
func GenerateInsertionQuery(model InsertionModel) string {
var valuePlaceholders []string
var updateOnConflict []string
for i := 0; i < len(model.OrderedColumns); i++ {
valuePlaceholder := fmt.Sprintf("$%d", 1+i)
valuePlaceholders = append(valuePlaceholders, valuePlaceholder)
updateOnConflict = append(updateOnConflict,
fmt.Sprintf("%s = %s", model.OrderedColumns[i], valuePlaceholder))
}
baseQuery := `INSERT INTO %v.%v (%v) VALUES(%v)
ON CONFLICT (header_id, log_id) DO UPDATE SET %v;`
return fmt.Sprintf(baseQuery,
model.SchemaName,
model.TableName,
joinOrderedColumns(model.OrderedColumns),
strings.Join(valuePlaceholders, ", "),
strings.Join(updateOnConflict, ", "))
}
/*
Create generates an insertion query and persists to the DB, given a slice of InsertionModels.
ColumnValues are restricted to []byte, bool, float64, int64, string, time.Time.
testModel = shared.InsertionModel{
SchemaName: "public"
TableName: "testEvent",
OrderedColumns: []string{"header_id", "log_id", "variable1"},
ColumnValues: ColumnValues{
"header_id": 303
"log_id": "808",
"variable1": "value1",
},
}
*/
func Create(models []InsertionModel, db *postgres.DB) error {
if len(models) == 0 {
return fmt.Errorf("repository got empty model slice")
}
tx, dbErr := db.Beginx()
if dbErr != nil {
return dbErr
}
for _, model := range models {
// Maps can't be iterated over in a reliable manner, so we rely on OrderedColumns to define the order to insert
// tx.Exec is variadically typed in the args, so if we wrap in []interface{} we can apply them all automatically
var args []interface{}
for _, col := range model.OrderedColumns {
value := model.ColumnValues[col]
// Check whether or not PG can accept the type of value in the model
okPgValue := driver.IsValue(value)
if !okPgValue {
logrus.WithField("model", model).Errorf("PG cannot handle value of this type: %T", value)
return ErrUnsupportedValue(value)
}
args = append(args, value)
}
insertionQuery := GetMemoizedQuery(model)
_, execErr := tx.Exec(insertionQuery, args...) // couldn't pass varying types in bulk with args :: []string
if execErr != nil {
rollbackErr := tx.Rollback()
if rollbackErr != nil {
logrus.Error("failed to rollback ", rollbackErr)
}
return execErr
}
_, logErr := tx.Exec(SetLogTransformedQuery, model.ColumnValues[LogFK])
if logErr != nil {
utils.RollbackAndLogFailure(tx, logErr, "header_sync_logs.transformed")
return logErr
}
}
return tx.Commit()
}
func joinOrderedColumns(columns []ColumnName) string {
var stringColumns []string
for _, columnName := range columns {
stringColumns = append(stringColumns, string(columnName))
}
return strings.Join(stringColumns, ", ")
}
@@ -0,0 +1,204 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program 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 Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package event_test
import (
"fmt"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/libraries/shared/factories/event"
"github.com/vulcanize/vulcanizedb/libraries/shared/test_data"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/test_config"
"math/big"
)
var _ = Describe("Repository", func() {
var db *postgres.DB
BeforeEach(func() {
db = test_config.NewTestDB(test_config.NewTestNode())
test_config.CleanTestDB(db)
})
Describe("Create", func() {
const createTestEventTableQuery = `CREATE TABLE public.testEvent(
id SERIAL PRIMARY KEY,
header_id INTEGER NOT NULL REFERENCES headers (id) ON DELETE CASCADE,
log_id BIGINT NOT NULL REFERENCES header_sync_logs (id) ON DELETE CASCADE,
variable1 TEXT,
UNIQUE (header_id, log_id)
);`
var (
headerID, logID int64
headerRepository repositories.HeaderRepository
testModel event.InsertionModel
)
BeforeEach(func() {
_, tableErr := db.Exec(createTestEventTableQuery)
Expect(tableErr).NotTo(HaveOccurred())
headerRepository = repositories.NewHeaderRepository(db)
var insertHeaderErr error
headerID, insertHeaderErr = headerRepository.CreateOrUpdateHeader(fakes.FakeHeader)
Expect(insertHeaderErr).NotTo(HaveOccurred())
headerSyncLog := test_data.CreateTestLog(headerID, db)
logID = headerSyncLog.ID
testModel = event.InsertionModel{
SchemaName: "public",
TableName: "testEvent",
OrderedColumns: []event.ColumnName{
event.HeaderFK, event.LogFK, "variable1",
},
ColumnValues: event.ColumnValues{
event.HeaderFK: headerID,
event.LogFK: logID,
"variable1": "value1",
},
}
})
AfterEach(func() {
db.MustExec(`DROP TABLE public.testEvent;`)
})
// Needs to run before the other tests, since those insert keys in map
It("memoizes queries", func() {
Expect(len(event.ModelToQuery)).To(Equal(0))
event.GetMemoizedQuery(testModel)
Expect(len(event.ModelToQuery)).To(Equal(1))
event.GetMemoizedQuery(testModel)
Expect(len(event.ModelToQuery)).To(Equal(1))
})
It("persists a model to postgres", func() {
createErr := event.Create([]event.InsertionModel{testModel}, db)
Expect(createErr).NotTo(HaveOccurred())
var res TestEvent
dbErr := db.Get(&res, `SELECT log_id, variable1 FROM public.testEvent;`)
Expect(dbErr).NotTo(HaveOccurred())
Expect(res.LogID).To(Equal(fmt.Sprint(testModel.ColumnValues[event.LogFK])))
Expect(res.Variable1).To(Equal(testModel.ColumnValues["variable1"]))
})
Describe("returns errors", func() {
It("for empty model slice", func() {
err := event.Create([]event.InsertionModel{}, db)
Expect(err).To(MatchError("repository got empty model slice"))
})
It("for failed SQL inserts", func() {
header := fakes.GetFakeHeader(1)
headerID, headerErr := headerRepository.CreateOrUpdateHeader(header)
Expect(headerErr).NotTo(HaveOccurred())
brokenModel := event.InsertionModel{
SchemaName: "public",
TableName: "testEvent",
// Wrong name of last column compared to DB, will generate incorrect query
OrderedColumns: []event.ColumnName{
event.HeaderFK, event.LogFK, "variable2",
},
ColumnValues: event.ColumnValues{
event.HeaderFK: headerID,
event.LogFK: logID,
"variable1": "value1",
},
}
// Remove cached queries, or we won't generate a new (incorrect) one
delete(event.ModelToQuery, "publictestEvent")
createErr := event.Create([]event.InsertionModel{brokenModel}, db)
// Remove incorrect query, so other tests won't get it
delete(event.ModelToQuery, "publictestEvent")
Expect(createErr).To(HaveOccurred())
})
It("for unsupported types in ColumnValue", func() {
unsupportedValue := big.NewInt(5)
testModel = event.InsertionModel{
SchemaName: "public",
TableName: "testEvent",
OrderedColumns: []event.ColumnName{
event.HeaderFK, event.LogFK, "variable1",
},
ColumnValues: event.ColumnValues{
event.HeaderFK: headerID,
event.LogFK: logID,
"variable1": unsupportedValue,
},
}
createErr := event.Create([]event.InsertionModel{testModel}, db)
Expect(createErr).To(MatchError(event.ErrUnsupportedValue(unsupportedValue)))
})
})
It("upserts queries with conflicting source", func() {
conflictingModel := event.InsertionModel{
SchemaName: "public",
TableName: "testEvent",
OrderedColumns: []event.ColumnName{
event.HeaderFK, event.LogFK, "variable1",
},
ColumnValues: event.ColumnValues{
event.HeaderFK: headerID,
event.LogFK: logID,
"variable1": "conflictingValue",
},
}
createErr := event.Create([]event.InsertionModel{testModel, conflictingModel}, db)
Expect(createErr).NotTo(HaveOccurred())
var res TestEvent
dbErr := db.Get(&res, `SELECT log_id, variable1 FROM public.testEvent;`)
Expect(dbErr).NotTo(HaveOccurred())
Expect(res.Variable1).To(Equal(conflictingModel.ColumnValues["variable1"]))
})
It("generates correct queries", func() {
actualQuery := event.GenerateInsertionQuery(testModel)
expectedQuery := `INSERT INTO public.testEvent (header_id, log_id, variable1) VALUES($1, $2, $3)
ON CONFLICT (header_id, log_id) DO UPDATE SET header_id = $1, log_id = $2, variable1 = $3;`
Expect(actualQuery).To(Equal(expectedQuery))
})
It("marks log transformed", func() {
createErr := event.Create([]event.InsertionModel{testModel}, db)
Expect(createErr).NotTo(HaveOccurred())
var logTransformed bool
getErr := db.Get(&logTransformed, `SELECT transformed FROM public.header_sync_logs WHERE id = $1`, logID)
Expect(getErr).NotTo(HaveOccurred())
Expect(logTransformed).To(BeTrue())
})
})
})
type TestEvent struct {
LogID string `db:"log_id"`
Variable1 string
}
@@ -30,6 +30,7 @@ type Transformer struct {
}
func (transformer Transformer) NewTransformer(db *postgres.DB) transformer.EventTransformer {
transformer.Converter.SetDB(db)
transformer.Repository.SetDB(db)
return transformer
}
@@ -42,13 +43,7 @@ func (transformer Transformer) Execute(logs []core.HeaderSyncLog) error {
return nil
}
entities, err := transformer.Converter.ToEntities(config.ContractAbi, logs)
if err != nil {
logrus.Errorf("error converting logs to entities in %v: %v", transformerName, err)
return err
}
models, err := transformer.Converter.ToModels(entities)
models, err := transformer.Converter.ToModels(config.ContractAbi, logs)
if err != nil {
logrus.Errorf("error converting entities to models in %v: %v", transformerName, err)
return err
@@ -66,12 +66,11 @@ var _ = Describe("Transformer", func() {
err := t.Execute([]core.HeaderSyncLog{})
Expect(err).NotTo(HaveOccurred())
Expect(converter.ToEntitiesCalledCounter).To(Equal(0))
Expect(converter.ToModelsCalledCounter).To(Equal(0))
Expect(repository.CreateCalledCounter).To(Equal(0))
})
It("converts an eth log to an entity", func() {
It("converts an eth log to a model", func() {
err := t.Execute(logs)
Expect(err).NotTo(HaveOccurred())
@@ -79,26 +78,7 @@ var _ = Describe("Transformer", func() {
Expect(converter.LogsToConvert).To(Equal(logs))
})
It("returns an error if converter fails", func() {
converter.ToEntitiesError = fakes.FakeError
err := t.Execute(logs)
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(fakes.FakeError))
})
It("converts an entity to a model", func() {
converter.EntitiesToReturn = []interface{}{test_data.GenericEntity{}}
err := t.Execute(logs)
Expect(err).NotTo(HaveOccurred())
Expect(converter.EntitiesToConvert[0]).To(Equal(test_data.GenericEntity{}))
})
It("returns an error if converting to models fails", func() {
converter.EntitiesToReturn = []interface{}{test_data.GenericEntity{}}
converter.ToModelsError = fakes.FakeError
err := t.Execute(logs)
@@ -108,12 +88,12 @@ var _ = Describe("Transformer", func() {
})
It("persists the record", func() {
converter.ModelsToReturn = []interface{}{test_data.GenericModel{}}
converter.ModelsToReturn = []event.InsertionModel{test_data.GenericModel}
err := t.Execute(logs)
Expect(err).NotTo(HaveOccurred())
Expect(repository.PassedModels[0]).To(Equal(test_data.GenericModel{}))
Expect(repository.PassedModels[0]).To(Equal(test_data.GenericModel))
})
It("returns error if persisting the record fails", func() {
+26 -10
View File
@@ -31,15 +31,15 @@ The storage transformer depends on contract-specific implementations of code cap
```golang
func (transformer Transformer) Execute(row shared.StorageDiffRow) error {
metadata, lookupErr := transformer.Mappings.Lookup(row.StorageKey)
metadata, lookupErr := transformer.StorageKeysLookup.Lookup(diff.StorageKey)
if lookupErr != nil {
return lookupErr
}
value, decodeErr := shared.Decode(row, metadata)
value, decodeErr := utils.Decode(diff, metadata)
if decodeErr != nil {
return decodeErr
}
return transformer.Repository.Create(row.BlockHeight, row.BlockHash.Hex(), metadata, value)
return transformer.Repository.Create(diff.BlockHeight, diff.BlockHash.Hex(), metadata, value)
}
```
@@ -47,20 +47,36 @@ func (transformer Transformer) Execute(row shared.StorageDiffRow) error {
In order to watch an additional smart contract, a developer must create three things:
1. Mappings - specify how to identify keys in the contract's storage trie.
1. StorageKeysLoader - identify keys in the contract's storage trie, providing metadata to describe how associated values should be decoded.
1. Repository - specify how to persist a parsed version of the storage value matching the recognized storage key.
1. Instance - create an instance of the storage transformer that uses your mappings and repository.
### Mappings
### StorageKeysLoader
A `StorageKeysLoader` is used by the `StorageKeysLookup` object on a storage transformer.
```golang
type Mappings interface {
Lookup(key common.Hash) (shared.StorageValueMetadata, error)
type KeysLoader interface {
LoadMappings() (map[common.Hash]utils.StorageValueMetadata, error)
SetDB(db *postgres.DB)
}
```
A contract-specific implementation of the mappings interface enables the storage transformer to fetch metadata associated with a storage key.
When a key is not found, the lookup object refreshes its known keys by calling the loader.
```golang
func (lookup *keysLookup) refreshMappings() error {
var err error
lookup.mappings, err = lookup.loader.LoadMappings()
if err != nil {
return err
}
lookup.mappings = utils.AddHashedKeys(lookup.mappings)
return nil
}
```
A contract-specific implementation of the loader enables the storage transformer to fetch metadata associated with a storage key.
Storage metadata contains: the name of the variable matching the storage key, a raw version of any keys associated with the variable (if the variable is a mapping), and the variable's type.
@@ -72,7 +88,7 @@ type StorageValueMetadata struct {
}
```
Keys are only relevant if the variable is a mapping. For example, in the following Solidity code:
The `Keys` field on the metadata is only relevant if the variable is a mapping. For example, in the following Solidity code:
```solidity
pragma solidity ^0.4.0;
@@ -85,7 +101,7 @@ contract Contract {
The metadata for variable `x` would not have any associated keys, but the metadata for a storage key associated with `y` would include the address used to specify that key's index in the mapping.
The `SetDB` function is required for the mappings to connect to the database.
The `SetDB` function is required for the storage key loader to connect to the database.
A database connection may be desired when keys in a mapping variable need to be read from log events (e.g. to lookup what addresses may exist in `y`, above).
### Repository
@@ -14,14 +14,15 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package repository
package storage
import "github.com/jmoiron/sqlx"
import (
"github.com/ethereum/go-ethereum/common"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage/utils"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
func MarkContractWatcherHeaderCheckedInTransaction(headerID int64, tx *sqlx.Tx, checkedHeadersColumn string) error {
_, err := tx.Exec(`INSERT INTO public.checked_headers (header_id, `+checkedHeadersColumn+`)
VALUES ($1, $2)
ON CONFLICT (header_id) DO
UPDATE SET `+checkedHeadersColumn+` = checked_headers.`+checkedHeadersColumn+` + 1`, headerID, 1)
return err
type KeysLoader interface {
LoadMappings() (map[common.Hash]utils.StorageValueMetadata, error)
SetDB(db *postgres.DB)
}
@@ -0,0 +1,66 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program 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 Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package storage
import (
"github.com/ethereum/go-ethereum/common"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage/utils"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type KeysLookup interface {
Lookup(key common.Hash) (utils.StorageValueMetadata, error)
SetDB(db *postgres.DB)
}
type keysLookup struct {
loader KeysLoader
mappings map[common.Hash]utils.StorageValueMetadata
}
func NewKeysLookup(loader KeysLoader) KeysLookup {
return &keysLookup{loader: loader, mappings: make(map[common.Hash]utils.StorageValueMetadata)}
}
func (lookup *keysLookup) Lookup(key common.Hash) (utils.StorageValueMetadata, error) {
metadata, ok := lookup.mappings[key]
if !ok {
refreshErr := lookup.refreshMappings()
if refreshErr != nil {
return metadata, refreshErr
}
metadata, ok = lookup.mappings[key]
if !ok {
return metadata, utils.ErrStorageKeyNotFound{Key: key.Hex()}
}
}
return metadata, nil
}
func (lookup *keysLookup) refreshMappings() error {
var err error
lookup.mappings, err = lookup.loader.LoadMappings()
if err != nil {
return err
}
lookup.mappings = utils.AddHashedKeys(lookup.mappings)
return nil
}
func (lookup *keysLookup) SetDB(db *postgres.DB) {
lookup.loader.SetDB(db)
}
@@ -0,0 +1,113 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program 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 Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package storage_test
import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/libraries/shared/factories/storage"
"github.com/vulcanize/vulcanizedb/libraries/shared/mocks"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage/utils"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/test_config"
)
var _ = Describe("Storage keys lookup", func() {
var (
fakeMetadata = utils.GetStorageValueMetadata("name", map[utils.Key]string{}, utils.Uint256)
lookup storage.KeysLookup
loader *mocks.MockStorageKeysLoader
)
BeforeEach(func() {
loader = &mocks.MockStorageKeysLoader{}
lookup = storage.NewKeysLookup(loader)
})
Describe("Lookup", func() {
Describe("when key not found", func() {
It("refreshes keys", func() {
loader.StorageKeyMappings = map[common.Hash]utils.StorageValueMetadata{fakes.FakeHash: fakeMetadata}
_, err := lookup.Lookup(fakes.FakeHash)
Expect(err).NotTo(HaveOccurred())
Expect(loader.LoadMappingsCallCount).To(Equal(1))
})
It("returns error if refreshing keys fails", func() {
loader.LoadMappingsError = fakes.FakeError
_, err := lookup.Lookup(fakes.FakeHash)
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(fakes.FakeError))
})
})
Describe("when key found", func() {
BeforeEach(func() {
loader.StorageKeyMappings = map[common.Hash]utils.StorageValueMetadata{fakes.FakeHash: fakeMetadata}
_, err := lookup.Lookup(fakes.FakeHash)
Expect(err).NotTo(HaveOccurred())
Expect(loader.LoadMappingsCallCount).To(Equal(1))
})
It("does not refresh keys", func() {
_, err := lookup.Lookup(fakes.FakeHash)
Expect(err).NotTo(HaveOccurred())
Expect(loader.LoadMappingsCallCount).To(Equal(1))
})
})
It("returns metadata for loaded static key", func() {
loader.StorageKeyMappings = map[common.Hash]utils.StorageValueMetadata{fakes.FakeHash: fakeMetadata}
metadata, err := lookup.Lookup(fakes.FakeHash)
Expect(err).NotTo(HaveOccurred())
Expect(metadata).To(Equal(fakeMetadata))
})
It("returns metadata for hashed version of key (accommodates keys emitted from Geth)", func() {
loader.StorageKeyMappings = map[common.Hash]utils.StorageValueMetadata{fakes.FakeHash: fakeMetadata}
hashedKey := common.BytesToHash(crypto.Keccak256(fakes.FakeHash.Bytes()))
metadata, err := lookup.Lookup(hashedKey)
Expect(err).NotTo(HaveOccurred())
Expect(metadata).To(Equal(fakeMetadata))
})
It("returns key not found error if key not found", func() {
_, err := lookup.Lookup(fakes.FakeHash)
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(utils.ErrStorageKeyNotFound{Key: fakes.FakeHash.Hex()}))
})
})
Describe("SetDB", func() {
It("sets the db on the loader", func() {
lookup.SetDB(test_config.NewTestDB(test_config.NewTestNode()))
Expect(loader.SetDBCalled).To(BeTrue())
})
})
})
@@ -18,20 +18,19 @@ package storage
import (
"github.com/ethereum/go-ethereum/common"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage/utils"
"github.com/vulcanize/vulcanizedb/libraries/shared/transformer"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type Transformer struct {
HashedAddress common.Hash
Mappings storage.Mappings
Repository Repository
HashedAddress common.Hash
StorageKeysLookup KeysLookup
Repository Repository
}
func (transformer Transformer) NewTransformer(db *postgres.DB) transformer.StorageTransformer {
transformer.Mappings.SetDB(db)
transformer.StorageKeysLookup.SetDB(db)
transformer.Repository.SetDB(db)
return transformer
}
@@ -41,7 +40,7 @@ func (transformer Transformer) KeccakContractAddress() common.Hash {
}
func (transformer Transformer) Execute(diff utils.StorageDiff) error {
metadata, lookupErr := transformer.Mappings.Lookup(diff.StorageKey)
metadata, lookupErr := transformer.StorageKeysLookup.Lookup(diff.StorageKey)
if lookupErr != nil {
return lookupErr
}
@@ -28,18 +28,18 @@ import (
var _ = Describe("Storage transformer", func() {
var (
mappings *mocks.MockMappings
repository *mocks.MockStorageRepository
t storage.Transformer
storageKeysLookup *mocks.MockStorageKeysLookup
repository *mocks.MockStorageRepository
t storage.Transformer
)
BeforeEach(func() {
mappings = &mocks.MockMappings{}
storageKeysLookup = &mocks.MockStorageKeysLookup{}
repository = &mocks.MockStorageRepository{}
t = storage.Transformer{
HashedAddress: common.Hash{},
Mappings: mappings,
Repository: repository,
HashedAddress: common.Hash{},
StorageKeysLookup: storageKeysLookup,
Repository: repository,
}
})
@@ -53,11 +53,11 @@ var _ = Describe("Storage transformer", func() {
It("looks up metadata for storage key", func() {
t.Execute(utils.StorageDiff{})
Expect(mappings.LookupCalled).To(BeTrue())
Expect(storageKeysLookup.LookupCalled).To(BeTrue())
})
It("returns error if lookup fails", func() {
mappings.LookupErr = fakes.FakeError
storageKeysLookup.LookupErr = fakes.FakeError
err := t.Execute(utils.StorageDiff{})
@@ -67,7 +67,7 @@ var _ = Describe("Storage transformer", func() {
It("creates storage row with decoded data", func() {
fakeMetadata := utils.StorageValueMetadata{Type: utils.Address}
mappings.Metadata = fakeMetadata
storageKeysLookup.Metadata = fakeMetadata
rawValue := common.HexToAddress("0x12345")
fakeBlockNumber := 123
fakeBlockHash := "0x67890"
@@ -91,7 +91,7 @@ var _ = Describe("Storage transformer", func() {
It("returns error if creating row fails", func() {
rawValue := common.HexToAddress("0x12345")
fakeMetadata := utils.StorageValueMetadata{Type: utils.Address}
mappings.Metadata = fakeMetadata
storageKeysLookup.Metadata = fakeMetadata
repository.CreateErr = fakes.FakeError
err := t.Execute(utils.StorageDiff{StorageValue: rawValue.Hash()})
@@ -118,7 +118,7 @@ var _ = Describe("Storage transformer", func() {
}
It("passes the decoded data items to the repository", func() {
mappings.Metadata = fakeMetadata
storageKeysLookup.Metadata = fakeMetadata
fakeRow := utils.StorageDiff{
HashedAddress: common.Hash{},
BlockHash: common.HexToHash(fakeBlockHash),
@@ -140,7 +140,7 @@ var _ = Describe("Storage transformer", func() {
})
It("returns error if creating a row fails", func() {
mappings.Metadata = fakeMetadata
storageKeysLookup.Metadata = fakeMetadata
repository.CreateErr = fakes.FakeError
err := t.Execute(utils.StorageDiff{StorageValue: rawValue.Hash()})
+14 -26
View File
@@ -16,41 +16,29 @@
package mocks
import "github.com/vulcanize/vulcanizedb/pkg/core"
import (
"github.com/vulcanize/vulcanizedb/libraries/shared/factories/event"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type MockConverter struct {
ToEntitiesError error
PassedContractAddresses []string
ToModelsError error
entityConverterError error
modelConverterError error
ContractAbi string
LogsToConvert []core.HeaderSyncLog
EntitiesToConvert []interface{}
EntitiesToReturn []interface{}
ModelsToReturn []interface{}
ToEntitiesCalledCounter int
ModelsToReturn []event.InsertionModel
PassedContractAddresses []string
SetDBCalled bool
ToModelsCalledCounter int
}
func (converter *MockConverter) ToEntities(contractAbi string, ethLogs []core.HeaderSyncLog) ([]interface{}, error) {
for _, log := range ethLogs {
converter.PassedContractAddresses = append(converter.PassedContractAddresses, log.Log.Address.Hex())
}
converter.ContractAbi = contractAbi
converter.LogsToConvert = ethLogs
return converter.EntitiesToReturn, converter.ToEntitiesError
}
func (converter *MockConverter) ToModels(entities []interface{}) ([]interface{}, error) {
converter.EntitiesToConvert = entities
func (converter *MockConverter) ToModels(abi string, logs []core.HeaderSyncLog) ([]event.InsertionModel, error) {
converter.LogsToConvert = logs
converter.ContractAbi = abi
converter.ToModelsCalledCounter = converter.ToModelsCalledCounter + 1
return converter.ModelsToReturn, converter.ToModelsError
}
func (converter *MockConverter) SetToEntityConverterError(err error) {
converter.entityConverterError = err
}
func (converter *MockConverter) SetToModelConverterError(err error) {
converter.modelConverterError = err
func (converter *MockConverter) SetDB(db *postgres.DB) {
converter.SetDBCalled = true
}
+3 -2
View File
@@ -17,17 +17,18 @@
package mocks
import (
"github.com/vulcanize/vulcanizedb/libraries/shared/factories/event"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type MockEventRepository struct {
createError error
PassedModels []interface{}
PassedModels []event.InsertionModel
SetDbCalled bool
CreateCalledCounter int
}
func (repository *MockEventRepository) Create(models []interface{}) error {
func (repository *MockEventRepository) Create(models []event.InsertionModel) error {
repository.PassedModels = models
repository.CreateCalledCounter++
@@ -0,0 +1,39 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program 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 Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package mocks
import (
"github.com/ethereum/go-ethereum/common"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage/utils"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type MockStorageKeysLoader struct {
LoadMappingsCallCount int
LoadMappingsError error
SetDBCalled bool
StorageKeyMappings map[common.Hash]utils.StorageValueMetadata
}
func (loader *MockStorageKeysLoader) LoadMappings() (map[common.Hash]utils.StorageValueMetadata, error) {
loader.LoadMappingsCallCount++
return loader.StorageKeyMappings, loader.LoadMappingsError
}
func (loader *MockStorageKeysLoader) SetDB(db *postgres.DB) {
loader.SetDBCalled = true
}
@@ -18,22 +18,21 @@ package mocks
import (
"github.com/ethereum/go-ethereum/common"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage/utils"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type MockMappings struct {
type MockStorageKeysLookup struct {
Metadata utils.StorageValueMetadata
LookupCalled bool
LookupErr error
}
func (mappings *MockMappings) Lookup(key common.Hash) (utils.StorageValueMetadata, error) {
func (mappings *MockStorageKeysLookup) Lookup(key common.Hash) (utils.StorageValueMetadata, error) {
mappings.LookupCalled = true
return mappings.Metadata, mappings.LookupErr
}
func (*MockMappings) SetDB(db *postgres.DB) {
func (*MockStorageKeysLookup) SetDB(db *postgres.DB) {
panic("implement me")
}
@@ -1,67 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program 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 Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package repository_test
import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/libraries/shared/repository"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/test_config"
)
var _ = Describe("", func() {
Describe("MarkContractWatcherHeaderCheckedInTransaction", func() {
var (
checkedHeadersColumn string
db *postgres.DB
)
BeforeEach(func() {
db = test_config.NewTestDB(test_config.NewTestNode())
test_config.CleanTestDB(db)
checkedHeadersColumn = "test_column_checked"
_, migrateErr := db.Exec(`ALTER TABLE public.checked_headers
ADD COLUMN ` + checkedHeadersColumn + ` integer`)
Expect(migrateErr).NotTo(HaveOccurred())
})
AfterEach(func() {
_, cleanupMigrateErr := db.Exec(`ALTER TABLE public.checked_headers DROP COLUMN ` + checkedHeadersColumn)
Expect(cleanupMigrateErr).NotTo(HaveOccurred())
})
It("marks passed header as checked within a passed transaction", func() {
headerRepository := repositories.NewHeaderRepository(db)
headerID, headerErr := headerRepository.CreateOrUpdateHeader(fakes.FakeHeader)
Expect(headerErr).NotTo(HaveOccurred())
tx, txErr := db.Beginx()
Expect(txErr).NotTo(HaveOccurred())
err := repository.MarkContractWatcherHeaderCheckedInTransaction(headerID, tx, checkedHeadersColumn)
Expect(err).NotTo(HaveOccurred())
commitErr := tx.Commit()
Expect(commitErr).NotTo(HaveOccurred())
var checkedCount int
fetchErr := db.Get(&checkedCount, `SELECT COUNT(*) FROM public.checked_headers WHERE header_id = $1`, headerID)
Expect(fetchErr).NotTo(HaveOccurred())
Expect(checkedCount).To(Equal(1))
})
})
})
@@ -14,23 +14,14 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package storage
package utils
import (
"math/big"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage/utils"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"math/big"
)
type Mappings interface {
Lookup(key common.Hash) (utils.StorageValueMetadata, error)
SetDB(db *postgres.DB)
}
const (
IndexZero = "0000000000000000000000000000000000000000000000000000000000000000"
IndexOne = "0000000000000000000000000000000000000000000000000000000000000001"
@@ -46,32 +37,17 @@ const (
IndexEleven = "000000000000000000000000000000000000000000000000000000000000000b"
)
func AddHashedKeys(currentMappings map[common.Hash]utils.StorageValueMetadata) map[common.Hash]utils.StorageValueMetadata {
copyOfCurrentMappings := make(map[common.Hash]utils.StorageValueMetadata)
for k, v := range currentMappings {
copyOfCurrentMappings[k] = v
}
for k, v := range copyOfCurrentMappings {
currentMappings[hashKey(k)] = v
}
return currentMappings
}
func hashKey(key common.Hash) common.Hash {
return crypto.Keccak256Hash(key.Bytes())
}
func GetMapping(indexOnContract, key string) common.Hash {
func GetStorageKeyForMapping(indexOnContract, key string) common.Hash {
keyBytes := common.FromHex(key + indexOnContract)
return crypto.Keccak256Hash(keyBytes)
}
func GetNestedMapping(indexOnContract, primaryKey, secondaryKey string) common.Hash {
func GetStorageKeyForNestedMapping(indexOnContract, primaryKey, secondaryKey string) common.Hash {
primaryMappingIndex := crypto.Keccak256(common.FromHex(primaryKey + indexOnContract))
return crypto.Keccak256Hash(common.FromHex(secondaryKey), primaryMappingIndex)
}
func GetIncrementedKey(original common.Hash, incrementBy int64) common.Hash {
func GetIncrementedStorageKey(original common.Hash, incrementBy int64) common.Hash {
originalMappingAsInt := original.Big()
incremented := big.NewInt(0).Add(originalMappingAsInt, big.NewInt(incrementBy))
return common.BytesToHash(incremented.Bytes())
@@ -1,78 +1,72 @@
package storage_test
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program 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 Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package utils_test
import (
"github.com/ethereum/go-ethereum/common"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage/utils"
)
var _ = Describe("Mappings", func() {
Describe("AddHashedKeys", func() {
It("returns a copy of the map with an additional slot for the hashed version of every key", func() {
fakeMap := map[common.Hash]utils.StorageValueMetadata{}
fakeStorageKey := common.HexToHash("72c72de6b203d67cb6cd54fc93300109fcc6fd6eac88e390271a3d548794d800")
var fakeMappingKey utils.Key = "fakeKey"
fakeMetadata := utils.StorageValueMetadata{
Name: "fakeName",
Keys: map[utils.Key]string{fakeMappingKey: "fakeValue"},
Type: utils.Uint48,
}
fakeMap[fakeStorageKey] = fakeMetadata
result := storage.AddHashedKeys(fakeMap)
Expect(len(result)).To(Equal(2))
expectedHashedStorageKey := common.HexToHash("2165edb4e1c37b99b60fa510d84f939dd35d5cd1d1c8f299d6456ea09df65a76")
Expect(fakeMap[fakeStorageKey]).To(Equal(fakeMetadata))
Expect(fakeMap[expectedHashedStorageKey]).To(Equal(fakeMetadata))
})
})
Describe("GetMapping", func() {
var _ = Describe("Storage keys loader utils", func() {
Describe("GetStorageKeyForMapping", func() {
It("returns the storage key for a mapping when passed the mapping's index on the contract and the desired value's key", func() {
// ex. solidity:
// mapping (bytes32 => uint) public amounts
// to access amounts, pass in the index of the mapping on the contract + the bytes32 key for the uint val being looked up
indexOfMappingOnContract := storage.IndexZero
indexOfMappingOnContract := utils.IndexZero
keyForDesiredValueInMapping := "1234567890abcdef"
storageKey := storage.GetMapping(indexOfMappingOnContract, keyForDesiredValueInMapping)
storageKey := utils.GetStorageKeyForMapping(indexOfMappingOnContract, keyForDesiredValueInMapping)
expectedStorageKey := common.HexToHash("0xee0c1b59a3856bafbfb8730e7694c4badc271eb5f01ce4a8d7a53d8a6499676f")
Expect(storageKey).To(Equal(expectedStorageKey))
})
It("returns same result if value includes hex prefix", func() {
indexOfMappingOnContract := storage.IndexZero
indexOfMappingOnContract := utils.IndexZero
keyForDesiredValueInMapping := "0x1234567890abcdef"
storageKey := storage.GetMapping(indexOfMappingOnContract, keyForDesiredValueInMapping)
storageKey := utils.GetStorageKeyForMapping(indexOfMappingOnContract, keyForDesiredValueInMapping)
expectedStorageKey := common.HexToHash("0xee0c1b59a3856bafbfb8730e7694c4badc271eb5f01ce4a8d7a53d8a6499676f")
Expect(storageKey).To(Equal(expectedStorageKey))
})
})
Describe("GetNestedMapping", func() {
Describe("GetStorageKeyForNestedMapping", func() {
It("returns the storage key for a nested mapping when passed the mapping's index on the contract and the desired value's keys", func() {
// ex. solidity:
// mapping (bytes32 => uint) public amounts
// mapping (address => mapping (uint => bytes32)) public addressNames
// to access addressNames, pass in the index of the mapping on the contract + the address and uint keys for the bytes32 val being looked up
indexOfMappingOnContract := storage.IndexOne
indexOfMappingOnContract := utils.IndexOne
keyForOuterMapping := "1234567890abcdef"
keyForInnerMapping := "123"
storageKey := storage.GetNestedMapping(indexOfMappingOnContract, keyForOuterMapping, keyForInnerMapping)
storageKey := utils.GetStorageKeyForNestedMapping(indexOfMappingOnContract, keyForOuterMapping, keyForInnerMapping)
expectedStorageKey := common.HexToHash("0x82113529f6cd61061d1a6f0de53f2bdd067a1addd3d2b46be50a99abfcdb1661")
Expect(storageKey).To(Equal(expectedStorageKey))
})
})
Describe("GetIncrementedKey", func() {
Describe("GetIncrementedStorageKey", func() {
It("returns the storage key for later values sharing an index on the contract with other earlier values", func() {
// ex. solidity:
// mapping (bytes32 => uint) public amounts
@@ -84,11 +78,11 @@ var _ = Describe("Mappings", func() {
// mapping (bytes32 => Data) public itemData;
// to access quality from itemData, pass in the storage key for the zero-indexed value (quantity) + the number of increments required.
// (For "quality", we must increment the storage key for the corresponding "quantity" by 1).
indexOfMappingOnContract := storage.IndexTwo
indexOfMappingOnContract := utils.IndexTwo
keyForDesiredValueInMapping := "1234567890abcdef"
storageKeyForFirstPropertyOnStruct := storage.GetMapping(indexOfMappingOnContract, keyForDesiredValueInMapping)
storageKeyForFirstPropertyOnStruct := utils.GetStorageKeyForMapping(indexOfMappingOnContract, keyForDesiredValueInMapping)
storageKey := storage.GetIncrementedKey(storageKeyForFirstPropertyOnStruct, 1)
storageKey := utils.GetIncrementedStorageKey(storageKeyForFirstPropertyOnStruct, 1)
expectedStorageKey := common.HexToHash("0x69b38749f0a8ed5d505c8474f7fb62c7828aad8a7627f1c67e07af1d2368cad4")
Expect(storageKey).To(Equal(expectedStorageKey))
@@ -0,0 +1,37 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program 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 Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package utils
import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
)
func AddHashedKeys(currentMappings map[common.Hash]StorageValueMetadata) map[common.Hash]StorageValueMetadata {
copyOfCurrentMappings := make(map[common.Hash]StorageValueMetadata)
for k, v := range currentMappings {
copyOfCurrentMappings[k] = v
}
for k, v := range copyOfCurrentMappings {
currentMappings[hashKey(k)] = v
}
return currentMappings
}
func hashKey(key common.Hash) common.Hash {
return crypto.Keccak256Hash(key.Bytes())
}
@@ -0,0 +1,47 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program 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 Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package utils_test
import (
"github.com/ethereum/go-ethereum/common"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage/utils"
)
var _ = Describe("Storage keys lookup utils", func() {
Describe("AddHashedKeys", func() {
It("returns a copy of the map with an additional slot for the hashed version of every key", func() {
fakeMap := map[common.Hash]utils.StorageValueMetadata{}
fakeStorageKey := common.HexToHash("72c72de6b203d67cb6cd54fc93300109fcc6fd6eac88e390271a3d548794d800")
var fakeMappingKey utils.Key = "fakeKey"
fakeMetadata := utils.StorageValueMetadata{
Name: "fakeName",
Keys: map[utils.Key]string{fakeMappingKey: "fakeValue"},
Type: utils.Uint48,
}
fakeMap[fakeStorageKey] = fakeMetadata
result := utils.AddHashedKeys(fakeMap)
Expect(len(result)).To(Equal(2))
expectedHashedStorageKey := common.HexToHash("2165edb4e1c37b99b60fa510d84f939dd35d5cd1d1c8f299d6456ea09df65a76")
Expect(fakeMap[fakeStorageKey]).To(Equal(fakeMetadata))
Expect(fakeMap[expectedHashedStorageKey]).To(Equal(fakeMetadata))
})
})
})
+3 -3
View File
@@ -20,14 +20,12 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/core/types"
"github.com/vulcanize/vulcanizedb/libraries/shared/factories/event"
"github.com/vulcanize/vulcanizedb/libraries/shared/transformer"
"math/rand"
"time"
)
type GenericModel struct{}
type GenericEntity struct{}
var startingBlockNumber = rand.Int63()
var topic0 = "0x" + randomString(64)
@@ -44,6 +42,8 @@ var GenericTestLog = func() types.Log {
}
}
var GenericModel = event.InsertionModel{}
var GenericTestConfig = transformer.EventTransformerConfig{
TransformerName: "generic-test-transformer",
ContractAddresses: []string{fakeAddress().Hex()},
@@ -0,0 +1,37 @@
package test_data
import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
"math/rand"
)
// Create a header sync log to reference in an event, returning inserted header sync log
func CreateTestLog(headerID int64, db *postgres.DB) core.HeaderSyncLog {
log := types.Log{
Address: common.Address{},
Topics: nil,
Data: nil,
BlockNumber: 0,
TxHash: common.Hash{},
TxIndex: uint(rand.Int31()),
BlockHash: common.Hash{},
Index: 0,
Removed: false,
}
headerSyncLogRepository := repositories.NewHeaderSyncLogRepository(db)
insertLogsErr := headerSyncLogRepository.CreateHeaderSyncLogs(headerID, []types.Log{log})
Expect(insertLogsErr).NotTo(HaveOccurred())
headerSyncLogs, getLogsErr := headerSyncLogRepository.GetUntransformedHeaderSyncLogs()
Expect(getLogsErr).NotTo(HaveOccurred())
for _, headerSyncLog := range headerSyncLogs {
if headerSyncLog.Log.TxIndex == log.TxIndex {
return headerSyncLog
}
}
panic("couldn't find inserted test log")
}
+2 -2
View File
@@ -19,7 +19,7 @@ package config
import (
log "github.com/sirupsen/logrus"
"github.com/spf13/viper"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/eth"
"strings"
)
@@ -98,7 +98,7 @@ func (contractConfig *ContractConfig) PrepConfig() {
}
}
if abi != "" {
if _, abiErr := geth.ParseAbi(abi); abiErr != nil {
if _, abiErr := eth.ParseAbi(abi); abiErr != nil {
log.Fatal(addr, "transformer `abi` not valid JSON")
}
}
@@ -17,7 +17,6 @@
package converter
import (
"errors"
"fmt"
"math/big"
"strconv"
@@ -32,22 +31,24 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/core"
)
// Converter is used to convert watched event logs to
// ConverterInterface is used to convert watched event logs to
// custom logs containing event input name => value maps
type ConverterInterface interface {
Convert(watchedEvent core.WatchedEvent, event types.Event) (*types.Log, error)
Update(info *contract.Contract)
}
// Converter is the underlying struct for the ConverterInterface
type Converter struct {
ContractInfo *contract.Contract
}
// Update configures the converter for a specific contract
func (c *Converter) Update(info *contract.Contract) {
c.ContractInfo = info
}
// Convert the given watched event log into a types.Log for the given event
// Convert converts the given watched event log into a types.Log for the given event
func (c *Converter) Convert(watchedEvent core.WatchedEvent, event types.Event) (*types.Log, error) {
boundContract := bind.NewBoundContract(common.HexToAddress(c.ContractInfo.Address), c.ContractInfo.ParsedAbi, nil, nil, nil)
values := make(map[string]interface{})
@@ -88,14 +89,14 @@ func (c *Converter) Convert(watchedEvent core.WatchedEvent, event types.Event) (
b := input.(byte)
strValues[fieldName] = string(b)
default:
return nil, errors.New(fmt.Sprintf("error: unhandled abi type %T", input))
return nil, fmt.Errorf("error: unhandled abi type %T", input)
}
}
// Only hold onto logs that pass our address filter, if any
if c.ContractInfo.PassesEventFilter(strValues) {
eventLog := &types.Log{
Id: watchedEvent.LogID,
ID: watchedEvent.LogID,
Values: strValues,
Block: watchedEvent.BlockNumber,
Tx: watchedEvent.TxHash,
@@ -18,11 +18,12 @@ package retriever
import (
"database/sql"
"github.com/vulcanize/vulcanizedb/libraries/shared/repository"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
// Block retriever is used to retrieve the first block for a given contract and the most recent block
// BlockRetriever is used to retrieve the first block for a given contract and the most recent block
// It requires a vDB synced database with blocks, transactions, receipts, and logs
type BlockRetriever interface {
RetrieveFirstBlock(contractAddr string) (int64, error)
@@ -33,13 +34,15 @@ type blockRetriever struct {
db *postgres.DB
}
func NewBlockRetriever(db *postgres.DB) (r *blockRetriever) {
// NewBlockRetriever returns a new BlockRetriever
func NewBlockRetriever(db *postgres.DB) BlockRetriever {
return &blockRetriever{
db: db,
}
}
// Try both methods of finding the first block, with the receipt method taking precedence
// RetrieveFirstBlock fetches the block number for the earliest block in the db
// Tries both methods of finding the first block, with the receipt method taking precedence
func (r *blockRetriever) RetrieveFirstBlock(contractAddr string) (int64, error) {
i, err := r.retrieveFirstBlockFromReceipts(contractAddr)
if err != nil {
@@ -55,7 +58,7 @@ func (r *blockRetriever) RetrieveFirstBlock(contractAddr string) (int64, error)
// For some contracts the contract creation transaction receipt doesn't have the contract address so this doesn't work (e.g. Sai)
func (r *blockRetriever) retrieveFirstBlockFromReceipts(contractAddr string) (int64, error) {
var firstBlock int64
addressId, getAddressErr := repository.GetOrCreateAddress(r.db, contractAddr)
addressID, getAddressErr := repository.GetOrCreateAddress(r.db, contractAddr)
if getAddressErr != nil {
return firstBlock, getAddressErr
}
@@ -66,7 +69,7 @@ func (r *blockRetriever) retrieveFirstBlockFromReceipts(contractAddr string) (in
WHERE contract_address_id = $1
ORDER BY block_id ASC
LIMIT 1)`,
addressId,
addressID,
)
return firstBlock, err
@@ -84,7 +87,7 @@ func (r *blockRetriever) retrieveFirstBlockFromLogs(contractAddr string) (int64,
return int64(firstBlock), err
}
// Method to retrieve the most recent block in vDB
// RetrieveMostRecentBlock retrieves the most recent block number in vDB
func (r *blockRetriever) RetrieveMostRecentBlock() (int64, error) {
var lastBlock int64
err := r.db.Get(
@@ -17,10 +17,11 @@
package retriever_test
import (
"strings"
"github.com/ethereum/go-ethereum/core/types"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"strings"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/full/retriever"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/constants"
@@ -17,12 +17,12 @@
package retriever_test
import (
"github.com/sirupsen/logrus"
"io/ioutil"
"testing"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/sirupsen/logrus"
)
func TestRetriever(t *testing.T) {
@@ -35,6 +35,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
)
// Transformer is the top level struct for transforming watched contract data
// Requires a fully synced vDB and a running eth node (or infura)
type Transformer struct {
// Database interfaces
@@ -60,7 +61,7 @@ type Transformer struct {
LastBlock int64
}
// Transformer takes in config for blockchain, database, and network id
// NewTransformer takes in contract config, blockchain, and database, and returns a new Transformer
func NewTransformer(con config.ContractConfig, BC core.BlockChain, DB *postgres.DB) *Transformer {
return &Transformer{
Poller: poller.NewPoller(BC, DB, types.FullSync),
@@ -75,6 +76,7 @@ func NewTransformer(con config.ContractConfig, BC core.BlockChain, DB *postgres.
}
}
// Init initializes the transformer
// Use after creating and setting transformer
// Loops over all of the addr => filter sets
// Uses parser to pull event info from abi
@@ -167,6 +169,7 @@ func (tr *Transformer) Init() error {
return nil
}
// Execute runs the transformation processes
// Iterates through stored, initialized contract objects
// Iterates through contract's event filters, grabbing watched event logs
// Uses converter to convert logs into custom log type
@@ -227,6 +230,7 @@ func (tr *Transformer) Execute() error {
return nil
}
// GetConfig returns the transformers config; satisfies the transformer interface
func (tr *Transformer) GetConfig() config.ContractConfig {
return tr.Config
}
@@ -17,12 +17,12 @@
package transformer_test
import (
"github.com/sirupsen/logrus"
"io/ioutil"
"testing"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/sirupsen/logrus"
)
func TestTransformer(t *testing.T) {
@@ -18,7 +18,6 @@ package converter
import (
"encoding/json"
"errors"
"fmt"
"math/big"
"strconv"
@@ -32,16 +31,19 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types"
)
// ConverterInterface is the interface for converting geth logs to our custom log type
type ConverterInterface interface {
Convert(logs []gethTypes.Log, event types.Event, headerID int64) ([]types.Log, error)
ConvertBatch(logs []gethTypes.Log, events map[string]types.Event, headerID int64) (map[string][]types.Log, error)
Update(info *contract.Contract)
}
// Converter is the underlying struct for the ConverterInterface
type Converter struct {
ContractInfo *contract.Contract
}
// Update is used to configure the converter with a specific contract
func (c *Converter) Update(info *contract.Contract) {
c.ContractInfo = info
}
@@ -98,7 +100,7 @@ func (c *Converter) Convert(logs []gethTypes.Log, event types.Event, headerID in
strValues[fieldName] = converted.String()
seenHashes = append(seenHashes, converted)
default:
return nil, errors.New(fmt.Sprintf("error: unhandled abi type %T", input))
return nil, fmt.Errorf("error: unhandled abi type %T", input)
}
}
@@ -114,7 +116,7 @@ func (c *Converter) Convert(logs []gethTypes.Log, event types.Event, headerID in
Values: strValues,
Raw: raw,
TransactionIndex: log.TxIndex,
Id: headerID,
ID: headerID,
})
// Cache emitted values if their caching is turned on
@@ -130,7 +132,7 @@ func (c *Converter) Convert(logs []gethTypes.Log, event types.Event, headerID in
return returnLogs, nil
}
// Convert the given watched event logs into types.Logs; returns a map of event names to a slice of their converted logs
// ConvertBatch converts the given watched event logs into types.Logs; returns a map of event names to a slice of their converted logs
func (c *Converter) ConvertBatch(logs []gethTypes.Log, events map[string]types.Event, headerID int64) (map[string][]types.Log, error) {
boundContract := bind.NewBoundContract(common.HexToAddress(c.ContractInfo.Address), c.ContractInfo.ParsedAbi, nil, nil, nil)
eventsToLogs := make(map[string][]types.Log)
@@ -182,7 +184,7 @@ func (c *Converter) ConvertBatch(logs []gethTypes.Log, events map[string]types.E
strValues[fieldName] = converted.String()
seenHashes = append(seenHashes, converted)
default:
return nil, errors.New(fmt.Sprintf("error: unhandled abi type %T", input))
return nil, fmt.Errorf("error: unhandled abi type %T", input)
}
}
@@ -198,7 +200,7 @@ func (c *Converter) ConvertBatch(logs []gethTypes.Log, events map[string]types.E
Values: strValues,
Raw: raw,
TransactionIndex: log.TxIndex,
Id: headerID,
ID: headerID,
})
// Cache emitted values that pass the argument filter if their caching is turned on
@@ -72,11 +72,11 @@ var _ = Describe("Converter", func() {
Expect(logs[0].Values["to"]).To(Equal(sender1.String()))
Expect(logs[0].Values["from"]).To(Equal(sender2.String()))
Expect(logs[0].Values["value"]).To(Equal(value.String()))
Expect(logs[0].Id).To(Equal(int64(232)))
Expect(logs[0].ID).To(Equal(int64(232)))
Expect(logs[1].Values["to"]).To(Equal(sender2.String()))
Expect(logs[1].Values["from"]).To(Equal(sender1.String()))
Expect(logs[1].Values["value"]).To(Equal(value.String()))
Expect(logs[1].Id).To(Equal(int64(232)))
Expect(logs[1].ID).To(Equal(int64(232)))
})
It("Keeps track of addresses it sees if they will be used for method polling", func() {
@@ -24,6 +24,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/core"
)
// Fetcher is the fetching interface
type Fetcher interface {
FetchLogs(contractAddresses []string, topics []common.Hash, missingHeader core.Header) ([]types.Log, error)
}
@@ -32,13 +33,14 @@ type fetcher struct {
blockChain core.BlockChain
}
func NewFetcher(blockchain core.BlockChain) *fetcher {
// NewFetcher returns a new Fetcher
func NewFetcher(blockchain core.BlockChain) Fetcher {
return &fetcher{
blockChain: blockchain,
}
}
// Checks all topic0s, on all addresses, fetching matching logs for the given header
// FetchLogs checks all topic0s, on all addresses, fetching matching logs for the given header
func (fetcher *fetcher) FetchLogs(contractAddresses []string, topic0s []common.Hash, header core.Header) ([]types.Log, error) {
addresses := hexStringsToAddresses(contractAddresses)
blockHash := common.HexToHash(header.Hash)
@@ -20,7 +20,6 @@ import (
"fmt"
"github.com/hashicorp/golang-lru"
"github.com/jmoiron/sqlx"
"github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/pkg/core"
@@ -29,6 +28,7 @@ import (
const columnCacheSize = 1000
// HeaderRepository interfaces with the header and checked_headers tables
type HeaderRepository interface {
AddCheckColumn(id string) error
AddCheckColumns(ids []string) error
@@ -46,7 +46,8 @@ type headerRepository struct {
columns *lru.Cache // Cache created columns to minimize db connections
}
func NewHeaderRepository(db *postgres.DB) *headerRepository {
// NewHeaderRepository returns a new HeaderRepository
func NewHeaderRepository(db *postgres.DB) HeaderRepository {
ccs, _ := lru.New(columnCacheSize)
return &headerRepository{
db: db,
@@ -54,7 +55,7 @@ func NewHeaderRepository(db *postgres.DB) *headerRepository {
}
}
// Adds a checked_header column for the provided column id
// AddCheckColumn adds a checked_header column for the provided column id
func (r *headerRepository) AddCheckColumn(id string) error {
// Check cache to see if column already exists before querying pg
_, ok := r.columns.Get(id)
@@ -75,7 +76,7 @@ func (r *headerRepository) AddCheckColumn(id string) error {
return nil
}
// Adds a checked_header column for all of the provided column ids
// AddCheckColumns adds a checked_header column for all of the provided column ids
func (r *headerRepository) AddCheckColumns(ids []string) error {
var err error
baseQuery := "ALTER TABLE public.checked_headers"
@@ -99,7 +100,7 @@ func (r *headerRepository) AddCheckColumns(ids []string) error {
return err
}
// Marks the header checked for the provided column id
// MarkHeaderChecked marks the header checked for the provided column id
func (r *headerRepository) MarkHeaderChecked(headerID int64, id string) error {
_, err := r.db.Exec(`INSERT INTO public.checked_headers (header_id, `+id+`)
VALUES ($1, $2)
@@ -108,7 +109,7 @@ func (r *headerRepository) MarkHeaderChecked(headerID int64, id string) error {
return err
}
// Marks the header checked for all of the provided column ids
// MarkHeaderCheckedForAll marks the header checked for all of the provided column ids
func (r *headerRepository) MarkHeaderCheckedForAll(headerID int64, ids []string) error {
pgStr := "INSERT INTO public.checked_headers (header_id, "
for _, id := range ids {
@@ -127,7 +128,7 @@ func (r *headerRepository) MarkHeaderCheckedForAll(headerID int64, ids []string)
return err
}
// Marks all of the provided headers checked for each of the provided column ids
// MarkHeadersCheckedForAll marks all of the provided headers checked for each of the provided column ids
func (r *headerRepository) MarkHeadersCheckedForAll(headers []core.Header, ids []string) error {
tx, err := r.db.Beginx()
if err != nil {
@@ -160,7 +161,7 @@ func (r *headerRepository) MarkHeadersCheckedForAll(headers []core.Header, ids [
return err
}
// Returns missing headers for the provided checked_headers column id
// MissingHeaders returns missing headers for the provided checked_headers column id
func (r *headerRepository) MissingHeaders(startingBlockNumber, endingBlockNumber int64, id string) ([]core.Header, error) {
var result []core.Header
var query string
@@ -186,7 +187,7 @@ func (r *headerRepository) MissingHeaders(startingBlockNumber, endingBlockNumber
return continuousHeaders(result), err
}
// Returns missing headers for all of the provided checked_headers column ids
// MissingHeadersForAll returns missing headers for all of the provided checked_headers column ids
func (r *headerRepository) MissingHeadersForAll(startingBlockNumber, endingBlockNumber int64, ids []string) ([]core.Header, error) {
var result []core.Header
var query string
@@ -214,7 +215,7 @@ func (r *headerRepository) MissingHeadersForAll(startingBlockNumber, endingBlock
return continuousHeaders(result), err
}
// Returns headers that have been checked for all of the provided event ids but not for the provided method ids
// MissingMethodsCheckedEventsIntersection returns headers that have been checked for all of the provided event ids but not for the provided method ids
func (r *headerRepository) MissingMethodsCheckedEventsIntersection(startingBlockNumber, endingBlockNumber int64, methodIds, eventIds []string) ([]core.Header, error) {
var result []core.Header
var query string
@@ -264,16 +265,7 @@ func continuousHeaders(headers []core.Header) []core.Header {
return headers
}
// Check the repositories column id cache for a value
// CheckCache checks the repositories column id cache for a value
func (r *headerRepository) CheckCache(key string) (interface{}, bool) {
return r.columns.Get(key)
}
// Used to mark a header checked as part of some external transaction so as to group into one commit
func (r *headerRepository) MarkHeaderCheckedInTransaction(headerID int64, tx *sqlx.Tx, eventID string) error {
_, err := tx.Exec(`INSERT INTO public.checked_headers (header_id, `+eventID+`)
VALUES ($1, $2)
ON CONFLICT (header_id) DO
UPDATE SET `+eventID+` = checked_headers.`+eventID+` + 1`, headerID, 1)
return err
}
@@ -18,7 +18,6 @@ package repository_test
import (
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/core"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
@@ -26,6 +25,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/header/repository"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers/mocks"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
)
@@ -17,12 +17,12 @@
package repository_test
import (
"github.com/sirupsen/logrus"
"io/ioutil"
"testing"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/sirupsen/logrus"
)
func TestRepository(t *testing.T) {
@@ -20,7 +20,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
// Block retriever is used to retrieve the first block for a given contract and the most recent block
// BlockRetriever is used to retrieve the first block for a given contract and the most recent block
// It requires a vDB synced database with blocks, transactions, receipts, and logs
type BlockRetriever interface {
RetrieveFirstBlock() (int64, error)
@@ -31,13 +31,14 @@ type blockRetriever struct {
db *postgres.DB
}
func NewBlockRetriever(db *postgres.DB) (r *blockRetriever) {
// NewBlockRetriever returns a new BlockRetriever
func NewBlockRetriever(db *postgres.DB) BlockRetriever {
return &blockRetriever{
db: db,
}
}
// Retrieve block number of earliest header in repo
// RetrieveFirstBlock retrieves block number of earliest header in repo
func (r *blockRetriever) RetrieveFirstBlock() (int64, error) {
var firstBlock int
err := r.db.Get(
@@ -48,7 +49,7 @@ func (r *blockRetriever) RetrieveFirstBlock() (int64, error) {
return int64(firstBlock), err
}
// Retrieve block number of latest header in repo
// RetrieveMostRecentBlock retrieves block number of latest header in repo
func (r *blockRetriever) RetrieveMostRecentBlock() (int64, error) {
var lastBlock int
err := r.db.Get(
@@ -17,12 +17,12 @@
package retriever_test
import (
"github.com/sirupsen/logrus"
"io/ioutil"
"testing"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/sirupsen/logrus"
)
func TestRetriever(t *testing.T) {
@@ -17,6 +17,7 @@
package transformer
import (
"database/sql"
"errors"
"fmt"
"strings"
@@ -39,6 +40,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
// Transformer is the top level struct for transforming watched contract data
// Requires a header synced vDB (headers) and a running eth node (or infura)
type Transformer struct {
// Database interfaces
@@ -75,7 +77,7 @@ type Transformer struct {
// 3. Init
// 4. Execute
// Transformer takes in config for blockchain, database, and network id
// NewTransformer takes in a contract config, blockchain, and database, and returns a new Transformer
func NewTransformer(con config.ContractConfig, bc core.BlockChain, db *postgres.DB) *Transformer {
return &Transformer{
@@ -91,6 +93,7 @@ func NewTransformer(con config.ContractConfig, bc core.BlockChain, db *postgres.
}
}
// Init initialized the Transformer
// Use after creating and setting transformer
// Loops over all of the addr => filter sets
// Uses parser to pull event info from abi
@@ -124,7 +127,12 @@ func (tr *Transformer) Init() error {
// Get first block and most recent block number in the header repo
firstBlock, retrieveErr := tr.Retriever.RetrieveFirstBlock()
if retrieveErr != nil {
return fmt.Errorf("error retrieving first block: %s", retrieveErr.Error())
if retrieveErr == sql.ErrNoRows {
logrus.Error(fmt.Errorf("error retrieving first block: %s", retrieveErr.Error()))
firstBlock = 0
} else {
return fmt.Errorf("error retrieving first block: %s", retrieveErr.Error())
}
}
// Set to specified range if it falls within the bounds
@@ -170,14 +178,14 @@ func (tr *Transformer) Init() error {
// Create checked_headers columns for each event id and append to list of all event ids
tr.sortedEventIds[con.Address] = make([]string, 0, len(con.Events))
for _, event := range con.Events {
eventId := strings.ToLower(event.Name + "_" + con.Address)
addColumnErr := tr.HeaderRepository.AddCheckColumn(eventId)
eventID := strings.ToLower(event.Name + "_" + con.Address)
addColumnErr := tr.HeaderRepository.AddCheckColumn(eventID)
if addColumnErr != nil {
return fmt.Errorf("error adding check column: %s", addColumnErr.Error())
}
// Keep track of this event id; sorted and unsorted
tr.sortedEventIds[con.Address] = append(tr.sortedEventIds[con.Address], eventId)
tr.eventIds = append(tr.eventIds, eventId)
tr.sortedEventIds[con.Address] = append(tr.sortedEventIds[con.Address], eventID)
tr.eventIds = append(tr.eventIds, eventID)
// Append this event sig to the filters
tr.eventFilters = append(tr.eventFilters, event.Sig())
}
@@ -185,12 +193,12 @@ func (tr *Transformer) Init() error {
// Create checked_headers columns for each method id and append list of all method ids
tr.sortedMethodIds[con.Address] = make([]string, 0, len(con.Methods))
for _, m := range con.Methods {
methodId := strings.ToLower(m.Name + "_" + con.Address)
addColumnErr := tr.HeaderRepository.AddCheckColumn(methodId)
methodID := strings.ToLower(m.Name + "_" + con.Address)
addColumnErr := tr.HeaderRepository.AddCheckColumn(methodID)
if addColumnErr != nil {
return fmt.Errorf("error adding check column: %s", addColumnErr.Error())
}
tr.sortedMethodIds[con.Address] = append(tr.sortedMethodIds[con.Address], methodId)
tr.sortedMethodIds[con.Address] = append(tr.sortedMethodIds[con.Address], methodID)
}
// Update start to the lowest block
@@ -202,6 +210,7 @@ func (tr *Transformer) Init() error {
return nil
}
// Execute runs the transformation processes
func (tr *Transformer) Execute() error {
if len(tr.Contracts) == 0 {
return errors.New("error: transformer has no initialized contracts")
@@ -243,7 +252,6 @@ func (tr *Transformer) Execute() error {
continue
}
// Sort logs by the contract they belong to
for _, log := range allLogs {
addr := strings.ToLower(log.Address.Hex())
sortedLogs[addr] = append(sortedLogs[addr], log)
@@ -268,16 +276,10 @@ func (tr *Transformer) Execute() error {
for eventName, logs := range convertedLogs {
// If logs for this event are empty, mark them checked at this header and continue
if len(logs) < 1 {
eventId := strings.ToLower(eventName + "_" + con.Address)
markCheckedErr := tr.HeaderRepository.MarkHeaderChecked(header.Id, eventId)
if markCheckedErr != nil {
return fmt.Errorf("error marking header checked: %s", markCheckedErr.Error())
}
logrus.Tracef("no logs found for event %s on contract %s at block %d, continuing", eventName, conAddr, header.BlockNumber)
continue
}
// If logs aren't empty, persist them
// Header is marked checked in the transactions
persistErr := tr.EventRepository.PersistLogs(logs, con.Events[eventName], con.Address, con.Name)
if persistErr != nil {
return fmt.Errorf("error persisting logs: %s", persistErr.Error())
@@ -285,6 +287,11 @@ func (tr *Transformer) Execute() error {
}
}
markCheckedErr := tr.HeaderRepository.MarkHeaderCheckedForAll(header.Id, tr.eventIds)
if markCheckedErr != nil {
return fmt.Errorf("error marking header checked: %s", markCheckedErr.Error())
}
// Poll contracts at this block height
pollingErr := tr.methodPolling(header, tr.sortedMethodIds)
if pollingErr != nil {
@@ -323,6 +330,7 @@ func (tr *Transformer) methodPolling(header core.Header, sortedMethodIds map[str
return nil
}
// GetConfig returns the transformers config; satisfies the transformer interface
func (tr *Transformer) GetConfig() config.ContractConfig {
return tr.Config
}
@@ -17,12 +17,12 @@
package transformer_test
import (
"github.com/sirupsen/logrus"
"io/ioutil"
"testing"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/sirupsen/logrus"
)
func TestTransformer(t *testing.T) {
@@ -17,6 +17,8 @@
package transformer_test
import (
"database/sql"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
@@ -101,7 +103,17 @@ var _ = Describe("Transformer", func() {
Expect(c.Address).To(Equal(fakeAddress))
})
It("Fails to initialize if first block cannot be fetched from vDB headers table", func() {
It("uses first block from config if vDB headers table has no rows", func() {
blockRetriever := &fakes.MockHeaderSyncBlockRetriever{}
blockRetriever.FirstBlockErr = sql.ErrNoRows
t := getFakeTransformer(blockRetriever, &fakes.MockParser{}, &fakes.MockPoller{})
err := t.Init()
Expect(err).ToNot(HaveOccurred())
})
It("returns error if fetching first block fails for other reason", func() {
blockRetriever := &fakes.MockHeaderSyncBlockRetriever{}
blockRetriever.FirstBlockErr = fakes.FakeError
t := getFakeTransformer(blockRetriever, &fakes.MockParser{}, &fakes.MockPoller{})
File diff suppressed because one or more lines are too long
@@ -20,11 +20,10 @@ import (
"github.com/ethereum/go-ethereum/common/hexutil"
)
// Basic abi needed to check which interfaces are adhered to
// SupportsInterfaceABI is the basic abi needed to check which interfaces are adhered to
var SupportsInterfaceABI = `[{"constant":true,"inputs":[{"name":"interfaceID","type":"bytes4"}],"name":"supportsInterface","outputs":[{"name":"","type":"bool"}],"payable":false,"type":"function"}]`
// Individual event interfaces for constructing ABI from
var SupportsInterface = `{"constant":true,"inputs":[{"name":"interfaceID","type":"bytes4"}],"name":"supportsInterface","outputs":[{"name":"","type":"bool"}],"payable":false,"type":"function"}`
var AddrChangeInterface = `{"anonymous":false,"inputs":[{"indexed":true,"name":"node","type":"bytes32"},{"indexed":false,"name":"a","type":"address"}],"name":"AddrChanged","type":"event"}`
var ContentChangeInterface = `{"anonymous":false,"inputs":[{"indexed":true,"name":"node","type":"bytes32"},{"indexed":false,"name":"hash","type":"bytes32"}],"name":"ContentChanged","type":"event"}`
var NameChangeInterface = `{"anonymous":false,"inputs":[{"indexed":true,"name":"node","type":"bytes32"},{"indexed":false,"name":"name","type":"string"}],"name":"NameChanged","type":"event"}`
@@ -34,11 +33,10 @@ var TextChangeInterface = `{"anonymous":false,"inputs":[{"indexed":true,"name":"
var MultihashChangeInterface = `{"anonymous":false,"inputs":[{"indexed":true,"name":"node","type":"bytes32"},{"indexed":false,"name":"hash","type":"bytes"}],"name":"MultihashChanged","type":"event"}`
var ContenthashChangeInterface = `{"anonymous":false,"inputs":[{"indexed":true,"name":"node","type":"bytes32"},{"indexed":false,"name":"hash","type":"bytes"}],"name":"ContenthashChanged","type":"event"}`
var StartingBlock = int64(3648359)
// Resolver interface signatures
type Interface int
// Interface enums
const (
MetaSig Interface = iota
AddrChangeSig
@@ -51,6 +49,7 @@ const (
ContentHashChangeSig
)
// Hex returns the hex signature for an interface
func (e Interface) Hex() string {
strings := [...]string{
"0x01ffc9a7",
@@ -71,6 +70,7 @@ func (e Interface) Hex() string {
return strings[e]
}
// Bytes returns the bytes signature for an interface
func (e Interface) Bytes() [4]uint8 {
if e < MetaSig || e > ContentHashChangeSig {
return [4]byte{}
@@ -86,6 +86,7 @@ func (e Interface) Bytes() [4]uint8 {
return byArray
}
// EventSig returns the event signature for an interface
func (e Interface) EventSig() string {
strings := [...]string{
"",
@@ -106,6 +107,7 @@ func (e Interface) EventSig() string {
return strings[e]
}
// MethodSig returns the method signature for an interface
func (e Interface) MethodSig() string {
strings := [...]string{
"supportsInterface(bytes4)",
@@ -48,6 +48,7 @@ type Contract struct {
Piping bool // Whether or not to pipe method results forward as arguments to subsequent methods
}
// Init initializes a contract object
// If we will be calling methods that use addr, hash, or byte arrays
// as arguments then we initialize maps to hold these types of values
func (c Contract) Init() *Contract {
@@ -66,7 +67,7 @@ func (c Contract) Init() *Contract {
return &c
}
// Use contract info to generate event filters - full sync contract watcher only
// GenerateFilters uses contract info to generate event filters - full sync contract watcher only
func (c *Contract) GenerateFilters() error {
c.Filters = map[string]filters.LogFilter{}
@@ -87,7 +88,7 @@ func (c *Contract) GenerateFilters() error {
return nil
}
// Returns true if address is in list of arguments to
// WantedEventArg returns true if address is in list of arguments to
// filter events for or if no filtering is specified
func (c *Contract) WantedEventArg(arg string) bool {
if c.FilterArgs == nil {
@@ -101,7 +102,7 @@ func (c *Contract) WantedEventArg(arg string) bool {
return false
}
// Returns true if address is in list of arguments to
// WantedMethodArg returns true if address is in list of arguments to
// poll methods with or if no filtering is specified
func (c *Contract) WantedMethodArg(arg interface{}) bool {
if c.MethodArgs == nil {
@@ -121,7 +122,7 @@ func (c *Contract) WantedMethodArg(arg interface{}) bool {
return false
}
// Returns true if any mapping value matches filtered for address or if no filter exists
// PassesEventFilter returns true if any mapping value matches filtered for address or if no filter exists
// Used to check if an event log name-value mapping should be filtered or not
func (c *Contract) PassesEventFilter(args map[string]string) bool {
for _, arg := range args {
@@ -133,7 +134,7 @@ func (c *Contract) PassesEventFilter(args map[string]string) bool {
return false
}
// Add event emitted address to our list if it passes filter and method polling is on
// AddEmittedAddr adds event emitted addresses to our list if it passes filter and method polling is on
func (c *Contract) AddEmittedAddr(addresses ...interface{}) {
for _, addr := range addresses {
if c.WantedMethodArg(addr) && c.Methods != nil {
@@ -142,7 +143,7 @@ func (c *Contract) AddEmittedAddr(addresses ...interface{}) {
}
}
// Add event emitted hash to our list if it passes filter and method polling is on
// AddEmittedHash adds event emitted hashes to our list if it passes filter and method polling is on
func (c *Contract) AddEmittedHash(hashes ...interface{}) {
for _, hash := range hashes {
if c.WantedMethodArg(hash) && c.Methods != nil {
@@ -151,6 +152,7 @@ func (c *Contract) AddEmittedHash(hashes ...interface{}) {
}
}
// StringifyArg resolves a method argument type to string type
func StringifyArg(arg interface{}) (str string) {
switch arg.(type) {
case string:
@@ -29,7 +29,7 @@ import (
// Fetcher serves as the lower level data fetcher that calls the underlying
// blockchain's FetchConctractData method for a given return type
// Interface definition for a Fetcher
// FetcherInterface is the interface definition for a fetcher
type FetcherInterface interface {
FetchBigInt(method, contractAbi, contractAddress string, blockNumber int64, methodArgs []interface{}) (big.Int, error)
FetchBool(method, contractAbi, contractAddress string, blockNumber int64, methodArgs []interface{}) (bool, error)
@@ -56,14 +56,14 @@ type fetcherError struct {
fetchMethod string
}
// Fetcher error method
// Error method
func (fe *fetcherError) Error() string {
return fmt.Sprintf("Error fetching %s: %s", fe.fetchMethod, fe.err)
}
// Generic Fetcher methods used by Getters to call contract methods
// Method used to fetch big.Int value from contract
// FetchBigInt is the method used to fetch big.Int value from contract
func (f Fetcher) FetchBigInt(method, contractAbi, contractAddress string, blockNumber int64, methodArgs []interface{}) (big.Int, error) {
var result = new(big.Int)
err := f.BlockChain.FetchContractData(contractAbi, contractAddress, method, methodArgs, &result, blockNumber)
@@ -75,7 +75,7 @@ func (f Fetcher) FetchBigInt(method, contractAbi, contractAddress string, blockN
return *result, nil
}
// Method used to fetch bool value from contract
// FetchBool is the method used to fetch bool value from contract
func (f Fetcher) FetchBool(method, contractAbi, contractAddress string, blockNumber int64, methodArgs []interface{}) (bool, error) {
var result = new(bool)
err := f.BlockChain.FetchContractData(contractAbi, contractAddress, method, methodArgs, &result, blockNumber)
@@ -87,7 +87,7 @@ func (f Fetcher) FetchBool(method, contractAbi, contractAddress string, blockNum
return *result, nil
}
// Method used to fetch address value from contract
// FetchAddress is the method used to fetch address value from contract
func (f Fetcher) FetchAddress(method, contractAbi, contractAddress string, blockNumber int64, methodArgs []interface{}) (common.Address, error) {
var result = new(common.Address)
err := f.BlockChain.FetchContractData(contractAbi, contractAddress, method, methodArgs, &result, blockNumber)
@@ -99,7 +99,7 @@ func (f Fetcher) FetchAddress(method, contractAbi, contractAddress string, block
return *result, nil
}
// Method used to fetch string value from contract
// FetchString is the method used to fetch string value from contract
func (f Fetcher) FetchString(method, contractAbi, contractAddress string, blockNumber int64, methodArgs []interface{}) (string, error) {
var result = new(string)
err := f.BlockChain.FetchContractData(contractAbi, contractAddress, method, methodArgs, &result, blockNumber)
@@ -111,7 +111,7 @@ func (f Fetcher) FetchString(method, contractAbi, contractAddress string, blockN
return *result, nil
}
// Method used to fetch hash value from contract
// FetchHash is the method used to fetch hash value from contract
func (f Fetcher) FetchHash(method, contractAbi, contractAddress string, blockNumber int64, methodArgs []interface{}) (common.Hash, error) {
var result = new(common.Hash)
err := f.BlockChain.FetchContractData(contractAbi, contractAddress, method, methodArgs, &result, blockNumber)
@@ -24,10 +24,10 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/constants"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/getter"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/geth/client"
rpc2 "github.com/vulcanize/vulcanizedb/pkg/geth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/geth/node"
"github.com/vulcanize/vulcanizedb/pkg/eth"
"github.com/vulcanize/vulcanizedb/pkg/eth/client"
rpc2 "github.com/vulcanize/vulcanizedb/pkg/eth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/eth/node"
"github.com/vulcanize/vulcanizedb/test_config"
)
@@ -45,11 +45,12 @@ var _ = Describe("Interface Getter", func() {
blockChainClient := client.NewEthClient(ethClient)
node := node.MakeNode(rpcClient)
transactionConverter := rpc2.NewRpcTransactionConverter(ethClient)
blockChain := geth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
blockChain := eth.NewBlockChain(blockChainClient, rpcClient, node, transactionConverter)
interfaceGetter := getter.NewInterfaceGetter(blockChain)
abi := interfaceGetter.GetABI(constants.PublicResolverAddress, blockNumber)
abi, err := interfaceGetter.GetABI(constants.PublicResolverAddress, blockNumber)
Expect(err).NotTo(HaveOccurred())
Expect(abi).To(Equal(expectedABI))
_, err = geth.ParseAbi(abi)
_, err = eth.ParseAbi(abi)
Expect(err).ToNot(HaveOccurred())
})
})
@@ -17,13 +17,16 @@
package getter
import (
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/constants"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/fetcher"
"github.com/vulcanize/vulcanizedb/pkg/core"
)
// InterfaceGetter is used to derive the interface of a contract
type InterfaceGetter interface {
GetABI(resolverAddr string, blockNumber int64) string
GetABI(resolverAddr string, blockNumber int64) (string, error)
GetBlockChain() core.BlockChain
}
@@ -31,7 +34,8 @@ type interfaceGetter struct {
fetcher.Fetcher
}
func NewInterfaceGetter(blockChain core.BlockChain) *interfaceGetter {
// NewInterfaceGetter returns a new InterfaceGetter
func NewInterfaceGetter(blockChain core.BlockChain) InterfaceGetter {
return &interfaceGetter{
Fetcher: fetcher.Fetcher{
BlockChain: blockChain,
@@ -39,15 +43,19 @@ func NewInterfaceGetter(blockChain core.BlockChain) *interfaceGetter {
}
}
// Used to construct a custom ABI based on the results from calling supportsInterface
func (g *interfaceGetter) GetABI(resolverAddr string, blockNumber int64) string {
// GetABI is used to construct a custom ABI based on the results from calling supportsInterface
func (g *interfaceGetter) GetABI(resolverAddr string, blockNumber int64) (string, error) {
a := constants.SupportsInterfaceABI
args := make([]interface{}, 1)
args[0] = constants.MetaSig.Bytes()
supports, err := g.getSupportsInterface(a, resolverAddr, blockNumber, args)
if err != nil || !supports {
return ""
if err != nil {
return "", fmt.Errorf("call to getSupportsInterface failed: %v", err)
}
if !supports {
return "", fmt.Errorf("contract does not support interface")
}
abiStr := `[`
args[0] = constants.AddrChangeSig.Bytes()
supports, err = g.getSupportsInterface(a, resolverAddr, blockNumber, args)
@@ -91,7 +99,7 @@ func (g *interfaceGetter) GetABI(resolverAddr string, blockNumber int64) string
}
abiStr = abiStr[:len(abiStr)-1] + `]`
return abiStr
return abiStr, nil
}
// Use this method to check whether or not a contract supports a given method/event interface
@@ -99,7 +107,7 @@ func (g *interfaceGetter) getSupportsInterface(contractAbi, contractAddress stri
return g.Fetcher.FetchBool("supportsInterface", contractAbi, contractAddress, blockNumber, methodArgs)
}
// Method to retrieve the Getter's blockchain
// GetBlockChain is a method to retrieve the Getter's blockchain
func (g *interfaceGetter) GetBlockChain() core.BlockChain {
return g.Fetcher.BlockChain
}
@@ -27,6 +27,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/core"
)
// ConvertToLog converts a watched event to a log
func ConvertToLog(watchedEvent core.WatchedEvent) types.Log {
allTopics := []string{watchedEvent.Topic0, watchedEvent.Topic1, watchedEvent.Topic2, watchedEvent.Topic3}
var nonNilTopics []string
@@ -56,12 +57,14 @@ func createTopics(topics ...string) []common.Hash {
return topicsArray
}
// BigFromString creates a big.Int from a string
func BigFromString(n string) *big.Int {
b := new(big.Int)
b.SetString(n, 10)
return b
}
// GenerateSignature returns the keccak256 hash hex of a string
func GenerateSignature(s string) string {
eventSignature := []byte(s)
hash := crypto.Keccak256Hash(eventSignature)
@@ -30,10 +30,10 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/geth/client"
rpc2 "github.com/vulcanize/vulcanizedb/pkg/geth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/geth/node"
"github.com/vulcanize/vulcanizedb/pkg/eth"
"github.com/vulcanize/vulcanizedb/pkg/eth/client"
rpc2 "github.com/vulcanize/vulcanizedb/pkg/eth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/eth/node"
"github.com/vulcanize/vulcanizedb/test_config"
)
@@ -117,7 +117,7 @@ func SetupDBandBC() (*postgres.DB, core.BlockChain) {
blockChainClient := client.NewEthClient(ethClient)
madeNode := node.MakeNode(rpcClient)
transactionConverter := rpc2.NewRpcTransactionConverter(ethClient)
blockChain := geth.NewBlockChain(blockChainClient, rpcClient, madeNode, transactionConverter)
blockChain := eth.NewBlockChain(blockChainClient, rpcClient, madeNode, transactionConverter)
db, err := postgres.NewDB(config.Database{
Hostname: "localhost",
@@ -294,8 +294,7 @@ func TearDown(db *postgres.DB) {
_, err = tx.Exec(`CREATE TABLE checked_headers (
id SERIAL PRIMARY KEY,
header_id INTEGER UNIQUE NOT NULL REFERENCES headers (id) ON DELETE CASCADE,
check_count INTEGER NOT NULL DEFAULT 1);`)
header_id INTEGER UNIQUE NOT NULL REFERENCES headers (id) ON DELETE CASCADE);`)
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DROP SCHEMA IF EXISTS full_0x8dd5fbce2f6a956c3022ba3663759011dd51e73e CASCADE`)
@@ -20,7 +20,7 @@ import (
"github.com/ethereum/go-ethereum/accounts/abi"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/eth"
)
// Mock parser
@@ -50,7 +50,7 @@ func (p *parser) ParsedAbi() abi.ABI {
// for the given contract address
func (p *parser) Parse() error {
var err error
p.parsedAbi, err = geth.ParseAbi(p.abi)
p.parsedAbi, err = eth.ParseAbi(p.abi)
return err
}
+17 -14
View File
@@ -24,7 +24,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/constants"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/eth"
)
// Parser is used to fetch and parse contract ABIs
@@ -40,28 +40,31 @@ type Parser interface {
}
type parser struct {
client *geth.EtherScanAPI
client *eth.EtherScanAPI
abi string
parsedAbi abi.ABI
}
func NewParser(network string) *parser {
url := geth.GenURL(network)
// NewParser returns a new Parser
func NewParser(network string) Parser {
url := eth.GenURL(network)
return &parser{
client: geth.NewEtherScanClient(url),
client: eth.NewEtherScanClient(url),
}
}
// Abi returns the parser's configured abi string
func (p *parser) Abi() string {
return p.abi
}
// ParsedAbi returns the parser's parsed abi
func (p *parser) ParsedAbi() abi.ABI {
return p.parsedAbi
}
// Retrieves and parses the abi string
// Parse retrieves and parses the abi string
// for the given contract address
func (p *parser) Parse(contractAddr string) error {
// If the abi is one our locally stored abis, fetch
@@ -69,7 +72,7 @@ func (p *parser) Parse(contractAddr string) error {
knownAbi, err := p.lookUp(contractAddr)
if err == nil {
p.abi = knownAbi
p.parsedAbi, err = geth.ParseAbi(knownAbi)
p.parsedAbi, err = eth.ParseAbi(knownAbi)
return err
}
// Try getting abi from etherscan
@@ -79,29 +82,29 @@ func (p *parser) Parse(contractAddr string) error {
}
//TODO: Implement other ways to fetch abi
p.abi = abiStr
p.parsedAbi, err = geth.ParseAbi(abiStr)
p.parsedAbi, err = eth.ParseAbi(abiStr)
return err
}
// Loads and parses an abi from a given abi string
// ParseAbiStr loads and parses an abi from a given abi string
func (p *parser) ParseAbiStr(abiStr string) error {
var err error
p.abi = abiStr
p.parsedAbi, err = geth.ParseAbi(abiStr)
p.parsedAbi, err = eth.ParseAbi(abiStr)
return err
}
func (p *parser) lookUp(contractAddr string) (string, error) {
if v, ok := constants.Abis[common.HexToAddress(contractAddr)]; ok {
if v, ok := constants.ABIs[common.HexToAddress(contractAddr)]; ok {
return v, nil
}
return "", errors.New("ABI not present in lookup table")
}
// Returns only specified methods, if they meet the criteria
// GetSelectMethods returns only specified methods, if they meet the criteria
// Returns as array with methods in same order they were specified
// Nil or empty wanted array => no events are returned
func (p *parser) GetSelectMethods(wanted []string) []types.Method {
@@ -121,7 +124,7 @@ func (p *parser) GetSelectMethods(wanted []string) []types.Method {
return methods
}
// Returns wanted methods
// GetMethods returns wanted methods
// Empty wanted array => all methods are returned
// Nil wanted array => no methods are returned
func (p *parser) GetMethods(wanted []string) []types.Method {
@@ -139,7 +142,7 @@ func (p *parser) GetMethods(wanted []string) []types.Method {
return methods
}
// Returns wanted events as map of types.Events
// GetEvents returns wanted events as map of types.Events
// Empty wanted array => all events are returned
// Nil wanted array => no events are returned
func (p *parser) GetEvents(wanted []string) map[string]types.Event {
@@ -25,7 +25,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers/mocks"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/parser"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/eth"
)
var _ = Describe("Parser", func() {
@@ -44,7 +44,7 @@ var _ = Describe("Parser", func() {
Expect(err).ToNot(HaveOccurred())
parsedAbi := mp.ParsedAbi()
expectedAbi, err := geth.ParseAbi(constants.DaiAbiString)
expectedAbi, err := eth.ParseAbi(constants.DaiAbiString)
Expect(err).ToNot(HaveOccurred())
Expect(parsedAbi).To(Equal(expectedAbi))
@@ -73,7 +73,7 @@ var _ = Describe("Parser", func() {
expectedAbi := constants.DaiAbiString
Expect(p.Abi()).To(Equal(expectedAbi))
expectedParsedAbi, err := geth.ParseAbi(expectedAbi)
expectedParsedAbi, err := eth.ParseAbi(expectedAbi)
Expect(err).ToNot(HaveOccurred())
Expect(p.ParsedAbi()).To(Equal(expectedParsedAbi))
})
+12 -8
View File
@@ -33,6 +33,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
// Poller is the interface for polling public contract methods
type Poller interface {
PollContract(con contract.Contract, lastBlock int64) error
PollContractAt(con contract.Contract, blockNumber int64) error
@@ -45,13 +46,15 @@ type poller struct {
contract contract.Contract
}
func NewPoller(blockChain core.BlockChain, db *postgres.DB, mode types.Mode) *poller {
// NewPoller returns a new Poller
func NewPoller(blockChain core.BlockChain, db *postgres.DB, mode types.Mode) Poller {
return &poller{
MethodRepository: repository.NewMethodRepository(db, mode),
bc: blockChain,
}
}
// PollContract polls a contract's public methods from the contracts starting block to specified last block
func (p *poller) PollContract(con contract.Contract, lastBlock int64) error {
for i := con.StartingBlock; i <= lastBlock; i++ {
if err := p.PollContractAt(con, i); err != nil {
@@ -62,6 +65,7 @@ func (p *poller) PollContract(con contract.Contract, lastBlock int64) error {
return nil
}
// PollContractAt polls a contract's public getter methods at the specified block height
func (p *poller) PollContractAt(con contract.Contract, blockNumber int64) error {
p.contract = con
for _, m := range con.Methods {
@@ -98,7 +102,7 @@ func (p *poller) pollNoArgAt(m types.Method, bn int64) error {
var out interface{}
err := p.bc.FetchContractData(p.contract.Abi, p.contract.Address, m.Name, nil, &out, bn)
if err != nil {
return errors.New(fmt.Sprintf("poller error calling 0 argument method\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err))
return fmt.Errorf("poller error calling 0 argument method\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err)
}
strOut, err := stringify(out)
if err != nil {
@@ -112,7 +116,7 @@ func (p *poller) pollNoArgAt(m types.Method, bn int64) error {
// Persist result immediately
err = p.PersistResults([]types.Result{result}, m, p.contract.Address, p.contract.Name)
if err != nil {
return errors.New(fmt.Sprintf("poller error persisting 0 argument method result\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err))
return fmt.Errorf("poller error persisting 0 argument method result\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err)
}
return nil
@@ -148,7 +152,7 @@ func (p *poller) pollSingleArgAt(m types.Method, bn int64) error {
var out interface{}
err := p.bc.FetchContractData(p.contract.Abi, p.contract.Address, m.Name, in, &out, bn)
if err != nil {
return errors.New(fmt.Sprintf("poller error calling 1 argument method\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err))
return fmt.Errorf("poller error calling 1 argument method\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err)
}
strOut, err := stringify(out)
if err != nil {
@@ -164,7 +168,7 @@ func (p *poller) pollSingleArgAt(m types.Method, bn int64) error {
// Persist result set as batch
err := p.PersistResults(results, m, p.contract.Address, p.contract.Name)
if err != nil {
return errors.New(fmt.Sprintf("poller error persisting 1 argument method result\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err))
return fmt.Errorf("poller error persisting 1 argument method result\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err)
}
return nil
@@ -212,7 +216,7 @@ func (p *poller) pollDoubleArgAt(m types.Method, bn int64) error {
var out interface{}
err := p.bc.FetchContractData(p.contract.Abi, p.contract.Address, m.Name, in, &out, bn)
if err != nil {
return errors.New(fmt.Sprintf("poller error calling 2 argument method\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err))
return fmt.Errorf("poller error calling 2 argument method\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err)
}
strOut, err := stringify(out)
if err != nil {
@@ -228,13 +232,13 @@ func (p *poller) pollDoubleArgAt(m types.Method, bn int64) error {
err := p.PersistResults(results, m, p.contract.Address, p.contract.Name)
if err != nil {
return errors.New(fmt.Sprintf("poller error persisting 2 argument method result\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err))
return fmt.Errorf("poller error persisting 2 argument method result\r\nblock: %d, method: %s, contract: %s\r\nerr: %v", bn, m.Name, p.contract.Address, err)
}
return nil
}
// This is just a wrapper around the poller blockchain's FetchContractData method
// FetchContractData is just a wrapper around the poller blockchain's FetchContractData method
func (p *poller) FetchContractData(contractAbi, contractAddress, method string, methodArgs []interface{}, result interface{}, blockNumber int64) error {
return p.bc.FetchContractData(contractAbi, contractAddress, method, methodArgs, result, blockNumber)
}
@@ -24,7 +24,6 @@ import (
"github.com/hashicorp/golang-lru"
"github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/libraries/shared/repository"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
@@ -35,7 +34,7 @@ const (
eventCacheSize = 1000
)
// Event repository is used to persist event data into custom tables
// EventRepository is used to persist event data into custom tables
type EventRepository interface {
PersistLogs(logs []types.Log, eventInfo types.Event, contractAddr, contractName string) error
CreateEventTable(contractAddr string, event types.Event) (bool, error)
@@ -51,7 +50,8 @@ type eventRepository struct {
tables *lru.Cache // Cache names of recently used tables to minimize db connections
}
func NewEventRepository(db *postgres.DB, mode types.Mode) *eventRepository {
// NewEventRepository returns a new EventRepository
func NewEventRepository(db *postgres.DB, mode types.Mode) EventRepository {
ccs, _ := lru.New(contractCacheSize)
ecs, _ := lru.New(eventCacheSize)
return &eventRepository{
@@ -62,7 +62,7 @@ func NewEventRepository(db *postgres.DB, mode types.Mode) *eventRepository {
}
}
// Creates a schema for the contract if needed
// PersistLogs creates a schema for the contract if needed
// Creates table for the watched contract event if needed
// Persists converted event log data into this custom table
func (r *eventRepository) PersistLogs(logs []types.Log, eventInfo types.Event, contractAddr, contractName string) error {
@@ -112,7 +112,7 @@ func (r *eventRepository) persistHeaderSyncLogs(logs []types.Log, eventInfo type
// Preallocate slice of needed capacity and proceed to pack variables into it in same order they appear in string
data := make([]interface{}, 0, 5+el)
data = append(data,
event.Id,
event.ID,
contractName,
event.Raw,
event.LogIndex,
@@ -143,17 +143,6 @@ func (r *eventRepository) persistHeaderSyncLogs(logs []types.Log, eventInfo type
}
}
// Mark header as checked for this eventId
eventId := strings.ToLower(eventInfo.Name + "_" + contractAddr)
markCheckedErr := repository.MarkContractWatcherHeaderCheckedInTransaction(logs[0].Id, tx, eventId) // This assumes all logs are from same block
if markCheckedErr != nil {
rollbackErr := tx.Rollback()
if rollbackErr != nil {
logrus.Warnf("error rolling back transaction while marking header checked: %s", rollbackErr.Error())
}
return fmt.Errorf("error marking header checked: %s", markCheckedErr.Error())
}
return tx.Commit()
}
@@ -171,7 +160,7 @@ func (r *eventRepository) persistFullSyncLogs(logs []types.Log, eventInfo types.
data := make([]interface{}, 0, 4+el)
data = append(data,
event.Id,
event.ID,
contractName,
event.Block,
event.Tx)
@@ -201,7 +190,7 @@ func (r *eventRepository) persistFullSyncLogs(logs []types.Log, eventInfo types.
return tx.Commit()
}
// Checks for event table and creates it if it does not already exist
// CreateEventTable checks for event table and creates it if it does not already exist
// Returns true if it created a new table; returns false if table already existed
func (r *eventRepository) CreateEventTable(contractAddr string, event types.Event) (bool, error) {
tableID := fmt.Sprintf("%s_%s.%s_event", r.mode.String(), strings.ToLower(contractAddr), strings.ToLower(event.Name))
@@ -270,7 +259,7 @@ func (r *eventRepository) checkForTable(contractAddr string, eventName string) (
return exists, err
}
// Checks for contract schema and creates it if it does not already exist
// CreateContractSchema checks for contract schema and creates it if it does not already exist
// Returns true if it created a new schema; returns false if schema already existed
func (r *eventRepository) CreateContractSchema(contractAddr string) (bool, error) {
if contractAddr == "" {
@@ -316,10 +305,12 @@ func (r *eventRepository) checkForSchema(contractAddr string) (bool, error) {
return exists, err
}
// CheckSchemaCache is used to query the schema name cache
func (r *eventRepository) CheckSchemaCache(key string) (interface{}, bool) {
return r.schemas.Get(key)
}
// CheckTableCache is used to query the table name cache
func (r *eventRepository) CheckTableCache(key string) (interface{}, bool) {
return r.tables.Get(key)
}
@@ -355,11 +355,6 @@ var _ = Describe("Repository", func() {
Expect(count).To(Equal(2))
})
It("Fails if the persisted event does not have a corresponding eventID column in the checked_headers table", func() {
err = dataStore.PersistLogs(logs, event, con.Address, con.Name)
Expect(err).To(HaveOccurred())
})
It("Fails with empty log", func() {
err = dataStore.PersistLogs([]types.Log{}, event, con.Address, con.Name)
Expect(err).To(HaveOccurred())
@@ -30,6 +30,7 @@ import (
const methodCacheSize = 1000
// MethodRepository is used to persist public getter method data
type MethodRepository interface {
PersistResults(results []types.Result, methodInfo types.Method, contractAddr, contractName string) error
CreateMethodTable(contractAddr string, method types.Method) (bool, error)
@@ -45,7 +46,8 @@ type methodRepository struct {
tables *lru.Cache // Cache names of recently used tables to minimize db connections
}
func NewMethodRepository(db *postgres.DB, mode types.Mode) *methodRepository {
// NewMethodRepository returns a new MethodRepository
func NewMethodRepository(db *postgres.DB, mode types.Mode) MethodRepository {
ccs, _ := lru.New(contractCacheSize)
mcs, _ := lru.New(methodCacheSize)
return &methodRepository{
@@ -56,7 +58,7 @@ func NewMethodRepository(db *postgres.DB, mode types.Mode) *methodRepository {
}
}
// Creates a schema for the contract if needed
// PersistResults creates a schema for the contract if needed
// Creates table for the contract method if needed
// Persists method polling data into this custom table
func (r *methodRepository) PersistResults(results []types.Result, methodInfo types.Method, contractAddr, contractName string) error {
@@ -124,7 +126,7 @@ func (r *methodRepository) persistResults(results []types.Result, methodInfo typ
return tx.Commit()
}
// Checks for event table and creates it if it does not already exist
// CreateMethodTable checks for event table and creates it if it does not already exist
func (r *methodRepository) CreateMethodTable(contractAddr string, method types.Method) (bool, error) {
tableID := fmt.Sprintf("%s_%s.%s_method", r.mode.String(), strings.ToLower(contractAddr), strings.ToLower(method.Name))
@@ -177,7 +179,7 @@ func (r *methodRepository) checkForTable(contractAddr string, methodName string)
return exists, err
}
// Checks for contract schema and creates it if it does not already exist
// CreateContractSchema checks for contract schema and creates it if it does not already exist
func (r *methodRepository) CreateContractSchema(contractAddr string) (bool, error) {
if contractAddr == "" {
return false, errors.New("error: no contract address specified")
@@ -222,10 +224,12 @@ func (r *methodRepository) checkForSchema(contractAddr string) (bool, error) {
return exists, err
}
// CheckSchemaCache is used to query the schema name cache
func (r *methodRepository) CheckSchemaCache(key string) (interface{}, bool) {
return r.schemas.Get(key)
}
// CheckTableCache is used to query the table name cache
func (r *methodRepository) CheckTableCache(key string) (interface{}, bool) {
return r.tables.Get(key)
}
@@ -17,12 +17,12 @@
package repository_test
import (
"github.com/sirupsen/logrus"
"io/ioutil"
"testing"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/sirupsen/logrus"
)
func TestRepository(t *testing.T) {
@@ -18,17 +18,17 @@ package retriever
import (
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types"
"strings"
"github.com/ethereum/go-ethereum/accounts/abi"
"github.com/ethereum/go-ethereum/common"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/contract"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
// Address retriever is used to retrieve the addresses associated with a contract
// AddressRetriever is used to retrieve the addresses associated with a contract
type AddressRetriever interface {
RetrieveTokenHolderAddresses(info contract.Contract) (map[common.Address]bool, error)
}
@@ -38,14 +38,15 @@ type addressRetriever struct {
mode types.Mode
}
func NewAddressRetriever(db *postgres.DB, mode types.Mode) (r *addressRetriever) {
// NewAddressRetriever returns a new AddressRetriever
func NewAddressRetriever(db *postgres.DB, mode types.Mode) AddressRetriever {
return &addressRetriever{
db: db,
mode: mode,
}
}
// Method to retrieve list of token-holding/contract-related addresses by iterating over available events
// RetrieveTokenHolderAddresses is used to retrieve list of token-holding/contract-related addresses by iterating over available events
// This generic method should work whether or not the argument/input names of the events meet the expected standard
// This could be generalized to iterate over ALL events and pull out any address arguments
func (r *addressRetriever) RetrieveTokenHolderAddresses(info contract.Contract) (map[common.Address]bool, error) {
@@ -17,12 +17,12 @@
package retriever_test
import (
"github.com/sirupsen/logrus"
"io/ioutil"
"testing"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/sirupsen/logrus"
)
func TestRetriever(t *testing.T) {
+6 -3
View File
@@ -25,20 +25,22 @@ import (
"github.com/ethereum/go-ethereum/crypto"
)
// Event is our custom event type
type Event struct {
Name string
Anonymous bool
Fields []Field
}
// Field is our custom event field type which associates a postgres type with the field
type Field struct {
abi.Argument // Name, Type, Indexed
PgType string // Holds type used when committing data held in this field to postgres
}
// Struct to hold instance of an event log data
// Log is used to hold instance of an event log data
type Log struct {
Id int64 // VulcanizeIdLog for full sync and header ID for header sync contract watcher
ID int64 // VulcanizeIdLog for full sync and header ID for header sync contract watcher
Values map[string]string // Map of event input names to their values
// Used for full sync only
@@ -51,7 +53,7 @@ type Log struct {
Raw []byte // json.Unmarshalled byte array of geth/core/types.Log{}
}
// Unpack abi.Event into our custom Event struct
// NewEvent unpacks abi.Event into our custom Event struct
func NewEvent(e abi.Event) Event {
fields := make([]Field, len(e.Inputs))
for i, input := range e.Inputs {
@@ -85,6 +87,7 @@ func NewEvent(e abi.Event) Event {
}
}
// Sig returns the hash signature for an event
func (e Event) Sig() common.Hash {
types := make([]string, len(e.Fields))
+4 -2
View File
@@ -25,6 +25,7 @@ import (
"github.com/ethereum/go-ethereum/crypto"
)
// Method is our custom method struct
type Method struct {
Name string
Const bool
@@ -32,7 +33,7 @@ type Method struct {
Return []Field
}
// Struct to hold instance of result from method call with given inputs and block
// Result is used to hold instance of result from method call with given inputs and block
type Result struct {
Method
Inputs []interface{} // Will only use addresses
@@ -41,7 +42,7 @@ type Result struct {
Block int64
}
// Unpack abi.Method into our custom Method struct
// NewMethod unpacks abi.Method into our custom Method struct
func NewMethod(m abi.Method) Method {
inputs := make([]Field, len(m.Inputs))
for i, input := range m.Inputs {
@@ -99,6 +100,7 @@ func NewMethod(m abi.Method) Method {
}
}
// Sig returns the hash signature for the method
func (m Method) Sig() common.Hash {
types := make([]string, len(m.Args))
i := 0
+4 -25
View File
@@ -16,19 +16,21 @@
package types
import "fmt"
// Mode is used to explicitly represent the operating mode of the transformer
type Mode int
// Mode enums
const (
HeaderSync Mode = iota
FullSync
)
// IsValid returns true is the Mode is valid
func (mode Mode) IsValid() bool {
return mode >= HeaderSync && mode <= FullSync
}
// String returns the string representation of the mode
func (mode Mode) String() string {
switch mode {
case HeaderSync:
@@ -39,26 +41,3 @@ func (mode Mode) String() string {
return "unknown"
}
}
func (mode Mode) MarshalText() ([]byte, error) {
switch mode {
case HeaderSync:
return []byte("header"), nil
case FullSync:
return []byte("full"), nil
default:
return nil, fmt.Errorf("contract watcher: unknown mode %d, want HeaderSync or FullSync", mode)
}
}
func (mode *Mode) UnmarshalText(text []byte) error {
switch string(text) {
case "header":
*mode = HeaderSync
case "full":
*mode = FullSync
default:
return fmt.Errorf(`contract watcher: unknown mode %q, want "header" or "full"`, text)
}
return nil
}
+1 -1
View File
@@ -20,7 +20,7 @@ import (
"context"
"github.com/ethereum/go-ethereum/rpc"
"github.com/vulcanize/vulcanizedb/pkg/geth/client"
"github.com/vulcanize/vulcanizedb/pkg/eth/client"
)
type RpcClient interface {
@@ -41,7 +41,7 @@ func (repository HeaderRepository) CreateOrUpdateHeader(header core.Header) (int
hash, err := repository.getHeaderHash(header)
if err != nil {
if headerDoesNotExist(err) {
return repository.insertHeader(header)
return repository.InternalInsertHeader(header)
}
log.Error("CreateOrUpdateHeader: error getting header hash: ", err)
return 0, err
@@ -128,13 +128,21 @@ func (repository HeaderRepository) getHeaderHash(header core.Header) (string, er
return hash, err
}
func (repository HeaderRepository) insertHeader(header core.Header) (int64, error) {
// Function is public so we can test insert being called for the same header
// Can happen when concurrent processes are inserting headers
// Otherwise should not occur since only called in CreateOrUpdateHeader
func (repository HeaderRepository) InternalInsertHeader(header core.Header) (int64, error) {
var headerId int64
err := repository.database.QueryRowx(
`INSERT INTO public.headers (block_number, hash, block_timestamp, raw, eth_node_id, eth_node_fingerprint) VALUES ($1, $2, $3::NUMERIC, $4, $5, $6) RETURNING id`,
header.BlockNumber, header.Hash, header.Timestamp, header.Raw, repository.database.NodeID, repository.database.Node.ID).Scan(&headerId)
row := repository.database.QueryRowx(
`INSERT INTO public.headers (block_number, hash, block_timestamp, raw, eth_node_id, eth_node_fingerprint)
VALUES ($1, $2, $3::NUMERIC, $4, $5, $6) ON CONFLICT DO NOTHING RETURNING id`,
header.BlockNumber, header.Hash, header.Timestamp, header.Raw, repository.database.NodeID, repository.database.Node.ID)
err := row.Scan(&headerId)
if err != nil {
log.Error("insertHeader: error inserting header: ", err)
if err == sql.ErrNoRows {
return 0, ErrValidHeaderExists
}
log.Error("InternalInsertHeader: error inserting header: ", err)
}
return headerId, err
}
@@ -146,5 +154,5 @@ func (repository HeaderRepository) replaceHeader(header core.Header) (int64, err
log.Error("replaceHeader: error deleting headers: ", err)
return 0, err
}
return repository.insertHeader(header)
return repository.InternalInsertHeader(header)
}
@@ -98,6 +98,20 @@ var _ = Describe("Block header repository", func() {
Expect(len(dbHeaders)).To(Equal(1))
})
It("does not duplicate headers in concurrent insert", func() {
_, err = repo.InternalInsertHeader(header)
Expect(err).NotTo(HaveOccurred())
_, err = repo.InternalInsertHeader(header)
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(repositories.ErrValidHeaderExists))
var dbHeaders []core.Header
err = db.Select(&dbHeaders, `SELECT block_number, hash, raw FROM public.headers WHERE block_number = $1`, header.BlockNumber)
Expect(err).NotTo(HaveOccurred())
Expect(len(dbHeaders)).To(Equal(1))
})
It("replaces header if hash is different", func() {
_, err = repo.CreateOrUpdateHeader(header)
Expect(err).NotTo(HaveOccurred())
+1 -1
View File
@@ -14,7 +14,7 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package geth
package eth
import (
"errors"
+16 -16
View File
@@ -14,7 +14,7 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package geth_test
package eth_test
import (
"net/http"
@@ -25,7 +25,7 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/onsi/gomega/ghttp"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/eth"
"github.com/vulcanize/vulcanizedb/test_config"
)
@@ -36,7 +36,7 @@ var _ = Describe("ABI files", func() {
It("loads a valid ABI file", func() {
path := test_config.ABIFilePath + "valid_abi.json"
contractAbi, err := geth.ParseAbiFile(path)
contractAbi, err := eth.ParseAbiFile(path)
Expect(contractAbi).NotTo(BeNil())
Expect(err).To(BeNil())
@@ -45,7 +45,7 @@ var _ = Describe("ABI files", func() {
It("reads the contents of a valid ABI file", func() {
path := test_config.ABIFilePath + "valid_abi.json"
contractAbi, err := geth.ReadAbiFile(path)
contractAbi, err := eth.ReadAbiFile(path)
Expect(contractAbi).To(Equal("[{\"foo\": \"bar\"}]"))
Expect(err).To(BeNil())
@@ -54,38 +54,38 @@ var _ = Describe("ABI files", func() {
It("returns an error when the file does not exist", func() {
path := test_config.ABIFilePath + "missing_abi.json"
contractAbi, err := geth.ParseAbiFile(path)
contractAbi, err := eth.ParseAbiFile(path)
Expect(contractAbi).To(Equal(abi.ABI{}))
Expect(err).To(Equal(geth.ErrMissingAbiFile))
Expect(err).To(Equal(eth.ErrMissingAbiFile))
})
It("returns an error when the file has invalid contents", func() {
path := test_config.ABIFilePath + "invalid_abi.json"
contractAbi, err := geth.ParseAbiFile(path)
contractAbi, err := eth.ParseAbiFile(path)
Expect(contractAbi).To(Equal(abi.ABI{}))
Expect(err).To(Equal(geth.ErrInvalidAbiFile))
Expect(err).To(Equal(eth.ErrInvalidAbiFile))
})
Describe("Request ABI from endpoint", func() {
var (
server *ghttp.Server
client *geth.EtherScanAPI
client *eth.EtherScanAPI
abiString string
err error
)
BeforeEach(func() {
server = ghttp.NewServer()
client = geth.NewEtherScanClient(server.URL())
client = eth.NewEtherScanClient(server.URL())
path := test_config.ABIFilePath + "sample_abi.json"
abiString, err = geth.ReadAbiFile(path)
abiString, err = eth.ReadAbiFile(path)
Expect(err).NotTo(HaveOccurred())
_, err = geth.ParseAbi(abiString)
_, err = eth.ParseAbi(abiString)
Expect(err).NotTo(HaveOccurred())
})
@@ -117,14 +117,14 @@ var _ = Describe("ABI files", func() {
Describe("Generating etherscan endpoints based on network", func() {
It("should return the main endpoint as the default", func() {
url := geth.GenURL("")
url := eth.GenURL("")
Expect(url).To(Equal("https://api.etherscan.io"))
})
It("generates various test network endpoint if test network is supplied", func() {
ropstenUrl := geth.GenURL("ropsten")
rinkebyUrl := geth.GenURL("rinkeby")
kovanUrl := geth.GenURL("kovan")
ropstenUrl := eth.GenURL("ropsten")
rinkebyUrl := eth.GenURL("rinkeby")
kovanUrl := eth.GenURL("kovan")
Expect(ropstenUrl).To(Equal("https://ropsten.etherscan.io"))
Expect(kovanUrl).To(Equal("https://kovan.etherscan.io"))
@@ -14,7 +14,7 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package geth
package eth
import (
"errors"
@@ -28,8 +28,8 @@ import (
"golang.org/x/net/context"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/geth/client"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/geth/converters/common"
"github.com/vulcanize/vulcanizedb/pkg/eth/client"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/eth/converters/common"
)
var ErrEmptyHeader = errors.New("empty header returned over RPC")
@@ -14,7 +14,7 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package geth_test
package eth_test
import (
"context"
@@ -29,14 +29,14 @@ import (
. "github.com/onsi/gomega"
vulcCore "github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/eth"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/geth"
)
var _ = Describe("Geth blockchain", func() {
var (
mockClient *fakes.MockEthClient
blockChain *geth.BlockChain
blockChain *eth.BlockChain
mockRpcClient *fakes.MockRpcClient
mockTransactionConverter *fakes.MockTransactionConverter
node vulcCore.Node
@@ -47,7 +47,7 @@ var _ = Describe("Geth blockchain", func() {
mockRpcClient = fakes.NewMockRpcClient()
mockTransactionConverter = fakes.NewMockTransactionConverter()
node = vulcCore.Node{}
blockChain = geth.NewBlockChain(mockClient, mockRpcClient, node, mockTransactionConverter)
blockChain = eth.NewBlockChain(mockClient, mockRpcClient, node, mockTransactionConverter)
})
Describe("getting a block", func() {
@@ -105,7 +105,7 @@ var _ = Describe("Geth blockchain", func() {
node.NetworkID = vulcCore.KOVAN_NETWORK_ID
blockNumber := hexutil.Big(*big.NewInt(100))
mockRpcClient.SetReturnPOAHeader(vulcCore.POAHeader{Number: &blockNumber})
blockChain = geth.NewBlockChain(mockClient, mockRpcClient, node, fakes.NewMockTransactionConverter())
blockChain = eth.NewBlockChain(mockClient, mockRpcClient, node, fakes.NewMockTransactionConverter())
_, err := blockChain.GetHeaderByNumber(100)
@@ -116,7 +116,7 @@ var _ = Describe("Geth blockchain", func() {
It("returns err if rpcClient returns err", func() {
node.NetworkID = vulcCore.KOVAN_NETWORK_ID
mockRpcClient.SetCallContextErr(fakes.FakeError)
blockChain = geth.NewBlockChain(mockClient, mockRpcClient, node, fakes.NewMockTransactionConverter())
blockChain = eth.NewBlockChain(mockClient, mockRpcClient, node, fakes.NewMockTransactionConverter())
_, err := blockChain.GetHeaderByNumber(100)
@@ -126,12 +126,12 @@ var _ = Describe("Geth blockchain", func() {
It("returns error if returned header is empty", func() {
node.NetworkID = vulcCore.KOVAN_NETWORK_ID
blockChain = geth.NewBlockChain(mockClient, mockRpcClient, node, fakes.NewMockTransactionConverter())
blockChain = eth.NewBlockChain(mockClient, mockRpcClient, node, fakes.NewMockTransactionConverter())
_, err := blockChain.GetHeaderByNumber(100)
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(geth.ErrEmptyHeader))
Expect(err).To(MatchError(eth.ErrEmptyHeader))
})
It("returns multiple headers with multiple blocknumbers", func() {
@@ -19,7 +19,7 @@ package cold_import
import (
"github.com/vulcanize/vulcanizedb/pkg/datastore"
"github.com/vulcanize/vulcanizedb/pkg/datastore/ethereum"
"github.com/vulcanize/vulcanizedb/pkg/geth/converters/common"
"github.com/vulcanize/vulcanizedb/pkg/eth/converters/common"
)
type ColdImporter struct {
@@ -21,9 +21,9 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
"github.com/vulcanize/vulcanizedb/pkg/eth/cold_import"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/eth/converters/common"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/geth/cold_import"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/geth/converters/common"
)
var _ = Describe("Geth cold importer", func() {
@@ -21,8 +21,8 @@ import (
"github.com/ethereum/go-ethereum/common"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/eth/cold_import"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/geth/cold_import"
)
var _ = Describe("Cold importer node builder", func() {
+1 -1
View File
@@ -14,7 +14,7 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package geth
package eth
import (
"context"
@@ -29,9 +29,9 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/eth/converters/common"
"github.com/vulcanize/vulcanizedb/pkg/eth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/geth/converters/common"
"github.com/vulcanize/vulcanizedb/pkg/geth/converters/rpc"
)
var _ = Describe("Conversion of GethBlock to core.Block", func() {
@@ -26,7 +26,7 @@ import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/geth/converters/common"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/eth/converters/common"
)
var _ = Describe("Conversion of GethLog to core.FullSyncLog", func() {
@@ -26,8 +26,8 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
common2 "github.com/vulcanize/vulcanizedb/pkg/eth/converters/common"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
common2 "github.com/vulcanize/vulcanizedb/pkg/geth/converters/common"
)
var _ = Describe("Block header converter", func() {
@@ -25,7 +25,7 @@ import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/geth/converters/common"
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/eth/converters/common"
)
var _ = Describe("Conversion of GethReceipt to core.Receipt", func() {

Some files were not shown because too many files have changed in this diff Show More