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

Log checked headers for MCD events (#43)

Browse Source 
* Log checked headers for Bite events

* Log checked headers for Drip drip events

* Log checked headers for Drip file events

- mark headers checked for 3 implementations of file on the Drip contract

* Log checked headers for Frob frob events

* Log checked headers for Pit file events

- mark headers checked for 3 implementations of file on the Pit contract

* Log checked headers for Vat init events
This commit is contained in:
Rob Mulholand
2018-10-02 17:25:38 -05:00
committed by Elizabeth
parent 5a2d05b8d0
commit bf44ec614a
93 changed files with 2439 additions and 772 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/transformers/vat_init/converter.go
+20 -12
View File
@@ -23,23 +23,31 @@ import (
)
type Converter interface {
ToModel(ethLog types.Log) (VatInitModel, error)
ToModels(ethLogs []types.Log) ([]VatInitModel, error)
}
type VatInitConverter struct{}
func (VatInitConverter) ToModel(ethLog types.Log) (VatInitModel, error) {
err := verifyLog(ethLog)
if err != nil {
return VatInitModel{}, err
func (VatInitConverter) ToModels(ethLogs []types.Log) ([]VatInitModel, error) {
var models []VatInitModel
for _, ethLog := range ethLogs {
err := verifyLog(ethLog)
if err != nil {
return nil, err
}
ilk := string(bytes.Trim(ethLog.Topics[1].Bytes(), "\x00"))
raw, err := json.Marshal(ethLog)
if err != nil {
return nil, err
}
model := VatInitModel{
Ilk: ilk,
TransactionIndex: ethLog.TxIndex,
Raw: raw,
}
models = append(models, model)
}
ilk := string(bytes.Trim(ethLog.Topics[1].Bytes(), "\x00"))
raw, err := json.Marshal(ethLog)
return VatInitModel{
Ilk: ilk,
TransactionIndex: ethLog.TxIndex,
Raw: raw,
}, err
return models, nil
}
func verifyLog(log types.Log) error {
pkg/transformers/vat_init/converter_test.go
+4 -3
View File
@@ -28,7 +28,7 @@ var _ = Describe("Vat init converter", func() {
converter := vat_init.VatInitConverter{}
badLog := types.Log{}
_, err := converter.ToModel(badLog)
_, err := converter.ToModels([]types.Log{badLog})
Expect(err).To(HaveOccurred())
})
@@ -36,9 +36,10 @@ var _ = Describe("Vat init converter", func() {
It("converts a log to an model", func() {
converter := vat_init.VatInitConverter{}
model, err := converter.ToModel(test_data.EthVatInitLog)
models, err := converter.ToModels([]types.Log{test_data.EthVatInitLog})
Expect(err).NotTo(HaveOccurred())
Expect(model).To(Equal(test_data.VatInitModel))
Expect(len(models)).To(Equal(1))
Expect(models[0]).To(Equal(test_data.VatInitModel))
})
})
pkg/transformers/vat_init/repository.go
+28 -9
View File
@@ -20,7 +20,8 @@ import (
)
type Repository interface {
Create(headerID int64, model VatInitModel) error
Create(headerID int64, models []VatInitModel) error
MarkHeaderChecked(headerID int64) error
MissingHeaders(startingBlockNumber, endingBlockNumber int64) ([]core.Header, error)
}
@@ -34,12 +35,31 @@ func NewVatInitRepository(db *postgres.DB) VatInitRepository {
}
}
func (repository VatInitRepository) Create(headerID int64, model VatInitModel) error {
_, err := repository.db.Exec(
`INSERT into maker.vat_init (header_id, ilk, tx_idx, raw_log)
func (repository VatInitRepository) Create(headerID int64, models []VatInitModel) error {
tx, err := repository.db.Begin()
if err != nil {
return err
}
for _, model := range models {
_, err = tx.Exec(
`INSERT into maker.vat_init (header_id, ilk, tx_idx, raw_log)
VALUES($1, $2, $3, $4)`,
headerID, model.Ilk, model.TransactionIndex, model.Raw,
)
headerID, model.Ilk, model.TransactionIndex, model.Raw,
)
if err != nil {
tx.Rollback()
return err
}
}
return tx.Commit()
}
func (repository VatInitRepository) MarkHeaderChecked(headerID int64) error {
_, err := repository.db.Exec(`INSERT INTO public.checked_headers (header_id, vat_init_checked)
VALUES ($1, $2)
ON CONFLICT (header_id) DO
UPDATE SET vat_init_checked = $2`, headerID, true)
return err
}
@@ -48,8 +68,8 @@ func (repository VatInitRepository) MissingHeaders(startingBlockNumber, endingBl
err := repository.db.Select(
&result,
`SELECT headers.id, headers.block_number FROM headers
LEFT JOIN maker.vat_init on headers.id = header_id
WHERE header_id ISNULL
LEFT JOIN checked_headers on headers.id = header_id
WHERE (header_id ISNULL OR vat_init_checked IS FALSE)
AND headers.block_number >= $1
AND headers.block_number <= $2
AND headers.eth_node_fingerprint = $3`,
@@ -57,6 +77,5 @@ func (repository VatInitRepository) MissingHeaders(startingBlockNumber, endingBl
endingBlockNumber,
repository.db.Node.ID,
)
return result, err
}
pkg/transformers/vat_init/repository_test.go
+76 -8
View File
@@ -21,13 +21,14 @@ import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data"
"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_init"
"github.com/vulcanize/vulcanizedb/test_config"
)
var _ = Describe("", func() {
var _ = Describe("Vat init repository", func() {
Describe("Create", func() {
It("adds a vat event", func() {
db := test_config.NewTestDB(core.Node{})
@@ -37,7 +38,7 @@ var _ = Describe("", func() {
Expect(err).NotTo(HaveOccurred())
vatInitRepository := vat_init.NewVatInitRepository(db)
err = vatInitRepository.Create(headerID, test_data.VatInitModel)
err = vatInitRepository.Create(headerID, []vat_init.VatInitModel{test_data.VatInitModel})
Expect(err).NotTo(HaveOccurred())
var dbVatInit vat_init.VatInitModel
@@ -55,10 +56,10 @@ var _ = Describe("", func() {
headerID, err := headerRepository.CreateOrUpdateHeader(core.Header{})
Expect(err).NotTo(HaveOccurred())
vatInitRepository := vat_init.NewVatInitRepository(db)
err = vatInitRepository.Create(headerID, test_data.VatInitModel)
err = vatInitRepository.Create(headerID, []vat_init.VatInitModel{test_data.VatInitModel})
Expect(err).NotTo(HaveOccurred())
err = vatInitRepository.Create(headerID, test_data.VatInitModel)
err = vatInitRepository.Create(headerID, []vat_init.VatInitModel{test_data.VatInitModel})
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
@@ -71,7 +72,7 @@ var _ = Describe("", func() {
headerID, err := headerRepository.CreateOrUpdateHeader(core.Header{})
Expect(err).NotTo(HaveOccurred())
vatInitRepository := vat_init.NewVatInitRepository(db)
err = vatInitRepository.Create(headerID, test_data.VatInitModel)
err = vatInitRepository.Create(headerID, []vat_init.VatInitModel{test_data.VatInitModel})
Expect(err).NotTo(HaveOccurred())
_, err = db.Exec(`DELETE FROM headers WHERE id = $1`, headerID)
@@ -84,8 +85,48 @@ var _ = Describe("", func() {
})
})
Describe("MarkHeaderChecked", func() {
var (
db *postgres.DB
vatInitRepository vat_init.Repository
err error
headerID int64
)
BeforeEach(func() {
db = test_config.NewTestDB(core.Node{})
test_config.CleanTestDB(db)
headerRepository := repositories.NewHeaderRepository(db)
headerID, err = headerRepository.CreateOrUpdateHeader(core.Header{})
Expect(err).NotTo(HaveOccurred())
vatInitRepository = vat_init.NewVatInitRepository(db)
})
It("creates a row for a new headerID", func() {
err = vatInitRepository.MarkHeaderChecked(headerID)
Expect(err).NotTo(HaveOccurred())
var headerChecked bool
err = db.Get(&headerChecked, `SELECT vat_init_checked FROM public.checked_headers WHERE header_id = $1`, headerID)
Expect(err).NotTo(HaveOccurred())
Expect(headerChecked).To(BeTrue())
})
It("updates row when headerID already exists", func() {
_, err = db.Exec(`INSERT INTO public.checked_headers (header_id) VALUES ($1)`, headerID)
err = vatInitRepository.MarkHeaderChecked(headerID)
Expect(err).NotTo(HaveOccurred())
var headerChecked bool
err = db.Get(&headerChecked, `SELECT vat_init_checked FROM public.checked_headers WHERE header_id = $1`, headerID)
Expect(err).NotTo(HaveOccurred())
Expect(headerChecked).To(BeTrue())
})
})
Describe("MissingHeaders", func() {
It("returns headers with no associated vat event", func() {
It("returns headers that haven't been checked", func() {
db := test_config.NewTestDB(core.Node{})
test_config.CleanTestDB(db)
headerRepository := repositories.NewHeaderRepository(db)
@@ -100,7 +141,7 @@ var _ = Describe("", func() {
Expect(err).NotTo(HaveOccurred())
}
vatInitRepository := vat_init.NewVatInitRepository(db)
err := vatInitRepository.Create(headerIDs[1], test_data.VatInitModel)
err := vatInitRepository.MarkHeaderChecked(headerIDs[1])
Expect(err).NotTo(HaveOccurred())
headers, err := vatInitRepository.MissingHeaders(startingBlockNumber, endingBlockNumber)
@@ -111,6 +152,33 @@ var _ = Describe("", func() {
Expect(headers[1].BlockNumber).To(Or(Equal(startingBlockNumber), Equal(endingBlockNumber)))
})
It("only treats headers as checked if drip drip logs have been checked", func() {
db := test_config.NewTestDB(core.Node{})
test_config.CleanTestDB(db)
headerRepository := repositories.NewHeaderRepository(db)
startingBlockNumber := int64(1)
vatInitBlockNumber := int64(2)
endingBlockNumber := int64(3)
blockNumbers := []int64{startingBlockNumber, vatInitBlockNumber, endingBlockNumber, endingBlockNumber + 1}
var headerIDs []int64
for _, n := range blockNumbers {
headerID, err := headerRepository.CreateOrUpdateHeader(core.Header{BlockNumber: n})
headerIDs = append(headerIDs, headerID)
Expect(err).NotTo(HaveOccurred())
}
VatInitRepository := vat_init.NewVatInitRepository(db)
_, err := db.Exec(`INSERT INTO public.checked_headers (header_id) VALUES ($1)`, headerIDs[1])
Expect(err).NotTo(HaveOccurred())
headers, err := VatInitRepository.MissingHeaders(startingBlockNumber, endingBlockNumber)
Expect(err).NotTo(HaveOccurred())
Expect(len(headers)).To(Equal(3))
Expect(headers[0].BlockNumber).To(Or(Equal(startingBlockNumber), Equal(endingBlockNumber), Equal(vatInitBlockNumber)))
Expect(headers[1].BlockNumber).To(Or(Equal(startingBlockNumber), Equal(endingBlockNumber), Equal(vatInitBlockNumber)))
Expect(headers[2].BlockNumber).To(Or(Equal(startingBlockNumber), Equal(endingBlockNumber), Equal(vatInitBlockNumber)))
})
It("only returns headers associated with the current node", func() {
db := test_config.NewTestDB(core.Node{})
test_config.CleanTestDB(db)
@@ -128,7 +196,7 @@ var _ = Describe("", func() {
}
vatInitRepository := vat_init.NewVatInitRepository(db)
vatInitRepositoryTwo := vat_init.NewVatInitRepository(dbTwo)
err := vatInitRepository.Create(headerIDs[0], test_data.VatInitModel)
err := vatInitRepository.MarkHeaderChecked(headerIDs[0])
Expect(err).NotTo(HaveOccurred())
nodeOneMissingHeaders, err := vatInitRepository.MissingHeaders(blockNumbers[0], blockNumbers[len(blockNumbers)-1])
pkg/transformers/vat_init/transformer.go
+10 -6
View File
@@ -59,16 +59,20 @@ func (transformer VatInitTransformer) Execute() error {
if err != nil {
return err
}
for _, log := range matchingLogs {
model, err := transformer.Converter.ToModel(log)
if err != nil {
return err
}
err = transformer.Repository.Create(header.Id, model)
if len(matchingLogs) < 1 {
err = transformer.Repository.MarkHeaderChecked(header.Id)
if err != nil {
return err
}
}
models, err := transformer.Converter.ToModels(matchingLogs)
if err != nil {
return err
}
err = transformer.Repository.Create(header.Id, models)
if err != nil {
return err
}
}
return nil
}
pkg/transformers/vat_init/transformer_test.go
+38 -2
View File
@@ -96,6 +96,42 @@ var _ = Describe("Vat init transformer", func() {
Expect(err).To(MatchError(fakes.FakeError))
})
It("marks header checked if no logs returned", func() {
mockConverter := &vat_init_mocks.MockVatInitConverter{}
mockRepository := &vat_init_mocks.MockVatInitRepository{}
headerID := int64(123)
mockRepository.SetMissingHeaders([]core.Header{{Id: headerID}})
mockFetcher := &mocks.MockLogFetcher{}
transformer := vat_init.VatInitTransformer{
Converter: mockConverter,
Fetcher: mockFetcher,
Repository: mockRepository,
}
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
mockRepository.AssertMarkHeaderCheckedCalledWith(headerID)
})
It("returns error if marking header checked returns err", func() {
mockConverter := &vat_init_mocks.MockVatInitConverter{}
mockRepository := &vat_init_mocks.MockVatInitRepository{}
mockRepository.SetMissingHeaders([]core.Header{{Id: int64(123)}})
mockRepository.SetMarkHeaderCheckedErr(fakes.FakeError)
mockFetcher := &mocks.MockLogFetcher{}
transformer := vat_init.VatInitTransformer{
Converter: mockConverter,
Fetcher: mockFetcher,
Repository: mockRepository,
}
err := transformer.Execute()
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(fakes.FakeError))
})
It("converts matching logs", func() {
converter := &vat_init_mocks.MockVatInitConverter{}
fetcher := &mocks.MockLogFetcher{}
@@ -111,7 +147,7 @@ var _ = Describe("Vat init transformer", func() {
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
Expect(converter.PassedLog).To(Equal(test_data.EthVatInitLog))
Expect(converter.PassedLogs).To(Equal([]types.Log{test_data.EthVatInitLog}))
})
It("returns error if converter returns error", func() {
@@ -150,7 +186,7 @@ var _ = Describe("Vat init transformer", func() {
Expect(err).NotTo(HaveOccurred())
Expect(repository.PassedHeaderID).To(Equal(fakeHeader.Id))
Expect(repository.PassedModel).To(Equal(test_data.VatInitModel))
Expect(repository.PassedModels).To(Equal([]vat_init.VatInitModel{test_data.VatInitModel}))
})
It("returns error if repository returns error for create", func() {