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

finishing porting omni watcher to work with light sync; split into full, light,

Browse Source 
and shared directories and refactor as much into shared; finish
lightSync omni watcher tests
This commit is contained in:
Ian Norden
2018-11-30 13:33:48 -06:00
parent 58b34548f3
commit e02b33547d
38 changed files with 1715 additions and 890 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/omni/light/transformer/transformer.go
+48 -31
View File
@@ -18,7 +18,8 @@ package transformer
import (
"errors"
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers"
"strings"
"github.com/ethereum/go-ethereum/common"
@@ -31,12 +32,14 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/omni/shared/contract"
"github.com/vulcanize/vulcanizedb/pkg/omni/shared/parser"
"github.com/vulcanize/vulcanizedb/pkg/omni/shared/poller"
srep "github.com/vulcanize/vulcanizedb/pkg/omni/shared/repository"
"github.com/vulcanize/vulcanizedb/pkg/omni/shared/types"
)
// Requires a light synced vDB (headers) and a running eth node (or infura)
type transformer struct {
// Database interfaces
repository.EventRepository // Holds transformed watched event log data
srep.EventRepository // Holds transformed watched event log data
repository.HeaderRepository // Interface for interaction with header repositories
// Pre-processing interfaces
@@ -72,14 +75,14 @@ type transformer struct {
func NewTransformer(network string, bc core.BlockChain, db *postgres.DB) *transformer {
return &transformer{
Poller: poller.NewPoller(bc, db),
Poller: poller.NewPoller(bc, db, types.LightSync),
Fetcher: fetcher.NewFetcher(bc),
Parser: parser.NewParser(network),
HeaderRepository: repository.NewHeaderRepository(db),
BlockRetriever: retriever.NewBlockRetriever(db),
Converter: converter.NewConverter(&contract.Contract{}),
Contracts: map[string]*contract.Contract{},
EventRepository: repository.NewEventRepository(db),
EventRepository: srep.NewEventRepository(db, types.LightSync),
WatchedEvents: map[string][]string{},
WantedMethods: map[string][]string{},
ContractRanges: map[string][2]int64{},
@@ -93,7 +96,7 @@ func NewTransformer(network string, bc core.BlockChain, db *postgres.DB) *transf
// Uses parser to pull event info from abi
// Use this info to generate event filters
func (tr *transformer) Init() error {
// Iterate through all internal contract addresses
for contractAddr, subset := range tr.WatchedEvents {
// Get Abi
err := tr.Parser.Parse(contractAddr)
@@ -101,7 +104,7 @@ func (tr *transformer) Init() error {
return err
}
// Get first block for contract and most recent block for the chain
// Get first block and most recent block number in the header repo
firstBlock, err := tr.BlockRetriever.RetrieveFirstBlock()
if err != nil {
return err
@@ -111,7 +114,7 @@ func (tr *transformer) Init() error {
return err
}
// Set to specified range if it falls within the contract's bounds
// Set to specified range if it falls within the bounds
if firstBlock < tr.ContractRanges[contractAddr][0] {
firstBlock = tr.ContractRanges[contractAddr][0]
}
@@ -119,14 +122,11 @@ func (tr *transformer) Init() error {
lastBlock = tr.ContractRanges[contractAddr][1]
}
// Get contract name
// Get contract name if it has one
var name = new(string)
err = tr.FetchContractData(tr.Abi(), contractAddr, "name", nil, &name, lastBlock)
if err != nil {
return errors.New(fmt.Sprintf("unable to fetch contract name: %v\r\n", err))
}
tr.FetchContractData(tr.Abi(), contractAddr, "name", nil, &name, lastBlock)
// Remove any accidental duplicate inputs in filter addresses
// Remove any potential accidental duplicate inputs in filter addresses
EventAddrs := map[string]bool{}
for _, addr := range tr.EventAddrs[contractAddr] {
EventAddrs[addr] = true
@@ -142,6 +142,7 @@ func (tr *transformer) Init() error {
Network: tr.Network,
Address: contractAddr,
Abi: tr.Abi(),
ParsedAbi: tr.ParsedAbi(),
StartingBlock: firstBlock,
LastBlock: lastBlock,
Events: tr.GetEvents(subset),
@@ -151,7 +152,7 @@ func (tr *transformer) Init() error {
TknHolderAddrs: map[string]bool{},
}
// Store contract info for further processing
// Store contract info for execution
tr.Contracts[contractAddr] = info
}
@@ -160,53 +161,69 @@ func (tr *transformer) Init() error {
func (tr *transformer) Execute() error {
if len(tr.Contracts) == 0 {
return errors.New("error: transformer has no initialized contracts to work with")
return errors.New("error: transformer has no initialized contracts")
}
// Iterate through all internal contracts
for _, con := range tr.Contracts {
// Update converter with current contract
tr.Update(con)
// Iterate through events
for _, event := range con.Events {
topics := [][]common.Hash{{common.HexToHash(event.Sig())}}
eventId := event.Name + "_" + con.Address
// Filter using the event signature
topics := [][]common.Hash{{common.HexToHash(helpers.GenerateSignature(event.Sig()))}}
// Generate eventID and use it to create a checked_header column if one does not already exist
eventId := strings.ToLower(event.Name + "_" + con.Address)
if err := tr.AddCheckColumn(eventId); err != nil {
return err
}
// Find unchecked headers for this event
missingHeaders, err := tr.MissingHeaders(con.StartingBlock, con.LastBlock, eventId)
if err != nil {
return err
}
// Iterate over headers
for _, header := range missingHeaders {
// And fetch event logs using the header, contract address, and topics filter
logs, err := tr.FetchLogs([]string{con.Address}, topics, header)
if err != nil {
return err
}
// Mark the header checked for this eventID and continue to next iteration if no logs are found
if len(logs) < 1 {
err = tr.MarkHeaderChecked(header.Id, eventId)
if err != nil {
return err
}
continue
}
for _, l := range logs {
mapping, err := tr.Convert(l, event)
if err != nil {
return err
}
if mapping == nil {
break
}
// Convert logs into custom type
convertedLogs, err := tr.Convert(logs, event, header.Id)
if err != nil {
return err
}
if len(convertedLogs) < 1 {
continue
}
err = tr.PersistLog(*mapping, con.Address, con.Name)
if err != nil {
return err
}
// If logs aren't empty, persist them
err = tr.PersistLogs(convertedLogs, event, con.Address, con.Name)
if err != nil {
return err
}
}
}
// After persisting all watched event logs
// poller polls select contract methods
// and persists the results into custom pg tables
if err := tr.PollContract(*con); err != nil {
return err
}
}
return nil
pkg/omni/light/transformer/transformer_test.go
+98 -9
View File
@@ -17,9 +17,7 @@
package transformer_test
import (
"math/rand"
"time"
"fmt"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
@@ -29,6 +27,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/omni/light/transformer"
"github.com/vulcanize/vulcanizedb/pkg/omni/shared/constants"
"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers"
"github.com/vulcanize/vulcanizedb/pkg/omni/shared/helpers/test_helpers/mocks"
)
var _ = Describe("Transformer", func() {
@@ -36,7 +35,7 @@ var _ = Describe("Transformer", func() {
var err error
var blockChain core.BlockChain
var headerRepository repositories.HeaderRepository
rand.Seed(time.Now().UnixNano())
var headerID int64
BeforeEach(func() {
db, blockChain = test_helpers.SetupDBandBC()
@@ -94,8 +93,8 @@ var _ = Describe("Transformer", func() {
Describe("Init", func() {
It("Initializes transformer's contract objects", func() {
headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader1)
headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader3)
headerRepository.CreateOrUpdateHeader(mocks.MockHeader1)
headerRepository.CreateOrUpdateHeader(mocks.MockHeader3)
t := transformer.NewTransformer("", blockChain, db)
t.SetEvents(constants.TusdContractAddress, []string{"Transfer"})
err = t.Init()
@@ -111,7 +110,7 @@ var _ = Describe("Transformer", func() {
Expect(c.Address).To(Equal(constants.TusdContractAddress))
})
It("Fails to initialize if first and most recent blocks cannot be fetched from vDB", func() {
It("Fails to initialize if first and most recent block numbers cannot be fetched from vDB headers table", func() {
t := transformer.NewTransformer("", blockChain, db)
t.SetEvents(constants.TusdContractAddress, []string{"Transfer"})
err = t.Init()
@@ -119,8 +118,8 @@ var _ = Describe("Transformer", func() {
})
It("Does nothing if watched events are unset", func() {
headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader1)
headerRepository.CreateOrUpdateHeader(test_helpers.MockHeader3)
headerRepository.CreateOrUpdateHeader(mocks.MockHeader1)
headerRepository.CreateOrUpdateHeader(mocks.MockHeader3)
t := transformer.NewTransformer("", blockChain, db)
err = t.Init()
Expect(err).ToNot(HaveOccurred())
@@ -131,6 +130,96 @@ var _ = Describe("Transformer", func() {
})
Describe("Execute", func() {
BeforeEach(func() {
header1, err := blockChain.GetHeaderByNumber(6791668)
Expect(err).ToNot(HaveOccurred())
header2, err := blockChain.GetHeaderByNumber(6791669)
Expect(err).ToNot(HaveOccurred())
header3, err := blockChain.GetHeaderByNumber(6791670)
Expect(err).ToNot(HaveOccurred())
headerRepository.CreateOrUpdateHeader(header1)
headerID, err = headerRepository.CreateOrUpdateHeader(header2)
Expect(err).ToNot(HaveOccurred())
headerRepository.CreateOrUpdateHeader(header3)
})
It("Transforms watched contract data into custom repositories", func() {
t := transformer.NewTransformer("", blockChain, db)
t.SetEvents(constants.TusdContractAddress, []string{"Transfer"})
t.SetMethods(constants.TusdContractAddress, nil)
err = t.Init()
Expect(err).ToNot(HaveOccurred())
err = t.Execute()
Expect(err).ToNot(HaveOccurred())
log := test_helpers.LightTransferLog{}
err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.transfer_event", constants.TusdContractAddress)).StructScan(&log)
// We don't know vulcID, so compare individual fields instead of complete structures
Expect(log.HeaderID).To(Equal(headerID))
Expect(log.From).To(Equal("0x1062a747393198f70F71ec65A582423Dba7E5Ab3"))
Expect(log.To).To(Equal("0x2930096dB16b4A44Ecd4084EA4bd26F7EeF1AEf0"))
Expect(log.Value).To(Equal("9998940000000000000000"))
})
It("Keeps track of contract-related addresses while transforming event data", func() {
t := transformer.NewTransformer("", blockChain, db)
t.SetEvents(constants.TusdContractAddress, []string{"Transfer"})
t.SetMethods(constants.TusdContractAddress, nil)
err = t.Init()
Expect(err).ToNot(HaveOccurred())
c, ok := t.Contracts[constants.TusdContractAddress]
Expect(ok).To(Equal(true))
err = t.Execute()
Expect(err).ToNot(HaveOccurred())
b, ok := c.TknHolderAddrs["0x1062a747393198f70F71ec65A582423Dba7E5Ab3"]
Expect(ok).To(Equal(true))
Expect(b).To(Equal(true))
b, ok = c.TknHolderAddrs["0x2930096dB16b4A44Ecd4084EA4bd26F7EeF1AEf0"]
Expect(ok).To(Equal(true))
Expect(b).To(Equal(true))
_, ok = c.TknHolderAddrs["0x09BbBBE21a5975cAc061D82f7b843b1234567890"]
Expect(ok).To(Equal(false))
_, ok = c.TknHolderAddrs["0x"]
Expect(ok).To(Equal(false))
})
It("Polls given methods using generated token holder address", func() {
t := transformer.NewTransformer("", blockChain, db)
t.SetEvents(constants.TusdContractAddress, []string{"Transfer"})
t.SetMethods(constants.TusdContractAddress, []string{"balanceOf"})
err = t.Init()
Expect(err).ToNot(HaveOccurred())
err = t.Execute()
Expect(err).ToNot(HaveOccurred())
res := test_helpers.BalanceOf{}
err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.balanceof_method WHERE who_ = '0x1062a747393198f70F71ec65A582423Dba7E5Ab3' AND block = '6791669'", constants.TusdContractAddress)).StructScan(&res)
Expect(err).ToNot(HaveOccurred())
Expect(res.Balance).To(Equal("55849938025000000000000"))
Expect(res.TokenName).To(Equal("TrueUSD"))
err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.balanceof_method WHERE who_ = '0x09BbBBE21a5975cAc061D82f7b843b1234567890' AND block = '6791669'", constants.TusdContractAddress)).StructScan(&res)
Expect(err).To(HaveOccurred())
})
It("Fails if initialization has not been done", func() {
t := transformer.NewTransformer("", blockChain, db)
t.SetEvents(constants.TusdContractAddress, []string{"Transfer"})
t.SetMethods(constants.TusdContractAddress, nil)
err = t.Execute()
Expect(err).To(HaveOccurred())
})
})
})