Add light sync command

- Only syncs block headers (excludes block bodies, transactions, receipts, and logs) - Modifies validation window to include the most recent block - Isolates validation window to the variable defined in the cmd directory (blocks have a separate variable defined in the block_repository for determining when to set a block as final)
2018-07-18 16:34:12 -05:00
parent a620fc6a51
commit 1355271011
43 changed files with 1092 additions and 272 deletions
@@ -0,0 +1,30 @@
+package inmemory
+
+import "github.com/vulcanize/vulcanizedb/pkg/core"
+
+type HeaderRepository struct {
+	memory *InMemory
+}
+
+func NewHeaderRepository(memory *InMemory) *HeaderRepository {
+	return &HeaderRepository{memory: memory}
+}
+
+func (repository *HeaderRepository) CreateOrUpdateHeader(header core.Header) (int64, error) {
+	repository.memory.headers[header.BlockNumber] = header
+	return 0, nil
+}
+
+func (repository *HeaderRepository) GetHeader(blockNumber int64) (core.Header, error) {
+	return repository.memory.headers[blockNumber], nil
+}
+
+func (repository *HeaderRepository) MissingBlockNumbers(startingBlockNumber, endingBlockNumber int64, nodeID string) []int64 {
+	missingNumbers := []int64{}
+	for blockNumber := int64(startingBlockNumber); blockNumber <= endingBlockNumber; blockNumber++ {
+		if _, ok := repository.memory.headers[blockNumber]; !ok {
+			missingNumbers = append(missingNumbers, blockNumber)
+		}
+	}
+	return missingNumbers
+}
@@ -10,21 +10,23 @@ const (
 )

 type InMemory struct {
-	blocks                       map[int64]core.Block
-	receipts                     map[string]core.Receipt
-	contracts                    map[string]core.Contract
-	logs                         map[string][]core.Log
-	logFilters                   map[string]filters.LogFilter
 	CreateOrUpdateBlockCallCount int
+	blocks                       map[int64]core.Block
+	contracts                    map[string]core.Contract
+	headers                      map[int64]core.Header
+	logFilters                   map[string]filters.LogFilter
+	logs                         map[string][]core.Log
+	receipts                     map[string]core.Receipt
 }

 func NewInMemory() *InMemory {
 	return &InMemory{
 		CreateOrUpdateBlockCallCount: 0,
 		blocks:     make(map[int64]core.Block),
-		receipts:   make(map[string]core.Receipt),
 		contracts:  make(map[string]core.Contract),
-		logs:       make(map[string][]core.Log),
+		headers:    make(map[int64]core.Header),
 		logFilters: make(map[string]filters.LogFilter),
+		logs:       make(map[string][]core.Log),
+		receipts:   make(map[string]core.Receipt),
 	}
 }
@@ -0,0 +1,80 @@
+package repositories
+
+import (
+	"database/sql"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+)
+
+type HeaderRepository struct {
+	database *postgres.DB
+}
+
+func NewHeaderRepository(database *postgres.DB) HeaderRepository {
+	return HeaderRepository{database: database}
+}
+
+func (repository HeaderRepository) CreateOrUpdateHeader(header core.Header) (int64, error) {
+	hash, err := repository.getHeaderHash(header)
+	if err != nil {
+		if headerDoesNotExist(err) {
+			return repository.insertHeader(header)
+		}
+		return 0, err
+	}
+	if headerMustBeReplaced(hash, header) {
+		return repository.replaceHeader(header)
+	}
+	return 0, err
+}
+
+func (repository HeaderRepository) GetHeader(blockNumber int64) (core.Header, error) {
+	var header core.Header
+	err := repository.database.Get(&header, `SELECT block_number, hash, raw FROM headers WHERE block_number = $1 AND eth_node_fingerprint = $2`,
+		blockNumber, repository.database.Node.ID)
+	return header, err
+}
+
+func (repository HeaderRepository) MissingBlockNumbers(startingBlockNumber, endingBlockNumber int64, nodeID string) []int64 {
+	numbers := make([]int64, 0)
+	repository.database.Select(&numbers, `SELECT all_block_numbers
+	  FROM (
+		  SELECT generate_series($1::INT, $2::INT) AS all_block_numbers) series
+	  WHERE all_block_numbers NOT IN (
+		  SELECT block_number FROM headers WHERE eth_node_fingerprint = $3
+	  ) `,
+		startingBlockNumber, endingBlockNumber, nodeID)
+	return numbers
+}
+
+func headerMustBeReplaced(hash string, header core.Header) bool {
+	return hash != header.Hash
+}
+
+func headerDoesNotExist(err error) bool {
+	return err == sql.ErrNoRows
+}
+
+func (repository HeaderRepository) getHeaderHash(header core.Header) (string, error) {
+	var hash string
+	err := repository.database.Get(&hash, `SELECT hash FROM headers WHERE block_number = $1 AND eth_node_fingerprint = $2`,
+		header.BlockNumber, repository.database.Node.ID)
+	return hash, err
+}
+
+func (repository HeaderRepository) insertHeader(header core.Header) (int64, error) {
+	var headerId int64
+	err := repository.database.QueryRowx(
+		`INSERT INTO public.headers (block_number, hash, raw, eth_node_id, eth_node_fingerprint) VALUES ($1, $2, $3, $4, $5) RETURNING id`,
+		header.BlockNumber, header.Hash, header.Raw, repository.database.NodeID, repository.database.Node.ID).Scan(&headerId)
+	return headerId, err
+}
+
+func (repository HeaderRepository) replaceHeader(header core.Header) (int64, error) {
+	_, err := repository.database.Exec(`DELETE FROM headers WHERE block_number = $1 AND eth_node_fingerprint = $2`,
+		header.BlockNumber, repository.database.Node.ID)
+	if err != nil {
+		return 0, err
+	}
+	return repository.insertHeader(header)
+}
@@ -0,0 +1,248 @@
+package repositories_test
+
+import (
+	"database/sql"
+	"github.com/ethereum/go-ethereum/common"
+	. "github.com/onsi/ginkgo"
+	. "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/test_config"
+)
+
+var _ = Describe("Block header repository", func() {
+	Describe("creating or updating a header", func() {
+
+		It("adds a header", func() {
+			node := core.Node{}
+			db := test_config.NewTestDB(node)
+			repo := repositories.NewHeaderRepository(db)
+			header := core.Header{
+				BlockNumber: 100,
+				Hash:        common.BytesToHash([]byte{1, 2, 3, 4, 5}).Hex(),
+				Raw:         []byte{1, 2, 3, 4, 5},
+			}
+
+			_, err := repo.CreateOrUpdateHeader(header)
+
+			Expect(err).NotTo(HaveOccurred())
+			var dbHeader core.Header
+			err = db.Get(&dbHeader, `SELECT block_number, hash, raw FROM public.headers WHERE block_number = $1`, header.BlockNumber)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(dbHeader.BlockNumber).To(Equal(header.BlockNumber))
+			Expect(dbHeader.Hash).To(Equal(header.Hash))
+			Expect(dbHeader.Raw).To(Equal(header.Raw))
+		})
+
+		It("adds node data to header", func() {
+			node := core.Node{ID: "EthNodeFingerprint"}
+			db := test_config.NewTestDB(node)
+			repo := repositories.NewHeaderRepository(db)
+			header := core.Header{BlockNumber: 100}
+
+			_, err := repo.CreateOrUpdateHeader(header)
+
+			Expect(err).NotTo(HaveOccurred())
+			var ethNodeId int64
+			err = db.Get(&ethNodeId, `SELECT eth_node_id FROM public.headers WHERE block_number = $1`, header.BlockNumber)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(ethNodeId).To(Equal(db.NodeID))
+			var ethNodeFingerprint string
+			err = db.Get(&ethNodeFingerprint, `SELECT eth_node_fingerprint FROM public.headers WHERE block_number = $1`, header.BlockNumber)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(ethNodeFingerprint).To(Equal(db.Node.ID))
+		})
+
+		It("does not duplicate headers", func() {
+			node := core.Node{}
+			db := test_config.NewTestDB(node)
+			repo := repositories.NewHeaderRepository(db)
+			header := core.Header{
+				BlockNumber: 100,
+				Hash:        common.BytesToHash([]byte{1, 2, 3, 4, 5}).Hex(),
+				Raw:         []byte{1, 2, 3, 4, 5},
+			}
+
+			_, err := repo.CreateOrUpdateHeader(header)
+			Expect(err).NotTo(HaveOccurred())
+
+			_, err = repo.CreateOrUpdateHeader(header)
+			Expect(err).NotTo(HaveOccurred())
+
+			var dbHeaders []core.Header
+			err = db.Select(&dbHeaders, `SELECT block_number, hash, raw FROM public.headers WHERE block_number = $1`, header.BlockNumber)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(dbHeaders)).To(Equal(1))
+		})
+
+		It("replaces header if hash is different", func() {
+			node := core.Node{}
+			db := test_config.NewTestDB(node)
+			repo := repositories.NewHeaderRepository(db)
+			header := core.Header{
+				BlockNumber: 100,
+				Hash:        common.BytesToHash([]byte{1, 2, 3, 4, 5}).Hex(),
+				Raw:         []byte{1, 2, 3, 4, 5},
+			}
+			_, err := repo.CreateOrUpdateHeader(header)
+			Expect(err).NotTo(HaveOccurred())
+			headerTwo := core.Header{
+				BlockNumber: header.BlockNumber,
+				Hash:        common.BytesToHash([]byte{5, 4, 3, 2, 1}).Hex(),
+				Raw:         []byte{5, 4, 3, 2, 1},
+			}
+
+			_, err = repo.CreateOrUpdateHeader(headerTwo)
+
+			Expect(err).NotTo(HaveOccurred())
+			var dbHeader core.Header
+			err = db.Get(&dbHeader, `SELECT block_number, hash, raw FROM headers WHERE block_number = $1`, header.BlockNumber)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(dbHeader.Hash).To(Equal(headerTwo.Hash))
+			Expect(dbHeader.Raw).To(Equal(headerTwo.Raw))
+		})
+
+		It("does not replace header if node fingerprint is different", func() {
+			node := core.Node{ID: "Fingerprint"}
+			db := test_config.NewTestDB(node)
+			repo := repositories.NewHeaderRepository(db)
+			header := core.Header{
+				BlockNumber: 100,
+				Hash:        common.BytesToHash([]byte{1, 2, 3, 4, 5}).Hex(),
+				Raw:         []byte{1, 2, 3, 4, 5},
+			}
+			_, err := repo.CreateOrUpdateHeader(header)
+			nodeTwo := core.Node{ID: "FingerprintTwo"}
+			dbTwo, err := postgres.NewDB(test_config.DBConfig, nodeTwo)
+			Expect(err).NotTo(HaveOccurred())
+			repoTwo := repositories.NewHeaderRepository(dbTwo)
+			headerTwo := core.Header{
+				BlockNumber: header.BlockNumber,
+				Hash:        common.BytesToHash([]byte{5, 4, 3, 2, 1}).Hex(),
+				Raw:         []byte{5, 4, 3, 2, 1},
+			}
+
+			_, err = repoTwo.CreateOrUpdateHeader(headerTwo)
+
+			Expect(err).NotTo(HaveOccurred())
+			var dbHeaders []core.Header
+			err = dbTwo.Select(&dbHeaders, `SELECT block_number, hash, raw FROM headers WHERE block_number = $1`, header.BlockNumber)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(dbHeaders)).To(Equal(2))
+		})
+
+		It("only replaces header with matching node fingerprint", func() {
+			node := core.Node{ID: "Fingerprint"}
+			db := test_config.NewTestDB(node)
+			repo := repositories.NewHeaderRepository(db)
+			header := core.Header{
+				BlockNumber: 100,
+				Hash:        common.BytesToHash([]byte{1, 2, 3, 4, 5}).Hex(),
+				Raw:         []byte{1, 2, 3, 4, 5},
+			}
+			_, err := repo.CreateOrUpdateHeader(header)
+			nodeTwo := core.Node{ID: "FingerprintTwo"}
+			dbTwo, err := postgres.NewDB(test_config.DBConfig, nodeTwo)
+			Expect(err).NotTo(HaveOccurred())
+			repoTwo := repositories.NewHeaderRepository(dbTwo)
+			headerTwo := core.Header{
+				BlockNumber: header.BlockNumber,
+				Hash:        common.BytesToHash([]byte{5, 4, 3, 2, 1}).Hex(),
+				Raw:         []byte{5, 4, 3, 2, 1},
+			}
+			_, err = repoTwo.CreateOrUpdateHeader(headerTwo)
+			headerThree := core.Header{
+				BlockNumber: header.BlockNumber,
+				Hash:        common.BytesToHash([]byte{1, 1, 1, 1, 1}).Hex(),
+				Raw:         []byte{1, 1, 1, 1, 1},
+			}
+
+			_, err = repoTwo.CreateOrUpdateHeader(headerThree)
+
+			Expect(err).NotTo(HaveOccurred())
+			var dbHeaders []core.Header
+			err = dbTwo.Select(&dbHeaders, `SELECT block_number, hash, raw FROM headers WHERE block_number = $1`, header.BlockNumber)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(dbHeaders)).To(Equal(2))
+			Expect(dbHeaders[0].Hash).To(Or(Equal(header.Hash), Equal(headerThree.Hash)))
+			Expect(dbHeaders[1].Hash).To(Or(Equal(header.Hash), Equal(headerThree.Hash)))
+			Expect(dbHeaders[0].Raw).To(Or(Equal(header.Raw), Equal(headerThree.Raw)))
+			Expect(dbHeaders[1].Raw).To(Or(Equal(header.Raw), Equal(headerThree.Raw)))
+		})
+	})
+
+	Describe("Getting a header", func() {
+		It("returns header if it exists", func() {
+			node := core.Node{}
+			db := test_config.NewTestDB(node)
+			repo := repositories.NewHeaderRepository(db)
+			header := core.Header{
+				BlockNumber: 100,
+				Hash:        common.BytesToHash([]byte{1, 2, 3, 4, 5}).Hex(),
+				Raw:         []byte{1, 2, 3, 4, 5},
+			}
+			_, err := repo.CreateOrUpdateHeader(header)
+			Expect(err).NotTo(HaveOccurred())
+
+			dbHeader, err := repo.GetHeader(header.BlockNumber)
+
+			Expect(err).NotTo(HaveOccurred())
+			Expect(dbHeader).To(Equal(header))
+		})
+
+		It("does not return header for a different node fingerprint", func() {
+			node := core.Node{}
+			db := test_config.NewTestDB(node)
+			repo := repositories.NewHeaderRepository(db)
+			header := core.Header{
+				BlockNumber: 100,
+				Hash:        common.BytesToHash([]byte{1, 2, 3, 4, 5}).Hex(),
+				Raw:         []byte{1, 2, 3, 4, 5},
+			}
+			_, err := repo.CreateOrUpdateHeader(header)
+			Expect(err).NotTo(HaveOccurred())
+			nodeTwo := core.Node{ID: "NodeFingerprintTwo"}
+			dbTwo, err := postgres.NewDB(test_config.DBConfig, nodeTwo)
+			Expect(err).NotTo(HaveOccurred())
+			repoTwo := repositories.NewHeaderRepository(dbTwo)
+
+			_, err = repoTwo.GetHeader(header.BlockNumber)
+
+			Expect(err).To(HaveOccurred())
+			Expect(err).To(MatchError(sql.ErrNoRows))
+		})
+	})
+
+	Describe("Getting missing headers", func() {
+		It("returns block numbers for headers not in the database", func() {
+			node := core.Node{}
+			db := test_config.NewTestDB(node)
+			repo := repositories.NewHeaderRepository(db)
+			repo.CreateOrUpdateHeader(core.Header{BlockNumber: 1})
+			repo.CreateOrUpdateHeader(core.Header{BlockNumber: 3})
+			repo.CreateOrUpdateHeader(core.Header{BlockNumber: 5})
+
+			missingBlockNumbers := repo.MissingBlockNumbers(1, 5, node.ID)
+
+			Expect(missingBlockNumbers).To(ConsistOf([]int64{2, 4}))
+		})
+
+		It("does not count headers created by a different node fingerprint", func() {
+			node := core.Node{ID: "NodeFingerprint"}
+			db := test_config.NewTestDB(node)
+			repo := repositories.NewHeaderRepository(db)
+			repo.CreateOrUpdateHeader(core.Header{BlockNumber: 1})
+			repo.CreateOrUpdateHeader(core.Header{BlockNumber: 3})
+			repo.CreateOrUpdateHeader(core.Header{BlockNumber: 5})
+			nodeTwo := core.Node{ID: "NodeFingerprintTwo"}
+			dbTwo, err := postgres.NewDB(test_config.DBConfig, nodeTwo)
+			Expect(err).NotTo(HaveOccurred())
+			repoTwo := repositories.NewHeaderRepository(dbTwo)
+
+			missingBlockNumbers := repoTwo.MissingBlockNumbers(1, 5, nodeTwo.ID)
+
+			Expect(missingBlockNumbers).To(ConsistOf([]int64{1, 2, 3, 4, 5}))
+		})
+	})
+})
@@ -14,7 +14,7 @@ var ErrBlockDoesNotExist = func(blockNumber int64) error {
 type BlockRepository interface {
 	CreateOrUpdateBlock(block core.Block) (int64, error)
 	GetBlock(blockNumber int64) (core.Block, error)
-	MissingBlockNumbers(startingBlockNumber int64, endingBlockNumber int64, nodeId string) []int64
+	MissingBlockNumbers(startingBlockNumber, endingBlockNumber int64, nodeID string) []int64
 	SetBlocksStatus(chainHead int64)
 }

@@ -37,6 +37,12 @@ type FilterRepository interface {
 	GetFilter(name string) (filters.LogFilter, error)
 }

+type HeaderRepository interface {
+	CreateOrUpdateHeader(header core.Header) (int64, error)
+	GetHeader(blockNumber int64) (core.Header, error)
+	MissingBlockNumbers(startingBlockNumber, endingBlockNumber int64, nodeID string) []int64
+}
+
 type LogRepository interface {
 	CreateLogs(logs []core.Log, receiptId int64) error
 	GetLogs(address string, blockNumber int64) []core.Log