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

updates for light sync transactions

Browse Source
This commit is contained in:
Rob Mulholand
2019-03-28 14:31:17 -05:00
parent 79e011aad2
commit 54d46638a8
39 changed files with 769 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libraries/shared/transactions/syncer.go
+55
View File
@@ -0,0 +1,55 @@
package transactions
import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
)
type ITransactionsSyncer interface {
SyncTransactions(headerID int64, logs []types.Log) error
}
type TransactionsSyncer struct {
BlockChain core.BlockChain
Repository datastore.HeaderRepository
}
func NewTransactionsSyncer(db *postgres.DB, blockChain core.BlockChain) TransactionsSyncer {
repository := repositories.NewHeaderRepository(db)
return TransactionsSyncer{
BlockChain: blockChain,
Repository: repository,
}
}
func (syncer TransactionsSyncer) SyncTransactions(headerID int64, logs []types.Log) error {
transactionHashes := getUniqueTransactionHashes(logs)
transactions, transactionErr := syncer.BlockChain.GetTransactions(transactionHashes)
if transactionErr != nil {
return transactionErr
}
for _, transaction := range transactions {
writeErr := syncer.Repository.CreateTransaction(headerID, transaction)
if writeErr != nil {
return writeErr
}
}
return nil
}
func getUniqueTransactionHashes(logs []types.Log) []common.Hash {
seen := make(map[common.Hash]struct{}, len(logs))
var result []common.Hash
for _, log := range logs {
if _, ok := seen[log.TxHash]; ok {
continue
}
seen[log.TxHash] = struct{}{}
result = append(result, log.TxHash)
}
return result
}
libraries/shared/transactions/syncer_test.go
+80
View File
@@ -0,0 +1,80 @@
package transactions_test
import (
"github.com/ethereum/go-ethereum/core/types"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/libraries/shared/transactions"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/test_config"
)
var _ = Describe("Transaction syncer", func() {
It("fetches transactions for logs", func() {
db := test_config.NewTestDB(test_config.NewTestNode())
blockChain := fakes.NewMockBlockChain()
syncer := transactions.NewTransactionsSyncer(db, blockChain)
err := syncer.SyncTransactions(0, []types.Log{})
Expect(err).NotTo(HaveOccurred())
Expect(blockChain.GetTransactionsCalled).To(BeTrue())
})
It("only fetches transactions with unique hashes", func() {
db := test_config.NewTestDB(test_config.NewTestNode())
blockChain := fakes.NewMockBlockChain()
syncer := transactions.NewTransactionsSyncer(db, blockChain)
err := syncer.SyncTransactions(0, []types.Log{{
TxHash: fakes.FakeHash,
}, {
TxHash: fakes.FakeHash,
}})
Expect(err).NotTo(HaveOccurred())
Expect(len(blockChain.GetTransactionsPassedHashes)).To(Equal(1))
})
It("returns error if fetching transactions fails", func() {
db := test_config.NewTestDB(test_config.NewTestNode())
blockChain := fakes.NewMockBlockChain()
blockChain.GetTransactionsError = fakes.FakeError
syncer := transactions.NewTransactionsSyncer(db, blockChain)
err := syncer.SyncTransactions(0, []types.Log{})
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(fakes.FakeError))
})
It("passes transactions to repository for persistence", func() {
db := test_config.NewTestDB(test_config.NewTestNode())
blockChain := fakes.NewMockBlockChain()
blockChain.Transactions = []core.TransactionModel{{}}
syncer := transactions.NewTransactionsSyncer(db, blockChain)
mockHeaderRepository := fakes.NewMockHeaderRepository()
syncer.Repository = mockHeaderRepository
err := syncer.SyncTransactions(0, []types.Log{})
Expect(err).NotTo(HaveOccurred())
Expect(mockHeaderRepository.CreateTransactionCalled).To(BeTrue())
})
It("returns error if persisting transactions fails", func() {
db := test_config.NewTestDB(test_config.NewTestNode())
blockChain := fakes.NewMockBlockChain()
blockChain.Transactions = []core.TransactionModel{{}}
syncer := transactions.NewTransactionsSyncer(db, blockChain)
mockHeaderRepository := fakes.NewMockHeaderRepository()
mockHeaderRepository.CreateTransactionError = fakes.FakeError
syncer.Repository = mockHeaderRepository
err := syncer.SyncTransactions(0, []types.Log{})
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(fakes.FakeError))
})
})
libraries/shared/transactions/transactions_suite_test.go
+13
View File
@@ -0,0 +1,13 @@
package transactions_test
import (
"testing"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
)
func TestTransactions(t *testing.T) {
RegisterFailHandler(Fail)
RunSpecs(t, "Shared Transactions Suite")
}
libraries/shared/watcher/event_watcher.go
+48 -28
View File
@@ -18,14 +18,16 @@ package watcher
import (
"fmt"
"github.com/vulcanize/vulcanizedb/libraries/shared/transactions"
"github.com/ethereum/go-ethereum/common"
log "github.com/sirupsen/logrus"
"github.com/ethereum/go-ethereum/core/types"
"github.com/sirupsen/logrus"
chunk "github.com/vulcanize/vulcanizedb/libraries/shared/chunker"
"github.com/vulcanize/vulcanizedb/libraries/shared/chunker"
"github.com/vulcanize/vulcanizedb/libraries/shared/constants"
fetch "github.com/vulcanize/vulcanizedb/libraries/shared/fetcher"
repo "github.com/vulcanize/vulcanizedb/libraries/shared/repository"
"github.com/vulcanize/vulcanizedb/libraries/shared/fetcher"
"github.com/vulcanize/vulcanizedb/libraries/shared/repository"
"github.com/vulcanize/vulcanizedb/libraries/shared/transformer"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
@@ -33,21 +35,26 @@ import (
type EventWatcher struct {
Transformers []transformer.EventTransformer
BlockChain core.BlockChain
DB *postgres.DB
Fetcher fetch.LogFetcher
Chunker chunk.Chunker
Fetcher fetcher.LogFetcher
Chunker chunker.Chunker
Addresses []common.Address
Topics []common.Hash
StartingBlock *int64
Syncer transactions.ITransactionsSyncer
}
func NewEventWatcher(db *postgres.DB, bc core.BlockChain) EventWatcher {
chunker := chunk.NewLogChunker()
fetcher := fetch.NewFetcher(bc)
logChunker := chunker.NewLogChunker()
logFetcher := fetcher.NewFetcher(bc)
transactionSyncer := transactions.NewTransactionsSyncer(db, bc)
return EventWatcher{
DB: db,
Fetcher: fetcher,
Chunker: chunker,
BlockChain: bc,
DB: db,
Fetcher: logFetcher,
Chunker: logChunker,
Syncer: transactionSyncer,
}
}
@@ -85,15 +92,15 @@ func (watcher *EventWatcher) Execute(recheckHeaders constants.TransformerExecuti
return fmt.Errorf("No transformers added to watcher")
}
checkedColumnNames, err := repo.GetCheckedColumnNames(watcher.DB)
checkedColumnNames, err := repository.GetCheckedColumnNames(watcher.DB)
if err != nil {
return err
}
notCheckedSQL := repo.CreateNotCheckedSQL(checkedColumnNames, recheckHeaders)
notCheckedSQL := repository.CreateNotCheckedSQL(checkedColumnNames, recheckHeaders)
missingHeaders, err := repo.MissingHeaders(*watcher.StartingBlock, -1, watcher.DB, notCheckedSQL)
missingHeaders, err := repository.MissingHeaders(*watcher.StartingBlock, -1, watcher.DB, notCheckedSQL)
if err != nil {
log.Error("Fetching of missing headers failed in watcher!")
logrus.Error("Fetching of missing headers failed in watcher!")
return err
}
@@ -101,28 +108,41 @@ func (watcher *EventWatcher) Execute(recheckHeaders constants.TransformerExecuti
// TODO Extend FetchLogs for doing several blocks at a time
logs, err := watcher.Fetcher.FetchLogs(watcher.Addresses, watcher.Topics, header)
if err != nil {
// TODO Handle fetch error in watcher
log.Errorf("Error while fetching logs for header %v in watcher", header.Id)
logrus.Errorf("Error while fetching logs for header %v in watcher", header.Id)
return err
}
chunkedLogs := watcher.Chunker.ChunkLogs(logs)
transactionsSyncErr := watcher.Syncer.SyncTransactions(header.Id, logs)
if transactionsSyncErr != nil {
logrus.Errorf("error syncing transactions: %s", transactionsSyncErr.Error())
return transactionsSyncErr
}
// Can't quit early and mark as checked if there are no logs. If we are running continuousLogSync,
// not all logs we're interested in might have been fetched.
for _, t := range watcher.Transformers {
transformerName := t.GetConfig().TransformerName
logChunk := chunkedLogs[transformerName]
err = t.Execute(logChunk, header, constants.HeaderMissing)
if err != nil {
log.Errorf("%v transformer failed to execute in watcher: %v", transformerName, err)
return err
}
transformErr := watcher.transformLogs(logs, header)
if transformErr != nil {
return transformErr
}
}
return err
}
func (watcher *EventWatcher) transformLogs(logs []types.Log, header core.Header) error {
chunkedLogs := watcher.Chunker.ChunkLogs(logs)
// Can't quit early and mark as checked if there are no logs. If we are running continuousLogSync,
// not all logs we're interested in might have been fetched.
for _, t := range watcher.Transformers {
transformerName := t.GetConfig().TransformerName
logChunk := chunkedLogs[transformerName]
err := t.Execute(logChunk, header, constants.HeaderMissing)
if err != nil {
logrus.Errorf("%v transformer failed to execute in watcher: %v", transformerName, err)
return err
}
}
return nil
}
func earlierStartingBlockNumber(transformerBlock, watcherBlock int64) bool {
return transformerBlock < watcherBlock
}
libraries/shared/watcher/event_watcher_test.go
+27
View File
@@ -121,6 +121,33 @@ var _ = Describe("Watcher", func() {
w = watcher.NewEventWatcher(db, &mockBlockChain)
})
It("syncs transactions for fetched logs", func() {
fakeTransformer := &mocks.MockTransformer{}
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
repository.SetMissingHeaders([]core.Header{fakes.FakeHeader})
mockTransactionSyncer := &fakes.MockTransactionSyncer{}
w.Syncer = mockTransactionSyncer
err := w.Execute(constants.HeaderMissing)
Expect(err).NotTo(HaveOccurred())
Expect(mockTransactionSyncer.SyncTransactionsCalled).To(BeTrue())
})
It("returns error if syncing transactions fails", func() {
fakeTransformer := &mocks.MockTransformer{}
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
repository.SetMissingHeaders([]core.Header{fakes.FakeHeader})
mockTransactionSyncer := &fakes.MockTransactionSyncer{}
mockTransactionSyncer.SyncTransactionsError = fakes.FakeError
w.Syncer = mockTransactionSyncer
err := w.Execute(constants.HeaderMissing)
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(fakes.FakeError))
})
It("executes each transformer", func() {
fakeTransformer := &mocks.MockTransformer{}
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer.FakeTransformerInitializer})