Merge pull request #93 from vulcanize/vdb-570-storage-duplicates

(VDB-570) Handle duplicate storage diffs
This commit is contained in:
Rob Mulholand 2019-05-02 13:40:44 -05:00 committed by GitHub
commit 3b531950f2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 42 additions and 16 deletions

View File

@ -17,9 +17,12 @@
package utils
import (
"errors"
"fmt"
)
var ErrRowExists = errors.New("parsed row for storage diff already exists")
type ErrContractNotFound struct {
Contract string
}

View File

@ -78,7 +78,7 @@ func (storageWatcher StorageWatcher) processRow(row utils.StorageDiffRow) {
return
}
executeErr := storageTransformer.Execute(row)
if executeErr != nil {
if executeErr != nil && executeErr != utils.ErrRowExists {
logrus.Warn(fmt.Sprintf("error executing storage transformer: %s", executeErr))
queueErr := storageWatcher.Queue.Add(row)
if queueErr != nil {
@ -100,7 +100,7 @@ func (storageWatcher StorageWatcher) processQueue() {
continue
}
executeErr := storageTransformer.Execute(row)
if executeErr == nil {
if executeErr == nil || executeErr == utils.ErrRowExists {
storageWatcher.deleteRow(row.Id)
}
}

View File

@ -109,6 +109,18 @@ var _ = Describe("Storage Watcher", func() {
close(done)
})
It("does not queue row if transformer execution fails because row already exists", func(done Done) {
mockTransformer.ExecuteErr = utils.ErrRowExists
go storageWatcher.Execute(rows, errs, time.Hour)
Expect(<-errs).To(BeNil())
Consistently(func() bool {
return mockQueue.AddCalled
}).Should(BeFalse())
close(done)
})
It("queues row for later processing if transformer execution fails", func(done Done) {
mockTransformer.ExecuteErr = fakes.FakeError
@ -187,6 +199,17 @@ var _ = Describe("Storage Watcher", func() {
close(done)
})
It("deletes row from queue if transformer execution errors because row already exists", func(done Done) {
mockTransformer.ExecuteErr = utils.ErrRowExists
go storageWatcher.Execute(rows, errs, time.Nanosecond)
Eventually(func() int {
return mockQueue.DeletePassedId
}).Should(Equal(row.Id))
close(done)
})
It("logs error if deleting persisted row fails", func(done Done) {
mockQueue.DeleteErr = fakes.FakeError
tempFile, fileErr := ioutil.TempFile("", "log")