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

(VDB-298) Consume Pit contract storage diffs

Browse Source 
- Continuously parse storage diffs CSV data to read Pit contract state
- Convert ilks in database to raw bytes32 value for use in generating
  storage keys dynamically
- Persist storage diffs with block number and hash for validation
This commit is contained in:
Rob Mulholand
2019-02-06 10:31:46 -06:00
parent c474933432
commit 867f92c431
85 changed files with 2143 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libraries/shared/watcher.go → libraries/shared/event_watcher.go
+5 -5
View File
@@ -25,7 +25,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
)
type Watcher struct {
type EventWatcher struct {
Transformers []shared.Transformer
DB *postgres.DB
Fetcher shared.LogFetcher
@@ -35,10 +35,10 @@ type Watcher struct {
StartingBlock *int64
}
func NewWatcher(db *postgres.DB, bc core.BlockChain) Watcher {
func NewEventWatcher(db *postgres.DB, bc core.BlockChain) EventWatcher {
chunker := shared.NewLogChunker()
fetcher := shared.NewFetcher(bc)
return Watcher{
return EventWatcher{
DB: db,
Fetcher: fetcher,
Chunker: chunker,
@@ -46,7 +46,7 @@ func NewWatcher(db *postgres.DB, bc core.BlockChain) Watcher {
}
// Adds transformers to the watcher and updates the chunker, so that it will consider the new transformers.
func (watcher *Watcher) AddTransformers(initializers []shared.TransformerInitializer) {
func (watcher *EventWatcher) AddTransformers(initializers []shared.TransformerInitializer) {
var contractAddresses []common.Address
var topic0s []common.Hash
var configs []shared.TransformerConfig
@@ -74,7 +74,7 @@ func (watcher *Watcher) AddTransformers(initializers []shared.TransformerInitial
watcher.Chunker.AddConfigs(configs)
}
func (watcher *Watcher) Execute() error {
func (watcher *EventWatcher) Execute() error {
if watcher.Transformers == nil {
return fmt.Errorf("No transformers added to watcher")
}
libraries/shared/watcher_test.go → libraries/shared/event_watcher_test.go
+9 -9
View File
@@ -35,12 +35,12 @@ import (
"github.com/vulcanize/vulcanizedb/test_config"
)
var _ = Describe("Watcher", func() {
var _ = Describe("EventWatcher", func() {
It("initialises correctly", func() {
db := test_config.NewTestDB(core.Node{ID: "testNode"})
bc := fakes.NewMockBlockChain()
watcher := shared.NewWatcher(db, bc)
watcher := shared.NewEventWatcher(db, bc)
Expect(watcher.DB).To(Equal(db))
Expect(watcher.Fetcher).NotTo(BeNil())
@@ -48,7 +48,7 @@ var _ = Describe("Watcher", func() {
})
It("adds transformers", func() {
watcher := shared.NewWatcher(nil, nil)
watcher := shared.NewEventWatcher(nil, nil)
fakeTransformer := &mocks.MockTransformer{}
fakeTransformer.SetTransformerConfig(mocks.FakeTransformerConfig)
watcher.AddTransformers([]shared2.TransformerInitializer{fakeTransformer.FakeTransformerInitializer})
@@ -60,7 +60,7 @@ var _ = Describe("Watcher", func() {
})
It("adds transformers from multiple sources", func() {
watcher := shared.NewWatcher(nil, nil)
watcher := shared.NewEventWatcher(nil, nil)
fakeTransformer1 := &mocks.MockTransformer{}
fakeTransformer1.SetTransformerConfig(mocks.FakeTransformerConfig)
@@ -84,7 +84,7 @@ var _ = Describe("Watcher", func() {
fakeTransformer2 := &mocks.MockTransformer{}
fakeTransformer2.SetTransformerConfig(shared2.TransformerConfig{StartingBlockNumber: 3})
watcher := shared.NewWatcher(nil, nil)
watcher := shared.NewEventWatcher(nil, nil)
watcher.AddTransformers([]shared2.TransformerInitializer{
fakeTransformer1.FakeTransformerInitializer,
fakeTransformer2.FakeTransformerInitializer,
@@ -94,7 +94,7 @@ var _ = Describe("Watcher", func() {
})
It("returns an error when run without transformers", func() {
watcher := shared.NewWatcher(nil, nil)
watcher := shared.NewEventWatcher(nil, nil)
err := watcher.Execute()
Expect(err).To(MatchError("No transformers added to watcher"))
})
@@ -102,7 +102,7 @@ var _ = Describe("Watcher", func() {
Describe("with missing headers", func() {
var (
db *postgres.DB
watcher shared.Watcher
watcher shared.EventWatcher
mockBlockChain fakes.MockBlockChain
headerRepository repositories.HeaderRepository
repository mocks.MockWatcherRepository
@@ -117,7 +117,7 @@ var _ = Describe("Watcher", func() {
Expect(err).NotTo(HaveOccurred())
repository = mocks.MockWatcherRepository{}
watcher = shared.NewWatcher(db, &mockBlockChain)
watcher = shared.NewEventWatcher(db, &mockBlockChain)
})
It("executes each transformer", func() {
@@ -163,7 +163,7 @@ var _ = Describe("Watcher", func() {
mockBlockChain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{logA, logB})
repository.SetMissingHeaders([]core.Header{fakes.FakeHeader})
watcher = shared.NewWatcher(db, &mockBlockChain)
watcher = shared.NewEventWatcher(db, &mockBlockChain)
watcher.AddTransformers([]shared2.TransformerInitializer{
transformerA.FakeTransformerInitializer, transformerB.FakeTransformerInitializer})
libraries/shared/shared_suite_test.go
+1 -1
View File
@@ -17,8 +17,8 @@
package shared_test
import (
log "github.com/sirupsen/logrus"
"io/ioutil"
"log"
"testing"
. "github.com/onsi/ginkgo"
libraries/shared/storage_watcher.go
+74
View File
@@ -0,0 +1,74 @@
// VulcanizeDB
// Copyright © 2018 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 shared
import (
"github.com/ethereum/go-ethereum/common"
"github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/transformers/shared/storage"
"github.com/vulcanize/vulcanizedb/pkg/transformers/storage_diffs/shared"
"strings"
"github.com/vulcanize/vulcanizedb/pkg/fs"
)
type StorageWatcher struct {
db *postgres.DB
tailer fs.Tailer
Transformers map[common.Address]storage.Transformer
}
func NewStorageWatcher(tailer fs.Tailer, db *postgres.DB) StorageWatcher {
transformers := make(map[common.Address]storage.Transformer)
return StorageWatcher{
db: db,
tailer: tailer,
Transformers: transformers,
}
}
func (watcher StorageWatcher) AddTransformers(initializers []storage.TransformerInitializer) {
for _, initializer := range initializers {
transformer := initializer(watcher.db)
watcher.Transformers[transformer.ContractAddress()] = transformer
}
}
func (watcher StorageWatcher) Execute() error {
t, tailErr := watcher.tailer.Tail()
if tailErr != nil {
return tailErr
}
for line := range t.Lines {
row, parseErr := shared.FromStrings(strings.Split(line.Text, ","))
if parseErr != nil {
return parseErr
}
transformer, ok := watcher.Transformers[row.Contract]
if !ok {
logrus.Warn(shared.ErrContractNotFound{Contract: row.Contract.Hex()}.Error())
continue
}
executeErr := transformer.Execute(row)
if executeErr != nil {
logrus.Warn(executeErr.Error())
continue
}
}
return nil
}
libraries/shared/storage_watcher_test.go
+175
View File
@@ -0,0 +1,175 @@
// VulcanizeDB
// Copyright © 2018 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 shared_test
import (
"errors"
"fmt"
"io/ioutil"
"os"
"strings"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/hpcloud/tail"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/libraries/shared"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/transformers/shared/storage"
shared2 "github.com/vulcanize/vulcanizedb/pkg/transformers/storage_diffs/shared"
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data/mocks"
"github.com/vulcanize/vulcanizedb/test_config"
)
var _ = Describe("Storage Watcher", func() {
It("adds transformers", func() {
fakeAddress := common.HexToAddress("0x12345")
fakeTransformer := &mocks.MockStorageTransformer{Address: fakeAddress}
watcher := shared.NewStorageWatcher(&fakes.MockTailer{}, test_config.NewTestDB(core.Node{}))
watcher.AddTransformers([]storage.TransformerInitializer{fakeTransformer.FakeTransformerInitializer})
Expect(watcher.Transformers[fakeAddress]).To(Equal(fakeTransformer))
})
It("reads the tail of the storage diffs file", func() {
mockTailer := fakes.NewMockTailer()
watcher := shared.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
assert(func(err error) {
Expect(err).To(BeNil())
Expect(mockTailer.TailCalled).To(BeTrue())
}, watcher, mockTailer, []*tail.Line{})
})
It("returns error if row parsing fails", func() {
mockTailer := fakes.NewMockTailer()
watcher := shared.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
line := &tail.Line{Text: "oops"}
assert(func(err error) {
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(shared2.ErrRowMalformed{Length: 1}))
}, watcher, mockTailer, []*tail.Line{line})
})
It("logs error if no transformer can parse storage row", func() {
mockTailer := fakes.NewMockTailer()
line := &tail.Line{
Text: "12345,block_hash,123,storage_key,storage_value",
Time: time.Time{},
Err: nil,
}
watcher := shared.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
tempFile, err := ioutil.TempFile("", "log")
defer os.Remove(tempFile.Name())
Expect(err).NotTo(HaveOccurred())
logrus.SetOutput(tempFile)
assert(func(err error) {
Expect(err).NotTo(HaveOccurred())
logContent, readErr := ioutil.ReadFile(tempFile.Name())
Expect(readErr).NotTo(HaveOccurred())
Expect(string(logContent)).To(ContainSubstring(shared2.ErrContractNotFound{Contract: common.HexToAddress("0x12345").Hex()}.Error()))
}, watcher, mockTailer, []*tail.Line{line})
})
It("executes transformer with storage row", func() {
address := []byte{1, 2, 3}
blockHash := []byte{4, 5, 6}
blockHeight := int64(789)
storageKey := []byte{9, 8, 7}
storageValue := []byte{6, 5, 4}
mockTailer := fakes.NewMockTailer()
line := &tail.Line{
Text: fmt.Sprintf("%s,%s,%d,%s,%s", common.Bytes2Hex(address), common.Bytes2Hex(blockHash), blockHeight, common.Bytes2Hex(storageKey), common.Bytes2Hex(storageValue)),
Time: time.Time{},
Err: nil,
}
watcher := shared.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
fakeTransformer := &mocks.MockStorageTransformer{Address: common.BytesToAddress(address)}
watcher.AddTransformers([]storage.TransformerInitializer{fakeTransformer.FakeTransformerInitializer})
assert(func(err error) {
Expect(err).To(BeNil())
expectedRow, err := shared2.FromStrings(strings.Split(line.Text, ","))
Expect(err).NotTo(HaveOccurred())
Expect(fakeTransformer.PassedRow).To(Equal(expectedRow))
}, watcher, mockTailer, []*tail.Line{line})
})
It("logs error if executing transformer fails", func() {
address := []byte{1, 2, 3}
blockHash := []byte{4, 5, 6}
blockHeight := int64(789)
storageKey := []byte{9, 8, 7}
storageValue := []byte{6, 5, 4}
mockTailer := fakes.NewMockTailer()
line := &tail.Line{
Text: fmt.Sprintf("%s,%s,%d,%s,%s", common.Bytes2Hex(address), common.Bytes2Hex(blockHash), blockHeight, common.Bytes2Hex(storageKey), common.Bytes2Hex(storageValue)),
Time: time.Time{},
Err: nil,
}
watcher := shared.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
executionError := errors.New("storage watcher failed attempting to execute transformer")
fakeTransformer := &mocks.MockStorageTransformer{Address: common.BytesToAddress(address), ExecuteErr: executionError}
watcher.AddTransformers([]storage.TransformerInitializer{fakeTransformer.FakeTransformerInitializer})
tempFile, err := ioutil.TempFile("", "log")
defer os.Remove(tempFile.Name())
Expect(err).NotTo(HaveOccurred())
logrus.SetOutput(tempFile)
assert(func(err error) {
Expect(err).NotTo(HaveOccurred())
logContent, readErr := ioutil.ReadFile(tempFile.Name())
Expect(readErr).NotTo(HaveOccurred())
Expect(string(logContent)).To(ContainSubstring(executionError.Error()))
}, watcher, mockTailer, []*tail.Line{line})
})
})
func assert(assertion func(err error), watcher shared.StorageWatcher, mockTailer *fakes.MockTailer, lines []*tail.Line) {
errs := make(chan error, 1)
done := make(chan bool, 1)
go execute(watcher, mockTailer, errs, done)
for _, line := range lines {
mockTailer.Lines <- line
}
close(mockTailer.Lines)
select {
case err := <-errs:
assertion(err)
break
case <-done:
assertion(nil)
break
}
}
func execute(watcher shared.StorageWatcher, tailer *fakes.MockTailer, errs chan error, done chan bool) {
err := watcher.Execute()
if err != nil {
errs <- err
} else {
done <- true
}
}