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

Handle different contract address format for geth vs csv

Browse Source
This commit is contained in:
Elizabeth Engelman 2019-07-10 14:34:14 -05:00
parent ee244ac6f5
commit 8c4a4d6587
2 changed files with 280 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
libraries/shared/watcher/storage_watcher.go
View File

@ -18,6 +18,7 @@ package watcher
import ( import (
"fmt" "fmt"
"github.com/ethereum/go-ethereum/crypto"
"reflect" "reflect"
"time" "time"
@ -33,6 +34,7 @@ import (
type StorageWatcher struct { type StorageWatcher struct {
db *postgres.DB db *postgres.DB
diffSource string
StorageFetcher fetcher.IStorageFetcher StorageFetcher fetcher.IStorageFetcher
Queue storage.IStorageQueue Queue storage.IStorageQueue
Transformers map[common.Address]transformer.StorageTransformer Transformers map[common.Address]transformer.StorageTransformer
@ -43,12 +45,17 @@ func NewStorageWatcher(fetcher fetcher.IStorageFetcher, db *postgres.DB) Storage
queue := storage.NewStorageQueue(db) queue := storage.NewStorageQueue(db)
return StorageWatcher{ return StorageWatcher{
db: db, db: db,
diffSource: "csv",
StorageFetcher: fetcher, StorageFetcher: fetcher,
Queue: queue, Queue: queue,
Transformers: transformers, Transformers: transformers,
} }
} }
func (storageWatcher *StorageWatcher) SetStorageDiffSource(source string) {
storageWatcher.diffSource = source
}
func (storageWatcher StorageWatcher) AddTransformers(initializers []transformer.StorageTransformerInitializer) { func (storageWatcher StorageWatcher) AddTransformers(initializers []transformer.StorageTransformerInitializer) {
for _, initializer := range initializers { for _, initializer := range initializers {
storageTransformer := initializer(storageWatcher.db) storageTransformer := initializer(storageWatcher.db)
@ -71,10 +78,29 @@ func (storageWatcher StorageWatcher) Execute(rows chan utils.StorageDiffRow, err
} }
} }
func (storageWatcher StorageWatcher) getTransformer(contractAddress common.Address) (transformer.StorageTransformer, bool) {
if storageWatcher.diffSource == "csv" {
storageTransformer, ok := storageWatcher.Transformers[contractAddress]
return storageTransformer, ok
} else if storageWatcher.diffSource == "geth" {
logrus.Debug("number of transformers", len(storageWatcher.Transformers))
for address, t := range storageWatcher.Transformers {
keccakOfTransformerAddress := common.BytesToAddress(crypto.Keccak256(address[:]))
if keccakOfTransformerAddress == contractAddress {
return t, true
}
}
return nil, false
}
return nil, false
}
func (storageWatcher StorageWatcher) processRow(row utils.StorageDiffRow) { func (storageWatcher StorageWatcher) processRow(row utils.StorageDiffRow) {
storageTransformer, ok := storageWatcher.Transformers[row.Contract] storageTransformer, ok := storageWatcher.getTransformer(row.Contract)
if !ok { if !ok {
// ignore rows from unwatched contracts logrus.Debug("ignoring a row from an unwatched contract")
return return
} }
executeErr := storageTransformer.Execute(row) executeErr := storageTransformer.Execute(row)
@ -93,7 +119,7 @@ func (storageWatcher StorageWatcher) processQueue() {
logrus.Warn(fmt.Sprintf("error getting queued storage: %s", fetchErr)) logrus.Warn(fmt.Sprintf("error getting queued storage: %s", fetchErr))
} }
for _, row := range rows { for _, row := range rows {
storageTransformer, ok := storageWatcher.Transformers[row.Contract] storageTransformer, ok := storageWatcher.getTransformer(row.Contract)
if !ok { if !ok {
// delete row from queue if address no longer watched // delete row from queue if address no longer watched
storageWatcher.deleteRow(row.Id) storageWatcher.deleteRow(row.Id)

371
libraries/shared/watcher/storage_watcher_test.go
View File

@ -17,6 +17,7 @@
package watcher_test package watcher_test
import ( import (
"github.com/ethereum/go-ethereum/crypto"
"io/ioutil" "io/ioutil"
"os" "os"
"time" "time"
@ -51,7 +52,8 @@ var _ = Describe("Storage Watcher", func() {
mockFetcher *mocks.MockStorageFetcher mockFetcher *mocks.MockStorageFetcher
mockQueue *mocks.MockStorageQueue mockQueue *mocks.MockStorageQueue
mockTransformer *mocks.MockStorageTransformer mockTransformer *mocks.MockStorageTransformer
row utils.StorageDiffRow csvRow utils.StorageDiffRow
gethRow utils.StorageDiffRow
rows chan utils.StorageDiffRow rows chan utils.StorageDiffRow
storageWatcher watcher.StorageWatcher storageWatcher watcher.StorageWatcher
) )
@ -63,7 +65,7 @@ var _ = Describe("Storage Watcher", func() {
mockFetcher = mocks.NewMockStorageFetcher() mockFetcher = mocks.NewMockStorageFetcher()
mockQueue = &mocks.MockStorageQueue{} mockQueue = &mocks.MockStorageQueue{}
mockTransformer = &mocks.MockStorageTransformer{Address: address} mockTransformer = &mocks.MockStorageTransformer{Address: address}
row = utils.StorageDiffRow{ csvRow = utils.StorageDiffRow{
Id: 1337, Id: 1337,
Contract: address, Contract: address,
BlockHash: common.HexToHash("0xfedcba9876543210"), BlockHash: common.HexToHash("0xfedcba9876543210"),
@ -71,6 +73,14 @@ var _ = Describe("Storage Watcher", func() {
StorageKey: common.HexToHash("0xabcdef1234567890"), StorageKey: common.HexToHash("0xabcdef1234567890"),
StorageValue: common.HexToHash("0x9876543210abcdef"), StorageValue: common.HexToHash("0x9876543210abcdef"),
} }
gethRow = utils.StorageDiffRow{
Id: 1338,
Contract: common.BytesToAddress(crypto.Keccak256(address[:])),
BlockHash: common.HexToHash("0xfedcba9876543210"),
BlockHeight: 0,
StorageKey: common.HexToHash("0xabcdef1234567890"),
StorageValue: common.HexToHash("0x9876543210abcdef"),
}
}) })
It("logs error if fetching storage diffs fails", func(done Done) { It("logs error if fetching storage diffs fails", func(done Done) {
@ -92,153 +102,274 @@ var _ = Describe("Storage Watcher", func() {
close(done) close(done)
}) })
Describe("transforming new storage diffs", func() { Describe("transforming new storage diffs from csv", func() {
BeforeEach(func() { Describe("where diff source is a csv file", func() {
mockFetcher.RowsToReturn = []utils.StorageDiffRow{row} BeforeEach(func() {
storageWatcher = watcher.NewStorageWatcher(mockFetcher, test_config.NewTestDB(test_config.NewTestNode())) mockFetcher.RowsToReturn = []utils.StorageDiffRow{csvRow}
storageWatcher.Queue = mockQueue storageWatcher = watcher.NewStorageWatcher(mockFetcher, test_config.NewTestDB(test_config.NewTestNode()))
storageWatcher.AddTransformers([]transformer.StorageTransformerInitializer{mockTransformer.FakeTransformerInitializer}) storageWatcher.Queue = mockQueue
storageWatcher.AddTransformers([]transformer.StorageTransformerInitializer{mockTransformer.FakeTransformerInitializer})
})
It("executes transformer for recognized storage row", func(done Done) {
go storageWatcher.Execute(rows, errs, time.Hour)
Eventually(func() utils.StorageDiffRow {
return mockTransformer.PassedRow
}).Should(Equal(csvRow))
close(done)
})
It("queues row for later processing if transformer execution fails", func(done Done) {
mockTransformer.ExecuteErr = fakes.FakeError
go storageWatcher.Execute(rows, errs, time.Hour)
Expect(<-errs).To(BeNil())
Eventually(func() bool {
return mockQueue.AddCalled
}).Should(BeTrue())
Eventually(func() utils.StorageDiffRow {
return mockQueue.AddPassedRow
}).Should(Equal(csvRow))
close(done)
})
It("logs error if queueing row fails", func(done Done) {
mockTransformer.ExecuteErr = utils.ErrStorageKeyNotFound{}
mockQueue.AddError = fakes.FakeError
tempFile, fileErr := ioutil.TempFile("", "log")
Expect(fileErr).NotTo(HaveOccurred())
defer os.Remove(tempFile.Name())
logrus.SetOutput(tempFile)
go storageWatcher.Execute(rows, errs, time.Hour)
Eventually(func() bool {
return mockQueue.AddCalled
}).Should(BeTrue())
Eventually(func() (string, error) {
logContent, err := ioutil.ReadFile(tempFile.Name())
return string(logContent), err
}).Should(ContainSubstring(fakes.FakeError.Error()))
close(done)
})
}) })
It("executes transformer for recognized storage row", func(done Done) { Describe("where diff source is geth RPC pub sub", func() {
go storageWatcher.Execute(rows, errs, time.Hour) BeforeEach(func() {
mockFetcher.RowsToReturn = []utils.StorageDiffRow{gethRow}
storageWatcher = watcher.NewStorageWatcher(mockFetcher, test_config.NewTestDB(test_config.NewTestNode()))
storageWatcher.SetStorageDiffSource("geth")
storageWatcher.Queue = mockQueue
storageWatcher.AddTransformers([]transformer.StorageTransformerInitializer{mockTransformer.FakeTransformerInitializer})
})
Eventually(func() utils.StorageDiffRow { It("executes transformer for recognized storage row", func(done Done) {
return mockTransformer.PassedRow go storageWatcher.Execute(rows, errs, time.Hour)
}).Should(Equal(row))
close(done)
})
It("queues row for later processing if transformer execution fails", func(done Done) { Eventually(func() utils.StorageDiffRow {
mockTransformer.ExecuteErr = fakes.FakeError return mockTransformer.PassedRow
}).Should(Equal(gethRow))
close(done)
})
go storageWatcher.Execute(rows, errs, time.Hour) It("queues row for later processing if transformer execution fails", func(done Done) {
mockTransformer.ExecuteErr = fakes.FakeError
Expect(<-errs).To(BeNil()) go storageWatcher.Execute(rows, errs, time.Hour)
Eventually(func() bool {
return mockQueue.AddCalled
}).Should(BeTrue())
Eventually(func() utils.StorageDiffRow {
return mockQueue.AddPassedRow
}).Should(Equal(row))
close(done)
})
It("logs error if queueing row fails", func(done Done) { Expect(<-errs).To(BeNil())
mockTransformer.ExecuteErr = utils.ErrStorageKeyNotFound{} Eventually(func() bool {
mockQueue.AddError = fakes.FakeError return mockQueue.AddCalled
tempFile, fileErr := ioutil.TempFile("", "log") }).Should(BeTrue())
Expect(fileErr).NotTo(HaveOccurred()) Eventually(func() utils.StorageDiffRow {
defer os.Remove(tempFile.Name()) return mockQueue.AddPassedRow
logrus.SetOutput(tempFile) }).Should(Equal(gethRow))
close(done)
})
go storageWatcher.Execute(rows, errs, time.Hour) It("logs error if queueing row fails", func(done Done) {
mockTransformer.ExecuteErr = utils.ErrStorageKeyNotFound{}
mockQueue.AddError = fakes.FakeError
tempFile, fileErr := ioutil.TempFile("", "log")
Expect(fileErr).NotTo(HaveOccurred())
defer os.Remove(tempFile.Name())
logrus.SetOutput(tempFile)
Eventually(func() bool { go storageWatcher.Execute(rows, errs, time.Hour)
return mockQueue.AddCalled
}).Should(BeTrue()) Eventually(func() bool {
Eventually(func() (string, error) { return mockQueue.AddCalled
logContent, err := ioutil.ReadFile(tempFile.Name()) }).Should(BeTrue())
return string(logContent), err Eventually(func() (string, error) {
}).Should(ContainSubstring(fakes.FakeError.Error())) logContent, err := ioutil.ReadFile(tempFile.Name())
close(done) return string(logContent), err
}).Should(ContainSubstring(fakes.FakeError.Error()))
close(done)
})
}) })
}) })
Describe("transforming queued storage diffs", func() { Describe("transforming queued storage diffs", func() {
BeforeEach(func() { Describe("where diff source is a csv file", func() {
mockQueue.RowsToReturn = []utils.StorageDiffRow{row} BeforeEach(func() {
storageWatcher = watcher.NewStorageWatcher(mockFetcher, test_config.NewTestDB(test_config.NewTestNode())) mockQueue.RowsToReturn = []utils.StorageDiffRow{csvRow}
storageWatcher.Queue = mockQueue storageWatcher = watcher.NewStorageWatcher(mockFetcher, test_config.NewTestDB(test_config.NewTestNode()))
storageWatcher.AddTransformers([]transformer.StorageTransformerInitializer{mockTransformer.FakeTransformerInitializer}) storageWatcher.Queue = mockQueue
storageWatcher.AddTransformers([]transformer.StorageTransformerInitializer{mockTransformer.FakeTransformerInitializer})
})
It("executes transformer for storage row", func(done Done) {
go storageWatcher.Execute(rows, errs, time.Nanosecond)
Eventually(func() utils.StorageDiffRow {
return mockTransformer.PassedRow
}).Should(Equal(csvRow))
close(done)
})
It("deletes row from queue if transformer execution successful", func(done Done) {
go storageWatcher.Execute(rows, errs, time.Nanosecond)
Eventually(func() int {
return mockQueue.DeletePassedId
}).Should(Equal(csvRow.Id))
close(done)
})
It("logs error if deleting persisted row fails", func(done Done) {
mockQueue.DeleteErr = fakes.FakeError
tempFile, fileErr := ioutil.TempFile("", "log")
Expect(fileErr).NotTo(HaveOccurred())
defer os.Remove(tempFile.Name())
logrus.SetOutput(tempFile)
go storageWatcher.Execute(rows, errs, time.Nanosecond)
Eventually(func() (string, error) {
logContent, err := ioutil.ReadFile(tempFile.Name())
return string(logContent), err
}).Should(ContainSubstring(fakes.FakeError.Error()))
close(done)
})
It("deletes obsolete row from queue if contract not recognized", func(done Done) {
obsoleteRow := utils.StorageDiffRow{
Id: csvRow.Id + 1,
Contract: common.HexToAddress("0xfedcba9876543210"),
}
mockQueue.RowsToReturn = []utils.StorageDiffRow{obsoleteRow}
go storageWatcher.Execute(rows, errs, time.Nanosecond)
Eventually(func() int {
return mockQueue.DeletePassedId
}).Should(Equal(obsoleteRow.Id))
close(done)
})
It("logs error if deleting obsolete row fails", func(done Done) {
obsoleteRow := utils.StorageDiffRow{
Id: csvRow.Id + 1,
Contract: common.HexToAddress("0xfedcba9876543210"),
}
mockQueue.RowsToReturn = []utils.StorageDiffRow{obsoleteRow}
mockQueue.DeleteErr = fakes.FakeError
tempFile, fileErr := ioutil.TempFile("", "log")
Expect(fileErr).NotTo(HaveOccurred())
defer os.Remove(tempFile.Name())
logrus.SetOutput(tempFile)
go storageWatcher.Execute(rows, errs, time.Nanosecond)
Eventually(func() (string, error) {
logContent, err := ioutil.ReadFile(tempFile.Name())
return string(logContent), err
}).Should(ContainSubstring(fakes.FakeError.Error()))
close(done)
})
}) })
It("logs error if getting queued storage fails", func(done Done) { Describe("where diff source is geth RPC pub sub", func() {
mockQueue.GetAllErr = fakes.FakeError BeforeEach(func() {
tempFile, fileErr := ioutil.TempFile("", "log") mockQueue.RowsToReturn = []utils.StorageDiffRow{gethRow}
Expect(fileErr).NotTo(HaveOccurred()) storageWatcher = watcher.NewStorageWatcher(mockFetcher, test_config.NewTestDB(test_config.NewTestNode()))
defer os.Remove(tempFile.Name()) storageWatcher.Queue = mockQueue
logrus.SetOutput(tempFile) storageWatcher.SetStorageDiffSource("geth")
storageWatcher.AddTransformers([]transformer.StorageTransformerInitializer{mockTransformer.FakeTransformerInitializer})
})
go storageWatcher.Execute(rows, errs, time.Nanosecond) It("executes transformer for storage row", func(done Done) {
go storageWatcher.Execute(rows, errs, time.Nanosecond)
Eventually(func() (string, error) { Eventually(func() utils.StorageDiffRow {
logContent, err := ioutil.ReadFile(tempFile.Name()) return mockTransformer.PassedRow
return string(logContent), err }).Should(Equal(gethRow))
}).Should(ContainSubstring(fakes.FakeError.Error())) close(done)
close(done) })
})
It("executes transformer for storage row", func(done Done) { It("deletes row from queue if transformer execution successful", func(done Done) {
go storageWatcher.Execute(rows, errs, time.Nanosecond) go storageWatcher.Execute(rows, errs, time.Nanosecond)
Eventually(func() utils.StorageDiffRow { Eventually(func() int {
return mockTransformer.PassedRow return mockQueue.DeletePassedId
}).Should(Equal(row)) }).Should(Equal(gethRow.Id))
close(done) close(done)
}) })
It("deletes row from queue if transformer execution successful", func(done Done) { It("logs error if deleting persisted row fails", func(done Done) {
go storageWatcher.Execute(rows, errs, time.Nanosecond) mockQueue.DeleteErr = fakes.FakeError
tempFile, fileErr := ioutil.TempFile("", "log")
Expect(fileErr).NotTo(HaveOccurred())
defer os.Remove(tempFile.Name())
logrus.SetOutput(tempFile)
Eventually(func() int { go storageWatcher.Execute(rows, errs, time.Nanosecond)
return mockQueue.DeletePassedId
}).Should(Equal(row.Id))
close(done)
})
It("logs error if deleting persisted row fails", func(done Done) { Eventually(func() (string, error) {
mockQueue.DeleteErr = fakes.FakeError logContent, err := ioutil.ReadFile(tempFile.Name())
tempFile, fileErr := ioutil.TempFile("", "log") return string(logContent), err
Expect(fileErr).NotTo(HaveOccurred()) }).Should(ContainSubstring(fakes.FakeError.Error()))
defer os.Remove(tempFile.Name()) close(done)
logrus.SetOutput(tempFile) })
go storageWatcher.Execute(rows, errs, time.Nanosecond) It("deletes obsolete row from queue if contract not recognized", func(done Done) {
obsoleteRow := utils.StorageDiffRow{
Id: gethRow.Id + 1,
Contract: common.HexToAddress("0xfedcba9876543210"),
}
mockQueue.RowsToReturn = []utils.StorageDiffRow{obsoleteRow}
Eventually(func() (string, error) { go storageWatcher.Execute(rows, errs, time.Nanosecond)
logContent, err := ioutil.ReadFile(tempFile.Name())
return string(logContent), err
}).Should(ContainSubstring(fakes.FakeError.Error()))
close(done)
})
It("deletes obsolete row from queue if contract not recognized", func(done Done) { Eventually(func() int {
obsoleteRow := utils.StorageDiffRow{ return mockQueue.DeletePassedId
Id: row.Id + 1, }).Should(Equal(obsoleteRow.Id))
Contract: common.HexToAddress("0xfedcba9876543210"), close(done)
} })
mockQueue.RowsToReturn = []utils.StorageDiffRow{obsoleteRow}
go storageWatcher.Execute(rows, errs, time.Nanosecond) It("logs error if deleting obsolete row fails", func(done Done) {
obsoleteRow := utils.StorageDiffRow{
Id: gethRow.Id + 1,
Contract: common.HexToAddress("0xfedcba9876543210"),
}
mockQueue.RowsToReturn = []utils.StorageDiffRow{obsoleteRow}
mockQueue.DeleteErr = fakes.FakeError
tempFile, fileErr := ioutil.TempFile("", "log")
Expect(fileErr).NotTo(HaveOccurred())
defer os.Remove(tempFile.Name())
logrus.SetOutput(tempFile)
Eventually(func() int { go storageWatcher.Execute(rows, errs, time.Nanosecond)
return mockQueue.DeletePassedId
}).Should(Equal(obsoleteRow.Id))
close(done)
})
It("logs error if deleting obsolete row fails", func(done Done) { Eventually(func() (string, error) {
obsoleteRow := utils.StorageDiffRow{ logContent, err := ioutil.ReadFile(tempFile.Name())
Id: row.Id + 1, return string(logContent), err
Contract: common.HexToAddress("0xfedcba9876543210"), }).Should(ContainSubstring(fakes.FakeError.Error()))
} close(done)
mockQueue.RowsToReturn = []utils.StorageDiffRow{obsoleteRow} })
mockQueue.DeleteErr = fakes.FakeError
tempFile, fileErr := ioutil.TempFile("", "log")
Expect(fileErr).NotTo(HaveOccurred())
defer os.Remove(tempFile.Name())
logrus.SetOutput(tempFile)
go storageWatcher.Execute(rows, errs, time.Nanosecond)
Eventually(func() (string, error) {
logContent, err := ioutil.ReadFile(tempFile.Name())
return string(logContent), err
}).Should(ContainSubstring(fakes.FakeError.Error()))
close(done)
}) })
}) })
}) })
}) })