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

Combine price feed transformers

Browse Source 
- fetches logs from all three price feeds in one query
- assumes eth/usd price feed will be updated to include LogValue event
- updates transformers to run separate from header sync
This commit is contained in:
Rob Mulholand
2018-08-16 11:22:16 -05:00
parent 9231d40369
commit 634604d0b5
68 changed files with 902 additions and 1384 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/history/header_validator.go
+2 -5
View File
@@ -3,28 +3,25 @@ package history
import (
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore"
"github.com/vulcanize/vulcanizedb/pkg/transformers"
)
type HeaderValidator struct {
blockChain core.BlockChain
headerRepository datastore.HeaderRepository
windowSize int
transformers []transformers.Transformer
}
func NewHeaderValidator(blockChain core.BlockChain, repository datastore.HeaderRepository, windowSize int, transformers []transformers.Transformer) HeaderValidator {
func NewHeaderValidator(blockChain core.BlockChain, repository datastore.HeaderRepository, windowSize int) HeaderValidator {
return HeaderValidator{
blockChain: blockChain,
headerRepository: repository,
windowSize: windowSize,
transformers: transformers,
}
}
func (validator HeaderValidator) ValidateHeaders() ValidationWindow {
window := MakeValidationWindow(validator.blockChain, validator.windowSize)
blockNumbers := MakeRange(window.LowerBound, window.UpperBound)
RetrieveAndUpdateHeaders(validator.blockChain, validator.headerRepository, blockNumbers, validator.transformers)
RetrieveAndUpdateHeaders(validator.blockChain, validator.headerRepository, blockNumbers)
return window
}
pkg/history/header_validator_test.go
+1 -17
View File
@@ -2,10 +2,8 @@ package history_test
import (
. "github.com/onsi/ginkgo"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/history"
"github.com/vulcanize/vulcanizedb/pkg/transformers"
"math/big"
)
@@ -15,24 +13,10 @@ var _ = Describe("Header validator", func() {
headerRepository.SetMissingBlockNumbers([]int64{})
blockChain := fakes.NewMockBlockChain()
blockChain.SetLastBlock(big.NewInt(3))
validator := history.NewHeaderValidator(blockChain, headerRepository, 2, []transformers.Transformer{})
validator := history.NewHeaderValidator(blockChain, headerRepository, 2)
validator.ValidateHeaders()
headerRepository.AssertCreateOrUpdateHeaderCallCountAndPassedBlockNumbers(3, []int64{1, 2, 3})
})
It("passes transformers for execution on new blocks", func() {
headerRepository := fakes.NewMockHeaderRepository()
headerRepository.SetMissingBlockNumbers([]int64{})
blockChain := fakes.NewMockBlockChain()
blockChain.SetLastBlock(big.NewInt(3))
transformer := fakes.NewMockTransformer()
validator := history.NewHeaderValidator(blockChain, headerRepository, 1, []transformers.Transformer{transformer})
validator.ValidateHeaders()
headerRepository.AssertCreateOrUpdateHeaderCallCountAndPassedBlockNumbers(2, []int64{2, 3})
transformer.AssertExecuteCalledWith(core.Header{BlockNumber: 3}, 0)
})
})
pkg/history/populate_headers.go
+4 -11
View File
@@ -6,40 +6,33 @@ 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/transformers"
)
func PopulateMissingHeaders(blockchain core.BlockChain, headerRepository datastore.HeaderRepository, startingBlockNumber int64, transformers []transformers.Transformer) (int, error) {
func PopulateMissingHeaders(blockchain core.BlockChain, headerRepository datastore.HeaderRepository, startingBlockNumber int64) (int, error) {
lastBlock := blockchain.LastBlock().Int64()
blockRange := headerRepository.MissingBlockNumbers(startingBlockNumber, lastBlock, blockchain.Node().ID)
log.SetPrefix("")
log.Printf("Backfilling %d blocks\n\n", len(blockRange))
_, err := RetrieveAndUpdateHeaders(blockchain, headerRepository, blockRange, transformers)
_, err := RetrieveAndUpdateHeaders(blockchain, headerRepository, blockRange)
if err != nil {
return 0, err
}
return len(blockRange), nil
}
func RetrieveAndUpdateHeaders(chain core.BlockChain, headerRepository datastore.HeaderRepository, blockNumbers []int64, transformers []transformers.Transformer) (int, error) {
func RetrieveAndUpdateHeaders(chain core.BlockChain, headerRepository datastore.HeaderRepository, blockNumbers []int64) (int, error) {
for _, blockNumber := range blockNumbers {
header, err := chain.GetHeaderByNumber(blockNumber)
if err != nil {
return 0, err
}
id, err := headerRepository.CreateOrUpdateHeader(header)
_, err = headerRepository.CreateOrUpdateHeader(header)
if err != nil {
if err == repositories.ErrValidHeaderExists {
continue
}
return 0, err
}
for _, transformer := range transformers {
err := transformer.Execute(header, id)
if err != nil {
return 0, err
}
}
}
return len(blockNumbers), nil
}
pkg/history/populate_headers_test.go
+2 -50
View File
@@ -6,11 +6,8 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/history"
"github.com/vulcanize/vulcanizedb/pkg/transformers"
)
var _ = Describe("Populating headers", func() {
@@ -26,7 +23,7 @@ var _ = Describe("Populating headers", func() {
blockChain.SetLastBlock(big.NewInt(2))
headerRepository.SetMissingBlockNumbers([]int64{2})
headersAdded, err := history.PopulateMissingHeaders(blockChain, headerRepository, 1, []transformers.Transformer{})
headersAdded, err := history.PopulateMissingHeaders(blockChain, headerRepository, 1)
Expect(err).NotTo(HaveOccurred())
Expect(headersAdded).To(Equal(1))
@@ -37,54 +34,9 @@ var _ = Describe("Populating headers", func() {
blockChain.SetLastBlock(big.NewInt(2))
headerRepository.SetMissingBlockNumbers([]int64{2})
_, err := history.PopulateMissingHeaders(blockChain, headerRepository, 1, []transformers.Transformer{})
_, err := history.PopulateMissingHeaders(blockChain, headerRepository, 1)
Expect(err).NotTo(HaveOccurred())
headerRepository.AssertCreateOrUpdateHeaderCallCountAndPassedBlockNumbers(1, []int64{2})
})
It("executes passed transformers with created headers", func() {
blockNumber := int64(54321)
blockChain := fakes.NewMockBlockChain()
blockChain.SetLastBlock(big.NewInt(blockNumber))
headerRepository.SetMissingBlockNumbers([]int64{blockNumber})
headerID := int64(12345)
headerRepository.SetCreateOrUpdateHeaderReturnID(headerID)
transformer := fakes.NewMockTransformer()
_, err := history.PopulateMissingHeaders(blockChain, headerRepository, blockNumber, []transformers.Transformer{transformer})
Expect(err).NotTo(HaveOccurred())
transformer.AssertExecuteCalledWith(core.Header{BlockNumber: blockNumber}, headerID)
})
It("does not execute transformer if repository indicates header already exists", func() {
blockNumber := int64(54321)
blockChain := fakes.NewMockBlockChain()
blockChain.SetLastBlock(big.NewInt(blockNumber))
headerRepository.SetMissingBlockNumbers([]int64{blockNumber})
headerRepository.SetCreateOrUpdateHeaderReturnErr(repositories.ErrValidHeaderExists)
transformer := fakes.NewMockTransformer()
_, err := history.PopulateMissingHeaders(blockChain, headerRepository, blockNumber, []transformers.Transformer{transformer})
Expect(err).NotTo(HaveOccurred())
transformer.AssertExecuteNotCalled()
})
It("returns error if executing transformer fails", func() {
blockNumber := int64(54321)
blockChain := fakes.NewMockBlockChain()
blockChain.SetLastBlock(big.NewInt(blockNumber))
headerRepository.SetMissingBlockNumbers([]int64{blockNumber})
headerID := int64(12345)
headerRepository.SetCreateOrUpdateHeaderReturnID(headerID)
transformer := fakes.NewMockTransformer()
transformer.SetExecuteErr(fakes.FakeError)
_, err := history.PopulateMissingHeaders(blockChain, headerRepository, blockNumber, []transformers.Transformer{transformer})
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(fakes.FakeError))
})
})