Decouple log extraction from transformer delegation

- limit missing headers results set to 100 so that extraction doesn't
  excessively block delegation
- wrap checked headers functions in repository struct
- move storage repository to factory, to correspond with event
  repository path
- remove unused files
- reformat sql
- remove line breaks in imports
This commit is contained in:
Rob Mulholand
2019-08-28 09:25:13 -05:00
parent cb819fa9a6
commit d496dad33c
81 changed files with 2049 additions and 1328 deletions
+34 -108
View File
@@ -17,141 +17,67 @@
package watcher
import (
"fmt"
"github.com/vulcanize/vulcanizedb/libraries/shared/transactions"
"github.com/ethereum/go-ethereum/common"
"github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/libraries/shared/chunker"
"github.com/vulcanize/vulcanizedb/libraries/shared/constants"
"github.com/vulcanize/vulcanizedb/libraries/shared/fetcher"
"github.com/vulcanize/vulcanizedb/libraries/shared/repository"
"github.com/vulcanize/vulcanizedb/libraries/shared/logs"
"github.com/vulcanize/vulcanizedb/libraries/shared/transactions"
"github.com/vulcanize/vulcanizedb/libraries/shared/transformer"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
)
type EventWatcher struct {
Transformers []transformer.EventTransformer
BlockChain core.BlockChain
DB *postgres.DB
Fetcher fetcher.ILogFetcher
Chunker chunker.Chunker
Addresses []common.Address
Topics []common.Hash
StartingBlock *int64
Syncer transactions.ITransactionsSyncer
blockChain core.BlockChain
db *postgres.DB
LogDelegator logs.ILogDelegator
LogExtractor logs.ILogExtractor
}
func NewEventWatcher(db *postgres.DB, bc core.BlockChain) EventWatcher {
logChunker := chunker.NewLogChunker()
logFetcher := fetcher.NewLogFetcher(bc)
transactionSyncer := transactions.NewTransactionsSyncer(db, bc)
extractor := &logs.LogExtractor{
Fetcher: fetcher.NewLogFetcher(bc),
CheckedHeadersRepository: repositories.NewCheckedHeadersRepository(db),
LogRepository: repositories.NewHeaderSyncLogRepository(db),
Syncer: transactions.NewTransactionsSyncer(db, bc),
}
logTransformer := &logs.LogDelegator{
Chunker: chunker.NewLogChunker(),
LogRepository: repositories.NewHeaderSyncLogRepository(db),
}
return EventWatcher{
BlockChain: bc,
DB: db,
Fetcher: logFetcher,
Chunker: logChunker,
Syncer: transactionSyncer,
blockChain: bc,
db: db,
LogExtractor: extractor,
LogDelegator: logTransformer,
}
}
// Adds transformers to the watcher and updates the chunker, so that it will consider the new transformers.
// Adds transformers to the watcher so that their logs will be extracted and delegated.
func (watcher *EventWatcher) AddTransformers(initializers []transformer.EventTransformerInitializer) {
var contractAddresses []common.Address
var topic0s []common.Hash
var configs []transformer.EventTransformerConfig
for _, initializer := range initializers {
t := initializer(watcher.DB)
watcher.Transformers = append(watcher.Transformers, t)
t := initializer(watcher.db)
config := t.GetConfig()
configs = append(configs, config)
if watcher.StartingBlock == nil {
watcher.StartingBlock = &config.StartingBlockNumber
} else if earlierStartingBlockNumber(config.StartingBlockNumber, *watcher.StartingBlock) {
watcher.StartingBlock = &config.StartingBlockNumber
}
addresses := transformer.HexStringsToAddresses(config.ContractAddresses)
contractAddresses = append(contractAddresses, addresses...)
topic0s = append(topic0s, common.HexToHash(config.Topic))
watcher.LogDelegator.AddTransformer(t)
watcher.LogExtractor.AddTransformerConfig(t.GetConfig())
}
watcher.Addresses = append(watcher.Addresses, contractAddresses...)
watcher.Topics = append(watcher.Topics, topic0s...)
watcher.Chunker.AddConfigs(configs)
}
// Extracts and delegates watched log events.
func (watcher *EventWatcher) Execute(recheckHeaders constants.TransformerExecution) error {
if watcher.Transformers == nil {
return fmt.Errorf("No transformers added to watcher")
extractErr := watcher.LogExtractor.ExtractLogs(recheckHeaders)
if extractErr != nil {
logrus.Errorf("error extracting logs in event watcher: %s", extractErr.Error())
return extractErr
}
checkedColumnNames, err := repository.GetCheckedColumnNames(watcher.DB)
if err != nil {
return err
}
notCheckedSQL := repository.CreateHeaderCheckedPredicateSQL(checkedColumnNames, recheckHeaders)
missingHeaders, err := repository.MissingHeaders(*watcher.StartingBlock, -1, watcher.DB, notCheckedSQL)
if err != nil {
logrus.Error("Couldn't fetch missing headers in watcher: ", err)
return err
delegateErr := watcher.LogDelegator.DelegateLogs()
if delegateErr != nil {
logrus.Errorf("error delegating logs in event watcher: %s", delegateErr.Error())
return delegateErr
}
for _, header := range missingHeaders {
// TODO Extend FetchLogs for doing several blocks at a time
logs, err := watcher.Fetcher.FetchLogs(watcher.Addresses, watcher.Topics, header)
if err != nil {
logrus.WithFields(logrus.Fields{
"headerId": header.Id,
"headerHash": header.Hash,
"blockNumber": header.BlockNumber,
}).Errorf("Couldn't fetch logs for header: %v", err)
return err
}
transactionsSyncErr := watcher.Syncer.SyncTransactions(header.Id, logs)
if transactionsSyncErr != nil {
logrus.Errorf("error syncing transactions: %s", transactionsSyncErr.Error())
return transactionsSyncErr
}
persistedLogs, createLogsErr := repository.CreateLogs(header.Id, logs, watcher.DB)
if createLogsErr != nil {
logrus.Errorf("error persisting logs: %s", createLogsErr.Error())
}
transformErr := watcher.transformLogs(persistedLogs, header.Id)
if transformErr != nil {
logrus.Error("Could not transform logs: ", transformErr)
return transformErr
}
}
return err
}
func (watcher *EventWatcher) transformLogs(logs []core.HeaderSyncLog, headerID int64) 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, headerID)
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
}
+57 -199
View File
@@ -17,239 +17,97 @@
package watcher_test
import (
"errors"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/libraries/shared/constants"
"github.com/vulcanize/vulcanizedb/libraries/shared/mocks"
"github.com/vulcanize/vulcanizedb/libraries/shared/transformer"
"github.com/vulcanize/vulcanizedb/libraries/shared/watcher"
"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/fakes"
"github.com/vulcanize/vulcanizedb/test_config"
)
var _ = Describe("Watcher", func() {
It("initialises correctly", func() {
db := test_config.NewTestDB(core.Node{ID: "testNode"})
bc := fakes.NewMockBlockChain()
var _ = Describe("Event Watcher", func() {
var (
delegator *mocks.MockLogDelegator
extractor *mocks.MockLogExtractor
eventWatcher watcher.EventWatcher
)
w := watcher.NewEventWatcher(db, bc)
Expect(w.DB).To(Equal(db))
Expect(w.Fetcher).NotTo(BeNil())
Expect(w.Chunker).NotTo(BeNil())
BeforeEach(func() {
delegator = &mocks.MockLogDelegator{}
extractor = &mocks.MockLogExtractor{}
eventWatcher = watcher.EventWatcher{
LogDelegator: delegator,
LogExtractor: extractor,
}
})
It("adds transformers", func() {
w := watcher.NewEventWatcher(nil, nil)
fakeTransformer := &mocks.MockTransformer{}
fakeTransformer.SetTransformerConfig(mocks.FakeTransformerConfig)
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
Expect(len(w.Transformers)).To(Equal(1))
Expect(w.Transformers).To(ConsistOf(fakeTransformer))
Expect(w.Topics).To(Equal([]common.Hash{common.HexToHash("FakeTopic")}))
Expect(w.Addresses).To(Equal([]common.Address{common.HexToAddress("FakeAddress")}))
})
It("adds transformers from multiple sources", func() {
w := watcher.NewEventWatcher(nil, nil)
fakeTransformer1 := &mocks.MockTransformer{}
fakeTransformer1.SetTransformerConfig(mocks.FakeTransformerConfig)
fakeTransformer2 := &mocks.MockTransformer{}
fakeTransformer2.SetTransformerConfig(mocks.FakeTransformerConfig)
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer1.FakeTransformerInitializer})
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer2.FakeTransformerInitializer})
Expect(len(w.Transformers)).To(Equal(2))
Expect(w.Topics).To(Equal([]common.Hash{common.HexToHash("FakeTopic"),
common.HexToHash("FakeTopic")}))
Expect(w.Addresses).To(Equal([]common.Address{common.HexToAddress("FakeAddress"),
common.HexToAddress("FakeAddress")}))
})
It("calculates earliest starting block number", func() {
fakeTransformer1 := &mocks.MockTransformer{}
fakeTransformer1.SetTransformerConfig(transformer.EventTransformerConfig{StartingBlockNumber: 5})
fakeTransformer2 := &mocks.MockTransformer{}
fakeTransformer2.SetTransformerConfig(transformer.EventTransformerConfig{StartingBlockNumber: 3})
w := watcher.NewEventWatcher(nil, nil)
w.AddTransformers([]transformer.EventTransformerInitializer{
fakeTransformer1.FakeTransformerInitializer,
fakeTransformer2.FakeTransformerInitializer,
})
Expect(*w.StartingBlock).To(Equal(int64(3)))
})
It("returns an error when run without transformers", func() {
w := watcher.NewEventWatcher(nil, nil)
err := w.Execute(constants.HeaderMissing)
Expect(err).To(MatchError("No transformers added to watcher"))
})
Describe("with missing headers", func() {
Describe("AddTransformers", func() {
var (
db *postgres.DB
w watcher.EventWatcher
mockBlockChain fakes.MockBlockChain
headerRepository repositories.HeaderRepository
fakeTransformerOne, fakeTransformerTwo *mocks.MockEventTransformer
)
BeforeEach(func() {
db = test_config.NewTestDB(test_config.NewTestNode())
test_config.CleanTestDB(db)
mockBlockChain = fakes.MockBlockChain{}
headerRepository = repositories.NewHeaderRepository(db)
_, err := headerRepository.CreateOrUpdateHeader(fakes.FakeHeader)
Expect(err).NotTo(HaveOccurred())
fakeTransformerOne = &mocks.MockEventTransformer{}
fakeTransformerOne.SetTransformerConfig(mocks.FakeTransformerConfig)
fakeTransformerTwo = &mocks.MockEventTransformer{}
fakeTransformerTwo.SetTransformerConfig(mocks.FakeTransformerConfig)
initializers := []transformer.EventTransformerInitializer{
fakeTransformerOne.FakeTransformerInitializer,
fakeTransformerTwo.FakeTransformerInitializer,
}
w = watcher.NewEventWatcher(db, &mockBlockChain)
eventWatcher.AddTransformers(initializers)
})
It("syncs transactions for fetched logs", func() {
fakeTransformer := &mocks.MockTransformer{}
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
mockTransactionSyncer := &fakes.MockTransactionSyncer{}
w.Syncer = mockTransactionSyncer
err := w.Execute(constants.HeaderMissing)
Expect(err).NotTo(HaveOccurred())
Expect(mockTransactionSyncer.SyncTransactionsCalled).To(BeTrue())
It("adds initialized transformer to log delegator", func() {
expectedTransformers := []transformer.EventTransformer{
fakeTransformerOne,
fakeTransformerTwo,
}
Expect(delegator.AddedTransformers).To(Equal(expectedTransformers))
})
It("returns error if syncing transactions fails", func() {
fakeTransformer := &mocks.MockTransformer{}
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
mockTransactionSyncer := &fakes.MockTransactionSyncer{}
mockTransactionSyncer.SyncTransactionsError = fakes.FakeError
w.Syncer = mockTransactionSyncer
It("adds transformer config to log extractor", func() {
expectedConfigs := []transformer.EventTransformerConfig{
mocks.FakeTransformerConfig,
mocks.FakeTransformerConfig,
}
Expect(extractor.AddedConfigs).To(Equal(expectedConfigs))
})
})
err := w.Execute(constants.HeaderMissing)
Describe("Execute", func() {
It("extracts watched logs", func() {
err := eventWatcher.Execute(constants.HeaderMissing)
Expect(err).NotTo(HaveOccurred())
Expect(extractor.ExtractLogsCalled).To(BeTrue())
})
It("returns error if extracting logs fails", func() {
extractor.ExtractLogsError = fakes.FakeError
err := eventWatcher.Execute(constants.HeaderMissing)
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(fakes.FakeError))
})
It("persists fetched logs", func() {
fakeTransformer := &mocks.MockTransformer{}
transformerConfig := transformer.EventTransformerConfig{TransformerName: "transformerA",
ContractAddresses: []string{"0x000000000000000000000000000000000000000A"},
Topic: "0xA"}
fakeTransformer.SetTransformerConfig(transformerConfig)
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
log := types.Log{Address: common.HexToAddress("0xA"),
Topics: []common.Hash{common.HexToHash("0xA")},
Index: 0,
}
mockBlockChain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{log})
err := w.Execute(constants.HeaderMissing)
It("delegates untransformed logs", func() {
err := eventWatcher.Execute(constants.HeaderMissing)
Expect(err).NotTo(HaveOccurred())
Expect(len(fakeTransformer.PassedLogs)).NotTo(BeZero())
Expect(fakeTransformer.PassedLogs[0].ID).NotTo(BeZero())
Expect(fakeTransformer.PassedLogs[0].Log).To(Equal(log))
Expect(delegator.DelegateCalled).To(BeTrue())
})
It("executes each transformer", func() {
fakeTransformer := &mocks.MockTransformer{}
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
It("returns error if delegating logs fails", func() {
delegator.DelegateError = fakes.FakeError
err := w.Execute(constants.HeaderMissing)
Expect(err).NotTo(HaveOccurred())
Expect(fakeTransformer.ExecuteWasCalled).To(BeTrue())
})
err := eventWatcher.Execute(constants.HeaderMissing)
It("returns an error if transformer returns an error", func() {
fakeTransformer := &mocks.MockTransformer{ExecuteError: errors.New("Something bad happened")}
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
err := w.Execute(constants.HeaderMissing)
Expect(err).To(HaveOccurred())
Expect(fakeTransformer.ExecuteWasCalled).To(BeFalse())
})
It("passes only relevant logs to each transformer", func() {
transformerA := &mocks.MockTransformer{}
transformerB := &mocks.MockTransformer{}
configA := transformer.EventTransformerConfig{TransformerName: "transformerA",
ContractAddresses: []string{"0x000000000000000000000000000000000000000A"},
Topic: "0xA"}
configB := transformer.EventTransformerConfig{TransformerName: "transformerB",
ContractAddresses: []string{"0x000000000000000000000000000000000000000b"},
Topic: "0xB"}
transformerA.SetTransformerConfig(configA)
transformerB.SetTransformerConfig(configB)
logA := types.Log{Address: common.HexToAddress("0xA"),
Topics: []common.Hash{common.HexToHash("0xA")},
Index: 0,
}
logB := types.Log{Address: common.HexToAddress("0xB"),
Topics: []common.Hash{common.HexToHash("0xB")},
Index: 1,
}
mockBlockChain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{logA, logB})
w = watcher.NewEventWatcher(db, &mockBlockChain)
w.AddTransformers([]transformer.EventTransformerInitializer{
transformerA.FakeTransformerInitializer, transformerB.FakeTransformerInitializer})
err := w.Execute(constants.HeaderMissing)
Expect(err).NotTo(HaveOccurred())
Expect(len(transformerA.PassedLogs)).NotTo(BeZero())
Expect(transformerA.PassedLogs[0].Log).To(Equal(logA))
Expect(len(transformerB.PassedLogs)).NotTo(BeZero())
Expect(transformerB.PassedLogs[0].Log).To(Equal(logB))
})
Describe("uses the LogFetcher correctly:", func() {
var fakeTransformer mocks.MockTransformer
BeforeEach(func() {
fakeTransformer = mocks.MockTransformer{}
})
It("fetches logs for added transformers", func() {
addresses := []string{"0xA", "0xB"}
topic := "0x1"
fakeTransformer.SetTransformerConfig(transformer.EventTransformerConfig{
Topic: topic, ContractAddresses: addresses})
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
err := w.Execute(constants.HeaderMissing)
Expect(err).NotTo(HaveOccurred())
fakeHash := common.HexToHash(fakes.FakeHeader.Hash)
mockBlockChain.AssertGetEthLogsWithCustomQueryCalledWith(ethereum.FilterQuery{
BlockHash: &fakeHash,
Addresses: transformer.HexStringsToAddresses(addresses),
Topics: [][]common.Hash{{common.HexToHash(topic)}},
})
})
It("propagates log fetcher errors", func() {
fetcherError := errors.New("FetcherError")
mockBlockChain.SetGetEthLogsWithCustomQueryErr(fetcherError)
w.AddTransformers([]transformer.EventTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
err := w.Execute(constants.HeaderMissing)
Expect(err).To(MatchError(fetcherError))
})
Expect(err).To(MatchError(fakes.FakeError))
})
})
})