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

(VDB-950) Write raw diffs before transforming

Browse Source 
- Raw field we can reference by FK for related data
- Enables replay for unwatched or mistransformed diffs
This commit is contained in:
Rob Mulholand
2019-12-03 14:51:17 -06:00
committed by Ian Norden
parent 1178940047
commit 56ce8bdb41
45 changed files with 543 additions and 1815 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libraries/shared/storage/backfiller.go
+3 -3
View File
@@ -36,7 +36,7 @@ const (
// BackFiller is the backfilling interface
type BackFiller interface {
BackFill(startingBlock, endingBlock uint64, backFill chan utils.StorageDiff, errChan chan error, done chan bool) error
BackFill(startingBlock, endingBlock uint64, backFill chan utils.StorageDiffInput, errChan chan error, done chan bool) error
}
// backFiller is the backfilling struct
@@ -59,7 +59,7 @@ func NewStorageBackFiller(fetcher fetcher.StateDiffFetcher, batchSize uint64) Ba
// BackFill fetches, processes, and returns utils.StorageDiffs over a range of blocks
// It splits a large range up into smaller chunks, batch fetching and processing those chunks concurrently
func (bf *backFiller) BackFill(startingBlock, endingBlock uint64, backFill chan utils.StorageDiff, errChan chan error, done chan bool) error {
func (bf *backFiller) BackFill(startingBlock, endingBlock uint64, backFill chan utils.StorageDiffInput, errChan chan error, done chan bool) error {
logrus.Infof("going to fill in gap from %d to %d", startingBlock, endingBlock)
// break the range up into bins of smaller ranges
@@ -113,7 +113,7 @@ func (bf *backFiller) BackFill(startingBlock, endingBlock uint64, backFill chan
return nil
}
func (bf *backFiller) backFillRange(blockHeights []uint64, diffChan chan utils.StorageDiff, errChan chan error, doneChan chan [2]uint64) {
func (bf *backFiller) backFillRange(blockHeights []uint64, diffChan chan utils.StorageDiffInput, errChan chan error, doneChan chan [2]uint64) {
payloads, fetchErr := bf.fetcher.FetchStateDiffsAt(blockHeights)
if fetchErr != nil {
errChan <- fetchErr
libraries/shared/storage/backfiller_test.go
+11 -11
View File
@@ -45,7 +45,7 @@ var _ = Describe("BackFiller", func() {
It("batch calls statediff_stateDiffAt", func() {
backFiller = storage.NewStorageBackFiller(mockFetcher, 100)
backFill := make(chan utils.StorageDiff)
backFill := make(chan utils.StorageDiffInput)
done := make(chan bool)
errChan := make(chan error)
backFillInitErr := backFiller.BackFill(
@@ -55,7 +55,7 @@ var _ = Describe("BackFiller", func() {
errChan,
done)
Expect(backFillInitErr).ToNot(HaveOccurred())
var diffs []utils.StorageDiff
var diffs []utils.StorageDiffInput
for {
select {
case diff := <-backFill:
@@ -79,7 +79,7 @@ var _ = Describe("BackFiller", func() {
It("has a configurable batch size", func() {
backFiller = storage.NewStorageBackFiller(mockFetcher, 1)
backFill := make(chan utils.StorageDiff)
backFill := make(chan utils.StorageDiffInput)
done := make(chan bool)
errChan := make(chan error)
backFillInitErr := backFiller.BackFill(
@@ -89,7 +89,7 @@ var _ = Describe("BackFiller", func() {
errChan,
done)
Expect(backFillInitErr).ToNot(HaveOccurred())
var diffs []utils.StorageDiff
var diffs []utils.StorageDiffInput
for {
select {
case diff := <-backFill:
@@ -119,7 +119,7 @@ var _ = Describe("BackFiller", func() {
mockFetcher.PayloadsToReturn = payloadsToReturn
// batch size of 2 with 1001 block range => 501 bins
backFiller = storage.NewStorageBackFiller(mockFetcher, 2)
backFill := make(chan utils.StorageDiff)
backFill := make(chan utils.StorageDiffInput)
done := make(chan bool)
errChan := make(chan error)
backFillInitErr := backFiller.BackFill(
@@ -129,7 +129,7 @@ var _ = Describe("BackFiller", func() {
errChan,
done)
Expect(backFillInitErr).ToNot(HaveOccurred())
var diffs []utils.StorageDiff
var diffs []utils.StorageDiffInput
for {
select {
case diff := <-backFill:
@@ -155,7 +155,7 @@ var _ = Describe("BackFiller", func() {
test_data.BlockNumber.Uint64(): errors.New("mock fetcher error"),
}
backFiller = storage.NewStorageBackFiller(mockFetcher, 1)
backFill := make(chan utils.StorageDiff)
backFill := make(chan utils.StorageDiffInput)
done := make(chan bool)
errChan := make(chan error)
backFillInitErr := backFiller.BackFill(
@@ -166,7 +166,7 @@ var _ = Describe("BackFiller", func() {
done)
Expect(backFillInitErr).ToNot(HaveOccurred())
var numOfErrs int
var diffs []utils.StorageDiff
var diffs []utils.StorageDiffInput
for {
select {
case diff := <-backFill:
@@ -193,7 +193,7 @@ var _ = Describe("BackFiller", func() {
}
mockFetcher.CalledTimes = 0
backFiller = storage.NewStorageBackFiller(mockFetcher, 1)
backFill = make(chan utils.StorageDiff)
backFill = make(chan utils.StorageDiffInput)
done = make(chan bool)
errChan = make(chan error)
backFillInitErr = backFiller.BackFill(
@@ -204,7 +204,7 @@ var _ = Describe("BackFiller", func() {
done)
Expect(backFillInitErr).ToNot(HaveOccurred())
numOfErrs = 0
diffs = []utils.StorageDiff{}
diffs = []utils.StorageDiffInput{}
for {
select {
case diff := <-backFill:
@@ -227,7 +227,7 @@ var _ = Describe("BackFiller", func() {
})
})
func containsDiff(diffs []utils.StorageDiff, diff utils.StorageDiff) bool {
func containsDiff(diffs []utils.StorageDiffInput, diff utils.StorageDiffInput) bool {
for _, d := range diffs {
if d == diff {
return true
libraries/shared/storage/storage_queue.go
+13 -13
View File
@@ -22,9 +22,9 @@ import (
)
type IStorageQueue interface {
Add(diff utils.StorageDiff) error
Delete(id int) error
GetAll() ([]utils.StorageDiff, error)
Add(diff utils.PersistedStorageDiff) error
Delete(id int64) error
GetAll() ([]utils.PersistedStorageDiff, error)
}
type StorageQueue struct {
@@ -35,21 +35,21 @@ func NewStorageQueue(db *postgres.DB) StorageQueue {
return StorageQueue{db: db}
}
func (queue StorageQueue) Add(diff utils.StorageDiff) error {
_, err := queue.db.Exec(`INSERT INTO public.queued_storage (contract,
block_hash, block_height, storage_key, storage_value) VALUES
($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING`, diff.HashedAddress.Bytes(), diff.BlockHash.Bytes(),
diff.BlockHeight, diff.StorageKey.Bytes(), diff.StorageValue.Bytes())
func (queue StorageQueue) Add(diff utils.PersistedStorageDiff) error {
_, err := queue.db.Exec(`INSERT INTO public.queued_storage (diff_id) VALUES
($1) ON CONFLICT DO NOTHING`, diff.ID)
return err
}
func (queue StorageQueue) Delete(id int) error {
_, err := queue.db.Exec(`DELETE FROM public.queued_storage WHERE id = $1`, id)
func (queue StorageQueue) Delete(diffID int64) error {
_, err := queue.db.Exec(`DELETE FROM public.queued_storage WHERE diff_id = $1`, diffID)
return err
}
func (queue StorageQueue) GetAll() ([]utils.StorageDiff, error) {
var result []utils.StorageDiff
err := queue.db.Select(&result, `SELECT * FROM public.queued_storage`)
func (queue StorageQueue) GetAll() ([]utils.PersistedStorageDiff, error) {
var result []utils.PersistedStorageDiff
err := queue.db.Select(&result, `SELECT storage_diff.id, hashed_address, block_height, block_hash, storage_key, storage_value
FROM public.queued_storage
LEFT JOIN public.storage_diff ON queued_storage.diff_id = storage_diff.id`)
return result, err
}
libraries/shared/storage/storage_queue_test.go
+19 -8
View File
@@ -23,19 +23,21 @@ import (
"github.com/vulcanize/vulcanizedb/libraries/shared/storage"
"github.com/vulcanize/vulcanizedb/libraries/shared/storage/utils"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
"github.com/vulcanize/vulcanizedb/test_config"
)
var _ = Describe("Storage queue", func() {
var (
db *postgres.DB
diff utils.StorageDiff
queue storage.IStorageQueue
db *postgres.DB
diff utils.PersistedStorageDiff
diffRepository repositories.StorageDiffRepository
queue storage.IStorageQueue
)
BeforeEach(func() {
fakeAddr := "0x123456"
diff = utils.StorageDiff{
rawDiff := utils.StorageDiffInput{
HashedAddress: utils.HexToKeccak256Hash(fakeAddr),
BlockHash: common.HexToHash("0x678901"),
BlockHeight: 987,
@@ -44,6 +46,10 @@ var _ = Describe("Storage queue", func() {
}
db = test_config.NewTestDB(test_config.NewTestNode())
test_config.CleanTestDB(db)
diffRepository = repositories.NewStorageDiffRepository(db)
diffID, insertDiffErr := diffRepository.CreateStorageDiff(rawDiff)
Expect(insertDiffErr).NotTo(HaveOccurred())
diff = utils.ToPersistedDiff(rawDiff, diffID)
queue = storage.NewStorageQueue(db)
addErr := queue.Add(diff)
Expect(addErr).NotTo(HaveOccurred())
@@ -51,8 +57,10 @@ var _ = Describe("Storage queue", func() {
Describe("Add", func() {
It("adds a storage diff to the db", func() {
var result utils.StorageDiff
getErr := db.Get(&result, `SELECT contract, block_hash, block_height, storage_key, storage_value FROM public.queued_storage`)
var result utils.PersistedStorageDiff
getErr := db.Get(&result, `SELECT storage_diff.id, hashed_address, block_hash, block_height, storage_key, storage_value
FROM public.queued_storage
LEFT JOIN public.storage_diff ON queued_storage.diff_id = storage_diff.id`)
Expect(getErr).NotTo(HaveOccurred())
Expect(result).To(Equal(diff))
})
@@ -82,14 +90,17 @@ var _ = Describe("Storage queue", func() {
It("gets all storage diffs from db", func() {
fakeAddr := "0x234567"
diffTwo := utils.StorageDiff{
diffTwo := utils.StorageDiffInput{
HashedAddress: utils.HexToKeccak256Hash(fakeAddr),
BlockHash: common.HexToHash("0x678902"),
BlockHeight: 988,
StorageKey: common.HexToHash("0x654322"),
StorageValue: common.HexToHash("0x198766"),
}
addErr := queue.Add(diffTwo)
persistedDiffTwoID, insertDiffErr := diffRepository.CreateStorageDiff(diffTwo)
Expect(insertDiffErr).NotTo(HaveOccurred())
persistedDiffTwo := utils.ToPersistedDiff(diffTwo, persistedDiffTwoID)
addErr := queue.Add(persistedDiffTwo)
Expect(addErr).NotTo(HaveOccurred())
diffs, err := queue.GetAll()
libraries/shared/storage/utils/decoder.go
+1 -1
View File
@@ -27,7 +27,7 @@ const (
bitsPerByte = 8
)
func Decode(diff StorageDiff, metadata StorageValueMetadata) (interface{}, error) {
func Decode(diff PersistedStorageDiff, metadata StorageValueMetadata) (interface{}, error) {
switch metadata.Type {
case Uint256:
return decodeInteger(diff.StorageValue.Bytes()), nil
libraries/shared/storage/utils/decoder_test.go
+8 -8
View File
@@ -29,7 +29,7 @@ import (
var _ = Describe("Storage decoder", func() {
It("decodes uint256", func() {
fakeInt := common.HexToHash("0000000000000000000000000000000000000000000000000000000000000539")
diff := utils.StorageDiff{StorageValue: fakeInt}
diff := utils.PersistedStorageDiff{StorageDiffInput: utils.StorageDiffInput{StorageValue: fakeInt}}
metadata := utils.StorageValueMetadata{Type: utils.Uint256}
result, err := utils.Decode(diff, metadata)
@@ -40,7 +40,7 @@ var _ = Describe("Storage decoder", func() {
It("decodes uint128", func() {
fakeInt := common.HexToHash("0000000000000000000000000000000000000000000000000000000000011123")
diff := utils.StorageDiff{StorageValue: fakeInt}
diff := utils.PersistedStorageDiff{StorageDiffInput: utils.StorageDiffInput{StorageValue: fakeInt}}
metadata := utils.StorageValueMetadata{Type: utils.Uint128}
result, err := utils.Decode(diff, metadata)
@@ -51,7 +51,7 @@ var _ = Describe("Storage decoder", func() {
It("decodes uint48", func() {
fakeInt := common.HexToHash("0000000000000000000000000000000000000000000000000000000000000123")
diff := utils.StorageDiff{StorageValue: fakeInt}
diff := utils.PersistedStorageDiff{StorageDiffInput: utils.StorageDiffInput{StorageValue: fakeInt}}
metadata := utils.StorageValueMetadata{Type: utils.Uint48}
result, err := utils.Decode(diff, metadata)
@@ -62,7 +62,7 @@ var _ = Describe("Storage decoder", func() {
It("decodes address", func() {
fakeAddress := common.HexToAddress("0x12345")
diff := utils.StorageDiff{StorageValue: fakeAddress.Hash()}
diff := utils.PersistedStorageDiff{StorageDiffInput: utils.StorageDiffInput{StorageValue: fakeAddress.Hash()}}
metadata := utils.StorageValueMetadata{Type: utils.Address}
result, err := utils.Decode(diff, metadata)
@@ -75,7 +75,7 @@ var _ = Describe("Storage decoder", func() {
It("decodes uint48 items", func() {
//this is a real storage data example
packedStorage := common.HexToHash("000000000000000000000000000000000000000000000002a300000000002a30")
diff := utils.StorageDiff{StorageValue: packedStorage}
diff := utils.PersistedStorageDiff{StorageDiffInput: utils.StorageDiffInput{StorageValue: packedStorage}}
packedTypes := map[int]utils.ValueType{}
packedTypes[0] = utils.Uint48
packedTypes[1] = utils.Uint48
@@ -99,7 +99,7 @@ var _ = Describe("Storage decoder", func() {
packedStorageHex := "0000000A5D1AFFFFFFFFFFFE00000009F3C600000002A300000000002A30"
packedStorage := common.HexToHash(packedStorageHex)
diff := utils.StorageDiff{StorageValue: packedStorage}
diff := utils.PersistedStorageDiff{StorageDiffInput: utils.StorageDiffInput{StorageValue: packedStorage}}
packedTypes := map[int]utils.ValueType{}
packedTypes[0] = utils.Uint48
packedTypes[1] = utils.Uint48
@@ -129,7 +129,7 @@ var _ = Describe("Storage decoder", func() {
packedStorageHex := "000000038D7EA4C67FF8E502B6730000" +
"0000000000000000AB54A98CEB1F0AD2"
packedStorage := common.HexToHash(packedStorageHex)
diff := utils.StorageDiff{StorageValue: packedStorage}
diff := utils.PersistedStorageDiff{StorageDiffInput: utils.StorageDiffInput{StorageValue: packedStorage}}
packedTypes := map[int]utils.ValueType{}
packedTypes[0] = utils.Uint128
packedTypes[1] = utils.Uint128
@@ -151,7 +151,7 @@ var _ = Describe("Storage decoder", func() {
//TODO: replace with real data when available
addressHex := "0000000000000000000000000000000000012345"
packedStorage := common.HexToHash("00000002a300" + "000000002a30" + addressHex)
row := utils.StorageDiff{StorageValue: packedStorage}
row := utils.PersistedStorageDiff{StorageDiffInput: utils.StorageDiffInput{StorageValue: packedStorage}}
packedTypes := map[int]utils.ValueType{}
packedTypes[0] = utils.Address
packedTypes[1] = utils.Uint48
libraries/shared/storage/utils/diff.go
+21 -10
View File
@@ -27,24 +27,28 @@ import (
const ExpectedRowLength = 5
type StorageDiff struct {
ID int
HashedAddress common.Hash `db:"contract"`
type StorageDiffInput struct {
HashedAddress common.Hash `db:"hashed_address"`
BlockHash common.Hash `db:"block_hash"`
BlockHeight int `db:"block_height"`
StorageKey common.Hash `db:"storage_key"`
StorageValue common.Hash `db:"storage_value"`
}
func FromParityCsvRow(csvRow []string) (StorageDiff, error) {
type PersistedStorageDiff struct {
StorageDiffInput
ID int64
}
func FromParityCsvRow(csvRow []string) (StorageDiffInput, error) {
if len(csvRow) != ExpectedRowLength {
return StorageDiff{}, ErrRowMalformed{Length: len(csvRow)}
return StorageDiffInput{}, ErrRowMalformed{Length: len(csvRow)}
}
height, err := strconv.Atoi(csvRow[2])
if err != nil {
return StorageDiff{}, err
return StorageDiffInput{}, err
}
return StorageDiff{
return StorageDiffInput{
HashedAddress: HexToKeccak256Hash(csvRow[0]),
BlockHash: common.HexToHash(csvRow[1]),
BlockHeight: height,
@@ -53,14 +57,14 @@ func FromParityCsvRow(csvRow []string) (StorageDiff, error) {
}, nil
}
func FromGethStateDiff(account statediff.AccountDiff, stateDiff *statediff.StateDiff, storage statediff.StorageDiff) (StorageDiff, error) {
func FromGethStateDiff(account statediff.AccountDiff, stateDiff *statediff.StateDiff, storage statediff.StorageDiff) (StorageDiffInput, error) {
var decodedValue []byte
err := rlp.DecodeBytes(storage.Value, &decodedValue)
if err != nil {
return StorageDiff{}, err
return StorageDiffInput{}, err
}
return StorageDiff{
return StorageDiffInput{
HashedAddress: common.BytesToHash(account.Key),
BlockHash: stateDiff.BlockHash,
BlockHeight: int(stateDiff.BlockNumber.Int64()),
@@ -69,6 +73,13 @@ func FromGethStateDiff(account statediff.AccountDiff, stateDiff *statediff.State
}, nil
}
func ToPersistedDiff(raw StorageDiffInput, id int64) PersistedStorageDiff {
return PersistedStorageDiff{
StorageDiffInput: raw,
ID: id,
}
}
func HexToKeccak256Hash(hex string) common.Hash {
return crypto.Keccak256Hash(common.FromHex(hex))
}