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

remove LIMIT 100; method to check continuity of headers returned but doesn't require

Browse Source 
the set to start at a specific block number; config for
account_transformer; review fixes; update schema
This commit is contained in:
Ian Norden
2019-04-05 17:21:59 -05:00
parent c1940c3e58
commit 9befc76fc6
20 changed files with 323 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/contract_watcher/light/repository/header_repository.go
+27 -23
View File
@@ -18,9 +18,9 @@ package repository
import (
"fmt"
"github.com/jmoiron/sqlx"
"github.com/hashicorp/golang-lru"
"github.com/jmoiron/sqlx"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
@@ -104,7 +104,6 @@ func (r *headerRepository) MarkHeaderChecked(headerID int64, id string) error {
VALUES ($1, $2)
ON CONFLICT (header_id) DO
UPDATE SET `+id+` = checked_headers.`+id+` + 1`, headerID, 1)
return err
}
@@ -124,7 +123,6 @@ func (r *headerRepository) MarkHeaderCheckedForAll(headerID int64, ids []string)
}
pgStr = pgStr[:len(pgStr)-2]
_, err := r.db.Exec(pgStr, headerID)
return err
}
@@ -134,7 +132,6 @@ func (r *headerRepository) MarkHeadersCheckedForAll(headers []core.Header, ids [
if err != nil {
return err
}
for _, header := range headers {
pgStr := "INSERT INTO public.checked_headers (header_id, "
for _, id := range ids {
@@ -155,8 +152,8 @@ func (r *headerRepository) MarkHeadersCheckedForAll(headers []core.Header, ids [
return err
}
}
return tx.Commit()
err = tx.Commit()
return err
}
// Returns missing headers for the provided checked_headers column id
@@ -164,14 +161,13 @@ func (r *headerRepository) MissingHeaders(startingBlockNumber, endingBlockNumber
var result []core.Header
var query string
var err error
if endingBlockNumber == -1 {
query = `SELECT headers.id, headers.block_number, headers.hash FROM headers
LEFT JOIN checked_headers on headers.id = header_id
WHERE (header_id ISNULL OR checked_headers.` + id + `=0)
AND headers.block_number >= $1
AND headers.eth_node_fingerprint = $2
ORDER BY headers.block_number LIMIT 100`
ORDER BY headers.block_number`
err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID)
} else {
query = `SELECT headers.id, headers.block_number, headers.hash FROM headers
@@ -180,10 +176,9 @@ func (r *headerRepository) MissingHeaders(startingBlockNumber, endingBlockNumber
AND headers.block_number >= $1
AND headers.block_number <= $2
AND headers.eth_node_fingerprint = $3
ORDER BY headers.block_number LIMIT 100`
ORDER BY headers.block_number`
err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
}
return contiguousHeaders(result, startingBlockNumber), err
}
@@ -192,34 +187,30 @@ func (r *headerRepository) MissingHeadersForAll(startingBlockNumber, endingBlock
var result []core.Header
var query string
var err error
baseQuery := `SELECT headers.id, headers.block_number, headers.hash FROM headers
LEFT JOIN checked_headers on headers.id = header_id
WHERE (header_id ISNULL`
for _, id := range ids {
baseQuery += ` OR checked_headers.` + id + `= 0`
}
if endingBlockNumber == -1 {
endStr := `) AND headers.block_number >= $1
AND headers.eth_node_fingerprint = $2
ORDER BY headers.block_number LIMIT 100`
ORDER BY headers.block_number`
query = baseQuery + endStr
err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID)
} else {
endStr := `) AND headers.block_number >= $1
AND headers.block_number <= $2
AND headers.eth_node_fingerprint = $3
ORDER BY headers.block_number LIMIT 100`
ORDER BY headers.block_number`
query = baseQuery + endStr
err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
}
return contiguousHeaders(result, startingBlockNumber), err
}
// Takes in an ordered sequence of headers and returns only the first contiguous segment
// Enforce continuity with previous segment with the appropriate startingBlockNumber
// Returns a continuous set of headers that is contiguous with the provided startingBlockNumber
func contiguousHeaders(headers []core.Header, startingBlockNumber int64) []core.Header {
if len(headers) < 1 {
return headers
@@ -243,7 +234,6 @@ func (r *headerRepository) MissingMethodsCheckedEventsIntersection(startingBlock
var result []core.Header
var query string
var err error
baseQuery := `SELECT headers.id, headers.block_number, headers.hash FROM headers
LEFT JOIN checked_headers on headers.id = header_id
WHERE (header_id IS NOT NULL`
@@ -255,23 +245,37 @@ func (r *headerRepository) MissingMethodsCheckedEventsIntersection(startingBlock
baseQuery += id + ` =0 AND `
}
baseQuery = baseQuery[:len(baseQuery)-5] + `) `
if endingBlockNumber == -1 {
endStr := `AND headers.block_number >= $1
AND headers.eth_node_fingerprint = $2
ORDER BY headers.block_number LIMIT 100`
ORDER BY headers.block_number`
query = baseQuery + endStr
err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID)
} else {
endStr := `AND headers.block_number >= $1
AND headers.block_number <= $2
AND headers.eth_node_fingerprint = $3
ORDER BY headers.block_number LIMIT 100`
ORDER BY headers.block_number`
query = baseQuery + endStr
err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
}
return continuousHeaders(result), err
}
return result, err
// Returns a continuous set of headers
func continuousHeaders(headers []core.Header) []core.Header {
if len(headers) < 1 {
return headers
}
previousHeader := headers[0].BlockNumber
for i := 1; i < len(headers); i++ {
previousHeader++
if headers[i].BlockNumber != previousHeader {
return headers[:i]
}
}
return headers
}
// Check the repositories column id cache for a value
@@ -280,7 +284,7 @@ func (r *headerRepository) CheckCache(key string) (interface{}, bool) {
}
// Used to mark a header checked as part of some external transaction so as to group into one commit
func MarkHeaderCheckedInTransaction(headerID int64, tx *sqlx.Tx, eventID string) error {
func (r *headerRepository) MarkHeaderCheckedInTransaction(headerID int64, tx *sqlx.Tx, eventID string) error {
_, err := tx.Exec(`INSERT INTO public.checked_headers (header_id, `+eventID+`)
VALUES ($1, $2)
ON CONFLICT (header_id) DO
pkg/contract_watcher/light/repository/header_repository_test.go
-21
View File
@@ -203,18 +203,6 @@ var _ = Describe("Repository", func() {
Expect(missingHeaders[1].BlockNumber).To(Equal(int64(6194633)))
})
It("Returns at most 100 headers", func() {
add102Headers(coreHeaderRepo)
err := contractHeaderRepo.AddCheckColumns(eventIDs)
Expect(err).ToNot(HaveOccurred())
missingHeaders, err := contractHeaderRepo.MissingHeadersForAll(6194632, 6194733, eventIDs)
Expect(err).ToNot(HaveOccurred())
Expect(len(missingHeaders)).To(Equal(100))
Expect(missingHeaders[0].BlockNumber).To(Equal(int64(6194632)))
Expect(missingHeaders[1].BlockNumber).To(Equal(int64(6194633)))
})
It("Fails if one of the eventIDs does not yet exist in check_headers table", func() {
addHeaders(coreHeaderRepo)
err := contractHeaderRepo.AddCheckColumns(eventIDs)
@@ -357,12 +345,3 @@ func addDiscontinuousHeaders(coreHeaderRepo repositories.HeaderRepository) {
coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader2)
coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader4)
}
func add102Headers(coreHeaderRepo repositories.HeaderRepository) {
baseHeader := mocks.MockHeader1
for i := 6194632; i < 6194733; i++ {
_, err := coreHeaderRepo.CreateOrUpdateHeader(baseHeader)
Expect(err).ToNot(HaveOccurred())
baseHeader.BlockNumber++
}
}
pkg/contract_watcher/shared/repository/event_repository.go
+1 -1
View File
@@ -23,7 +23,7 @@ import (
"github.com/hashicorp/golang-lru"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/light/repository"
"github.com/vulcanize/vulcanizedb/libraries/shared/repository"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)